//! `libpanto-lua` — a native Lua 5.4 C-module for `libpanto`, in pure Zig.
//!
//! This file `@cImport`s the Lua 5.4 C headers and builds the module table
//! plus two `luaL_newmetatable` userdata types — `Agent` and `Stream` —
//! against the translated C types, calling the **Zig** `libpanto` API
//! directly (no `libpanto-c` dependency, no C translation units). The
//! `build.zig` emits a loadable `panto.so` exporting `luaopen_panto`,
//! discovered on `package.cpath` and loaded by `require('panto')`.
//!
//! The unifying streaming contract (see `docs/libpanto-bindings.md`) maps
//! onto Lua exactly like Python — Lua is a pull-iterator language too:
//!
//!   | layer | progress / terminal   | exhausted     | failure                 |
//!   | ----- | --------------------- | ------------- | ----------------------- |
//!   | Lua   | event table (yielded) | iterator ends | error(...) / pcall false |
//!
//!   - `Event`  -> push the event as a Lua table and return it.
//!   - `null`   -> return `nil`, ending the `for` loop.
//!   - `error.X` -> `lua_error` with a mapped message, catchable via `pcall`.
//!
//! ## The surface
//!
//!   local panto = require('panto')
//!   local agent = panto.agent {
//!     api_style = "openai_chat",       -- or "anthropic_messages"
//!     api_key   = "...",
//!     base_url  = "https://...",
//!     model     = "...",
//!     max_tokens = 64000,              -- optional
//!     reasoning  = "medium",           -- optional (openai_chat)
//!     thinking   = "enabled",          -- optional (anthropic_messages)
//!     -- ...see config.zig parsing below for the complete set...
//!   }
//!
//!   agent:register_tool {             -- give the model a tool
//!     name = "...", description = "...",
//!     schema = { type = "object", ... },
//!     handler = function(input) ... return "result" end,
//!   }
//!
//!   agent:set_config { ... }           -- swap provider/model/policy
//!   agent:add_system_message("...")    -- append to the system prompt
//!   agent:set_system_prompt("...")     -- replace it
//!   agent:compact()                    -- summarize older turns
//!   print(agent:session_id())
//!
//!   for ev in agent:run("hello"):events() do
//!     if ev.type == "content_delta" then io.write(ev.delta) end
//!   end
//!
//!   -- Borrowed host agents only (`panto.ext.agent` in the CLI): queue the
//!   -- next user message; the HOST opens and pumps the turn, not the caller.
//!   panto.ext.agent:submit("prompt")   -- or a { <block>, ... } array
//!
//! ## Tools, coroutines, and luv
//!
//! Tool handlers run as Lua coroutines driven by libuv (via the `luv`
//! rock, a hard dependency of this package). The contract mirrors the CLI
//! runtime exactly: a handler may block **only** by yielding on a pending
//! libuv operation whose callback resumes it, so a single
//! `uv.run("default")` pass guarantees every handler has finished. A
//! purely synchronous handler runs to completion on its first resume and
//! needs no event loop.
//!
//! libpanto delivers every Lua-tool call for a turn in one `invoke_batch`
//! on a single thread (the source-grouped dispatch contract); that thread
//! is *not* the host thread, but the host thread is parked inside
//! `Stream.next()` for the whole batch, so only one thread ever touches
//! the `lua_State` at a time. We start one coroutine per call and drive
//! `uv.run` to completion.
//!
//! ## Lua-version pinning
//!
//! Lua has no stable ABI; this module is built against 5.4 headers and the
//! `luaL_checkversion` call in `luaopen_panto` makes a wrong-version load
//! fail loud. 5.3/5.2/5.1(LuaJIT)/5.5 are out of scope for v1.

const std = @import("std");
const panto = @import("panto");

pub const c = @cImport({
    @cInclude("lua.h");
    @cInclude("lauxlib.h");
    @cInclude("lualib.h");
});

// translate_c surfaces Lua's `#define`d type tags as inline functions that
// aren't usable as comptime constants in switch prongs; mirror the
// canonical 5.4 values as clean `c_int`s.
const T_NIL: c_int = 0;
const T_BOOLEAN: c_int = 1;
const T_LIGHTUSERDATA: c_int = 2;
const T_NUMBER: c_int = 3;
const T_STRING: c_int = 4;
const T_TABLE: c_int = 5;
const T_FUNCTION: c_int = 6;

// LUAI_MAXSTACK defaults to 1_000_000 on 64-bit, so the registry pseudo-
// index is -1_001_000 (matches lua.h). Parity with `src/lua_bridge.zig`.
const LUA_REGISTRYINDEX: c_int = -1001000;

const c_allocator = std.heap.c_allocator;

const SOURCE_NAME = "panto-lua";

// ===========================================================================
// Process-global I/O context
// ===========================================================================
//
// `libpanto` needs a process-global HTTP client (`panto.init`) and a
// `std.Io` to drive blocking provider reads. A standalone Lua interpreter
// has no `std.process.Init`, so we own a `std.Io.Threaded` here and
// initialize it lazily on first `panto.agent{}`, tearing it down when the
// last Agent is collected. Single-interpreter assumption: a standalone
// `require('panto')` runs in one `lua_State` and the embedded CLI VM is
// single-threaded; we do not guard the refcount with an atomic.

const IoContext = struct {
    threaded: std.Io.Threaded,
    io: std.Io,
    agent_refs: usize = 0,
    initialized: bool = false,
};

var io_ctx: IoContext = .{ .threaded = undefined, .io = undefined };

fn retainIo() void {
    if (!io_ctx.initialized) {
        io_ctx.threaded = .init(c_allocator, .{});
        io_ctx.io = io_ctx.threaded.io();
        panto.init(c_allocator, io_ctx.io);
        io_ctx.initialized = true;
    }
    io_ctx.agent_refs += 1;
}

fn releaseIo() void {
    if (io_ctx.agent_refs == 0) return;
    io_ctx.agent_refs -= 1;
    if (io_ctx.agent_refs == 0 and io_ctx.initialized) {
        panto.deinit();
        io_ctx.threaded.deinit();
        io_ctx.initialized = false;
    }
}

// ===========================================================================
// Userdata payloads
// ===========================================================================

const AGENT_MT = "panto.Agent";
const STREAM_MT = "panto.Stream";
const CONV_MT = "panto.Conversation";
const STORE_MT = "panto.SessionStore";

/// Boxed `*Agent` userdata.
///
/// `libpanto` borrows `*const Config`, so we own the live `Config` and
/// every string it points at, kept alive for the agent's lifetime. A
/// `set_config` swap pins a *new* config without freeing the old one
/// eagerly: an in-flight turn (or a concurrent tool worker) may still be
/// reading the old snapshot, so old configs and their strings accumulate
/// in `string_arena`/`old_configs` and are freed together at agent
/// teardown. (Bounded by the number of `set_config` calls — a tiny,
/// deliberate retain-until-deinit.)
///
/// A `NullStore` is embedded so a standalone consumer needs no session
/// wiring; the store is borrowed by the agent and must outlive it, so it
/// lives in the same box.
///
/// The Lua tool source is created lazily on the first `register_tool` and
/// registered with the agent then.
const AgentBox = struct {
    agent: *panto.Agent,
    /// The live config the agent reads. Heap-pinned so `set_config` can
    /// swap the pointee the agent holds without moving this box.
    config: *panto.Config,
    /// Superseded configs, freed at teardown (see the struct doc).
    old_configs: std.ArrayList(*panto.Config),
    store_ref: c_int = c.LUA_NOREF,
    /// Arena owning every config string (api_key/base_url/model/prompt…)
    /// across the live config and all superseded ones. Freed at teardown.
    string_arena: std.heap.ArenaAllocator,
    /// The lazily-created Lua tool source (null until first register_tool).
    tools: ?*LuaToolSource,
    closed: bool = false,
    /// A borrowed box wraps an agent owned by the host embedder (see
    /// `_wrap_agent`): teardown is the host's job, `config` is undefined,
    /// and ownership-dependent methods (`set_config`, `register_tool`)
    /// are refused.
    borrowed: bool = false,

    fn deinit(self: *AgentBox, L: *c.lua_State) void {
        if (self.closed) return;
        self.closed = true;
        // Borrowed: the host owns the agent and everything it borrows;
        // this box holds nothing else (empty arena, no store ref, no
        // tools, no Io retain).
        if (self.borrowed) return;
        // Tear down the agent first (it borrows config + store + source).
        self.agent.deinit();
        if (self.tools) |ts| ts.deinit(L);
        // Free the live config + every superseded one. Their string bytes
        // live in `string_arena`, freed in one shot below.
        c_allocator.destroy(self.config);
        for (self.old_configs.items) |cfg| c_allocator.destroy(cfg);
        self.old_configs.deinit(c_allocator);
        self.string_arena.deinit();
        if (self.store_ref != c.LUA_NOREF) {
            c.luaL_unref(L, LUA_REGISTRYINDEX, self.store_ref);
            self.store_ref = c.LUA_NOREF;
        }
        releaseIo();
    }
};

/// Boxed `*Stream` userdata. Holds a Lua reference to the owning Agent
/// userdata so the Agent cannot be GC'd while a live Stream still borrows
/// its conversation + config.
const StreamBox = struct {
    stream: *panto.Stream,
    agent_ref: c_int,
    closed: bool = false,

    fn deinit(self: *StreamBox, L: *c.lua_State) void {
        if (self.closed) return;
        self.closed = true;
        self.stream.deinit();
        if (self.agent_ref != c.LUA_NOREF) {
            c.luaL_unref(L, LUA_REGISTRYINDEX, self.agent_ref);
            self.agent_ref = c.LUA_NOREF;
        }
    }
};

/// Boxed `Conversation` userdata. The same userdata type serves three
/// ownership states, because `libpanto`'s `Conversation` is both a
/// standalone buildable value and the agent's live transparent field:
///
///   - `.owned`: built by `panto.conversation()`; this box owns `inner`
///     and frees it at `__gc`. Can be handed to `panto.agent{conversation=}`,
///     which moves ownership to the agent and flips this box to `.adopted`.
///   - `.adopted`: ownership moved into an `Agent`; the box is inert —
///     `__gc` frees nothing and methods raise. (The agent now owns it.)
///   - `.borrowed`: a live view returned by `agent:conversation()`. It
///     points at the agent's `*Agent.conversation` field; `__gc` frees
///     nothing. A Lua ref pins the owning agent so the view can't dangle.
///
/// `view` is the pointer methods operate through: for `.owned` it is
/// `&self.inner`; for `.borrowed` it is `&agent.conversation`.
const StoreBox = struct {
    impl: union(enum) {
        null_store: panto.NullStore,
        fs_jsonl: panto.FileSystemJSONLStore,
        lua: LuaSessionStore,
    },
    closed: bool = false,

    fn store(self: *StoreBox) panto.SessionStore {
        return switch (self.impl) {
            .null_store => |*s| s.store(),
            .fs_jsonl => |*s| s.store(),
            .lua => |*s| s.store(),
        };
    }

    fn deinit(self: *StoreBox, L: *c.lua_State) void {
        if (self.closed) return;
        self.closed = true;
        switch (self.impl) {
            .null_store => {},
            .fs_jsonl => |*s| s.deinit(),
            .lua => |*s| s.deinit(L),
        }
    }
};

const LuaSessionStore = struct {
    L: *c.lua_State,
    table_ref: c_int,

    fn init(L: *c.lua_State, idx: c_int) LuaSessionStore {
        c.lua_pushvalue(L, idx);
        return .{ .L = L, .table_ref = c.luaL_ref(L, LUA_REGISTRYINDEX) };
    }

    fn deinit(self: *LuaSessionStore, L: *c.lua_State) void {
        if (self.table_ref != c.LUA_NOREF) {
            c.luaL_unref(L, LUA_REGISTRYINDEX, self.table_ref);
            self.table_ref = c.LUA_NOREF;
        }
    }

    fn store(self: *LuaSessionStore) panto.SessionStore {
        return .{ .ptr = self, .vtable = &lua_store_vtable };
    }
};

const ConvBox = struct {
    state: enum { owned, adopted, borrowed }, 
    /// Storage for an owned conversation. Unused (undefined) when borrowed.
    inner: panto.Conversation,
    /// The conversation methods act on. Always valid while usable.
    view: *panto.Conversation,
    /// For `.borrowed`: a ref pinning the owning Agent userdata. NOREF
    /// otherwise.
    agent_ref: c_int = c.LUA_NOREF,

    fn deinit(self: *ConvBox, L: *c.lua_State) void {
        switch (self.state) {
            .owned => self.inner.deinit(),
            .adopted => {}, // the agent owns and frees it
            .borrowed => {
                if (self.agent_ref != c.LUA_NOREF) {
                    c.luaL_unref(L, LUA_REGISTRYINDEX, self.agent_ref);
                    self.agent_ref = c.LUA_NOREF;
                }
            },
        }
    }
};

// ===========================================================================
// Module entry point
// ===========================================================================

/// `luaopen_panto` — the fixed-name init function the Lua loader calls.
/// Builds and returns a **fresh** module table on every call (standard
/// C-module behavior; never a process-shared singleton — the CLI relies on
/// this to safely attach its `ext` field to its own copy).
pub export fn luaopen_panto(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    // Fail loud if loaded into a wrong-version host (the wrong-ABI guard).
    c.luaL_checkversion(L);

    registerMetatables(L);

    c.lua_createtable(L, 0, 6);
    c.lua_pushcclosure(L, agentNew, 0);
    c.lua_setfield(L, -2, "agent");
    c.lua_pushcclosure(L, conversationNew, 0);
    c.lua_setfield(L, -2, "conversation");
    c.lua_pushcclosure(L, nullStoreNew, 0);
    c.lua_setfield(L, -2, "null_store");
    c.lua_pushcclosure(L, fsJSONLStoreNew, 0);
    c.lua_setfield(L, -2, "file_system_jsonl_store");
    // Internal/unstable: wrap a host-owned `*panto.Agent` (lightuserdata)
    // as a *borrowed* Agent userdata. Used by the panto CLI to expose its
    // session agent to extensions (`panto.ext.agent`).
    c.lua_pushcclosure(L, wrapAgent, 0);
    c.lua_setfield(L, -2, "_wrap_agent");
    _ = c.lua_pushstring(L, "libpanto-lua 0.0.0 (Lua 5.4)");
    c.lua_setfield(L, -2, "_VERSION");
    return 1;
}

