diff options
Diffstat (limited to 'libpanto/src')
| -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 |
5 files changed, 670 insertions, 158 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; +} |