//! Session-lived Lua runtime, registered with libpanto as a single
//! `ToolSource`.
//!
//! This replaces the per-call `lua_State` model of phase 3 (LuaTool +
//! LuaStatePool). The CLI maintains exactly one `lua_State` per SESSION:
//! the runtime's lifetime IS the session's, and `/new`//`/resume` tear it
//! down and boot a fresh one through the startup path (see the lifetime
//! contract in `extension_loader.zig`). Every extension is loaded into it
//! once; extension top-level code runs exactly once at session start. Tool
//! handlers are stored in the Lua registry and looked up by tool name on
//! each call.
//!
//! libpanto delivers all tool calls targeting Lua-defined tools in one
//! `invoke_batch` per turn, on a single thread (see
//! `code.tjp.lol/libpantograph.git/src/tool_source.zig`). This runtime then runs each call as
//! a Lua *coroutine*. When (later) we wire in libuv via `luv`, a yield
//! inside a coroutine returns control to the runtime, which drives
//! `uv.run()` until any coroutine is resumable.
//!
//! For now (step 2 of LUA_MAKEOVER.md): no batteries yet. Each call's
//! coroutine runs to completion synchronously. A handler that yields
//! to nothing currently leaves the call permanently suspended — we
//! surface that as a `LuaHandlerYielded` error so it's at least visible.
//! Step 4 (install `luv`) and step 5 (wire `coro-*`) make yields
//! productive.
//!
//! Concurrency contract for source-backed tools: "coroutine-safe within
//! this runtime". Concurrent host entry into the same `lua_State` is
//! *not* safe; libpanto's grouped-dispatch guarantees this never happens.

const std = @import("std");
const Allocator = std.mem.Allocator;
const panto = @import("panto");
const lua_bridge = @import("lua_bridge.zig");
const lua_event_bridge = @import("lua_event_bridge.zig");
const panto_home = @import("panto_home.zig");
const ui_event = @import("tui_event.zig");

const c = lua_bridge.c;
const Io = std.Io;
const EventBridge = lua_event_bridge.EventBridge;

pub const SOURCE_NAME = "panto-lua";

/// Errors produced by the runtime above and beyond bridge errors.
pub const RuntimeError = error{
    LuaInitFailed,
    LuaHandlerNotFound,
    LuaHandlerYielded,
    LuaHandlerCrashed,
    BadHandlerReturn,
    InputNotJsonObject,
    OutOfMemory,
};

/// A slash command declared by a Lua extension. Name and description
/// point into the runtime's `strings` pool; the handler lives in the Lua
/// registry, addressed by `handler_ref`.
pub const LuaCommand = struct {
    name: []const u8,
    description: []const u8,
    handler_ref: c_int,
};