fn registerMetatables(L: *c.lua_State) void {
    if (c.luaL_newmetatable(L, AGENT_MT) != 0) {
        c.lua_createtable(L, 0, 6); // methods
        setMethod(L, "run", agentRun);
        setMethod(L, "submit", agentSubmit);
        setMethod(L, "register_tool", agentRegisterTool);
        setMethod(L, "set_config", agentSetConfig);
        setMethod(L, "add_system_message", agentAddSystemMessage);
        setMethod(L, "set_system_prompt", agentSetSystemPrompt);
        setMethod(L, "compact", agentCompact);
        setMethod(L, "session_id", agentSessionId);
        setMethod(L, "conversation", agentConversation);
        // Internal/unstable: drive a tool batch directly, bypassing a live
        // provider, so the luv-coroutine dispatch path is testable without
        // a real turn. Not part of the public surface.
        setMethod(L, "_dispatch_tools", agentDispatchTools);
        c.lua_setfield(L, -2, "__index");
        setMethod(L, "__gc", agentGc);
    }
    c.lua_settop(L, c.lua_gettop(L) - 1);

    if (c.luaL_newmetatable(L, STREAM_MT) != 0) {
        c.lua_createtable(L, 0, 3); // methods
        setMethod(L, "next", streamNext);
        setMethod(L, "events", streamEvents);
        setMethod(L, "reopen", streamReopen);
        c.lua_setfield(L, -2, "__index");
        setMethod(L, "__gc", streamGc);
    }
    c.lua_settop(L, c.lua_gettop(L) - 1);

    if (c.luaL_newmetatable(L, STORE_MT) != 0) {
        c.lua_createtable(L, 0, 5); // methods
        setMethod(L, "list", storeList);
        setMethod(L, "resolve", storeResolve);
        setMethod(L, "latest", storeLatest);
        setMethod(L, "load", storeLoad);
        c.lua_setfield(L, -2, "__index");
        setMethod(L, "__gc", storeGc);
    }
    c.lua_settop(L, c.lua_gettop(L) - 1);

    if (c.luaL_newmetatable(L, CONV_MT) != 0) {
        c.lua_createtable(L, 0, 9); // methods
        setMethod(L, "add_system_message", convAddSystemMessage);
        setMethod(L, "replace_system_message", convReplaceSystemMessage);
        setMethod(L, "add_user_message", convAddUserMessage);
        setMethod(L, "add_user_blocks", convAddUserBlocks);
        setMethod(L, "add_assistant_message", convAddAssistantMessage);
        setMethod(L, "add_compaction_summary", convAddCompactionSummary);
        setMethod(L, "messages", convMessages);
        setMethod(L, "len", convLen);
        c.lua_setfield(L, -2, "__index");
        setMethod(L, "__len", convLen);
        setMethod(L, "__gc", convGc);
    }
    c.lua_settop(L, c.lua_gettop(L) - 1);
}

fn setMethod(L: *c.lua_State, name: [:0]const u8, f: c.lua_CFunction) void {
    c.lua_pushcclosure(L, f, 0);
    c.lua_setfield(L, -2, name.ptr);
}

// ===========================================================================
// panto.agent { ... } -> Agent userdata
// ===========================================================================

fn agentNew(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    c.luaL_checktype(L, 1, T_TABLE);

    const ud = c.lua_newuserdatauv(L, @sizeOf(AgentBox), 0) orelse
        return luaErr(L, "panto.agent: out of memory");
    const box: *AgentBox = @ptrCast(@alignCast(ud));
    box.* = .{
        .agent = undefined,
        .config = undefined,
        .old_configs = .empty,
        .store_ref = c.LUA_NOREF,
        .string_arena = std.heap.ArenaAllocator.init(c_allocator),
        .tools = null,
        .closed = false,
    };

    buildAgent(L, box) catch |err| {
        box.string_arena.deinit();
        box.old_configs.deinit(c_allocator);
        return luaErr(L, configErrorMessage(err, "panto.agent"));
    };

    _ = c.luaL_setmetatable(L, AGENT_MT);
    return 1;
}

fn buildAgent(L: *c.lua_State, box: *AgentBox) ConfigError!void {
    const cfg = try c_allocator.create(panto.Config);
    errdefer c_allocator.destroy(cfg);
    cfg.* = try parseConfig(L, 1, box.string_arena.allocator());

    // Optional `conversation = <panto.conversation()>`: adopt it for
    // resumption. The userdata must be an `.owned` Conversation; on
    // success its inner is moved into the agent and the box flips to
    // `.adopted`. We resolve the ConvBox pointer here but only consume it
    // *after* `Agent.init` succeeds, so a failed init leaves the caller's
    // conversation owned and intact.
    var adopt_box: ?*ConvBox = null;
    {
        const t = c.lua_getfield(L, 1, "conversation");
        defer c.lua_settop(L, c.lua_gettop(L) - 1);
        if (t != T_NIL) {
            const ud = c.luaL_testudata(L, -1, CONV_MT) orelse return error.BadField;
            const cb: *ConvBox = @ptrCast(@alignCast(ud));
            if (cb.state != .owned) return error.ConversationNotOwned;
            adopt_box = cb;
        }
    }

    retainIo();
    errdefer releaseIo();

    const store_box = try parseStore(L, 1);
    const session = store_box.store().create();

    box.config = cfg;
    box.agent = panto.Agent.init(
        c_allocator,
        io_ctx.io,
        box.config,
        session,
        if (adopt_box) |cb| cb.inner else null,
    ) catch return error.AgentInit;

    // Pin the store after Agent.init succeeds; the agent/session borrow it.
    c.lua_pushvalue(L, -1);
    box.store_ref = c.luaL_ref(L, LUA_REGISTRYINDEX);

    // Ownership of the inner conversation moved into the agent; neuter the
    // source box so its `__gc` won't double-free.
    if (adopt_box) |cb| {
        cb.state = .adopted;
        cb.view = &box.agent.conversation;
    }
}

/// `panto._wrap_agent(ptr)` — wrap a host-owned `*panto.Agent`, passed as
/// a lightuserdata, into a *borrowed* Agent userdata sharing the AGENT_MT
/// method surface. The host retains ownership: `__gc` tears nothing down,
/// and methods that would entangle box-owned state with the host's agent
/// (`set_config`, `register_tool`) are refused.
fn wrapAgent(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    if (c.lua_type(L, 1) != T_LIGHTUSERDATA)
        return luaErr(L, "panto._wrap_agent: expected an agent pointer (lightuserdata)");
    const ptr = c.lua_touserdata(L, 1) orelse
        return luaErr(L, "panto._wrap_agent: null agent pointer");

    const ud = c.lua_newuserdatauv(L, @sizeOf(AgentBox), 0) orelse
        return luaErr(L, "panto._wrap_agent: out of memory");
    const box: *AgentBox = @ptrCast(@alignCast(ud));
    box.* = .{
        .agent = @ptrCast(@alignCast(ptr)),
        .config = undefined, // never read: set_config is refused on borrowed
        .old_configs = .empty,
        .store_ref = c.LUA_NOREF,
        .string_arena = std.heap.ArenaAllocator.init(c_allocator),
        .tools = null,
        .closed = false,
        .borrowed = true,
    };
    _ = c.luaL_setmetatable(L, AGENT_MT);
    return 1;
}

/// `agent:set_config{...}` — swap the active provider/model/policy. Takes
/// effect at the next turn boundary (libpanto re-reads the snapshot each
/// turn). The new config is parsed with the *same* parser as construction,
/// so the two argument shapes are identical by construction. The old
/// config is retained (not freed) until agent teardown, since an in-flight
/// turn may still read it.
fn agentSetConfig(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const box = checkAgent(L, 1);
    if (box.borrowed) return luaErr(L, "panto: set_config is not supported on a borrowed agent");
    if (box.closed) return luaErr(L, "panto: agent is closed");
    c.luaL_checktype(L, 2, T_TABLE);

    const new_cfg = c_allocator.create(panto.Config) catch
        return luaErr(L, "panto: out of memory");
    new_cfg.* = parseConfig(L, 2, box.string_arena.allocator()) catch |err| {
        c_allocator.destroy(new_cfg);
        return luaErr(L, configErrorMessage(err, "set_config"));
    };

    // Retain the prior config so an in-flight turn keeps a valid snapshot.
    box.old_configs.append(c_allocator, box.config) catch {
        c_allocator.destroy(new_cfg);
        return luaErr(L, "panto: out of memory");
    };
    box.config = new_cfg;
    box.agent.setConfig(new_cfg);
    return 0;
}

fn agentAddSystemMessage(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const box = checkAgent(L, 1);
    if (box.closed) return luaErr(L, "panto: agent is closed");
    var len: usize = 0;
    const ptr = c.luaL_checklstring(L, 2, &len);
    box.agent.addSystemMessage(ptr[0..len]) catch
        return luaErr(L, "panto: failed to add system message");
    return 0;
}

fn agentSetSystemPrompt(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const box = checkAgent(L, 1);
    if (box.closed) return luaErr(L, "panto: agent is closed");
    var len: usize = 0;
    const ptr = c.luaL_checklstring(L, 2, &len);
    box.agent.setSystemPrompt(ptr[0..len]) catch
        return luaErr(L, "panto: failed to set system prompt");
    return 0;
}

/// `agent:compact([override_prompt[, extra_instructions]])` -> table.
/// Returns `{ compacted=, kept_turns=, summarized_messages= }`. Both
/// arguments are optional strings; nil falls back to the configured
/// compaction prompt. Errors (e.g. no compaction prompt configured)
/// surface as a Lua error.
fn agentCompact(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const box = checkAgent(L, 1);
    if (box.closed) return luaErr(L, "panto: agent is closed");

    const override = optArgString(L, 2);
    const extra = optArgString(L, 3);

    const res = box.agent.compact(override, extra) catch |err| {
        return switch (err) {
            error.NoCompactionPrompt => luaErr(L, "panto: no compaction prompt configured (set compaction.prompt in the config)"),
            else => luaErr(L, "panto: compaction failed"),
        };
    };

    c.lua_createtable(L, 0, 3);
    c.lua_pushboolean(L, if (res.compacted) 1 else 0);
    c.lua_setfield(L, -2, "compacted");
    setIntField(L, "kept_turns", @intCast(res.kept_turns));
    setIntField(L, "summarized_messages", @intCast(res.summarized_messages));
    return 1;
}

fn agentSessionId(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const box = checkAgent(L, 1);
    if (box.closed) return luaErr(L, "panto: agent is closed");
    const id = box.agent.sessionId();
    _ = c.lua_pushlstring(L, id.ptr, id.len);
    return 1;
}

/// `agent:conversation()` -> a **borrowed** Conversation view of the
/// agent's live `conversation` field. Mutations and reads go straight
/// through to the agent's conversation (the transparent-field semantics of
/// the Zig API). The view pins the agent (a Lua ref) so it cannot dangle,
/// and frees nothing at `__gc`.
fn agentConversation(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const box = checkAgent(L, 1);
    if (box.closed) return luaErr(L, "panto: agent is closed");

    const ud = c.lua_newuserdatauv(L, @sizeOf(ConvBox), 0) orelse
        return luaErr(L, "panto: out of memory");
    const cb: *ConvBox = @ptrCast(@alignCast(ud));
    cb.* = .{
        .state = .borrowed,
        .inner = undefined,
        .view = &box.agent.conversation,
        .agent_ref = c.LUA_NOREF,
    };
    _ = c.luaL_setmetatable(L, CONV_MT);

    // Pin the agent so the borrowed view can't outlive it.
    c.lua_pushvalue(L, 1);
    cb.agent_ref = c.luaL_ref(L, LUA_REGISTRYINDEX);
    return 1;
}

fn agentRun(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const box = checkAgent(L, 1);
    if (box.closed) return luaErr(L, "panto: agent is closed");

    var len: usize = 0;
    const ptr = c.luaL_checklstring(L, 2, &len);

    // Open the turn from a single user text block; `run` adopts the block.
    const alloc = box.agent.conversation.allocator;
    var blocks = [_]panto.ContentBlock{
        .{ .Text = panto.textualBlockFromSlice(alloc, ptr[0..len]) catch
            return luaErr(L, "panto: out of memory") },
    };
    const stream = box.agent.run(.{ .blocks = &blocks }) catch
        return luaErr(L, "panto: failed to start turn");

    const ud = c.lua_newuserdatauv(L, @sizeOf(StreamBox), 0) orelse {
        stream.deinit();
        return luaErr(L, "panto: out of memory");
    };
    const sbox: *StreamBox = @ptrCast(@alignCast(ud));
    sbox.* = .{ .stream = stream, .agent_ref = c.LUA_NOREF, .closed = false };
    _ = c.luaL_setmetatable(L, STREAM_MT);

    // Pin the agent so it outlives the stream.
    c.lua_pushvalue(L, 1);
    sbox.agent_ref = c.luaL_ref(L, LUA_REGISTRYINDEX);
    return 1;
}

/// `agent:submit(text | { <block>, ... })` — queue user-message blocks for
/// the HOST front end to open the next turn with, instead of opening (and
/// pumping) a turn here like `run` does. Only available on the borrowed
/// host session agent (`panto.ext.agent`): the panto CLI drains the queue
/// once slash-command handling completes and drives the turn through its
/// native renderer (streaming paint, Esc interrupt, auth fallbacks).
/// Accepts a plain string or the same block-array shape as
/// `conv:add_user_blocks`. Multiple calls append to one pending message.
fn agentSubmit(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const box = checkAgent(L, 1);
    if (!box.borrowed)
        return luaErr(L, "panto: submit is only available on the host session agent (panto.ext.agent); use run");
    if (box.closed) return luaErr(L, "panto: agent is closed");

    const alloc = box.agent.conversation.allocator;
    var blocks: std.ArrayList(panto.ContentBlock) = .empty;
    defer blocks.deinit(c_allocator);

    if (c.lua_type(L, 2) == T_STRING) {
        var len: usize = 0;
        const ptr = c.lua_tolstring(L, 2, &len);
        const tb = panto.textualBlockFromSlice(alloc, ptr[0..len]) catch
            return luaErr(L, "panto: out of memory");
        blocks.append(c_allocator, .{ .Text = tb }) catch {
            var b: panto.ContentBlock = .{ .Text = tb };
            b.deinit(alloc);
            return luaErr(L, "panto: out of memory");
        };
    } else {
        c.luaL_checktype(L, 2, T_TABLE);
        collectBlocks(L, alloc, 2, &blocks) catch |e| {
            for (blocks.items) |*b| b.deinit(alloc);
            return luaErr(L, blockErrorMessage(e));
        };
    }

    box.agent.queueSubmission(blocks.items) catch {
        for (blocks.items) |*b| b.deinit(alloc);
        return luaErr(L, "panto: out of memory");
    };
    return 0;
}

fn agentGc(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const box = checkAgent(L, 1);
    box.deinit(L);
    return 0;
}

// ===========================================================================
// panto.conversation() -> Conversation userdata
// ===========================================================================

/// Build a fresh, standalone (`.owned`) `Conversation`. Hand it to
/// `panto.agent { conversation = <conv> }` to resume against a
/// pre-built history; ownership transfers to the agent at that point.
fn conversationNew(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const ud = c.lua_newuserdatauv(L, @sizeOf(ConvBox), 0) orelse
        return luaErr(L, "panto.conversation: out of memory");
    const cb: *ConvBox = @ptrCast(@alignCast(ud));
    cb.* = .{
        .state = .owned,
        .inner = panto.Conversation.init(c_allocator),
        .view = undefined,
        .agent_ref = c.LUA_NOREF,
    };
    cb.view = &cb.inner;
    _ = c.luaL_setmetatable(L, CONV_MT);
    return 1;
}

