From ccdaef8e682960ecadf73d3b2df70559a0baaf56 Mon Sep 17 00:00:00 2001 From: t Date: Fri, 5 Jun 2026 11:43:29 -0600 Subject: Move session persistence into the Agent; collapse CLI plumbing The Agent now owns its Conversation and a SessionStore, and persists everything it generates as turns progress. Embedders get persistence for free; the CLI no longer drives the session log. libpanto: - Agent.init takes a SessionStore + optional Conversation to adopt (resume path). Agent owns/tears down the conversation; turn-driving methods drop the *Conversation param and operate on self.conversation. - New Agent methods: submitUserMessage (adds+persists the user prompt immediately), addSystemMessage(text, mode), runStep persists the turn on every exit path (incl. partial turns on error), compactAndPersist for the explicit /compact path. Auto-compaction persists its window via runStep's tail. - turn_persist.zig: DiskMessage mapping moved out of the CLI; reads usage off Message.usage (no per-message-usage list). System mode rides on DiskMessage.mode, derived from the System block. - NullStore is now an allocator-bearing struct (frees consumed messages per the store-consumes-messages contract). - Tests: in-memory CapturingStore round-trip + NullStore turn test. panto CLI: - Delete src/session_persist.zig. REPL is submitUserMessage + runStep. - Reorder construction: store -> registry -> agent -> bootstrap -> seed/ reconcile system prompt through the agent -> load extensions. - command.Context drops conv/session_mgr; commands reach ctx.agent. - system_prompt seed/reconcile call agent.addSystemMessage. - CLIReceiver is display-only (per_message_usage removed). Phases 3-5 of docs/pluggable-session-store.md. Behavior preserved; full build + test suite green. --- libpanto/src/turn_persist.zig | 170 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 170 insertions(+) create mode 100644 libpanto/src/turn_persist.zig (limited to 'libpanto/src/turn_persist.zig') 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; +} -- cgit v1.3