/// Owned state for the runtime.
pub const LuaRuntime = struct {
    allocator: Allocator,
    L: *c.lua_State,
    /// Tool declarations for the `ToolSource`, owned by this runtime.
    decls: std.array_list.Managed(panto.ToolDecl),
    /// Backing byte buffers for every string referenced by `decls`.
    strings: std.array_list.Managed([]u8),
    /// Map from tool name (borrowed from `decls`) to its handler ref in
    /// the Lua registry (`luaL_ref` index).
    handlers: std.StringHashMap(c_int),

    /// Slash commands declared by extensions via `panto.ext.register_command`.
    /// Owned by this runtime; the CLI reads these to build its command
    /// registry. Each entry's name/description point into `strings`.
    commands: std.array_list.Managed(LuaCommand),
    /// Map from command name (borrowed from `commands`) to its handler
    /// ref in the Lua registry.
    command_handlers: std.StringHashMap(c_int),

    /// Registry ref to the wrapper closure that runs a user handler
    /// inside a `pcall` and reports the result back to Zig via
    /// `panto._record_result`. Allocated once at `create`; reused for
    /// every call. `0` until `installScheduler` runs.
    wrapper_ref: c_int = 0,
    /// Registry ref to `require("luv").update_time`. libuv caches the
    /// loop's notion of "now" and only refreshes it while `uv.run`
    /// executes; the loop is idle between batches (and during a slow
    /// model's streaming), so its clock goes stale. The scheduler calls
    /// this at the start of each batch so handlers that arm timers
    /// (e.g. the shell tool's timeout) compute deadlines against the
    /// real current time, not a frozen one. `0` until
    /// `installScheduler` runs.
    uv_update_time_ref: c_int = 0,

    /// Pointer to the in-flight batch, valid only for the duration of
    /// one `invoke_batch` call. The `panto._record_result` C function
    /// writes through this. `null` between batches; not concurrently
    /// accessible (libpanto's source-grouped dispatch guarantees one
    /// thread per source per turn).
    current_batch: ?*BatchState = null,

    /// The extension UI event bridge (plan §7.6): owns Lua `on` handlers
    /// and Lua-defined components, and drives the App's native `EventBus`.
    /// Heap-allocated so its address is stable (the bridge stores `*self`
    /// pointers in Lua light-userdata upvalues and handler ctx).
    event_bridge: *EventBridge,

    /// Create a new runtime. The `lua_State` is opened, standard libs
    /// loaded, and the `panto.ext.register_tool` bridge installed.
    pub fn create(allocator: Allocator) !*LuaRuntime {
        const self = try allocator.create(LuaRuntime);
        errdefer allocator.destroy(self);

        const L = c.luaL_newstate() orelse return RuntimeError.LuaInitFailed;
        errdefer c.lua_close(L);
        c.luaL_openlibs(L);
        lua_bridge.install(L);

        const eb = try allocator.create(EventBridge);
        errdefer allocator.destroy(eb);
        eb.* = EventBridge.init(allocator, L);

        self.* = .{
            .allocator = allocator,
            .L = L,
            .decls = std.array_list.Managed(panto.ToolDecl).init(allocator),
            .strings = std.array_list.Managed([]u8).init(allocator),
            .handlers = std.StringHashMap(c_int).init(allocator),
            .commands = std.array_list.Managed(LuaCommand).init(allocator),
            .command_handlers = std.StringHashMap(c_int).init(allocator),
            .event_bridge = eb,
        };

        // Install the real `panto.ext.emit` now that the bridge exists, so
        // a Lua `emit(name, data)` reaches the native bus once attached.
        lua_bridge.installEmit(L, @ptrCast(eb), lua_event_bridge.emitThunk);

        // In production, `require('panto')` is wired to the native module
        // by `installPantoModule` (after bootstrap stages `panto.so`).
        // Unit tests have no staged `.so`, and load extension scripts that
        // `require('panto')` for `panto.ext.*`; give them a fabricated
        // `{ ext = ... }` module table so those `require`s resolve.
        if (@import("builtin").is_test) {
            lua_bridge.installTestPantoTable(L);
        }
        return self;
    }

    /// The extension UI event bridge. The embedder wires its `attachBus`
    /// to the App's `EventBus` during startup (after the App exists).
    pub fn eventBridge(self: *LuaRuntime) *EventBridge {
        return self.event_bridge;
    }

    /// Wire `require('panto')` to the native `libpanto-lua` module plus the
    /// CLI's `ext` subtable. Run **after** luarocks bootstrap has staged
    /// `panto.so` and configured `package.cpath`, and **before**
    /// `installScheduler` (whose `_record_result` / wrapper closure live on
    /// the module table).
    ///
    /// The sequence (all in Zig, via the C API, so the CLI never calls
    /// `luaopen_panto` itself — it only speaks `require`):
    ///
    ///   1. `require('panto')` resolves the staged `panto.so` on `cpath`,
    ///      running its `luaopen_panto`, which returns a **fresh** table
    ///      (the agent/stream surface).
    ///   2. Attach the `ext` subtable (built by `lua_bridge.install`) onto
    ///      that fresh native table — mutating the host's own copy, which
    ///      is safe precisely because the table is fresh, not shared.
    ///   3. Install a `package.preload['panto']` loader returning that same
    ///      augmented table, so any *later* `require('panto')` (e.g. from
    ///      an extension) gets `ext` too. `require` already cached the
    ///      table in `package.loaded` from step 1; the preload entry wins
    ///      over `cpath` for any cache miss.
    ///   4. Stash the augmented table under `panto_table_key` so internal
    ///      Zig (`pushPantoTable`, the scheduler) reaches it.
    pub fn installPantoModule(self: *LuaRuntime) !void {
        const L = self.L;

        // 1. require('panto') -> native agent/stream table on the stack.
        _ = c.lua_getglobal(L, "require");
        _ = c.lua_pushlstring(L, "panto", 5);
        if (c.lua_pcallk(L, 1, 1, 0, 0, null) != 0) {
            logTopAsError(L, "panto-lua: require('panto') failed (was panto.so staged on cpath?)");
            c.lua_settop(L, c.lua_gettop(L) - 1);
            return RuntimeError.LuaInitFailed;
        }
        if (c.lua_type(L, -1) != lua_bridge.T_TABLE) {
            c.lua_settop(L, c.lua_gettop(L) - 1);
            return RuntimeError.LuaInitFailed;
        }
        // Stack: ..., panto (native, fresh)

        // 2. panto.ext = <ext table>.
        lua_bridge.pushExtTable(L);
        if (c.lua_type(L, -1) != lua_bridge.T_TABLE) {
            c.lua_settop(L, c.lua_gettop(L) - 2);
            return RuntimeError.LuaInitFailed;
        }
        c.lua_setfield(L, -2, "ext"); // panto.ext = ext; pops ext
        // Stack: ..., panto (augmented)

        // 3. Set package.preload['panto'] to return this augmented table.
        lua_bridge.installPantoPreload(L); // leaves `panto` on the stack

        // 4. Stash in the registry slot for internal Zig access.
        lua_bridge.setPantoTable(L); // dups + stores; leaves `panto`
        c.lua_settop(L, c.lua_gettop(L) - 1); // pop the leftover `panto`
    }

    /// Expose the CLI's session agent to extensions as `panto.ext.agent`:
    /// a *borrowed* native Agent userdata built by the module's
    /// `_wrap_agent`, giving extensions the full agent method surface
    /// (`add_system_message`, `compact`, …) on the live session. Must run
    /// after `installPantoModule`. Borrowing is safe because the CLI
    /// deinits this runtime (closing the Lua state) before the agent.
    pub fn setExtAgent(self: *LuaRuntime, agent: *panto.Agent) !void {
        const L = self.L;
        lua_bridge.pushPantoTable(L);
        if (c.lua_type(L, -1) != lua_bridge.T_TABLE) {
            c.lua_settop(L, c.lua_gettop(L) - 1);
            return RuntimeError.LuaInitFailed;
        }
        _ = c.lua_getfield(L, -1, "_wrap_agent");
        if (c.lua_type(L, -1) != lua_bridge.T_FUNCTION) {
            c.lua_settop(L, c.lua_gettop(L) - 2);
            return RuntimeError.LuaInitFailed;
        }
        c.lua_pushlightuserdata(L, agent);
        if (c.lua_pcallk(L, 1, 1, 0, 0, null) != 0) {
            logTopAsError(L, "panto-lua: _wrap_agent failed");
            c.lua_settop(L, c.lua_gettop(L) - 2);
            return RuntimeError.LuaInitFailed;
        }
        // Stack: ..., panto, agent_ud
        lua_bridge.pushExtTable(L);
        c.lua_pushvalue(L, -2); // copy agent_ud above ext
        c.lua_setfield(L, -2, "agent"); // ext.agent = agent_ud
        c.lua_settop(L, c.lua_gettop(L) - 3); // pop ext, agent_ud, panto
    }

    /// Install the libuv-driven coroutine scheduler:
    ///   - Register `panto._record_result` (C function with `self` as
    ///     light-userdata upvalue) so the wrapper closure can hand
    ///     results back to Zig.
    ///   - Create the wrapper closure that runs a user handler in
    ///     `pcall` and reports the result.
    ///   - Cache `require("luv").update_time` for the per-batch clock
    ///     refresh.
    ///
    /// Must be called after luarocks bootstrap has installed luv,
    /// otherwise the `require("luv")` step will fail. Also after
    /// `installPantoModule`, since the scheduler writes onto the module
    /// table.
    pub fn installScheduler(self: *LuaRuntime) !void {
        try installRecordResult(self);
        try installWrapperClosure(self);
        try cacheUvUpdateTime(self);
    }

    /// Tear down the runtime: free every owned string, unref every
    /// handler, close the Lua state.
    pub fn deinit(self: *LuaRuntime) void {
        // Unref handlers so future GCs collect them. Not strictly
        // necessary since we close the state next, but it documents
        // intent.
        var hit = self.handlers.iterator();
        while (hit.next()) |entry| {
            c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, entry.value_ptr.*);
        }
        self.handlers.deinit();
        var cit = self.command_handlers.iterator();
        while (cit.next()) |entry| {
            c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, entry.value_ptr.*);
        }
        self.command_handlers.deinit();
        if (self.wrapper_ref != 0) {
            c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, self.wrapper_ref);
        }
        if (self.uv_update_time_ref != 0) {
            c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, self.uv_update_time_ref);
        }

        // Free the UI event bridge (Lua handler/component refs + caches)
        // BEFORE closing the state, since it unrefs registry entries.
        self.event_bridge.deinit();
        self.allocator.destroy(self.event_bridge);

        c.lua_close(self.L);

        self.decls.deinit();
        self.commands.deinit();
        for (self.strings.items) |s| self.allocator.free(s);
        self.strings.deinit();
        self.allocator.destroy(self);
    }

    /// A declared but not-yet-activated extension entry: its `name` (the
    /// identity that the availability policy gates on) and a Lua registry
    /// ref to either its `activate` function or — for the sugar tool form —
    /// the tool table itself. Produced by `evalEntries`; consumed by exactly
    /// one of `activateEntry` / `dropEntry`, both of which unref the ref and
    /// free `name`.
    pub const Entry = struct {
        name: []u8,
        ref: c_int,
        is_tool_table: bool,
    };

    /// Evaluate one Lua source file and append the entries it declares to
    /// `out`, WITHOUT activating them — no tools/commands are registered yet.
    ///
    /// The chunk must be side-effect-free (this runs over *every* candidate,
    /// including ones that will be shadowed or denied) and return either:
    ///   - an entry table `{ name = <string>, activate = <function> }`,
    ///   - the sugar tool form `{ name, description, schema, handler }`
    ///     (no `activate`; activation registers it as a tool), or
    ///   - a list `{ <entry|tool>, <entry|tool>, ... }` of the above.
    ///
    /// `package_root`, if provided, is prepended to `package.path`.
    pub fn evalEntries(
        self: *LuaRuntime,
        script_path: []const u8,
        package_root: ?[]const u8,
        out: *std.array_list.Managed(Entry),
    ) !void {
        const L = self.L;
        const path_z = try self.allocator.dupeZ(u8, script_path);
        defer self.allocator.free(path_z);

        if (package_root) |root| {
            const root_z = try self.allocator.dupeZ(u8, root);
            defer self.allocator.free(root_z);
            try prependPackagePath(L, root_z);
        }

        if (c.luaL_loadfilex(L, path_z.ptr, null) != 0) {
            logTopAsError(L, "lua: failed to load extension");
            c.lua_settop(L, c.lua_gettop(L) - 1);
            return error.LuaLoadFailed;
        }
        // The chunk is user Lua: run it through the shared entrypoint
        // machinery like everything else (eval must still be
        // side-effect-free per the extension contract, but it gets the
        // same execution environment).
        try lua_bridge.runEntrypointResult(L, 0, "extension eval", null, null);
        // Stack: ..., returned_value
        defer c.lua_settop(L, c.lua_gettop(L) - 1); // pop returned value
        try self.collectEntriesFromTop(script_path, out);
    }

    /// Evaluate an installed rock (or any `require`-able module) and append
    /// the entries it declares to `out`. Same contract as `evalEntries`: the
    /// module's chunk must be side-effect-free and return an entry, sugar
    /// tool, or list. Used to load `extensions.rocks`.
    pub fn evalEntriesFromModule(
        self: *LuaRuntime,
        module_name: []const u8,
        out: *std.array_list.Managed(Entry),
    ) !void {
        const L = self.L;
        // `require` runs the rock's chunk — user Lua — so it goes through
        // the shared entrypoint machinery like every other entrypoint.
        _ = c.lua_getglobal(L, "require");
        _ = c.lua_pushlstring(L, module_name.ptr, module_name.len);
        try lua_bridge.runEntrypointResult(L, 1, "rock eval", null, null);
        // Stack: ..., returned_value
        defer c.lua_settop(L, c.lua_gettop(L) - 1);
        try self.collectEntriesFromTop(module_name, out);
    }

    /// Classify the value at the top of the stack (does not pop it) as an
    /// entry, a sugar tool, or a list of those, appending each to `out`.
    fn collectEntriesFromTop(
        self: *LuaRuntime,
        what: []const u8,
        out: *std.array_list.Managed(Entry),
    ) !void {
        const L = self.L;
        if (c.lua_type(L, -1) != lua_bridge.T_TABLE) {
            std.log.err("lua: extension '{s}' must return an entry table or list", .{what});
            return error.BadRegistration;
        }

        // A single entry/tool has a string `name`; otherwise it's a list.
        _ = c.lua_getfield(L, -1, "name");
        const has_name = c.lua_type(L, -1) == lua_bridge.T_STRING;
        c.lua_settop(L, c.lua_gettop(L) - 1);

        if (has_name) {
            try self.readOneEntry(what, out);
            return;
        }

        const n: usize = @intCast(c.lua_rawlen(L, -1));
        if (n == 0) {
            std.log.err("lua: extension '{s}' returned a table with no `name` and no entries", .{what});
            return error.BadRegistration;
        }
        var i: usize = 1;
        while (i <= n) : (i += 1) {
            _ = c.lua_rawgeti(L, -1, @intCast(i)); // push element
            if (c.lua_type(L, -1) != lua_bridge.T_TABLE) {
                c.lua_settop(L, c.lua_gettop(L) - 1);
                std.log.err("lua: extension '{s}' entry {d} is not a table", .{ what, i });
                return error.BadRegistration;
            }
            try self.readOneEntry(what, out);
            c.lua_settop(L, c.lua_gettop(L) - 1); // pop element
        }
    }

    /// Read one entry table sitting at the top of the stack (does not pop
    /// it), ref its activator, and append an `Entry` to `out`.
    fn readOneEntry(
        self: *LuaRuntime,
        script_path: []const u8,
        out: *std.array_list.Managed(Entry),
    ) !void {
        const L = self.L;
        const t = c.lua_gettop(L); // absolute index of the entry table

        _ = c.lua_getfield(L, t, "name");
        if (c.lua_type(L, -1) != lua_bridge.T_STRING) {
            c.lua_settop(L, c.lua_gettop(L) - 1);
            std.log.err("lua: extension '{s}' entry has no string `name`", .{script_path});
            return error.BadRegistration;
        }
        var nlen: usize = 0;
        const nptr = c.lua_tolstring(L, -1, &nlen);
        const name = try self.allocator.dupe(u8, nptr[0..nlen]);
        errdefer self.allocator.free(name);
        c.lua_settop(L, c.lua_gettop(L) - 1); // pop name

        var ref: c_int = undefined;
        var is_tool_table = false;

        _ = c.lua_getfield(L, t, "activate");
        if (c.lua_type(L, -1) == lua_bridge.T_FUNCTION) {
            ref = c.luaL_ref(L, lua_bridge.LUA_REGISTRYINDEX); // refs + pops the fn
        } else {
            c.lua_settop(L, c.lua_gettop(L) - 1); // pop non-function activate
            _ = c.lua_getfield(L, t, "handler");
            const is_handler = c.lua_type(L, -1) == lua_bridge.T_FUNCTION;
            c.lua_settop(L, c.lua_gettop(L) - 1); // pop handler probe
            if (!is_handler) {
                std.log.err("lua: extension '{s}' entry '{s}' has neither `activate` nor `handler`", .{ script_path, name });
                return error.BadRegistration;
            }
            // Sugar tool form: ref the table itself; activation registers it.
            c.lua_pushvalue(L, t); // copy of the tool table
            ref = c.luaL_ref(L, lua_bridge.LUA_REGISTRYINDEX); // refs + pops copy
            is_tool_table = true;
        }

        try out.append(.{ .name = name, .ref = ref, .is_tool_table = is_tool_table });
    }

    /// Activate an entry: run its `activate` function (or, for the sugar
    /// form, `register_tool(<tool table>)`), then harvest the tools,
    /// commands, and event handlers it registered. Consumes the entry
    /// (unrefs its ref, frees its name).
    pub fn activateEntry(self: *LuaRuntime, entry: Entry) !void {
        const L = self.L;
        // Isolate this entry's registrations so harvest picks up only its own.
        lua_bridge.resetRegistrations(L);

        if (entry.is_tool_table) {
            lua_bridge.pushExtTable(L);
            _ = c.lua_getfield(L, -1, "register_tool");
            c.lua_copy(L, -1, -2); // overwrite `ext` with `register_tool`
            c.lua_settop(L, c.lua_gettop(L) - 1); // Stack: register_tool
            _ = c.lua_rawgeti(L, lua_bridge.LUA_REGISTRYINDEX, entry.ref); // push tool table
            if (c.lua_pcallk(L, 1, 0, 0, 0, null) != 0) {
                logTopAsError(L, "lua: register_tool failed for tool");
                c.lua_settop(L, c.lua_gettop(L) - 1);
                self.finishEntry(entry);
                return error.LuaRunFailed;
            }
        } else {
            // Run activate() as a user-Lua entrypoint: inside a coroutine
            // with the uv loop driven to completion, so activation code may
            // await libuv work like any other user Lua.
            _ = c.lua_rawgeti(L, lua_bridge.LUA_REGISTRYINDEX, entry.ref); // push activate fn
            lua_bridge.runEntrypoint(L, 0, "extension activate()", null, null) catch {
                self.finishEntry(entry);
                return error.LuaRunFailed;
            };
        }

        self.finishEntry(entry);
        try self.harvestAndStoreHandlers();
    }

    /// Discard an entry without activating it (shadowed or denied).
    pub fn dropEntry(self: *LuaRuntime, entry: Entry) void {
        self.finishEntry(entry);
    }

    fn finishEntry(self: *LuaRuntime, entry: Entry) void {
        c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, entry.ref);
        self.allocator.free(entry.name);
    }

    /// Walk the registrations table that the script just populated.
    /// For each entry:
    ///   - Copy `name`, `description`, `schema_json` into owned bytes.
    ///   - Pop the `handler` function and `luaL_ref` it into the
    ///     registry; record the ref under `handlers[name]`.
    ///   - Append a `ToolDecl` to `self.decls`.
    fn harvestAndStoreHandlers(self: *LuaRuntime) !void {
        const L = self.L;

        // Push the registrations table onto the stack.
        _ = c.lua_rawgetp(L, lua_bridge.LUA_REGISTRYINDEX, &lua_bridge.registrations_key);
        defer c.lua_settop(L, c.lua_gettop(L) - 1);

        const n: usize = @intCast(c.lua_rawlen(L, -1));
        var i: usize = 1;
        while (i <= n) : (i += 1) {
            _ = c.lua_rawgeti(L, -1, @intCast(i)); // push record
            // record at -1; bridge's records are 4-field tables.

            const name = try self.readStringFieldOwned("name");
            errdefer {
                // If anything below fails after the name was added to
                // strings, the global deinit still cleans up; nothing
                // extra to undo here for the string itself. But we
                // *do* need to make sure the handlers map and decls
                // remain consistent. We allocate after the string adds,
                // so partial state is "string captured but no decl"
                // — harmless.
            }
            const desc = try self.readStringFieldOwned("description");
            const schema = try self.readStringFieldOwned("schema_json");

            // Pop handler function -> luaL_ref into the registry.
            _ = c.lua_getfield(L, -1, "handler");
            if (c.lua_type(L, -1) != lua_bridge.T_FUNCTION) {
                c.lua_settop(L, c.lua_gettop(L) - 2); // pop handler + record
                return RuntimeError.LuaHandlerNotFound;
            }
            const ref = c.luaL_ref(L, lua_bridge.LUA_REGISTRYINDEX);
            // Stack: ..., regs_table, record

            const decl: panto.ToolDecl = .{
                .name = name,
                .description = desc,
                .schema_json = schema,
            };

            // Duplicate names within the runtime are not allowed —
            // libpanto will also catch them at registry insertion, but
            // we want a Lua-side error before we've started talking to
            // libpanto.
            const gop = try self.handlers.getOrPut(name);
            if (gop.found_existing) {
                c.luaL_unref(L, lua_bridge.LUA_REGISTRYINDEX, ref);
                c.lua_settop(L, c.lua_gettop(L) - 1); // pop record
                return error.DuplicateTool;
            }
            gop.value_ptr.* = ref;

            try self.decls.append(decl);

            c.lua_settop(L, c.lua_gettop(L) - 1); // pop record
        }

        try self.harvestAndStoreCommands();
        try self.harvestAndStoreOnHandlers();
    }

    /// Walk the `panto.ext.on` registrations table populated by the script
    /// just loaded. For each `{ event, handler }` record, hand the handler
    /// function and event name to the `EventBridge`, which `luaL_ref`s the
    /// function and remembers the subscription (in registration order).
    /// Actual registration into the App's `EventBus` happens later via
    /// `EventBridge.attachBus` (the bus may not exist at load time).
    fn harvestAndStoreOnHandlers(self: *LuaRuntime) !void {
        const L = self.L;

        _ = c.lua_rawgetp(L, lua_bridge.LUA_REGISTRYINDEX, &lua_bridge.on_registrations_key);
        defer c.lua_settop(L, c.lua_gettop(L) - 1);

        const n: usize = @intCast(c.lua_rawlen(L, -1));
        var i: usize = 1;
        while (i <= n) : (i += 1) {
            _ = c.lua_rawgeti(L, -1, @intCast(i)); // push record
            defer c.lua_settop(L, c.lua_gettop(L) - 1); // pop record

            // Read the event name (borrowed; the bridge dupes it).
            _ = c.lua_getfield(L, -1, "event");
            var elen: usize = 0;
            const eptr = c.lua_tolstring(L, -1, &elen);
            const event_name = if (eptr != null) eptr[0..elen] else "";
            c.lua_settop(L, c.lua_gettop(L) - 1); // pop event string

            // Push the handler function and hand it to the bridge (which
            // pushes a copy + luaL_refs it).
            _ = c.lua_getfield(L, -1, "handler");
            if (c.lua_type(L, -1) != lua_bridge.T_FUNCTION) {
                c.lua_settop(L, c.lua_gettop(L) - 1); // pop non-function
                return RuntimeError.LuaHandlerNotFound;
            }
            try self.event_bridge.registerOnHandler(event_name, -1);
            c.lua_settop(L, c.lua_gettop(L) - 1); // pop handler
        }
    }

    /// Walk the command registrations table populated by the script just
    /// loaded. For each `{ name, description, handler }` record: copy the
    /// strings, `luaL_ref` the handler, and append a `LuaCommand`.
    /// Duplicate command names within the runtime are rejected.
    fn harvestAndStoreCommands(self: *LuaRuntime) !void {
        const L = self.L;

        _ = c.lua_rawgetp(L, lua_bridge.LUA_REGISTRYINDEX, &lua_bridge.command_registrations_key);
        defer c.lua_settop(L, c.lua_gettop(L) - 1);

        const n: usize = @intCast(c.lua_rawlen(L, -1));
        var i: usize = 1;
        while (i <= n) : (i += 1) {
            _ = c.lua_rawgeti(L, -1, @intCast(i)); // push record

            const name = try self.readStringFieldOwned("name");
            const desc = try self.readStringFieldOwned("description");

            _ = c.lua_getfield(L, -1, "handler");
            if (c.lua_type(L, -1) != lua_bridge.T_FUNCTION) {
                c.lua_settop(L, c.lua_gettop(L) - 2); // pop handler + record
                return RuntimeError.LuaHandlerNotFound;
            }
            const ref = c.luaL_ref(L, lua_bridge.LUA_REGISTRYINDEX);
            // Stack: ..., regs_table, record

            const gop = try self.command_handlers.getOrPut(name);
            if (gop.found_existing) {
                c.luaL_unref(L, lua_bridge.LUA_REGISTRYINDEX, ref);
                c.lua_settop(L, c.lua_gettop(L) - 1); // pop record
                return error.DuplicateCommand;
            }
            gop.value_ptr.* = ref;

            try self.commands.append(.{
                .name = name,
                .description = desc,
                .handler_ref = ref,
            });

            c.lua_settop(L, c.lua_gettop(L) - 1); // pop record
        }
    }

    /// Run a Lua slash-command handler. `handler_ref` is a registry ref
    /// to a `function(args)` closure; `args` is the trimmed remainder of
    /// the command line. Like every user-Lua entrypoint it runs inside a
    /// coroutine with the uv loop driven to completion, so command
    /// handlers may await libuv work; the call is synchronous from the
    /// REPL's point of view.
    ///
    /// Slash-command handlers act by side effect (printing, mutating
    /// state) and their return value is ignored. On Lua error, returns
    /// `RuntimeError.LuaHandlerCrashed` after copying the error message
    /// (with traceback) into `err_buf` (truncated to fit); `*err_len` is
    /// set to the number of bytes written.
    pub fn runCommand(
        self: *LuaRuntime,
        handler_ref: c_int,
        args: []const u8,
        err_buf: []u8,
        err_len: *usize,
    ) RuntimeError!void {
        const L = self.L;
        err_len.* = 0;

        _ = c.lua_rawgeti(L, lua_bridge.LUA_REGISTRYINDEX, @intCast(handler_ref));
        _ = c.lua_pushlstring(L, args.ptr, args.len);
        lua_bridge.runEntrypoint(L, 1, "slash command", err_buf, err_len) catch
            return RuntimeError.LuaHandlerCrashed;
    }

    /// Slash commands declared by loaded extensions, in registration
    /// order. The CLI walks these to populate its command registry.
    pub fn commandList(self: *const LuaRuntime) []const LuaCommand {
        return self.commands.items;
    }

    fn readStringFieldOwned(self: *LuaRuntime, field_name: [:0]const u8) ![]const u8 {
        const L = self.L;
        _ = c.lua_getfield(L, -1, field_name.ptr);
        defer c.lua_settop(L, c.lua_gettop(L) - 1);
        if (c.lua_type(L, -1) != lua_bridge.T_STRING) return error.BadRegistration;
        var len: usize = 0;
        const ptr = c.lua_tolstring(L, -1, &len);
        if (ptr == null) return error.BadRegistration;
        const owned = try self.allocator.dupe(u8, ptr[0..len]);
        try self.strings.append(owned);
        return owned;
    }

    /// Build a `ToolSource` that hands `invoke_batch` calls back to
    /// this runtime. The source's `ctx` is `self`. The runtime keeps
    /// ownership of `self`'s allocation; libpanto's registry only
    /// frees `ctx` via the source's `vtable.deinit` (which we make a
    /// no-op — the runtime is owned by the embedder).
    ///
    /// Callers must keep the LuaRuntime alive at least as long as the
    /// registry holds the source.
    pub fn toolSource(self: *LuaRuntime) panto.ToolSource {
        return .{
            .name = SOURCE_NAME,
            .tools = self.decls.items,
            .ctx = self,
            .vtable = &source_vtable,
        };
    }

    /// Number of tools currently declared by extensions loaded into
    /// this runtime.
    pub fn toolCount(self: *const LuaRuntime) usize {
        return self.decls.items.len;
    }

    /// Drop every declared tool whose registered name is rejected by
    /// `permits`. The handler ref is unref'd and the decl removed.
    /// Returns the number of tools dropped.
    ///
    /// String storage for a dropped tool's name/description/schema is
    /// left in `self.strings` (freed at `deinit`); only the decl entry
    /// and the Lua handler ref are reclaimed eagerly. This keeps the
    /// filter simple — we never need to find-and-free individual
    /// strings out of the shared pool.
    pub fn filterTools(
        self: *LuaRuntime,
        ctx: anytype,
        comptime permits: fn (@TypeOf(ctx), []const u8) bool,
    ) usize {
        var dropped: usize = 0;
        var i: usize = 0;
        while (i < self.decls.items.len) {
            const name = self.decls.items[i].name;
            if (permits(ctx, name)) {
                i += 1;
                continue;
            }
            // Unref the handler, if present.
            if (self.handlers.fetchRemove(name)) |kv| {
                c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, kv.value);
            }
            _ = self.decls.orderedRemove(i);
            dropped += 1;
        }
        return dropped;
    }
};