// ===========================================================================
// panto.null_store() / panto.file_system_jsonl_store{} / custom Lua stores
// ===========================================================================

fn nullStoreNew(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const ud = c.lua_newuserdatauv(L, @sizeOf(StoreBox), 0) orelse
        return luaErr(L, "panto.null_store: out of memory");
    const sb: *StoreBox = @ptrCast(@alignCast(ud));
    sb.* = .{ .impl = .{ .null_store = panto.NullStore.init(c_allocator) } };
    _ = c.luaL_setmetatable(L, STORE_MT);
    return 1;
}

fn fsJSONLStoreNew(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    c.luaL_checktype(L, 1, T_TABLE);
    const dir = requiredString(L, 1, "dir") catch return luaErr(L, "panto.file_system_jsonl_store: missing dir");
    const metadata = optString(L, 1, "metadata") catch return luaErr(L, "panto.file_system_jsonl_store: bad metadata");
    const store = panto.FileSystemJSONLStore.initWithMetadata(c_allocator, io_ctx.io, dir, metadata) catch
        return luaErr(L, "panto.file_system_jsonl_store: failed to initialize store");

    const ud = c.lua_newuserdatauv(L, @sizeOf(StoreBox), 0) orelse {
        var s = store;
        s.deinit();
        return luaErr(L, "panto.file_system_jsonl_store: out of memory");
    };
    const sb: *StoreBox = @ptrCast(@alignCast(ud));
    sb.* = .{ .impl = .{ .fs_jsonl = store } };
    _ = c.luaL_setmetatable(L, STORE_MT);
    return 1;
}

fn parseStore(L: *c.lua_State, agent_tbl: c_int) ConfigError!*StoreBox {
    const t = c.lua_getfield(L, agent_tbl, "store");
    if (t == T_NIL) {
        c.lua_settop(L, c.lua_gettop(L) - 1);
        if (nullStoreNew(L) != 1) return error.OutOfMemory;
        return checkStore(L, -1);
    }
    if (c.luaL_testudata(L, -1, STORE_MT)) |ud| return @ptrCast(@alignCast(ud));
    if (t != T_TABLE) return error.BadField;

    const ud = c.lua_newuserdatauv(L, @sizeOf(StoreBox), 0) orelse return error.OutOfMemory;
    const sb: *StoreBox = @ptrCast(@alignCast(ud));
    sb.* = .{ .impl = .{ .lua = LuaSessionStore.init(L, -2) } };
    _ = c.luaL_setmetatable(L, STORE_MT);
    return sb;
}

fn checkStore(L: *c.lua_State, idx: c_int) *StoreBox {
    return @ptrCast(@alignCast(c.luaL_checkudata(L, idx, STORE_MT)));
}

fn storeGc(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    checkStore(L, 1).deinit(L);
    return 0;
}

fn storeList(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const sb = checkStore(L, 1);
    const infos = sb.store().list() catch return luaErr(L, "panto: store list failed");
    defer sb.store().freeSessionInfos(infos);
    pushSessionInfoList(L, infos);
    return 1;
}

fn storeResolve(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const sb = checkStore(L, 1);
    var len: usize = 0;
    const ptr = c.luaL_checklstring(L, 2, &len);
    const s = sb.store().resolve(ptr[0..len]) catch return luaErr(L, "panto: store resolve failed");
    if (s) |sess| pushSessionInfo(L, sess.info) else c.lua_pushnil(L);
    return 1;
}

fn storeLatest(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const sb = checkStore(L, 1);
    const s = sb.store().latest() catch return luaErr(L, "panto: store latest failed");
    if (s) |sess| pushSessionInfo(L, sess.info) else c.lua_pushnil(L);
    return 1;
}

fn storeLoad(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const sb = checkStore(L, 1);
    var len: usize = 0;
    const ptr = c.luaL_checklstring(L, 2, &len);
    const conv = (sb.store().load(ptr[0..len]) catch return luaErr(L, "panto: store load failed")) orelse {
        c.lua_pushnil(L);
        return 1;
    };
    pushConversationOwned(L, conv);
    return 1;
}

fn pushConversationOwned(L: *c.lua_State, conv: panto.Conversation) void {
    const ud = c.lua_newuserdatauv(L, @sizeOf(ConvBox), 0).?;
    const cb: *ConvBox = @ptrCast(@alignCast(ud));
    cb.* = .{ .state = .owned, .inner = conv, .view = undefined, .agent_ref = c.LUA_NOREF };
    cb.view = &cb.inner;
    _ = c.luaL_setmetatable(L, CONV_MT);
}

fn pushSessionInfoList(L: *c.lua_State, infos: []panto.SessionInfo) void {
    c.lua_createtable(L, @intCast(infos.len), 0);
    for (infos, 0..) |info, i| {
        pushSessionInfo(L, info);
        c.lua_rawseti(L, -2, @intCast(i + 1));
    }
}

fn pushSessionInfo(L: *c.lua_State, info: panto.SessionInfo) void {
    c.lua_createtable(L, 0, 9);
    setStringField(L, "id", info.id);
    setStringField(L, "created", info.created);
    setStringField(L, "modified", info.modified);
    setIntField(L, "message_count", @intCast(info.message_count));
    setStringField(L, "last_user_message", info.last_user_message);
    setStringField(L, "api_style", @tagName(info.api_style));
    setStringField(L, "base_url", info.base_url);
    setStringField(L, "model", info.model);
    setStringField(L, "reasoning", @tagName(info.reasoning));
}

fn checkConv(L: *c.lua_State, idx: c_int) *ConvBox {
    return @ptrCast(@alignCast(c.luaL_checkudata(L, idx, CONV_MT)));
}

/// Resolve the usable `*Conversation` for a method, or raise if the box was
/// adopted (the agent owns it now — operate through `agent:conversation()`).
fn convView(L: *c.lua_State, cb: *ConvBox) ?*panto.Conversation {
    if (cb.state == .adopted) {
        _ = luaErr(L, "panto: this conversation was adopted by an agent; use agent:conversation() to operate on it");
        return null;
    }
    return cb.view;
}

fn convAddSystemMessage(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const cb = checkConv(L, 1);
    const conv = convView(L, cb) orelse return 0;
    var len: usize = 0;
    const ptr = c.luaL_checklstring(L, 2, &len);
    conv.addSystemMessage(ptr[0..len]) catch return luaErr(L, "panto: out of memory");
    return 0;
}

fn convReplaceSystemMessage(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const cb = checkConv(L, 1);
    const conv = convView(L, cb) orelse return 0;
    var len: usize = 0;
    const ptr = c.luaL_checklstring(L, 2, &len);
    conv.replaceSystemMessage(ptr[0..len]) catch return luaErr(L, "panto: out of memory");
    return 0;
}

/// `conv:add_user_message(text)` — the ergonomic single-text case. The
/// general block-slice form is `conv:add_user_blocks{...}`.
fn convAddUserMessage(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const cb = checkConv(L, 1);
    const conv = convView(L, cb) orelse return 0;
    var len: usize = 0;
    const ptr = c.luaL_checklstring(L, 2, &len);
    const tb = panto.textualBlockFromSlice(conv.allocator, ptr[0..len]) catch
        return luaErr(L, "panto: out of memory");
    var block: panto.ContentBlock = .{ .Text = tb };
    conv.addUserMessage(&.{block}) catch {
        block.deinit(conv.allocator);
        return luaErr(L, "panto: out of memory");
    };
    return 0;
}

fn convAddCompactionSummary(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const cb = checkConv(L, 1);
    const conv = convView(L, cb) orelse return 0;
    var len: usize = 0;
    const ptr = c.luaL_checklstring(L, 2, &len);
    conv.addCompactionSummary(ptr[0..len]) catch return luaErr(L, "panto: out of memory");
    return 0;
}

/// `conv:add_user_blocks{ <block>, ... }` — the general user-message
/// builder, symmetric with `add_assistant_message`. Each block is a table
/// `{ type = "text"|"tool_result", ... }`. Lets a Lua app reconstruct full
/// user turns from serialized state: plain/queued text plus tool results.
fn convAddUserBlocks(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const cb = checkConv(L, 1);
    const conv = convView(L, cb) orelse return 0;
    c.luaL_checktype(L, 2, T_TABLE);
    addMessageFromBlocks(L, conv, 2, .user) catch |e| return luaErr(L, blockErrorMessage(e));
    return 0;
}

/// `conv:add_assistant_message{ blocks = { <block>, ... }, usage = {...} }`
/// — reconstruct an assistant turn: `text`, `thinking`, and `tool_use`
/// blocks, plus optional provider `usage`. Together with `add_user_blocks`
/// this makes a serialize→rebuild cycle lossless for tool-using turns.
fn convAddAssistantMessage(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const cb = checkConv(L, 1);
    const conv = convView(L, cb) orelse return 0;
    c.luaL_checktype(L, 2, T_TABLE);

    // Accept either `{ blocks = {...}, usage = ? }` or a bare block array.
    const bt = c.lua_getfield(L, 2, "blocks");
    const blocks_idx: c_int = if (bt == T_TABLE) c.lua_gettop(L) else blk: {
        c.lua_settop(L, c.lua_gettop(L) - 1);
        break :blk 2;
    };
    const has_wrapper = bt == T_TABLE;
    defer if (has_wrapper) c.lua_settop(L, c.lua_gettop(L) - 1);

    const usage = parseUsage(L, 2) catch |e| return luaErr(L, blockErrorMessage(e));
    addAssistantFromBlocks(L, conv, blocks_idx, usage) catch |e| return luaErr(L, blockErrorMessage(e));
    return 0;
}

const BlockError = error{ BadBlock, UnknownBlockType, OutOfMemory };

fn blockErrorMessage(e: BlockError) [:0]const u8 {
    return switch (e) {
        error.BadBlock => "panto: malformed content block (check field names/types)",
        error.UnknownBlockType => "panto: unknown block type (want text/thinking/tool_use/tool_result)",
        error.OutOfMemory => "panto: out of memory",
    };
}

/// Build a `[]ContentBlock` from the Lua array at `arr_idx`, then hand it
/// to the conversation (user or compaction-style role via addUserMessage).
fn addMessageFromBlocks(L: *c.lua_State, conv: *panto.Conversation, arr_idx: c_int, comptime role: enum { user }) BlockError!void {
    _ = role;
    var blocks: std.ArrayList(panto.ContentBlock) = .empty;
    defer blocks.deinit(c_allocator);
    errdefer for (blocks.items) |*b| b.deinit(conv.allocator);

    try collectBlocks(L, conv.allocator, arr_idx, &blocks);
    conv.addUserMessage(blocks.items) catch return error.OutOfMemory;
    // Ownership transferred; clear so errdefer/deinit don't double-free.
    blocks.clearRetainingCapacity();
}

fn addAssistantFromBlocks(L: *c.lua_State, conv: *panto.Conversation, arr_idx: c_int, usage: ?panto.Usage) BlockError!void {
    var blocks: std.ArrayList(panto.ContentBlock) = .empty;
    defer blocks.deinit(c_allocator);
    errdefer for (blocks.items) |*b| b.deinit(conv.allocator);

    try collectBlocks(L, conv.allocator, arr_idx, &blocks);
    conv.addAssistantMessage(blocks.items, usage) catch return error.OutOfMemory;
    blocks.clearRetainingCapacity();
}

/// Walk the Lua array at `arr_idx`, append one `ContentBlock` per entry to
/// `out`. Bytes are owned by `alloc` (the conversation's allocator). On
/// error the caller frees whatever was appended.
fn collectBlocks(L: *c.lua_State, alloc: std.mem.Allocator, arr_idx: c_int, out: *std.ArrayList(panto.ContentBlock)) BlockError!void {
    const abs = c.lua_absindex(L, arr_idx);
    const n: usize = @intCast(c.lua_rawlen(L, abs));
    var i: usize = 1;
    while (i <= n) : (i += 1) {
        _ = c.lua_rawgeti(L, abs, @intCast(i));
        defer c.lua_settop(L, c.lua_gettop(L) - 1);
        if (c.lua_type(L, -1) != T_TABLE) return error.BadBlock;
        const block = try buildContentBlock(L, alloc, c.lua_gettop(L));
        out.append(c_allocator, block) catch {
            var b = block;
            b.deinit(alloc);
            return error.OutOfMemory;
        };
    }
}

/// Construct a single `ContentBlock` from the table at `idx`. Every owned
/// byte is allocated with `alloc` so the conversation can free it.
fn buildContentBlock(L: *c.lua_State, alloc: std.mem.Allocator, idx: c_int) BlockError!panto.ContentBlock {
    const type_str = blockField(L, idx, "type") orelse return error.BadBlock;

    if (std.mem.eql(u8, type_str, "text")) {
        const text = blockField(L, idx, "text") orelse return error.BadBlock;
        const tb = panto.textualBlockFromSlice(alloc, text) catch return error.OutOfMemory;
        return .{ .Text = tb };
    } else if (std.mem.eql(u8, type_str, "thinking")) {
        const text = blockField(L, idx, "text") orelse "";
        var tb = panto.textualBlockFromSlice(alloc, text) catch return error.OutOfMemory;
        errdefer tb.deinit(alloc);
        const sig: ?[]const u8 = if (blockField(L, idx, "signature")) |s|
            (alloc.dupe(u8, s) catch return error.OutOfMemory)
        else
            null;
        return .{ .Thinking = .{ .text = tb, .signature = sig } };
    } else if (std.mem.eql(u8, type_str, "tool_use")) {
        const id = blockField(L, idx, "id") orelse return error.BadBlock;
        const name = blockField(L, idx, "name") orelse return error.BadBlock;
        const input = blockField(L, idx, "input") orelse "";
        const id_owned = alloc.dupe(u8, id) catch return error.OutOfMemory;
        errdefer alloc.free(id_owned);
        const name_owned = alloc.dupe(u8, name) catch return error.OutOfMemory;
        errdefer alloc.free(name_owned);
        const input_tb = panto.textualBlockFromSlice(alloc, input) catch return error.OutOfMemory;
        return .{ .ToolUse = .{ .id = id_owned, .name = name_owned, .input = input_tb } };
    } else if (std.mem.eql(u8, type_str, "tool_result")) {
        return buildToolResultBlock(L, alloc, idx);
    }
    return error.UnknownBlockType;
}

