diff options
| -rw-r--r-- | libpanto/src/agent.zig | 550 | ||||
| -rw-r--r-- | libpanto/src/null_store.zig | 97 | ||||
| -rw-r--r-- | libpanto/src/root.zig | 1 | ||||
| -rw-r--r-- | libpanto/src/session_store.zig | 10 | ||||
| -rw-r--r-- | libpanto/src/turn_persist.zig | 170 | ||||
| -rw-r--r-- | src/command.zig | 8 | ||||
| -rw-r--r-- | src/compaction.zig | 10 | ||||
| -rw-r--r-- | src/main.zig | 147 | ||||
| -rw-r--r-- | src/session_persist.zig | 153 | ||||
| -rw-r--r-- | src/system_prompt.zig | 114 |
10 files changed, 789 insertions, 471 deletions
diff --git a/libpanto/src/agent.zig b/libpanto/src/agent.zig index c7d0536..5a78722 100644 --- a/libpanto/src/agent.zig +++ b/libpanto/src/agent.zig @@ -36,6 +36,9 @@ const tool_mod = @import("tool.zig"); const image_mod = @import("image.zig"); const tool_source_mod = @import("tool_source.zig"); const tool_registry_mod = @import("tool_registry.zig"); +const session_store_mod = @import("session_store.zig"); +const null_store_mod = @import("null_store.zig"); +const turn_persist = @import("turn_persist.zig"); pub const Tool = tool_mod.Tool; pub const ToolSource = tool_source_mod.ToolSource; @@ -205,7 +208,6 @@ fn toolErrorResult( return tool_mod.ownedTextResult(allocator, msg); } - pub const Agent = struct { allocator: Allocator, io: Io, @@ -215,6 +217,24 @@ pub const Agent = struct { /// the visible tool set atomically. The pointee and its registry are /// owned by the embedder, not the agent. config: *const Config, + /// The live conversation the agent drives. Owned by the agent (adopted + /// at `init`); torn down in `deinit`. Turn-driving methods operate on + /// this directly rather than taking a `*Conversation` parameter. + conversation: conversation.Conversation, + /// Pluggable persistence backend. Every message that enters + /// `conversation` is persisted here as it is added. Defaults to a no-op + /// `NullStore` so embedders/tests that don't care about persistence + /// keep working. Borrowed: the embedder owns the concrete store and + /// must outlive the agent. + session_store: session_store_mod.SessionStore, + /// Display provider/model names used to stamp persisted entries. These + /// are the embedder's user-facing identity strings (e.g. the + /// `"provider:model"` ref's parts), which can't be derived from the + /// `ProviderConfig` snapshot (it only carries the `APIStyle` tag and + /// the raw model string). Borrowed; lifetime owned by the embedder. The + /// model defaults to the config snapshot's model when left empty. + persist_provider: []const u8 = "", + persist_model: []const u8 = "", /// Injectable streaming seam. Defaults to the real provider dispatch /// (`provider_mod.streamStep`); tests override it with a stub. stream_fn: provider_mod.StreamFn = provider_mod.streamStep, @@ -232,18 +252,52 @@ pub const Agent = struct { /// no synchronization is needed. retry_prng: ?std.Random.DefaultPrng = null, - pub fn init(allocator: Allocator, io: Io, config: *const Config) Agent { + /// Construct an agent. + /// + /// `store` is the persistence backend (use `null_store.store()` to opt + /// out). `maybe_conversation` is adopted (ownership transferred) when + /// non-null — the resume path: open a store, ask it for the + /// conversation, hand it here. When null, a fresh empty conversation is + /// created. Either way the agent owns and tears down the conversation. + pub fn init( + allocator: Allocator, + io: Io, + config: *const Config, + store: session_store_mod.SessionStore, + maybe_conversation: ?conversation.Conversation, + ) Agent { return .{ .allocator = allocator, .io = io, .config = config, + .conversation = maybe_conversation orelse conversation.Conversation.init(allocator), + .session_store = store, }; } pub fn deinit(self: *Agent) void { - // The agent owns neither the config snapshot nor the registry it - // borrows; the embedder tears those down. - _ = self; + // The agent owns the conversation; it borrows the config snapshot, + // the registry, and the session store, which the embedder tears + // down. + self.conversation.deinit(); + } + + /// The provider/model identity used to stamp persisted entries. Uses + /// the embedder-supplied display names when set, falling back to the + /// active config snapshot's model string (and the `APIStyle` tag name + /// for the provider) when not. + fn providerModel(self: *const Agent) struct { provider: []const u8, model: []const u8 } { + const cfg_model = switch (self.config.provider) { + inline else => |c| c.model, + }; + const provider = if (self.persist_provider.len > 0) + self.persist_provider + else switch (self.config.provider) { + .openai_chat => "openai_chat", + .anthropic_messages => "anthropic_messages", + }; + const model = if (self.persist_model.len > 0) self.persist_model else cfg_model; + return .{ .provider = provider, .model = model }; } /// Swap the active configuration snapshot. Takes effect at the start of @@ -259,18 +313,74 @@ pub const Agent = struct { return self.config.registry; } + /// Add a user message to the conversation and durably persist it + /// immediately (before any provider call). The user prompt is logged on + /// submission, so a crash before the model replies leaves a recoverable + /// dangling prompt in the store. + pub fn submitUserMessage(self: *Agent, text: []const u8) !void { + const start = self.conversation.messages.items.len; + try self.conversation.addUserMessage(text); + const pm = self.providerModel(); + try turn_persist.persistTurn( + self.allocator, + self.session_store, + &self.conversation, + start, + pm.provider, + pm.model, + ); + } + + /// Add a system message (append or replace mode) to the conversation + /// and persist it. The persisted entry records the mode so replay + /// reconstructs the same effective system prompt. + pub fn addSystemMessage( + self: *Agent, + text: []const u8, + mode: conversation.SystemMode, + ) !void { + const start = self.conversation.messages.items.len; + switch (mode) { + .append => try self.conversation.addSystemMessage(text), + .replace => try self.conversation.replaceSystemMessage(text), + } + const pm = self.providerModel(); + try turn_persist.persistTurn( + self.allocator, + self.session_store, + &self.conversation, + start, + pm.provider, + pm.model, + ); + } + /// Drive the conversation forward until the model stops calling tools. + /// + /// Persists everything the turn appends to the conversation: assistant + /// messages and tool-result user messages. Persistence runs on exit + /// regardless of how the turn ended — including error paths that + /// committed messages before failing — so a partial turn is durably + /// logged. An automatic compaction during the turn is persisted as a + /// fresh compaction window instead. pub fn runStep( self: *Agent, - conv: *conversation.Conversation, receiver: *provider_mod.Receiver, ) !void { self.auto_compacted = false; + const conv = &self.conversation; + const start = conv.messages.items.len; + + // Persist whatever the turn committed, on every exit path. + defer self.persistTurnTail(start) catch |e| { + std.log.err("session: failed to persist turn: {t}", .{e}); + }; + while (true) { // Re-read the config snapshot at the top of each turn so a // mid-conversation swap takes effect here, never mid-stream. const cfg = self.config; - try self.streamWithRetries(cfg, conv, receiver); + try self.streamWithRetries(cfg, receiver); const last = conv.messages.items[conv.messages.items.len - 1]; std.debug.assert(last.role == .assistant); @@ -282,7 +392,33 @@ pub const Agent = struct { if (!hasToolUseBlock(last)) return; - try self.dispatchToolCalls(conv, last); + try self.dispatchToolCalls(last); + } + } + + /// Persist the messages a turn produced. When the turn auto-compacted, + /// message indices shifted (the conversation was rewritten to + /// `[system..., summary, kept-suffix...]`), so persist the whole + /// post-compaction window instead of `[start..]`. + fn persistTurnTail(self: *Agent, start: usize) !void { + const pm = self.providerModel(); + if (self.auto_compacted) { + try turn_persist.persistCompaction( + self.allocator, + self.session_store, + &self.conversation, + pm.provider, + pm.model, + ); + } else { + try turn_persist.persistTurn( + self.allocator, + self.session_store, + &self.conversation, + start, + pm.provider, + pm.model, + ); } } @@ -312,16 +448,15 @@ pub const Agent = struct { fn streamWithRetries( self: *Agent, cfg: *const Config, - conv: *conversation.Conversation, receiver: *provider_mod.Receiver, ) !void { const policy = cfg.retry; var attempt: usize = 1; while (true) { var diag: provider_mod.ProviderDiagnostic = .{}; - self.stream_fn(self.allocator, self.io, cfg, conv, receiver, &diag) catch |err| { + self.stream_fn(self.allocator, self.io, cfg, &self.conversation, receiver, &diag) catch |err| { if (err == error.ContextOverflow) { - try self.handleContextOverflow(cfg, conv, receiver, err); + try self.handleContextOverflow(cfg, receiver, err); return; } if (!provider_mod.isRetryableProviderError(err)) return err; @@ -356,13 +491,12 @@ pub const Agent = struct { fn handleContextOverflow( self: *Agent, cfg: *const Config, - conv: *conversation.Conversation, receiver: *provider_mod.Receiver, err: anyerror, ) !void { if (self.auto_compacted) return err; // already retried once this turn const sys = self.compaction_system_prompt orelse return err; - const res = try self.compact(conv, sys, null); + const res = try self.compact(sys, null); if (!res.compacted) return err; // nothing to shed; give up self.auto_compacted = true; receiver.onProviderRetry(.{ @@ -375,7 +509,7 @@ pub const Agent = struct { // Retry the same request against the compacted context. A second // overflow (or any other error) propagates. var diag: provider_mod.ProviderDiagnostic = .{}; - try self.stream_fn(self.allocator, self.io, cfg, conv, receiver, &diag); + try self.stream_fn(self.allocator, self.io, cfg, &self.conversation, receiver, &diag); } /// Compute the backoff delay (ms) for the just-failed `attempt` @@ -422,9 +556,12 @@ pub const Agent = struct { /// Compact the conversation: summarize an older prefix into a single /// `.CompactionSummary` block and keep a recent suffix of whole turns - /// verbatim. Mutates `conv` in place. The embedder is responsible for - /// persisting the resulting new messages (the agent never touches the - /// session log). + /// verbatim. Mutates `self.conversation` in place. + /// + /// This is the pure transform — it does **not** persist. The explicit + /// `/compact` path uses `compactAndPersist`; the automatic + /// (context-overflow) path persists via `runStep`'s turn tail (the + /// rewritten window is logged as a fresh compaction window). /// /// The system prompt survives untouched: all `.system`-role messages /// are preserved in order, and no `replace` block is written. Only the @@ -442,10 +579,10 @@ pub const Agent = struct { /// embedder from its `COMPACTION.md` layers, or a built-in default). pub fn compact( self: *Agent, - conv: *conversation.Conversation, system_prompt: []const u8, extra_instructions: ?[]const u8, ) !CompactionResult { + const conv = &self.conversation; const messages = conv.messages.items; // Project per-message usage off the conversation for sizing. @@ -495,6 +632,30 @@ pub const Agent = struct { }; } + /// Compact and persist the result to the session store. This is the + /// explicit `/compact` entry point: it summarizes (via `compact`) and, + /// if anything was compacted, appends the new compaction window + /// (summary + restated kept suffix) to the store. Returns the same + /// `CompactionResult` for the embedder to report. + pub fn compactAndPersist( + self: *Agent, + system_prompt: []const u8, + extra_instructions: ?[]const u8, + ) !CompactionResult { + const res = try self.compact(system_prompt, extra_instructions); + if (res.compacted) { + const pm = self.providerModel(); + try turn_persist.persistCompaction( + self.allocator, + self.session_store, + &self.conversation, + pm.provider, + pm.model, + ); + } + return res; + } + /// Rewrite `conv.messages` to `[all system messages..., summary, /// kept-suffix...]`. The summarized conversation prefix (everything /// before `prefix_end` that isn't a system message) is dropped; system @@ -642,9 +803,9 @@ pub const Agent = struct { /// original call order. fn dispatchToolCalls( self: *Agent, - conv: *conversation.Conversation, assistant_msg: conversation.Message, ) !void { + const conv = &self.conversation; // Build the flat call list (in original order) and group calls // by owning registration. var calls: std.array_list.Managed(FlatCall) = .init(self.allocator); @@ -1354,6 +1515,141 @@ const NoopReceiver = struct { fn noop7(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {} }; +/// An in-memory `SessionStore` test double: records every appended +/// `DiskMessage` (role + provider/model stamp) so tests can assert the +/// agent persisted the right turn without touching disk. Honors the store +/// ownership contract by freeing each consumed message after recording its +/// salient fields. +const CapturingStore = struct { + allocator: Allocator, + roles: std.ArrayList(session_store_mod.DiskMessageRole) = .empty, + providers: std.ArrayList(?[]const u8) = .empty, + + fn init(allocator: Allocator) CapturingStore { + return .{ .allocator = allocator }; + } + + fn deinit(self: *CapturingStore) void { + for (self.providers.items) |p| if (p) |s| self.allocator.free(s); + self.providers.deinit(self.allocator); + self.roles.deinit(self.allocator); + } + + fn appendMessagesVT( + ctx: *anyopaque, + messages: []session_store_mod.DiskMessage, + providers: []const ?[]const u8, + _: []const ?[]const u8, + ) anyerror!void { + const self: *CapturingStore = @ptrCast(@alignCast(ctx)); + for (messages, 0..) |m, i| { + try self.roles.append(self.allocator, m.role); + // Prefer the assistant message's own provider stamp, else the + // entry-level provider stamp (user messages). + const prov = m.provider orelse providers[i]; + const dup: ?[]const u8 = if (prov) |p| try self.allocator.dupe(u8, p) else null; + try self.providers.append(self.allocator, dup); + } + // Contract: appends consume their messages. + for (messages) |m| m.deinit(self.allocator); + } + + fn loadConversationVT(_: *anyopaque, alloc: Allocator) anyerror!session_store_mod.LoadedSession { + return .{ .conversation = conversation.Conversation.init(alloc), .dangling_user = null }; + } + fn sessionIdVT(_: *anyopaque) []const u8 { + return "cap"; + } + fn activeModelVT(_: *anyopaque) ?session_store_mod.ActiveModel { + return null; + } + + const vtable: session_store_mod.SessionStore.VTable = .{ + .appendMessages = appendMessagesVT, + .loadConversation = loadConversationVT, + .sessionId = sessionIdVT, + .activeModel = activeModelVT, + }; + + fn store(self: *CapturingStore) session_store_mod.SessionStore { + return .{ .ptr = self, .vtable = &vtable }; + } +}; + +test "agent persists user, assistant, and tool-result messages of a turn" { + const allocator = testing.allocator; + + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{ + .{ .ToolUse = .{ .id = "tc_1", .name = "echo", .input = "hello" } }, + } }, + .{ .blocks = &.{ + .{ .Text = "ok" }, + } }, + }; + var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var h = TestHarness.init(allocator); + defer h.deinit(); + try h.registry.register(try EchoTool.create(allocator, "echo", "ECHO:")); + h.activate(); + + var cap = CapturingStore.init(allocator); + defer cap.deinit(); + var agent = Agent.init(allocator, io, &h.config, cap.store(), null); + defer agent.deinit(); + agent.stream_fn = stub.install(); + agent.persist_provider = "openai"; + agent.persist_model = "gpt-4o"; + + var recv = NoopReceiver.make(); + try agent.submitUserMessage("call a tool"); + try agent.runStep(&recv); + + // Persisted, in order: user prompt, assistant(ToolUse), user(ToolResult), + // assistant(text). + try testing.expectEqual(@as(usize, 4), cap.roles.items.len); + try testing.expectEqual(session_store_mod.DiskMessageRole.user, cap.roles.items[0]); + try testing.expectEqual(session_store_mod.DiskMessageRole.assistant, cap.roles.items[1]); + try testing.expectEqual(session_store_mod.DiskMessageRole.user, cap.roles.items[2]); + try testing.expectEqual(session_store_mod.DiskMessageRole.assistant, cap.roles.items[3]); + + // The display provider stamp rode through on every entry. + for (cap.providers.items) |p| { + try testing.expect(p != null); + try testing.expectEqualStrings("openai", p.?); + } +} + +test "agent runs a turn against NullStore without persisting or erroring" { + const allocator = testing.allocator; + + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{.{ .Text = "hi" }} }, + }; + var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var h = TestHarness.init(allocator); + defer h.deinit(); + h.activate(); + + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); + agent.stream_fn = stub.install(); + + var recv = NoopReceiver.make(); + try agent.submitUserMessage("hello"); + try agent.runStep(&recv); + + // Nothing crashed; the conversation has the user + assistant messages. + try testing.expectEqual(@as(usize, 2), agent.conversation.messages.items.len); +} + /// A configurable ToolSource for testing the grouped-dispatch path. /// Stores every batch it receives so tests can assert "calls X and Y /// arrived in the same batch on the same thread". @@ -1632,15 +1928,16 @@ test "runStep dispatches a tool call and loops to a final text turn" { defer h.deinit(); try h.registry.register(try EchoTool.create(allocator, "echo", "ECHO:")); h.activate(); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("call a tool"); var recv = NoopReceiver.make(); - try agent.runStep(&conv, &recv); + try agent.runStep(&recv); try testing.expectEqual(@as(usize, 4), conv.messages.items.len); @@ -1683,15 +1980,16 @@ test "runStep dispatches multiple tool calls in parallel" { try h.registry.register(try BarrierTool.create(allocator, "barrierB", &barrier)); try h.registry.register(try BarrierTool.create(allocator, "barrierC", &barrier)); h.activate(); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("go"); var recv = NoopReceiver.make(); - try agent.runStep(&conv, &recv); + try agent.runStep(&recv); const tr_msg = conv.messages.items[2]; try testing.expectEqual(@as(usize, 3), tr_msg.content.items.len); @@ -1723,15 +2021,16 @@ test "runStep: native tool handler error becomes an error result and the model g defer h.deinit(); try h.registry.register(try FailingTool.create(allocator, "boom")); h.activate(); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("break it"); var recv = NoopReceiver.make(); - try agent.runStep(&conv, &recv); + try agent.runStep(&recv); // user, assistant(tool_use), user(tool_result), assistant(text) try testing.expectEqual(@as(usize, 4), conv.messages.items.len); @@ -1757,15 +2056,16 @@ test "runStep: unknown tool becomes an error tool result and the loop continues" var h = TestHarness.init(allocator); defer h.deinit(); h.activate(); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("call a ghost"); var recv = NoopReceiver.make(); - try agent.runStep(&conv, &recv); + try agent.runStep(&recv); // messages: user, assistant(tool_use), user(tool_result), assistant(text) try testing.expectEqual(@as(usize, 4), conv.messages.items.len); @@ -1790,15 +2090,16 @@ test "runStep with no tool calls returns after one provider step" { var h = TestHarness.init(allocator); defer h.deinit(); h.activate(); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("hello"); var recv = NoopReceiver.make(); - try agent.runStep(&conv, &recv); + try agent.runStep(&recv); try testing.expectEqual(@as(usize, 2), conv.messages.items.len); try testing.expectEqualStrings("hi", conv.messages.items[1].content.items[0].Text.items); @@ -1817,15 +2118,16 @@ test "runStep surfaces EmptyAssistantResponse when provider commits an empty mes var h = TestHarness.init(allocator); defer h.deinit(); h.activate(); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("hi"); var recv = NoopReceiver.make(); - try testing.expectError(error.EmptyAssistantResponse, agent.runStep(&conv, &recv)); + try testing.expectError(error.EmptyAssistantResponse, agent.runStep(&recv)); } // ------------ ToolSource tests ------------ @@ -1849,15 +2151,16 @@ test "runStep delivers all source-backed calls in one batch on one thread" { defer h.deinit(); try h.registry.registerSource(try TestSource.create(allocator, "panto-lua", &.{ "lua_x", "lua_y" })); h.activate(); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("go"); var recv = NoopReceiver.make(); - try agent.runStep(&conv, &recv); + try agent.runStep(&recv); // Locate the source and inspect its observed batches. const view = h.registry.lookup("lua_x") orelse return error.NotFound; @@ -1901,15 +2204,16 @@ test "runStep: distinct sources run on distinct threads in parallel" { try h.registry.registerSource(try TestSource.create(allocator, "src_a", &.{"src_a_t"})); try h.registry.registerSource(try TestSource.create(allocator, "src_b", &.{"src_b_t"})); h.activate(); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("go"); var recv = NoopReceiver.make(); - try agent.runStep(&conv, &recv); + try agent.runStep(&recv); const view_a = h.registry.lookup("src_a_t") orelse return error.NotFound; const view_b = h.registry.lookup("src_b_t") orelse return error.NotFound; @@ -1940,15 +2244,16 @@ test "runStep: source whole-batch error becomes per-call error results and conti defer h.deinit(); try h.registry.registerSource(try FailingSource.create(allocator, "fs", &.{ "fa", "fb" })); h.activate(); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("kaboom"); var recv = NoopReceiver.make(); - try agent.runStep(&conv, &recv); + try agent.runStep(&recv); // user, assistant(tool_use x2), user(tool_result x2), assistant(text) try testing.expectEqual(@as(usize, 4), conv.messages.items.len); @@ -1984,15 +2289,16 @@ test "runStep: mixed single Tools and source-backed tools coexist in one turn" { try h.registry.register(try EchoTool.create(allocator, "single", "S:")); try h.registry.registerSource(try TestSource.create(allocator, "src", &.{ "src_t1", "src_t2" })); h.activate(); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("go"); var recv = NoopReceiver.make(); - try agent.runStep(&conv, &recv); + try agent.runStep(&recv); const tr_msg = conv.messages.items[2]; try testing.expectEqual(@as(usize, 3), tr_msg.content.items.len); @@ -2036,7 +2342,9 @@ test "setConfig swaps the visible tool set between turns" { .registry = ®_b, }; - var agent = Agent.init(allocator, io, &cfg_a); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &cfg_a, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); // Under A: `echo` visible, `late` not. @@ -2050,11 +2358,10 @@ test "setConfig swaps the visible tool set between turns" { // A real turn under B resolves `late` (which would have been // UnknownTool under A), then loops to the final text turn. - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("go"); var recv = NoopReceiver.make(); - try agent.runStep(&conv, &recv); + try agent.runStep(&recv); const tr = conv.messages.items[2].content.items[0].ToolResult; try testing.expectEqualStrings("2", tr.tool_use_id); @@ -2079,11 +2386,12 @@ test "compact: summarizes prefix, keeps suffix, system survives" { // 3 words each => ceil(3*1.3)=4 tokens each => 8 total <= 10, while // adding the longer first turn exceeds it. h.config.compaction = .{ .keep_verbatim = 10 }; - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addSystemMessage("you are helpful"); try conv.addUserMessage("first question here with several words"); try conv.addAssistantMessage(&.{ @@ -2094,7 +2402,7 @@ test "compact: summarizes prefix, keeps suffix, system survives" { .{ .Text = try conversation.textualBlockFromSlice(allocator, "second recent answer") }, }); - const res = try agent.compact(&conv, "Summarize the conversation.", null); + const res = try agent.compact("Summarize the conversation.", null); try testing.expect(res.compacted); // Expected rebuilt: [system, compaction summary(user), user q2, asst a2] @@ -2133,18 +2441,19 @@ test "compact: no-op when conversation already fits the budget" { defer h.deinit(); h.activate(); h.config.compaction = .{ .keep_verbatim = 1_000_000 }; - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addSystemMessage("sys"); try conv.addUserMessage("hi"); try conv.addAssistantMessage(&.{ .{ .Text = try conversation.textualBlockFromSlice(allocator, "hello") }, }); - const res = try agent.compact(&conv, "Summarize.", null); + const res = try agent.compact("Summarize.", null); try testing.expect(!res.compacted); try testing.expectEqual(@as(usize, 3), conv.messages.items.len); // Stub was never consumed. @@ -2169,11 +2478,12 @@ test "compact: extra instructions are appended to the system prompt" { defer h.deinit(); h.activate(); h.config.compaction = .{ .keep_verbatim = 1 }; - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("question one two three"); try conv.addAssistantMessage(&.{ .{ .Text = try conversation.textualBlockFromSlice(allocator, "answer one two three") }, @@ -2183,7 +2493,7 @@ test "compact: extra instructions are appended to the system prompt" { .{ .Text = try conversation.textualBlockFromSlice(allocator, "answer two") }, }); - const res = try agent.compact(&conv, "Base prompt.", "keep bug #3 details"); + const res = try agent.compact("Base prompt.", "keep bug #3 details"); try testing.expect(res.compacted); } @@ -2208,12 +2518,13 @@ test "runStep: auto-compacts on context overflow and retries once" { defer h.deinit(); h.activate(); h.config.compaction = .{ .keep_verbatim = 10 }; - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); agent.compaction_system_prompt = "Summarize the conversation."; - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addSystemMessage("you are helpful"); try conv.addUserMessage("first question with several words here"); try conv.addAssistantMessage(&.{ @@ -2222,7 +2533,7 @@ test "runStep: auto-compacts on context overflow and retries once" { try conv.addUserMessage("second recent question"); var recv = NoopReceiver.make(); - try agent.runStep(&conv, &recv); + try agent.runStep(&recv); try testing.expect(agent.auto_compacted); // After compaction + retry: [system, summary, user q2, assistant final]. @@ -2253,16 +2564,17 @@ test "runStep: context overflow without compaction prompt propagates" { var h = TestHarness.init(allocator); defer h.deinit(); h.activate(); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); // No compaction_system_prompt set -> overflow propagates. - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("hi"); var recv = NoopReceiver.make(); - try testing.expectError(error.ContextOverflow, agent.runStep(&conv, &recv)); + try testing.expectError(error.ContextOverflow, agent.runStep(&recv)); } // ----------------------------------------------------------------------------- @@ -2339,17 +2651,18 @@ test "runStep: provider 429 retries then succeeds without duplicate messages" { var h = TestHarness.init(allocator); defer h.deinit(); fastRetryHarness(&h); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("hi"); var rr = RetryRecordingReceiver{ .allocator = allocator }; defer rr.deinit(); var recv = rr.make(); - try agent.runStep(&conv, &recv); + try agent.runStep(&recv); // Two failures + one success. try testing.expectEqual(@as(usize, 3), stub.calls_made); @@ -2383,17 +2696,18 @@ test "runStep: provider 500 retries with backoff notification" { var h = TestHarness.init(allocator); defer h.deinit(); fastRetryHarness(&h); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("hi"); var rr = RetryRecordingReceiver{ .allocator = allocator }; defer rr.deinit(); var recv = rr.make(); - try agent.runStep(&conv, &recv); + try agent.runStep(&recv); try testing.expectEqual(@as(usize, 2), stub.calls_made); try testing.expectEqual(@as(usize, 1), rr.infos.items.len); @@ -2422,17 +2736,18 @@ test "runStep: provider auth failure does not retry" { var h = TestHarness.init(allocator); defer h.deinit(); fastRetryHarness(&h); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("hi"); var rr = RetryRecordingReceiver{ .allocator = allocator }; defer rr.deinit(); var recv = rr.make(); - try testing.expectError(error.ProviderAuthFailed, agent.runStep(&conv, &recv)); + try testing.expectError(error.ProviderAuthFailed, agent.runStep(&recv)); // Exactly one attempt, no retry notification. try testing.expectEqual(@as(usize, 1), stub.calls_made); @@ -2462,17 +2777,18 @@ test "runStep: retries exhaust and hard-fail after max_attempts" { var h = TestHarness.init(allocator); defer h.deinit(); fastRetryHarness(&h); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("hi"); var rr = RetryRecordingReceiver{ .allocator = allocator }; defer rr.deinit(); var recv = rr.make(); - try testing.expectError(error.ProviderUnavailable, agent.runStep(&conv, &recv)); + try testing.expectError(error.ProviderUnavailable, agent.runStep(&recv)); // 4 attempts total (max_attempts), 3 retry notifications. try testing.expectEqual(@as(usize, 4), stub.calls_made); @@ -2501,17 +2817,18 @@ test "runStep: Retry-After is honored and reported" { h.activate(); // Cap below the Retry-After to verify the policy cap applies. h.config.retry = .{ .initial_delay_ms = 0, .max_delay_ms = 1, .jitter = false }; - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("hi"); var rr = RetryRecordingReceiver{ .allocator = allocator }; defer rr.deinit(); var recv = rr.make(); - try agent.runStep(&conv, &recv); + try agent.runStep(&recv); try testing.expectEqual(@as(usize, 1), rr.infos.items.len); // Reported Retry-After is the raw provider value... @@ -2537,15 +2854,16 @@ test "runStep: cancellation from a tool still hard-fails" { defer h.deinit(); try h.registry.register(try HardFailTool.create(allocator, "hard")); h.activate(); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("go"); var recv = NoopReceiver.make(); - try testing.expectError(error.Canceled, agent.runStep(&conv, &recv)); + try testing.expectError(error.Canceled, agent.runStep(&recv)); // Turn aborts: no tool result appended (user + assistant only). try testing.expectEqual(@as(usize, 2), conv.messages.items.len); } @@ -2568,15 +2886,16 @@ test "runStep: source per-call error produces a per-call error result and contin defer h.deinit(); try h.registry.registerSource(try PartialSource.create(allocator, "ps", &.{ "pa", "pb" })); h.activate(); - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addUserMessage("go"); var recv = NoopReceiver.make(); - try agent.runStep(&conv, &recv); + try agent.runStep(&recv); try testing.expectEqual(@as(usize, 4), conv.messages.items.len); const tr_msg = conv.messages.items[2]; @@ -2607,12 +2926,13 @@ test "runStep: context-overflow compaction fires a compaction retry notification defer h.deinit(); h.activate(); h.config.compaction = .{ .keep_verbatim = 10 }; - var agent = Agent.init(allocator, io, &h.config); + var ns = null_store_mod.NullStore.init(allocator); + var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + defer agent.deinit(); agent.stream_fn = stub.install(); agent.compaction_system_prompt = "Summarize the conversation."; - var conv = conversation.Conversation.init(allocator); - defer conv.deinit(); + const conv = &agent.conversation; try conv.addSystemMessage("you are helpful"); try conv.addUserMessage("first question with several words here"); try conv.addAssistantMessage(&.{ @@ -2623,7 +2943,7 @@ test "runStep: context-overflow compaction fires a compaction retry notification var rr = RetryRecordingReceiver{ .allocator = allocator }; defer rr.deinit(); var recv = rr.make(); - try agent.runStep(&conv, &recv); + try agent.runStep(&recv); try testing.expect(agent.auto_compacted); // Exactly one notification, flagged as a compaction retry with no delay. diff --git a/libpanto/src/null_store.zig b/libpanto/src/null_store.zig index a9bd813..d2337ee 100644 --- a/libpanto/src/null_store.zig +++ b/libpanto/src/null_store.zig @@ -1,11 +1,16 @@ //! `NullStore`: a no-op `SessionStore` for embedders who opt out of //! persistence (and the default backing for an `Agent` constructed without -//! an explicit store). +//! a real store). //! -//! Every append is dropped. `loadConversation` returns an empty -//! conversation with no dangling prompt. `activeModel` is null and the -//! session id is the empty string. The store is stateless, so a single -//! process-global instance backs every `NullStore` handle. +//! Every append is dropped (but its messages are freed, honoring the store +//! ownership contract — appends consume their messages). `loadConversation` +//! returns an empty conversation with no dangling prompt. `activeModel` is +//! null and the session id is the empty string. +//! +//! Because freeing the dropped messages needs an allocator, `NullStore` +//! carries one. Construct it with `init(alloc)` and call `.store()` for the +//! interface handle. The struct is trivially copyable and holds no other +//! state, so the handle may borrow it for the agent's lifetime. const std = @import("std"); const Allocator = std.mem.Allocator; @@ -18,54 +23,64 @@ const LoadedSession = session_store_mod.LoadedSession; const DiskMessage = session_store_mod.DiskMessage; const ActiveModel = session_store_mod.ActiveModel; -/// Stateless singleton context. The vtable ignores `ctx` entirely; we hand -/// out a pointer to this so the `*anyopaque` is always valid. -var singleton: u8 = 0; +pub const NullStore = struct { + allocator: Allocator, -fn appendMessagesVT( - _: *anyopaque, - _: []DiskMessage, - _: []const ?[]const u8, - _: []const ?[]const u8, -) anyerror!void {} + pub fn init(allocator: Allocator) NullStore { + return .{ .allocator = allocator }; + } -fn loadConversationVT(_: *anyopaque, alloc: Allocator) anyerror!LoadedSession { - return .{ - .conversation = conversation_mod.Conversation.init(alloc), - .dangling_user = null, - }; -} + fn appendMessagesVT( + ctx: *anyopaque, + messages: []DiskMessage, + _: []const ?[]const u8, + _: []const ?[]const u8, + ) anyerror!void { + const self: *NullStore = @ptrCast(@alignCast(ctx)); + // Appends consume their messages; we drop them, so free them. + for (messages) |m| m.deinit(self.allocator); + } -fn sessionIdVT(_: *anyopaque) []const u8 { - return ""; -} + fn loadConversationVT(ctx: *anyopaque, alloc: Allocator) anyerror!LoadedSession { + _ = ctx; + return .{ + .conversation = conversation_mod.Conversation.init(alloc), + .dangling_user = null, + }; + } -fn activeModelVT(_: *anyopaque) ?ActiveModel { - return null; -} + fn sessionIdVT(_: *anyopaque) []const u8 { + return ""; + } -const vtable: SessionStore.VTable = .{ - .appendMessages = appendMessagesVT, - .loadConversation = loadConversationVT, - .sessionId = sessionIdVT, - .activeModel = activeModelVT, -}; + fn activeModelVT(_: *anyopaque) ?ActiveModel { + return null; + } -/// A no-op `SessionStore`. Cheap to call repeatedly; all handles share one -/// stateless context. -pub fn store() SessionStore { - return .{ .ptr = &singleton, .vtable = &vtable }; -} + const vtable: SessionStore.VTable = .{ + .appendMessages = appendMessagesVT, + .loadConversation = loadConversationVT, + .sessionId = sessionIdVT, + .activeModel = activeModelVT, + }; + + /// Wrap this `NullStore` as a `SessionStore`. The handle borrows + /// `self`; `self` must outlive it. + pub fn store(self: *NullStore) SessionStore { + return .{ .ptr = self, .vtable = &vtable }; + } +}; const testing = std.testing; -test "NullStore: appends are dropped and load returns empty" { +test "NullStore: appends are dropped (and freed) and load returns empty" { + var ns = NullStore.init(testing.allocator); + const s = ns.store(); + var msg_content = try testing.allocator.alloc(session_store_mod.DiskContentBlock, 1); msg_content[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hi") } }; var messages = [_]DiskMessage{.{ .role = .user, .content = msg_content }}; - defer messages[0].deinit(testing.allocator); - - const s = store(); + // No defer-free: the store consumes (frees) the message. try s.appendMessages(&messages, &.{null}, &.{null}); var loaded = try s.loadConversation(testing.allocator); diff --git a/libpanto/src/root.zig b/libpanto/src/root.zig index 39797dd..a49a428 100644 --- a/libpanto/src/root.zig +++ b/libpanto/src/root.zig @@ -12,6 +12,7 @@ pub const session = @import("session.zig"); pub const session_manager = @import("session_manager.zig"); pub const session_store = @import("session_store.zig"); pub const null_store = @import("null_store.zig"); +pub const turn_persist = @import("turn_persist.zig"); pub const pricing = @import("pricing.zig"); pub const compaction = @import("compaction.zig"); pub const image = @import("image.zig"); diff --git a/libpanto/src/session_store.zig b/libpanto/src/session_store.zig index 83bf136..b929c66 100644 --- a/libpanto/src/session_store.zig +++ b/libpanto/src/session_store.zig @@ -85,8 +85,14 @@ pub const SessionStore = struct { /// Append a batch of messages atomically. `providers`/`models` are /// parallel arrays (one per message) carrying the top-level entry /// stamps (recorded on user entries). A single append is a - /// length-1 batch. Implementations consume nothing — the caller - /// retains ownership of `messages` and frees them after return. + /// length-1 batch. + /// + /// Ownership: the store **consumes** each `DiskMessage` on success + /// (takes ownership of its heap allocations). On error the store + /// frees any messages it had already taken and the caller frees the + /// rest — i.e. after this call returns the caller must not free the + /// `DiskMessage`s regardless of outcome. (The `messages` *slice* + /// itself remains the caller's.) appendMessages: *const fn ( ctx: *anyopaque, messages: []DiskMessage, diff --git a/libpanto/src/turn_persist.zig b/libpanto/src/turn_persist.zig new file mode 100644 index 0000000..a8b3faa --- /dev/null +++ b/libpanto/src/turn_persist.zig @@ -0,0 +1,170 @@ +//! Turn → session-log persistence: map in-memory `Conversation` messages +//! to neutral `DiskMessage`s and append them to a `SessionStore`. +//! +//! This logic used to live in the `panto` CLI (`src/session_persist.zig`), +//! driven externally around `runStep`. It now lives in `libpanto` and is +//! called by the `Agent` itself, so every embedder gets persistence for +//! free. The functions here are stateless helpers over a `SessionStore`; +//! the agent owns the store and the conversation. +//! +//! Per-message usage is read directly off `Message.usage` (canonical: both +//! providers stamp it via `addAssistantMessageWithUsage`). There is no +//! separate usage list to thread through. + +const std = @import("std"); +const Allocator = std.mem.Allocator; + +const conversation = @import("conversation.zig"); +const session = @import("session.zig"); +const session_store = @import("session_store.zig"); + +const DiskMessage = session_store.DiskMessage; +const DiskContentBlock = session_store.DiskContentBlock; +const SessionStore = session_store.SessionStore; + +/// Persist every conversation message at index `>= start_index` to `store` +/// as a single atomic batch. +/// +/// Mapping: +/// - assistant → assistant entry with provider/model/stop_reason and the +/// message's own `usage`. +/// - user (incl. ToolResult-only messages) → user entry stamped with +/// provider/model. +/// - system → system entry, verbatim, unstamped (mid-turn system +/// messages aren't produced by pantograph today, but persist harmless). +/// +/// An assistant message carrying a ToolUse with no following matching +/// ToolResult is skipped (a dangling tool call from an interrupted turn); +/// persisting it would make the log un-replayable. +/// +/// `stop_reason` is `"stop"` for now; the real wire value needs provider +/// plumbing (separate future work). +pub fn persistTurn( + alloc: Allocator, + store: SessionStore, + conv: *const conversation.Conversation, + start_index: usize, + provider: []const u8, + model: []const u8, +) !void { + // Ownership note: the messages we build here are handed to + // `store.appendMessages`, which *consumes* them. We therefore never + // free entries in `batch_messages` ourselves — only the parallel + // metadata arrays and the per-block transient on the error path before + // a message is appended to the batch. This mirrors the original CLI + // `persistTurn` exactly (it transferred ownership to the manager and + // never freed the DiskMessages). + var batch_messages: std.ArrayList(DiskMessage) = .empty; + defer batch_messages.deinit(alloc); + var batch_providers: std.ArrayList(?[]const u8) = .empty; + defer batch_providers.deinit(alloc); + var batch_models: std.ArrayList(?[]const u8) = .empty; + defer batch_models.deinit(alloc); + + var i = start_index; + while (i < conv.messages.items.len) : (i += 1) { + const msg = conv.messages.items[i]; + const blocks = try alloc.alloc(DiskContentBlock, msg.content.items.len); + var allocated: usize = 0; + errdefer { + for (blocks[0..allocated]) |b| b.deinit(alloc); + alloc.free(blocks); + } + for (msg.content.items) |block| { + blocks[allocated] = try session.contentBlockToDisk(alloc, block); + allocated += 1; + } + if (msg.role == .assistant and hasToolUseWithoutFollowingResults(conv, i)) { + for (blocks[0..allocated]) |b| b.deinit(alloc); + alloc.free(blocks); + continue; + } + switch (msg.role) { + .system => { + // The System block carries the append/replace mode; it + // rides on `DiskMessage.mode`, not on the disk block. Derive + // it from the in-memory message so reconciliation on replay + // reconstructs the same effective prompt. + try batch_messages.append(alloc, .{ + .role = .system, + .mode = systemModeOf(msg), + .content = blocks, + }); + try batch_providers.append(alloc, null); + try batch_models.append(alloc, null); + }, + .user => { + try batch_messages.append(alloc, .{ .role = .user, .content = blocks }); + try batch_providers.append(alloc, provider); + try batch_models.append(alloc, model); + }, + .assistant => { + try batch_messages.append(alloc, .{ + .role = .assistant, + .content = blocks, + .provider = try alloc.dupe(u8, provider), + .model = try alloc.dupe(u8, model), + .stop_reason = try alloc.dupe(u8, "stop"), + .usage = msg.usage, + }); + try batch_providers.append(alloc, null); + try batch_models.append(alloc, null); + }, + } + } + + if (batch_messages.items.len == 0) return; + try store.appendMessages(batch_messages.items, batch_providers.items, batch_models.items); +} + +/// Persist a compaction result. The agent rewrote the conversation to +/// `[system..., summary, kept-suffix...]`; persist everything from the +/// latest compaction summary onward as fresh entries. On replay the latest +/// summary resets effective context, so the duplicated suffix is what +/// survives. Usage isn't re-derived for the restated suffix (those turns +/// were already sized when first logged). +pub fn persistCompaction( + alloc: Allocator, + store: SessionStore, + conv: *const conversation.Conversation, + provider: []const u8, + model: []const u8, +) !void { + const start = conversation.latestCompactionIndex(conv.messages.items) orelse return; + try persistTurn(alloc, store, conv, start, provider, model); +} + +/// Derive the disk system-mode from a system message's blocks. A `.System` +/// block carries the mode; a `replace` anywhere makes the message a +/// replace. Defaults to `append`. +fn systemModeOf(msg: conversation.Message) session.DiskSystemMode { + for (msg.content.items) |block| { + if (block == .System and block.System.mode == .replace) return .replace; + } + return .append; +} + +/// True when the assistant message at `index` contains a ToolUse block +/// with no matching ToolResult in the immediately following user message +/// — i.e. a dangling tool call (e.g. a turn cut short by an error). +pub fn hasToolUseWithoutFollowingResults( + conv: *const conversation.Conversation, + index: usize, +) bool { + const msg = conv.messages.items[index]; + for (msg.content.items) |block| { + if (block != .ToolUse) continue; + if (index + 1 >= conv.messages.items.len) return true; + const next = conv.messages.items[index + 1]; + if (next.role != .user) return true; + var found = false; + for (next.content.items) |next_block| { + if (next_block == .ToolResult and std.mem.eql(u8, next_block.ToolResult.tool_use_id, block.ToolUse.id)) { + found = true; + break; + } + } + if (!found) return true; + } + return false; +} diff --git a/src/command.zig b/src/command.zig index 26acc7a..49ad5b8 100644 --- a/src/command.zig +++ b/src/command.zig @@ -32,12 +32,10 @@ const panto = @import("panto"); pub const Context = struct { allocator: std.mem.Allocator, - /// In-memory conversation the agent is driving. - conv: *panto.conversation.Conversation, - /// The active agent (model/provider/tooling snapshot). + /// The active agent. Owns the conversation (`agent.conversation`) and + /// the session store (`agent.session_store`); commands reach both + /// through it rather than holding separate pointers. agent: *panto.agent.Agent, - /// Session log to persist any changes a command makes. - session_mgr: *panto.session_manager.SessionManager, /// REPL output writer and its backing file writer (for flushing). stdout: *std.Io.Writer, diff --git a/src/compaction.zig b/src/compaction.zig index a1f6e89..4ec5732 100644 --- a/src/compaction.zig +++ b/src/compaction.zig @@ -8,7 +8,6 @@ const std = @import("std"); const panto = @import("panto"); const command = @import("command.zig"); -const session_persist = @import("session_persist.zig"); /// Install the `/compact` command into `registry`. pub fn register(registry: *command.Registry) !void { @@ -23,20 +22,13 @@ pub fn register(registry: *command.Registry) !void { fn run(args: []const u8, ctx: *command.Context) anyerror!void { const extra: ?[]const u8 = if (args.len > 0) args else null; - const res = ctx.agent.compact(ctx.conv, ctx.compaction_prompt, extra) catch |err| { + const res = ctx.agent.compactAndPersist(ctx.compaction_prompt, extra) catch |err| { try ctx.stdout.print("\n[compaction failed: {s}]\n> ", .{@errorName(err)}); try ctx.stdout_file.flush(); return; }; if (res.compacted) { - try session_persist.persistCompaction( - ctx.allocator, - ctx.session_mgr, - ctx.conv, - ctx.provider_name, - ctx.model_name, - ); try ctx.stdout.print( "\n[compacted: summarized {d} message(s), kept {d} recent turn(s)]\n> ", .{ res.summarized_messages, res.kept_turns }, diff --git a/src/main.zig b/src/main.zig index 61e78d5..9059a6f 100644 --- a/src/main.zig +++ b/src/main.zig @@ -14,7 +14,6 @@ const glob = @import("glob.zig"); const system_prompt = @import("system_prompt.zig"); const command = @import("command.zig"); const command_compaction = @import("compaction.zig"); -const session_persist = @import("session_persist.zig"); // Shorthand alias for the Lua C API. The bridge module owns the actual // `@cImport`; we re-use it here so the smoke check uses identical types. @@ -44,7 +43,6 @@ test { _ = system_prompt; _ = command; _ = command_compaction; - _ = session_persist; } const Receiver = panto.provider.Receiver; @@ -55,34 +53,24 @@ const MessageRole = panto.conversation.MessageRole; /// Receiver that prints streaming deltas to stdout. Thinking blocks are /// dimmed with ANSI escape codes; text blocks render plain. /// -/// Also captures one `?Usage` per assistant response during a turn. In -/// a tool-using turn the agent loop drives multiple streamStep calls; -/// each one ends with `onMessageComplete(msg, usage)`. We append the -/// usage — including `null` when the wire didn't deliver any — to -/// `per_message_usage` in order so `persistTurn` can pair each captured -/// usage with its corresponding assistant message. +/// Persistence is owned by the agent now; the receiver is display-only. const CLIReceiver = struct { stdout: *std.Io.Writer, file: *std.Io.File.Writer, allocator: std.mem.Allocator, - /// One slot per assistant message completed during the current - /// turn, in completion order. `null` means the provider did not - /// report usage on the wire for that message (typical of - /// OpenAI-compatible proxies that ignore `stream_options.include_usage`). - per_message_usage: std.ArrayList(?panto.session_manager.Usage) = .empty, - pub fn receiver(self: *CLIReceiver) Receiver { return .{ .ptr = self, .vtable = &vtable }; } - /// Reset usage state at the start of each turn. + /// Per-turn reset hook. Currently a no-op (no per-turn receiver state), + /// retained as the REPL's turn-boundary signal. pub fn beginTurn(self: *CLIReceiver) void { - self.per_message_usage.clearRetainingCapacity(); + _ = self; } pub fn deinit(self: *CLIReceiver) void { - self.per_message_usage.deinit(self.allocator); + _ = self; } const vtable: ReceiverVTable = .{ @@ -184,13 +172,8 @@ const CLIReceiver = struct { usage: ?panto.session_manager.Usage, ) anyerror!void { const self: *CLIReceiver = @ptrCast(@alignCast(ptr)); - // Only assistant messages come through streaming. The receiver - // contract says onMessageComplete fires exactly once per - // streamStep, with role=.assistant; record the usage slot in - // turn order regardless of whether the wire actually had usage. - if (message.role == .assistant) { - try self.per_message_usage.append(self.allocator, usage); - } + _ = message; + _ = usage; try self.stdout.writeAll("\n"); try self.file.flush(); } @@ -403,17 +386,19 @@ pub fn main(init: std.process.Init) !void { }; defer session_mgr.deinit(); - var conv = panto.conversation.Conversation.init(alloc); - defer conv.deinit(); - const is_resume = session_mgr.getEntries().len > 0; + + // On resume, reconstruct the conversation from the store. The dangling + // user prompt (a trailing user entry with no assistant reply) is split + // out and currently ignored (a future TUI will prefill it for editing). + // On a fresh session, the agent starts with an empty conversation. + var adopted_conversation: ?panto.conversation.Conversation = null; if (is_resume) { - // Resumed an existing session — rebuild the conversation from the - // log. The system prompt is part of the log. (Reconciliation - // against the on-disk SYSTEM.md happens after the agent tree is - // staged, below.) - conv.deinit(); - conv = try session_mgr.rebuildConversation(); + var loaded = try session_mgr.loadConversation(alloc); + // We don't use the dangling prompt yet; free it. + if (loaded.dangling_user) |d| alloc.free(d); + loaded.dangling_user = null; + adopted_conversation = loaded.conversation; try stdout.print( "resumed session {s} ({d} entries)\n", .{ session_mgr.getSessionId()[0..@min(8, session_mgr.getSessionId().len)], session_mgr.getEntries().len }, @@ -421,11 +406,33 @@ pub fn main(init: std.process.Init) !void { } // The tool registry is owned here and referenced by the active config - // snapshot. Extensions register their tool source into it *before* the - // snapshot is built, so the agent's first turn sees them. + // snapshot. Extensions register their tool source into it *after* the + // agent is built; the agent holds a `*const Config` pointing at this + // registry, so in-place registration before the first turn is visible. var registry = panto.ToolRegistry.init(alloc); defer registry.deinit(); + // Assemble the active configuration snapshot and build the agent now, + // before luarocks bootstrap and extension loading. The agent adopts the + // store (for free persistence) and the resumed conversation (or starts + // fresh). System-prompt seeding/reconciliation below runs *through* the + // agent so those entries persist. + const active_config: panto.config.Config = .{ + .provider = provider_config, + .registry = ®istry, + .compaction = compaction_cfg, + }; + var agent = panto.agent.Agent.init( + alloc, + io, + &active_config, + session_mgr.store(), + adopted_conversation, + ); + defer agent.deinit(); + agent.persist_provider = banner_provider_initial; + agent.persist_model = banner_model_initial; + // Spin up the long-lived Lua runtime. All Lua extensions load into // one `lua_State`; module-global state survives across calls. The // runtime registers with the agent as a single `ToolSource` named @@ -461,8 +468,7 @@ pub fn main(init: std.process.Init) !void { io, init.environ_map, luarocks_rt.layout.agent_dir, - &conv, - &session_mgr, + &agent, ); } else { // Fresh session — source and install the system prompt. @@ -471,8 +477,7 @@ pub fn main(init: std.process.Init) !void { io, init.environ_map, luarocks_rt.layout.agent_dir, - &conv, - &session_mgr, + &agent, ); } @@ -506,17 +511,6 @@ pub fn main(init: std.process.Init) !void { try registry.registerSource(rt.toolSource()); } - // Assemble the active configuration snapshot (provider + tool registry) - // and create the agent holding a pointer to it. Swapping this pointer - // later (model/provider/tool changes) takes effect at the next turn. - const active_config: panto.config.Config = .{ - .provider = provider_config, - .registry = ®istry, - .compaction = compaction_cfg, - }; - var agent = panto.agent.Agent.init(alloc, io, &active_config); - defer agent.deinit(); - // Resolve the compaction system prompt (COMPACTION.md across layers, // last wins; built-in default otherwise) and arm automatic compaction. // Owned by `sp_arena`, which lives for the whole REPL. @@ -572,9 +566,7 @@ pub fn main(init: std.process.Init) !void { var cmd_ctx: command.Context = .{ .allocator = alloc, - .conv = &conv, .agent = &agent, - .session_mgr = &session_mgr, .stdout = stdout, .stdout_file = &stdout_file, .compaction_prompt = compaction_prompt, @@ -617,54 +609,18 @@ pub fn main(init: std.process.Init) !void { continue; } - try conv.addUserMessage(line); - try session_persist.appendUserPromptToSession( - alloc, - &session_mgr, - line, - banner_provider_initial, - banner_model_initial, - ); - - const entries_before_step = conv.messages.items.len; + // Submit + persist the user prompt, then drive the turn. The agent + // owns the conversation and persists everything it generates + // (assistant/tool-result messages, and the compaction window on + // auto-compaction) through its session store — the CLI no longer + // touches persistence. + try agent.submitUserMessage(line); cli_recv.beginTurn(); - agent.runStep(&conv, &recv) catch |err| { + agent.runStep(&recv) catch |err| { try stdout.print("\n[error: {s}]\n", .{@errorName(err)}); }; - if (agent.auto_compacted) { - // Automatic compaction rewrote the in-memory conversation, so - // message indices no longer align with `entries_before_step`. - // Persist the whole post-compaction window (summary + duplicated - // kept suffix + the retried turn) as fresh entries; the - // superseded prefix stays on disk and is ignored on replay. - try session_persist.persistCompaction( - alloc, - &session_mgr, - &conv, - banner_provider_initial, - banner_model_initial, - ); - } else { - // Persist whatever new entries the agent produced this turn. The - // agent loop may have appended: - // - assistant message(s) (one per provider response) - // - user messages containing ToolResult blocks (one per tool round) - // Each assistant message gets paired with the Usage that the - // receiver captured at its onMessageComplete time (or null if - // the provider didn't emit usage that round). - try session_persist.persistTurn( - alloc, - &session_mgr, - &conv, - entries_before_step, - banner_provider_initial, - banner_model_initial, - cli_recv.per_message_usage.items, - ); - } - try stdout.writeAll("\n> "); try stdout_file.flush(); } @@ -784,4 +740,3 @@ fn openSession( }, } } - diff --git a/src/session_persist.zig b/src/session_persist.zig deleted file mode 100644 index d5349bc..0000000 --- a/src/session_persist.zig +++ /dev/null @@ -1,153 +0,0 @@ -//! Session-log persistence helpers shared between the REPL loop -//! (`main.zig`) and slash-command handlers (e.g. `compaction.zig`). -//! -//! These were previously private to `main.zig`. They were lifted into -//! their own module so that command handlers living outside the root -//! source file can reuse them without depending on `main.zig` (which, -//! being the executable root, can't be imported as a normal module). - -const std = @import("std"); -const panto = @import("panto"); - -/// Append a single user-prompt message to the session log. -pub fn appendUserPromptToSession( - alloc: std.mem.Allocator, - mgr: *panto.session_manager.SessionManager, - text: []const u8, - provider: []const u8, - model: []const u8, -) !void { - const blocks = try alloc.alloc(panto.session_manager.DiskContentBlock, 1); - blocks[0] = .{ .text = .{ .text = try alloc.dupe(u8, text) } }; - _ = try mgr.appendMessage( - .{ .role = .user, .content = blocks }, - provider, - model, - ); -} - -/// After the agent loop has driven a turn to completion, persist every -/// new in-memory message at index `>= start_index` to the session log. -/// -/// Each in-memory message is mapped to disk: -/// - assistant → assistant entry with provider/model/stop_reason metadata -/// and Usage (from `per_message_usage`, one slot per assistant message in -/// completion order). -/// - user (with ToolResult blocks) → user entry stamped with provider/model. -/// -/// `stop_reason` is recorded as `"stop"` for now; surfacing the real wire -/// value requires plumbing it through the Receiver vtable (future work, -/// separate from token plumbing). -pub fn persistTurn( - alloc: std.mem.Allocator, - mgr: *panto.session_manager.SessionManager, - conv: *const panto.conversation.Conversation, - start_index: usize, - provider: []const u8, - model: []const u8, - per_message_usage: []const ?panto.session_manager.Usage, -) !void { - var batch_messages: std.ArrayList(panto.session_manager.DiskMessage) = .empty; - defer batch_messages.deinit(alloc); - var batch_providers: std.ArrayList(?[]const u8) = .empty; - defer batch_providers.deinit(alloc); - var batch_models: std.ArrayList(?[]const u8) = .empty; - defer batch_models.deinit(alloc); - - var i = start_index; - var assistant_seen: usize = 0; - while (i < conv.messages.items.len) : (i += 1) { - const msg = conv.messages.items[i]; - const blocks = try alloc.alloc(panto.session_manager.DiskContentBlock, msg.content.items.len); - var allocated: usize = 0; - errdefer { - for (blocks[0..allocated]) |b| b.deinit(alloc); - alloc.free(blocks); - } - for (msg.content.items) |block| { - blocks[allocated] = try panto.session.contentBlockToDisk(alloc, block); - allocated += 1; - } - if (msg.role == .assistant and hasToolUseWithoutFollowingResults(conv, i)) { - for (blocks[0..allocated]) |b| b.deinit(alloc); - alloc.free(blocks); - continue; - } - switch (msg.role) { - .system => { - // Mid-turn system messages aren't a thing in pantograph today. - // Treat as harmless and persist verbatim, sans stamps. - try batch_messages.append(alloc, .{ .role = .system, .content = blocks }); - try batch_providers.append(alloc, null); - try batch_models.append(alloc, null); - }, - .user => { - try batch_messages.append(alloc, .{ .role = .user, .content = blocks }); - try batch_providers.append(alloc, provider); - try batch_models.append(alloc, model); - }, - .assistant => { - // Pair this assistant message with its usage, if the - // receiver captured one for this position. (A turn - // ending in a stream error can have fewer usage entries - // than assistant messages; treat as null and move on.) - const usage: ?panto.session_manager.Usage = if (assistant_seen < per_message_usage.len) - per_message_usage[assistant_seen] - else - null; - assistant_seen += 1; - try batch_messages.append(alloc, .{ - .role = .assistant, - .content = blocks, - .provider = try alloc.dupe(u8, provider), - .model = try alloc.dupe(u8, model), - .stop_reason = try alloc.dupe(u8, "stop"), - .usage = usage, - }); - try batch_providers.append(alloc, null); - try batch_models.append(alloc, null); - }, - } - } - try mgr.appendMessagesAtomic(batch_messages.items, batch_providers.items, batch_models.items); -} - -/// Persist a compaction to the session log. The agent rewrote the -/// in-memory conversation to `[system..., summary, kept-suffix...]`; the -/// superseded prefix stays on disk untouched. We append everything from -/// the compaction summary onward as fresh entries: the summary message -/// (carrying the `compactionSummary` block) followed by the duplicated -/// kept-verbatim suffix. On replay, the latest summary resets effective -/// context, so the duplicated suffix is what survives. -/// -/// Usage is not threaded through here: the duplicated entries are replays -/// of already-sized turns, and the summary itself isn't an assistant turn. -pub fn persistCompaction( - alloc: std.mem.Allocator, - mgr: *panto.session_manager.SessionManager, - conv: *const panto.conversation.Conversation, - provider: []const u8, - model: []const u8, -) !void { - const start = panto.conversation.latestCompactionIndex(conv.messages.items) orelse return; - try persistTurn(alloc, mgr, conv, start, provider, model, &.{}); -} - -pub fn hasToolUseWithoutFollowingResults(conv: *const panto.conversation.Conversation, index: usize) bool { - const msg = conv.messages.items[index]; - for (msg.content.items) |block| { - if (block != .ToolUse) continue; - if (index + 1 >= conv.messages.items.len) return true; - const next = conv.messages.items[index + 1]; - if (next.role != .user) return true; - var found = false; - for (next.content.items) |next_block| { - if (next_block == .ToolResult and std.mem.eql(u8, next_block.ToolResult.tool_use_id, block.ToolUse.id)) { - found = true; - break; - } - } - if (!found) return true; - } - return false; -} diff --git a/src/system_prompt.zig b/src/system_prompt.zig index 0b449dc..b5423f8 100644 --- a/src/system_prompt.zig +++ b/src/system_prompt.zig @@ -137,12 +137,11 @@ pub fn seedFresh( io: Io, environ_map: *const std.process.Environ.Map, base_dir: []const u8, - conv: *panto.conversation.Conversation, - mgr: *panto.session_manager.SessionManager, + agent: *panto.agent.Agent, ) !void { const user_dir = try userLayerDir(arena, environ_map); const project_dir = try projectLayerDir(arena, io); - return seedFreshLayers(arena, io, base_dir, user_dir, project_dir, conv, mgr); + return seedFreshLayers(arena, io, base_dir, user_dir, project_dir, agent); } /// Layer-explicit variant of `seedFresh` (see `resolveLayers`). @@ -152,14 +151,13 @@ pub fn seedFreshLayers( base_dir: ?[]const u8, user_dir: ?[]const u8, project_dir: ?[]const u8, - conv: *panto.conversation.Conversation, - mgr: *panto.session_manager.SessionManager, + agent: *panto.agent.Agent, ) !void { const resolved = try resolveLayers(arena, io, base_dir, user_dir, project_dir); const blocks = try resolved.blocks(arena); for (blocks) |text| { - try conv.addSystemMessage(text); - try appendSystemEntry(mgr.allocator, mgr, text, .append); + // The agent adds to its conversation and persists the entry. + try agent.addSystemMessage(text, .append); } } @@ -176,12 +174,11 @@ pub fn reconcileResume( io: Io, environ_map: *const std.process.Environ.Map, base_dir: []const u8, - conv: *panto.conversation.Conversation, - mgr: *panto.session_manager.SessionManager, + agent: *panto.agent.Agent, ) !void { const user_dir = try userLayerDir(arena, environ_map); const project_dir = try projectLayerDir(arena, io); - return reconcileResumeLayers(arena, io, base_dir, user_dir, project_dir, conv, mgr); + return reconcileResumeLayers(arena, io, base_dir, user_dir, project_dir, agent); } /// Layer-explicit variant of `reconcileResume` (see `resolveLayers`). @@ -191,22 +188,21 @@ pub fn reconcileResumeLayers( base_dir: ?[]const u8, user_dir: ?[]const u8, project_dir: ?[]const u8, - conv: *panto.conversation.Conversation, - mgr: *panto.session_manager.SessionManager, + agent: *panto.agent.Agent, ) !void { const resolved = try resolveLayers(arena, io, base_dir, user_dir, project_dir); const config_blocks = try resolved.blocks(arena); - const window = try effectiveConfigWindow(arena, conv.messages.items); + const window = try effectiveConfigWindow(arena, agent.conversation.messages.items); if (blocksEqual(config_blocks, window)) return; // Re-seed: one `replace` for the seed, then one `append` per append. - try conv.replaceSystemMessage(config_blocks[0]); - try appendSystemEntry(mgr.allocator, mgr, config_blocks[0], .replace); + // The agent adds to its conversation and persists each entry (with the + // correct system mode). + try agent.addSystemMessage(config_blocks[0], .replace); for (config_blocks[1..]) |text| { - try conv.addSystemMessage(text); - try appendSystemEntry(mgr.allocator, mgr, text, .append); + try agent.addSystemMessage(text, .append); } } @@ -292,22 +288,6 @@ fn blocksEqual(a: []const []const u8, b: []const []const u8) bool { return true; } -/// Append a system-role entry to the session log carrying `mode`. -fn appendSystemEntry( - alloc: Allocator, - mgr: *panto.session_manager.SessionManager, - text: []const u8, - mode: panto.session_manager.DiskSystemMode, -) !void { - const blocks = try alloc.alloc(panto.session_manager.DiskContentBlock, 1); - blocks[0] = .{ .text = .{ .text = try alloc.dupe(u8, text) } }; - _ = try mgr.appendMessage( - .{ .role = .system, .mode = mode, .content = blocks }, - null, - null, - ); -} - /// Read `dir/name`, returning its contents (owned by `arena`) or null if the /// file does not exist. Other I/O errors propagate. fn readLayerFile( @@ -487,6 +467,34 @@ fn openTmpSession(arena: Allocator, root: []const u8) !panto.session_manager.Ses return panto.session_manager.SessionManager.init(testing.allocator, testing.io, sessions, "/cwd"); } +/// Minimal agent harness for system-prompt tests: an empty registry and a +/// throwaway provider config, wrapping a session store and (optionally) an +/// adopted conversation. Holds the registry/config so the agent's borrowed +/// pointers stay valid for the agent's lifetime. +const SPAgentHarness = struct { + registry: panto.ToolRegistry, + config: panto.config.Config, + agent: panto.agent.Agent, + + fn init( + self: *SPAgentHarness, + store: panto.session_store.SessionStore, + adopted: ?panto.conversation.Conversation, + ) void { + self.registry = panto.ToolRegistry.init(testing.allocator); + self.config = .{ + .provider = .{ .openai_chat = .{ .api_key = "k", .base_url = "u", .model = "m" } }, + .registry = &self.registry, + }; + self.agent = panto.agent.Agent.init(testing.allocator, testing.io, &self.config, store, adopted); + } + + fn deinit(self: *SPAgentHarness) void { + self.agent.deinit(); + self.registry.deinit(); + } +}; + /// Force the session to flush to disk by appending an assistant entry. /// (System/user entries buffer in memory until the first assistant entry; /// a realistic resume only sees what a completed turn persisted.) @@ -512,12 +520,13 @@ test "seedFresh then no-config-change resume is a no-op" { try layers.writeProject("APPEND_SYSTEM.md", "project append"); const project_dir = try layers.projectDir(arena); - // Fresh session: seed + persist. + // Fresh session: seed + persist (through the agent). var mgr = try openTmpSession(arena, layers.root); { - var conv = panto.conversation.Conversation.init(alloc); - defer conv.deinit(); - try seedFreshLayers(arena, testing.io, null, null, project_dir, &conv, &mgr); + var h: SPAgentHarness = undefined; + h.init(mgr.store(), null); + defer h.deinit(); + try seedFreshLayers(arena, testing.io, null, null, project_dir, &h.agent); } try forceFlush(&mgr); // persist seed + append (+ assistant) to disk const entries_on_disk = mgr.getEntries().len; @@ -525,13 +534,15 @@ test "seedFresh then no-config-change resume is a no-op" { const session_file = try arena.dupe(u8, mgr.getSessionFile()); mgr.deinit(); - // Resume: reopen the file, rebuild + reconcile against unchanged config - // → no new entries. + // Resume: reopen the file, adopt the rebuilt conversation + reconcile + // against unchanged config → no new entries. var mgr2 = try panto.session_manager.SessionManager.open(alloc, testing.io, session_file); defer mgr2.deinit(); - var conv2 = try mgr2.rebuildConversation(); - defer conv2.deinit(); - try reconcileResumeLayers(arena, testing.io, null, null, project_dir, &conv2, &mgr2); + const conv2 = try mgr2.rebuildConversation(); + var h2: SPAgentHarness = undefined; + h2.init(mgr2.store(), conv2); + defer h2.deinit(); + try reconcileResumeLayers(arena, testing.io, null, null, project_dir, &h2.agent); try testing.expectEqual(entries_on_disk, mgr2.getEntries().len); } @@ -548,9 +559,10 @@ test "resume after config change appends replace + append sequence" { var mgr = try openTmpSession(arena, layers.root); { - var conv = panto.conversation.Conversation.init(alloc); - defer conv.deinit(); - try seedFreshLayers(arena, testing.io, null, null, project_dir, &conv, &mgr); + var h: SPAgentHarness = undefined; + h.init(mgr.store(), null); + defer h.deinit(); + try seedFreshLayers(arena, testing.io, null, null, project_dir, &h.agent); } try forceFlush(&mgr); // persist old seed (+ assistant) to disk try testing.expectEqual(@as(usize, 2), mgr.getEntries().len); // seed + assistant @@ -563,15 +575,17 @@ test "resume after config change appends replace + append sequence" { var mgr2 = try panto.session_manager.SessionManager.open(alloc, testing.io, session_file); defer mgr2.deinit(); - var conv2 = try mgr2.rebuildConversation(); - defer conv2.deinit(); - try reconcileResumeLayers(arena, testing.io, null, null, project_dir, &conv2, &mgr2); + const conv2 = try mgr2.rebuildConversation(); + var h2: SPAgentHarness = undefined; + h2.init(mgr2.store(), conv2); + defer h2.deinit(); + try reconcileResumeLayers(arena, testing.io, null, null, project_dir, &h2.agent); // old seed + assistant + replace(new seed) + append(new append) = 4. try testing.expectEqual(@as(usize, 4), mgr2.getEntries().len); // The effective prompt now reflects only the new config blocks. - const eff = try panto.conversation.effectiveSystemBlocks(arena, conv2.messages.items); + const eff = try panto.conversation.effectiveSystemBlocks(arena, h2.agent.conversation.messages.items); try testing.expectEqual(@as(usize, 2), eff.items.len); try testing.expectEqualStrings("new seed", eff.items[0]); try testing.expectEqualStrings("new append", eff.items[1]); @@ -579,7 +593,7 @@ test "resume after config change appends replace + append sequence" { // A second no-op resume must not append again (anchors to the new // `replace` window, not the stale original seed). const after_first = mgr2.getEntries().len; - try reconcileResumeLayers(arena, testing.io, null, null, project_dir, &conv2, &mgr2); + try reconcileResumeLayers(arena, testing.io, null, null, project_dir, &h2.agent); try testing.expectEqual(after_first, mgr2.getEntries().len); } |