const source_vtable: panto.ToolSource.VTable = .{
    .invoke_batch = invokeBatch,
    .deinit = deinitSrc,
};

fn deinitSrc(_: *anyopaque, _: Allocator) void {
    // The runtime is owned by the embedder (main()). It explicitly
    // calls `runtime.deinit()` after the agent has been torn down.
    // libpanto's source.deinit here is a no-op.
}

// ===========================================================================
// Scheduler: libuv-driven cooperative coroutine dispatch
// ===========================================================================
//
// libpanto's `invoke_batch` delivers all of a turn's tool-call requests
// at once, on a single thread. We answer the contract by running each
// call as a Lua coroutine inside our long-lived `lua_State`, then
// driving `uv.run("once")` to wake any of those coroutines that are
// blocked on libuv-aware I/O. This is the entire scheduler — luv's
// libuv binding does the actual event-loop work; we just resume
// coroutines and call `run` between resumes.
//
// Capturing return values requires a wrapper. When a coroutine is
// resumed by a luv callback after yielding, the eventual return value
// of the coroutine flows back to *that callback*, not to us. So we
// install a Lua wrapper closure that does
//
//     pcall(handler, input) → panto._record_result(idx, ok, val)
//
// before the handler returns. `_record_result` is a C function that
// stores into a per-runtime `BatchState`, accessed via a light-userdata
// upvalue carrying the runtime pointer.