fn buildToolResultBlock(L: *c.lua_State, alloc: std.mem.Allocator, idx: c_int) BlockError!panto.ContentBlock {
    const tuid = blockField(L, idx, "tool_use_id") orelse return error.BadBlock;
    const tuid_owned = alloc.dupe(u8, tuid) catch return error.OutOfMemory;
    errdefer alloc.free(tuid_owned);

    const is_error: bool = blk: {
        const t = c.lua_getfield(L, idx, "is_error");
        defer c.lua_settop(L, c.lua_gettop(L) - 1);
        break :blk (t == T_BOOLEAN and c.lua_toboolean(L, -1) != 0);
    };

    var parts: std.ArrayList(panto.ResultPartStored) = .empty;
    errdefer {
        for (parts.items) |*p| p.deinit(alloc);
        parts.deinit(alloc);
    }

    const pt = c.lua_getfield(L, idx, "parts");
    defer c.lua_settop(L, c.lua_gettop(L) - 1);
    if (pt == T_TABLE) {
        const pn: usize = @intCast(c.lua_rawlen(L, -1));
        var i: usize = 1;
        while (i <= pn) : (i += 1) {
            _ = c.lua_rawgeti(L, -1, @intCast(i));
            defer c.lua_settop(L, c.lua_gettop(L) - 1);
            if (c.lua_type(L, -1) != T_TABLE) return error.BadBlock;
            const part = try buildStoredPart(L, alloc, c.lua_gettop(L));
            parts.append(alloc, part) catch {
                var p = part;
                p.deinit(alloc);
                return error.OutOfMemory;
            };
        }
    } else if (pt != T_NIL) {
        return error.BadBlock;
    }

    return .{ .ToolResult = .{ .tool_use_id = tuid_owned, .parts = parts, .is_error = is_error } };
}

/// A stored result part: `{ text = "..." }` or `{ media_type = "...",
/// data = "<base64>" }`. Media `data` is the already-encoded bytes (this
/// is resumption of *stored* state, not raw tool output).
fn buildStoredPart(L: *c.lua_State, alloc: std.mem.Allocator, idx: c_int) BlockError!panto.ResultPartStored {
    if (blockField(L, idx, "text")) |t| {
        const tb = panto.textualBlockFromSlice(alloc, t) catch return error.OutOfMemory;
        return .{ .text = tb };
    }
    const media_type = blockField(L, idx, "media_type") orelse return error.BadBlock;
    const data = blockField(L, idx, "data") orelse return error.BadBlock;
    const mt_owned = alloc.dupe(u8, media_type) catch return error.OutOfMemory;
    errdefer alloc.free(mt_owned);
    const data_tb = panto.textualBlockFromSlice(alloc, data) catch return error.OutOfMemory;
    return .{ .media = .{ .media_type = mt_owned, .data = data_tb } };
}

/// Read string field `name` from the table at `idx`, returning a borrowed
/// slice valid until the field value is popped. We pop immediately (Lua
/// keeps the string alive as long as it's referenced by the table), so the
/// returned slice stays valid for the synchronous dupe that follows.
fn blockField(L: *c.lua_State, idx: c_int, name: [*:0]const u8) ?[]const u8 {
    const t = c.lua_getfield(L, idx, name);
    defer c.lua_settop(L, c.lua_gettop(L) - 1);
    if (t != T_STRING) return null;
    var len: usize = 0;
    const ptr = c.lua_tolstring(L, -1, &len);
    if (ptr == null) return null;
    return ptr[0..len];
}

fn parseUsage(L: *c.lua_State, tbl_idx: c_int) BlockError!?panto.Usage {
    const t = c.lua_getfield(L, tbl_idx, "usage");
    defer c.lua_settop(L, c.lua_gettop(L) - 1);
    if (t == T_NIL) return null;
    if (t != T_TABLE) return error.BadBlock;
    const sub = c.lua_gettop(L);
    return panto.Usage{
        .input = usageField(L, sub, "input"),
        .output = usageField(L, sub, "output"),
        .cache_read = usageField(L, sub, "cache_read"),
        .cache_write = usageField(L, sub, "cache_write"),
        .reasoning = usageField(L, sub, "reasoning"),
    };
}

fn usageField(L: *c.lua_State, tbl_idx: c_int, name: [*:0]const u8) u64 {
    const t = c.lua_getfield(L, tbl_idx, name);
    defer c.lua_settop(L, c.lua_gettop(L) - 1);
    if (t != T_NUMBER) return 0;
    const v = c.lua_tointegerx(L, -1, null);
    return if (v < 0) 0 else @intCast(v);
}

fn convLen(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const cb = checkConv(L, 1);
    const conv = convView(L, cb) orelse return 0;
    c.lua_pushinteger(L, @intCast(conv.messages.items.len));
    return 1;
}

/// `conv:messages()` -> array of `{ role=, blocks={ {type=, ...} } }`.
/// A read-only inspection snapshot: bytes are copied into Lua, so the
/// returned tables are independent of later conversation mutation.
fn convMessages(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const cb = checkConv(L, 1);
    const conv = convView(L, cb) orelse return 0;

    const msgs = conv.messages.items;
    c.lua_createtable(L, @intCast(msgs.len), 0);
    for (msgs, 0..) |msg, mi| {
        c.lua_createtable(L, 0, 3);
        setStringField(L, "role", roleName(msg.role));
        if (msg.usage) |u| pushUsage(L, u); // emits a `usage` field
        // blocks = { ... } — faithful per-block detail for lossless
        // round-trip: each block table is shaped to be fed straight back
        // into add_user_blocks / add_assistant_message.
        c.lua_createtable(L, @intCast(msg.content.items.len), 0);
        for (msg.content.items, 0..) |block, bi| {
            pushInspectBlock(L, block);
            c.lua_rawseti(L, -2, @intCast(bi + 1));
        }
        c.lua_setfield(L, -2, "blocks");
        c.lua_rawseti(L, -2, @intCast(mi + 1));
    }
    return 1;
}

/// Push a fully-detailed block table (leaves it on the stack top). The
/// shape mirrors exactly what `add_user_blocks`/`add_assistant_message`
/// accept, so `conv:messages()` output can be replayed verbatim. The block
/// `type` strings here match `buildContentBlock`'s expectations.
fn pushInspectBlock(L: *c.lua_State, block: panto.ContentBlock) void {
    c.lua_createtable(L, 0, 4);
    switch (block) {
        .Text => |t| {
            setStringField(L, "type", "text");
            setStringField(L, "text", t.items);
        },
        .Thinking => |t| {
            setStringField(L, "type", "thinking");
            setStringField(L, "text", t.text.items);
            if (t.signature) |s| setStringField(L, "signature", s);
        },
        .ToolUse => |tu| {
            setStringField(L, "type", "tool_use");
            setStringField(L, "id", tu.id);
            setStringField(L, "name", tu.name);
            setStringField(L, "input", tu.input.items);
        },
        .ToolResult => |tr| {
            setStringField(L, "type", "tool_result");
            setStringField(L, "tool_use_id", tr.tool_use_id);
            c.lua_pushboolean(L, if (tr.is_error) 1 else 0);
            c.lua_setfield(L, -2, "is_error");
            c.lua_createtable(L, @intCast(tr.parts.items.len), 0);
            for (tr.parts.items, 0..) |part, pi| {
                c.lua_createtable(L, 0, 2);
                switch (part) {
                    .text => |t| setStringField(L, "text", t.items),
                    .media => |m| {
                        setStringField(L, "media_type", m.media_type);
                        setStringField(L, "data", m.data.items);
                    },
                }
                c.lua_rawseti(L, -2, @intCast(pi + 1));
            }
            c.lua_setfield(L, -2, "parts");
        },
        .System => |s| {
            setStringField(L, "type", "system");
            setStringField(L, "text", s.text.items);
            setStringField(L, "mode", switch (s.mode) {
                .append => "append",
                .replace => "replace",
            });
        },
        .CompactionSummary => |s| {
            setStringField(L, "type", "compaction_summary");
            setStringField(L, "text", s.text.items);
        },
    }
}

fn convGc(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const cb = checkConv(L, 1);
    cb.deinit(L);
    return 0;
}

// ===========================================================================
// Config parsing (shared by panto.agent{} and agent:set_config{})
// ===========================================================================

const ConfigError = error{
    MissingField,
    BadField,
    UnknownApiStyle,
    BadEnum,
    ConversationNotOwned,
    OutOfMemory,
    AgentInit,
};

fn configErrorMessage(err: ConfigError, comptime ctx: [:0]const u8) [:0]const u8 {
    return switch (err) {
        error.MissingField => ctx ++ ": missing required field (need api_style, api_key, base_url, model)",
        error.BadField => ctx ++ ": a field has the wrong type",
        error.UnknownApiStyle => ctx ++ ": api_style must be 'openai_chat' or 'anthropic_messages'",
        error.BadEnum => ctx ++ ": an enum field has an unrecognized value",
        error.ConversationNotOwned => ctx ++ ": the `conversation` was already adopted by an agent or is a borrowed view; build a fresh one with panto.conversation()",
        error.OutOfMemory => ctx ++ ": out of memory",
        error.AgentInit => ctx ++ ": failed to initialize agent",
    };
}

/// Parse a full `panto.Config` from the named-args table at `tbl_idx`.
/// Every string is duped into `arena` (owned by the AgentBox for the
/// config's lifetime). This is the single source of truth for both
/// construction and `set_config`, so the two argument shapes never drift.
///
/// Provider fields (all required): `api_style`, `api_key`, `base_url`,
/// `model`. Common optional: `max_tokens`.
/// openai_chat optional: `reasoning`.
/// anthropic_messages optional: `api_version`, `thinking`, `effort`,
///   `thinking_budget_tokens`, `thinking_interleaved`.
/// Optional `compaction = { keep_verbatim=, prompt= }`.
/// Optional `retry = { max_attempts=, initial_delay_ms=, max_delay_ms=,
///   multiplier=, jitter= }`.
fn parseConfig(L: *c.lua_State, tbl_idx: c_int, arena: std.mem.Allocator) ConfigError!panto.Config {
    const api_style = try requiredString(L, tbl_idx, "api_style");
    const api_key = try dupArena(arena, try requiredString(L, tbl_idx, "api_key"));
    const base_url = try dupArena(arena, try requiredString(L, tbl_idx, "base_url"));
    const model = try dupArena(arena, try requiredString(L, tbl_idx, "model"));
    const max_tokens = try optU32(L, tbl_idx, "max_tokens");

    var provider: panto.ProviderConfig = undefined;
    if (std.mem.eql(u8, api_style, "openai_chat")) {
        var p: panto.OpenAIChatConfig = .{ .api_key = api_key, .base_url = base_url, .model = model };
        if (max_tokens) |mt| p.max_tokens = mt;
        if (try optString(L, tbl_idx, "reasoning")) |r|
            p.reasoning = parseEnum(panto.ReasoningEffort, r) orelse return error.BadEnum;
        provider = .{ .openai_chat = p };
    } else if (std.mem.eql(u8, api_style, "anthropic_messages")) {
        var p: panto.AnthropicMessagesConfig = .{ .api_key = api_key, .base_url = base_url, .model = model };
        if (max_tokens) |mt| p.max_tokens = mt;
        if (try optString(L, tbl_idx, "api_version")) |v| p.api_version = try dupArena(arena, v);
        if (try optString(L, tbl_idx, "thinking")) |t|
            p.thinking = parseEnum(panto.Thinking, t) orelse return error.BadEnum;
        if (try optString(L, tbl_idx, "effort")) |e|
            p.effort = parseEnum(panto.Effort, e) orelse return error.BadEnum;
        if (try optU32(L, tbl_idx, "thinking_budget_tokens")) |b| p.thinking_budget_tokens = b;
        if (try optBool(L, tbl_idx, "thinking_interleaved")) |i| p.thinking_interleaved = i;
        provider = .{ .anthropic_messages = p };
    } else {
        return error.UnknownApiStyle;
    }

    var config: panto.Config = .{ .provider = provider };
    try parseCompaction(L, tbl_idx, arena, &config.compaction);
    try parseRetry(L, tbl_idx, &config.retry);
    return config;
}

fn parseCompaction(
    L: *c.lua_State,
    tbl_idx: c_int,
    arena: std.mem.Allocator,
    out: *panto.CompactionConfig,
) ConfigError!void {
    const t = c.lua_getfield(L, tbl_idx, "compaction");
    defer c.lua_settop(L, c.lua_gettop(L) - 1);
    if (t == T_NIL) return;
    if (t != T_TABLE) return error.BadField;
    const sub = c.lua_gettop(L);
    if (try optU32(L, sub, "keep_verbatim")) |k| out.keep_verbatim = k;
    if (try optString(L, sub, "prompt")) |p| out.compaction_prompt = try dupArena(arena, p);
    // Note: a nested compaction `model` override is intentionally omitted
    // for v1 — it would require recursively parsing a second provider
    // block; the active model is used for compaction unless added later.
}

fn parseRetry(L: *c.lua_State, tbl_idx: c_int, out: *panto.RetryConfig) ConfigError!void {
    const t = c.lua_getfield(L, tbl_idx, "retry");
    defer c.lua_settop(L, c.lua_gettop(L) - 1);
    if (t == T_NIL) return;
    if (t != T_TABLE) return error.BadField;
    const sub = c.lua_gettop(L);
    if (try optU32(L, sub, "max_attempts")) |v| out.max_attempts = v;
    if (try optU64(L, sub, "initial_delay_ms")) |v| out.initial_delay_ms = v;
    if (try optU64(L, sub, "max_delay_ms")) |v| out.max_delay_ms = v;
    if (try optNumber(L, sub, "multiplier")) |v| out.multiplier = v;
    if (try optBool(L, sub, "jitter")) |v| out.jitter = v;
}

fn parseEnum(comptime E: type, s: []const u8) ?E {
    inline for (@typeInfo(E).@"enum".fields) |f| {
        if (std.mem.eql(u8, s, f.name)) return @enumFromInt(f.value);
    }
    return null;
}

fn dupArena(arena: std.mem.Allocator, s: []const u8) ConfigError![]const u8 {
    return arena.dupe(u8, s) catch error.OutOfMemory;
}

// ===========================================================================
// Lua tool source: luv-driven cooperative coroutines
// ===========================================================================
//
// Ported from the CLI's `src/lua_runtime.zig` scheduler, operating on the
// host `lua_State` passed to `luaopen_panto`. Each registered tool stores
// its handler in the Lua registry (`luaL_ref`). On `invoke_batch` we run
// each call as a coroutine through a wrapper that `pcall`s the handler and
// reports the result via a registered `_record_result` C closure, then
// drive `uv.run("default")` to completion. Synchronous handlers finish on
// their first resume; async handlers yield on a libuv op and are resumed
// by its callback.

/// One coroutine's outcome, written by `recordResult` and read after the
/// event loop drains.
const Slot = struct {
    recorded: bool = false,
    ok: bool = false,
    value: ?panto.ResultParts = null,
    err_msg: ?[]u8 = null,
};

const BatchState = struct {
    allocator: std.mem.Allocator,
    slots: []Slot,
};

