//! Long-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` for its
//! entire lifetime. Every extension is loaded into it once; extension
//! top-level code runs exactly once at startup. 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
//! `libpanto/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 c = lua_bridge.c;
const Io = std.Io;

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.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").run`, the function the scheduler
    /// calls to drive libuv to completion. `0` until
    /// `installScheduler` runs.
    uv_run_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,

    /// Create a new runtime. The `lua_State` is opened, standard libs
    /// loaded, and the `panto.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);

        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),
        };
        return self;
    }

    /// 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").run` for fast per-tick access.
    ///
    /// Must be called after luarocks bootstrap has installed luv,
    /// otherwise the `require("luv")` step will fail.
    pub fn installScheduler(self: *LuaRuntime) !void {
        try installRecordResult(self);
        try installWrapperClosure(self);
        try cacheUvRun(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_run_ref != 0) {
            c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, self.uv_run_ref);
        }

        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);
    }

    /// Load and execute one Lua extension script in this runtime.
    ///
    /// `package_root`, if provided, is prepended to `package.path` so
    /// `require` finds sibling modules.
    ///
    /// All `panto.register_tool` calls in the script run during this
    /// call. The runtime then harvests the registrations table,
    /// transfers handler functions into the Lua registry (one `luaL_ref`
    /// per tool), and records each tool's metadata in `self.decls`.
    pub fn loadExtension(
        self: *LuaRuntime,
        script_path: []const u8,
        package_root: ?[]const u8,
    ) !void {
        const path_z = try self.allocator.dupeZ(u8, script_path);
        defer self.allocator.free(path_z);

        // Reset the registrations table to empty so we only harvest the
        // calls made by *this* script (not accumulated from prior ones).
        // The bridge re-installs the registrations table when called;
        // we want to call only that subset. Instead of re-installing
        // everything (which would also reset the panto global, fine),
        // create a fresh registrations table directly via the bridge.
        lua_bridge.resetRegistrations(self.L);

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

        lua_bridge.loadFile(self.L, path_z) catch |err| {
            logTopAsError(self.L, "lua: failed to load extension");
            return err;
        };

        // Harvest the registrations table into our state.
        try self.harvestAndStoreHandlers();
    }

    /// Load a single-tool Lua script and register the table it returns
    /// as if `panto.register_tool` had been called on that table.
    ///
    /// The script's top-level chunk must return a table with the same
    /// shape that `panto.register_tool` accepts:
    /// `{ name, description, schema, handler }`. This is the ergonomic
    /// form supported under a `tools/` directory.
    pub fn loadTool(
        self: *LuaRuntime,
        script_path: []const u8,
        package_root: ?[]const u8,
    ) !void {
        const path_z = try self.allocator.dupeZ(u8, script_path);
        defer self.allocator.free(path_z);

        lua_bridge.resetRegistrations(self.L);

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

        // Run the file expecting exactly one returned value (the tool
        // table). Use luaL_loadfilex + lua_pcallk directly so we can
        // ask for a return value (the bridge's loadFile discards them).
        //
        // Push `panto.register_tool` *first*, then load+run the chunk so
        // its return value naturally lands above it; calling pcall then
        // consumes both in the right order.
        const L = self.L;
        _ = c.lua_getglobal(L, "panto");
        _ = c.lua_getfield(L, -1, "register_tool");
        c.lua_copy(L, -1, -2); // overwrite `panto` with `register_tool`
        c.lua_settop(L, c.lua_gettop(L) - 1); // pop the duplicate
        // Stack: ..., register_tool

        if (c.luaL_loadfilex(L, path_z.ptr, null) != 0) {
            logTopAsError(L, "lua: failed to load tool");
            c.lua_settop(L, c.lua_gettop(L) - 2); // pop err + register_tool
            return error.LuaLoadFailed;
        }
        if (c.lua_pcallk(L, 0, 1, 0, 0, null) != 0) {
            logTopAsError(L, "lua: failed to run tool");
            c.lua_settop(L, c.lua_gettop(L) - 2); // pop err + register_tool
            return error.LuaRunFailed;
        }
        // Stack: ..., register_tool, returned_value
        if (c.lua_type(L, -1) != lua_bridge.T_TABLE) {
            c.lua_settop(L, c.lua_gettop(L) - 2); // pop both
            std.log.err(
                "lua: tool script '{s}' must return a table",
                .{script_path},
            );
            return error.BadRegistration;
        }

        // Invoke register_tool(returned_table). Same validation, schema
        // serialization, and registrations-table append logic as an
        // extension's `panto.register_tool` call.
        if (c.lua_pcallk(L, 1, 0, 0, 0, null) != 0) {
            logTopAsError(L, "lua: register_tool failed for tool script");
            return error.LuaRunFailed;
        }

        try self.harvestAndStoreHandlers();
    }

    /// 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();
    }

    /// 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 synchronously. `handler_ref` is a
    /// registry ref to a `function(args)` closure; `args` is the trimmed
    /// remainder of the command line. The handler runs under `pcall` with
    /// a `debug.traceback` error handler so failures surface a readable
    /// message instead of crashing the REPL.
    ///
    /// 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
    /// 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;

        // Install a traceback error handler beneath the call frame.
        _ = c.lua_getglobal(L, "debug");
        _ = c.lua_getfield(L, -1, "traceback");
        c.lua_copy(L, -1, -2);
        c.lua_settop(L, c.lua_gettop(L) - 1); // pop the `debug` table
        const errfunc_idx = c.lua_gettop(L);

        // Push handler + args string.
        _ = c.lua_rawgeti(L, lua_bridge.LUA_REGISTRYINDEX, @intCast(handler_ref));
        _ = c.lua_pushlstring(L, args.ptr, args.len);

        const rc = c.lua_pcallk(L, 1, 0, errfunc_idx, 0, null);
        if (rc != 0) {
            var len: usize = 0;
            const ptr = c.lua_tolstring(L, -1, &len);
            if (ptr != null) {
                const n = @min(len, err_buf.len);
                @memcpy(err_buf[0..n], ptr[0..n]);
                err_len.* = n;
            }
            c.lua_settop(L, c.lua_gettop(L) - 2); // pop error + errfunc
            return RuntimeError.LuaHandlerCrashed;
        }
        c.lua_settop(L, c.lua_gettop(L) - 1); // pop errfunc
    }

    /// 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.freeResultParts`.
    value: ?[]panto.ResultPart = 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 or self.uv_run_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;
    }

    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.textResult(allocator, "") };
            // Free the err_msg if both ended up set somehow.
            if (slot.err_msg) |m| allocator.free(m);
        } else {
            if (slot.value) |v| panto.freeResultParts(allocator, v);
            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.ResultPart {
    const text = try std.fmt.allocPrint(
        allocator,
        "panto-lua: tool '{s}' failed: {s}",
        .{ tool_name, message },
    );
    return panto.ownedTextResult(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,
    };
}

/// Call `uv.run("default")`, which runs the event loop until there are
/// no active handles or requests left. 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. The boolean return value is discarded.
fn driveUvToCompletion(self: *LuaRuntime) !void {
    const L = self.L;
    _ = c.lua_rawgeti(L, lua_bridge.LUA_REGISTRYINDEX, @intCast(self.uv_run_ref));
    _ = c.lua_pushlstring(L, "default", 7);
    if (c.lua_pcallk(L, 1, 1, 0, 0, null) != 0) {
        logTopAsError(L, "panto-lua: uv.run failed");
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return error.UvRunFailed;
    }
    // Discard the boolean return value.
    c.lua_settop(L, c.lua_gettop(L) - 1);
}

/// 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 synchronously, with no event loop. On Lua-level
/// failure the error message string 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.ResultPart {
    const co = c.lua_newthread(L) orelse return RuntimeError.LuaInitFailed;
    defer c.lua_settop(L, c.lua_gettop(L) - 1);

    _ = c.lua_rawgeti(co, lua_bridge.LUA_REGISTRYINDEX, @intCast(handler_ref));
    if (c.lua_type(co, -1) != lua_bridge.T_FUNCTION) {
        return RuntimeError.LuaHandlerNotFound;
    }

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

    var nresults: c_int = 0;
    const status = c.lua_resume(co, L, 1, &nresults);

    switch (status) {
        c.LUA_OK => {
            if (nresults < 1) return RuntimeError.BadHandlerReturn;
            return try lua_bridge.readHandlerResult(co, -1, allocator);
        },
        c.LUA_YIELD => {
            const msg = "lua: tool handler yielded with no event loop installed; call installScheduler() before dispatching";
            if (@import("builtin").is_test) {
                std.log.warn("{s}", .{msg});
            } else {
                std.log.err("{s}", .{msg});
            }
            return RuntimeError.LuaHandlerYielded;
        },
        else => {
            // Capture the Lua error message into the caller's slot
            // *before* logging+returning, so the legacy sync path can
            // surface it to the model rather than just the typed
            // error name.
            var msg_len: usize = 0;
            const msg_ptr = c.lua_tolstring(co, -1, &msg_len);
            if (msg_ptr != null) {
                err_msg_out.* = allocator.dupe(u8, msg_ptr[0..msg_len]) catch null;
            }
            logTopAsError(co, "lua: handler crashed");
            return RuntimeError.LuaHandlerCrashed;
        },
    }
}

