diff options
Diffstat (limited to 'libpanto')
| -rw-r--r-- | libpanto/src/agent.zig | 551 | ||||
| -rw-r--r-- | libpanto/src/anthropic_messages_json.zig | 16 | ||||
| -rw-r--r-- | libpanto/src/compaction.zig | 743 | ||||
| -rw-r--r-- | libpanto/src/config.zig | 19 | ||||
| -rw-r--r-- | libpanto/src/conversation.zig | 176 | ||||
| -rw-r--r-- | libpanto/src/openai_chat_json.zig | 48 | ||||
| -rw-r--r-- | libpanto/src/provider.zig | 31 | ||||
| -rw-r--r-- | libpanto/src/provider_anthropic_messages.zig | 54 | ||||
| -rw-r--r-- | libpanto/src/provider_openai_chat.zig | 10 | ||||
| -rw-r--r-- | libpanto/src/root.zig | 1 | ||||
| -rw-r--r-- | libpanto/src/session.zig | 112 | ||||
| -rw-r--r-- | libpanto/src/session_manager.zig | 69 |
12 files changed, 1753 insertions, 77 deletions
diff --git a/libpanto/src/agent.zig b/libpanto/src/agent.zig index a42c16c..411abc5 100644 --- a/libpanto/src/agent.zig +++ b/libpanto/src/agent.zig @@ -31,6 +31,7 @@ const Io = std.Io; const provider_mod = @import("provider.zig"); const config_mod = @import("config.zig"); const conversation = @import("conversation.zig"); +const compaction_mod = @import("compaction.zig"); const tool_mod = @import("tool.zig"); const tool_source_mod = @import("tool_source.zig"); const tool_registry_mod = @import("tool_registry.zig"); @@ -43,6 +44,91 @@ const Entry = tool_registry_mod.Entry; pub const Config = config_mod.Config; +/// Re-export for the `compact` usages parameter (provider-reported token +/// usage per message, used for retention sizing). +pub const conversation_Usage = @import("session.zig").Usage; + +/// Deep-copy a message (role + all content blocks) into fresh owned +/// allocations. Used when rebuilding the conversation after compaction. +fn cloneMessage(alloc: Allocator, msg: conversation.Message) !conversation.Message { + var content: std.ArrayList(conversation.ContentBlock) = .empty; + errdefer { + for (content.items) |*b| b.deinit(alloc); + content.deinit(alloc); + } + try content.ensureTotalCapacity(alloc, msg.content.items.len); + for (msg.content.items) |block| { + content.appendAssumeCapacity(try cloneBlock(alloc, block)); + } + return .{ .role = msg.role, .content = content, .usage = msg.usage }; +} + +fn cloneBlock(alloc: Allocator, block: conversation.ContentBlock) !conversation.ContentBlock { + return switch (block) { + .Text => |b| .{ .Text = try conversation.textualBlockFromSlice(alloc, b.items) }, + .Thinking => |b| blk: { + const tb = try conversation.textualBlockFromSlice(alloc, b.text.items); + errdefer { + var mut = tb; + mut.deinit(alloc); + } + const sig: ?[]const u8 = if (b.signature) |s| try alloc.dupe(u8, s) else null; + break :blk .{ .Thinking = .{ .text = tb, .signature = sig } }; + }, + .ToolUse => |b| blk: { + const id = try alloc.dupe(u8, b.id); + errdefer alloc.free(id); + const name = try alloc.dupe(u8, b.name); + errdefer alloc.free(name); + const input = try conversation.textualBlockFromSlice(alloc, b.input.items); + break :blk .{ .ToolUse = .{ .id = id, .name = name, .input = input } }; + }, + .ToolResult => |b| blk: { + const tuid = try alloc.dupe(u8, b.tool_use_id); + errdefer alloc.free(tuid); + const content = try conversation.textualBlockFromSlice(alloc, b.content.items); + break :blk .{ .ToolResult = .{ .tool_use_id = tuid, .content = content } }; + }, + .System => |b| .{ .System = .{ + .text = try conversation.textualBlockFromSlice(alloc, b.text.items), + .mode = b.mode, + } }, + .CompactionSummary => |b| .{ .CompactionSummary = .{ + .text = try conversation.textualBlockFromSlice(alloc, b.text.items), + } }, + }; +} + +/// A minimal receiver that captures the assistant's streamed message for +/// compaction. We don't need incremental events — the assembled message is +/// read off the conversation after the turn — so all callbacks are no-ops. +const CompactionCapture = struct { + allocator: Allocator, + + fn receiver(self: *CompactionCapture) provider_mod.Receiver { + return .{ .ptr = self, .vtable = &vt }; + } + fn deinit(self: *CompactionCapture) void { + _ = self; + } + const vt: provider_mod.ReceiverVTable = .{ + .onMessageStart = onMessageStart, + .onBlockStart = onBlockStart, + .onToolDetails = onToolDetails, + .onContentDelta = onContentDelta, + .onBlockComplete = onBlockComplete, + .onMessageComplete = onMessageComplete, + .onError = onError, + }; + fn onMessageStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {} + fn onBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {} + fn onToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {} + fn onContentDelta(_: *anyopaque, _: usize, _: []const u8) anyerror!void {} + fn onBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {} + fn onMessageComplete(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {} + fn onError(_: *anyopaque, _: anyerror) void {} +}; + fn isValidToolInput(input: []const u8) bool { if (input.len == 0) return true; if (input[0] != '{') return true; // legacy tests/tools may use opaque bytes @@ -72,6 +158,15 @@ pub const Agent = struct { /// 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, + /// Compaction system prompt used for automatic compaction on context + /// overflow. Borrowed; set by the embedder (resolved from its + /// `COMPACTION.md` layers). When null, auto-compaction is disabled and + /// a context-overflow error propagates unchanged. + compaction_system_prompt: ?[]const u8 = null, + /// Set by the embedder after `runStep` returns to learn whether an + /// automatic compaction occurred this turn (so it can persist the + /// rewritten conversation). Reset at the top of each `runStep`. + auto_compacted: bool = false, pub fn init(allocator: Allocator, io: Io, config: *const Config) Agent { return .{ @@ -106,11 +201,24 @@ pub const Agent = struct { conv: *conversation.Conversation, receiver: *provider_mod.Receiver, ) !void { + self.auto_compacted = false; 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.stream_fn(self.allocator, self.io, cfg, conv, receiver); + self.stream_fn(self.allocator, self.io, cfg, conv, receiver) catch |err| { + // Automatic compaction on context overflow: compact the + // conversation once and retry the failed request a single + // time. If the retry also overflows, surface the error. + if (err != error.ContextOverflow) return err; + if (self.auto_compacted) return err; // already retried once + const sys = self.compaction_system_prompt orelse return err; + const res = try self.compact(conv, sys, null); + if (!res.compacted) return err; // nothing to shed; give up + self.auto_compacted = true; + // Retry the same request against the compacted context. + try self.stream_fn(self.allocator, self.io, cfg, conv, receiver); + }; const last = conv.messages.items[conv.messages.items.len - 1]; std.debug.assert(last.role == .assistant); @@ -133,6 +241,235 @@ pub const Agent = struct { return false; } + /// Outcome of a compaction attempt. + pub const CompactionResult = struct { + /// Whether the conversation was actually compacted. False means the + /// active conversation already fit within the keep-verbatim budget + /// (nothing to summarize) — the conversation is unchanged. + compacted: bool, + /// Number of whole turns kept verbatim after the summary. + kept_turns: usize = 0, + /// Number of conversation messages folded into the summary. + summarized_messages: usize = 0, + }; + + /// 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). + /// + /// The system prompt survives untouched: all `.system`-role messages + /// are preserved in order, and no `replace` block is written. Only the + /// conversation (user/assistant) prefix is summarized. + /// + /// Per-message provider usage is read directly off the conversation + /// (`Message.usage`, set live by the provider and on replay from disk). + /// `computeSplit` uses it to size the retention window; messages + /// lacking usage fall back to word counting. + /// + /// `extra_instructions`, when non-null, is appended to the compaction + /// system prompt for this run (the `/compact $ARGUMENTS` path). + /// + /// `system_prompt` is the compaction system prompt (resolved by the + /// 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 messages = conv.messages.items; + + // Project per-message usage off the conversation for sizing. + const usages = try self.allocator.alloc(?conversation_Usage, messages.len); + defer self.allocator.free(usages); + for (messages, 0..) |m, i| usages[i] = m.usage; + + const split = compaction_mod.computeSplit(messages, usages, self.config.compaction.keep_verbatim); + + // Determine the active conversation start (after any prior summary). + const active_start: usize = if (conversation.latestCompactionIndex(messages)) |a| a + 1 else 0; + + // Nothing to summarize: the active conversation already fits, or the + // prefix boundary is at/under the first active turn. + if (split.prefix_end <= active_start) { + return .{ .compacted = false }; + } + // Count how many *conversation* (non-system) messages are in the + // summarized prefix. If none, this is also a no-op. + var summarized: usize = 0; + for (messages[active_start..split.prefix_end]) |m| { + if (m.role != .system) summarized += 1; + } + if (summarized == 0) return .{ .compacted = false }; + + // Serialize the prefix transcript and carry forward the latest + // existing summary (chained-compaction invariant). + const transcript = try compaction_mod.serializeTranscript( + self.allocator, + messages[active_start..split.prefix_end], + ); + defer self.allocator.free(transcript); + + const previous_summary = compaction_mod.latestSummaryText(messages); + const body = try compaction_mod.buildRequestBody(self.allocator, transcript, previous_summary); + defer self.allocator.free(body); + + const summary = try self.runCompactionRequest(system_prompt, body, extra_instructions); + defer self.allocator.free(summary); + + try self.rewriteWithSummary(conv, split.prefix_end, summary); + + return .{ + .compacted = true, + .kept_turns = split.kept_turns, + .summarized_messages = summarized, + }; + } + + /// 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 + /// messages survive in order; a `.CompactionSummary` user message is + /// inserted; the kept suffix (`messages[prefix_end..]`) is preserved. + fn rewriteWithSummary( + self: *Agent, + conv: *conversation.Conversation, + prefix_end: usize, + summary: []const u8, + ) !void { + const alloc = self.allocator; + const old = conv.messages.items; + + var rebuilt: std.ArrayList(conversation.Message) = .empty; + errdefer { + for (rebuilt.items) |*m| m.deinit(alloc); + rebuilt.deinit(alloc); + } + + // 1. All system messages from the summarized prefix survive, in + // order. (System messages in the kept suffix come along with it + // below, so only scan the prefix here.) + for (old[0..prefix_end]) |*m| { + if (m.role != .system) continue; + try rebuilt.append(alloc, try cloneMessage(alloc, m.*)); + } + + // 2. The compaction summary, alone in a user message. + { + const tb = try conversation.textualBlockFromSlice(alloc, summary); + var content: std.ArrayList(conversation.ContentBlock) = .empty; + errdefer { + for (content.items) |*b| b.deinit(alloc); + content.deinit(alloc); + } + try content.append(alloc, .{ .CompactionSummary = .{ .text = tb } }); + try rebuilt.append(alloc, .{ .role = .user, .content = content }); + } + + // 3. The kept verbatim suffix. + for (old[prefix_end..]) |*m| { + try rebuilt.append(alloc, try cloneMessage(alloc, m.*)); + } + + // Swap in the rebuilt list and free the old one. + for (conv.messages.items) |*m| m.deinit(alloc); + conv.messages.deinit(alloc); + conv.messages = rebuilt; + } + + /// Run a single compaction provider call against a throwaway + /// conversation. Returns the assistant's summary text (caller owns). + /// + /// Model selection: try `config.compaction.model` if set; on failure, + /// fall back to the active chat model. Compaction runs with an empty + /// tool registry and a single user message (the request body); no tools + /// are exposed and no session logging occurs. + fn runCompactionRequest( + self: *Agent, + system_prompt: []const u8, + body: []const u8, + extra_instructions: ?[]const u8, + ) ![]u8 { + const alloc = self.allocator; + + // Assemble the effective compaction system prompt (+ extra + // instructions for a `/compact $ARGUMENTS` run). + var sys_text: []const u8 = system_prompt; + var sys_owned: ?[]u8 = null; + defer if (sys_owned) |s| alloc.free(s); + if (extra_instructions) |extra| { + if (extra.len > 0) { + const combined = try std.fmt.allocPrint( + alloc, + "{s}\n\n## Additional instructions for this compaction run\n\n{s}", + .{ system_prompt, extra }, + ); + sys_owned = combined; + sys_text = combined; + } + } + + var empty_registry = ToolRegistry.init(alloc); + defer empty_registry.deinit(); + + // Try the configured compaction model first, then fall back to the + // active chat model on any failure. + if (self.config.compaction.model) |comp_provider| { + const cfg: config_mod.Config = .{ + .provider = comp_provider, + .registry = &empty_registry, + .compaction = self.config.compaction, + }; + if (self.runSingleCompactionTurn(&cfg, sys_text, body)) |summary| { + return summary; + } else |err| { + std.log.warn("compaction model failed ({t}); falling back to active model", .{err}); + } + } + + const cfg: config_mod.Config = .{ + .provider = self.config.provider, + .registry = &empty_registry, + .compaction = self.config.compaction, + }; + return self.runSingleCompactionTurn(&cfg, sys_text, body); + } + + /// One provider call for compaction. Builds a throwaway conversation + /// (system prompt + one user message), streams a single turn through a + /// capturing receiver, and returns the assembled assistant text. + fn runSingleCompactionTurn( + self: *Agent, + cfg: *const config_mod.Config, + system_prompt: []const u8, + body: []const u8, + ) ![]u8 { + const alloc = self.allocator; + var conv = conversation.Conversation.init(alloc); + defer conv.deinit(); + try conv.addSystemMessage(system_prompt); + try conv.addUserMessage(body); + + var capture = CompactionCapture{ .allocator = alloc }; + defer capture.deinit(); + var recv = capture.receiver(); + + try self.stream_fn(alloc, self.io, cfg, &conv, &recv); + + // The provider appended an assistant message; gather its text. + const last = conv.messages.items[conv.messages.items.len - 1]; + if (last.role != .assistant) return error.CompactionNoResponse; + var out: std.ArrayList(u8) = .empty; + errdefer out.deinit(alloc); + for (last.content.items) |block| { + if (block == .Text) try out.appendSlice(alloc, block.Text.items); + } + if (out.items.len == 0) return error.CompactionEmptySummary; + return out.toOwnedSlice(alloc); + } + /// Dispatch every ToolUse block in `assistant_msg`. Groups by owning /// registration; one OS thread per group; results assembled in the /// original call order. @@ -431,6 +768,10 @@ const StubProvider = struct { allocator: Allocator, scripted: []const ScriptedTurn, next: usize = 0, + /// Number of leading stream calls that should fail with + /// `error.ContextOverflow` before any scripted turn is served. Used to + /// drive the auto-compaction path. Decremented on each overflow. + overflow_calls: usize = 0, const ScriptedTurn = struct { blocks: []const TestBlock, @@ -463,6 +804,10 @@ fn stubStreamStep( ) anyerror!void { const self = stub_active orelse return error.NoStubInstalled; _ = allocator; + if (self.overflow_calls > 0) { + self.overflow_calls -= 1; + return error.ContextOverflow; + } if (self.next >= self.scripted.len) return error.NoMoreScriptedTurns; const turn = self.scripted[self.next]; self.next += 1; @@ -1285,3 +1630,207 @@ test "setConfig swaps the visible tool set between turns" { try testing.expectEqualStrings("2", tr.tool_use_id); try testing.expectEqualStrings("B:B", tr.content.items); } + +test "compact: summarizes prefix, keeps suffix, system survives" { + const allocator = testing.allocator; + + // The stub returns a single text turn — used as the summary text. + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{.{ .Text = "SUMMARY OF EARLIER" }} }, + }; + 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(); + // keep_verbatim sized so only the last (short) turn fits: q2+a2 are + // 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); + agent.stream_fn = stub.install(); + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addSystemMessage("you are helpful"); + try conv.addUserMessage("first question here with several words"); + try conv.addAssistantMessage(&.{ + .{ .Text = try conversation.textualBlockFromSlice(allocator, "first answer with several words") }, + }); + try conv.addUserMessage("second recent question"); + try conv.addAssistantMessage(&.{ + .{ .Text = try conversation.textualBlockFromSlice(allocator, "second recent answer") }, + }); + + const res = try agent.compact(&conv, "Summarize the conversation.", null); + try testing.expect(res.compacted); + + // Expected rebuilt: [system, compaction summary(user), user q2, asst a2] + try testing.expectEqual(@as(usize, 4), conv.messages.items.len); + try testing.expectEqual(conversation.MessageRole.system, conv.messages.items[0].role); + try testing.expectEqualStrings( + "you are helpful", + conv.messages.items[0].content.items[0].System.text.items, + ); + try testing.expectEqual(conversation.MessageRole.user, conv.messages.items[1].role); + try testing.expectEqualStrings( + "SUMMARY OF EARLIER", + conv.messages.items[1].content.items[0].CompactionSummary.text.items, + ); + try testing.expectEqualStrings( + "second recent question", + conv.messages.items[2].content.items[0].Text.items, + ); + try testing.expectEqualStrings( + "second recent answer", + conv.messages.items[3].content.items[0].Text.items, + ); +} + +test "compact: no-op when conversation already fits the budget" { + const allocator = testing.allocator; + + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{.{ .Text = "should not be used" }} }, + }; + 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(); + h.config.compaction = .{ .keep_verbatim = 1_000_000 }; + var agent = Agent.init(allocator, io, &h.config); + agent.stream_fn = stub.install(); + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + 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); + try testing.expect(!res.compacted); + try testing.expectEqual(@as(usize, 3), conv.messages.items.len); + // Stub was never consumed. + try testing.expectEqual(@as(usize, 0), stub.next); +} + +test "compact: extra instructions are appended to the system prompt" { + const allocator = testing.allocator; + + // Capture the system prompt the stub sees by scripting a turn and + // inspecting the throwaway conversation isn't directly possible via the + // current stub; instead we just assert compaction succeeds with extra + // instructions present (smoke test of the append path). + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{.{ .Text = "S" }} }, + }; + 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(); + h.config.compaction = .{ .keep_verbatim = 1 }; + var agent = Agent.init(allocator, io, &h.config); + agent.stream_fn = stub.install(); + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("question one two three"); + try conv.addAssistantMessage(&.{ + .{ .Text = try conversation.textualBlockFromSlice(allocator, "answer one two three") }, + }); + try conv.addUserMessage("question two"); + try conv.addAssistantMessage(&.{ + .{ .Text = try conversation.textualBlockFromSlice(allocator, "answer two") }, + }); + + const res = try agent.compact(&conv, "Base prompt.", "keep bug #3 details"); + try testing.expect(res.compacted); +} + +test "runStep: auto-compacts on context overflow and retries once" { + const allocator = testing.allocator; + + // First stream call overflows; then the compaction request returns a + // summary; then the retried main request returns a final text turn. + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{.{ .Text = "COMPACTED SUMMARY" }} }, // compaction call + .{ .blocks = &.{.{ .Text = "final answer" }} }, // retried main call + }; + var stub = StubProvider{ + .allocator = allocator, + .scripted = &scripted, + .overflow_calls = 1, + }; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var h = TestHarness.init(allocator); + defer h.deinit(); + h.activate(); + h.config.compaction = .{ .keep_verbatim = 10 }; + var agent = Agent.init(allocator, io, &h.config); + agent.stream_fn = stub.install(); + agent.compaction_system_prompt = "Summarize the conversation."; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addSystemMessage("you are helpful"); + try conv.addUserMessage("first question with several words here"); + try conv.addAssistantMessage(&.{ + .{ .Text = try conversation.textualBlockFromSlice(allocator, "first answer with several words") }, + }); + try conv.addUserMessage("second recent question"); + + var recv = NoopReceiver.make(); + try agent.runStep(&conv, &recv); + + try testing.expect(agent.auto_compacted); + // After compaction + retry: [system, summary, user q2, assistant final]. + const msgs = conv.messages.items; + try testing.expectEqual(conversation.MessageRole.system, msgs[0].role); + try testing.expectEqualStrings( + "COMPACTED SUMMARY", + msgs[1].content.items[0].CompactionSummary.text.items, + ); + try testing.expectEqualStrings("second recent question", msgs[2].content.items[0].Text.items); + try testing.expectEqualStrings("final answer", msgs[msgs.len - 1].content.items[0].Text.items); +} + +test "runStep: context overflow without compaction prompt propagates" { + const allocator = testing.allocator; + + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{.{ .Text = "unused" }} }, + }; + var stub = StubProvider{ + .allocator = allocator, + .scripted = &scripted, + .overflow_calls = 1, + }; + 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 agent = Agent.init(allocator, io, &h.config); + agent.stream_fn = stub.install(); + // No compaction_system_prompt set -> overflow propagates. + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("hi"); + + var recv = NoopReceiver.make(); + try testing.expectError(error.ContextOverflow, agent.runStep(&conv, &recv)); +} diff --git a/libpanto/src/anthropic_messages_json.zig b/libpanto/src/anthropic_messages_json.zig index d382de8..c54db86 100644 --- a/libpanto/src/anthropic_messages_json.zig +++ b/libpanto/src/anthropic_messages_json.zig @@ -79,10 +79,12 @@ pub fn serializeRequest( try s.endArray(); } - // Emit messages (everything that isn't .system). + // Emit messages (everything that isn't .system). If the conversation + // has been compacted, only the latest compaction summary and the + // messages after it are active; the superseded prefix is dropped. try s.objectField("messages"); try s.beginArray(); - for (conv.messages.items) |msg| { + for (conversation.activeMessageWindow(conv.messages.items)) |msg| { if (msg.role == .system) continue; try writeMessage(&s, msg, allocator); } @@ -239,6 +241,16 @@ fn writeBlock(s: *std.json.Stringify, block: conversation.ContentBlock) !void { // system text is hoisted into the top-level `system` string by // `collectSystemPrompt`. Present only to keep the switch exhaustive. .System => {}, + // A compaction summary is the synthetic seed text standing in for + // a compacted prefix; the model reads it as ordinary user text. + .CompactionSummary => |cs| { + try s.beginObject(); + try s.objectField("type"); + try s.write("text"); + try s.objectField("text"); + try s.write(cs.text.items); + try s.endObject(); + }, } } diff --git a/libpanto/src/compaction.zig b/libpanto/src/compaction.zig new file mode 100644 index 0000000..254b876 --- /dev/null +++ b/libpanto/src/compaction.zig @@ -0,0 +1,743 @@ +//! Compaction sizing and retention policy. +//! +//! Compaction reduces context size by summarizing an older prefix of the +//! conversation into a synthetic seed (a `.CompactionSummary` block) and +//! keeping a recent suffix of whole turns verbatim. This module owns the +//! *policy* half — deciding where to split prefix-to-summarize from +//! suffix-to-keep — and the transcript serialization of the prefix. The +//! *execution* half (running the compaction request, mutating the +//! conversation) lives in `agent.zig`. +//! +//! ## Retention model +//! +//! The retention unit is a whole **turn**: a user message followed by the +//! assistant message(s) it elicited (including any interleaved tool-result +//! user messages that belong to the same request/response cycle). v1 never +//! splits a turn. +//! +//! Splitting walks the active conversation backward, accumulating a token +//! estimate per turn, and stops as soon as the running total *exceeds* +//! `keep_verbatim`. The turn that crosses the threshold is the last turn +//! folded into the summary; turns after it are kept verbatim. So +//! `keep_verbatim` is an upper bound on the kept-suffix size, not a target. +//! +//! ## Sizing +//! +//! When provider-reported `Usage` is available for a message, its token +//! total drives sizing. Otherwise we fall back to a cheap monotonic +//! heuristic: word count scaled by `word_to_token_factor`. The fallback is +//! not exact but is sufficient to choose a conservative recent suffix. + +const std = @import("std"); +const Allocator = std.mem.Allocator; + +const conversation = @import("conversation.zig"); +const session_mod = @import("session.zig"); + +pub const Message = conversation.Message; +pub const Usage = session_mod.Usage; + +/// Multiplier applied to a word count to approximate token count when no +/// provider usage data is available. Tokens-per-word runs ~1.3 for English +/// prose and code on common BPE tokenizers (rough, deliberately so). +pub const word_to_token_factor: f64 = 1.3; + +/// Estimate the token size of a single message. +/// +/// If `usage` is non-null, its prompt+output token total is used directly. +/// Otherwise the message's textual content is word-counted and scaled by +/// `word_to_token_factor`. +pub fn messageTokenEstimate(msg: Message, usage: ?Usage) u64 { + if (usage) |u| { + // Total tokens attributable to this message: all input categories + // plus output. (For a user message only inputs are typically set; + // for an assistant message output dominates.) + return u.input + u.cache_read + u.cache_write + u.output; + } + const words = countWords(msg); + const est = @as(f64, @floatFromInt(words)) * word_to_token_factor; + return @intFromFloat(@ceil(est)); +} + +/// The full prompt size (all input categories) reported by a turn's usage. +/// Provider `input` may exclude cached tokens, which appear under +/// `cache_read`/`cache_write`; summing them recovers the true context size +/// that was sent to the model. +fn promptTokens(u: Usage) u64 { + return u.input + u.cache_read + u.cache_write; +} + +/// The cumulative context footprint at a turn's final assistant message: +/// the full prompt it consumed plus the output it produced. This is the +/// running total of *all* prior conversation tokens as of that turn. +fn cumulativeAt(u: Usage) u64 { + return promptTokens(u) + u.output; +} + +/// Estimate the incremental token cost of the turn spanning +/// `messages[start..end)` using cumulative provider usage. +/// +/// Provider `usage.input` is *cumulative*: it is the entire prompt sent for +/// that turn (the whole prior conversation), not the size of one message. +/// So a turn's own contribution is the delta between its final assistant's +/// cumulative footprint and the previous turn's: +/// +/// cumulativeAt(this_turn_last_assistant) +/// - cumulativeAt(prev_turn_last_assistant) +/// +/// This single delta captures the turn-start user prompt, every +/// tool-result, every intermediate assistant output, and the final +/// assistant output at once — no per-message decomposition needed. At +/// session start there is no previous assistant, so `before` is zero and +/// the first turn's cost is its own full cumulative footprint (which fairly +/// includes the system prompt). +/// +/// `prev_usage` is the usage of the assistant message immediately before +/// `start` (the previous turn's final assistant), or null at session start. +/// +/// Returns null when usage data is insufficient (no usage on this turn's +/// assistant), signalling the caller to fall back to word counting. +fn turnTokenEstimate( + messages: []const Message, + usages: []const ?Usage, + start: usize, + end: usize, + prev_usage: ?Usage, +) ?u64 { + // Find this turn's final assistant usage (cumulative as of turn end). + var after: ?Usage = null; + var j = end; + while (j > start) { + j -= 1; + if (messages[j].role == .assistant) { + if (usages[j]) |u| { + after = u; + break; + } + } + } + const after_usage = after orelse return null; + const after_total = cumulativeAt(after_usage); + const before_total = if (prev_usage) |b| cumulativeAt(b) else 0; + // Saturating subtraction: usage is an estimate and can be non-monotone. + return if (after_total > before_total) after_total - before_total else 0; +} + +/// Find the usage of the assistant message immediately before index +/// `start`, scanning backward. Returns null if none precedes it. +fn priorAssistantUsage( + messages: []const Message, + usages: []const ?Usage, + start: usize, +) ?Usage { + var j = start; + while (j > 0) { + j -= 1; + if (messages[j].role == .assistant) { + if (usages[j]) |u| return u; + } + } + return null; +} + +/// Count whitespace-delimited words across all textual content in a +/// message. Tool-use input JSON and tool-result content count too — they +/// occupy real context. +fn countWords(msg: Message) u64 { + var total: u64 = 0; + for (msg.content.items) |block| { + const text: []const u8 = switch (block) { + .Text => |b| b.items, + .Thinking => |b| b.text.items, + .ToolUse => |b| b.input.items, + .ToolResult => |b| b.content.items, + .System => |b| b.text.items, + .CompactionSummary => |b| b.text.items, + }; + total += countWordsIn(text); + } + return total; +} + +fn countWordsIn(text: []const u8) u64 { + var count: u64 = 0; + var in_word = false; + for (text) |c| { + const is_space = c == ' ' or c == '\t' or c == '\n' or c == '\r'; + if (is_space) { + in_word = false; + } else if (!in_word) { + in_word = true; + count += 1; + } + } + return count; +} + +/// The result of a retention split over a conversation. +pub const Split = struct { + /// Index into the *full* message list. Messages in `[0, prefix_end)` + /// are summarized; messages in `[prefix_end, len)` are kept verbatim. + /// A value equal to the message count means "summarize everything". + prefix_end: usize, + /// Number of whole turns kept verbatim. + kept_turns: usize, +}; + +/// Decide the prefix/suffix split for compaction. +/// +/// `messages` is the conversation to compact. Only the active window +/// (everything after the latest existing `.CompactionSummary`, if any) is +/// considered — an earlier compacted prefix is already summarized and is +/// never re-walked. `usages`, when provided, must be the same length as +/// `messages` and supplies per-message provider usage (null entries fall +/// back to word counting). +/// +/// Leading system messages are never part of a turn and are always treated +/// as belonging to the (kept) structural frame, not the summarized prefix: +/// the split index is reported relative to the full list but the walk only +/// accumulates user/assistant conversation turns. +/// +/// Returns the split. When the entire active conversation fits within +/// `keep_verbatim`, `prefix_end` points at the first active turn (nothing +/// to summarize) — callers should treat that as a no-op. +pub fn computeSplit( + messages: []const Message, + usages: ?[]const ?Usage, + keep_verbatim: u64, +) Split { + if (usages) |u| std.debug.assert(u.len == messages.len); + + const active_start = blk: { + if (conversation.latestCompactionIndex(messages)) |anchor| { + // The summary message itself anchors the window; active turns + // begin after it. + break :blk anchor + 1; + } + break :blk 0; + }; + + // Identify turn boundaries within the active window. A turn starts at a + // user message that is NOT purely tool-results (a fresh human/user + // prompt or compaction-follow user prompt) and runs until the next such + // boundary. Tool-result user messages and assistant messages attach to + // the current turn. + // + // We collect the start index of each turn, then walk turns backward. + var turn_starts_buf: [4096]usize = undefined; + var n_turns: usize = 0; + var i = active_start; + // Skip leading system messages (structural frame, not a turn). + while (i < messages.len and messages[i].role == .system) : (i += 1) {} + const first_turn_msg = i; + while (i < messages.len) : (i += 1) { + const msg = messages[i]; + if (msg.role == .system) continue; + if (isTurnStart(msg)) { + if (n_turns < turn_starts_buf.len) { + turn_starts_buf[n_turns] = i; + } + n_turns += 1; + } + } + const turn_starts = turn_starts_buf[0..@min(n_turns, turn_starts_buf.len)]; + + if (turn_starts.len == 0) { + // No turns to keep or summarize. + return .{ .prefix_end = messages.len, .kept_turns = 0 }; + } + + // Walk turns backward, accumulating token totals. Stop as soon as the + // running total exceeds the budget: the crossing turn is summarized. + var running: u64 = 0; + var kept_turns: usize = 0; + var t = turn_starts.len; + while (t > 0) { + t -= 1; + const start = turn_starts[t]; + const end = if (t + 1 < turn_starts.len) turn_starts[t + 1] else messages.len; + var turn_tokens: u64 = 0; + // Prefer cumulative-usage deltas: provider `input` is the whole + // prior conversation, so a turn's true cost is the delta between + // its final assistant footprint and the previous turn's. Fall back + // to per-message word counting when usage is unavailable. + const turn_estimate: ?u64 = if (usages) |u| + turnTokenEstimate(messages, u, start, end, priorAssistantUsage(messages, u, start)) + else + null; + if (turn_estimate) |est| { + turn_tokens = est; + } else { + var j = start; + while (j < end) : (j += 1) { + const usage: ?Usage = if (usages) |u| u[j] else null; + turn_tokens += messageTokenEstimate(messages[j], usage); + } + } + if (running + turn_tokens > keep_verbatim) { + // This turn crosses the budget: it (and everything before it) + // is summarized. Kept suffix starts at the next turn. + const kept_start = if (t + 1 < turn_starts.len) turn_starts[t + 1] else messages.len; + return .{ .prefix_end = kept_start, .kept_turns = kept_turns }; + } + running += turn_tokens; + kept_turns += 1; + } + + // Everything fits: nothing to summarize. Prefix ends at the first turn. + return .{ .prefix_end = first_turn_msg, .kept_turns = kept_turns }; +} + +/// A user message starts a new turn unless it carries any tool-result +/// block (in which case it's a continuation of the assistant's tool round +/// within the current turn). System and assistant messages never start a +/// turn. +fn isTurnStart(msg: Message) bool { + if (msg.role != .user) return false; + for (msg.content.items) |block| { + if (block == .ToolResult) return false; + } + return true; +} + +// ============================================================================= +// Transcript serialization +// ============================================================================= + +/// Serialize a range of conversation messages into a plain-text transcript +/// artifact. The output marks message roles and structure so the +/// compaction model treats it as material under analysis rather than live +/// chat to continue. Caller owns the returned bytes. +/// +/// Format (one labelled section per block, blank-line separated): +/// [User]: ... +/// [Assistant]: ... +/// [Assistant thinking]: ... +/// [Assistant tool call: <name> (<id>)]: <input> +/// [Tool result (<id>)]: ... +/// [Previous summary]: ... +/// +/// System messages are skipped: the system prompt survives compaction and +/// is not part of the material being summarized. +pub fn serializeTranscript( + allocator: Allocator, + messages: []const Message, +) ![]u8 { + var out: std.ArrayList(u8) = .empty; + errdefer out.deinit(allocator); + const w = &out; + + var first = true; + for (messages) |msg| { + if (msg.role == .system) continue; + for (msg.content.items) |block| { + if (!first) try w.appendSlice(allocator, "\n\n"); + first = false; + switch (block) { + .Text => |b| { + const label = if (msg.role == .assistant) "[Assistant]: " else "[User]: "; + try w.appendSlice(allocator, label); + try w.appendSlice(allocator, b.items); + }, + .Thinking => |b| { + try w.appendSlice(allocator, "[Assistant thinking]: "); + try w.appendSlice(allocator, b.text.items); + }, + .ToolUse => |b| { + const line = try std.fmt.allocPrint(allocator, "[Assistant tool call: {s} ({s})]: {s}", .{ + b.name, b.id, b.input.items, + }); + defer allocator.free(line); + try w.appendSlice(allocator, line); + }, + .ToolResult => |b| { + const line = try std.fmt.allocPrint(allocator, "[Tool result ({s})]: {s}", .{ + b.tool_use_id, b.content.items, + }); + defer allocator.free(line); + try w.appendSlice(allocator, line); + }, + .CompactionSummary => |b| { + try w.appendSlice(allocator, "[Previous summary]: "); + try w.appendSlice(allocator, b.text.items); + }, + .System => {}, + } + } + } + return out.toOwnedSlice(allocator); +} + +// ============================================================================= +// Compaction request user-prompt assembly +// ============================================================================= + +/// Build the user-prompt body for a compaction request: a brief framing, +/// the optional previous summary in a `<previous-summary>` section, and the +/// transcript of the prefix being summarized in a `<transcript>` section. +/// +/// `transcript` is the serialized prefix (see `serializeTranscript`). +/// `previous_summary`, when non-null, is the latest active compaction +/// summary text carried forward (the chained-compaction invariant: at most +/// one previous summary, never an accumulating stack). Caller owns the +/// returned bytes. +pub fn buildRequestBody( + allocator: Allocator, + transcript: []const u8, + previous_summary: ?[]const u8, +) ![]u8 { + var out: std.ArrayList(u8) = .empty; + errdefer out.deinit(allocator); + + if (previous_summary) |prev| { + try out.appendSlice(allocator, + "The previous compaction generated this summary:\n\n<previous-summary>\n"); + try out.appendSlice(allocator, prev); + try out.appendSlice(allocator, + "\n</previous-summary>\n\nThe conversation since that previous compaction:\n\n<transcript>\n"); + } else { + try out.appendSlice(allocator, + "The conversation to summarize:\n\n<transcript>\n"); + } + try out.appendSlice(allocator, transcript); + try out.appendSlice(allocator, "\n</transcript>"); + return out.toOwnedSlice(allocator); +} + +/// The latest active compaction summary text in `messages`, or null if the +/// conversation has never been compacted. Borrowed from `messages`. +pub fn latestSummaryText(messages: []const Message) ?[]const u8 { + const anchor = conversation.latestCompactionIndex(messages) orelse return null; + for (messages[anchor].content.items) |block| { + if (block == .CompactionSummary) return block.CompactionSummary.text.items; + } + return null; +} + +// ============================================================================= +// Tests +// ============================================================================= + +const testing = std.testing; + +fn userMsg(alloc: Allocator, text: []const u8) !Message { + var content: std.ArrayList(conversation.ContentBlock) = .empty; + try content.append(alloc, .{ .Text = try conversation.textualBlockFromSlice(alloc, text) }); + return .{ .role = .user, .content = content }; +} + +fn asstMsg(alloc: Allocator, text: []const u8) !Message { + var content: std.ArrayList(conversation.ContentBlock) = .empty; + try content.append(alloc, .{ .Text = try conversation.textualBlockFromSlice(alloc, text) }); + return .{ .role = .assistant, .content = content }; +} + +fn freeMsgs(alloc: Allocator, msgs: []Message) void { + for (msgs) |*m| m.deinit(alloc); +} + +test "countWordsIn - basic word counting" { + try testing.expectEqual(@as(u64, 0), countWordsIn("")); + try testing.expectEqual(@as(u64, 0), countWordsIn(" \n\t ")); + try testing.expectEqual(@as(u64, 1), countWordsIn("hello")); + try testing.expectEqual(@as(u64, 3), countWordsIn(" hello there world ")); +} + +test "messageTokenEstimate - usage wins when present" { + const a = testing.allocator; + var m = try userMsg(a, "one two three"); + defer m.deinit(a); + const u: Usage = .{ .input = 100, .output = 7, .cache_read = 3 }; + try testing.expectEqual(@as(u64, 110), messageTokenEstimate(m, u)); +} + +test "messageTokenEstimate - word-count fallback when usage absent" { + const a = testing.allocator; + var m = try userMsg(a, "one two three four"); // 4 words * 1.3 = 5.2 -> ceil 6 + defer m.deinit(a); + try testing.expectEqual(@as(u64, 6), messageTokenEstimate(m, null)); +} + +test "computeSplit - everything fits keeps all turns, summarizes nothing" { + const a = testing.allocator; + var msgs = [_]Message{ + try userMsg(a, "q1"), + try asstMsg(a, "a1"), + try userMsg(a, "q2"), + try asstMsg(a, "a2"), + }; + defer freeMsgs(a, &msgs); + + const split = computeSplit(&msgs, null, 1_000_000); + try testing.expectEqual(@as(usize, 0), split.prefix_end); + try testing.expectEqual(@as(usize, 2), split.kept_turns); +} + +test "computeSplit - tiny budget summarizes all but the last turn" { + const a = testing.allocator; + // Each message ~ a few words; usage forces deterministic sizes. + var msgs = [_]Message{ + try userMsg(a, "q1"), + try asstMsg(a, "a1"), + try userMsg(a, "q2"), + try asstMsg(a, "a2"), + try userMsg(a, "q3"), + try asstMsg(a, "a3"), + }; + defer freeMsgs(a, &msgs); + + // Each turn contributes ~200 incremental tokens (100 user + 100 + // output). Provider `input` is cumulative: it is the whole prior + // conversation as of that assistant turn. So the assistant inputs grow + // 100, 300, 500 and the per-turn delta is a steady 200. Budget 250 + // keeps only the last turn (200 <= 250); adding the prior turn (400) + // exceeds it. + const usages = [_]?Usage{ + .{ .input = 0 }, .{ .input = 100, .output = 100 }, + .{ .input = 0 }, .{ .input = 300, .output = 100 }, + .{ .input = 0 }, .{ .input = 500, .output = 100 }, + }; + const split = computeSplit(&msgs, &usages, 250); + // Kept suffix = turn starting at index 4 (q3/a3). + try testing.expectEqual(@as(usize, 4), split.prefix_end); + try testing.expectEqual(@as(usize, 1), split.kept_turns); +} + +test "computeSplit - crossing turn goes into the summary (upper-bound semantics)" { + const a = testing.allocator; + var msgs = [_]Message{ + try userMsg(a, "q1"), + try asstMsg(a, "a1"), + try userMsg(a, "q2"), + try asstMsg(a, "a2"), + }; + defer freeMsgs(a, &msgs); + + // 200 incremental tokens per turn via cumulative inputs (100, 300). + // Budget exactly 200: last turn (delta 200) does NOT exceed 200, so + // it's kept. Adding turn 1 => 400 > 200, so turn 1 summarized. + const usages = [_]?Usage{ + .{ .input = 0 }, .{ .input = 100, .output = 100 }, + .{ .input = 0 }, .{ .input = 300, .output = 100 }, + }; + const split = computeSplit(&msgs, &usages, 200); + try testing.expectEqual(@as(usize, 2), split.prefix_end); + try testing.expectEqual(@as(usize, 1), split.kept_turns); +} + +test "computeSplit - tool-result user messages attach to the current turn" { + const a = testing.allocator; + + // Turn 1: user q1 -> assistant(tool_use) -> user(tool_result) -> assistant a1 + var tr_content: std.ArrayList(conversation.ContentBlock) = .empty; + try tr_content.append(a, .{ .ToolResult = .{ + .tool_use_id = try a.dupe(u8, "t1"), + .content = try conversation.textualBlockFromSlice(a, "result"), + } }); + + var msgs = [_]Message{ + try userMsg(a, "q1"), + try asstMsg(a, "calling tool"), + .{ .role = .user, .content = tr_content }, + try asstMsg(a, "a1"), + try userMsg(a, "q2"), + try asstMsg(a, "a2"), + }; + defer freeMsgs(a, &msgs); + + // Budget small enough to keep only the last turn. The tool-result + // user message must NOT be mistaken for a turn boundary. Turn 1's + // final assistant (index 3) carries the cumulative input; the inner + // tool-round assistant (index 1) is subsumed by it. Turn 1 delta = + // 300; turn 2 delta = cumAt(a2) - cumAt(a1) = 500 - 300 = 200. + const usages = [_]?Usage{ + .{ .input = 0 }, .{ .input = 100, .output = 50 }, .{ .input = 0 }, .{ .input = 200, .output = 100 }, + .{ .input = 0 }, .{ .input = 400, .output = 100 }, + }; + const split = computeSplit(&msgs, &usages, 250); + // Kept suffix = q2/a2 turn at index 4. The whole 4-message first turn + // is summarized. + try testing.expectEqual(@as(usize, 4), split.prefix_end); + try testing.expectEqual(@as(usize, 1), split.kept_turns); +} + +test "computeSplit - active window starts after an existing compaction summary" { + const a = testing.allocator; + var msgs = [_]Message{ + try userMsg(a, "old q"), + try asstMsg(a, "old a"), + undefined, // compaction summary, filled below + try userMsg(a, "new q"), + try asstMsg(a, "new a"), + }; + var cs_content: std.ArrayList(conversation.ContentBlock) = .empty; + try cs_content.append(a, .{ .CompactionSummary = .{ + .text = try conversation.textualBlockFromSlice(a, "S1"), + } }); + msgs[2] = .{ .role = .user, .content = cs_content }; + defer freeMsgs(a, &msgs); + + // Large budget: the whole *active* window (new q / new a) fits, so + // nothing new is summarized. prefix_end points at first active turn. + const split = computeSplit(&msgs, null, 1_000_000); + try testing.expectEqual(@as(usize, 3), split.prefix_end); + try testing.expectEqual(@as(usize, 1), split.kept_turns); +} + +test "computeSplit - leading system messages are not turns" { + const a = testing.allocator; + var sys_content: std.ArrayList(conversation.ContentBlock) = .empty; + try sys_content.append(a, .{ .System = .{ + .text = try conversation.textualBlockFromSlice(a, "you are helpful"), + .mode = .append, + } }); + var msgs = [_]Message{ + .{ .role = .system, .content = sys_content }, + try userMsg(a, "q1"), + try asstMsg(a, "a1"), + }; + defer freeMsgs(a, &msgs); + + const split = computeSplit(&msgs, null, 1_000_000); + // First turn is index 1 (after the system message). + try testing.expectEqual(@as(usize, 1), split.prefix_end); + try testing.expectEqual(@as(usize, 1), split.kept_turns); +} + +test "computeSplit - regression: real cumulative session overflows keep_verbatim" { + // Reproduces a real session whose final assistant reported + // usage.input=28098, usage.output=5477 (total context 33,575). Earlier + // code treated cumulative `input` as a per-message cost and so either + // double-counted or (when usages were absent) word-counted below the + // budget, yielding a spurious "nothing to compact". With the + // turn-delta model this conversation must split at keep_verbatim=20000. + const a = testing.allocator; + // 12 user/assistant turns. User messages carry null usage; assistant + // messages carry the cumulative inputs observed on disk. + var msgs = [_]Message{ + try userMsg(a, "q1"), try asstMsg(a, "a1"), + try userMsg(a, "q2"), try asstMsg(a, "a2"), + try userMsg(a, "q3"), try asstMsg(a, "a3"), + try userMsg(a, "q4"), try asstMsg(a, "a4"), + try userMsg(a, "q5"), try asstMsg(a, "a5"), + try userMsg(a, "q6"), try asstMsg(a, "a6"), + try userMsg(a, "q7"), try asstMsg(a, "a7"), + try userMsg(a, "q8"), try asstMsg(a, "a8"), + try userMsg(a, "q9"), try asstMsg(a, "a9"), + try userMsg(a, "q10"), try asstMsg(a, "a10"), + try userMsg(a, "q11"), try asstMsg(a, "a11"), + try userMsg(a, "q12"), try asstMsg(a, "a12"), + }; + defer freeMsgs(a, &msgs); + + const U = struct { + fn mk(input: u64, output: u64) ?Usage { + return .{ .input = input, .output = output }; + } + }; + const usages = [_]?Usage{ + null, U.mk(1286, 118), + null, U.mk(2167, 62), + null, U.mk(2352, 146), + null, U.mk(9710, 153), + null, U.mk(10233, 180), + null, U.mk(12562, 127), + null, U.mk(13087, 110), + null, U.mk(14153, 68), + null, U.mk(16794, 838), + null, U.mk(17666, 228), + null, U.mk(24762, 203), + null, U.mk(28098, 5477), + }; + + const split = computeSplit(&msgs, &usages, 20_000); + // Something must be summarized: the split cannot sit at/before the + // first active turn (index 0). + try testing.expect(split.prefix_end > 0); + // The total context (33,575) exceeds the 20k budget, so at least the + // oldest turn(s) are shed and fewer than all 12 turns are kept. + try testing.expect(split.kept_turns < 12); +} + +test "serializeTranscript - labels roles and skips system" { + const a = testing.allocator; + var sys_content: std.ArrayList(conversation.ContentBlock) = .empty; + try sys_content.append(a, .{ .System = .{ + .text = try conversation.textualBlockFromSlice(a, "sys prompt"), + .mode = .append, + } }); + var msgs = [_]Message{ + .{ .role = .system, .content = sys_content }, + try userMsg(a, "hello"), + try asstMsg(a, "hi there"), + }; + defer freeMsgs(a, &msgs); + + const t = try serializeTranscript(a, &msgs); + defer a.free(t); + try testing.expect(std.mem.indexOf(u8, t, "sys prompt") == null); + try testing.expect(std.mem.indexOf(u8, t, "[User]: hello") != null); + try testing.expect(std.mem.indexOf(u8, t, "[Assistant]: hi there") != null); +} + +test "serializeTranscript - tool call and result framing" { + const a = testing.allocator; + var tu_content: std.ArrayList(conversation.ContentBlock) = .empty; + try tu_content.append(a, .{ .ToolUse = .{ + .id = try a.dupe(u8, "tc1"), + .name = try a.dupe(u8, "bash"), + .input = try conversation.textualBlockFromSlice(a, "{\"cmd\":\"ls\"}"), + } }); + var tr_content: std.ArrayList(conversation.ContentBlock) = .empty; + try tr_content.append(a, .{ .ToolResult = .{ + .tool_use_id = try a.dupe(u8, "tc1"), + .content = try conversation.textualBlockFromSlice(a, "file.txt"), + } }); + var msgs = [_]Message{ + .{ .role = .assistant, .content = tu_content }, + .{ .role = .user, .content = tr_content }, + }; + defer freeMsgs(a, &msgs); + + const t = try serializeTranscript(a, &msgs); + defer a.free(t); + try testing.expect(std.mem.indexOf(u8, t, "[Assistant tool call: bash (tc1)]: {\"cmd\":\"ls\"}") != null); + try testing.expect(std.mem.indexOf(u8, t, "[Tool result (tc1)]: file.txt") != null); +} + +test "buildRequestBody - without previous summary" { + const a = testing.allocator; + const body = try buildRequestBody(a, "TRANSCRIPT", null); + defer a.free(body); + try testing.expect(std.mem.indexOf(u8, body, "previous-summary") == null); + try testing.expect(std.mem.indexOf(u8, body, "<transcript>\nTRANSCRIPT\n</transcript>") != null); +} + +test "buildRequestBody - with previous summary" { + const a = testing.allocator; + const body = try buildRequestBody(a, "TRANSCRIPT", "PRIOR"); + defer a.free(body); + try testing.expect(std.mem.indexOf(u8, body, "<previous-summary>\nPRIOR\n</previous-summary>") != null); + try testing.expect(std.mem.indexOf(u8, body, "<transcript>\nTRANSCRIPT\n</transcript>") != null); +} + +test "latestSummaryText - returns latest, null when none" { + const a = testing.allocator; + { + var msgs = [_]Message{ try userMsg(a, "hi") }; + defer freeMsgs(a, &msgs); + try testing.expect(latestSummaryText(&msgs) == null); + } + { + var cs: std.ArrayList(conversation.ContentBlock) = .empty; + try cs.append(a, .{ .CompactionSummary = .{ + .text = try conversation.textualBlockFromSlice(a, "S2"), + } }); + var msgs = [_]Message{ + try userMsg(a, "hi"), + .{ .role = .user, .content = cs }, + }; + defer freeMsgs(a, &msgs); + try testing.expectEqualStrings("S2", latestSummaryText(&msgs).?); + } +} diff --git a/libpanto/src/config.zig b/libpanto/src/config.zig index 6b8d9a8..114280c 100644 --- a/libpanto/src/config.zig +++ b/libpanto/src/config.zig @@ -69,6 +69,24 @@ pub const ProviderConfig = union(APIStyle) { } }; +/// Compaction settings the agent consults when summarizing old history. +/// +/// `keep_verbatim` is the budget (in tokens) governing how much recent +/// conversation is kept verbatim after compaction. The retention walk +/// accumulates whole turns backward and stops once the running total +/// exceeds this value, so `keep_verbatim` is an *upper bound* on the size +/// of the kept suffix (the turn that crosses the threshold is summarized, +/// not kept). +/// +/// `model`, when set, is the provider/model used to run the compaction +/// request itself. On failure (e.g. the compaction model rejects the +/// transcript for context length) the agent falls back to the active chat +/// model. When null, compaction uses the active chat model directly. +pub const CompactionConfig = struct { + keep_verbatim: u32 = 20_000, + model: ?ProviderConfig = null, +}; + /// An immutable snapshot of everything the agent consults per turn: which /// provider/model to talk to, and which tools to expose. The agent holds a /// `*const Config` and re-reads it each turn; replacing the pointer swaps @@ -80,6 +98,7 @@ pub const ProviderConfig = union(APIStyle) { pub const Config = struct { provider: ProviderConfig, registry: *const ToolRegistry, + compaction: CompactionConfig = .{}, pub fn style(self: Config) APIStyle { return self.provider.style(); diff --git a/libpanto/src/conversation.zig b/libpanto/src/conversation.zig index f145d7f..780f650 100644 --- a/libpanto/src/conversation.zig +++ b/libpanto/src/conversation.zig @@ -68,12 +68,31 @@ pub const SystemBlock = struct { } }; +/// A compaction summary block. Carries the synthetic seed text that +/// stands in for a compacted conversation prefix. Like a `replace`-mode +/// `.System` block, it changes replay semantics: when the effective +/// conversation is rebuilt, a `.CompactionSummary` block resets all prior +/// *conversation* turns (user/assistant), and only the latest summary +/// plus the messages after it contribute to active context. System blocks +/// are unaffected — they are derived separately by `effectiveSystemBlocks` +/// and survive compaction untouched. +/// +/// A `.CompactionSummary` block sits alone in a `user`-role message. +pub const CompactionSummaryBlock = struct { + text: TextualBlock = .empty, + + pub fn deinit(self: *CompactionSummaryBlock, alloc: Allocator) void { + self.text.deinit(alloc); + } +}; + pub const ContentBlock = union(enum) { Text: TextualBlock, Thinking: ThinkingBlock, ToolUse: ToolUseBlock, ToolResult: ToolResultBlock, System: SystemBlock, + CompactionSummary: CompactionSummaryBlock, pub fn deinit(self: *ContentBlock, alloc: Allocator) void { switch (self.*) { @@ -82,6 +101,7 @@ pub const ContentBlock = union(enum) { .ToolUse => |*b| b.deinit(alloc), .ToolResult => |*b| b.deinit(alloc), .System => |*b| b.deinit(alloc), + .CompactionSummary => |*b| b.deinit(alloc), } } }; @@ -92,9 +112,52 @@ pub const MessageRole = enum { assistant, }; +/// Token usage reported by a provider for a single assistant turn. +/// +/// All input categories sum to the total prompt tokens billed for the turn: +/// `input + cache_read + cache_write` = total prompt tokens. +/// +/// Crucially, `input` is *cumulative*: it is the entire prompt sent for the +/// turn (the whole prior conversation), not the size of the turn's own +/// message. Compaction relies on this to back out per-turn sizes from +/// neighbor deltas (see `compaction.computeSplit`). +/// +/// `reasoning` is a **subset** of `output`, not additive — the portion of +/// the output spent on internal reasoning (OpenAI o-series, Anthropic +/// extended thinking). Cost is computed from `output` only; `reasoning` is +/// tracked for display. +/// +/// All fields are `u64` and default to 0. Providers that don't report a +/// given category leave it at 0. +/// +/// Defined here (rather than in `session.zig`) so in-memory `Message`s can +/// carry usage without a module cycle; `session.zig` re-exports it. +pub const Usage = struct { + /// Fresh input tokens billed at the model's base input rate. + input: u64 = 0, + /// Output tokens billed at the model's base output rate. + output: u64 = 0, + /// Input tokens served from cache (typ. 0.1× base input rate on + /// Anthropic, 0.5× on OpenAI). + cache_read: u64 = 0, + /// Input tokens written to a new cache entry (Anthropic only at + /// time of writing; typ. 1.25× base input rate). OpenAI's cache + /// hits don't bill a write premium, so this stays 0 there. + cache_write: u64 = 0, + /// Subset of `output` spent on reasoning. For OpenAI o-series + /// models and Anthropic extended thinking. Display-only — cost is + /// already accounted for via `output`. + reasoning: u64 = 0, +}; + pub const Message = struct { role: MessageRole, content: std.ArrayList(ContentBlock) = .empty, + /// Provider-reported token usage for this message's turn. Set on + /// assistant messages (live, via the provider; replayed, from disk); + /// null on user/system messages and when the provider emitted no usage. + /// Used by compaction to size the retention window. + usage: ?Usage = null, pub fn deinit(self: *Message, alloc: Allocator) void { for (self.content.items) |*block| { @@ -154,9 +217,36 @@ pub const Conversation = struct { }); } + /// Append a `user`-role message carrying a single `.CompactionSummary` + /// block holding `text` (copied). This is the seed that stands in for + /// a compacted conversation prefix; see `CompactionSummaryBlock`. + pub fn addCompactionSummary(self: *Conversation, text: []const u8) !void { + const tb = try textualBlockFromSlice(self.allocator, text); + var content: std.ArrayList(ContentBlock) = .empty; + errdefer { + for (content.items) |*b| b.deinit(self.allocator); + content.deinit(self.allocator); + } + try content.append(self.allocator, .{ .CompactionSummary = .{ .text = tb } }); + try self.messages.append(self.allocator, .{ + .role = .user, + .content = content, + }); + } + /// Append an assistant message. Ownership of the blocks is transferred /// to the conversation; the caller must not deinit them after this call. pub fn addAssistantMessage(self: *Conversation, blocks: []const ContentBlock) !void { + return self.addAssistantMessageWithUsage(blocks, null); + } + + /// Append an assistant message tagged with its provider-reported + /// `usage`. Ownership of the blocks is transferred to the conversation. + pub fn addAssistantMessageWithUsage( + self: *Conversation, + blocks: []const ContentBlock, + usage: ?Usage, + ) !void { var content: std.ArrayList(ContentBlock) = .empty; try content.ensureTotalCapacity(self.allocator, blocks.len); for (blocks) |block| { @@ -165,6 +255,7 @@ pub const Conversation = struct { try self.messages.append(self.allocator, .{ .role = .assistant, .content = content, + .usage = usage, }); } @@ -220,6 +311,42 @@ pub fn effectiveSystemBlocks( return out; } +/// Index of the message carrying the latest `.CompactionSummary` block, +/// or null if the conversation has never been compacted. +/// +/// Compaction replay is reset-like: only the latest compaction summary and +/// the messages after it contribute to active conversation context. This +/// returns the anchor message (the summary itself); active conversation is +/// `messages[anchor..]`. System messages are unaffected and are derived +/// independently via `effectiveSystemBlocks`. +pub fn latestCompactionIndex(messages: []const Message) ?usize { + var anchor: ?usize = null; + for (messages, 0..) |msg, i| { + for (msg.content.items) |block| { + if (block == .CompactionSummary) { + anchor = i; + break; + } + } + } + return anchor; +} + +/// The active (post-compaction) conversation message window that should be +/// sent to a provider. If the conversation has been compacted, this is the +/// latest compaction summary message plus everything after it; otherwise +/// it is the whole message list. System messages within the window are +/// still emitted/handled by the caller's own role filtering — this only +/// trims the *prefix* superseded by compaction. +/// +/// The returned slice borrows from `messages`. +pub fn activeMessageWindow(messages: []const Message) []const Message { + if (latestCompactionIndex(messages)) |anchor| { + return messages[anchor..]; + } + return messages; +} + test "Conversation - add messages and verify content" { const allocator = std.testing.allocator; @@ -360,3 +487,52 @@ test "effectiveSystemBlocks - prefix reconstructs prompt as of that point" { try std.testing.expectEqual(@as(usize, 1), blocks.items.len); try std.testing.expectEqualStrings("a", blocks.items[0]); } + +test "addCompactionSummary - sits alone in a user message" { + const allocator = std.testing.allocator; + + var conv = Conversation.init(allocator); + defer conv.deinit(); + + try conv.addUserMessage("hi"); + try conv.addCompactionSummary("summary of earlier history"); + + const m = conv.messages.items[1]; + try std.testing.expectEqual(MessageRole.user, m.role); + try std.testing.expectEqual(@as(usize, 1), m.content.items.len); + try std.testing.expectEqualStrings( + "summary of earlier history", + m.content.items[0].CompactionSummary.text.items, + ); +} + +test "latestCompactionIndex - null when never compacted" { + const allocator = std.testing.allocator; + + var conv = Conversation.init(allocator); + defer conv.deinit(); + + try conv.addSystemMessage("sys"); + try conv.addUserMessage("hi"); + try conv.addAssistantMessage(&.{ + .{ .Text = try textualBlockFromSlice(allocator, "hello") }, + }); + + try std.testing.expect(latestCompactionIndex(conv.messages.items) == null); +} + +test "latestCompactionIndex - returns the latest summary anchor" { + const allocator = std.testing.allocator; + + var conv = Conversation.init(allocator); + defer conv.deinit(); + + try conv.addSystemMessage("sys"); + try conv.addUserMessage("old"); + try conv.addCompactionSummary("S1"); + try conv.addUserMessage("mid"); + try conv.addCompactionSummary("S2"); // index 4 + try conv.addUserMessage("recent"); + + try std.testing.expectEqual(@as(?usize, 4), latestCompactionIndex(conv.messages.items)); +} diff --git a/libpanto/src/openai_chat_json.zig b/libpanto/src/openai_chat_json.zig index 8e6c9be..8035338 100644 --- a/libpanto/src/openai_chat_json.zig +++ b/libpanto/src/openai_chat_json.zig @@ -158,8 +158,11 @@ pub fn serializeRequest( try s.write(text); try s.endObject(); } - // Then every non-system message, in its original order. - for (conv.messages.items) |msg| { + // Then every non-system message, in its original order. If the + // conversation has been compacted, only the latest compaction summary + // and the messages after it are active; the superseded prefix is + // dropped. + for (conversation.activeMessageWindow(conv.messages.items)) |msg| { if (msg.role == .system) continue; try writeMessage(&s, msg, allocator); } @@ -309,6 +312,9 @@ fn concatTextBlocks( for (blocks) |block| { switch (block) { .Text => |tb| try out.appendSlice(allocator, tb.items), + // A compaction summary is the synthetic seed text standing in + // for a compacted prefix; emit it as ordinary message text. + .CompactionSummary => |cs| try out.appendSlice(allocator, cs.text.items), // Thinking, ToolUse, ToolResult: handled elsewhere or dropped. else => {}, } @@ -945,3 +951,41 @@ test "parseStreamEvent - bare-string error" { try testing.expectEqualStrings("something went wrong", pd.delta.error_message.?); try testing.expect(pd.delta.error_type == null); } + +test "serializeRequest - compaction summary trims superseded prefix" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + + // System prompt survives compaction. + try conv.addSystemMessage("you are helpful"); + // Old prefix that must be dropped. + try conv.addUserMessage("ancient question"); + try conv.addAssistantMessage(&.{ + .{ .Text = try conversation.textualBlockFromSlice(allocator, "ancient answer") }, + }); + // Compaction summary resets conversation context. + try conv.addCompactionSummary("summary of the ancient exchange"); + // Kept-verbatim suffix replayed after the summary. + try conv.addUserMessage("recent question"); + + var tools = emptyTools(); + defer tools.deinit(); + const cfg = testConfig("gpt-4o"); + const body = try serializeRequest(allocator, &cfg, &conv, &tools); + defer allocator.free(body); + + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); + defer parsed.deinit(); + + const msgs = parsed.value.object.get("messages").?.array.items; + // system + compaction summary (user) + recent question (user) = 3. + try testing.expectEqual(@as(usize, 3), msgs.len); + try testing.expectEqualStrings("system", msgs[0].object.get("role").?.string); + try testing.expectEqualStrings("you are helpful", msgs[0].object.get("content").?.string); + try testing.expectEqualStrings("user", msgs[1].object.get("role").?.string); + try testing.expectEqualStrings("summary of the ancient exchange", msgs[1].object.get("content").?.string); + try testing.expectEqualStrings("user", msgs[2].object.get("role").?.string); + try testing.expectEqualStrings("recent question", msgs[2].object.get("content").?.string); +} diff --git a/libpanto/src/provider.zig b/libpanto/src/provider.zig index 1a8ee4e..6ccbefe 100644 --- a/libpanto/src/provider.zig +++ b/libpanto/src/provider.zig @@ -14,6 +14,28 @@ pub const ContentBlockType = enum { ToolResult, }; +/// Heuristic detector for provider context-overflow rejections, applied to +/// an HTTP 400 error response body. Both OpenAI-compatible and Anthropic +/// APIs reject oversized requests on the input side with HTTP 400 and a +/// recognizable message; matching it lets the agent compact and retry +/// instead of surfacing a hard error. +/// +/// Markers (case-sensitive substrings, as the wire emits them): +/// - OpenAI: `context_length_exceeded`, `maximum context length` +/// - Anthropic: `prompt is too long` +pub fn isContextOverflowBody(body: []const u8) bool { + const markers = [_][]const u8{ + "context_length_exceeded", + "maximum context length", + "prompt is too long", + "context window", + }; + for (markers) |m| { + if (std.mem.indexOf(u8, body, m) != null) return true; + } + return false; +} + /// Vtable for receiving streaming events from a Provider. /// /// The lifecycle callbacks (`onMessageStart` ... `onMessageComplete`) return @@ -143,3 +165,12 @@ pub const StreamFn = *const fn ( conv: *conversation.Conversation, receiver: *Receiver, ) anyerror!void; + +test "isContextOverflowBody - matches known markers, rejects others" { + const t2 = std.testing; + try t2.expect(isContextOverflowBody("{\"error\":{\"code\":\"context_length_exceeded\"}}")); + try t2.expect(isContextOverflowBody("This model's maximum context length is 8192 tokens")); + try t2.expect(isContextOverflowBody("prompt is too long: 250000 tokens > 200000 maximum")); + try t2.expect(!isContextOverflowBody("{\"error\":{\"code\":\"invalid_api_key\"}}")); + try t2.expect(!isContextOverflowBody("rate limit exceeded")); +} diff --git a/libpanto/src/provider_anthropic_messages.zig b/libpanto/src/provider_anthropic_messages.zig index 5c24b8f..e77d6a5 100644 --- a/libpanto/src/provider_anthropic_messages.zig +++ b/libpanto/src/provider_anthropic_messages.zig @@ -110,6 +110,12 @@ pub const AnthropicMessagesRequest = struct { @intFromEnum(response.head.status), err_buf.items, }); + // Anthropic rejects oversized requests with HTTP 400 and a + // "prompt is too long" message; surface as ContextOverflow so + // the caller can compact and retry. + if (@intFromEnum(response.head.status) == 400 and provider_mod.isContextOverflowBody(err_buf.items)) { + return error.ContextOverflow; + } return error.HttpError; } @@ -329,35 +335,6 @@ const StreamState = struct { self.stop_reason = if (reason) |r| try self.allocator.dupe(u8, r) else null; } - fn isJSONObject(input: []const u8) bool { - if (input.len == 0) return true; - var parsed = std.json.parseFromSlice(std.json.Value, std.heap.page_allocator, input, .{}) catch return false; - defer parsed.deinit(); - return parsed.value == .object; - } - - fn appendAnthropicFailureResult( - allocator: Allocator, - blocks: *std.ArrayList(conversation.ContentBlock), - tool_use_id: []const u8, - input: []const u8, - stop_reason: ?[]const u8, - ) !void { - const id_copy = try allocator.dupe(u8, tool_use_id); - errdefer allocator.free(id_copy); - const msg = try std.fmt.allocPrint( - allocator, - "Tool call was not executed: Anthropic stopped before completing valid tool input JSON (stop_reason={s}). Partial input: {s}", - .{ stop_reason orelse "unknown", input }, - ); - errdefer allocator.free(msg); - var content: conversation.TextualBlock = .empty; - errdefer content.deinit(allocator); - try content.appendSlice(allocator, msg); - allocator.free(msg); - try blocks.append(allocator, .{ .ToolResult = .{ .tool_use_id = id_copy, .content = content } }); - } - /// Close the active block: append it to `blocks` and emit onBlockComplete. fn closeBlock( self: *StreamState, @@ -391,16 +368,13 @@ const StreamState = struct { .text = a.text_buf, .signature = a.signature, } }, + // An interrupted/malformed tool_use (incomplete or non-object + // input JSON) is preserved as-is. The agent's dispatch path + // detects invalid input and answers it with a synthetic error + // ToolResult in the *following user message* — emitting a + // ToolResult here would wrongly place it in this assistant + // message, which Anthropic rejects. .tool_use => blk: { - if (!isJSONObject(a.text_buf.items)) { - try appendAnthropicFailureResult( - self.allocator, - &self.blocks, - a.tool_id.?, - a.text_buf.items, - self.stop_reason, - ); - } break :blk .{ .ToolUse = .{ .id = a.tool_id.?, .name = a.tool_name.?, @@ -444,10 +418,10 @@ const StreamState = struct { const moved_blocks = try self.blocks.toOwnedSlice(self.allocator); defer self.allocator.free(moved_blocks); - try conv.addAssistantMessage(moved_blocks); + const usage: ?provider_mod.Usage = if (self.usage_seen) self.usage else null; + try conv.addAssistantMessageWithUsage(moved_blocks, usage); const msg = conv.messages.items[conv.messages.items.len - 1]; - const usage: ?provider_mod.Usage = if (self.usage_seen) self.usage else null; try receiver.onMessageComplete(msg, usage); } }; diff --git a/libpanto/src/provider_openai_chat.zig b/libpanto/src/provider_openai_chat.zig index d70eb30..c454679 100644 --- a/libpanto/src/provider_openai_chat.zig +++ b/libpanto/src/provider_openai_chat.zig @@ -137,6 +137,14 @@ pub const OpenAIChatRequest = struct { @intFromEnum(response.head.status), err_buf.items, }); + // Detect a context-overflow rejection so the caller can + // compact and retry rather than treating it as a hard error. + // OpenAI-compatible APIs return HTTP 400 with a + // `context_length_exceeded` code / "maximum context length" + // message on the input side. + if (@intFromEnum(response.head.status) == 400 and provider_mod.isContextOverflowBody(err_buf.items)) { + return error.ContextOverflow; + } return error.HttpError; } @@ -524,7 +532,7 @@ const StreamState = struct { const moved_blocks = try self.blocks.toOwnedSlice(self.allocator); defer self.allocator.free(moved_blocks); - try conv.addAssistantMessage(moved_blocks); + try conv.addAssistantMessageWithUsage(moved_blocks, self.usage); const msg = conv.messages.items[conv.messages.items.len - 1]; try receiver.onMessageComplete(msg, self.usage); diff --git a/libpanto/src/root.zig b/libpanto/src/root.zig index 68f8e7d..b882be5 100644 --- a/libpanto/src/root.zig +++ b/libpanto/src/root.zig @@ -11,6 +11,7 @@ pub const tool_registry = @import("tool_registry.zig"); pub const session = @import("session.zig"); pub const session_manager = @import("session_manager.zig"); pub const pricing = @import("pricing.zig"); +pub const compaction = @import("compaction.zig"); // Re-exports for ergonomic embedder use. pub const Tool = tool.Tool; diff --git a/libpanto/src/session.zig b/libpanto/src/session.zig index c1fd1c4..21193c8 100644 --- a/libpanto/src/session.zig +++ b/libpanto/src/session.zig @@ -129,36 +129,10 @@ pub const DiskMessage = struct { /// Token usage reported by a provider for a single assistant turn. /// -/// All four input categories sum to the total prompt tokens that were -/// billed for the turn: -/// `input + cache_read + cache_write` = total prompt tokens. -/// -/// `reasoning` is a **subset** of `output`, not additive — it's the -/// portion of the output that the model spent on internal reasoning -/// (OpenAI o-series, Anthropic extended thinking). Cost is computed -/// from `output` only; `reasoning` is tracked for display. -/// -/// All fields are `u64` and default to 0. Providers that don't report a -/// given category leave it at 0. Future cache-tier breakdowns (e.g. -/// Anthropic's 5m vs 1h tiers) can be added as new fields without -/// breaking the format. -pub const Usage = struct { - /// Fresh input tokens billed at the model's base input rate. - input: u64 = 0, - /// Output tokens billed at the model's base output rate. - output: u64 = 0, - /// Input tokens served from cache (typ. 0.1× base input rate on - /// Anthropic, 0.5× on OpenAI). - cache_read: u64 = 0, - /// Input tokens written to a new cache entry (Anthropic only at - /// time of writing; typ. 1.25× base input rate). OpenAI's cache - /// hits don't bill a write premium, so this stays 0 there. - cache_write: u64 = 0, - /// Subset of `output` spent on reasoning. For OpenAI o-series - /// models and Anthropic extended thinking. Display-only — cost is - /// already accounted for via `output`. - reasoning: u64 = 0, -}; +/// Defined in `conversation.zig` (so in-memory `Message`s can carry it +/// without a module cycle) and re-exported here for the on-disk types and +/// historical call sites that import it as `session.Usage`. +pub const Usage = conversation.Usage; // ============================================================================= // Content blocks @@ -169,6 +143,7 @@ pub const DiskContentBlock = union(enum) { thinking: DiskThinkingBlock, tool_use: DiskToolUseBlock, tool_result: DiskToolResultBlock, + compaction_summary: DiskCompactionSummaryBlock, pub fn deinit(self: DiskContentBlock, alloc: Allocator) void { switch (self) { @@ -176,6 +151,7 @@ pub const DiskContentBlock = union(enum) { .thinking => |b| b.deinit(alloc), .tool_use => |b| b.deinit(alloc), .tool_result => |b| b.deinit(alloc), + .compaction_summary => |b| b.deinit(alloc), } } }; @@ -219,6 +195,16 @@ pub const DiskToolResultBlock = struct { } }; +/// A compaction summary block: the synthetic seed text standing in for a +/// compacted conversation prefix. Sits alone in a `user`-role message. See +/// `conversation.CompactionSummaryBlock`. +pub const DiskCompactionSummaryBlock = struct { + text: []const u8, // owned + pub fn deinit(self: DiskCompactionSummaryBlock, alloc: Allocator) void { + alloc.free(self.text); + } +}; + // ============================================================================= // File entry (header or entry) // ============================================================================= @@ -400,6 +386,14 @@ fn writeDiskBlock(s: *std.json.Stringify, block: DiskContentBlock) !void { try s.write(b.content); try s.endObject(); }, + .compaction_summary => |b| { + try s.beginObject(); + try s.objectField("type"); + try s.write("compactionSummary"); + try s.objectField("text"); + try s.write(b.text); + try s.endObject(); + }, } } @@ -575,6 +569,9 @@ fn parseDiskBlock(allocator: Allocator, obj: std.json.ObjectMap) ParseError!Disk errdefer allocator.free(tuid); const content = try dupeStringField(allocator, obj, "content"); return .{ .tool_result = .{ .tool_use_id = tuid, .content = content } }; + } else if (std.mem.eql(u8, t, "compactionSummary")) { + const text = try dupeStringField(allocator, obj, "text"); + return .{ .compaction_summary = .{ .text = text } }; } else { return error.UnknownBlockType; } @@ -643,6 +640,10 @@ pub fn contentBlockToDisk( const text = try allocator.dupe(u8, sb.text.items); return .{ .text = .{ .text = text } }; }, + .CompactionSummary => |cs| { + const text = try allocator.dupe(u8, cs.text.items); + return .{ .compaction_summary = .{ .text = text } }; + }, } } @@ -681,6 +682,10 @@ pub fn diskContentBlockToInternal( const content = try conversation.textualBlockFromSlice(allocator, b.content); return .{ .ToolResult = .{ .tool_use_id = tuid, .content = content } }; }, + .compaction_summary => |b| { + const tb = try conversation.textualBlockFromSlice(allocator, b.text); + return .{ .CompactionSummary = .{ .text = tb } }; + }, } } @@ -1031,3 +1036,50 @@ test "diskContentBlockToInternal: Thinking preserves signature" { try testing.expectEqualStrings("reasoning...", inmem.Thinking.text.items); try testing.expectEqualStrings("sig123", inmem.Thinking.signature.?); } + +test "compactionSummary block round-trips through serialize/parse" { + const a = testing.allocator; + + var content = try a.alloc(DiskContentBlock, 1); + content[0] = .{ .compaction_summary = .{ .text = try dupe(a, "earlier history summary") } }; + + const entry: SessionEntry = .{ .message = .{ + .base = .{ + .id = try dupe(a, "cafef00d"), + .parent_id = null, + .timestamp = try dupe(a, "2026-04-25T17:40:00Z"), + }, + .message = .{ .role = .user, .content = content }, + } }; + defer entry.deinit(a); + + const line = try serializeEntry(a, entry); + defer a.free(line); + try testing.expect(std.mem.indexOf(u8, line, "\"type\":\"compactionSummary\"") != null); + + var fe = try parseLine(a, line); + defer fe.deinit(a); + const got = fe.entry.message; + try testing.expectEqual(DiskMessageRole.user, got.message.role); + try testing.expectEqualStrings("earlier history summary", got.message.content[0].compaction_summary.text); +} + +test "compactionSummary bridges in-memory <-> disk both directions" { + const a = testing.allocator; + + // in-memory -> disk + const tb = try conversation.textualBlockFromSlice(a, "S1"); + const block: conversation.ContentBlock = .{ .CompactionSummary = .{ .text = tb } }; + defer { + var mut = block; + mut.deinit(a); + } + const disk = try contentBlockToDisk(a, block); + defer disk.deinit(a); + try testing.expectEqualStrings("S1", disk.compaction_summary.text); + + // disk -> in-memory + var inmem = try diskContentBlockToInternal(a, disk); + defer inmem.deinit(a); + try testing.expectEqualStrings("S1", inmem.CompactionSummary.text.items); +} diff --git a/libpanto/src/session_manager.zig b/libpanto/src/session_manager.zig index 9e8e2c8..c6256e0 100644 --- a/libpanto/src/session_manager.zig +++ b/libpanto/src/session_manager.zig @@ -750,7 +750,13 @@ fn appendMessageToConv( .user => .user, .assistant => .assistant, }; - try conv.messages.append(allocator, .{ .role = role, .content = content }); + // Carry the recorded usage forward so compaction can size the retention + // window after a session is reopened (it's null for user/system). + try conv.messages.append(allocator, .{ + .role = role, + .content = content, + .usage = disk_msg.usage, + }); } // ============================================================================= @@ -1621,3 +1627,64 @@ test "resolveSessionId: unique prefix → match, ambiguous → error" { try testing.expectError(error.SessionNotFound, resolveSessionId(testing.allocator, io, sessions, "ffffffff")); } + +test "compaction summary round-trips through persist + resume + rebuild" { + const io = testing.io; + + var td = try TmpSessionDir.init(testing.allocator); + defer td.deinit(testing.allocator); + const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); + defer testing.allocator.free(sessions); + + const session_file: []u8 = blk: { + var mgr = try SessionManager.init(testing.allocator, io, sessions, "/c"); + defer mgr.deinit(); + + // System + an old turn that will be superseded. + const sys = try testing.allocator.alloc(DiskContentBlock, 1); + sys[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "you are helpful") } }; + _ = try mgr.appendMessage(.{ .role = .system, .content = sys }, null, null); + + const uo = try testing.allocator.alloc(DiskContentBlock, 1); + uo[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "old q") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = uo }, "openai", "gpt-4o"); + const ao = try testing.allocator.alloc(DiskContentBlock, 1); + ao[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "old a") } }; + _ = try mgr.appendMessage(.{ .role = .assistant, .content = ao }, null, null); + + // Compaction: summary message + duplicated kept suffix. + const cs = try testing.allocator.alloc(DiskContentBlock, 1); + cs[0] = .{ .compaction_summary = .{ .text = try testing.allocator.dupe(u8, "SUMMARY") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = cs }, "openai", "gpt-4o"); + + const ur = try testing.allocator.alloc(DiskContentBlock, 1); + ur[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "recent q") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = ur }, "openai", "gpt-4o"); + + break :blk try testing.allocator.dupe(u8, mgr.getSessionFile()); + }; + defer testing.allocator.free(session_file); + + var resumed = try SessionManager.open(testing.allocator, io, session_file); + defer resumed.deinit(); + var conv = try resumed.rebuildConversation(); + defer conv.deinit(); + + // The compaction summary block survived as a CompactionSummary. + const anchor = conversation_mod.latestCompactionIndex(conv.messages.items).?; + try testing.expectEqualStrings( + "SUMMARY", + conv.messages.items[anchor].content.items[0].CompactionSummary.text.items, + ); + + // The active window is [summary, recent q]. + const window = conversation_mod.activeMessageWindow(conv.messages.items); + try testing.expectEqual(@as(usize, 2), window.len); + try testing.expectEqualStrings("recent q", window[1].content.items[0].Text.items); + + // System prompt survives (derived independently). + var sys_blocks = try conversation_mod.effectiveSystemBlocks(testing.allocator, conv.messages.items); + defer sys_blocks.deinit(testing.allocator); + try testing.expectEqual(@as(usize, 1), sys_blocks.items.len); + try testing.expectEqualStrings("you are helpful", sys_blocks.items[0]); +} |