const LuaToolSource = struct {
    L: *c.lua_State,
    /// Tool decls handed to libpanto (borrowed strings live in `arena`).
    decls: std.ArrayList(panto.ToolDecl),
    /// name -> handler registry ref.
    handlers: std.StringHashMap(c_int),
    /// Owns every decl string + handler-name key.
    arena: std.heap.ArenaAllocator,
    /// Wrapper closure ref: pcall(handler,input) -> _record_result(...).
    wrapper_ref: c_int = c.LUA_NOREF,
    /// Cached `require("luv").run`.
    uv_run_ref: c_int = c.LUA_NOREF,
    /// In-flight batch, valid only during one `invoke_batch`.
    current_batch: ?*BatchState = null,
    /// True once the source has been handed to the agent.
    registered: bool = false,

    fn create(L: *c.lua_State) !*LuaToolSource {
        const self = try c_allocator.create(LuaToolSource);
        self.* = .{
            .L = L,
            .decls = .empty,
            .handlers = std.StringHashMap(c_int).init(c_allocator),
            .arena = std.heap.ArenaAllocator.init(c_allocator),
        };
        return self;
    }

    fn deinit(self: *LuaToolSource, L: *c.lua_State) void {
        var it = self.handlers.iterator();
        while (it.next()) |e| c.luaL_unref(L, LUA_REGISTRYINDEX, e.value_ptr.*);
        self.handlers.deinit();
        if (self.wrapper_ref != c.LUA_NOREF) c.luaL_unref(L, LUA_REGISTRYINDEX, self.wrapper_ref);
        if (self.uv_run_ref != c.LUA_NOREF) c.luaL_unref(L, LUA_REGISTRYINDEX, self.uv_run_ref);
        self.decls.deinit(c_allocator);
        self.arena.deinit();
        c_allocator.destroy(self);
    }

    fn toolSource(self: *LuaToolSource) panto.ToolSource {
        return .{
            .name = SOURCE_NAME,
            .tools = self.decls.items,
            .ctx = self,
            .vtable = &source_vtable,
        };
    }

    /// Lazily build the wrapper closure + cache `uv.run`. Requires `luv`
    /// to be `require`-able in the host (a hard package dependency).
    fn ensureScheduler(self: *LuaToolSource) !void {
        if (self.wrapper_ref != c.LUA_NOREF) return;
        const L = self.L;

        // Register `_record_result` as a C closure carrying `self`, and
        // build the wrapper that closes over it. We keep them in a private
        // table referenced only by the wrapper, so we never touch the
        // module table (which is the host's, possibly augmented, copy).
        const snippet =
            \\local record = ...
            \\return function(idx, handler, input)
            \\  local ok, val = pcall(handler, input)
            \\  record(idx, ok, val)
            \\end
        ;
        if (c.luaL_loadstring(L, snippet) != 0) {
            c.lua_settop(L, c.lua_gettop(L) - 1);
            return error.LuaInitFailed;
        }
        c.lua_pushlightuserdata(L, @ptrCast(self));
        c.lua_pushcclosure(L, recordResultC, 1); // the `record` arg
        if (c.lua_pcallk(L, 1, 1, 0, 0, null) != 0) {
            c.lua_settop(L, c.lua_gettop(L) - 1);
            return error.LuaInitFailed;
        }
        self.wrapper_ref = c.luaL_ref(L, LUA_REGISTRYINDEX);

        // Cache uv.run.
        const uv_snippet =
            \\return require("luv").run
        ;
        if (c.luaL_loadstring(L, uv_snippet) != 0) {
            c.lua_settop(L, c.lua_gettop(L) - 1);
            return error.LuaInitFailed;
        }
        if (c.lua_pcallk(L, 0, 1, 0, 0, null) != 0) {
            c.lua_settop(L, c.lua_gettop(L) - 1);
            return error.LuvMissing;
        }
        if (c.lua_type(L, -1) != T_FUNCTION) {
            c.lua_settop(L, c.lua_gettop(L) - 1);
            return error.LuvMissing;
        }
        self.uv_run_ref = c.luaL_ref(L, LUA_REGISTRYINDEX);
    }
};

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

/// The source's `ctx` is owned by the AgentBox (freed in `AgentBox.deinit`
/// after `agent.deinit`), so libpanto's source teardown is a no-op.
fn deinitSrc(_: *anyopaque, _: std.mem.Allocator) void {}

/// `agent:register_tool { name=, description=, schema=, handler= }`.
/// Validates the named-args table, serializes the schema to JSON, refs the
/// handler, and appends a decl. The first call creates the source and
/// hands it to the agent; later calls extend the source (visible at the
/// next turn boundary, per libpanto's registry semantics).
fn agentRegisterTool(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const box = checkAgent(L, 1);
    if (box.closed) return luaErr(L, "panto: agent is closed");
    if (box.borrowed) return luaErr(L, "panto: register_tool is not supported on a borrowed agent (use panto.ext.register_tool)");
    c.luaL_checktype(L, 2, T_TABLE);

    if (box.tools == null) {
        box.tools = LuaToolSource.create(L) catch
            return luaErr(L, "panto: out of memory");
    }
    const ts = box.tools.?;

    // Make sure luv + the wrapper are ready before we accept any tool, so
    // a missing `luv` fails at registration (clear) not mid-turn.
    ts.ensureScheduler() catch |err| {
        return switch (err) {
            error.LuvMissing => luaErr(L, "panto: require('luv') failed — libpanto-lua needs the 'luv' rock for tool dispatch"),
            else => luaErr(L, "panto: failed to initialize the tool scheduler"),
        };
    };

    addTool(L, ts, 2) catch |err| {
        return luaErr(L, configErrorMessage2(err));
    };

    // Register the source with the agent on first tool.
    if (!ts.registered) {
        box.agent.registerToolSource(ts.toolSource()) catch
            return luaErr(L, "panto: failed to register tool source");
        ts.registered = true;
    }
    return 0;
}

const ToolError = error{ MissingField, BadField, OutOfMemory };

fn configErrorMessage2(err: ToolError) [:0]const u8 {
    return switch (err) {
        error.MissingField => "register_tool: need name (string), description (string), schema (table), handler (function)",
        error.BadField => "register_tool: a field has the wrong type",
        error.OutOfMemory => "register_tool: out of memory",
    };
}

fn addTool(L: *c.lua_State, ts: *LuaToolSource, tbl_idx: c_int) ToolError!void {
    const arena = ts.arena.allocator();

    const name = try arenaString(L, tbl_idx, "name", arena);
    const description = try arenaString(L, tbl_idx, "description", arena);

    // schema (table) -> JSON.
    const sty = c.lua_getfield(L, tbl_idx, "schema");
    if (sty != T_TABLE) {
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return error.BadField;
    }
    const schema_json = serializeSchema(L, c.lua_gettop(L), arena) catch {
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return error.OutOfMemory;
    };
    c.lua_settop(L, c.lua_gettop(L) - 1); // pop schema

    // handler (function) -> registry ref. luaL_ref pops it.
    const hty = c.lua_getfield(L, tbl_idx, "handler");
    if (hty != T_FUNCTION) {
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return error.BadField;
    }
    const handler_ref = c.luaL_ref(L, LUA_REGISTRYINDEX);

    ts.handlers.put(name, handler_ref) catch {
        c.luaL_unref(L, LUA_REGISTRYINDEX, handler_ref);
        return error.OutOfMemory;
    };
    ts.decls.append(c_allocator, .{
        .name = name,
        .description = description,
        .schema_json = schema_json,
    }) catch {
        _ = ts.handlers.remove(name);
        c.luaL_unref(L, LUA_REGISTRYINDEX, handler_ref);
        return error.OutOfMemory;
    };
}

fn arenaString(L: *c.lua_State, tbl_idx: c_int, name: [:0]const u8, arena: std.mem.Allocator) ToolError![]const u8 {
    const t = c.lua_getfield(L, tbl_idx, name.ptr);
    defer c.lua_settop(L, c.lua_gettop(L) - 1);
    if (t == T_NIL) return error.MissingField;
    if (t != T_STRING) return error.BadField;
    var len: usize = 0;
    const ptr = c.lua_tolstring(L, -1, &len);
    if (ptr == null) return error.BadField;
    return arena.dupe(u8, ptr[0..len]) catch error.OutOfMemory;
}

/// Internal test hook: `agent:_dispatch_tools({ {name=,input=}, ... })`
/// runs the calls through the real `invoke_batch` (coroutines + luv) and
/// returns an array of `{ ok=bool, text=string }`. Lets us validate the
/// dispatch path without a live provider emitting tool calls.
fn agentDispatchTools(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const box = checkAgent(L, 1);
    if (box.closed) return luaErr(L, "panto: agent is closed");
    c.luaL_checktype(L, 2, T_TABLE);
    const ts = box.tools orelse return luaErr(L, "panto: no tools registered");

    const n: usize = @intCast(c.lua_rawlen(L, 2));
    var arena_state = std.heap.ArenaAllocator.init(c_allocator);
    defer arena_state.deinit();
    const arena = arena_state.allocator();

    const calls = arena.alloc(panto.ToolCall, n) catch return luaErr(L, "panto: oom");
    var i: usize = 0;
    while (i < n) : (i += 1) {
        _ = c.lua_rawgeti(L, 2, @intCast(i + 1));
        const name = (optTableString(L, -1, "name") catch null) orelse
            return luaErr(L, "_dispatch_tools: each call needs name + input strings");
        const name_owned = arena.dupe(u8, name) catch return luaErr(L, "panto: oom");
        c.lua_settop(L, c.lua_gettop(L) - 1); // pop name value
        const input = (optTableString(L, -1, "input") catch null) orelse "{}";
        const input_owned = arena.dupe(u8, input) catch return luaErr(L, "panto: oom");
        c.lua_settop(L, c.lua_gettop(L) - 1); // pop input value
        c.lua_settop(L, c.lua_gettop(L) - 1); // pop call table
        calls[i] = .{ .tool_name = name_owned, .input = input_owned };
    }

    const results = arena.alloc(panto.ToolCallResult, n) catch return luaErr(L, "panto: oom");
    invokeBatch(ts, calls, results, arena) catch return luaErr(L, "panto: dispatch failed");

    c.lua_createtable(L, @intCast(n), 0);
    for (results, 0..) |r, idx| {
        c.lua_createtable(L, 0, 2);
        switch (r) {
            .ok => |parts| {
                c.lua_pushboolean(L, 1);
                c.lua_setfield(L, -2, "ok");
                const text: []const u8 = if (parts.items.len > 0 and parts.items[0] == .text)
                    parts.items[0].text
                else
                    "";
                setStringField(L, "text", text);
                parts.deinit(arena);
            },
            .err => {
                c.lua_pushboolean(L, 0);
                c.lua_setfield(L, -2, "ok");
            },
        }
        c.lua_rawseti(L, -2, @intCast(idx + 1));
    }
    return 1;
}

fn invokeBatch(
    ctx: *anyopaque,
    calls: []const panto.ToolCall,
    results: []panto.ToolCallResult,
    allocator: std.mem.Allocator,
) anyerror!void {
    const self: *LuaToolSource = @ptrCast(@alignCast(ctx));
    const L = self.L;

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

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

    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(L, LUA_REGISTRYINDEX, r);
    };

    // Step 1: start every call's coroutine. Sync handlers complete here.
    var any_pending = false;
    for (calls, 0..) |call, i| {
        const handler_ref = self.handlers.get(call.tool_name) orelse {
            slots[i] = .{ .recorded = true, .ok = false, .err_msg = try allocator.dupe(u8, "unknown tool name") };
            continue;
        };
        const started = startCoroutine(self, i, handler_ref, call.input, allocator) catch {
            slots[i] = .{ .recorded = true, .ok = false, .err_msg = try allocator.dupe(u8, "failed to start handler coroutine") };
            continue;
        };
        thread_refs[i] = started.thread_ref;
        if (started.still_pending) any_pending = true;
    }

    // Step 2: drive libuv to completion (wakes any yielded coroutine).
    if (any_pending) try driveUvToCompletion(self);

    // Step 3: reap. A still-suspended coroutine violated the contract.
    for (thread_refs, 0..) |tref, i| {
        if (tref == 0) continue;
        _ = c.lua_rawgeti(L, LUA_REGISTRYINDEX, tref);
        const co: *c.lua_State = @ptrCast(c.lua_tothread(L, -1).?);
        const status = c.lua_status(co);
        c.lua_settop(L, c.lua_gettop(L) - 1);
        if (status == c.LUA_YIELD and !slots[i].recorded) {
            slots[i] = .{ .recorded = true, .ok = false, .err_msg = try allocator.dupe(
                u8,
                "handler still suspended after the event loop drained (yielded without a pending libuv op)",
            ) };
        }
        c.luaL_unref(L, LUA_REGISTRYINDEX, tref);
        thread_refs[i] = 0;
    }

    // Step 4: translate slots into libpanto results. A Lua-level failure
    // is surfaced to the model as a textual `.ok` result rather than a
    // `.err` (which libpanto treats as turn-aborting) — same policy as the
    // CLI runtime.
    for (slots, 0..) |slot, i| {
        if (slot.ok and slot.value != null) {
            results[i] = .{ .ok = slot.value.? };
            if (slot.err_msg) |m| allocator.free(m);
        } else {
            if (slot.value) |v| v.deinit(allocator);
            results[i] = .{ .ok = try formatToolError(
                allocator,
                calls[i].tool_name,
                slot.err_msg orelse "(no message)",
            ) };
            if (slot.err_msg) |m| allocator.free(m);
        }
    }
}

fn formatToolError(allocator: std.mem.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);
}

fn startCoroutine(
    self: *LuaToolSource,
    idx: usize,
    handler_ref: c_int,
    input: []const u8,
    allocator: std.mem.Allocator,
) !struct { thread_ref: c_int, still_pending: bool } {
    const L = self.L;
    const co = c.lua_newthread(L) orelse return error.LuaInitFailed;
    const thread_ref = c.luaL_ref(L, LUA_REGISTRYINDEX);

    _ = c.lua_rawgeti(co, LUA_REGISTRYINDEX, @intCast(self.wrapper_ref));
    c.lua_pushinteger(co, @intCast(idx));
    _ = c.lua_rawgeti(co, LUA_REGISTRYINDEX, @intCast(handler_ref));
    var arena_state = std.heap.ArenaAllocator.init(allocator);
    defer arena_state.deinit();
    try 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 };
}

fn driveUvToCompletion(self: *LuaToolSource) !void {
    const L = self.L;
    _ = c.lua_rawgeti(L, LUA_REGISTRYINDEX, @intCast(self.uv_run_ref));
    _ = c.lua_pushlstring(L, "default", 7);
    if (c.lua_pcallk(L, 1, 1, 0, 0, null) != 0) {
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return error.UvRunFailed;
    }
    c.lua_settop(L, c.lua_gettop(L) - 1); // discard the bool return
}