// ---------------------------------------------------------------------------
// 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;
    _ = c.lua_getglobal(L, "panto");
    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;
    const snippet =
        \\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;
    }
    if (c.lua_pcallk(L, 0, 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.
fn cacheUvRun(self: *LuaRuntime) !void {
    const L = self.L;
    const snippet =
        \\local uv = require("luv")
        \\return uv.run
    ;
    if (c.luaL_loadstring(L, snippet) != 0) {
        logTopAsError(L, "panto-lua: failed to compile luv 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;
    }
    // Stack: [..., uv.run]. luaL_ref pops the top.
    if (c.lua_type(L, -1) != lua_bridge.T_FUNCTION) {
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return RuntimeError.LuaInitFailed;
    }
    self.uv_run_ref = c.luaL_ref(L, lua_bridge.LUA_REGISTRYINDEX);
}

// ---------------------------------------------------------------------------
// 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) |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| panto.freeResultParts(testing.allocator, b),
        .err => {},
    };
}

fn writeTempScript(dir: Io.Dir, name: []const u8, source: []const u8) ![]const u8 {
    try dir.writeFile(testing.io, .{ .sub_path = name, .data = source });
    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 "loadExtension records tool decls" {
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();

    const source =
        \\panto.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 rt.loadExtension(path, null);
    try testing.expectEqual(@as(usize, 1), rt.toolCount());
    try testing.expectEqualStrings("greet", rt.decls.items[0].name);
}

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

    const source =
        \\_G.last_args = nil
        \\panto.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 rt.loadExtension(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.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 rt.loadExtension(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.register_command { name = "dup", description = "a", handler = function() end }
        \\panto.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, rt.loadExtension(path, null));
}

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

    const source =
        \\panto.register_tool {
        \\  name = "echo", description = "echoes",
        \\  schema = { type = "object", properties = { msg = { type = "string" } } },
        \\  handler = function(input) return "got: " .. input.msg end,
        \\}
        \\panto.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 rt.loadExtension(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.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 rt.loadExtension(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.register_tool {
        \\  name = "ok", description = "ok",
        \\  schema = { type = "object" },
        \\  handler = function(input) return "fine" end,
        \\}
        \\panto.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 rt.loadExtension(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 util = require("util")
        \\panto.register_tool {
        \\  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 rt.loadExtension(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 LuaHandlerYielded" {
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();

    const source =
        \\panto.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 rt.loadExtension(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 error type name
    // (`LuaHandlerYielded`) is included via `@errorName` in the legacy
    // sync path's formatToolError call.
    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]), "LuaHandlerYielded") != null);
}

// Integration test: requires a `$PANTO_HOME` with luv already
// installed. Skipped if luv isn't on disk — unit tests stay offline.
test "scheduler: yielding handler is resumed by libuv" {
    const home_z = std.c.getenv("PANTO_HOME") orelse return error.SkipZigTest;
    const panto_home_env = std.mem.sliceTo(home_z, 0);
    // 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",
        .{ panto_home_env, 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.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. We use the real environment so the test picks
    // up the same PANTO_HOME the developer's machine has.
    var env: std.process.Environ.Map = .init(testing.allocator);
    defer env.deinit();
    try env.put("PANTO_HOME", panto_home_env);

    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 rt.loadExtension(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.register_tool {
        \\  name = "abandon", description = "yields into the void",
        \\  schema = { type = "object" },
        \\  handler = function() coroutine.yield(); return "never" end,
        \\}
        \\panto.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 rt.loadExtension(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 {
    const home_z = std.c.getenv("PANTO_HOME") orelse return error.SkipZigTest;
    const panto_home_env = std.mem.sliceTo(home_z, 0);
    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",
        .{ panto_home_env, manifest.lua_version, manifest.lua_short_version },
    );
    std.Io.Dir.cwd().access(testing.io, so_path, .{}) catch return error.SkipZigTest;

    var env: std.process.Environ.Map = .init(testing.allocator);
    defer env.deinit();
    try env.put("PANTO_HOME", panto_home_env);
    const luarocks_runtime = @import("luarocks_runtime.zig");
    const luarocks_rt = try luarocks_runtime.bootstrap(
        testing.allocator,
        testing.io,
        &env,
        rt.L,
        "/usr/bin/true",
    );
    try rt.installScheduler();
    return luarocks_rt;
}

// 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 rt.loadTool(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);
}

// 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 rt.loadTool(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 "loadExtension: duplicate tool name from a second extension errors" {
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();

    const a =
        \\panto.register_tool {
        \\  name = "clash", description = "a",
        \\  schema = { type = "object" },
        \\  handler = function(input) return "a" end,
        \\}
    ;
    const b =
        \\panto.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 rt.loadExtension(pa, null);

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