/// One coroutine's outcome, recorded by `_record_result` and read by
/// `invokeBatch` once the coroutine has terminated.
const Slot = struct {
    /// Set true the moment `_record_result` writes a result for this
    /// index. Used to detect coroutines that terminated without
    /// calling the wrapper (a bug / API misuse).
    recorded: bool = false,
    /// `true` if the handler returned cleanly, `false` if it raised
    /// via the `pcall` wrapping.
    ok: bool = false,
    /// Result payload as owned parts. Allocated from `allocator`.
    /// Caller frees via `panto.ResultParts.deinit`.
    value: ?panto.ResultParts = null,
    /// On `ok = false`, an owned copy of the error message.
    err_msg: ?[]u8 = null,
};

/// State shared between Zig and the in-flight Lua wrapper closure.
const BatchState = struct {
    allocator: Allocator,
    slots: []Slot,
};

fn invokeBatch(
    ctx: *anyopaque,
    calls: []const panto.ToolCall,
    results: []panto.ToolCallResult,
    allocator: Allocator,
) anyerror!void {
    const self: *LuaRuntime = @ptrCast(@alignCast(ctx));
    return runBatch(self, calls, results, allocator);
}

fn runBatch(
    self: *LuaRuntime,
    calls: []const panto.ToolCall,
    results: []panto.ToolCallResult,
    allocator: Allocator,
) !void {
    if (self.wrapper_ref == 0) {
        // Scheduler not installed. We can still run synchronous
        // handlers — use the legacy path that drives one coroutine at
        // a time without an event loop.
        for (calls, 0..) |call, i| {
            results[i] = runLegacySync(self, call, allocator);
        }
        return;
    }

    // Refresh libuv's cached loop clock before any handler arms a timer.
    // The loop sits idle between batches (and while a slow model streams),
    // so its notion of "now" is stale; a timeout armed against it would
    // otherwise fire the instant `uv.run` refreshes time below.
    refreshLoopClock(self);

    var slots = try allocator.alloc(Slot, calls.len);
    defer allocator.free(slots);
    for (slots) |*s| s.* = .{};

    var batch_state: BatchState = .{ .allocator = allocator, .slots = slots };
    self.current_batch = &batch_state;
    defer self.current_batch = null;

    // Track each call's coroutine reference in the parent stack (we
    // hold them in registry refs so they survive across `uv.run`
    // ticks). `0` after a coroutine has been reaped.
    var thread_refs = try allocator.alloc(c_int, calls.len);
    defer allocator.free(thread_refs);
    @memset(thread_refs, 0);
    defer for (thread_refs) |r| {
        if (r != 0) c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, r);
    };

    // Step 1: start every call's coroutine. A *synchronous* handler
    // (one that never yields) runs to completion right here and
    // records its result via the wrapper; `still_pending` is false.
    // An *async* handler arms one or more libuv operations and yields
    // — that armed work is what keeps the event loop alive in step 2.
    var any_pending = false;
    for (calls, 0..) |call, i| {
        const handler_ref = self.handlers.get(call.tool_name) orelse {
            // Synthesize a recorded "err" result; don't even bother
            // spawning a coroutine.
            slots[i] = .{
                .recorded = true,
                .ok = false,
                .err_msg = try allocator.dupe(u8, "panto: unknown tool name"),
            };
            continue;
        };
        const t = try startCoroutine(self, i, handler_ref, call.input, allocator);
        thread_refs[i] = t.thread_ref;
        if (t.still_pending) any_pending = true;
    }

    // Step 2: drive the event loop to completion. The contract for
    // async handlers is that the *only* way a handler coroutine may
    // block is by yielding on a pending libuv operation whose callback
    // will eventually resume it. Under that invariant, `uv.run`
    // returns exactly when no active handles remain — which means
    // every coroutine has either finished or violated the contract
    // (yielded without keeping libuv work alive). Either way there is
    // nothing left to wake, so a single run-to-completion pass is all
    // we need; the per-tick reap/`loop_alive` dance the scheduler used
    // to do is unnecessary.
    if (any_pending) {
        try driveUvToCompletion(self);
    }

    // Step 3: reap. Every coroutine should now be terminated. Any that
    // is still suspended (`LUA_YIELD`) violated the contract — it
    // yielded without arming libuv work that would resume it (or armed
    // work, then tore it down without resolving). Surface that as a
    // per-call error rather than hanging the process.
    for (thread_refs, 0..) |tref, i| {
        if (tref == 0) continue;
        _ = c.lua_rawgeti(self.L, lua_bridge.LUA_REGISTRYINDEX, tref);
        const co: *c.lua_State = @ptrCast(c.lua_tothread(self.L, -1).?);
        const status = c.lua_status(co);
        c.lua_settop(self.L, c.lua_gettop(self.L) - 1);
        if (status == c.LUA_YIELD) {
            slots[i] = .{
                .recorded = true,
                .ok = false,
                .err_msg = try allocator.dupe(
                    u8,
                    "panto: handler is still suspended after the event loop " ++
                        "drained; it yielded without a pending libuv operation " ++
                        "to resume it (did a uv callback fail to resume the " ++
                        "coroutine, or was its handle closed without resolving?)",
                ),
            };
        } else if (!slots[i].recorded) {
            // Terminated cleanly or with an error, but the wrapper
            // never recorded — should not happen, but don't drop it.
            slots[i] = .{
                .recorded = true,
                .ok = false,
                .err_msg = try allocator.dupe(
                    u8,
                    "panto: handler terminated without recording a result",
                ),
            };
        }
        c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, tref);
        thread_refs[i] = 0;
    }

    // Translate slots into the libpanto-shaped results.
    //
    // Important: a handler that raised (`ok == false`) or otherwise
    // misbehaved (`!recorded`) is surfaced to the model as an `.ok`
    // result whose body is the formatted error message. We do *not*
    // return `.err` here, because libpanto treats any per-call `.err`
    // as an unrecoverable failure that aborts the entire turn (see
    // `agent.dispatchToolCalls`). Aborting the turn over a Lua-level
    // bug — in either a builtin tool or a user extension — is far
    // more disruptive than handing the model a readable error and
    // letting it correct course on the next turn.
    for (slots, 0..) |slot, i| {
        if (!slot.recorded) {
            results[i] = .{
                .ok = try formatToolError(
                    allocator,
                    calls[i].tool_name,
                    "handler terminated without recording a result",
                ),
            };
            continue;
        }
        if (slot.ok) {
            results[i] = .{ .ok = slot.value orelse try panto.ResultParts.fromText(allocator, "") };
            // Free the err_msg if both ended up set somehow.
            if (slot.err_msg) |m| allocator.free(m);
        } else {
            if (slot.value) |v| v.deinit(allocator);
            std.log.warn(
                "panto-lua: tool '{s}' failed: {s}",
                .{
                    calls[i].tool_name,
                    slot.err_msg orelse "(no message)",
                },
            );
            results[i] = .{
                .ok = try formatToolError(
                    allocator,
                    calls[i].tool_name,
                    slot.err_msg orelse "(no message)",
                ),
            };
            if (slot.err_msg) |m| allocator.free(m);
        }
    }
}

/// Format a tool-level failure as a textual result the model can read.
/// The prefix mirrors what the user sees in `panto-lua` log lines so
/// the model and the developer are looking at the same string.
fn formatToolError(
    allocator: Allocator,
    tool_name: []const u8,
    message: []const u8,
) !panto.ResultParts {
    const text = try std.fmt.allocPrint(
        allocator,
        "panto-lua: tool '{s}' failed: {s}",
        .{ tool_name, message },
    );
    return panto.ResultParts.fromTextOwned(allocator, text);
}