/// `_record_result(idx, ok, value)` — carries `*LuaToolSource` as upvalue
/// 1, writes into the in-flight batch's slot.
fn recordResultC(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const self_ptr = c.lua_touserdata(L, c.lua_upvalueindex(1)) orelse return 0;
    const self: *LuaToolSource = @ptrCast(@alignCast(self_ptr));
    const batch = self.current_batch orelse return 0;

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

    if (ok) {
        const value = readHandlerResult(L, 3, batch.allocator) catch |e| {
            const msg = std.fmt.allocPrint(batch.allocator, "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 {
        var len: usize = 0;
        const ptr = c.luaL_tolstring(L, 3, &len);
        const owned = if (ptr != null) batch.allocator.dupe(u8, ptr[0..len]) catch null else null;
        c.lua_settop(L, c.lua_gettop(L) - 1); // pop tolstring's string
        batch.slots[idx] = .{ .recorded = true, .ok = false, .err_msg = owned };
    }
    return 0;
}

// ===========================================================================
// Stream methods
// ===========================================================================

fn streamNext(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const sbox = checkStream(L, 1);
    if (sbox.closed) return luaErr(L, "panto: stream is closed");

    const maybe_event = sbox.stream.next() catch
        return luaErr(L, "panto: stream failed");
    if (maybe_event) |ev| {
        pushEvent(L, ev) catch return luaErr(L, "panto: out of memory building event");
        return 1;
    }
    c.lua_pushnil(L);
    return 1;
}

/// `stream:events()` -> (iterator, stream, nil): the generic-`for` triple,
/// so `for ev in stream:events() do ... end` drives `next()`.
fn streamEvents(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    _ = checkStream(L, 1);
    c.lua_pushcclosure(L, streamIterStep, 0);
    c.lua_pushvalue(L, 1);
    c.lua_pushnil(L);
    return 3;
}

fn streamIterStep(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const sbox = checkStream(L, 1);
    if (sbox.closed) {
        c.lua_pushnil(L);
        return 1;
    }
    const maybe_event = sbox.stream.next() catch
        return luaErr(L, "panto: stream failed");
    if (maybe_event) |ev| {
        pushEvent(L, ev) catch return luaErr(L, "panto: out of memory building event");
        return 1;
    }
    c.lua_pushnil(L);
    return 1;
}

/// `stream:reopen()`: reset a failed stream back to its turn-open boundary so
/// the caller can resume `next()` after changing the agent config (e.g. catch
/// a terminal error, swap the provider config via `agent:set_config(...)`, and
/// retry the SAME turn without re-appending the user message). Errors if the
/// stream has not failed (or is closed).
fn streamReopen(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const sbox = checkStream(L, 1);
    if (sbox.closed) return luaErr(L, "panto: stream is closed");
    sbox.stream.reopen() catch return luaErr(L, "panto: stream reopen failed");
    return 0;
}

fn streamGc(L_opt: ?*c.lua_State) callconv(.c) c_int {
    const L = L_opt.?;
    const sbox = checkStream(L, 1);
    sbox.deinit(L);
    return 0;
}

// ===========================================================================
// Event -> Lua table marshalling
// ===========================================================================

fn pushEvent(L: *c.lua_State, ev: panto.Event) !void {
    c.lua_createtable(L, 0, 4);
    switch (ev) {
        .message_start => |role| {
            setStringField(L, "type", "message_start");
            setStringField(L, "role", roleName(role));
        },
        .block_start => |bs| {
            setStringField(L, "type", "block_start");
            setStringField(L, "block_type", blockTypeName(bs.block_type));
            setIntField(L, "index", @intCast(bs.index));
        },
        .tool_details => |td| {
            setStringField(L, "type", "tool_details");
            setIntField(L, "index", @intCast(td.index));
            setStringField(L, "id", td.id);
            setStringField(L, "name", td.name);
        },
        .content_delta => |cd| {
            setStringField(L, "type", "content_delta");
            setIntField(L, "index", @intCast(cd.index));
            setStringField(L, "delta", cd.delta);
        },
        .block_complete => |bc| {
            setStringField(L, "type", "block_complete");
            setIntField(L, "index", @intCast(bc.index));
            setStringField(L, "block_type", contentBlockTypeName(bc.block));
            pushBlockText(L, bc.block);
        },
        .message_complete => |mc| {
            setStringField(L, "type", "message_complete");
            setStringField(L, "role", roleName(mc.message.role));
            if (mc.usage) |u| pushUsage(L, u);
        },
        .provider_retry => |pr| {
            setStringField(L, "type", "provider_retry");
            setIntField(L, "attempt", @intCast(pr.attempt));
            setIntField(L, "max_attempts", @intCast(pr.max_attempts));
            setIntField(L, "delay_ms", @intCast(pr.delay_ms));
        },
        .tool_dispatch_start => |tds| {
            setStringField(L, "type", "tool_dispatch_start");
            setIntField(L, "count", @intCast(tds.count));
        },
        .tool_dispatch_result => setStringField(L, "type", "tool_dispatch_result"),
        .tool_dispatch_complete => setStringField(L, "type", "tool_dispatch_complete"),
        .turn_complete => setStringField(L, "type", "turn_complete"),
    }
}

fn pushBlockText(L: *c.lua_State, block: panto.ContentBlock) void {
    switch (block) {
        .Text => |t| setStringField(L, "text", t.items),
        .Thinking => |t| setStringField(L, "text", t.text.items),
        .ToolUse => |tu| {
            setStringField(L, "text", tu.input.items);
            setStringField(L, "id", tu.id);
            setStringField(L, "name", tu.name);
        },
        .ToolResult => {},
        .System => |s| setStringField(L, "text", s.text.items),
        .CompactionSummary => |s| setStringField(L, "text", s.text.items),
    }
}

fn pushUsage(L: *c.lua_State, u: panto.Usage) void {
    c.lua_createtable(L, 0, 5);
    setIntField(L, "input", @intCast(u.input));
    setIntField(L, "output", @intCast(u.output));
    setIntField(L, "cache_read", @intCast(u.cache_read));
    setIntField(L, "cache_write", @intCast(u.cache_write));
    setIntField(L, "reasoning", @intCast(u.reasoning));
    c.lua_setfield(L, -2, "usage");
}

fn roleName(role: panto.MessageRole) []const u8 {
    return switch (role) {
        .system => "system",
        .user => "user",
        .assistant => "assistant",
    };
}

fn blockTypeName(t: panto.ContentBlockType) []const u8 {
    return switch (t) {
        .Text => "text",
        .Thinking => "thinking",
        .ToolUse => "tool_use",
        .ToolResult => "tool_result",
    };
}

fn contentBlockTypeName(block: panto.ContentBlock) []const u8 {
    return switch (block) {
        .Text => "text",
        .Thinking => "thinking",
        .ToolUse => "tool_use",
        .ToolResult => "tool_result",
        .System => "system",
        .CompactionSummary => "compaction_summary",
    };
}

// ===========================================================================
// JSON <-> Lua (ported from src/lua_bridge.zig; standalone copy so this
// package has no cross-package dependency on the CLI)
// ===========================================================================

const lua_store_vtable: panto.SessionStore.VTable = .{
    .create = luaStoreCreate,
    .list = luaStoreList,
    .freeSessionInfos = luaStoreFreeInfos,
    .resolve = luaStoreResolve,
    .latest = luaStoreLatest,
    .load = luaStoreLoad,
    .appendMessages = luaStoreAppend,
};

fn luaStoreCreate(ctx: *anyopaque) panto.Session {
    const self: *LuaSessionStore = @ptrCast(@alignCast(ctx));
    callStoreMethod(self, "create", 0) catch return fallbackNullSession();
    defer c.lua_settop(self.L, c.lua_gettop(self.L) - 1);
    return sessionFromLuaInfo(self, -1) catch fallbackNullSession();
}

fn luaStoreList(ctx: *anyopaque) anyerror![]panto.SessionInfo {
    const self: *LuaSessionStore = @ptrCast(@alignCast(ctx));
    try callStoreMethod(self, "list", 0);
    defer c.lua_settop(self.L, c.lua_gettop(self.L) - 1);
    if (c.lua_type(self.L, -1) != T_TABLE) return error.BadLuaStore;
    const n: usize = @intCast(c.lua_rawlen(self.L, -1));
    const out = try c_allocator.alloc(panto.SessionInfo, n);
    var built: usize = 0;
    errdefer { for (out[0..built]) |i| i.deinit(c_allocator); c_allocator.free(out); }
    var i: usize = 1;
    while (i <= n) : (i += 1) {
        _ = c.lua_rawgeti(self.L, -1, @intCast(i));
        defer c.lua_settop(self.L, c.lua_gettop(self.L) - 1);
        out[built] = try parseSessionInfo(self.L, -1);
        built += 1;
    }
    return out;
}

fn luaStoreFreeInfos(_: *anyopaque, infos: []panto.SessionInfo) void {
    for (infos) |i| i.deinit(c_allocator);
    c_allocator.free(infos);
}

fn luaStoreResolve(ctx: *anyopaque, id: []const u8) anyerror!?panto.Session {
    const self: *LuaSessionStore = @ptrCast(@alignCast(ctx));
    _ = c.lua_pushlstring(self.L, id.ptr, id.len);
    try callStoreMethod(self, "resolve", 1);
    defer c.lua_settop(self.L, c.lua_gettop(self.L) - 1);
    if (c.lua_type(self.L, -1) == T_NIL) return null;
    return try sessionFromLuaInfo(self, -1);
}

fn luaStoreLatest(ctx: *anyopaque) anyerror!?panto.Session {
    const self: *LuaSessionStore = @ptrCast(@alignCast(ctx));
    try callStoreMethod(self, "latest", 0);
    defer c.lua_settop(self.L, c.lua_gettop(self.L) - 1);
    if (c.lua_type(self.L, -1) == T_NIL) return null;
    return try sessionFromLuaInfo(self, -1);
}

fn luaStoreLoad(ctx: *anyopaque, id: []const u8) anyerror!?panto.Conversation {
    const self: *LuaSessionStore = @ptrCast(@alignCast(ctx));
    _ = c.lua_pushlstring(self.L, id.ptr, id.len);
    try callStoreMethod(self, "load", 1);
    defer c.lua_settop(self.L, c.lua_gettop(self.L) - 1);
    if (c.lua_type(self.L, -1) == T_NIL) return null;
    const cb = @as(?*ConvBox, if (c.luaL_testudata(self.L, -1, CONV_MT)) |ud| @ptrCast(@alignCast(ud)) else null) orelse return error.BadLuaStore;
    if (cb.state != .owned) return error.BadLuaStore;
    const conv = cb.inner;
    cb.inner = panto.Conversation.init(c_allocator);
    return conv;
}

fn luaStoreAppend(ctx: *anyopaque, session_id: []const u8, messages: []panto.PersistentMessage) anyerror!void {
    const self: *LuaSessionStore = @ptrCast(@alignCast(ctx));
    _ = c.lua_pushlstring(self.L, session_id.ptr, session_id.len);
    c.lua_createtable(self.L, @intCast(messages.len), 0);
    for (messages, 0..) |m, i| {
        pushPersistentMessage(self.L, m);
        c.lua_rawseti(self.L, -2, @intCast(i + 1));
    }
    try callStoreMethod(self, "append_messages", 2);
    c.lua_settop(self.L, c.lua_gettop(self.L) - 1);
}

fn callStoreMethod(self: *LuaSessionStore, name: [:0]const u8, nargs: c_int) !void {
    const L = self.L;
    _ = c.lua_rawgeti(L, LUA_REGISTRYINDEX, @intCast(self.table_ref));
    const tt = c.lua_getfield(L, -1, name.ptr);
    if (tt != T_FUNCTION) return error.BadLuaStore;
    c.lua_insert(L, -(nargs + 2));
    c.lua_settop(L, c.lua_gettop(L) - 1);
    if (c.lua_pcallk(L, nargs, 1, 0, 0, null) != 0) return error.BadLuaStore;
}

fn sessionFromLuaInfo(self: *LuaSessionStore, idx: c_int) !panto.Session {
    return .{ .info = try parseSessionInfo(self.L, idx), .store = self.store() };
}

fn parseSessionInfo(L: *c.lua_State, idx: c_int) !panto.SessionInfo {
    if (c.lua_type(L, idx) != T_TABLE) return error.BadLuaStore;
    return .{
        .id = try dupLuaField(L, idx, "id"),
        .created = try dupLuaField(L, idx, "created"),
        .modified = try dupLuaField(L, idx, "modified"),
        .message_count = @intCast(intLuaField(L, idx, "message_count")),
        .last_user_message = try dupLuaField(L, idx, "last_user_message"),
        .api_style = parseEnum(panto.APIStyle, luaFieldOrDefault(L, idx, "api_style", "openai_chat") catch "openai_chat") orelse .openai_chat,
        .base_url = try dupLuaField(L, idx, "base_url"),
        .model = try dupLuaField(L, idx, "model"),
        .reasoning = parseEnum(panto.ReasoningEffort, luaFieldOrDefault(L, idx, "reasoning", "default") catch "default") orelse .default,
    };
}

fn dupLuaField(L: *c.lua_State, idx: c_int, name: [:0]const u8) ![]u8 {
    return c_allocator.dupe(u8, try luaFieldOrDefault(L, idx, name, ""));
}

fn luaFieldOrDefault(L: *c.lua_State, idx: c_int, name: [:0]const u8, default: []const u8) ![]const u8 {
    const abs = c.lua_absindex(L, idx);
    const t = c.lua_getfield(L, abs, name.ptr);
    defer c.lua_settop(L, c.lua_gettop(L) - 1);
    if (t == T_NIL) return default;
    if (t != T_STRING) return error.BadLuaStore;
    var len: usize = 0;
    const ptr = c.lua_tolstring(L, -1, &len) orelse return error.BadLuaStore;
    return ptr[0..len];
}

fn intLuaField(L: *c.lua_State, idx: c_int, name: [:0]const u8) c.lua_Integer {
    const abs = c.lua_absindex(L, idx);
    const t = c.lua_getfield(L, abs, name.ptr);
    defer c.lua_settop(L, c.lua_gettop(L) - 1);
    if (t != T_NUMBER) return 0;
    return c.lua_tointegerx(L, -1, null);
}

fn fallbackNullSession() panto.Session {
    var ns = panto.NullStore.init(c_allocator);
    return ns.store().create();
}

fn pushPersistentMessage(L: *c.lua_State, pm: panto.PersistentMessage) void {
    c.lua_createtable(L, 0, 4);
    setStringField(L, "role", roleName(pm.message.role));
    if (pm.usage) |u| pushUsage(L, u);
    setStringField(L, "metadata", pm.message.metadata orelse "");
    c.lua_createtable(L, @intCast(pm.message.content.items.len), 0);
    for (pm.message.content.items, 0..) |block, i| {
        pushInspectBlock(L, block);
        c.lua_rawseti(L, -2, @intCast(i + 1));
    }
    c.lua_setfield(L, -2, "blocks");
}

const JsonError = error{ OutOfMemory, BadHandlerReturn, InputNotJsonObject };

/// Parse JSON bytes (must be a top-level object — tool input convention)
/// and push the value onto `L`.
fn pushJsonAsLua(L: *c.lua_State, arena: std.mem.Allocator, input: []const u8) !void {
    var parsed = std.json.parseFromSlice(std.json.Value, arena, input, .{}) catch
        return JsonError.InputNotJsonObject;
    defer parsed.deinit();
    if (parsed.value != .object) return JsonError.InputNotJsonObject;
    try pushJsonValue(L, parsed.value);
}

fn pushJsonValue(L: *c.lua_State, v: std.json.Value) !void {
    switch (v) {
        .null => c.lua_pushnil(L),
        .bool => |b| c.lua_pushboolean(L, if (b) 1 else 0),
        .integer => |i| c.lua_pushinteger(L, @intCast(i)),
        .float => |f| c.lua_pushnumber(L, f),
        .number_string => |s| {
            if (std.fmt.parseInt(c.lua_Integer, s, 10)) |i| {
                c.lua_pushinteger(L, i);
            } else |_| {
                const f = try std.fmt.parseFloat(c.lua_Number, s);
                c.lua_pushnumber(L, f);
            }
        },
        .string => |s| _ = c.lua_pushlstring(L, s.ptr, s.len),
        .array => |arr| {
            c.lua_createtable(L, @intCast(arr.items.len), 0);
            for (arr.items, 0..) |item, i| {
                try pushJsonValue(L, item);
                c.lua_rawseti(L, -2, @intCast(i + 1));
            }
        },
        .object => |obj| {
            c.lua_createtable(L, 0, @intCast(obj.count()));
            var it = obj.iterator();
            while (it.next()) |kv| {
                const key = kv.key_ptr.*;
                _ = c.lua_pushlstring(L, key.ptr, key.len);
                try pushJsonValue(L, kv.value_ptr.*);
                c.lua_rawset(L, -3);
            }
        },
    }
}

/// Read a handler's return at `idx` into owned `ResultParts`. A plain
/// string -> one text part; a table `{ text=, attachments={{media_type=,
/// data=}} }` -> optional text + one media part per attachment.
fn readHandlerResult(L: *c.lua_State, idx: c_int, allocator: std.mem.Allocator) !panto.ResultParts {
    const ty = c.lua_type(L, idx);
    if (ty == T_STRING) {
        var len: usize = 0;
        const ptr = c.lua_tolstring(L, idx, &len);
        if (ptr == null) return JsonError.BadHandlerReturn;
        return panto.ResultParts.fromText(allocator, ptr[0..len]);
    }
    if (ty != T_TABLE) {
        // Coerce other scalars (numbers/bools) via tostring for ergonomics.
        var len: usize = 0;
        const ptr = c.luaL_tolstring(L, idx, &len);
        defer c.lua_settop(L, c.lua_gettop(L) - 1);
        if (ptr == null) return JsonError.BadHandlerReturn;
        return panto.ResultParts.fromText(allocator, ptr[0..len]);
    }
    return readHandlerResultTable(L, idx, allocator);
}

fn readHandlerResultTable(L: *c.lua_State, idx: c_int, allocator: std.mem.Allocator) !panto.ResultParts {
    var parts: std.ArrayList(panto.ResultPart) = .empty;
    errdefer {
        for (parts.items) |p| p.deinit(allocator);
        parts.deinit(allocator);
    }
    const abs = c.lua_absindex(L, idx);

    // Optional `text`.
    if (try optTableString(L, abs, "text")) |s| {
        const owned = try allocator.dupe(u8, s);
        c.lua_settop(L, c.lua_gettop(L) - 1);
        errdefer allocator.free(owned);
        try parts.append(allocator, .{ .text = owned });
    }

    // Optional `attachments`.
    const at = c.lua_getfield(L, abs, "attachments");
    defer c.lua_settop(L, c.lua_gettop(L) - 1);
    if (at == T_TABLE) {
        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));
            defer c.lua_settop(L, c.lua_gettop(L) - 1);
            if (c.lua_type(L, -1) != T_TABLE) return JsonError.BadHandlerReturn;

            const media_type: ?[]const u8 = if (try optTableString(L, -1, "media_type")) |mt| blk: {
                const owned = try allocator.dupe(u8, mt);
                c.lua_settop(L, c.lua_gettop(L) - 1);
                break :blk owned;
            } else null;
            errdefer if (media_type) |mt| allocator.free(mt);

            const data_slice = (try optTableString(L, -1, "data")) orelse return JsonError.BadHandlerReturn;
            const data = try allocator.dupe(u8, data_slice);
            c.lua_settop(L, c.lua_gettop(L) - 1);
            errdefer allocator.free(data);

            try parts.append(allocator, .{ .media = .{ .media_type = media_type, .data = data } });
        }
    } else if (at != T_NIL) {
        return JsonError.BadHandlerReturn;
    }

    const items = try parts.toOwnedSlice(allocator);
    return .{ .items = items };
}

