support pending submissions on an Agent object

3 files changed, 174 insertions, 1 deletions

diff --git a/libpanto-lua/src/module.zig b/libpanto-lua/src/module.zig
index 17d0fb2..910334f 100644
--- a/libpanto-lua/src/module.zig
+++ b/libpanto-lua/src/module.zig
@@ -48,6 +48,10 @@
 //!     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`
@@ -354,6 +358,7 @@ 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);
@@ -670,6 +675,50 @@ fn agentRun(L_opt: ?*c.lua_State) callconv(.c) c_int {
     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);
@@ -2568,6 +2617,43 @@ test "_wrap_agent: borrowed agent shares the host's, and __gc leaves it alive" {
     );
 }
 
+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.
diff --git a/libpanto/src/agent.zig b/libpanto/src/agent.zig
index 3616ab6..09b0994 100644
--- a/libpanto/src/agent.zig
+++ b/libpanto/src/agent.zig
@@ -270,6 +270,13 @@ pub const Agent = struct {
     /// rewound by `rewriteWithSummary` so a compaction re-persists its summary
     /// + restated suffix.
     _persisted_through: usize = 0,
+    /// User-message blocks queued by an embedder seam (the CLI's
+    /// `panto.ext.agent:submit`) for the host front end to open the next
+    /// turn with, so the turn runs on the host's own driver (rendering,
+    /// interrupts) rather than inside the queuing callback. Blocks and the
+    /// list storage are owned by `conversation.allocator` — the same
+    /// ownership `run` adopts.
+    _pending_submission: std.ArrayList(conversation.ContentBlock) = .empty,
 
     /// Construct an agent.
     ///
@@ -311,6 +318,8 @@ pub const Agent = struct {
         // self-heap-pinned (`init` allocated it), so it frees itself last.
         const allocator = self._allocator;
         self._registry.deinit();
+        for (self._pending_submission.items) |*b| b.deinit(self.conversation.allocator);
+        self._pending_submission.deinit(self.conversation.allocator);
         self.conversation.deinit();
         self._session.info.deinit(allocator);
         allocator.destroy(self);
@@ -454,6 +463,22 @@ pub const Agent = struct {
     /// The agent re-reads its `config` snapshot at the top of each provider
     /// response inside the stream, so a mid-conversation `setConfig` takes
     /// effect at the next response boundary, never mid-stream.
+    /// Append blocks to the pending submission (see `_pending_submission`).
+    /// Adopts the block *contents*; the `blocks` slice itself stays the
+    /// caller's. Multiple calls accumulate into one pending user message.
+    pub fn queueSubmission(self: *Agent, blocks: []const conversation.ContentBlock) !void {
+        try self._pending_submission.appendSlice(self.conversation.allocator, blocks);
+    }
+
+    /// Take the pending submission, or null when none is queued. The caller
+    /// owns the returned slice: typically hand it to `run` (which adopts the
+    /// block contents) and free the slice itself with
+    /// `conversation.allocator`.
+    pub fn takeSubmission(self: *Agent) !?[]conversation.ContentBlock {
+        if (self._pending_submission.items.len == 0) return null;
+        return try self._pending_submission.toOwnedSlice(self.conversation.allocator);
+    }
+
     pub fn run(self: *Agent, message: UserMessage) !*Stream {
         self._auto_compacted = false;
 
diff --git a/src/tui_app.zig b/src/tui_app.zig
index d097225..98b94be 100644
--- a/src/tui_app.zig
+++ b/src/tui_app.zig
@@ -2008,6 +2008,12 @@ fn handleSubmittedLine(app: *App, term: *Terminal, line: []const u8, opts: RunOp
         if (captured.len != 0) {
             _ = try app.spawnStatus(captured);
         }
+        // Drain any user message the command queued via
+        // `panto.ext.agent:submit` and drive it as a native turn. A loop,
+        // not an if: a handler firing during that turn may queue another.
+        while (try opts.agent.takeSubmission()) |blocks| {
+            try runSubmission(app, term, opts, blocks);
+        }
         try app.renderNow();
         return;
     }
@@ -2038,6 +2044,56 @@ fn handleSubmittedLine(app: *App, term: *Terminal, line: []const u8, opts: RunOp
     try app.renderNow();
 }
 
+/// Drive one turn from user-message blocks queued by an extension
+/// (`panto.ext.agent:submit`). Mirrors the typed-line path: echo the
+/// message (fires `user_message`), honor a text override, resolve auth,
+/// drive the turn natively. Owns `queued` (allocated by `takeSubmission`
+/// with the conversation allocator): `run` adopts the block contents, the
+/// slice is freed here.
+fn runSubmission(app: *App, term: *Terminal, opts: RunOptions, queued: []panto.ContentBlock) !void {
+    const alloc = opts.agent.conversation.allocator;
+    var blocks = queued;
+    defer alloc.free(blocks);
+
+    // Echo the concatenated text blocks (a message with no text at all
+    // echoes as a placeholder).
+    var echo: std.ArrayList(u8) = .empty;
+    defer echo.deinit(app.alloc);
+    for (blocks) |b| switch (b) {
+        .Text => |t| {
+            if (echo.items.len != 0) try echo.append(app.alloc, '\n');
+            try echo.appendSlice(app.alloc, t.items);
+        },
+        else => {},
+    };
+    try app.spawnUser(if (echo.items.len != 0) echo.items else "[non-text message]"); // fires `user_message`
+
+    // A `user_message` handler override has the same authority as on a
+    // typed line: it replaces the queued message wholesale.
+    if (app.bus.takeOverride()) |t| {
+        defer app.alloc.free(t);
+        for (blocks) |*b| b.deinit(alloc);
+        alloc.free(blocks);
+        blocks = &.{}; // keep the defer safe if the realloc below fails
+        blocks = try alloc.alloc(panto.ContentBlock, 1);
+        blocks[0] = .{ .Text = try panto.textualBlockFromSlice(alloc, t) };
+    }
+
+    app.beginTurn();
+    try app.renderNow();
+    resolveAuthForTurn(app, opts, false) catch |err| {
+        // The turn never opened; the block contents are still ours.
+        for (blocks) |*b| b.deinit(alloc);
+        try app.routeError(err);
+        try app.renderNow();
+        return;
+    };
+    driveTurnBlocks(app, term, opts, blocks) catch |err| {
+        try app.routeError(err);
+    };
+    try app.renderNow();
+}
+
 /// The provider name (left of `provider:alias`) currently selected.
 fn currentProviderName(app: *App, opts: RunOptions) []const u8 {
     const label = if (app.selectors) |c| c.model_label else opts.model_label;
@@ -2115,7 +2171,13 @@ fn driveTurn(app: *App, term: *Terminal, opts: RunOptions, message_text: []const
     var blocks = [_]panto.ContentBlock{
         .{ .Text = try panto.textualBlockFromSlice(app.alloc, message_text) },
     };
-    var stream = try opts.agent.run(.{ .blocks = &blocks });
+    return driveTurnBlocks(app, term, opts, &blocks);
+}
+
+/// `driveTurn` for a pre-built user message. `run` adopts the block
+/// contents; the slice itself stays the caller's.
+fn driveTurnBlocks(app: *App, term: *Terminal, opts: RunOptions, blocks: []panto.ContentBlock) !void {
+    var stream = try opts.agent.run(.{ .blocks = blocks });
     defer stream.deinit();
     // Two single-shot fallbacks can re-open the SAME turn (no user-message
     // duplication):