/// Start one coroutine: create a thread under the runtime's lua_State,
/// push the wrapper closure + (idx, handler, input), `lua_resume` once.
///
/// If the coroutine returns immediately (sync handler), the wrapper
/// has already recorded its result via `panto._record_result` —
/// `still_pending` will be `false`.
fn startCoroutine(
    self: *LuaRuntime,
    idx: usize,
    handler_ref: c_int,
    input: []const u8,
    allocator: Allocator,
) !struct { thread_ref: c_int, still_pending: bool } {
    const L = self.L;

    const co = c.lua_newthread(L) orelse return RuntimeError.LuaInitFailed;
    // luaL_ref pops the topmost value (the thread) and returns a
    // registry ref to it. We keep the ref alive for the lifetime of
    // the call so GC doesn't collect the thread mid-yield.
    const thread_ref = c.luaL_ref(L, lua_bridge.LUA_REGISTRYINDEX);

    // Push the wrapper onto the coroutine's stack.
    _ = c.lua_rawgeti(co, lua_bridge.LUA_REGISTRYINDEX, @intCast(self.wrapper_ref));
    // Push (idx, handler, input) as the resume args.
    c.lua_pushinteger(co, @intCast(idx));
    _ = c.lua_rawgeti(co, lua_bridge.LUA_REGISTRYINDEX, @intCast(handler_ref));
    var arena_state = std.heap.ArenaAllocator.init(allocator);
    defer arena_state.deinit();
    try lua_bridge.pushJsonAsLua(co, arena_state.allocator(), input);

    var nres: c_int = 0;
    const status = c.lua_resume(co, L, 3, &nres);
    return .{
        .thread_ref = thread_ref,
        .still_pending = status == c.LUA_YIELD,
    };
}

/// Drain the uv loop (`lua_bridge.drainLoop` — the same drain every
/// single-fn entrypoint uses). Under the async-tool contract (a handler
/// may only block by yielding on a pending libuv operation whose callback
/// resumes it), this returns exactly when every handler coroutine has
/// finished.
fn driveUvToCompletion(self: *LuaRuntime) !void {
    const L = self.L;
    lua_bridge.drainLoop(L) catch {
        logTopAsError(L, "panto-lua: uv.run failed");
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return error.UvRunFailed;
    };
}

/// Pre-scheduler fallback (used in unit tests and during early
/// startup before `installScheduler` has run).
fn runLegacySync(
    self: *LuaRuntime,
    call: panto.ToolCall,
    allocator: Allocator,
) panto.ToolCallResult {
    const handler_ref = self.handlers.get(call.tool_name) orelse {
        const bytes = formatToolError(
            allocator,
            call.tool_name,
            "unknown tool name",
        ) catch return .{ .err = error.OutOfMemory };
        return .{ .ok = bytes };
    };
    var err_msg: ?[]u8 = null;
    defer if (err_msg) |m| allocator.free(m);
    const out_parts = invokeCoroutineSync(
        self.L,
        handler_ref,
        call.input,
        allocator,
        &err_msg,
    ) catch |e| {
        // Surface the failure to the model as a textual `.ok` result
        // rather than aborting the turn. Prefer the Lua-side error
        // message (captured into `err_msg` by invokeCoroutineSync
        // when available); fall back to the typed error name.
        const message: []const u8 = err_msg orelse @errorName(e);
        const bytes = formatToolError(
            allocator,
            call.tool_name,
            message,
        ) catch return .{ .err = error.OutOfMemory };
        return .{ .ok = bytes };
    };
    return .{ .ok = out_parts };
}

/// Run a handler through the shared entrypoint machinery (coroutine +
/// loop drain), keeping its return value. On Lua-level failure the error
/// message is duped into `*err_msg_out` (caller owns) before returning
/// the typed error. `err_msg_out` is only written on the error path; on
/// success it is left untouched.
fn invokeCoroutineSync(
    L: *c.lua_State,
    handler_ref: c_int,
    input: []const u8,
    allocator: Allocator,
    err_msg_out: *?[]u8,
) !panto.ResultParts {
    _ = c.lua_rawgeti(L, lua_bridge.LUA_REGISTRYINDEX, @intCast(handler_ref));
    if (c.lua_type(L, -1) != lua_bridge.T_FUNCTION) {
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return RuntimeError.LuaHandlerNotFound;
    }

    var arena_state = std.heap.ArenaAllocator.init(allocator);
    defer arena_state.deinit();
    try lua_bridge.pushJsonAsLua(L, arena_state.allocator(), input);

    var err_buf: [1024]u8 = undefined;
    var err_len: usize = 0;
    lua_bridge.runEntrypointResult(L, 1, "tool handler", &err_buf, &err_len) catch {
        err_msg_out.* = allocator.dupe(u8, err_buf[0..err_len]) catch null;
        return RuntimeError.LuaHandlerCrashed;
    };
    // Success: the handler's result is on top of L.
    defer c.lua_settop(L, c.lua_gettop(L) - 1);
    if (c.lua_type(L, -1) == lua_bridge.T_NIL) return RuntimeError.BadHandlerReturn;
    return try lua_bridge.readHandlerResult(L, -1, allocator);
}

// ---------------------------------------------------------------------------
// Scheduler setup (called once at startup, after luarocks bootstrap)
// ---------------------------------------------------------------------------

/// Register `panto._record_result(idx, ok, value)` on the `panto`
/// global. The C function carries the runtime pointer as an upvalue,
/// reaches the in-flight `BatchState` through `current_batch`, and
/// stores into the matching slot.
fn installRecordResult(self: *LuaRuntime) !void {
    const L = self.L;
    lua_bridge.pushPantoTable(L);
    if (c.lua_type(L, -1) != lua_bridge.T_TABLE) {
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return RuntimeError.LuaInitFailed;
    }
    c.lua_pushlightuserdata(L, @ptrCast(self));
    c.lua_pushcclosure(L, recordResultC, 1);
    c.lua_setfield(L, -2, "_record_result");
    c.lua_settop(L, c.lua_gettop(L) - 1); // pop `panto`
}

fn recordResultC(L: ?*c.lua_State) callconv(.c) c_int {
    const Lst = L.?;
    const self_ptr = c.lua_touserdata(Lst, c.lua_upvalueindex(1));
    if (self_ptr == null) return 0;
    const self: *LuaRuntime = @ptrCast(@alignCast(self_ptr.?));
    const batch = self.current_batch orelse return 0;

    const idx_i64 = c.lua_tointegerx(Lst, 1, null);
    const ok = c.lua_toboolean(Lst, 2) != 0;
    const idx: usize = @intCast(idx_i64);
    if (idx >= batch.slots.len) return 0;

    if (ok) {
        // Result value at index 3. The handler's return type is
        // free-form; we serialize via the existing `readHandlerResult`
        // helper which already knows how to JSON-encode any Lua value.
        const value = lua_bridge.readHandlerResult(Lst, 3, batch.allocator) catch |e| {
            // Allocation failure mid-callback is unrecoverable from
            // Lua's POV; record a synthetic error and bail.
            const msg = std.fmt.allocPrint(
                batch.allocator,
                "panto: failed to serialize handler result: {s}",
                .{@errorName(e)},
            ) catch null;
            batch.slots[idx] = .{ .recorded = true, .ok = false, .err_msg = msg };
            return 0;
        };
        batch.slots[idx] = .{ .recorded = true, .ok = true, .value = value };
    } else {
        // Error message at index 3. May be any Lua value; coerce to
        // string via `tostring`-equivalent semantics.
        var len: usize = 0;
        const ptr = c.luaL_tolstring(Lst, 3, &len);
        const owned = if (ptr != null)
            batch.allocator.dupe(u8, ptr[0..len]) catch null
        else
            null;
        c.lua_settop(Lst, c.lua_gettop(Lst) - 1); // pop tolstring's pushed string
        batch.slots[idx] = .{ .recorded = true, .ok = false, .err_msg = owned };
    }
    return 0;
}

/// Create the per-call wrapper closure:
///
///     local function wrapper(idx, handler, input)
///       local ok, val = pcall(handler, input)
///       panto._record_result(idx, ok, val)
///     end
///
/// Stored in the Lua registry under `self.wrapper_ref`.
fn installWrapperClosure(self: *LuaRuntime) !void {
    const L = self.L;
    // `panto` is not a global, so the wrapper closes over the table
    // (passed as an argument) rather than looking it up globally.
    const snippet =
        \\local panto = ...
        \\return function(idx, handler, input)
        \\  local ok, val = pcall(handler, input)
        \\  panto._record_result(idx, ok, val)
        \\end
    ;
    if (c.luaL_loadstring(L, snippet) != 0) {
        logTopAsError(L, "panto-lua: wrapper closure failed to compile");
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return RuntimeError.LuaInitFailed;
    }
    // Pass the `panto` table as the chunk's vararg argument.
    lua_bridge.pushPantoTable(L);
    if (c.lua_pcallk(L, 1, 1, 0, 0, null) != 0) {
        logTopAsError(L, "panto-lua: wrapper closure failed to evaluate");
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return RuntimeError.LuaInitFailed;
    }
    // Top of stack: the wrapper function. luaL_ref pops it.
    self.wrapper_ref = c.luaL_ref(L, lua_bridge.LUA_REGISTRYINDEX);
}

/// Cache `require("luv").run` in the registry so the scheduler can
/// invoke it cheaply without re-resolving `require` each batch.
/// Cache `require("luv").update_time` so the scheduler can refresh the
/// loop clock at the start of each batch. See `uv_update_time_ref`.
fn cacheUvUpdateTime(self: *LuaRuntime) !void {
    const L = self.L;
    const snippet =
        \\local uv = require("luv")
        \\return uv.update_time
    ;
    if (c.luaL_loadstring(L, snippet) != 0) {
        logTopAsError(L, "panto-lua: failed to compile luv.update_time lookup");
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return RuntimeError.LuaInitFailed;
    }
    if (c.lua_pcallk(L, 0, 1, 0, 0, null) != 0) {
        logTopAsError(L, "panto-lua: require('luv') failed (was the bootstrap successful?)");
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return RuntimeError.LuaInitFailed;
    }
    if (c.lua_type(L, -1) != lua_bridge.T_FUNCTION) {
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return RuntimeError.LuaInitFailed;
    }
    self.uv_update_time_ref = c.luaL_ref(L, lua_bridge.LUA_REGISTRYINDEX);
}

/// Refresh libuv's cached loop clock (`uv.update_time()`). Best-effort:
/// a failure here only means timeout deadlines may be computed against a
/// slightly stale clock, so we log and continue rather than failing the
/// batch.
fn refreshLoopClock(self: *LuaRuntime) void {
    if (self.uv_update_time_ref == 0) return;
    const L = self.L;
    _ = c.lua_rawgeti(L, lua_bridge.LUA_REGISTRYINDEX, @intCast(self.uv_update_time_ref));
    if (c.lua_pcallk(L, 0, 0, 0, 0, null) != 0) {
        logTopAsError(L, "panto-lua: uv.update_time failed");
        c.lua_settop(L, c.lua_gettop(L) - 1);
    }
}