/// Read string field `name` from the table at `tbl_idx`. Leaves the value
/// on the stack (caller pops after copying); null if absent.
fn optTableString(L: *c.lua_State, tbl_idx: c_int, name: [*:0]const u8) !?[]const u8 {
    const t = c.lua_getfield(L, tbl_idx, name);
    if (t == T_NIL) {
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return null;
    }
    if (t != T_STRING) {
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return JsonError.BadHandlerReturn;
    }
    var len: usize = 0;
    const ptr = c.lua_tolstring(L, -1, &len);
    if (ptr == null) {
        c.lua_settop(L, c.lua_gettop(L) - 1);
        return JsonError.BadHandlerReturn;
    }
    return ptr[0..len];
}

/// Serialize the Lua table at `idx` to a JSON string (arena-owned).
fn serializeSchema(L: *c.lua_State, idx: c_int, arena: std.mem.Allocator) ![]const u8 {
    var aw: std.Io.Writer.Allocating = .init(arena);
    var s: std.json.Stringify = .{ .writer = &aw.writer };
    try writeLuaValueAsJson(L, idx, &s);
    return aw.written();
}

const SchemaJsonError = error{ UnsupportedLuaType, UnsupportedLuaKey, WriteFailed };

fn writeLuaValueAsJson(L: *c.lua_State, idx: c_int, w: *std.json.Stringify) SchemaJsonError!void {
    switch (c.lua_type(L, idx)) {
        T_NIL => w.write(null) catch return error.WriteFailed,
        T_BOOLEAN => w.write(c.lua_toboolean(L, idx) != 0) catch return error.WriteFailed,
        T_NUMBER => {
            var isnum: c_int = 0;
            const as_int = c.lua_tointegerx(L, idx, &isnum);
            if (isnum != 0)
                w.write(as_int) catch return error.WriteFailed
            else
                w.write(c.lua_tonumberx(L, idx, null)) catch return error.WriteFailed;
        },
        T_STRING => {
            var len: usize = 0;
            const ptr = c.lua_tolstring(L, idx, &len);
            w.write(ptr[0..len]) catch return error.WriteFailed;
        },
        T_TABLE => try writeLuaTableAsJson(L, idx, w),
        else => return error.UnsupportedLuaType,
    }
}

fn writeLuaTableAsJson(L: *c.lua_State, idx_in: c_int, w: *std.json.Stringify) SchemaJsonError!void {
    const abs = c.lua_absindex(L, idx_in);
    const len = c.lua_rawlen(L, abs);
    if (len > 0) {
        w.beginArray() catch return error.WriteFailed;
        var i: c.lua_Integer = 1;
        while (i <= @as(c.lua_Integer, @intCast(len))) : (i += 1) {
            _ = c.lua_rawgeti(L, abs, i);
            try writeLuaValueAsJson(L, -1, w);
            c.lua_settop(L, c.lua_gettop(L) - 1);
        }
        w.endArray() catch return error.WriteFailed;
        return;
    }
    w.beginObject() catch return error.WriteFailed;
    c.lua_pushnil(L);
    while (c.lua_next(L, abs) != 0) {
        if (c.lua_type(L, -2) != T_STRING) {
            c.lua_settop(L, c.lua_gettop(L) - 1);
            return error.UnsupportedLuaKey;
        }
        var klen: usize = 0;
        const kptr = c.lua_tolstring(L, -2, &klen);
        w.objectField(kptr[0..klen]) catch return error.WriteFailed;
        try writeLuaValueAsJson(L, -1, w);
        c.lua_settop(L, c.lua_gettop(L) - 1);
    }
    w.endObject() catch return error.WriteFailed;
}

// ===========================================================================
// Stack / argument helpers
// ===========================================================================

fn checkAgent(L: *c.lua_State, idx: c_int) *AgentBox {
    return @ptrCast(@alignCast(c.luaL_checkudata(L, idx, AGENT_MT)));
}

fn checkStream(L: *c.lua_State, idx: c_int) *StreamBox {
    return @ptrCast(@alignCast(c.luaL_checkudata(L, idx, STREAM_MT)));
}

/// Raise a Lua error with a static message. Never returns (longjmp), typed
/// `c_int` so callers can `return` it.
fn luaErr(L: *c.lua_State, msg: [:0]const u8) c_int {
    _ = c.luaL_error(L, "%s", msg.ptr);
    return 0;
}

fn requiredString(L: *c.lua_State, tbl_idx: c_int, name: [:0]const u8) ConfigError![]const u8 {
    const t = c.lua_getfield(L, tbl_idx, name.ptr);
    defer c.lua_settop(L, c.lua_gettop(L) - 1);
    if (t == T_NIL) return error.MissingField;
    if (t != T_STRING) return error.BadField;
    var len: usize = 0;
    const ptr = c.lua_tolstring(L, -1, &len);
    if (ptr == null) return error.BadField;
    return ptr[0..len];
}

fn optString(L: *c.lua_State, tbl_idx: c_int, name: [:0]const u8) ConfigError!?[]const u8 {
    const t = c.lua_getfield(L, tbl_idx, name.ptr);
    defer c.lua_settop(L, c.lua_gettop(L) - 1);
    if (t == T_NIL) return null;
    if (t != T_STRING) return error.BadField;
    var len: usize = 0;
    const ptr = c.lua_tolstring(L, -1, &len);
    if (ptr == null) return error.BadField;
    return ptr[0..len];
}

/// An optional positional string argument (for compact's override/extra).
fn optArgString(L: *c.lua_State, idx: c_int) ?[]const u8 {
    if (c.lua_type(L, idx) != T_STRING) return null;
    var len: usize = 0;
    const ptr = c.lua_tolstring(L, idx, &len);
    if (ptr == null) return null;
    return ptr[0..len];
}

fn optU32(L: *c.lua_State, tbl_idx: c_int, name: [:0]const u8) ConfigError!?u32 {
    const v = try optInteger(L, tbl_idx, name) orelse return null;
    if (v < 0 or v > std.math.maxInt(u32)) return error.BadField;
    return @intCast(v);
}

fn optU64(L: *c.lua_State, tbl_idx: c_int, name: [:0]const u8) ConfigError!?u64 {
    const v = try optInteger(L, tbl_idx, name) orelse return null;
    if (v < 0) return error.BadField;
    return @intCast(v);
}

fn optInteger(L: *c.lua_State, tbl_idx: c_int, name: [:0]const u8) ConfigError!?c.lua_Integer {
    const t = c.lua_getfield(L, tbl_idx, name.ptr);
    defer c.lua_settop(L, c.lua_gettop(L) - 1);
    if (t == T_NIL) return null;
    if (t != T_NUMBER) return error.BadField;
    var isnum: c_int = 0;
    const v = c.lua_tointegerx(L, -1, &isnum);
    if (isnum == 0) return error.BadField;
    return v;
}

fn optNumber(L: *c.lua_State, tbl_idx: c_int, name: [:0]const u8) ConfigError!?f64 {
    const t = c.lua_getfield(L, tbl_idx, name.ptr);
    defer c.lua_settop(L, c.lua_gettop(L) - 1);
    if (t == T_NIL) return null;
    if (t != T_NUMBER) return error.BadField;
    return c.lua_tonumberx(L, -1, null);
}

fn optBool(L: *c.lua_State, tbl_idx: c_int, name: [:0]const u8) ConfigError!?bool {
    const t = c.lua_getfield(L, tbl_idx, name.ptr);
    defer c.lua_settop(L, c.lua_gettop(L) - 1);
    if (t == T_NIL) return null;
    if (t != T_BOOLEAN) return error.BadField;
    return c.lua_toboolean(L, -1) != 0;
}

fn setStringField(L: *c.lua_State, key: [:0]const u8, value: []const u8) void {
    _ = c.lua_pushlstring(L, value.ptr, value.len);
    c.lua_setfield(L, -2, key.ptr);
}

fn setIntField(L: *c.lua_State, key: [:0]const u8, value: c.lua_Integer) void {
    c.lua_pushinteger(L, value);
    c.lua_setfield(L, -2, key.ptr);
}

// ===========================================================================
// Tests
// ===========================================================================

const testing = std.testing;

fn newTestState() !*c.lua_State {
    const L = c.luaL_newstate() orelse return error.LuaInitFailed;
    c.luaL_openlibs(L);
    _ = luaopen_panto(L);
    c.lua_setglobal(L, "panto");
    return L;
}

fn runScript(L: *c.lua_State, src: [:0]const u8) !void {
    if (c.luaL_loadstring(L, src.ptr) != 0 or c.lua_pcallk(L, 0, c.LUA_MULTRET, 0, 0, null) != 0) {
        var len: usize = 0;
        const msg = c.lua_tolstring(L, -1, &len);
        if (msg != null) std.debug.print("lua error: {s}\n", .{msg[0..len]});
        return error.LuaScriptFailed;
    }
}

test "luaopen_panto returns a module table with agent + _VERSION" {
    const L = c.luaL_newstate() orelse return error.LuaInitFailed;
    defer c.lua_close(L);
    c.luaL_openlibs(L);

    try testing.expectEqual(@as(c_int, 1), luaopen_panto(L));
    try testing.expectEqual(@as(c_int, T_TABLE), c.lua_type(L, -1));
    _ = c.lua_getfield(L, -1, "agent");
    try testing.expectEqual(@as(c_int, T_FUNCTION), c.lua_type(L, -1));
    c.lua_settop(L, c.lua_gettop(L) - 1);
    _ = c.lua_getfield(L, -1, "_VERSION");
    try testing.expectEqual(@as(c_int, T_STRING), c.lua_type(L, -1));
}

test "luaopen_panto returns a FRESH table each call (not a shared singleton)" {
    const L = c.luaL_newstate() orelse return error.LuaInitFailed;
    defer c.lua_close(L);
    c.luaL_openlibs(L);

    _ = luaopen_panto(L);
    _ = luaopen_panto(L);
    _ = c.lua_pushstring(L, "marker");
    c.lua_setfield(L, -2, "_probe");
    _ = c.lua_getfield(L, -2, "_probe");
    try testing.expectEqual(@as(c_int, T_NIL), c.lua_type(L, -1));
}

test "agent: missing required fields raises (caught by pcall)" {
    const L = try newTestState();
    defer c.lua_close(L);
    try runScript(L,
        \\local ok, err = pcall(function() return panto.agent { api_style = "openai_chat" } end)
        \\assert(ok == false and type(err) == "string", "expected failure")
        \\assert(err:find("panto.agent"), "expected contextual message")
    );
}

test "agent: unknown api_style raises" {
    const L = try newTestState();
    defer c.lua_close(L);
    try runScript(L,
        \\local ok = pcall(function()
        \\  return panto.agent { api_style="nope", api_key="k", base_url="u", model="m" }
        \\end)
        \\assert(ok == false, "expected unknown api_style to fail")
    );
}

test "agent: bad enum value raises" {
    const L = try newTestState();
    defer c.lua_close(L);
    try runScript(L,
        \\local ok = pcall(function()
        \\  return panto.agent {
        \\    api_style="openai_chat", api_key="k", base_url="u", model="m",
        \\    reasoning="banana",
        \\  }
        \\end)
        \\assert(ok == false, "expected bad enum to fail")
    );
}

test "agent: full config surface parses (anthropic + compaction + retry)" {
    const L = try newTestState();
    defer c.lua_close(L);
    try runScript(L,
        \\local a = panto.agent {
        \\  api_style = "anthropic_messages",
        \\  api_key = "k", base_url = "http://127.0.0.1:1", model = "m",
        \\  max_tokens = 1000,
        \\  thinking = "adaptive", effort = "high",
        \\  thinking_budget_tokens = 2048, thinking_interleaved = true,
        \\  api_version = "2023-06-01",
        \\  compaction = { keep_verbatim = 5000, prompt = "summarize" },
        \\  retry = { max_attempts = 2, initial_delay_ms = 100, max_delay_ms = 500, multiplier = 1.5, jitter = false },
        \\}
        \\assert(type(a) == "userdata")
        \\assert(type(a.session_id) == "function")
        \\assert(type(a:session_id()) == "string")
    );
}

test "agent: methods exist and stream surface works" {
    const L = try newTestState();
    defer c.lua_close(L);
    try runScript(L,
        \\local a = panto.agent { api_style="openai_chat", api_key="k", base_url="http://127.0.0.1:1", model="m" }
        \\for _, m in ipairs({"run","register_tool","set_config","add_system_message","set_system_prompt","compact","session_id"}) do
        \\  assert(type(a[m]) == "function", "missing method "..m)
        \\end
        \\local s = a:run("hi")
        \\assert(type(s) == "userdata")
        \\local f, st, ctrl = s:events()
        \\assert(type(f) == "function" and st == s and ctrl == nil)
    );
}

test "agent: set_config swaps without error; system prompt setters work" {
    const L = try newTestState();
    defer c.lua_close(L);
    try runScript(L,
        \\local a = panto.agent { api_style="openai_chat", api_key="k", base_url="http://127.0.0.1:1", model="m" }
        \\a:add_system_message("be terse")
        \\a:set_system_prompt("you are a test")
        \\a:set_config { api_style="anthropic_messages", api_key="k2", base_url="http://127.0.0.1:2", model="m2" }
        \\-- another swap, to exercise old-config retention
        \\a:set_config { api_style="openai_chat", api_key="k3", base_url="http://127.0.0.1:3", model="m3" }
    );
}

test "_wrap_agent: borrowed agent shares the host's, and __gc leaves it alive" {
    const L = try newTestState();
    defer c.lua_close(L);
    try runScript(L,
        \\a = panto.agent { api_style="openai_chat", api_key="k", base_url="http://127.0.0.1:1", model="m" }
    );
    // Recover the underlying `*panto.Agent`, as a host embedder would
    // hold it, and wrap it borrowed.
    _ = c.lua_getglobal(L, "a");
    const owner = checkAgent(L, -1);
    c.lua_settop(L, c.lua_gettop(L) - 1);
    _ = c.lua_getglobal(L, "panto");
    _ = c.lua_getfield(L, -1, "_wrap_agent");
    c.lua_pushlightuserdata(L, owner.agent);
    if (c.lua_pcallk(L, 1, 1, 0, 0, null) != 0) return error.WrapFailed;
    c.lua_setglobal(L, "b");
    c.lua_settop(L, c.lua_gettop(L) - 1); // pop panto

    try runScript(L,
        \\-- Writes through the borrowed wrapper land on the shared agent.
        \\b:add_system_message("from borrowed")
        \\assert(a:conversation():len() == 1)
        \\-- Ownership-dependent methods are refused.
        \\local ok, err = pcall(function()
        \\  b:set_config { api_style="openai_chat", api_key="k", base_url="http://127.0.0.1:1", model="m" }
        \\end)
        \\assert(not ok and err:find("borrowed"), tostring(err))
        \\ok, err = pcall(function()
        \\  b:register_tool { name="t", description="d", schema={type="object"}, handler=function() end }
        \\end)
        \\assert(not ok and err:find("borrowed"), tostring(err))
        \\-- Collecting the borrowed wrapper must not tear the agent down.
        \\b = nil
        \\collectgarbage("collect")
        \\a:add_system_message("still alive")
        \\assert(a:conversation():len() == 2)
    );
}

test "submit: queues on borrowed agents, refused on owned ones" {
    const L = try newTestState();
    defer c.lua_close(L);
    try runScript(L,
        \\a = panto.agent { api_style="openai_chat", api_key="k", base_url="http://127.0.0.1:1", model="m" }
        \\local ok, err = pcall(function() a:submit("nope") end)
        \\assert(not ok and err:find("panto.ext.agent"), tostring(err))
    );
    // Wrap borrowed, as the CLI's `setExtAgent` does.
    _ = c.lua_getglobal(L, "a");
    const owner = checkAgent(L, -1);
    c.lua_settop(L, c.lua_gettop(L) - 1);
    _ = c.lua_getglobal(L, "panto");
    _ = c.lua_getfield(L, -1, "_wrap_agent");
    c.lua_pushlightuserdata(L, owner.agent);
    if (c.lua_pcallk(L, 1, 1, 0, 0, null) != 0) return error.WrapFailed;
    c.lua_setglobal(L, "b");
    c.lua_settop(L, c.lua_gettop(L) - 1); // pop panto

    try runScript(L,
        \\b:submit("hello")
        \\b:submit({ { type = "text", text = "world" } })
        \\local ok, err = pcall(function() b:submit({ { type = "bogus" } }) end)
        \\assert(not ok, "bad block shape must error")
    );
    // Both submits accumulated into ONE pending message on the shared agent.
    const taken = (try owner.agent.takeSubmission()) orelse return error.NothingQueued;
    defer {
        for (taken) |*b| b.deinit(owner.agent.conversation.allocator);
        owner.agent.conversation.allocator.free(taken);
    }
    try std.testing.expectEqual(@as(usize, 2), taken.len);
    try std.testing.expectEqualStrings("hello", taken[0].Text.items);
    try std.testing.expectEqualStrings("world", taken[1].Text.items);
    try std.testing.expectEqual(@as(?[]panto.ContentBlock, null), try owner.agent.takeSubmission());
}

test "register_tool without luv fails loud at registration" {
    // The test binary links Lua but NOT luv, so require('luv') fails. We
    // assert that surfaces as a clear error at register time, not later.
    const L = try newTestState();
    defer c.lua_close(L);
    try runScript(L,
        \\local a = panto.agent { api_style="openai_chat", api_key="k", base_url="http://127.0.0.1:1", model="m" }
        \\local ok, err = pcall(function()
        \\  a:register_tool {
        \\    name="echo", description="echoes", schema={ type="object" },
        \\    handler=function(input) return "ok" end,
        \\  }
        \\end)
        \\assert(ok == false, "expected register_tool to fail without luv")
        \\assert(err:find("luv"), "error should mention luv: "..tostring(err))
    );
}

test "conversation: build standalone, inspect via messages()" {
    const L = try newTestState();
    defer c.lua_close(L);
    try runScript(L,
        \\local conv = panto.conversation()
        \\assert(type(conv) == "userdata")
        \\conv:add_system_message("be terse")
        \\conv:add_user_message("hello")
        \\conv:add_user_message("again")
        \\assert(conv:len() == 3, "expected 3 messages, got "..conv:len())
        \\assert(#conv == 3, "__len should work too")
        \\local m = conv:messages()
        \\assert(m[1].role == "system" and m[1].blocks[1].type == "system")
        \\assert(m[1].blocks[1].text == "be terse")
        \\assert(m[2].role == "user" and m[2].blocks[1].text == "hello")
        \\assert(m[3].blocks[1].text == "again")
    );
    try runScript(L, "collectgarbage('collect')"); // owned conv freed cleanly
}

test "conversation: full-fidelity tool turn round-trips through messages()" {
    const L = try newTestState();
    defer c.lua_close(L);
    try runScript(L,
        \\local conv = panto.conversation()
        \\conv:add_user_message("use a tool")
        \\conv:add_assistant_message {
        \\  blocks = {
        \\    { type="thinking", text="hmm", signature="sig" },
        \\    { type="text", text="calling" },
        \\    { type="tool_use", id="c1", name="echo", input='{"m":"x"}' },
        \\  },
        \\  usage = { input=10, output=20, reasoning=5 },
        \\}
        \\conv:add_user_blocks {
        \\  { type="tool_result", tool_use_id="c1", is_error=false,
        \\    parts = { { text="echoed" }, { media_type="image/png", data="AAAA" } } },
        \\  { type="text", text="queued note" },
        \\}
        \\assert(conv:len() == 3)
        \\local m = conv:messages()
        \\-- assistant message: usage + three typed blocks
        \\assert(m[2].role == "assistant")
        \\assert(m[2].usage.input == 10 and m[2].usage.reasoning == 5)
        \\assert(m[2].blocks[1].type == "thinking" and m[2].blocks[1].signature == "sig")
        \\assert(m[2].blocks[3].type == "tool_use" and m[2].blocks[3].id == "c1")
        \\assert(m[2].blocks[3].input == '{"m":"x"}')
        \\-- user tool-result message: result block + queued text
        \\assert(m[3].blocks[1].type == "tool_result" and m[3].blocks[1].tool_use_id == "c1")
        \\assert(#m[3].blocks[1].parts == 2)
        \\assert(m[3].blocks[1].parts[1].text == "echoed")
        \\assert(m[3].blocks[1].parts[2].media_type == "image/png" and m[3].blocks[1].parts[2].data == "AAAA")
        \\assert(m[3].blocks[2].type == "text" and m[3].blocks[2].text == "queued note")
        \\-- lossless replay into a fresh conversation
        \\local c2 = panto.conversation()
        \\for _, msg in ipairs(m) do
        \\  if msg.role == "assistant" then c2:add_assistant_message { blocks = msg.blocks, usage = msg.usage }
        \\  elseif msg.role == "user" then c2:add_user_blocks(msg.blocks) end
        \\end
        \\assert(c2:len() == conv:len(), "round-trip length mismatch")
        \\local m2 = c2:messages()
        \\assert(m2[2].blocks[3].id == "c1" and m2[3].blocks[1].parts[2].data == "AAAA")
    );
    try runScript(L, "collectgarbage('collect')");
}

test "conversation: malformed block raises" {
    const L = try newTestState();
    defer c.lua_close(L);
    try runScript(L,
        \\local conv = panto.conversation()
        \\local ok = pcall(function() conv:add_user_blocks { { type="tool_use" } } end) -- missing id/name
        \\assert(ok == false, "tool_use without id/name should raise")
        \\local ok2 = pcall(function() conv:add_user_blocks { { type="frobnicate" } } end)
        \\assert(ok2 == false, "unknown block type should raise")
    );
}

test "conversation: adopt into agent for resumption, then operate via agent:conversation()" {
    const L = try newTestState();
    defer c.lua_close(L);
    try runScript(L,
        \\local conv = panto.conversation()
        \\conv:add_system_message("resumed prompt")
        \\conv:add_user_message("earlier question")
        \\local a = panto.agent {
        \\  api_style="openai_chat", api_key="k", base_url="http://127.0.0.1:1", model="m",
        \\  conversation = conv,
        \\}
        \\-- The standalone handle is now adopted; operating on it must raise.
        \\local ok = pcall(function() conv:add_user_message("nope") end)
        \\assert(ok == false, "adopted conversation should reject mutation")
        \\-- The live view reflects the adopted history.
        \\local live = a:conversation()
        \\assert(live:len() == 2, "expected 2 adopted messages, got "..live:len())
        \\live:add_system_message("added live")
        \\assert(live:len() == 3)
        \\assert(a:conversation():messages()[3].blocks[1].text == "added live")
    );
    try runScript(L, "collectgarbage('collect')");
}

test "conversation: reusing an adopted conversation in a second agent is rejected" {
    const L = try newTestState();
    defer c.lua_close(L);
    try runScript(L,
        \\local conv = panto.conversation()
        \\conv:add_user_message("x")
        \\local a1 = panto.agent { api_style="openai_chat", api_key="k", base_url="http://127.0.0.1:1", model="m", conversation = conv }
        \\local ok = pcall(function()
        \\  return panto.agent { api_style="openai_chat", api_key="k", base_url="http://127.0.0.1:1", model="m", conversation = conv }
        \\end)
        \\assert(ok == false, "an already-adopted conversation must not be adopted again")
    );
}

test "stream: a turn against an unreachable host surfaces as a Lua error" {
    const L = try newTestState();
    defer c.lua_close(L);
    try runScript(L,
        \\local a = panto.agent { api_style="openai_chat", api_key="k", base_url="http://127.0.0.1:1", model="m" }
        \\local s = a:run("hi")
        \\local ok, err = pcall(function() for ev in s:events() do end end)
        \\assert(ok == false and type(err) == "string", "unreachable host should raise")
    );
}

test "json round-trip helpers: schema serialize + input parse" {
    const L = c.luaL_newstate() orelse return error.LuaInitFailed;
    defer c.lua_close(L);
    c.luaL_openlibs(L);

    // serializeSchema on a table.
    try runScript(L, "schema = { type = \"object\", properties = { x = { type = \"number\" } } }");
    _ = c.lua_getglobal(L, "schema");
    var arena = std.heap.ArenaAllocator.init(testing.allocator);
    defer arena.deinit();
    const json = try serializeSchema(L, -1, arena.allocator());
    try testing.expect(std.mem.indexOf(u8, json, "\"type\"") != null);
    try testing.expect(std.mem.indexOf(u8, json, "\"object\"") != null);
    c.lua_settop(L, c.lua_gettop(L) - 1);

    // pushJsonAsLua + readHandlerResult round trip.
    try pushJsonAsLua(L, arena.allocator(), "{\"msg\":\"hello\"}");
    _ = c.lua_getfield(L, -1, "msg");
    var len: usize = 0;
    const ptr = c.lua_tolstring(L, -1, &len);
    try testing.expectEqualStrings("hello", ptr[0..len]);
}