// ---------------------------------------------------------------------------
// Internals
// ---------------------------------------------------------------------------

fn prependPackagePath(L: *c.lua_State, root: [:0]const u8) !void {
    const snippet =
        \\local root = ...
        \\package.path = root .. "/?.lua;" .. root .. "/?/init.lua;" .. package.path
    ;
    if (c.luaL_loadstring(L, snippet) != 0) {
        logTopAsError(L, "lua: package.path loader failed to compile");
        return error.LuaPackagePathLoadFailed;
    }
    _ = c.lua_pushlstring(L, root.ptr, root.len);
    if (c.lua_pcallk(L, 1, 0, 0, 0, null) != 0) {
        logTopAsError(L, "lua: package.path setup failed");
        return error.LuaPackagePathSetupFailed;
    }
}

fn logTopAsError(L: *c.lua_State, prefix: []const u8) void {
    var len: usize = 0;
    const msg = c.lua_tolstring(L, -1, &len);
    const is_test = @import("builtin").is_test;
    if (msg != null) {
        if (is_test) {
            std.log.warn("{s}: {s}", .{ prefix, msg[0..len] });
        } else {
            std.log.err("{s}: {s}", .{ prefix, msg[0..len] });
        }
    } else {
        if (is_test) {
            std.log.warn("{s} (no error message)", .{prefix});
        } else {
            std.log.err("{s} (no error message)", .{prefix});
        }
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

const testing = std.testing;

/// Test helper: concatenated text across a result's text parts.
/// Returns the first text part's items (sufficient for current tests,
/// which return a single text part).
fn okText(result: panto.ToolCallResult) []const u8 {
    switch (result) {
        .ok => |parts| {
            for (parts.items) |p| {
                if (p == .text) return p.text;
            }
            return "";
        },
        .err => return "",
    }
}

/// Test helper: free a results slice (parts on `.ok`).
fn freeResults(results: []panto.ToolCallResult) void {
    for (results) |r| switch (r) {
        .ok => |b| b.deinit(testing.allocator),
        .err => {},
    };
}

fn writeTempScript(dir: Io.Dir, name: []const u8, source: []const u8) ![]const u8 {
    // `panto` is not a global; extensions reach it via `require('panto')`.
    // The register-based `source` body is wrapped in an entry whose
    // `activate()` runs it, matching the new deferred-registration model.
    var src_buf: [16384]u8 = undefined;
    const data = try std.fmt.bufPrint(
        &src_buf,
        "local panto = require(\"panto\")\nreturn {{ name = \"_t\", activate = function()\n{s}\nend }}\n",
        .{source},
    );
    try dir.writeFile(testing.io, .{ .sub_path = name, .data = data });
    var buf: [std.fs.max_path_bytes]u8 = undefined;
    const n = try dir.realPathFile(testing.io, name, &buf);
    return testing.allocator.dupe(u8, buf[0..n]);
}

/// Test helper: eval one script and activate every entry it declares
/// (no policy, no shadowing) — the single-file analogue of the loader.
fn loadScript(rt: *LuaRuntime, path: []const u8, package_root: ?[]const u8) !void {
    var entries: std.array_list.Managed(LuaRuntime.Entry) = .init(rt.allocator);
    defer entries.deinit();
    rt.evalEntries(path, package_root, &entries) catch |e| {
        for (entries.items) |it| rt.dropEntry(it);
        return e;
    };
    for (entries.items, 0..) |it, idx| {
        rt.activateEntry(it) catch |e| {
            var j = idx + 1;
            while (j < entries.items.len) : (j += 1) rt.dropEntry(entries.items[j]);
            return e;
        };
    }
}

test "evalEntriesFromModule loads a rock's entries via require" {
    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();

    // Stand in for an installed rock: preload a module returning entries.
    const preload =
        \\package.preload["fakerock"] = function()
        \\  local panto = require("panto")
        \\  return {
        \\    { name = "rock.a", activate = function()
        \\        panto.ext.register_tool { name = "rock.a", description = "a",
        \\          schema = { type = "object" }, handler = function() return "A" end }
        \\      end },
        \\    { name = "rock.b", description = "b", schema = { type = "object" },
        \\      handler = function() return "B" end },
        \\  }
        \\end
    ;
    if (c.luaL_loadstring(rt.L, preload) != 0) return error.TestSetupFailed;
    if (c.lua_pcallk(rt.L, 0, 0, 0, 0, null) != 0) return error.TestSetupFailed;

    var entries: std.array_list.Managed(LuaRuntime.Entry) = .init(testing.allocator);
    defer entries.deinit();
    try rt.evalEntriesFromModule("fakerock", &entries);
    try testing.expectEqual(@as(usize, 2), entries.items.len);
    for (entries.items) |e| try rt.activateEntry(e);
    try testing.expectEqual(@as(usize, 2), rt.toolCount());
}

test "loadScript records tool decls" {
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();

    const source =
        \\panto.ext.register_tool {
        \\  name = "greet", description = "Says hi.",
        \\  schema = { type = "object", properties = { name = { type = "string" } } },
        \\  handler = function(input) return "hi, " .. input.name end,
        \\}
    ;
    const path = try writeTempScript(tmp.dir, "greet.lua", source);
    defer testing.allocator.free(path);

    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();

    try loadScript(rt, path, null);
    try testing.expectEqual(@as(usize, 1), rt.toolCount());
    try testing.expectEqualStrings("greet", rt.decls.items[0].name);
}

test "loadScript records command decls" {
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();

    const source =
        \\_G.last_args = nil
        \\panto.ext.register_command {
        \\  name = "shout", description = "Uppercases its args.",
        \\  handler = function(args) _G.last_args = args:upper() end,
        \\}
    ;
    const path = try writeTempScript(tmp.dir, "shout.lua", source);
    defer testing.allocator.free(path);

    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();

    try loadScript(rt, path, null);
    const cmds = rt.commandList();
    try testing.expectEqual(@as(usize, 1), cmds.len);
    try testing.expectEqualStrings("shout", cmds[0].name);
    try testing.expectEqualStrings("Uppercases its args.", cmds[0].description);
    // No tool was registered.
    try testing.expectEqual(@as(usize, 0), rt.toolCount());

    // Invoke the handler and confirm the side effect ran with our args.
    var err_buf: [256]u8 = undefined;
    var err_len: usize = 0;
    try rt.runCommand(cmds[0].handler_ref, "hi there", &err_buf, &err_len);

    _ = c.lua_getglobal(rt.L, "last_args");
    var len: usize = 0;
    const got = c.lua_tolstring(rt.L, -1, &len);
    try testing.expectEqualStrings("HI THERE", got[0..len]);
    c.lua_settop(rt.L, c.lua_gettop(rt.L) - 1);
}

test "runCommand surfaces a Lua error" {
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();

    const source =
        \\panto.ext.register_command {
        \\  name = "boom", description = "crashes",
        \\  handler = function(args) error("kaboom") end,
        \\}
    ;
    const path = try writeTempScript(tmp.dir, "boom.lua", source);
    defer testing.allocator.free(path);

    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();
    try loadScript(rt, path, null);

    const cmds = rt.commandList();
    var err_buf: [4096]u8 = undefined;
    var err_len: usize = 0;
    try testing.expectError(
        RuntimeError.LuaHandlerCrashed,
        rt.runCommand(cmds[0].handler_ref, "", &err_buf, &err_len),
    );
    try testing.expect(std.mem.indexOf(u8, err_buf[0..err_len], "kaboom") != null);
}

test "duplicate command name within runtime is rejected" {
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();

    const source =
        \\panto.ext.register_command { name = "dup", description = "a", handler = function() end }
        \\panto.ext.register_command { name = "dup", description = "b", handler = function() end }
    ;
    const path = try writeTempScript(tmp.dir, "dup.lua", source);
    defer testing.allocator.free(path);

    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();
    try testing.expectError(error.DuplicateCommand, loadScript(rt, path, null));
}

test "invokeBatch runs each call through a coroutine and returns the result" {
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();

    const source =
        \\panto.ext.register_tool {
        \\  name = "echo", description = "echoes",
        \\  schema = { type = "object", properties = { msg = { type = "string" } } },
        \\  handler = function(input) return "got: " .. input.msg end,
        \\}
        \\panto.ext.register_tool {
        \\  name = "shout", description = "shouts",
        \\  schema = { type = "object", properties = { msg = { type = "string" } } },
        \\  handler = function(input) return input.msg:upper() .. "!" end,
        \\}
    ;
    const path = try writeTempScript(tmp.dir, "ext.lua", source);
    defer testing.allocator.free(path);

    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();
    try loadScript(rt, path, null);

    var src = rt.toolSource();

    const calls = [_]panto.ToolCall{
        .{ .tool_name = "echo", .input = "{\"msg\":\"hello\"}" },
        .{ .tool_name = "shout", .input = "{\"msg\":\"hi\"}" },
        .{ .tool_name = "echo", .input = "{\"msg\":\"again\"}" },
    };
    var results: [3]panto.ToolCallResult = .{
        .{ .err = error.SourceDroppedCall },
        .{ .err = error.SourceDroppedCall },
        .{ .err = error.SourceDroppedCall },
    };
    try src.vtable.invoke_batch(src.ctx, &calls, &results, testing.allocator);

    defer freeResults(&results);

    try testing.expectEqualStrings("got: hello", okText(results[0]));
    try testing.expectEqualStrings("HI!", okText(results[1]));
    try testing.expectEqualStrings("got: again", okText(results[2]));
}

test "module-global state survives across calls in the same runtime" {
    // This is the headline reason the runtime exists. Verify it.
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();

    const source =
        \\local count = 0
        \\panto.ext.register_tool {
        \\  name = "bump", description = "increment counter",
        \\  schema = { type = "object" },
        \\  handler = function(input)
        \\    count = count + 1
        \\    return tostring(count)
        \\  end,
        \\}
    ;
    const path = try writeTempScript(tmp.dir, "counter.lua", source);
    defer testing.allocator.free(path);

    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();
    try loadScript(rt, path, null);
    var src = rt.toolSource();

    const calls = [_]panto.ToolCall{
        .{ .tool_name = "bump", .input = "{}" },
        .{ .tool_name = "bump", .input = "{}" },
        .{ .tool_name = "bump", .input = "{}" },
    };
    var results: [3]panto.ToolCallResult = .{
        .{ .err = error.SourceDroppedCall },
        .{ .err = error.SourceDroppedCall },
        .{ .err = error.SourceDroppedCall },
    };
    try src.vtable.invoke_batch(src.ctx, &calls, &results, testing.allocator);

    defer freeResults(&results);

    try testing.expectEqualStrings("1", okText(results[0]));
    try testing.expectEqualStrings("2", okText(results[1]));
    try testing.expectEqualStrings("3", okText(results[2]));

    // And a second batch keeps the counter going.
    var more: [1]panto.ToolCallResult = .{.{ .err = error.SourceDroppedCall }};
    try src.vtable.invoke_batch(
        src.ctx,
        &[_]panto.ToolCall{.{ .tool_name = "bump", .input = "{}" }},
        &more,
        testing.allocator,
    );
    defer freeResults(&more);
    try testing.expectEqualStrings("4", okText(more[0]));
}

test "handler crash: per-call error surfaces, sibling calls succeed" {
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();

    const source =
        \\panto.ext.register_tool {
        \\  name = "ok", description = "ok",
        \\  schema = { type = "object" },
        \\  handler = function(input) return "fine" end,
        \\}
        \\panto.ext.register_tool {
        \\  name = "boom", description = "bad",
        \\  schema = { type = "object" },
        \\  handler = function(input) error("kaboom") end,
        \\}
    ;
    const path = try writeTempScript(tmp.dir, "mix.lua", source);
    defer testing.allocator.free(path);

    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();
    try loadScript(rt, path, null);
    var src = rt.toolSource();

    const calls = [_]panto.ToolCall{
        .{ .tool_name = "ok", .input = "{}" },
        .{ .tool_name = "boom", .input = "{}" },
        .{ .tool_name = "ok", .input = "{}" },
    };
    var results: [3]panto.ToolCallResult = .{
        .{ .err = error.SourceDroppedCall },
        .{ .err = error.SourceDroppedCall },
        .{ .err = error.SourceDroppedCall },
    };
    try src.vtable.invoke_batch(src.ctx, &calls, &results, testing.allocator);
    defer freeResults(&results);

    try testing.expectEqualStrings("fine", okText(results[0]));
    // A handler crash is surfaced as an *ok* result whose payload is
    // the formatted error message — not as `.err` — because libpanto
    // would otherwise abort the entire turn on per-call `.err`. The
    // payload starts with the well-known `panto-lua: tool '...' failed:`
    // prefix and includes the Lua-side error message.
    try testing.expect(std.mem.startsWith(
        u8,
        okText(results[1]),
        "panto-lua: tool 'boom' failed:",
    ));
    try testing.expect(std.mem.indexOf(u8, okText(results[1]), "kaboom") != null);
    try testing.expectEqualStrings("fine", okText(results[2]));
}

test "directory-style extension can require sibling modules" {
    var tmp = testing.tmpDir(.{ .iterate = true });
    defer tmp.cleanup();

    try tmp.dir.createDirPath(testing.io, "ext");

    try tmp.dir.writeFile(testing.io, .{
        .sub_path = "ext/util.lua",
        .data =
        \\local M = {}
        \\function M.shout(s) return s:upper() .. "!" end
        \\return M
        ,
    });
    try tmp.dir.writeFile(testing.io, .{
        .sub_path = "ext/init.lua",
        .data =
        \\local panto = require("panto")
        \\local util = require("util")
        \\return {
        \\  name = "shout", description = "uppercase + bang",
        \\  schema = { type = "object", properties = { text = { type = "string" } } },
        \\  handler = function(input) return util.shout(input.text) end,
        \\}
        ,
    });

    var path_buf: [std.fs.max_path_bytes]u8 = undefined;
    const ext_len = try tmp.dir.realPathFile(testing.io, "ext", &path_buf);
    const ext_dir = try testing.allocator.dupe(u8, path_buf[0..ext_len]);
    defer testing.allocator.free(ext_dir);

    const init_path = try std.fs.path.join(testing.allocator, &.{ ext_dir, "init.lua" });
    defer testing.allocator.free(init_path);

    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();
    try loadScript(rt, init_path, ext_dir);

    var src = rt.toolSource();

    const calls = [_]panto.ToolCall{.{ .tool_name = "shout", .input = "{\"text\":\"hi\"}" }};
    var results: [1]panto.ToolCallResult = .{.{ .err = error.SourceDroppedCall }};
    try src.vtable.invoke_batch(src.ctx, &calls, &results, testing.allocator);
    defer freeResults(&results);
    try testing.expectEqualStrings("HI!", okText(results[0]));
}

test "yielding handler with no event loop surfaces a suspension error" {
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();

    const source =
        \\panto.ext.register_tool {
        \\  name = "sleeper", description = "yields forever",
        \\  schema = { type = "object" },
        \\  handler = function(input) coroutine.yield() ; return "never" end,
        \\}
    ;
    const path = try writeTempScript(tmp.dir, "y.lua", source);
    defer testing.allocator.free(path);

    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();
    try loadScript(rt, path, null);
    var src = rt.toolSource();

    const calls = [_]panto.ToolCall{.{ .tool_name = "sleeper", .input = "{}" }};
    var results: [1]panto.ToolCallResult = .{.{ .err = error.SourceDroppedCall }};
    try src.vtable.invoke_batch(src.ctx, &calls, &results, testing.allocator);
    defer freeResults(&results);

    // Same policy as the crash test: the failure is surfaced as `.ok`
    // text so libpanto doesn't abort the turn. The message is the shared
    // entrypoint machinery's post-drain suspension diagnosis (without luv
    // the drain is a no-op, so the yield can never be resumed).
    try testing.expect(std.mem.startsWith(
        u8,
        okText(results[0]),
        "panto-lua: tool 'sleeper' failed:",
    ));
    try testing.expect(std.mem.indexOf(u8, okText(results[0]), "suspended after the uv loop drained") != null);
}

// Integration test: requires a data home with luv already installed.
// Skipped if luv isn't on disk — unit tests stay offline.
test "scheduler: yielding handler is resumed by libuv" {
    var env = try processDataHomeEnv(testing.allocator);
    defer env.deinit();
    const data_home = panto_home.homePath(testing.allocator, &env) catch |err| switch (err) {
        error.NoHomeDirectory => return error.SkipZigTest,
        else => return err,
    };
    defer testing.allocator.free(data_home);

    // Check for `<home>/rocks/lua-<version>/lib/lua/5.4/luv.so`.
    const manifest = @import("manifest.zig");
    var path_buf: [std.fs.max_path_bytes]u8 = undefined;
    const so_path = try std.fmt.bufPrint(
        &path_buf,
        "{s}/rocks/lua-{s}/lib/lua/{s}/luv.so",
        .{ data_home, manifest.lua_version, manifest.lua_short_version },
    );
    std.Io.Dir.cwd().access(testing.io, so_path, .{}) catch return error.SkipZigTest;

    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();

    const source =
        \\local uv = require("luv")
        \\panto.ext.register_tool {
        \\  name = "timer_say", description = "sleep then return",
        \\  schema = { type = "object" },
        \\  handler = function(input)
        \\    local co = coroutine.running()
        \\    local timer = uv.new_timer()
        \\    uv.timer_start(timer, 5, 0, function()
        \\      uv.timer_stop(timer)
        \\      uv.close(timer)
        \\      coroutine.resume(co)
        \\    end)
        \\    coroutine.yield()
        \\    return "awake"
        \\  end,
        \\}
    ;
    const path = try writeTempScript(tmp.dir, "timer.lua", source);
    defer testing.allocator.free(path);

    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();

    // Bootstrap luarocks (so `require("luv")` works), then install
    // the scheduler from the normal data-home location.
    const luarocks_runtime = @import("luarocks_runtime.zig");
    // The bootstrap needs a panto executable path for the wrapper
    // script; tests don't actually invoke it, so a placeholder is
    // fine (the wrapper is only consulted when luarocks itself
    // shells out, which the test never triggers).
    const luarocks_rt = try luarocks_runtime.bootstrap(
        testing.allocator,
        testing.io,
        &env,
        rt.L,
        "/usr/bin/true",
    );
    defer luarocks_rt.deinit();

    try rt.installScheduler();
    try loadScript(rt, path, null);

    var src = rt.toolSource();
    const calls = [_]panto.ToolCall{
        .{ .tool_name = "timer_say", .input = "{}" },
        .{ .tool_name = "timer_say", .input = "{}" },
    };
    var results: [2]panto.ToolCallResult = .{
        .{ .err = error.SourceDroppedCall },
        .{ .err = error.SourceDroppedCall },
    };
    try src.vtable.invoke_batch(src.ctx, &calls, &results, testing.allocator);
    defer freeResults(&results);

    try testing.expectEqualStrings("awake", okText(results[0]));
    try testing.expectEqualStrings("awake", okText(results[1]));
}

// Contract-violation guard: a handler that yields WITHOUT arming any
// libuv work can never be resumed. Under the run-to-completion model,
// `uv.run` drains immediately (nothing to wait on) and the scheduler
// must surface the still-suspended coroutine as a per-call error
// rather than hanging the process. A sibling well-behaved call in the
// same batch must still succeed.
test "scheduler: handler that yields without arming libuv work is surfaced as an error" {
    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();
    const luarocks_rt = bootstrapRealRuntime(rt) catch |e| return e;
    defer luarocks_rt.deinit();

    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    const source =
        \\panto.ext.register_tool {
        \\  name = "abandon", description = "yields into the void",
        \\  schema = { type = "object" },
        \\  handler = function() coroutine.yield(); return "never" end,
        \\}
        \\panto.ext.register_tool {
        \\  name = "fine", description = "sync ok",
        \\  schema = { type = "object" },
        \\  handler = function() return "ok" end,
        \\}
    ;
    const path = try writeTempScript(tmp.dir, "abandon.lua", source);
    defer testing.allocator.free(path);
    try loadScript(rt, path, null);

    var src = rt.toolSource();
    const calls = [_]panto.ToolCall{
        .{ .tool_name = "abandon", .input = "{}" },
        .{ .tool_name = "fine", .input = "{}" },
    };
    var results: [2]panto.ToolCallResult = .{
        .{ .err = error.SourceDroppedCall },
        .{ .err = error.SourceDroppedCall },
    };
    try src.vtable.invoke_batch(src.ctx, &calls, &results, testing.allocator);
    defer freeResults(&results);

    try testing.expect(std.mem.startsWith(u8, okText(results[0]), "panto-lua: tool 'abandon' failed:"));
    try testing.expect(std.mem.indexOf(u8, okText(results[0]), "still suspended") != null);
    try testing.expectEqualStrings("ok", okText(results[1]));
}

/// Absolute path to the repo's `agent/tools` directory, derived from
/// the build-time repo root (the test runner's cwd is unreliable).
/// Caller owns the returned slice. Skips the test if the directory
/// isn't present.
fn findAgentToolsDir() ![]const u8 {
    const versions = @import("versions");
    const tools = try std.fs.path.join(testing.allocator, &.{ versions.repo_root, "agent", "tools" });
    errdefer testing.allocator.free(tools);
    const probe = try std.fs.path.join(testing.allocator, &.{ tools, "shell.lua" });
    defer testing.allocator.free(probe);
    std.Io.Dir.cwd().access(testing.io, probe, .{}) catch {
        testing.allocator.free(tools);
        return error.SkipZigTest;
    };
    return tools;
}

fn bootstrapRealRuntime(rt: *LuaRuntime) !*@import("luarocks_runtime.zig").LuarocksRuntime {
    var env = try processDataHomeEnv(testing.allocator);
    errdefer env.deinit();
    const data_home = panto_home.homePath(testing.allocator, &env) catch |err| switch (err) {
        error.NoHomeDirectory => return error.SkipZigTest,
        else => return err,
    };
    defer testing.allocator.free(data_home);

    const manifest = @import("manifest.zig");
    var path_buf: [std.fs.max_path_bytes]u8 = undefined;
    const so_path = try std.fmt.bufPrint(
        &path_buf,
        "{s}/rocks/lua-{s}/lib/lua/{s}/luv.so",
        .{ data_home, manifest.lua_version, manifest.lua_short_version },
    );
    std.Io.Dir.cwd().access(testing.io, so_path, .{}) catch return error.SkipZigTest;

    const luarocks_runtime = @import("luarocks_runtime.zig");
    const luarocks_rt = try luarocks_runtime.bootstrap(
        testing.allocator,
        testing.io,
        &env,
        rt.L,
        "/usr/bin/true",
    );
    env.deinit();
    try rt.installScheduler();
    return luarocks_rt;
}

fn processDataHomeEnv(allocator: Allocator) !std.process.Environ.Map {
    var env = std.process.Environ.Map.init(allocator);
    errdefer env.deinit();

    if (std.c.getenv("XDG_DATA_HOME")) |value| {
        try env.put("XDG_DATA_HOME", std.mem.sliceTo(value, 0));
    }
    if (std.c.getenv("HOME")) |value| {
        try env.put("HOME", std.mem.sliceTo(value, 0));
    }
    return env;
}

// Reproduction: two REAL `std.shell` calls in one batch. shell.lua
// uses spawn + two pipes + a timeout timer (3+ libuv events per call)
// and resumes its own coroutine from a libuv callback. This exercises
// the exact C-resume topology that the synthetic `timer_say` test does
// not (spawn, dual pipes, spill-file fs calls). Skipped without luv.
test "scheduler: two real std.shell calls in one batch do not deadlock" {
    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();
    const luarocks_rt = bootstrapRealRuntime(rt) catch |e| return e;
    defer luarocks_rt.deinit();

    const tools_dir = try findAgentToolsDir();
    defer testing.allocator.free(tools_dir);
    const shell_path = try std.fs.path.join(testing.allocator, &.{ tools_dir, "shell.lua" });
    defer testing.allocator.free(shell_path);
    try loadScript(rt, shell_path, tools_dir);

    var src = rt.toolSource();
    const calls = [_]panto.ToolCall{
        .{ .tool_name = "std.shell", .input = "{\"command\":\"echo first; sleep 0.05; echo first-done\"}" },
        .{ .tool_name = "std.shell", .input = "{\"command\":\"echo second; sleep 0.05; echo second-done\"}" },
    };
    var results: [2]panto.ToolCallResult = .{
        .{ .err = error.SourceDroppedCall },
        .{ .err = error.SourceDroppedCall },
    };
    try src.vtable.invoke_batch(src.ctx, &calls, &results, testing.allocator);
    defer freeResults(&results);

    try testing.expect(std.mem.indexOf(u8, okText(results[0]), "first-done") != null);
    try testing.expect(std.mem.indexOf(u8, okText(results[1]), "second-done") != null);
}

test "scheduler: shell timeout is measured from dispatch, not a stale loop clock" {
    // Regression: libuv caches the loop's "now" and only refreshes it while
    // `uv.run` runs. The loop is idle between batches, so a tool that arms a
    // timeout timer against the stale clock would have it fire the instant
    // `uv.run` refreshes time. We force staleness with `uv.sleep` (a blocking
    // sleep that does NOT run the loop), then dispatch a 1s-timeout shell call
    // and assert it does not spuriously time out.
    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();
    const luarocks_rt = bootstrapRealRuntime(rt) catch |e| return e;
    defer luarocks_rt.deinit();

    const tools_dir = try findAgentToolsDir();
    defer testing.allocator.free(tools_dir);
    const shell_path = try std.fs.path.join(testing.allocator, &.{ tools_dir, "shell.lua" });
    defer testing.allocator.free(shell_path);
    try loadScript(rt, shell_path, tools_dir);

    var src = rt.toolSource();

    // Prime the loop clock with one run.
    {
        const warm = [_]panto.ToolCall{.{ .tool_name = "std.shell", .input = "{\"command\":\"true\"}" }};
        var wres: [1]panto.ToolCallResult = .{.{ .err = error.SourceDroppedCall }};
        try src.vtable.invoke_batch(src.ctx, &warm, &wres, testing.allocator);
        freeResults(&wres);
    }

    // Advance real time ~1.2s WITHOUT running the loop, so its cached clock
    // goes stale relative to the wall clock by more than the timeout below.
    {
        const snippet = "require('luv').sleep(1200)";
        if (c.luaL_loadstring(rt.L, snippet) != 0) return error.SleepSnippetFailed;
        if (c.lua_pcallk(rt.L, 0, 0, 0, 0, null) != 0) return error.SleepSnippetFailed;
    }

    const calls = [_]panto.ToolCall{
        .{ .tool_name = "std.shell", .input = "{\"command\":\"echo prompt-done\",\"timeout\":1}" },
    };
    var results: [1]panto.ToolCallResult = .{.{ .err = error.SourceDroppedCall }};
    try src.vtable.invoke_batch(src.ctx, &calls, &results, testing.allocator);
    defer freeResults(&results);

    try testing.expect(std.mem.indexOf(u8, okText(results[0]), "prompt-done") != null);
    try testing.expect(std.mem.indexOf(u8, okText(results[0]), "timed out") == null);
}

test "scheduler: three real std.shell calls with explicit timeout all succeed" {
    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();
    const luarocks_rt = bootstrapRealRuntime(rt) catch |e| return e;
    defer luarocks_rt.deinit();

    const tools_dir = try findAgentToolsDir();
    defer testing.allocator.free(tools_dir);
    const shell_path = try std.fs.path.join(testing.allocator, &.{ tools_dir, "shell.lua" });
    defer testing.allocator.free(shell_path);
    try loadScript(rt, shell_path, tools_dir);

    var src = rt.toolSource();
    const calls = [_]panto.ToolCall{
        .{ .tool_name = "std.shell", .input = "{\"command\":\"echo alpha-done\",\"timeout\":30}" },
        .{ .tool_name = "std.shell", .input = "{\"command\":\"echo bravo-done\",\"timeout\":30}" },
        .{ .tool_name = "std.shell", .input = "{\"command\":\"echo charlie-done\",\"timeout\":30}" },
    };
    var results: [3]panto.ToolCallResult = .{
        .{ .err = error.SourceDroppedCall },
        .{ .err = error.SourceDroppedCall },
        .{ .err = error.SourceDroppedCall },
    };
    try src.vtable.invoke_batch(src.ctx, &calls, &results, testing.allocator);
    defer freeResults(&results);

    try testing.expect(std.mem.indexOf(u8, okText(results[0]), "alpha-done") != null);
    try testing.expect(std.mem.indexOf(u8, okText(results[1]), "bravo-done") != null);
    try testing.expect(std.mem.indexOf(u8, okText(results[2]), "charlie-done") != null);
    // None should have hit the timeout path.
    for (results) |r| {
        try testing.expect(std.mem.indexOf(u8, okText(r), "timed out") == null);
    }
}

// Reproduction: two REAL `std.read` calls in one batch. read.lua uses
// only synchronous `uv.fs_*` calls and never yields, so both should
// complete during dispatch without entering the drive loop.
test "scheduler: two real std.read calls in one batch do not deadlock" {
    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();
    const luarocks_rt = bootstrapRealRuntime(rt) catch |e| return e;
    defer luarocks_rt.deinit();

    const tools_dir = try findAgentToolsDir();
    defer testing.allocator.free(tools_dir);
    const read_path = try std.fs.path.join(testing.allocator, &.{ tools_dir, "read.lua" });
    defer testing.allocator.free(read_path);
    try loadScript(rt, read_path, tools_dir);

    const in0 = try std.fmt.allocPrint(testing.allocator, "{{\"path\":\"{s}/shell.lua\"}}", .{tools_dir});
    defer testing.allocator.free(in0);
    const in1 = try std.fmt.allocPrint(testing.allocator, "{{\"path\":\"{s}/read.lua\"}}", .{tools_dir});
    defer testing.allocator.free(in1);

    var src = rt.toolSource();
    const calls = [_]panto.ToolCall{
        .{ .tool_name = "std.read", .input = in0 },
        .{ .tool_name = "std.read", .input = in1 },
    };
    var results: [2]panto.ToolCallResult = .{
        .{ .err = error.SourceDroppedCall },
        .{ .err = error.SourceDroppedCall },
    };

    try src.vtable.invoke_batch(src.ctx, &calls, &results, testing.allocator);
    defer freeResults(&results);

    try testing.expect(okText(results[0]).len > 0);
    try testing.expect(okText(results[1]).len > 0);
}

test "loadScript: duplicate tool name from a second extension errors" {
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();

    const a =
        \\panto.ext.register_tool {
        \\  name = "clash", description = "a",
        \\  schema = { type = "object" },
        \\  handler = function(input) return "a" end,
        \\}
    ;
    const b =
        \\panto.ext.register_tool {
        \\  name = "clash", description = "b",
        \\  schema = { type = "object" },
        \\  handler = function(input) return "b" end,
        \\}
    ;
    const pa = try writeTempScript(tmp.dir, "a.lua", a);
    defer testing.allocator.free(pa);
    const pb = try writeTempScript(tmp.dir, "b.lua", b);
    defer testing.allocator.free(pb);

    var rt = try LuaRuntime.create(testing.allocator);
    defer rt.deinit();
    try loadScript(rt, pa, null);

    try testing.expectError(error.DuplicateTool, loadScript(rt, pb, null));
}