From b14859b9726185ab873356390068e887b7f486d3 Mon Sep 17 00:00:00 2001 From: t Date: Sun, 7 Jun 2026 11:35:49 -0600 Subject: R2: redesign session store with wire-format identity Replace the single-session SessionManager seam with a directory-backed catalog. session_manager.zig -> file_system_jsonl_store.zig; the old machinery becomes internal SessionFile, and a new FileSystemJSONLStore implements the redesigned SessionStore vtable (create/list/resolve/latest/ load/appendMessages) minting Session/SessionInfo handles. Session logs now record wire-format provider identity ({api_style, base_url, model, reasoning}) instead of a single provider string or CLI aliases; no api_key material is ever stored. Disk* content types renamed Stored*; the rich audit-oriented write record is PersistentMessage (in-memory Message + WireIdentity + provenance). Agent.init now takes a Session; persist_provider/persist_model display strings deleted (banner stays alias-based, resume picks default model). Message.metadata round-trips. Dangling-prompt recovery dropped. CLI migrated to the catalog store for run/resume/list. Clean break, no version bump (old logs wiped). --- libpanto/src/agent.zig | 240 ++-- libpanto/src/config.zig | 31 + libpanto/src/conversation.zig | 7 + libpanto/src/file_system_jsonl_store.zig | 2075 ++++++++++++++++++++++++++++++ libpanto/src/null_store.zig | 106 +- libpanto/src/root.zig | 2 +- libpanto/src/session.zig | 302 +++-- libpanto/src/session_manager.zig | 1886 --------------------------- libpanto/src/session_store.zig | 262 ++-- libpanto/src/turn_persist.zig | 153 +-- 10 files changed, 2677 insertions(+), 2387 deletions(-) create mode 100644 libpanto/src/file_system_jsonl_store.zig delete mode 100644 libpanto/src/session_manager.zig (limited to 'libpanto') diff --git a/libpanto/src/agent.zig b/libpanto/src/agent.zig index e8be325..604ee55 100644 --- a/libpanto/src/agent.zig +++ b/libpanto/src/agent.zig @@ -197,20 +197,12 @@ pub const Agent = struct { /// at `init`); torn down in `deinit`. Turn-driving methods operate on /// this directly rather than taking a `*Conversation` parameter. conversation: conversation.Conversation, - /// Pluggable persistence backend. Every message that enters - /// `conversation` is persisted here as it is added. Defaults to a no-op - /// `NullStore` so embedders/tests that don't care about persistence - /// keep working. Borrowed: the embedder owns the concrete store and - /// must outlive the agent. - session_store: session_store_mod.SessionStore, - /// Display provider/model names used to stamp persisted entries. These - /// are the embedder's user-facing identity strings (e.g. the - /// `"provider:model"` ref's parts), which can't be derived from the - /// `ProviderConfig` snapshot (it only carries the `APIStyle` tag and - /// the raw model string). Borrowed; lifetime owned by the embedder. The - /// model defaults to the config snapshot's model when left empty. - persist_provider: []const u8 = "", - persist_model: []const u8 = "", + /// The session this agent appends to. Minted from the store at `init` + /// (fresh: `store.create()`) or supplied by the embedder on resume + /// (`resolve`/`latest`). `Session.append` proxies to the store and + /// updates the session's last-used wire identity. The embedder owns the + /// underlying store, which must outlive the agent. + session: session_store_mod.Session, /// Injectable streaming seam. Defaults to the real provider dispatch /// (`provider_mod.openStream`); tests override it with a stub. open_stream_fn: provider_mod.OpenStreamFn = provider_mod.openStream, @@ -239,7 +231,7 @@ pub const Agent = struct { allocator: Allocator, io: Io, config: *const Config, - store: session_store_mod.SessionStore, + session: session_store_mod.Session, maybe_conversation: ?conversation.Conversation, ) Agent { return .{ @@ -248,16 +240,18 @@ pub const Agent = struct { .config = config, .registry = ToolRegistry.init(allocator), .conversation = maybe_conversation orelse conversation.Conversation.init(allocator), - .session_store = store, + .session = session, }; } pub fn deinit(self: *Agent) void { - // The agent owns the conversation and the tool registry; it borrows - // the config snapshot and the session store, which the embedder - // tears down. + // The agent owns the conversation, the tool registry, and the + // session handle's `info` (minted by `store.create()` or resolved + // by the embedder and handed in). It borrows the config snapshot + // and the underlying store, which the embedder tears down. self.registry.deinit(); self.conversation.deinit(); + self.session.info.deinit(self.allocator); } /// Add a single tool to this agent's tool set. Visible at the next turn. @@ -271,22 +265,11 @@ pub const Agent = struct { try self.registry.registerSource(src); } - /// The provider/model identity used to stamp persisted entries. Uses - /// the embedder-supplied display names when set, falling back to the - /// active config snapshot's model string (and the `APIStyle` tag name - /// for the provider) when not. - fn providerModel(self: *const Agent) struct { provider: []const u8, model: []const u8 } { - const cfg_model = switch (self.config.provider) { - inline else => |c| c.model, - }; - const provider = if (self.persist_provider.len > 0) - self.persist_provider - else switch (self.config.provider) { - .openai_chat => "openai_chat", - .anthropic_messages => "anthropic_messages", - }; - const model = if (self.persist_model.len > 0) self.persist_model else cfg_model; - return .{ .provider = provider, .model = model }; + /// The wire-format provider identity stamped on persisted entries, + /// derived from the active config snapshot. Ground truth: never a CLI + /// alias, never any `api_key` material. + fn wireIdentity(self: *const Agent) session_store_mod.WireIdentity { + return self.config.provider.wireIdentity(); } /// Swap the active configuration snapshot. Takes effect at the start of @@ -312,14 +295,13 @@ pub const Agent = struct { .append => try self.conversation.addSystemMessage(text), .replace => try self.conversation.replaceSystemMessage(text), } - const pm = self.providerModel(); try turn_persist.persistTurn( self.allocator, - self.session_store, + &self.session, &self.conversation, start, - pm.provider, - pm.model, + self.wireIdentity(), + &.{}, ); } @@ -355,14 +337,13 @@ pub const Agent = struct { // recovery guarantee). const user_start = self.conversation.messages.items.len; try self.conversation.addUserMessage(message.text); - const pm = self.providerModel(); try turn_persist.persistTurn( self.allocator, - self.session_store, + &self.session, &self.conversation, user_start, - pm.provider, - pm.model, + self.wireIdentity(), + &.{}, ); const s = try self.allocator.create(Stream); @@ -375,23 +356,23 @@ pub const Agent = struct { /// `[system..., summary, kept-suffix...]`), so persist the whole /// post-compaction window instead of `[start..]`. fn persistTurnTail(self: *Agent, start: usize) !void { - const pm = self.providerModel(); + const id = self.wireIdentity(); if (self.auto_compacted) { try turn_persist.persistCompaction( self.allocator, - self.session_store, + &self.session, &self.conversation, - pm.provider, - pm.model, + id, + &.{}, ); } else { try turn_persist.persistTurn( self.allocator, - self.session_store, + &self.session, &self.conversation, start, - pm.provider, - pm.model, + id, + &.{}, ); } } @@ -621,13 +602,12 @@ pub const Agent = struct { ) !CompactionResult { const res = try self.compact(system_prompt, extra_instructions); if (res.compacted) { - const pm = self.providerModel(); try turn_persist.persistCompaction( self.allocator, - self.session_store, + &self.session, &self.conversation, - pm.provider, - pm.model, + self.wireIdentity(), + &.{}, ); } return res; @@ -1798,59 +1778,81 @@ const HardFailTool = struct { }; /// An in-memory `SessionStore` test double: records every appended -/// `DiskMessage` (role + provider/model stamp) so tests can assert the +/// `StoredMessage` (role + provider/model stamp) so tests can assert the /// agent persisted the right turn without touching disk. Honors the store /// ownership contract by freeing each consumed message after recording its /// salient fields. const CapturingStore = struct { allocator: Allocator, - roles: std.ArrayList(session_store_mod.DiskMessageRole) = .empty, - providers: std.ArrayList(?[]const u8) = .empty, + roles: std.ArrayList(conversation.MessageRole) = .empty, + base_urls: std.ArrayList([]const u8) = .empty, fn init(allocator: Allocator) CapturingStore { return .{ .allocator = allocator }; } fn deinit(self: *CapturingStore) void { - for (self.providers.items) |p| if (p) |s| self.allocator.free(s); - self.providers.deinit(self.allocator); + for (self.base_urls.items) |s| self.allocator.free(s); + self.base_urls.deinit(self.allocator); self.roles.deinit(self.allocator); } - fn appendMessagesVT( - ctx: *anyopaque, - messages: []session_store_mod.DiskMessage, - providers: []const ?[]const u8, - _: []const ?[]const u8, - ) anyerror!void { + fn createVT(ctx: *anyopaque) session_store_mod.Session { const self: *CapturingStore = @ptrCast(@alignCast(ctx)); - for (messages, 0..) |m, i| { - try self.roles.append(self.allocator, m.role); - // Prefer the assistant message's own provider stamp, else the - // entry-level provider stamp (user messages). - const prov = m.provider orelse providers[i]; - const dup: ?[]const u8 = if (prov) |p| try self.allocator.dupe(u8, p) else null; - try self.providers.append(self.allocator, dup); - } - // Contract: appends consume their messages. - for (messages) |m| m.deinit(self.allocator); + const a = self.allocator; + const info: session_store_mod.SessionInfo = .{ + .id = a.dupe(u8, "cap") catch "cap", + .created = a.dupe(u8, "") catch "", + .modified = a.dupe(u8, "") catch "", + .message_count = 0, + .last_user_message = a.dupe(u8, "") catch "", + .api_style = .openai_chat, + .base_url = a.dupe(u8, "") catch "", + .model = a.dupe(u8, "") catch "", + .reasoning = .default, + }; + return .{ .info = info, .store = self.store() }; } - fn loadConversationVT(_: *anyopaque, alloc: Allocator) anyerror!session_store_mod.LoadedSession { - return .{ .conversation = conversation.Conversation.init(alloc), .dangling_user = null }; + fn listVT(ctx: *anyopaque) anyerror![]session_store_mod.SessionInfo { + const self: *CapturingStore = @ptrCast(@alignCast(ctx)); + return self.allocator.alloc(session_store_mod.SessionInfo, 0); + } + fn freeSessionInfosVT(ctx: *anyopaque, infos: []session_store_mod.SessionInfo) void { + const self: *CapturingStore = @ptrCast(@alignCast(ctx)); + for (infos) |i| i.deinit(self.allocator); + self.allocator.free(infos); } - fn sessionIdVT(_: *anyopaque) []const u8 { - return "cap"; + fn resolveVT(_: *anyopaque, _: []const u8) anyerror!?session_store_mod.Session { + return null; + } + fn latestVT(_: *anyopaque) anyerror!?session_store_mod.Session { + return null; } - fn activeModelVT(_: *anyopaque) ?session_store_mod.ActiveModel { + fn loadVT(_: *anyopaque, _: []const u8) anyerror!?conversation.Conversation { return null; } + fn appendMessagesVT( + ctx: *anyopaque, + _: []const u8, + messages: []session_store_mod.PersistentMessage, + ) anyerror!void { + const self: *CapturingStore = @ptrCast(@alignCast(ctx)); + for (messages) |m| { + try self.roles.append(self.allocator, m.message.role); + try self.base_urls.append(self.allocator, try self.allocator.dupe(u8, m.identity.base_url)); + } + } + const vtable: session_store_mod.SessionStore.VTable = .{ + .create = createVT, + .list = listVT, + .freeSessionInfos = freeSessionInfosVT, + .resolve = resolveVT, + .latest = latestVT, + .load = loadVT, .appendMessages = appendMessagesVT, - .loadConversation = loadConversationVT, - .sessionId = sessionIdVT, - .activeModel = activeModelVT, }; fn store(self: *CapturingStore) session_store_mod.SessionStore { @@ -1880,27 +1882,25 @@ test "agent persists user, assistant, and tool-result messages of a turn" { var cap = CapturingStore.init(allocator); defer cap.deinit(); - var agent = Agent.init(allocator, io, &h.config, cap.store(), null); + var agent = Agent.init(allocator, io, &h.config, cap.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); - agent.persist_provider = "openai"; - agent.persist_model = "gpt-4o"; try drainTurn(&agent, "call a tool"); // Persisted, in order: user prompt, assistant(ToolUse), user(ToolResult), // assistant(text). try testing.expectEqual(@as(usize, 4), cap.roles.items.len); - try testing.expectEqual(session_store_mod.DiskMessageRole.user, cap.roles.items[0]); - try testing.expectEqual(session_store_mod.DiskMessageRole.assistant, cap.roles.items[1]); - try testing.expectEqual(session_store_mod.DiskMessageRole.user, cap.roles.items[2]); - try testing.expectEqual(session_store_mod.DiskMessageRole.assistant, cap.roles.items[3]); + try testing.expectEqual(conversation.MessageRole.user, cap.roles.items[0]); + try testing.expectEqual(conversation.MessageRole.assistant, cap.roles.items[1]); + try testing.expectEqual(conversation.MessageRole.user, cap.roles.items[2]); + try testing.expectEqual(conversation.MessageRole.assistant, cap.roles.items[3]); - // The display provider stamp rode through on every entry. - for (cap.providers.items) |p| { - try testing.expect(p != null); - try testing.expectEqualStrings("openai", p.?); + // The wire identity (base_url from the active config) rode through on + // every entry. + for (cap.base_urls.items) |b| { + try testing.expectEqualStrings("u", b); } } @@ -1919,7 +1919,7 @@ test "agent runs a turn against NullStore without persisting or erroring" { h.activate(); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2209,7 +2209,7 @@ test "runStep dispatches a tool call and loops to a final text turn" { try h.registry.register(try EchoTool.create(allocator, "echo", "ECHO:")); h.activate(); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2260,7 +2260,7 @@ test "runStep dispatches multiple tool calls in parallel" { try h.registry.register(try BarrierTool.create(allocator, "barrierC", &barrier)); h.activate(); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2300,7 +2300,7 @@ test "runStep: native tool handler error becomes an error result and the model g try h.registry.register(try FailingTool.create(allocator, "boom")); h.activate(); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2334,7 +2334,7 @@ test "runStep: unknown tool becomes an error tool result and the loop continues" defer h.deinit(); h.activate(); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2367,7 +2367,7 @@ test "runStep with no tool calls returns after one provider step" { defer h.deinit(); h.activate(); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2394,7 +2394,7 @@ test "runStep surfaces EmptyAssistantResponse when provider commits an empty mes defer h.deinit(); h.activate(); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2425,7 +2425,7 @@ test "runStep delivers all source-backed calls in one batch on one thread" { try h.registry.registerSource(try TestSource.create(allocator, "panto-lua", &.{ "lua_x", "lua_y" })); h.activate(); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2477,7 +2477,7 @@ test "runStep: distinct sources run on distinct threads in parallel" { try h.registry.registerSource(try TestSource.create(allocator, "src_b", &.{"src_b_t"})); h.activate(); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2515,7 +2515,7 @@ test "runStep: source whole-batch error becomes per-call error results and conti try h.registry.registerSource(try FailingSource.create(allocator, "fs", &.{ "fa", "fb" })); h.activate(); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2559,7 +2559,7 @@ test "runStep: mixed single Tools and source-backed tools coexist in one turn" { try h.registry.registerSource(try TestSource.create(allocator, "src", &.{ "src_t1", "src_t2" })); h.activate(); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2599,7 +2599,7 @@ test "setConfig swaps provider between turns; agent tool set persists" { }; var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &cfg_a, ns.store(), null); + var agent = Agent.init(allocator, io, &cfg_a, ns.store().create(), null); defer agent.deinit(); try agent.registerTool(try EchoTool.create(allocator, "late", "B:")); agent.open_stream_fn = stub.install(); @@ -2638,7 +2638,7 @@ test "compact: summarizes prefix, keeps suffix, system survives" { // adding the longer first turn exceeds it. h.config.compaction = .{ .keep_verbatim = 10 }; var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2701,7 +2701,7 @@ test "compact: restated suffix usage reconstructs a fresh cumulative chain" { // summarizes the prefix. h.config.compaction = .{ .keep_verbatim = 20 }; var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2775,7 +2775,7 @@ test "compact: no-op when conversation already fits the budget" { h.activate(); h.config.compaction = .{ .keep_verbatim = 1_000_000 }; var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2813,7 +2813,7 @@ test "compact: extra instructions are appended to the system prompt" { h.activate(); h.config.compaction = .{ .keep_verbatim = 1 }; var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2854,7 +2854,7 @@ test "runStep: auto-compacts on context overflow and retries once" { h.activate(); h.config.compaction = .{ .keep_verbatim = 10 }; var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2899,7 +2899,7 @@ test "runStep: context overflow without compaction prompt propagates" { defer h.deinit(); h.activate(); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -2971,7 +2971,7 @@ test "runStep: provider 429 retries then succeeds without duplicate messages" { defer h.deinit(); fastRetryHarness(&h); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -3015,7 +3015,7 @@ test "runStep: provider 500 retries with backoff notification" { defer h.deinit(); fastRetryHarness(&h); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -3053,7 +3053,7 @@ test "runStep: provider auth failure does not retry" { defer h.deinit(); fastRetryHarness(&h); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -3092,7 +3092,7 @@ test "runStep: retries exhaust and hard-fail after max_attempts" { defer h.deinit(); fastRetryHarness(&h); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -3130,7 +3130,7 @@ test "runStep: Retry-After is honored and reported" { // Cap below the Retry-After to verify the policy cap applies. h.config.retry = .{ .initial_delay_ms = 0, .max_delay_ms = 1, .jitter = false }; var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -3165,7 +3165,7 @@ test "runStep: cancellation from a tool still hard-fails" { try h.registry.register(try HardFailTool.create(allocator, "hard")); h.activate(); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -3196,7 +3196,7 @@ test "runStep: source per-call error produces a per-call error result and contin try h.registry.registerSource(try PartialSource.create(allocator, "ps", &.{ "pa", "pb" })); h.activate(); var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); @@ -3235,7 +3235,7 @@ test "runStep: context-overflow compaction fires a compaction retry notification h.activate(); h.config.compaction = .{ .keep_verbatim = 10 }; var ns = null_store_mod.NullStore.init(allocator); - var agent = Agent.init(allocator, io, &h.config, ns.store(), null); + var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); agent.open_stream_fn = stub.install(); diff --git a/libpanto/src/config.zig b/libpanto/src/config.zig index 8efa810..203a091 100644 --- a/libpanto/src/config.zig +++ b/libpanto/src/config.zig @@ -67,6 +67,37 @@ pub const ProviderConfig = union(APIStyle) { pub fn style(self: ProviderConfig) APIStyle { return @as(APIStyle, self); } + + /// The wire-format provider identity for this config: the ground-truth + /// `{api_style, base_url, model, reasoning}` that a turn is sent with. + /// Anthropic has no reasoning-effort knob, so it reports `.default`. + /// Borrowed slices; valid as long as the config is. + pub fn wireIdentity(self: ProviderConfig) WireIdentity { + return switch (self) { + .openai_chat => |c| .{ + .api_style = .openai_chat, + .base_url = c.base_url, + .model = c.model, + .reasoning = c.reasoning, + }, + .anthropic_messages => |c| .{ + .api_style = .anthropic_messages, + .base_url = c.base_url, + .model = c.model, + .reasoning = .default, + }, + }; + } +}; + +/// Wire-format provider identity (see `ProviderConfig.wireIdentity`). This +/// is the same shape as `session_store.WireIdentity`; defined here to avoid +/// a module cycle (config must not import session_store). +pub const WireIdentity = struct { + api_style: APIStyle, + base_url: []const u8, + model: []const u8, + reasoning: ReasoningEffort = .default, }; /// Compaction settings the agent consults when summarizing old history. diff --git a/libpanto/src/conversation.zig b/libpanto/src/conversation.zig index f0c822d..1fc5016 100644 --- a/libpanto/src/conversation.zig +++ b/libpanto/src/conversation.zig @@ -209,6 +209,13 @@ pub const Message = struct { /// null on user/system messages and when the provider emitted no usage. /// Used by compaction to size the retention window. usage: ?Usage = null, + /// Opaque per-message metadata bag. `libpanto` never interprets these + /// bytes; the documented contract is that, when present, they are valid + /// JSON (so a store may keep them as a JSON column and tools may + /// deserialize them). Round-trips through persistence: set before a turn + /// commits, read back off the `Message` after `load`. Borrowed; owned by + /// whoever set it (the conversation's allocator on the replay path). + metadata: ?[]const u8 = null, pub fn deinit(self: *Message, alloc: Allocator) void { for (self.content.items) |*block| { diff --git a/libpanto/src/file_system_jsonl_store.zig b/libpanto/src/file_system_jsonl_store.zig new file mode 100644 index 0000000..10b65b3 --- /dev/null +++ b/libpanto/src/file_system_jsonl_store.zig @@ -0,0 +1,2075 @@ +//! Session lifecycle: create, open, replay, append. +//! +//! Backed by an append-only JSONL file on disk. The on-disk types live in +//! `session.zig`. This module owns: +//! +//! - Path resolution (sessions dir is supplied by the caller; we own the +//! filename and writes within it). +//! - The in-memory entry index (`by_id` map + leaf pointer). +//! - Deferred file creation: the file is not written until the first +//! assistant message persists. Until that point, all entries are +//! buffered in memory. +//! - Append semantics: once flushed, every completed entry is written +//! and synced to disk immediately. +//! - Crash recovery: on open, the file is parsed line-by-line; the first +//! line that fails to parse causes everything from that line onward to +//! be truncated from the file. +//! - One-time format migration when a future version reads a v1 file +//! (currently a no-op; the hook is in place). +//! - Rebuilding a `Conversation` from the entry tree, plus determining +//! the active provider/model. +//! +//! The library-vs-CLI boundary: callers pass an absolute path to the +//! per-cwd sessions directory. We compute the per-session filename +//! ourselves (`.jsonl`) and lazily mkdir the directory on the +//! first flush. The CLI owns XDG resolution, encoded-cwd grouping, and +//! the `--resume` flag plumbing. + +const std = @import("std"); +const Allocator = std.mem.Allocator; +const Io = std.Io; + +const session_mod = @import("session.zig"); +const conversation_mod = @import("conversation.zig"); +const session_store_mod = @import("session_store.zig"); + +pub const SessionHeader = session_mod.SessionHeader; +pub const SessionEntry = session_mod.SessionEntry; +pub const MessageEntry = session_mod.MessageEntry; +pub const StoredMessage = session_mod.StoredMessage; +pub const StoredMessageRole = session_mod.StoredMessageRole; +pub const StoredSystemMode = session_mod.StoredSystemMode; +pub const StoredContentBlock = session_mod.StoredContentBlock; +pub const Usage = session_mod.Usage; +pub const CURRENT_VERSION = session_mod.CURRENT_VERSION; + +// ============================================================================= +// IDs and timestamps +// ============================================================================= + +/// Generate a UUIDv7 (RFC 9562 §5.7). Returns a 36-character canonical +/// hex string with hyphens. Caller owns. +/// +/// Layout: +/// - 48 bits: unix_ts_ms (big-endian) +/// - 4 bits: version (7) +/// - 12 bits: random +/// - 2 bits: variant (10) +/// - 62 bits: random +pub fn newUuidV7(allocator: Allocator, io: Io) ![]u8 { + const ts = Io.Timestamp.now(io, .real); + const now_ms: u64 = @intCast(@max(ts.toMilliseconds(), 0)); + + var rand_bytes: [10]u8 = undefined; + io.random(&rand_bytes); + + var b: [16]u8 = undefined; + // Timestamp (48 bits, big-endian). + b[0] = @intCast((now_ms >> 40) & 0xFF); + b[1] = @intCast((now_ms >> 32) & 0xFF); + b[2] = @intCast((now_ms >> 24) & 0xFF); + b[3] = @intCast((now_ms >> 16) & 0xFF); + b[4] = @intCast((now_ms >> 8) & 0xFF); + b[5] = @intCast(now_ms & 0xFF); + // Version (4 high bits = 0x7) + 12 bits random. + b[6] = 0x70 | (rand_bytes[0] & 0x0F); + b[7] = rand_bytes[1]; + // Variant (2 high bits = 10) + 62 bits random. + b[8] = 0x80 | (rand_bytes[2] & 0x3F); + b[9] = rand_bytes[3]; + b[10] = rand_bytes[4]; + b[11] = rand_bytes[5]; + b[12] = rand_bytes[6]; + b[13] = rand_bytes[7]; + b[14] = rand_bytes[8]; + b[15] = rand_bytes[9]; + + return try std.fmt.allocPrint( + allocator, + "{x:0>2}{x:0>2}{x:0>2}{x:0>2}-{x:0>2}{x:0>2}-{x:0>2}{x:0>2}-{x:0>2}{x:0>2}-{x:0>2}{x:0>2}{x:0>2}{x:0>2}{x:0>2}{x:0>2}", + .{ b[0], b[1], b[2], b[3], b[4], b[5], b[6], b[7], b[8], b[9], b[10], b[11], b[12], b[13], b[14], b[15] }, + ); +} + +/// Generate a fresh 8-character hex entry id. Caller owns. +fn newEntryIdInto(buf: []u8, io: Io) void { + std.debug.assert(buf.len == 8); + var bytes: [4]u8 = undefined; + io.random(&bytes); + _ = std.fmt.bufPrint(buf, "{x:0>2}{x:0>2}{x:0>2}{x:0>2}", .{ bytes[0], bytes[1], bytes[2], bytes[3] }) catch unreachable; +} + +/// Format `now` as an ISO 8601 UTC string with millisecond precision. +/// Example: `2026-04-25T17:40:15.990Z`. Caller owns. +pub fn isoTimestamp(allocator: Allocator, io: Io) ![]u8 { + const ts = Io.Timestamp.now(io, .real); + const ms_total: i64 = ts.toMilliseconds(); + const seconds_total: i64 = @divTrunc(ms_total, 1000); + const ms: u64 = @intCast(@mod(ms_total, 1000)); + + const epoch_secs = std.time.epoch.EpochSeconds{ .secs = @intCast(seconds_total) }; + const epoch_day = epoch_secs.getEpochDay(); + const day_secs = epoch_secs.getDaySeconds(); + const year_day = epoch_day.calculateYearDay(); + const month_day = year_day.calculateMonthDay(); + + return try std.fmt.allocPrint( + allocator, + "{d:0>4}-{d:0>2}-{d:0>2}T{d:0>2}:{d:0>2}:{d:0>2}.{d:0>3}Z", + .{ + @as(u32, year_day.year), + month_day.month.numeric(), + @as(u32, month_day.day_index) + 1, + day_secs.getHoursIntoDay(), + day_secs.getMinutesIntoHour(), + day_secs.getSecondsIntoMinute(), + ms, + }, + ); +} + +// ============================================================================= +// FileInfo (per-file listing scan result) +// ============================================================================= + +/// Internal scan result for one session file. Carries the file `path` (the +/// catalog needs it to open/resolve) plus everything the public +/// `session_store.SessionInfo` needs. +const FileInfo = struct { + path: []u8, + id: []u8, + created: []u8, // ISO 8601 from header timestamp + modified: []u8, // ISO 8601 from last activity, falling back to header + message_count: usize, + last_user_message: []u8, + stamp: ?session_mod.WireStamp, // last-used wire identity + + pub fn deinit(self: FileInfo, alloc: Allocator) void { + alloc.free(self.path); + alloc.free(self.id); + alloc.free(self.created); + alloc.free(self.modified); + alloc.free(self.last_user_message); + if (self.stamp) |s| s.deinit(alloc); + } +}; + +// ============================================================================= +// SessionFile +// ============================================================================= + +pub const Error = error{ + NoSessionsFound, + AmbiguousSessionId, + SessionNotFound, + InvalidSessionFile, +} || Allocator.Error || Io.Cancelable; + +pub const SessionFile = struct { + allocator: Allocator, + io: Io, + + /// Absolute path to the per-cwd sessions directory. Lazily created. + session_dir: []u8, + /// Absolute path to the file we *will* write to (computed at init). + /// May not yet exist on disk if `flushed = false`. + session_file: []u8, + + /// Header. Allocated at init for new sessions; reloaded from the file + /// on resume. + header: SessionHeader, + + /// Entries indexed in insertion order. The first entry's `parent_id` + /// is null; each subsequent entry's `parent_id` points to its parent + /// (currently always the previous entry). + entries: std.ArrayList(SessionEntry), + /// id → entry index in `entries`. Used both for parent-id lookups + /// and for collision detection in `newEntryId`. + by_id: std.StringHashMap(usize), + /// id of the most recently appended entry, or null if no entries yet. + /// Borrowed from the entry; do not free. + leaf_id: ?[]const u8, + + /// True once the file exists on disk. False during the "buffered" + /// pre-assistant phase. See module-level docs. + flushed: bool, + /// Number of bytes written to `session_file` so far. Used as the + /// offset for the next positional write. Only meaningful when + /// `flushed = true`. + written_bytes: u64, + + // ---------- Construction ---------- + + /// Create a new session in memory. Allocates a UUIDv7, computes the + /// file path, but does NOT touch the filesystem. The file is created + /// on the first assistant-message flush. + /// + /// `session_dir` is duplicated; the caller retains ownership of the + /// passed slice. + pub fn init( + allocator: Allocator, + io: Io, + session_dir: []const u8, + cwd: []const u8, + ) !SessionFile { + const id = try newUuidV7(allocator, io); + defer allocator.free(id); + return initWithId(allocator, io, session_dir, cwd, id); + } + + /// Like `init`, but uses a caller-supplied session id (duped here) + /// rather than minting a fresh UUIDv7. Used by the catalog when a + /// `Session` handle was minted (with its id) before the first append. + pub fn initWithId( + allocator: Allocator, + io: Io, + session_dir: []const u8, + cwd: []const u8, + session_id: []const u8, + ) !SessionFile { + const dir = try allocator.dupe(u8, session_dir); + errdefer allocator.free(dir); + + const id = try allocator.dupe(u8, session_id); + errdefer allocator.free(id); + + const timestamp = try isoTimestamp(allocator, io); + errdefer allocator.free(timestamp); + + const cwd_copy = try allocator.dupe(u8, cwd); + errdefer allocator.free(cwd_copy); + + const filename = try std.fmt.allocPrint(allocator, "{s}.jsonl", .{id}); + defer allocator.free(filename); + const file_path = try std.fs.path.join(allocator, &.{ dir, filename }); + errdefer allocator.free(file_path); + + return .{ + .allocator = allocator, + .io = io, + .session_dir = dir, + .session_file = file_path, + .header = .{ + .version = CURRENT_VERSION, + .id = id, + .timestamp = timestamp, + .cwd = cwd_copy, + }, + .entries = .empty, + .by_id = std.StringHashMap(usize).init(allocator), + .leaf_id = null, + .flushed = false, + .written_bytes = 0, + }; + } + + /// Open and replay an existing session file. Truncates from the first + /// corrupted line. Runs format migration if needed and rewrites the + /// file once. + pub fn open( + allocator: Allocator, + io: Io, + file_path: []const u8, + ) !SessionFile { + const path_copy = try allocator.dupe(u8, file_path); + errdefer allocator.free(path_copy); + + // The session dir is the file's parent directory. + const dir_path = std.fs.path.dirname(path_copy) orelse "."; + const dir = try allocator.dupe(u8, dir_path); + errdefer allocator.free(dir); + + const bytes = try readWholeFile(allocator, io, path_copy); + defer allocator.free(bytes); + + // Walk line-by-line. The first failure causes a truncation back to + // the start of that line. + var entries: std.ArrayList(SessionEntry) = .empty; + errdefer { + for (entries.items) |e| e.deinit(allocator); + entries.deinit(allocator); + } + var by_id = std.StringHashMap(usize).init(allocator); + errdefer by_id.deinit(); + + var header_opt: ?SessionHeader = null; + errdefer if (header_opt) |h| h.deinit(allocator); + + var cursor: usize = 0; + var valid_bytes: u64 = 0; // length of the file prefix that parses cleanly + var saw_corruption: bool = false; + + while (cursor < bytes.len) { + // Find the next newline (or EOF). + const rest = bytes[cursor..]; + const nl_rel = std.mem.indexOfScalar(u8, rest, '\n'); + const line_end_excl: usize = if (nl_rel) |n| cursor + n else bytes.len; + const line = bytes[cursor..line_end_excl]; + const next_cursor: usize = if (nl_rel != null) line_end_excl + 1 else bytes.len; + + // Allow blank lines silently (just whitespace), but a non-empty + // trimmed line that won't parse triggers truncation. + const trimmed = std.mem.trim(u8, line, " \t\r"); + if (trimmed.len == 0) { + if (nl_rel == null) break; + cursor = next_cursor; + valid_bytes = cursor; + continue; + } + + // If the final line has no trailing newline AND we hit EOF, it + // is presumed truncated mid-write. Treat as corruption. + if (nl_rel == null) { + saw_corruption = true; + break; + } + + const fe = session_mod.parseLine(allocator, line) catch { + saw_corruption = true; + break; + }; + + switch (fe) { + .header => |h| { + if (header_opt != null) { + // Two headers — treat as corruption from this line on. + h.deinit(allocator); + saw_corruption = true; + break; + } + if (entries.items.len != 0) { + // Header arrived after entries — malformed. + h.deinit(allocator); + saw_corruption = true; + break; + } + header_opt = h; + }, + .entry => |e| { + const idx = entries.items.len; + entries.append(allocator, e) catch |err| { + e.deinit(allocator); + return err; + }; + by_id.put(e.base().id, idx) catch |err| { + // Rolling back the append is awkward; in practice + // OOM here is fatal anyway. + return err; + }; + }, + } + + cursor = next_cursor; + valid_bytes = cursor; + } + + // No header at all — refuse to load. + const header = header_opt orelse return error.InvalidSessionFile; + + // Truncate the file if anything beyond `valid_bytes` is corrupt. + if (saw_corruption and valid_bytes < bytes.len) { + try truncateFileTo(io, path_copy, valid_bytes); + } + + var migrated_header = header; + var did_migrate = migrate(allocator, &migrated_header, &entries); + if (elideDanglingToolUses(allocator, &entries)) { + did_migrate = true; + } + // We didn't reassign by_id during migration; rebuild if needed. + if (did_migrate) { + by_id.clearRetainingCapacity(); + for (entries.items, 0..) |e, i| { + try by_id.put(e.base().id, i); + } + try rewriteFile(allocator, io, path_copy, migrated_header, entries.items); + } + + const leaf_id: ?[]const u8 = if (entries.items.len > 0) + entries.items[entries.items.len - 1].base().id + else + null; + + // Compute final file length on disk so future appends use the + // correct offset. + const stat = try statFileForLength(io, path_copy); + + return .{ + .allocator = allocator, + .io = io, + .session_dir = dir, + .session_file = path_copy, + .header = migrated_header, + .entries = entries, + .by_id = by_id, + .leaf_id = leaf_id, + .flushed = true, + .written_bytes = stat, + }; + } + + pub fn deinit(self: *SessionFile) void { + self.header.deinit(self.allocator); + for (self.entries.items) |e| e.deinit(self.allocator); + self.entries.deinit(self.allocator); + self.by_id.deinit(); + self.allocator.free(self.session_dir); + self.allocator.free(self.session_file); + } + + // ---------- Accessors ---------- + + pub fn getCwd(self: *const SessionFile) []const u8 { + return self.header.cwd; + } + + pub fn getSessionId(self: *const SessionFile) []const u8 { + return self.header.id; + } + + pub fn getSessionFile(self: *const SessionFile) []const u8 { + return self.session_file; + } + + pub fn getSessionDir(self: *const SessionFile) []const u8 { + return self.session_dir; + } + + pub fn getLeafId(self: *const SessionFile) ?[]const u8 { + return self.leaf_id; + } + + pub fn getEntry(self: *const SessionFile, id: []const u8) ?*const SessionEntry { + const idx = self.by_id.get(id) orelse return null; + return &self.entries.items[idx]; + } + + pub fn getEntries(self: *const SessionFile) []const SessionEntry { + return self.entries.items; + } + + pub fn isFlushed(self: *const SessionFile) bool { + return self.flushed; + } + + // ---------- Active model resolution ---------- + + /// Determine the active provider/model by walking entries leaf→root + /// and finding the last user-message entry with provider/model + /// stamped. Returns null only when no user message has been appended + /// yet, which is only reachable on a freshly-`init`'d session before + /// the first user prompt (and therefore before any disk flush). + /// + /// Returns a borrowed stamp owned by the manager; do not free. + pub fn activeStamp(self: *const SessionFile) ?session_mod.WireStamp { + var i = self.entries.items.len; + while (i > 0) : (i -= 1) { + const e = self.entries.items[i - 1]; + switch (e) { + .message => |m| { + if (m.stamp) |st| return st; + }, + } + } + return null; + } + + // ---------- Appending ---------- + + /// Append a message entry. `msg` is consumed (ownership transferred) + /// regardless of success — on error, the message is deinit'd before + /// the error is returned. + /// + /// If `flushed`: writes the new line immediately. + /// If not flushed and `msg.role == .assistant`: writes the header + + /// all buffered entries + the new entry, then sets `flushed`. + /// Otherwise: buffers in memory only. + pub fn appendMessage( + self: *SessionFile, + msg: StoredMessage, + // Wire-format provider identity for the entry. Null on system + // messages. Borrowed; duplicated here into the entry. + stamp: ?session_mod.WireStamp, + ) ![]const u8 { + // Build the entry up-front, taking ownership of the inputs. + var msg_local = msg; + errdefer msg_local.deinit(self.allocator); + + const id_buf = try self.newEntryId(); + errdefer self.allocator.free(id_buf); + + const timestamp = try isoTimestamp(self.allocator, self.io); + errdefer self.allocator.free(timestamp); + + const parent_id_copy: ?[]const u8 = if (self.leaf_id) |l| try self.allocator.dupe(u8, l) else null; + errdefer if (parent_id_copy) |p| self.allocator.free(p); + + const stamp_copy: ?session_mod.WireStamp = if (stamp) |st| try st.dupe(self.allocator) else null; + errdefer if (stamp_copy) |st| st.deinit(self.allocator); + + const entry: SessionEntry = .{ .message = .{ + .base = .{ .id = id_buf, .parent_id = parent_id_copy, .timestamp = timestamp }, + .stamp = stamp_copy, + .message = msg_local, + } }; + + // The entry now owns msg_local + id_buf + timestamp + parent_id_copy + + // provider/model copies. Cancel the errdefers individually. + // (Zig's errdefer behavior: they only run on error returns; pushing the + // entry into entries.items before any further fallible step means an + // error in by_id.put() would double-free. Instead, do the put first + // against a not-yet-stored id.) + + const idx = self.entries.items.len; + + // Ensure capacity before touching anything. + try self.entries.ensureUnusedCapacity(self.allocator, 1); + try self.by_id.ensureUnusedCapacity(1); + + // Persist BEFORE inserting into the in-memory structures, so that on + // I/O failure we don't have a dangling in-memory entry the caller + // thinks was saved. (Failure leaves the file unchanged for an + // unflushed session, and unchanged-except-for-EOF for a flushed one.) + const is_assistant = entry.message.message.role == .assistant; + if (self.flushed) { + try self.persistEntry(entry); + } else if (is_assistant) { + try self.flushBuffered(entry); + } + // If not flushed and not assistant: nothing to do; the entry will be + // flushed alongside the eventual first assistant entry. + + // Now insert into in-memory structures. All allocations are kept. + self.entries.appendAssumeCapacity(entry); + self.by_id.putAssumeCapacity(entry.base().id, idx); + self.leaf_id = entry.base().id; + return entry.base().id; + } + + /// Returns a freshly allocated 8-character hex id, guaranteed not to + /// collide with any existing entry id in this session. + fn newEntryId(self: *SessionFile) ![]u8 { + const max_tries = 100; + var i: usize = 0; + while (i < max_tries) : (i += 1) { + const buf = try self.allocator.alloc(u8, 8); + errdefer self.allocator.free(buf); + newEntryIdInto(buf[0..8], self.io); + if (!self.by_id.contains(buf)) { + return buf; + } + self.allocator.free(buf); + } + // Fall back to a UUID prefix if 100 retries all collided. With 4 + // random bytes per id and a session with <<2^16 entries, the + // probability of getting here is effectively zero, but we want a + // hard guarantee. + const long = try newUuidV7(self.allocator, self.io); + defer self.allocator.free(long); + const buf = try self.allocator.alloc(u8, 8); + @memcpy(buf, long[0..8]); + return buf; + } + + // ---------- Persistence ---------- + + /// Write the header + all currently-buffered entries + `new_entries` + /// to the file as a single batch. Creates the directory and file. + fn flushBufferedMany(self: *SessionFile, new_entries: []const SessionEntry) !void { + try mkdirP(self.io, self.session_dir); + + const file = try Io.Dir.cwd().createFile(self.io, self.session_file, .{ + .truncate = true, + .read = false, + }); + defer file.close(self.io); + + var offset: u64 = 0; + const header_line = try session_mod.serializeHeader(self.allocator, self.header); + defer self.allocator.free(header_line); + try file.writePositionalAll(self.io, header_line, offset); + offset += header_line.len; + try file.writePositionalAll(self.io, "\n", offset); + offset += 1; + + for (self.entries.items) |e| { + const line = try session_mod.serializeEntry(self.allocator, e); + defer self.allocator.free(line); + try file.writePositionalAll(self.io, line, offset); + offset += line.len; + try file.writePositionalAll(self.io, "\n", offset); + offset += 1; + } + + for (new_entries) |entry| { + const line = try session_mod.serializeEntry(self.allocator, entry); + defer self.allocator.free(line); + try file.writePositionalAll(self.io, line, offset); + offset += line.len; + try file.writePositionalAll(self.io, "\n", offset); + offset += 1; + } + + file.sync(self.io) catch {}; + self.flushed = true; + self.written_bytes = offset; + } + + fn flushBuffered(self: *SessionFile, final_entry: SessionEntry) !void { + try self.flushBufferedMany(&.{final_entry}); + } + + /// Append a single line for `entry` to the open session file. Caller + /// must have already verified `flushed`. + fn persistEntry(self: *SessionFile, entry: SessionEntry) !void { + try self.persistEntries(&.{entry}); + } + + pub fn appendMessagesAtomic( + self: *SessionFile, + messages: []StoredMessage, + stamps: []const ?session_mod.WireStamp, + ) !void { + std.debug.assert(messages.len == stamps.len); + if (messages.len == 0) return; + + const base_len = self.entries.items.len; + try self.entries.ensureUnusedCapacity(self.allocator, messages.len); + try self.by_id.ensureUnusedCapacity(@intCast(messages.len)); + + var entries = try self.allocator.alloc(SessionEntry, messages.len); + defer self.allocator.free(entries); + var built: usize = 0; + errdefer { + for (entries[0..built]) |*e| e.deinit(self.allocator); + } + + var prev_leaf = self.leaf_id; + for (messages, 0..) |msg, i| { + const msg_local = msg; + const id_buf = try self.newEntryId(); + errdefer self.allocator.free(id_buf); + const timestamp = try isoTimestamp(self.allocator, self.io); + errdefer self.allocator.free(timestamp); + const parent_id_copy: ?[]const u8 = if (prev_leaf) |l| try self.allocator.dupe(u8, l) else null; + errdefer if (parent_id_copy) |p| self.allocator.free(p); + const stamp_copy: ?session_mod.WireStamp = if (stamps[i]) |st| try st.dupe(self.allocator) else null; + errdefer if (stamp_copy) |st| st.deinit(self.allocator); + entries[i] = .{ .message = .{ + .base = .{ .id = id_buf, .parent_id = parent_id_copy, .timestamp = timestamp }, + .stamp = stamp_copy, + .message = msg_local, + } }; + built += 1; + prev_leaf = entries[i].base().id; + } + + if (self.flushed) { + try self.persistEntries(entries); + } else { + try self.flushBufferedMany(entries); + } + + for (entries, 0..) |entry, i| { + self.entries.appendAssumeCapacity(entry); + self.by_id.putAssumeCapacity(entry.base().id, base_len + i); + } + self.leaf_id = entries[entries.len - 1].base().id; + built = 0; + } + + fn persistEntries(self: *SessionFile, entries: []const SessionEntry) !void { + const file = try Io.Dir.cwd().openFile(self.io, self.session_file, .{ + .mode = .write_only, + }); + defer file.close(self.io); + + var offset = self.written_bytes; + for (entries) |entry| { + const line = try session_mod.serializeEntry(self.allocator, entry); + defer self.allocator.free(line); + try file.writePositionalAll(self.io, line, offset); + offset += line.len; + try file.writePositionalAll(self.io, "\n", offset); + offset += 1; + } + file.sync(self.io) catch {}; + self.written_bytes = offset; + } + + // ============================================================================= + // Conversation rebuild + // ============================================================================= + + /// Build a fresh `Conversation` from the entry log. Caller owns the + /// returned conversation (call `deinit`). + pub fn rebuildConversation(self: *const SessionFile) !conversation_mod.Conversation { + var conv = conversation_mod.Conversation.init(self.allocator); + errdefer conv.deinit(); + + for (self.entries.items) |entry| { + switch (entry) { + .message => |me| try appendMessageToConv(&conv, self.allocator, me.message), + } + } + return conv; + } + +}; + +/// Best-effort extraction of plain prompt text from a user `StoredMessage`. +/// Used to populate `SessionInfo.last_user_message`. Returns null if the +/// message carries no plain text block. Caller owns the returned slice. +fn extractUserText(alloc: Allocator, msg: StoredMessage) !?[]u8 { + for (msg.content) |block| { + if (block == .text) { + return try alloc.dupe(u8, block.text.text); + } + } + return null; +} + +fn appendMessageToConv( + conv: *conversation_mod.Conversation, + allocator: Allocator, + disk_msg: StoredMessage, +) !void { + var content: std.ArrayList(conversation_mod.ContentBlock) = .empty; + errdefer { + for (content.items) |*b| { + var mut = b.*; + mut.deinit(allocator); + } + content.deinit(allocator); + } + try content.ensureTotalCapacity(allocator, disk_msg.content.len); + const sys_mode: conversation_mod.SystemMode = switch (disk_msg.mode) { + .append => .append, + .replace => .replace, + }; + for (disk_msg.content) |db| { + var block = try session_mod.diskContentBlockToInternal(allocator, db); + // System-role text blocks become `.System` blocks carrying the + // message's recorded mode, so the append/replace derivation works + // on the rebuilt conversation exactly as it did when written. + if (disk_msg.role == .system and block == .Text) { + const tb = block.Text; + block = .{ .System = .{ .text = tb, .mode = sys_mode } }; + } + content.appendAssumeCapacity(block); + } + const role: conversation_mod.MessageRole = switch (disk_msg.role) { + .system => .system, + .user => .user, + .assistant => .assistant, + }; + // 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, + }); +} + +// ============================================================================= +// Migration +// ============================================================================= + +/// Future format migrations land here. Returns true if anything changed +/// (which triggers a one-time file rewrite). +fn migrate( + allocator: Allocator, + header: *SessionHeader, + entries: *std.ArrayList(SessionEntry), +) bool { + _ = allocator; + _ = entries; + if (header.version >= CURRENT_VERSION) return false; + // No earlier versions exist yet. When v2 lands, transform v1 entries + // here and bump `header.version`. + return false; +} + +fn elideDanglingToolUses(allocator: Allocator, entries: *std.ArrayList(SessionEntry)) bool { + var needed: std.StringHashMap(void) = .init(allocator); + defer needed.deinit(); + var changed = false; + + var i = entries.items.len; + while (i > 0) { + i -= 1; + const entry = &entries.items[i]; + if (entry.* != .message) continue; + const msg = &entry.message.message; + + if (msg.role == .user) { + for (msg.content) |block| { + if (block == .tool_result) { + needed.put(block.tool_result.tool_use_id, {}) catch {}; + } + } + continue; + } + + if (msg.role != .assistant) continue; + var kept: std.ArrayList(StoredContentBlock) = .empty; + defer kept.deinit(allocator); + var removed = false; + for (msg.content) |block| { + if (block == .tool_use and !needed.contains(block.tool_use.id)) { + block.deinit(allocator); + removed = true; + continue; + } + kept.append(allocator, block) catch unreachable; + } + if (!removed) continue; + allocator.free(msg.content); + msg.content = kept.toOwnedSlice(allocator) catch unreachable; + changed = true; + } + return changed; +} + +// ============================================================================= +// File utilities +// ============================================================================= + +fn readWholeFile(allocator: Allocator, io: Io, path: []const u8) ![]u8 { + const file = Io.Dir.cwd().openFile(io, path, .{ .mode = .read_only }) catch |err| switch (err) { + error.FileNotFound => return error.InvalidSessionFile, + else => return err, + }; + defer file.close(io); + + const len = file.length(io) catch { + // Fall back to a streaming read of a reasonable upper bound. + // Sessions over ~10 MB are out of scope for phase 4. + var list: std.ArrayList(u8) = .empty; + defer list.deinit(allocator); + var chunk: [4096]u8 = undefined; + while (true) { + const n = file.readStreaming(io, &.{&chunk}) catch break; + if (n == 0) break; + try list.appendSlice(allocator, chunk[0..n]); + } + return try list.toOwnedSlice(allocator); + }; + + const buf = try allocator.alloc(u8, @intCast(len)); + errdefer allocator.free(buf); + _ = try file.readPositionalAll(io, buf, 0); + return buf; +} + +fn statFileForLength(io: Io, path: []const u8) !u64 { + const file = try Io.Dir.cwd().openFile(io, path, .{ .mode = .read_only }); + defer file.close(io); + return try file.length(io); +} + +fn truncateFileTo(io: Io, path: []const u8, new_length: u64) !void { + const file = try Io.Dir.cwd().openFile(io, path, .{ .mode = .write_only }); + defer file.close(io); + try file.setLength(io, new_length); + file.sync(io) catch {}; +} + +/// Write a fresh file containing `header` followed by `entries`. Truncates +/// any existing content. Used after a migration rewrites the format. +fn rewriteFile( + allocator: Allocator, + io: Io, + path: []const u8, + header: SessionHeader, + entries: []const SessionEntry, +) !void { + const file = try Io.Dir.cwd().createFile(io, path, .{ + .truncate = true, + .read = false, + }); + defer file.close(io); + + var offset: u64 = 0; + const header_line = try session_mod.serializeHeader(allocator, header); + defer allocator.free(header_line); + try file.writePositionalAll(io, header_line, offset); + offset += header_line.len; + try file.writePositionalAll(io, "\n", offset); + offset += 1; + for (entries) |e| { + const line = try session_mod.serializeEntry(allocator, e); + defer allocator.free(line); + try file.writePositionalAll(io, line, offset); + offset += line.len; + try file.writePositionalAll(io, "\n", offset); + offset += 1; + } + file.sync(io) catch {}; +} + +fn mkdirP(io: Io, path: []const u8) !void { + Io.Dir.cwd().createDirPath(io, path) catch |err| switch (err) { + error.PathAlreadyExists => {}, + else => return err, + }; +} + +// ============================================================================= +// Listing +// ============================================================================= + +/// List sessions in `session_dir`. Returns a slice of `SessionInfo`s +/// sorted by `modified` descending (most recent first). Caller owns the +/// slice and each `SessionInfo`. +/// +/// If the directory does not exist, returns an empty slice (no error). +/// +/// If `on_progress` is non-null, it is invoked after each file is parsed. +pub fn listSessions( + allocator: Allocator, + io: Io, + session_dir: []const u8, + on_progress: ?*const fn (loaded: usize, total: usize) void, +) ![]FileInfo { + var dir = Io.Dir.cwd().openDir(io, session_dir, .{}) catch |err| switch (err) { + error.FileNotFound => return try allocator.alloc(FileInfo, 0), + else => return err, + }; + defer dir.close(io); + + var names: std.ArrayList([]u8) = .empty; + defer { + for (names.items) |n| allocator.free(n); + names.deinit(allocator); + } + + var it = dir.iterate(); + while (try it.next(io)) |entry| { + if (entry.kind != .file) continue; + if (!std.mem.endsWith(u8, entry.name, ".jsonl")) continue; + const copy = try allocator.dupe(u8, entry.name); + errdefer allocator.free(copy); + try names.append(allocator, copy); + } + + var infos: std.ArrayList(FileInfo) = .empty; + errdefer { + for (infos.items) |i| i.deinit(allocator); + infos.deinit(allocator); + } + try infos.ensureTotalCapacity(allocator, names.items.len); + + var loaded: usize = 0; + for (names.items) |name| { + const full = try std.fs.path.join(allocator, &.{ session_dir, name }); + defer allocator.free(full); + const info_opt = buildFileInfo(allocator, io, full) catch null; + if (info_opt) |info| { + infos.appendAssumeCapacity(info); + } + loaded += 1; + if (on_progress) |cb| cb(loaded, names.items.len); + } + + const slice = try infos.toOwnedSlice(allocator); + std.sort.pdq(FileInfo, slice, {}, fileInfoNewerFirst); + return slice; +} + +fn fileInfoNewerFirst(_: void, a: FileInfo, b: FileInfo) bool { + return std.mem.order(u8, a.modified, b.modified) == .gt; +} + +fn buildFileInfo( + allocator: Allocator, + io: Io, + file_path: []const u8, +) !?FileInfo { + const bytes = readWholeFile(allocator, io, file_path) catch return null; + defer allocator.free(bytes); + + var header_opt: ?SessionHeader = null; + defer if (header_opt) |h| h.deinit(allocator); + + var message_count: usize = 0; + var last_activity: ?[]u8 = null; + defer if (last_activity) |la| allocator.free(la); + var last_user: ?[]u8 = null; + defer if (last_user) |lu| allocator.free(lu); + var last_stamp: ?session_mod.WireStamp = null; + defer if (last_stamp) |st| st.deinit(allocator); + + var lines = std.mem.splitScalar(u8, bytes, '\n'); + while (lines.next()) |line| { + const trimmed = std.mem.trim(u8, line, " \t\r"); + if (trimmed.len == 0) continue; + const fe = session_mod.parseLine(allocator, trimmed) catch break; + switch (fe) { + .header => |h| { + if (header_opt != null) { + h.deinit(allocator); + } else { + header_opt = h; + } + }, + .entry => |e| { + defer e.deinit(allocator); + switch (e) { + .message => |m| { + if (m.message.role == .user or m.message.role == .assistant) { + message_count += 1; + if (last_activity) |la| allocator.free(la); + last_activity = try allocator.dupe(u8, m.base.timestamp); + } + if (m.stamp) |st| { + if (last_stamp) |old| old.deinit(allocator); + last_stamp = try st.dupe(allocator); + } + if (m.message.role == .user) { + if (try extractUserText(allocator, m.message)) |ut| { + if (last_user) |lu| allocator.free(lu); + last_user = ut; + } + } + }, + } + }, + } + } + + const header = header_opt orelse return null; + + const path = try allocator.dupe(u8, file_path); + errdefer allocator.free(path); + const id = try allocator.dupe(u8, header.id); + errdefer allocator.free(id); + const created = try allocator.dupe(u8, header.timestamp); + errdefer allocator.free(created); + const modified = if (last_activity) |la| blk: { + last_activity = null; + break :blk la; + } else try allocator.dupe(u8, header.timestamp); + errdefer allocator.free(modified); + const last_user_message = if (last_user) |lu| blk: { + last_user = null; + break :blk lu; + } else try allocator.dupe(u8, ""); + errdefer allocator.free(last_user_message); + const stamp_out = if (last_stamp) |st| blk: { + last_stamp = null; + break :blk st; + } else null; + + return .{ + .path = path, + .id = id, + .created = created, + .modified = modified, + .message_count = message_count, + .last_user_message = last_user_message, + .stamp = stamp_out, + }; +} + +// ============================================================================= +// Recent / resume helpers +// ============================================================================= + +/// Find the most recent session file in `session_dir`. Returns null if +/// none exist. Caller owns the returned path. +pub fn findMostRecentSession(allocator: Allocator, io: Io, session_dir: []const u8) !?[]u8 { + var dir = Io.Dir.cwd().openDir(io, session_dir, .{}) catch |err| switch (err) { + error.FileNotFound => return null, + else => return err, + }; + defer dir.close(io); + + var best_name: ?[]u8 = null; + errdefer if (best_name) |b| allocator.free(b); + + var it = dir.iterate(); + while (try it.next(io)) |entry| { + if (entry.kind != .file) continue; + if (!std.mem.endsWith(u8, entry.name, ".jsonl")) continue; + if (best_name) |b| { + // Lexicographic compare. UUIDv7 filenames sort chronologically. + if (std.mem.order(u8, entry.name, b) == .gt) { + allocator.free(b); + best_name = try allocator.dupe(u8, entry.name); + } + } else { + best_name = try allocator.dupe(u8, entry.name); + } + } + + const name = best_name orelse return null; + defer allocator.free(name); + best_name = null; + return try std.fs.path.join(allocator, &.{ session_dir, name }); +} + +/// Resolve a (possibly abbreviated) session id to a session file path +/// within `session_dir`. Errors if no match or ambiguous prefix. +pub fn resolveSessionId( + allocator: Allocator, + io: Io, + session_dir: []const u8, + id_or_prefix: []const u8, +) ![]u8 { + var dir = Io.Dir.cwd().openDir(io, session_dir, .{}) catch |err| switch (err) { + error.FileNotFound => return error.SessionNotFound, + else => return err, + }; + defer dir.close(io); + + var match: ?[]u8 = null; + errdefer if (match) |m| allocator.free(m); + + var it = dir.iterate(); + while (try it.next(io)) |entry| { + if (entry.kind != .file) continue; + if (!std.mem.endsWith(u8, entry.name, ".jsonl")) continue; + // Strip `.jsonl` for the prefix match. + const stem = entry.name[0 .. entry.name.len - ".jsonl".len]; + if (!std.mem.startsWith(u8, stem, id_or_prefix)) continue; + if (match != null) return error.AmbiguousSessionId; + match = try allocator.dupe(u8, entry.name); + } + + const name = match orelse return error.SessionNotFound; + defer allocator.free(name); + match = null; + return try std.fs.path.join(allocator, &.{ session_dir, name }); +} + +// ============================================================================= +// FileSystemJSONLStore — the directory-backed catalog (SessionStore impl) +// ============================================================================= + +/// A directory-backed `SessionStore`: each session is one `.jsonl` file +/// under `dir`. The catalog mints `Session` handles, lists/resolves files, +/// loads conversations, and routes appends to the right `SessionFile`. +/// +/// Open `SessionFile`s are cached by id for the catalog's lifetime so the +/// buffered-until-first-assistant write discipline survives across the +/// separate user-prompt and assistant-turn appends of a single turn. +/// +/// `dir` is the already-resolved sessions directory (the panto CLI derives +/// the per-cwd grouping; the store itself is cwd-agnostic). `cwd` is stamped +/// into new session headers for display/provenance only. +pub const FileSystemJSONLStore = struct { + allocator: Allocator, + io: Io, + dir: []u8, // owned: the sessions directory + cwd: []u8, // owned: recorded in new session headers + open: std.StringHashMap(*SessionFile), + + pub fn init(allocator: Allocator, io: Io, dir: []const u8, cwd: []const u8) !FileSystemJSONLStore { + const dir_copy = try allocator.dupe(u8, dir); + errdefer allocator.free(dir_copy); + const cwd_copy = try allocator.dupe(u8, cwd); + return .{ + .allocator = allocator, + .io = io, + .dir = dir_copy, + .cwd = cwd_copy, + .open = std.StringHashMap(*SessionFile).init(allocator), + }; + } + + pub fn deinit(self: *FileSystemJSONLStore) void { + var it = self.open.iterator(); + while (it.next()) |e| { + e.value_ptr.*.deinit(); + self.allocator.destroy(e.value_ptr.*); + self.allocator.free(e.key_ptr.*); + } + self.open.deinit(); + self.allocator.free(self.dir); + self.allocator.free(self.cwd); + } + + /// Borrow (opening if needed) the `SessionFile` for `id`, caching it. + /// Returns null if the file does not exist on disk and `create_missing` + /// is false. + fn fileFor(self: *FileSystemJSONLStore, id: []const u8, create_missing: bool) !?*SessionFile { + if (self.open.get(id)) |sf| return sf; + // Locate the file by exact id. + const path = try std.fs.path.join(self.allocator, &.{ self.dir, id }); + defer self.allocator.free(path); + const full = try std.fmt.allocPrint(self.allocator, "{s}.jsonl", .{path}); + defer self.allocator.free(full); + + const exists = blk: { + Io.Dir.cwd().access(self.io, full, .{}) catch break :blk false; + break :blk true; + }; + if (!exists and !create_missing) return null; + + const sf = try self.allocator.create(SessionFile); + errdefer self.allocator.destroy(sf); + sf.* = if (exists) + try SessionFile.open(self.allocator, self.io, full) + else + try SessionFile.initWithId(self.allocator, self.io, self.dir, self.cwd, id); + errdefer sf.deinit(); + + const key = try self.allocator.dupe(u8, id); + errdefer self.allocator.free(key); + try self.open.put(key, sf); + return sf; + } + + fn infoFromFileInfo(self: *FileSystemJSONLStore, fi: FileInfo) !session_store_mod.SessionInfo { + const id = try self.allocator.dupe(u8, fi.id); + errdefer self.allocator.free(id); + const created = try self.allocator.dupe(u8, fi.created); + errdefer self.allocator.free(created); + const modified = try self.allocator.dupe(u8, fi.modified); + errdefer self.allocator.free(modified); + const last_user = try self.allocator.dupe(u8, fi.last_user_message); + errdefer self.allocator.free(last_user); + const base_url = try self.allocator.dupe(u8, if (fi.stamp) |s| s.base_url else ""); + errdefer self.allocator.free(base_url); + const model = try self.allocator.dupe(u8, if (fi.stamp) |s| s.model else ""); + return .{ + .id = id, + .created = created, + .modified = modified, + .message_count = fi.message_count, + .last_user_message = last_user, + .api_style = if (fi.stamp) |s| s.api_style else .openai_chat, + .base_url = base_url, + .model = model, + .reasoning = if (fi.stamp) |s| s.reasoning else .default, + }; + } + + // ---------- vtable ---------- + + fn createVT(ctx: *anyopaque) session_store_mod.Session { + const self: *FileSystemJSONLStore = @ptrCast(@alignCast(ctx)); + // Mint a fresh id; nothing hits disk until the first append. + const id = newUuidV7(self.allocator, self.io) catch ""; + // The SessionFile is created lazily on first append via fileFor. + const info: session_store_mod.SessionInfo = .{ + .id = id, + .created = self.allocator.dupe(u8, "") catch "", + .modified = self.allocator.dupe(u8, "") catch "", + .message_count = 0, + .last_user_message = self.allocator.dupe(u8, "") catch "", + .api_style = .openai_chat, + .base_url = self.allocator.dupe(u8, "") catch "", + .model = self.allocator.dupe(u8, "") catch "", + .reasoning = .default, + }; + return .{ .info = info, .store = self.store() }; + } + + fn listVT(ctx: *anyopaque) anyerror![]session_store_mod.SessionInfo { + const self: *FileSystemJSONLStore = @ptrCast(@alignCast(ctx)); + const fis = try listSessions(self.allocator, self.io, self.dir, null); + defer { + for (fis) |fi| fi.deinit(self.allocator); + self.allocator.free(fis); + } + var out = try self.allocator.alloc(session_store_mod.SessionInfo, fis.len); + var built: usize = 0; + errdefer { + for (out[0..built]) |i| i.deinit(self.allocator); + self.allocator.free(out); + } + for (fis, 0..) |fi, i| { + out[i] = try self.infoFromFileInfo(fi); + built += 1; + } + return out; + } + + fn freeSessionInfosVT(ctx: *anyopaque, infos: []session_store_mod.SessionInfo) void { + const self: *FileSystemJSONLStore = @ptrCast(@alignCast(ctx)); + for (infos) |i| i.deinit(self.allocator); + self.allocator.free(infos); + } + + fn resolveVT(ctx: *anyopaque, id_or_prefix: []const u8) anyerror!?session_store_mod.Session { + const self: *FileSystemJSONLStore = @ptrCast(@alignCast(ctx)); + const path = resolveSessionId(self.allocator, self.io, self.dir, id_or_prefix) catch |err| switch (err) { + error.SessionNotFound => return null, + else => return err, + }; + defer self.allocator.free(path); + return try self.sessionFromPath(path); + } + + fn latestVT(ctx: *anyopaque) anyerror!?session_store_mod.Session { + const self: *FileSystemJSONLStore = @ptrCast(@alignCast(ctx)); + const path = (try findMostRecentSession(self.allocator, self.io, self.dir)) orelse return null; + defer self.allocator.free(path); + return try self.sessionFromPath(path); + } + + fn sessionFromPath(self: *FileSystemJSONLStore, path: []const u8) !?session_store_mod.Session { + const fi = (try buildFileInfo(self.allocator, self.io, path)) orelse return null; + defer fi.deinit(self.allocator); + const info = try self.infoFromFileInfo(fi); + return .{ .info = info, .store = self.store() }; + } + + fn loadVT(ctx: *anyopaque, id: []const u8) anyerror!?conversation_mod.Conversation { + const self: *FileSystemJSONLStore = @ptrCast(@alignCast(ctx)); + const sf = (try self.fileFor(id, false)) orelse return null; + return try sf.rebuildConversation(); + } + + fn appendMessagesVT( + ctx: *anyopaque, + session_id: []const u8, + messages: []session_store_mod.PersistentMessage, + ) anyerror!void { + const self: *FileSystemJSONLStore = @ptrCast(@alignCast(ctx)); + if (messages.len == 0) return; + const sf = (try self.fileFor(session_id, true)).?; + + // Convert each rich PersistentMessage to a StoredMessage + wire stamp. + // The FS store deliberately ignores the `conversation` and + // `tools_available` provenance fields. + var stored = try self.allocator.alloc(StoredMessage, messages.len); + var stamps = try self.allocator.alloc(?session_mod.WireStamp, messages.len); + defer self.allocator.free(stored); + defer self.allocator.free(stamps); + var built: usize = 0; + errdefer for (stored[0..built]) |sm| sm.deinit(self.allocator); + + for (messages, 0..) |pm, i| { + stored[i] = try persistentToStored(self.allocator, pm); + // System entries carry no wire stamp; user/assistant do. + stamps[i] = if (pm.message.role == .system) null else .{ + .api_style = pm.identity.api_style, + .base_url = pm.identity.base_url, + .model = pm.identity.model, + .reasoning = pm.identity.reasoning, + }; + built += 1; + } + + // appendMessagesAtomic consumes the StoredMessages; it dupes stamps. + try sf.appendMessagesAtomic(stored, stamps); + } + + const store_vtable: session_store_mod.SessionStore.VTable = .{ + .create = createVT, + .list = listVT, + .freeSessionInfos = freeSessionInfosVT, + .resolve = resolveVT, + .latest = latestVT, + .load = loadVT, + .appendMessages = appendMessagesVT, + }; + + /// Wrap this catalog as a neutral `SessionStore`. The handle borrows + /// `self`; `self` must outlive it. + pub fn store(self: *FileSystemJSONLStore) session_store_mod.SessionStore { + return .{ .ptr = self, .vtable = &store_vtable }; + } +}; + +/// Convert a rich in-memory `PersistentMessage` to the on-disk +/// `StoredMessage`. Strings are duplicated; the source is untouched. +fn persistentToStored( + alloc: Allocator, + pm: session_store_mod.PersistentMessage, +) !StoredMessage { + const msg = pm.message; + const blocks = try alloc.alloc(session_mod.StoredContentBlock, 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_mod.contentBlockToDisk(alloc, block); + allocated += 1; + } + const mode: session_mod.StoredSystemMode = blk: { + for (msg.content.items) |block| { + if (block == .System and block.System.mode == .replace) break :blk .replace; + } + break :blk .append; + }; + const stop_reason: ?[]const u8 = if (msg.role == .assistant) try alloc.dupe(u8, "stop") else null; + errdefer if (stop_reason) |s| alloc.free(s); + const metadata: ?[]const u8 = if (msg.metadata) |m| try alloc.dupe(u8, m) else null; + const role: session_mod.StoredMessageRole = switch (msg.role) { + .system => .system, + .user => .user, + .assistant => .assistant, + }; + return .{ + .role = role, + .content = blocks, + .mode = mode, + .stop_reason = stop_reason, + .usage = pm.usage, + .metadata = metadata, + }; +} + +// ============================================================================= +// Tests +// ============================================================================= + +const testing = std.testing; + +/// Borrowed wire stamps for tests (no allocation; the manager dupes them). +fn oaStamp() session_mod.WireStamp { + return .{ .api_style = .openai_chat, .base_url = "https://api.openai.com/v1", .model = "gpt-4o" }; +} +fn anStamp() session_mod.WireStamp { + return .{ .api_style = .anthropic_messages, .base_url = "https://api.anthropic.com", .model = "claude-sonnet-4-20250514" }; +} + +test "newUuidV7: produces 36-char hyphenated string with version 7" { + const io = testing.io; + const id = try newUuidV7(testing.allocator, io); + defer testing.allocator.free(id); + try testing.expectEqual(@as(usize, 36), id.len); + // Position 14 is the version nibble — should be '7'. + try testing.expectEqual(@as(u8, '7'), id[14]); + // Hyphens at canonical positions. + try testing.expectEqual(@as(u8, '-'), id[8]); + try testing.expectEqual(@as(u8, '-'), id[13]); + try testing.expectEqual(@as(u8, '-'), id[18]); + try testing.expectEqual(@as(u8, '-'), id[23]); +} + +test "isoTimestamp: well-formed ISO 8601 with millisecond precision" { + const ts = try isoTimestamp(testing.allocator, testing.io); + defer testing.allocator.free(ts); + try testing.expectEqual(@as(usize, 24), ts.len); + try testing.expectEqual(@as(u8, '-'), ts[4]); + try testing.expectEqual(@as(u8, 'T'), ts[10]); + try testing.expectEqual(@as(u8, '.'), ts[19]); + try testing.expectEqual(@as(u8, 'Z'), ts[23]); +} + +// ---- In-memory + filesystem tests (use a tmp dir) ---- + +const TmpSessionDir = struct { + parent: std.testing.TmpDir, + abs_path: []u8, + + fn init(allocator: Allocator) !TmpSessionDir { + var parent = std.testing.tmpDir(.{}); + errdefer parent.cleanup(); + var path_buf: [std.fs.max_path_bytes]u8 = undefined; + const n = try parent.dir.realPath(testing.io, &path_buf); + const abs = try allocator.dupe(u8, path_buf[0..n]); + return .{ .parent = parent, .abs_path = abs }; + } + + fn deinit(self: *TmpSessionDir, allocator: Allocator) void { + allocator.free(self.abs_path); + self.parent.cleanup(); + } +}; + +test "SessionFile.init: does not create file yet" { + const io = testing.io; + + var td = try TmpSessionDir.init(testing.allocator); + defer td.deinit(testing.allocator); + + // Use a non-existent subdirectory inside the tmp dir to also exercise + // lazy directory creation. + const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); + defer testing.allocator.free(sessions); + + var mgr = try SessionFile.init( + testing.allocator, + io, + sessions, + "/some/cwd", + ); + defer mgr.deinit(); + + try testing.expect(!mgr.isFlushed()); + + // The directory should not exist yet. + const stat_err = Io.Dir.cwd().openDir(io, sessions, .{}); + try testing.expectError(error.FileNotFound, stat_err); +} + +test "SessionFile: full flow — buffer, flush on assistant, append, resume" { + 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 SessionFile.init( + testing.allocator, + io, + sessions, + "/proj/foo", + ); + defer mgr.deinit(); + + // System message — non-assistant, should NOT trigger flush. + const sys_blocks = try testing.allocator.alloc(StoredContentBlock, 1); + sys_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "You are helpful.") } }; + _ = try mgr.appendMessage( + .{ .role = .system, .content = sys_blocks }, + null, + ); + try testing.expect(!mgr.isFlushed()); + + // User message — also doesn't flush. + const usr_blocks = try testing.allocator.alloc(StoredContentBlock, 1); + usr_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hi there") } }; + _ = try mgr.appendMessage( + .{ .role = .user, .content = usr_blocks }, + oaStamp(), + ); + try testing.expect(!mgr.isFlushed()); + + // Assistant message — triggers flush. + const a_blocks = try testing.allocator.alloc(StoredContentBlock, 1); + a_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hello") } }; + _ = try mgr.appendMessage( + .{ + .role = .assistant, + .content = a_blocks, + .stop_reason = try testing.allocator.dupe(u8, "stop"), + }, + oaStamp(), + ); + try testing.expect(mgr.isFlushed()); + + // Append another user/assistant round. + const u_two = try testing.allocator.alloc(StoredContentBlock, 1); + u_two[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "what's 2+2?") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = u_two }, oaStamp()); + + const a2 = try testing.allocator.alloc(StoredContentBlock, 1); + a2[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "4") } }; + _ = try mgr.appendMessage( + .{ .role = .assistant, .content = a2, .stop_reason = try testing.allocator.dupe(u8, "stop") }, + oaStamp(), + ); + + try testing.expectEqual(@as(usize, 5), mgr.getEntries().len); + + break :blk try testing.allocator.dupe(u8, mgr.getSessionFile()); + }; + defer testing.allocator.free(session_file); + + // Verify the file exists and is well-formed. + { + const bytes = try readWholeFile(testing.allocator, io, session_file); + defer testing.allocator.free(bytes); + // 1 header + 5 entries + trailing \n on each = 6 newlines. + var nl_count: usize = 0; + for (bytes) |b| if (b == '\n') { + nl_count += 1; + }; + try testing.expectEqual(@as(usize, 6), nl_count); + } + + // Resume. + var resumed = try SessionFile.open(testing.allocator, io, session_file); + defer resumed.deinit(); + try testing.expect(resumed.isFlushed()); + try testing.expectEqual(@as(usize, 5), resumed.getEntries().len); + try testing.expectEqualStrings("/proj/foo", resumed.getCwd()); + + // Continue the conversation. + const u_three = try testing.allocator.alloc(StoredContentBlock, 1); + u_three[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "thanks") } }; + _ = try resumed.appendMessage(.{ .role = .user, .content = u_three }, oaStamp()); + try testing.expectEqual(@as(usize, 6), resumed.getEntries().len); +} + +test "SessionFile: assistant message tags the message metadata and the entry leaf id is the assistant entry" { + 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); + + var mgr = try SessionFile.init(testing.allocator, io, sessions, "/c"); + defer mgr.deinit(); + + const u_blocks = try testing.allocator.alloc(StoredContentBlock, 1); + u_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "ping") } }; + const user_id = try mgr.appendMessage(.{ .role = .user, .content = u_blocks }, oaStamp()); + + const a_blocks = try testing.allocator.alloc(StoredContentBlock, 1); + a_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "pong") } }; + const asst_id = try mgr.appendMessage(.{ .role = .assistant, .content = a_blocks }, null); + + // Leaf is the assistant entry. + try testing.expectEqualStrings(asst_id, mgr.getLeafId().?); + // Parent of assistant is the user entry. + const assistant_entry = mgr.getEntry(asst_id).?; + try testing.expectEqualStrings(user_id, assistant_entry.base().parent_id.?); + // User entry's parent is null (no system). + const user_entry = mgr.getEntry(user_id).?; + try testing.expect(user_entry.base().parent_id == null); +} + +test "SessionFile: activeStamp is null before any user message, then tracks the latest user stamp" { + 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); + + var mgr = try SessionFile.init(testing.allocator, io, sessions, "/c"); + defer mgr.deinit(); + + // No user messages yet — there is no "active" model on disk yet. + try testing.expect(mgr.activeStamp() == null); + + // Stamp a user message with anthropic. + const u_blocks = try testing.allocator.alloc(StoredContentBlock, 1); + u_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hi") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = u_blocks }, anStamp()); + + { + const am = mgr.activeStamp().?; + try testing.expectEqual(session_mod.APIStyle.anthropic_messages, am.api_style); + try testing.expectEqualStrings("claude-sonnet-4-20250514", am.model); + } +} + +test "SessionFile: rebuildConversation reconstructs system/user/assistant turn" { + 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); + + var mgr = try SessionFile.init(testing.allocator, io, sessions, "/c"); + defer mgr.deinit(); + + const sys = try testing.allocator.alloc(StoredContentBlock, 1); + sys[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "you are helpful") } }; + _ = try mgr.appendMessage(.{ .role = .system, .content = sys }, null); + + const u = try testing.allocator.alloc(StoredContentBlock, 1); + u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hello") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = u }, oaStamp()); + + const a = try testing.allocator.alloc(StoredContentBlock, 1); + a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hi!") } }; + _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, null); + + var conv = try mgr.rebuildConversation(); + defer conv.deinit(); + try testing.expectEqual(@as(usize, 3), conv.messages.items.len); + try testing.expectEqual(conversation_mod.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_mod.MessageRole.user, conv.messages.items[1].role); + try testing.expectEqualStrings("hello", conv.messages.items[1].content.items[0].Text.items); + try testing.expectEqual(conversation_mod.MessageRole.assistant, conv.messages.items[2].role); + try testing.expectEqualStrings("hi!", conv.messages.items[2].content.items[0].Text.items); +} + +test "SessionFile: crash recovery truncates corrupted trailing line" { + 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); + + // Build a valid session first. + const session_file: []u8 = blk: { + var mgr = try SessionFile.init(testing.allocator, io, sessions, "/c"); + defer mgr.deinit(); + const u = try testing.allocator.alloc(StoredContentBlock, 1); + u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "ping") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = u }, oaStamp()); + const a = try testing.allocator.alloc(StoredContentBlock, 1); + a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "pong") } }; + _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, null); + break :blk try testing.allocator.dupe(u8, mgr.getSessionFile()); + }; + defer testing.allocator.free(session_file); + + // Corrupt the file: append a partial JSON line at the end. + const garbage = "{\"type\":\"message\",\"id\":\"deadbeef\",\"parent"; + { + const file = try Io.Dir.cwd().openFile(io, session_file, .{ .mode = .write_only }); + defer file.close(io); + const len = try file.length(io); + try file.writePositionalAll(io, garbage, len); + } + // Confirm the file got bigger. + { + const f = try Io.Dir.cwd().openFile(io, session_file, .{ .mode = .read_only }); + defer f.close(io); + const corrupted_len = try f.length(io); + try testing.expect(corrupted_len > garbage.len); + } + + // Now resume — the partial line should be truncated. + var resumed = try SessionFile.open(testing.allocator, io, session_file); + defer resumed.deinit(); + try testing.expectEqual(@as(usize, 2), resumed.getEntries().len); + + // And the file on disk should match. + { + const bytes = try readWholeFile(testing.allocator, io, session_file); + defer testing.allocator.free(bytes); + try testing.expect(!std.mem.endsWith(u8, bytes, "parent")); + // Should end with a newline after the assistant entry. + try testing.expectEqual(@as(u8, '\n'), bytes[bytes.len - 1]); + } +} + +test "listSessions: returns most recent first, with counts" { + 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); + + // Create two sessions. + for (0..2) |i| { + var mgr = try SessionFile.init(testing.allocator, io, sessions, "/c"); + defer mgr.deinit(); + const u = try testing.allocator.alloc(StoredContentBlock, 1); + u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "u") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = u }, oaStamp()); + const a = try testing.allocator.alloc(StoredContentBlock, 1); + a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a") } }; + _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, null); + // Small sleep so UUIDv7 timestamps differ. + io.sleep(.fromMilliseconds(2), .real) catch {}; + _ = i; + } + + const infos = try listSessions(testing.allocator, io, sessions, null); + defer { + for (infos) |fi| fi.deinit(testing.allocator); + testing.allocator.free(infos); + } + try testing.expectEqual(@as(usize, 2), infos.len); + try testing.expectEqual(@as(usize, 2), infos[0].message_count); + try testing.expect(std.mem.order(u8, infos[0].modified, infos[1].modified) != .lt); +} + +test "findMostRecentSession: picks lexicographically greatest" { + 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); + + // Pre-resolution before any sessions exist → null. + try testing.expect((try findMostRecentSession(testing.allocator, io, sessions)) == null); + + // Create two. + var second_file: ?[]u8 = null; + defer if (second_file) |s| testing.allocator.free(s); + + for (0..2) |i| { + var mgr = try SessionFile.init(testing.allocator, io, sessions, "/c"); + defer mgr.deinit(); + const u = try testing.allocator.alloc(StoredContentBlock, 1); + u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "u") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = u }, oaStamp()); + const a = try testing.allocator.alloc(StoredContentBlock, 1); + a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a") } }; + _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, null); + if (i == 1) second_file = try testing.allocator.dupe(u8, mgr.getSessionFile()); + io.sleep(.fromMilliseconds(2), .real) catch {}; + } + + const found = (try findMostRecentSession(testing.allocator, io, sessions)).?; + defer testing.allocator.free(found); + try testing.expectEqualStrings(second_file.?, found); +} + +test "SessionFile: tool-use round-trip — assistant w/ ToolUse, user w/ ToolResult, assistant" { + 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 SessionFile.init(testing.allocator, io, sessions, "/c"); + defer mgr.deinit(); + + const u = try testing.allocator.alloc(StoredContentBlock, 1); + u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "list files") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = u }, oaStamp()); + + // Assistant emits a ToolUse. + const am1 = try testing.allocator.alloc(StoredContentBlock, 2); + am1[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "checking...") } }; + am1[1] = .{ .tool_use = .{ + .id = try testing.allocator.dupe(u8, "tool_abc"), + .name = try testing.allocator.dupe(u8, "bash"), + .input = try testing.allocator.dupe(u8, "{\"command\":\"ls\"}"), + } }; + _ = try mgr.appendMessage(.{ .role = .assistant, .content = am1 }, null); + + // Tool-result user message. + const tr = try testing.allocator.alloc(StoredContentBlock, 1); + const trp = try testing.allocator.alloc(session_mod.StoredResultPart, 1); + trp[0] = .{ .text = try testing.allocator.dupe(u8, "a.txt\nb.txt") }; + tr[0] = .{ .tool_result = .{ + .tool_use_id = try testing.allocator.dupe(u8, "tool_abc"), + .parts = trp, + } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = tr }, oaStamp()); + + // Final assistant reply. + const a2 = try testing.allocator.alloc(StoredContentBlock, 1); + a2[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "two files: a.txt and b.txt") } }; + _ = try mgr.appendMessage(.{ .role = .assistant, .content = a2 }, null); + + break :blk try testing.allocator.dupe(u8, mgr.getSessionFile()); + }; + defer testing.allocator.free(session_file); + + // Reopen and verify content blocks survive. + var resumed = try SessionFile.open(testing.allocator, io, session_file); + defer resumed.deinit(); + const entries = resumed.getEntries(); + try testing.expectEqual(@as(usize, 4), entries.len); + + // [1] = assistant with ToolUse + try testing.expectEqual(StoredMessageRole.assistant, entries[1].message.message.role); + try testing.expectEqual(@as(usize, 2), entries[1].message.message.content.len); + try testing.expect(entries[1].message.message.content[1] == .tool_use); + try testing.expectEqualStrings("bash", entries[1].message.message.content[1].tool_use.name); + try testing.expectEqualStrings("{\"command\":\"ls\"}", entries[1].message.message.content[1].tool_use.input); + + // [2] = user with ToolResult, stamped with wire identity. + try testing.expectEqual(StoredMessageRole.user, entries[2].message.message.role); + try testing.expectEqual(session_mod.APIStyle.openai_chat, entries[2].message.stamp.?.api_style); + try testing.expect(entries[2].message.message.content[0] == .tool_result); + try testing.expectEqualStrings("tool_abc", entries[2].message.message.content[0].tool_result.tool_use_id); + try testing.expectEqualStrings("a.txt\nb.txt", entries[2].message.message.content[0].tool_result.parts[0].text); + + // Conversation rebuild yields the same shape. + var conv = try resumed.rebuildConversation(); + defer conv.deinit(); + try testing.expectEqual(@as(usize, 4), conv.messages.items.len); + try testing.expect(conv.messages.items[1].content.items[1] == .ToolUse); + try testing.expect(conv.messages.items[2].content.items[0] == .ToolResult); +} + +test "SessionFile: linear chain — each entry's parent_id is the previous entry's id" { + 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); + + var mgr = try SessionFile.init(testing.allocator, io, sessions, "/c"); + defer mgr.deinit(); + + // Three rounds: sys, user, asst, user, asst. + const sys = try testing.allocator.alloc(StoredContentBlock, 1); + sys[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "sys") } }; + _ = try mgr.appendMessage(.{ .role = .system, .content = sys }, null); + + const u_one = try testing.allocator.alloc(StoredContentBlock, 1); + u_one[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "u1") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = u_one }, oaStamp()); + + const a_one = try testing.allocator.alloc(StoredContentBlock, 1); + a_one[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a1") } }; + _ = try mgr.appendMessage(.{ .role = .assistant, .content = a_one }, null); + + const u_two = try testing.allocator.alloc(StoredContentBlock, 1); + u_two[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "u2") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = u_two }, oaStamp()); + + const a_two = try testing.allocator.alloc(StoredContentBlock, 1); + a_two[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a2") } }; + _ = try mgr.appendMessage(.{ .role = .assistant, .content = a_two }, null); + + const entries = mgr.getEntries(); + try testing.expectEqual(@as(usize, 5), entries.len); + try testing.expect(entries[0].base().parent_id == null); + for (entries[1..], 1..) |e, i| { + try testing.expectEqualStrings(entries[i - 1].base().id, e.base().parent_id.?); + } +} + +test "resolveSessionId: unique prefix → match, ambiguous → error" { + 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); + + // Create one session. + var mgr = try SessionFile.init(testing.allocator, io, sessions, "/c"); + defer mgr.deinit(); + const u = try testing.allocator.alloc(StoredContentBlock, 1); + u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "u") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = u }, oaStamp()); + const a = try testing.allocator.alloc(StoredContentBlock, 1); + a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a") } }; + _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, null); + + const id = mgr.getSessionId(); + const prefix = id[0..8]; + + const resolved = try resolveSessionId(testing.allocator, io, sessions, prefix); + defer testing.allocator.free(resolved); + try testing.expectEqualStrings(mgr.getSessionFile(), resolved); + + 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 SessionFile.init(testing.allocator, io, sessions, "/c"); + defer mgr.deinit(); + + // System + an old turn that will be superseded. + const sys = try testing.allocator.alloc(StoredContentBlock, 1); + sys[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "you are helpful") } }; + _ = try mgr.appendMessage(.{ .role = .system, .content = sys }, null); + + const uo = try testing.allocator.alloc(StoredContentBlock, 1); + uo[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "old q") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = uo }, oaStamp()); + const ao = try testing.allocator.alloc(StoredContentBlock, 1); + ao[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "old a") } }; + _ = try mgr.appendMessage(.{ .role = .assistant, .content = ao }, null); + + // Compaction: summary message + duplicated kept suffix. + const cs = try testing.allocator.alloc(StoredContentBlock, 1); + cs[0] = .{ .compaction_summary = .{ .text = try testing.allocator.dupe(u8, "SUMMARY") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = cs }, oaStamp()); + + const ur = try testing.allocator.alloc(StoredContentBlock, 1); + ur[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "recent q") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = ur }, oaStamp()); + + break :blk try testing.allocator.dupe(u8, mgr.getSessionFile()); + }; + defer testing.allocator.free(session_file); + + var resumed = try SessionFile.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]); +} + +test "loadConversation: trailing user prompt is split out as dangling, excluded from conversation" { + 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); + + var mgr = try SessionFile.init(testing.allocator, io, sessions, "/c"); + defer mgr.deinit(); + + // A completed user/assistant round, then a trailing user prompt with no + // following assistant. The dangling-prompt recovery feature was dropped + // in R2: the trailing user message simply round-trips into the rebuilt + // conversation like any other. + const um1 = try testing.allocator.alloc(StoredContentBlock, 1); + um1[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hi") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = um1 }, oaStamp()); + const am1 = try testing.allocator.alloc(StoredContentBlock, 1); + am1[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hello") } }; + _ = try mgr.appendMessage(.{ .role = .assistant, .content = am1 }, null); + const um2 = try testing.allocator.alloc(StoredContentBlock, 1); + um2[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "what's 2+2?") } }; + _ = try mgr.appendMessage(.{ .role = .user, .content = um2 }, oaStamp()); + + var conv = try mgr.rebuildConversation(); + defer conv.deinit(); + // All three messages are present (dangling recovery dropped). + try testing.expectEqual(@as(usize, 3), conv.messages.items.len); + try testing.expectEqual(conversation_mod.MessageRole.user, conv.messages.items[2].role); +} + +test "FileSystemJSONLStore catalog: create → append → load round-trips" { + 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); + + var catalog = try FileSystemJSONLStore.init(testing.allocator, io, sessions, "/c"); + defer catalog.deinit(); + const store = catalog.store(); + + var sess = store.create(); + defer sess.info.deinit(testing.allocator); + + // Build a user + assistant PersistentMessage batch (borrows in-memory + // messages owned here). + var conv = conversation_mod.Conversation.init(testing.allocator); + defer conv.deinit(); + try conv.addUserMessage("ping"); + try conv.addAssistantMessage(&.{}); + + const id: session_store_mod.WireIdentity = .{ .api_style = .openai_chat, .base_url = "u", .model = "m" }; + var batch = [_]session_store_mod.PersistentMessage{ + .{ .message = conv.messages.items[0], .identity = id }, + .{ .message = conv.messages.items[1], .identity = id }, + }; + try sess.append(&batch); + + // The session's last-used api_style updated after append. + try testing.expectEqual(session_store_mod.APIStyle.openai_chat, sess.info.api_style); + + // Load it back by id. + var loaded = (try store.load(sess.info.id)).?; + defer loaded.deinit(); + try testing.expectEqual(@as(usize, 2), loaded.messages.items.len); + try testing.expectEqual(conversation_mod.MessageRole.user, loaded.messages.items[0].role); +} diff --git a/libpanto/src/null_store.zig b/libpanto/src/null_store.zig index d2337ee..1f3ab74 100644 --- a/libpanto/src/null_store.zig +++ b/libpanto/src/null_store.zig @@ -2,15 +2,10 @@ //! persistence (and the default backing for an `Agent` constructed without //! a real store). //! -//! Every append is dropped (but its messages are freed, honoring the store -//! ownership contract — appends consume their messages). `loadConversation` -//! returns an empty conversation with no dangling prompt. `activeModel` is -//! null and the session id is the empty string. -//! -//! Because freeing the dropped messages needs an allocator, `NullStore` -//! carries one. Construct it with `init(alloc)` and call `.store()` for the -//! interface handle. The struct is trivially copyable and holds no other -//! state, so the handle may borrow it for the agent's lifetime. +//! Every append is dropped. `load` returns null. `list` returns an empty +//! slice. `create`/`resolve`/`latest` mint/return empty handles. The struct +//! holds an allocator (needed to satisfy the `SessionInfo` ownership +//! contract for the empty handles it mints) and is trivially copyable. const std = @import("std"); const Allocator = std.mem.Allocator; @@ -19,9 +14,10 @@ const session_store_mod = @import("session_store.zig"); const conversation_mod = @import("conversation.zig"); const SessionStore = session_store_mod.SessionStore; -const LoadedSession = session_store_mod.LoadedSession; -const DiskMessage = session_store_mod.DiskMessage; -const ActiveModel = session_store_mod.ActiveModel; +const Session = session_store_mod.Session; +const SessionInfo = session_store_mod.SessionInfo; +const PersistentMessage = session_store_mod.PersistentMessage; +const Conversation = conversation_mod.Conversation; pub const NullStore = struct { allocator: Allocator, @@ -30,38 +26,62 @@ pub const NullStore = struct { return .{ .allocator = allocator }; } - fn appendMessagesVT( - ctx: *anyopaque, - messages: []DiskMessage, - _: []const ?[]const u8, - _: []const ?[]const u8, - ) anyerror!void { + fn emptyInfo(self: *NullStore) SessionInfo { + const a = self.allocator; + return .{ + .id = a.dupe(u8, "") catch "", + .created = a.dupe(u8, "") catch "", + .modified = a.dupe(u8, "") catch "", + .message_count = 0, + .last_user_message = a.dupe(u8, "") catch "", + .api_style = .openai_chat, + .base_url = a.dupe(u8, "") catch "", + .model = a.dupe(u8, "") catch "", + .reasoning = .default, + }; + } + + fn createVT(ctx: *anyopaque) Session { const self: *NullStore = @ptrCast(@alignCast(ctx)); - // Appends consume their messages; we drop them, so free them. - for (messages) |m| m.deinit(self.allocator); + return .{ .info = self.emptyInfo(), .store = self.store() }; } - fn loadConversationVT(ctx: *anyopaque, alloc: Allocator) anyerror!LoadedSession { - _ = ctx; - return .{ - .conversation = conversation_mod.Conversation.init(alloc), - .dangling_user = null, - }; + fn listVT(ctx: *anyopaque) anyerror![]SessionInfo { + const self: *NullStore = @ptrCast(@alignCast(ctx)); + return self.allocator.alloc(SessionInfo, 0); + } + + fn freeSessionInfosVT(ctx: *anyopaque, infos: []SessionInfo) void { + const self: *NullStore = @ptrCast(@alignCast(ctx)); + for (infos) |i| i.deinit(self.allocator); + self.allocator.free(infos); } - fn sessionIdVT(_: *anyopaque) []const u8 { - return ""; + fn resolveVT(_: *anyopaque, _: []const u8) anyerror!?Session { + return null; } - fn activeModelVT(_: *anyopaque) ?ActiveModel { + fn latestVT(_: *anyopaque) anyerror!?Session { return null; } + fn loadVT(_: *anyopaque, _: []const u8) anyerror!?Conversation { + return null; + } + + fn appendMessagesVT(_: *anyopaque, _: []const u8, _: []PersistentMessage) anyerror!void { + // Drop everything. The PersistentMessages borrow in-memory data + // owned by the caller (the conversation); nothing to free here. + } + const vtable: SessionStore.VTable = .{ + .create = createVT, + .list = listVT, + .freeSessionInfos = freeSessionInfosVT, + .resolve = resolveVT, + .latest = latestVT, + .load = loadVT, .appendMessages = appendMessagesVT, - .loadConversation = loadConversationVT, - .sessionId = sessionIdVT, - .activeModel = activeModelVT, }; /// Wrap this `NullStore` as a `SessionStore`. The handle borrows @@ -73,20 +93,16 @@ pub const NullStore = struct { const testing = std.testing; -test "NullStore: appends are dropped (and freed) and load returns empty" { +test "NullStore: appends are dropped and load returns null" { var ns = NullStore.init(testing.allocator); const s = ns.store(); - var msg_content = try testing.allocator.alloc(session_store_mod.DiskContentBlock, 1); - msg_content[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hi") } }; - var messages = [_]DiskMessage{.{ .role = .user, .content = msg_content }}; - // No defer-free: the store consumes (frees) the message. - try s.appendMessages(&messages, &.{null}, &.{null}); - - var loaded = try s.loadConversation(testing.allocator); - defer loaded.deinit(testing.allocator); - try testing.expectEqual(@as(usize, 0), loaded.conversation.messages.items.len); - try testing.expect(loaded.dangling_user == null); - try testing.expect(s.activeModel() == null); - try testing.expectEqualStrings("", s.sessionId()); + try s.appendMessages("sid", &.{}); + try testing.expect((try s.load("sid")) == null); + try testing.expect((try s.resolve("sid")) == null); + try testing.expect((try s.latest()) == null); + + const infos = try s.list(); + defer s.freeSessionInfos(infos); + try testing.expectEqual(@as(usize, 0), infos.len); } diff --git a/libpanto/src/root.zig b/libpanto/src/root.zig index b5fab1a..fb8c1df 100644 --- a/libpanto/src/root.zig +++ b/libpanto/src/root.zig @@ -10,7 +10,7 @@ pub const tool = @import("tool.zig"); pub const tool_source = @import("tool_source.zig"); pub const tool_registry = @import("tool_registry.zig"); pub const session = @import("session.zig"); -pub const session_manager = @import("session_manager.zig"); +pub const session_manager = @import("file_system_jsonl_store.zig"); pub const session_store = @import("session_store.zig"); pub const null_store = @import("null_store.zig"); pub const turn_persist = @import("turn_persist.zig"); diff --git a/libpanto/src/session.zig b/libpanto/src/session.zig index 0491157..ca7046c 100644 --- a/libpanto/src/session.zig +++ b/libpanto/src/session.zig @@ -23,6 +23,38 @@ const Allocator = std.mem.Allocator; const Writer = std.Io.Writer; const conversation = @import("conversation.zig"); +const config = @import("config.zig"); + +pub const APIStyle = config.APIStyle; +pub const ReasoningEffort = config.ReasoningEffort; + +/// Wire-format provider identity stamped on a message entry. This is the +/// ground truth of which endpoint a turn was sent to — never a CLI config +/// alias, and never any `api_key` material. Recorded on user/assistant +/// entries; null on system entries. +pub const WireStamp = struct { + api_style: APIStyle, + base_url: []const u8, // owned + model: []const u8, // owned + reasoning: ReasoningEffort = .default, + + pub fn deinit(self: WireStamp, alloc: Allocator) void { + alloc.free(self.base_url); + alloc.free(self.model); + } + + pub fn dupe(self: WireStamp, alloc: Allocator) !WireStamp { + const burl = try alloc.dupe(u8, self.base_url); + errdefer alloc.free(burl); + const mdl = try alloc.dupe(u8, self.model); + return .{ + .api_style = self.api_style, + .base_url = burl, + .model = mdl, + .reasoning = self.reasoning, + }; + } +}; /// Bumped whenever the on-disk format changes in a way that older readers /// cannot tolerate. Older files are upgraded by `migrate()` on load and @@ -83,47 +115,45 @@ pub const SessionEntry = union(enum) { pub const MessageEntry = struct { base: EntryBase, - /// Recorded on user message entries (both human prompts and tool-result - /// messages). Both are submissions to a provider API; the stamp says - /// which one. Null on system and assistant entries. - provider: ?[]const u8 = null, // owned - model: ?[]const u8 = null, // owned - message: DiskMessage, + /// Wire-format provider identity for this entry. Recorded on user and + /// assistant message entries (both are tied to a provider API call); + /// null on system entries. + stamp: ?WireStamp = null, + message: StoredMessage, pub fn deinit(self: MessageEntry, alloc: Allocator) void { self.base.deinit(alloc); - if (self.provider) |p| alloc.free(p); - if (self.model) |m| alloc.free(m); + if (self.stamp) |s| s.deinit(alloc); self.message.deinit(alloc); } }; -pub const DiskMessageRole = enum { system, user, assistant }; +pub const StoredMessageRole = enum { system, user, assistant }; /// Mode for a system-role message. Mirrors `conversation.SystemMode`. /// `append` adds to the effective prompt; `replace` discards all prior /// system text. Only meaningful on system messages; absent on disk means /// `append` (back-compatible with pre-mode logs). -pub const DiskSystemMode = enum { append, replace }; +pub const StoredSystemMode = enum { append, replace }; -pub const DiskMessage = struct { - role: DiskMessageRole, - content: []DiskContentBlock, // owned +pub const StoredMessage = struct { + role: StoredMessageRole, + content: []StoredContentBlock, // owned /// System-message mode. Recorded only for system-role messages; an /// absent `mode` on disk parses back as `.append`. - mode: DiskSystemMode = .append, - // Assistant-only metadata. Null for system/user messages. - provider: ?[]const u8 = null, // owned - model: ?[]const u8 = null, // owned + mode: StoredSystemMode = .append, + /// Assistant-only stop reason. Null for system/user messages. stop_reason: ?[]const u8 = null, // owned usage: ?Usage = null, + /// Opaque per-message metadata bag (see `conversation.Message.metadata`). + /// Round-trips verbatim; `libpanto` never interprets it. + metadata: ?[]const u8 = null, // owned - pub fn deinit(self: DiskMessage, alloc: Allocator) void { + pub fn deinit(self: StoredMessage, alloc: Allocator) void { for (self.content) |block| block.deinit(alloc); alloc.free(self.content); - if (self.provider) |p| alloc.free(p); - if (self.model) |m| alloc.free(m); if (self.stop_reason) |s| alloc.free(s); + if (self.metadata) |m| alloc.free(m); } }; @@ -138,14 +168,14 @@ pub const Usage = conversation.Usage; // Content blocks // ============================================================================= -pub const DiskContentBlock = union(enum) { - text: DiskTextBlock, - thinking: DiskThinkingBlock, - tool_use: DiskToolUseBlock, - tool_result: DiskToolResultBlock, - compaction_summary: DiskCompactionSummaryBlock, +pub const StoredContentBlock = union(enum) { + text: StoredTextBlock, + thinking: StoredThinkingBlock, + tool_use: StoredToolUseBlock, + tool_result: StoredToolResultBlock, + compaction_summary: StoredCompactionSummaryBlock, - pub fn deinit(self: DiskContentBlock, alloc: Allocator) void { + pub fn deinit(self: StoredContentBlock, alloc: Allocator) void { switch (self) { .text => |b| b.deinit(alloc), .thinking => |b| b.deinit(alloc), @@ -156,30 +186,30 @@ pub const DiskContentBlock = union(enum) { } }; -pub const DiskTextBlock = struct { +pub const StoredTextBlock = struct { text: []const u8, // owned - pub fn deinit(self: DiskTextBlock, alloc: Allocator) void { + pub fn deinit(self: StoredTextBlock, alloc: Allocator) void { alloc.free(self.text); } }; -pub const DiskThinkingBlock = struct { +pub const StoredThinkingBlock = struct { thinking: []const u8, // owned /// Anthropic's opaque integrity token. Other providers do not produce /// one. Preserved here so resumed sessions can be sent back to /// Anthropic with the original thinking block intact. signature: ?[]const u8 = null, // owned - pub fn deinit(self: DiskThinkingBlock, alloc: Allocator) void { + pub fn deinit(self: StoredThinkingBlock, alloc: Allocator) void { alloc.free(self.thinking); if (self.signature) |s| alloc.free(s); } }; -pub const DiskToolUseBlock = struct { +pub const StoredToolUseBlock = struct { id: []const u8, // owned name: []const u8, // owned input: []const u8, // raw JSON bytes, owned - pub fn deinit(self: DiskToolUseBlock, alloc: Allocator) void { + pub fn deinit(self: StoredToolUseBlock, alloc: Allocator) void { alloc.free(self.id); alloc.free(self.name); alloc.free(self.input); @@ -188,13 +218,13 @@ pub const DiskToolUseBlock = struct { /// One on-disk tool-result part: either text or an inline base64 media /// attachment (no sidecar files). -pub const DiskResultPart = union(enum) { +pub const StoredResultPart = union(enum) { text: []const u8, // owned media: struct { media_type: []const u8, // owned data: []const u8, // owned (base64) }, - pub fn deinit(self: DiskResultPart, alloc: Allocator) void { + pub fn deinit(self: StoredResultPart, alloc: Allocator) void { switch (self) { .text => |t| alloc.free(t), .media => |m| { @@ -205,11 +235,11 @@ pub const DiskResultPart = union(enum) { } }; -pub const DiskToolResultBlock = struct { +pub const StoredToolResultBlock = struct { tool_use_id: []const u8, // owned - parts: []DiskResultPart, // owned + parts: []StoredResultPart, // owned is_error: bool = false, - pub fn deinit(self: DiskToolResultBlock, alloc: Allocator) void { + pub fn deinit(self: StoredToolResultBlock, alloc: Allocator) void { alloc.free(self.tool_use_id); for (self.parts) |p| p.deinit(alloc); alloc.free(self.parts); @@ -219,9 +249,9 @@ 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 { +pub const StoredCompactionSummaryBlock = struct { text: []const u8, // owned - pub fn deinit(self: DiskCompactionSummaryBlock, alloc: Allocator) void { + pub fn deinit(self: StoredCompactionSummaryBlock, alloc: Allocator) void { alloc.free(self.text); } }; @@ -294,21 +324,25 @@ fn writeMessageEntry(s: *std.json.Stringify, m: MessageEntry) !void { if (m.base.parent_id) |p| try s.write(p) else try s.write(null); try s.objectField("timestamp"); try s.write(m.base.timestamp); - // Top-level provider/model on user message entries. - if (m.provider) |p| { - try s.objectField("provider"); - try s.write(p); - } - if (m.model) |mm| { - try s.objectField("model"); - try s.write(mm); - } + // Wire-format provider identity on user/assistant entries. + if (m.stamp) |st| try writeWireStamp(s, st); try s.objectField("message"); try writeDiskMessage(s, m.message); try s.endObject(); } -fn writeDiskMessage(s: *std.json.Stringify, msg: DiskMessage) !void { +fn writeWireStamp(s: *std.json.Stringify, st: WireStamp) !void { + try s.objectField("apiStyle"); + try s.write(@tagName(st.api_style)); + try s.objectField("baseUrl"); + try s.write(st.base_url); + try s.objectField("model"); + try s.write(st.model); + try s.objectField("reasoning"); + try s.write(@tagName(st.reasoning)); +} + +fn writeDiskMessage(s: *std.json.Stringify, msg: StoredMessage) !void { try s.beginObject(); try s.objectField("role"); try s.write(@tagName(msg.role)); @@ -324,18 +358,14 @@ fn writeDiskMessage(s: *std.json.Stringify, msg: DiskMessage) !void { try writeDiskBlock(s, block); } try s.endArray(); - if (msg.provider) |p| { - try s.objectField("provider"); - try s.write(p); - } - if (msg.model) |mm| { - try s.objectField("model"); - try s.write(mm); - } if (msg.stop_reason) |sr| { try s.objectField("stopReason"); try s.write(sr); } + if (msg.metadata) |md| { + try s.objectField("metadata"); + try s.write(md); + } if (msg.usage) |u| { try s.objectField("usage"); try s.beginObject(); @@ -363,7 +393,7 @@ fn writeDiskMessage(s: *std.json.Stringify, msg: DiskMessage) !void { try s.endObject(); } -fn writeDiskBlock(s: *std.json.Stringify, block: DiskContentBlock) !void { +fn writeDiskBlock(s: *std.json.Stringify, block: StoredContentBlock) !void { switch (block) { .text => |b| { try s.beginObject(); @@ -518,10 +548,8 @@ fn parseMessageEntry(allocator: Allocator, obj: std.json.ObjectMap) ParseError!M }; errdefer if (parent_id) |p| allocator.free(p); - const provider: ?[]const u8 = try dupeOptionalStringField(allocator, obj, "provider"); - errdefer if (provider) |p| allocator.free(p); - const model: ?[]const u8 = try dupeOptionalStringField(allocator, obj, "model"); - errdefer if (model) |m| allocator.free(m); + const stamp = try parseWireStamp(allocator, obj); + errdefer if (stamp) |st| st.deinit(allocator); const msg_v = obj.get("message") orelse return error.MissingField; if (msg_v != .object) return error.MissingField; @@ -529,28 +557,45 @@ fn parseMessageEntry(allocator: Allocator, obj: std.json.ObjectMap) ParseError!M return .{ .base = .{ .id = id, .parent_id = parent_id, .timestamp = timestamp }, - .provider = provider, - .model = model, + .stamp = stamp, .message = msg, }; } -fn parseDiskMessage(allocator: Allocator, obj: std.json.ObjectMap) ParseError!DiskMessage { +/// Parse the wire-format provider stamp from a message entry object. +/// Returns null when no `apiStyle` field is present (system entries). +fn parseWireStamp(allocator: Allocator, obj: std.json.ObjectMap) ParseError!?WireStamp { + const style_v = obj.get("apiStyle") orelse return null; + if (style_v != .string) return null; + const api_style = std.meta.stringToEnum(APIStyle, style_v.string) orelse return error.MissingField; + const base_url = try dupeStringField(allocator, obj, "baseUrl"); + errdefer allocator.free(base_url); + const model = try dupeStringField(allocator, obj, "model"); + errdefer allocator.free(model); + const reasoning: ReasoningEffort = blk: { + const rv = obj.get("reasoning") orelse break :blk .default; + if (rv != .string) break :blk .default; + break :blk std.meta.stringToEnum(ReasoningEffort, rv.string) orelse .default; + }; + return .{ .api_style = api_style, .base_url = base_url, .model = model, .reasoning = reasoning }; +} + +fn parseDiskMessage(allocator: Allocator, obj: std.json.ObjectMap) ParseError!StoredMessage { const role_v = obj.get("role") orelse return error.MissingField; if (role_v != .string) return error.MissingField; - const role = std.meta.stringToEnum(DiskMessageRole, role_v.string) orelse return error.UnknownRole; + const role = std.meta.stringToEnum(StoredMessageRole, role_v.string) orelse return error.UnknownRole; // `mode` is optional; absent defaults to `.append`. Unknown values are // tolerated as `.append` rather than rejecting an otherwise-valid log. - const mode: DiskSystemMode = blk: { + const mode: StoredSystemMode = blk: { const mv = obj.get("mode") orelse break :blk .append; if (mv != .string) break :blk .append; - break :blk std.meta.stringToEnum(DiskSystemMode, mv.string) orelse .append; + break :blk std.meta.stringToEnum(StoredSystemMode, mv.string) orelse .append; }; const content_v = obj.get("content") orelse return error.MissingField; if (content_v != .array) return error.MissingField; - var content_list = try std.ArrayList(DiskContentBlock).initCapacity(allocator, content_v.array.items.len); + var content_list = try std.ArrayList(StoredContentBlock).initCapacity(allocator, content_v.array.items.len); errdefer { for (content_list.items) |b| b.deinit(allocator); content_list.deinit(allocator); @@ -566,12 +611,10 @@ fn parseDiskMessage(allocator: Allocator, obj: std.json.ObjectMap) ParseError!Di allocator.free(content); } - const provider: ?[]const u8 = try dupeOptionalStringField(allocator, obj, "provider"); - errdefer if (provider) |p| allocator.free(p); - const model: ?[]const u8 = try dupeOptionalStringField(allocator, obj, "model"); - errdefer if (model) |m| allocator.free(m); const stop_reason: ?[]const u8 = try dupeOptionalStringField(allocator, obj, "stopReason"); errdefer if (stop_reason) |s| allocator.free(s); + const metadata: ?[]const u8 = try dupeOptionalStringField(allocator, obj, "metadata"); + errdefer if (metadata) |m| allocator.free(m); var usage: ?Usage = null; if (obj.get("usage")) |uv| { @@ -590,14 +633,13 @@ fn parseDiskMessage(allocator: Allocator, obj: std.json.ObjectMap) ParseError!Di .role = role, .content = content, .mode = mode, - .provider = provider, - .model = model, .stop_reason = stop_reason, .usage = usage, + .metadata = metadata, }; } -fn parseDiskBlock(allocator: Allocator, obj: std.json.ObjectMap) ParseError!DiskContentBlock { +fn parseDiskBlock(allocator: Allocator, obj: std.json.ObjectMap) ParseError!StoredContentBlock { const type_v = obj.get("type") orelse return error.MissingField; if (type_v != .string) return error.MissingField; const t = type_v.string; @@ -634,8 +676,8 @@ fn parseDiskBlock(allocator: Allocator, obj: std.json.ObjectMap) ParseError!Disk /// Parse the `parts` array of a `toolResult` disk block. Falls back to a /// legacy single `content` string field (older session logs) -> one text /// part. Each element is {type:"text",text} or {type:"image",mimeType,data}. -fn parseDiskResultParts(allocator: Allocator, obj: std.json.ObjectMap) ParseError![]DiskResultPart { - var list: std.ArrayList(DiskResultPart) = .empty; +fn parseDiskResultParts(allocator: Allocator, obj: std.json.ObjectMap) ParseError![]StoredResultPart { + var list: std.ArrayList(StoredResultPart) = .empty; errdefer { for (list.items) |p| p.deinit(allocator); list.deinit(allocator); @@ -698,13 +740,13 @@ fn dupeOptionalStringField(allocator: Allocator, obj: std.json.ObjectMap, name: // Bridge between in-memory and on-disk content blocks // ============================================================================= -/// Convert an in-memory `ContentBlock` to a `DiskContentBlock`. All strings +/// Convert an in-memory `ContentBlock` to a `StoredContentBlock`. All strings /// are duplicated; the source block remains untouched and the resulting /// disk block is independently owned. pub fn contentBlockToDisk( allocator: Allocator, block: conversation.ContentBlock, -) !DiskContentBlock { +) !StoredContentBlock { switch (block) { .Text => |tb| { const text = try allocator.dupe(u8, tb.items); @@ -727,7 +769,7 @@ pub fn contentBlockToDisk( .ToolResult => |tr| { const tuid = try allocator.dupe(u8, tr.tool_use_id); errdefer allocator.free(tuid); - var parts: std.ArrayList(DiskResultPart) = .empty; + var parts: std.ArrayList(StoredResultPart) = .empty; errdefer { for (parts.items) |p| p.deinit(allocator); parts.deinit(allocator); @@ -751,7 +793,7 @@ pub fn contentBlockToDisk( } }; }, // A `.System` block becomes a disk text block; its mode rides on - // the enclosing `DiskMessage.mode` (set by the session manager), + // the enclosing `StoredMessage.mode` (set by the session manager), // not on the block itself. .System => |sb| { const text = try allocator.dupe(u8, sb.text.items); @@ -764,12 +806,12 @@ pub fn contentBlockToDisk( } } -/// Convert a `DiskContentBlock` to an in-memory `ContentBlock`. Allocates +/// Convert a `StoredContentBlock` to an in-memory `ContentBlock`. Allocates /// fresh owned buffers for every string field. The returned block is /// independently owned. pub fn diskContentBlockToInternal( allocator: Allocator, - block: DiskContentBlock, + block: StoredContentBlock, ) !conversation.ContentBlock { switch (block) { .text => |b| { @@ -856,7 +898,7 @@ test "serialize/parse header round-trip" { test "serialize/parse user message entry round-trip (with provider/model stamp)" { const a = testing.allocator; - var content = try a.alloc(DiskContentBlock, 1); + var content = try a.alloc(StoredContentBlock, 1); content[0] = .{ .text = .{ .text = try dupe(a, "hello world") } }; const entry: SessionEntry = .{ .message = .{ @@ -865,8 +907,12 @@ test "serialize/parse user message entry round-trip (with provider/model stamp)" .parent_id = try dupe(a, "00000000"), .timestamp = try dupe(a, "2026-04-25T17:40:16.000Z"), }, - .provider = try dupe(a, "openai"), - .model = try dupe(a, "gpt-4o"), + .stamp = .{ + .api_style = .openai_chat, + .base_url = try dupe(a, "https://api.openai.com/v1"), + .model = try dupe(a, "gpt-4o"), + .reasoning = .high, + }, .message = .{ .role = .user, .content = content, @@ -883,9 +929,11 @@ test "serialize/parse user message entry round-trip (with provider/model stamp)" const got = fe.entry.message; try testing.expectEqualStrings("a1b2c3d4", got.base.id); try testing.expectEqualStrings("00000000", got.base.parent_id.?); - try testing.expectEqualStrings("openai", got.provider.?); - try testing.expectEqualStrings("gpt-4o", got.model.?); - try testing.expectEqual(DiskMessageRole.user, got.message.role); + try testing.expectEqual(APIStyle.openai_chat, got.stamp.?.api_style); + try testing.expectEqualStrings("https://api.openai.com/v1", got.stamp.?.base_url); + try testing.expectEqualStrings("gpt-4o", got.stamp.?.model); + try testing.expectEqual(ReasoningEffort.high, got.stamp.?.reasoning); + try testing.expectEqual(StoredMessageRole.user, got.message.role); try testing.expectEqual(@as(usize, 1), got.message.content.len); try testing.expectEqualStrings("hello world", got.message.content[0].text.text); } @@ -893,7 +941,7 @@ test "serialize/parse user message entry round-trip (with provider/model stamp)" test "serialize/parse assistant message entry with metadata" { const a = testing.allocator; - var content = try a.alloc(DiskContentBlock, 3); + var content = try a.alloc(StoredContentBlock, 3); content[0] = .{ .thinking = .{ .thinking = try dupe(a, "let me think"), .signature = try dupe(a, "sig-xyz"), @@ -911,13 +959,17 @@ test "serialize/parse assistant message entry with metadata" { .parent_id = try dupe(a, "a1b2c3d4"), .timestamp = try dupe(a, "2026-04-25T17:40:17.000Z"), }, + .stamp = .{ + .api_style = .anthropic_messages, + .base_url = try dupe(a, "https://api.anthropic.com"), + .model = try dupe(a, "claude-sonnet-4-20250514"), + }, .message = .{ .role = .assistant, .content = content, - .provider = try dupe(a, "anthropic"), - .model = try dupe(a, "claude-sonnet-4-20250514"), .stop_reason = try dupe(a, "toolUse"), .usage = .{ .input = 1500, .output = 85 }, + .metadata = try dupe(a, "{\"k\":1}"), }, } }; defer entry.deinit(a); @@ -928,14 +980,15 @@ test "serialize/parse assistant message entry with metadata" { var fe = try parseLine(a, line); defer fe.deinit(a); const got = fe.entry.message; - try testing.expectEqual(DiskMessageRole.assistant, got.message.role); + try testing.expectEqual(StoredMessageRole.assistant, got.message.role); try testing.expectEqual(@as(usize, 3), got.message.content.len); try testing.expectEqualStrings("let me think", got.message.content[0].thinking.thinking); try testing.expectEqualStrings("sig-xyz", got.message.content[0].thinking.signature.?); try testing.expectEqualStrings("bash", got.message.content[2].tool_use.name); try testing.expectEqualStrings("{\"command\":\"ls\"}", got.message.content[2].tool_use.input); - try testing.expectEqualStrings("anthropic", got.message.provider.?); + try testing.expectEqualStrings("anthropic", @tagName(got.stamp.?.api_style)[0..9]); try testing.expectEqualStrings("toolUse", got.message.stop_reason.?); + try testing.expectEqualStrings("{\"k\":1}", got.message.metadata.?); try testing.expect(got.message.usage != null); try testing.expectEqual(@as(u64, 1500), got.message.usage.?.input); try testing.expectEqual(@as(u64, 85), got.message.usage.?.output); @@ -944,8 +997,8 @@ test "serialize/parse assistant message entry with metadata" { test "serialize/parse tool result message entry" { const a = testing.allocator; - var content = try a.alloc(DiskContentBlock, 1); - var trp = try a.alloc(DiskResultPart, 1); + var content = try a.alloc(StoredContentBlock, 1); + var trp = try a.alloc(StoredResultPart, 1); trp[0] = .{ .text = try dupe(a, "file1.txt\nfile2.txt") }; content[0] = .{ .tool_result = .{ .tool_use_id = try dupe(a, "tool_abc"), @@ -958,8 +1011,11 @@ test "serialize/parse tool result message entry" { .parent_id = try dupe(a, "b2c3d4e5"), .timestamp = try dupe(a, "2026-04-25T17:40:18.000Z"), }, - .provider = try dupe(a, "anthropic"), - .model = try dupe(a, "claude-sonnet-4-20250514"), + .stamp = .{ + .api_style = .anthropic_messages, + .base_url = try dupe(a, "https://api.anthropic.com"), + .model = try dupe(a, "claude-sonnet-4-20250514"), + }, .message = .{ .role = .user, .content = content, @@ -973,11 +1029,11 @@ test "serialize/parse tool result message entry" { 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.expectEqual(StoredMessageRole.user, got.message.role); try testing.expectEqualStrings("tool_abc", got.message.content[0].tool_result.tool_use_id); try testing.expectEqual(@as(usize, 1), got.message.content[0].tool_result.parts.len); try testing.expectEqualStrings("file1.txt\nfile2.txt", got.message.content[0].tool_result.parts[0].text); - try testing.expectEqualStrings("anthropic", got.provider.?); + try testing.expectEqual(APIStyle.anthropic_messages, got.stamp.?.api_style); // Unset is_error defaults to false and serializes without the field. try testing.expect(!got.message.content[0].tool_result.is_error); try testing.expect(std.mem.indexOf(u8, line, "isError") == null); @@ -986,8 +1042,8 @@ test "serialize/parse tool result message entry" { test "serialize/parse tool result preserves is_error = true" { const a = testing.allocator; - var content = try a.alloc(DiskContentBlock, 1); - var trp = try a.alloc(DiskResultPart, 1); + var content = try a.alloc(StoredContentBlock, 1); + var trp = try a.alloc(StoredResultPart, 1); trp[0] = .{ .text = try dupe(a, "file not found") }; content[0] = .{ .tool_result = .{ .tool_use_id = try dupe(a, "tool_err"), @@ -1001,8 +1057,11 @@ test "serialize/parse tool result preserves is_error = true" { .parent_id = try dupe(a, "e0"), .timestamp = try dupe(a, "2026-04-25T17:40:18.000Z"), }, - .provider = try dupe(a, "anthropic"), - .model = try dupe(a, "claude-sonnet-4-20250514"), + .stamp = .{ + .api_style = .anthropic_messages, + .base_url = try dupe(a, "https://api.anthropic.com"), + .model = try dupe(a, "claude-sonnet-4-20250514"), + }, .message = .{ .role = .user, .content = content }, } }; defer entry.deinit(a); @@ -1030,8 +1089,8 @@ test "parse tool result without isError defaults to false" { test "serialize/parse tool result with text + image part round-trips" { const a = testing.allocator; - var content = try a.alloc(DiskContentBlock, 1); - var trp = try a.alloc(DiskResultPart, 2); + var content = try a.alloc(StoredContentBlock, 1); + var trp = try a.alloc(StoredResultPart, 2); trp[0] = .{ .text = try dupe(a, "here is the image") }; trp[1] = .{ .media = .{ .media_type = try dupe(a, "image/png"), @@ -1048,8 +1107,11 @@ test "serialize/parse tool result with text + image part round-trips" { .parent_id = try dupe(a, "img00000"), .timestamp = try dupe(a, "2026-04-25T17:40:18.000Z"), }, - .provider = try dupe(a, "anthropic"), - .model = try dupe(a, "claude-sonnet-4-20250514"), + .stamp = .{ + .api_style = .anthropic_messages, + .base_url = try dupe(a, "https://api.anthropic.com"), + .model = try dupe(a, "claude-sonnet-4-20250514"), + }, .message = .{ .role = .user, .content = content }, } }; defer entry.deinit(a); @@ -1072,7 +1134,7 @@ test "system message mode round-trips; absent mode defaults to append" { // replace-mode system entry round-trips. { - var content = try a.alloc(DiskContentBlock, 1); + var content = try a.alloc(StoredContentBlock, 1); content[0] = .{ .text = .{ .text = try dupe(a, "fresh seed") } }; const entry: SessionEntry = .{ .message = .{ .base = .{ @@ -1094,7 +1156,7 @@ test "system message mode round-trips; absent mode defaults to append" { var fe = try parseLine(a, line); defer fe.deinit(a); - try testing.expectEqual(DiskSystemMode.replace, fe.entry.message.message.mode); + try testing.expectEqual(StoredSystemMode.replace, fe.entry.message.message.mode); } // A legacy system entry with no `mode` parses back as append. @@ -1104,7 +1166,7 @@ test "system message mode round-trips; absent mode defaults to append" { ; var fe = try parseLine(a, line); defer fe.deinit(a); - try testing.expectEqual(DiskSystemMode.append, fe.entry.message.message.mode); + try testing.expectEqual(StoredSystemMode.append, fe.entry.message.message.mode); } } @@ -1147,7 +1209,7 @@ test "contentBlockToDisk: Text round-trips via in-memory" { test "diskContentBlockToInternal: ToolUse preserves id/name/input" { const a = testing.allocator; - const disk: DiskContentBlock = .{ .tool_use = .{ + const disk: StoredContentBlock = .{ .tool_use = .{ .id = try a.dupe(u8, "tu_1"), .name = try a.dupe(u8, "bash"), .input = try a.dupe(u8, "{\"command\":\"ls\"}"), @@ -1164,7 +1226,7 @@ test "diskContentBlockToInternal: ToolUse preserves id/name/input" { test "Usage: all five fields round-trip; zero-valued fields omitted from JSON" { const a = testing.allocator; - var content = try a.alloc(DiskContentBlock, 1); + var content = try a.alloc(StoredContentBlock, 1); content[0] = .{ .text = .{ .text = try dupe(a, "hi") } }; const entry: SessionEntry = .{ .message = .{ @@ -1176,8 +1238,6 @@ test "Usage: all five fields round-trip; zero-valued fields omitted from JSON" { .message = .{ .role = .assistant, .content = content, - .provider = try dupe(a, "anthropic"), - .model = try dupe(a, "claude-sonnet-4-20250514"), .stop_reason = try dupe(a, "stop"), .usage = .{ .input = 100, @@ -1213,7 +1273,7 @@ test "Usage: all five fields round-trip; zero-valued fields omitted from JSON" { test "Usage: zero-valued auxiliary fields are omitted but parse back as 0" { const a = testing.allocator; - var content = try a.alloc(DiskContentBlock, 1); + var content = try a.alloc(StoredContentBlock, 1); content[0] = .{ .text = .{ .text = try dupe(a, "hi") } }; const entry: SessionEntry = .{ .message = .{ @@ -1248,7 +1308,7 @@ test "Usage: zero-valued auxiliary fields are omitted but parse back as 0" { test "diskContentBlockToInternal: Thinking preserves signature" { const a = testing.allocator; - const disk: DiskContentBlock = .{ .thinking = .{ + const disk: StoredContentBlock = .{ .thinking = .{ .thinking = try a.dupe(u8, "reasoning..."), .signature = try a.dupe(u8, "sig123"), } }; @@ -1263,7 +1323,7 @@ test "diskContentBlockToInternal: Thinking preserves signature" { test "compactionSummary block round-trips through serialize/parse" { const a = testing.allocator; - var content = try a.alloc(DiskContentBlock, 1); + var content = try a.alloc(StoredContentBlock, 1); content[0] = .{ .compaction_summary = .{ .text = try dupe(a, "earlier history summary") } }; const entry: SessionEntry = .{ .message = .{ @@ -1283,7 +1343,7 @@ test "compactionSummary block round-trips through serialize/parse" { 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.expectEqual(StoredMessageRole.user, got.message.role); try testing.expectEqualStrings("earlier history summary", got.message.content[0].compaction_summary.text); } diff --git a/libpanto/src/session_manager.zig b/libpanto/src/session_manager.zig deleted file mode 100644 index b659506..0000000 --- a/libpanto/src/session_manager.zig +++ /dev/null @@ -1,1886 +0,0 @@ -//! Session lifecycle: create, open, replay, append. -//! -//! Backed by an append-only JSONL file on disk. The on-disk types live in -//! `session.zig`. This module owns: -//! -//! - Path resolution (sessions dir is supplied by the caller; we own the -//! filename and writes within it). -//! - The in-memory entry index (`by_id` map + leaf pointer). -//! - Deferred file creation: the file is not written until the first -//! assistant message persists. Until that point, all entries are -//! buffered in memory. -//! - Append semantics: once flushed, every completed entry is written -//! and synced to disk immediately. -//! - Crash recovery: on open, the file is parsed line-by-line; the first -//! line that fails to parse causes everything from that line onward to -//! be truncated from the file. -//! - One-time format migration when a future version reads a v1 file -//! (currently a no-op; the hook is in place). -//! - Rebuilding a `Conversation` from the entry tree, plus determining -//! the active provider/model. -//! -//! The library-vs-CLI boundary: callers pass an absolute path to the -//! per-cwd sessions directory. We compute the per-session filename -//! ourselves (`.jsonl`) and lazily mkdir the directory on the -//! first flush. The CLI owns XDG resolution, encoded-cwd grouping, and -//! the `--resume` flag plumbing. - -const std = @import("std"); -const Allocator = std.mem.Allocator; -const Io = std.Io; - -const session_mod = @import("session.zig"); -const conversation_mod = @import("conversation.zig"); -const session_store_mod = @import("session_store.zig"); - -pub const SessionHeader = session_mod.SessionHeader; -pub const SessionEntry = session_mod.SessionEntry; -pub const MessageEntry = session_mod.MessageEntry; -pub const DiskMessage = session_mod.DiskMessage; -pub const DiskMessageRole = session_mod.DiskMessageRole; -pub const DiskSystemMode = session_mod.DiskSystemMode; -pub const DiskContentBlock = session_mod.DiskContentBlock; -pub const Usage = session_mod.Usage; -pub const CURRENT_VERSION = session_mod.CURRENT_VERSION; - -// ============================================================================= -// IDs and timestamps -// ============================================================================= - -/// Generate a UUIDv7 (RFC 9562 §5.7). Returns a 36-character canonical -/// hex string with hyphens. Caller owns. -/// -/// Layout: -/// - 48 bits: unix_ts_ms (big-endian) -/// - 4 bits: version (7) -/// - 12 bits: random -/// - 2 bits: variant (10) -/// - 62 bits: random -pub fn newUuidV7(allocator: Allocator, io: Io) ![]u8 { - const ts = Io.Timestamp.now(io, .real); - const now_ms: u64 = @intCast(@max(ts.toMilliseconds(), 0)); - - var rand_bytes: [10]u8 = undefined; - io.random(&rand_bytes); - - var b: [16]u8 = undefined; - // Timestamp (48 bits, big-endian). - b[0] = @intCast((now_ms >> 40) & 0xFF); - b[1] = @intCast((now_ms >> 32) & 0xFF); - b[2] = @intCast((now_ms >> 24) & 0xFF); - b[3] = @intCast((now_ms >> 16) & 0xFF); - b[4] = @intCast((now_ms >> 8) & 0xFF); - b[5] = @intCast(now_ms & 0xFF); - // Version (4 high bits = 0x7) + 12 bits random. - b[6] = 0x70 | (rand_bytes[0] & 0x0F); - b[7] = rand_bytes[1]; - // Variant (2 high bits = 10) + 62 bits random. - b[8] = 0x80 | (rand_bytes[2] & 0x3F); - b[9] = rand_bytes[3]; - b[10] = rand_bytes[4]; - b[11] = rand_bytes[5]; - b[12] = rand_bytes[6]; - b[13] = rand_bytes[7]; - b[14] = rand_bytes[8]; - b[15] = rand_bytes[9]; - - return try std.fmt.allocPrint( - allocator, - "{x:0>2}{x:0>2}{x:0>2}{x:0>2}-{x:0>2}{x:0>2}-{x:0>2}{x:0>2}-{x:0>2}{x:0>2}-{x:0>2}{x:0>2}{x:0>2}{x:0>2}{x:0>2}{x:0>2}", - .{ b[0], b[1], b[2], b[3], b[4], b[5], b[6], b[7], b[8], b[9], b[10], b[11], b[12], b[13], b[14], b[15] }, - ); -} - -/// Generate a fresh 8-character hex entry id. Caller owns. -fn newEntryIdInto(buf: []u8, io: Io) void { - std.debug.assert(buf.len == 8); - var bytes: [4]u8 = undefined; - io.random(&bytes); - _ = std.fmt.bufPrint(buf, "{x:0>2}{x:0>2}{x:0>2}{x:0>2}", .{ bytes[0], bytes[1], bytes[2], bytes[3] }) catch unreachable; -} - -/// Format `now` as an ISO 8601 UTC string with millisecond precision. -/// Example: `2026-04-25T17:40:15.990Z`. Caller owns. -pub fn isoTimestamp(allocator: Allocator, io: Io) ![]u8 { - const ts = Io.Timestamp.now(io, .real); - const ms_total: i64 = ts.toMilliseconds(); - const seconds_total: i64 = @divTrunc(ms_total, 1000); - const ms: u64 = @intCast(@mod(ms_total, 1000)); - - const epoch_secs = std.time.epoch.EpochSeconds{ .secs = @intCast(seconds_total) }; - const epoch_day = epoch_secs.getEpochDay(); - const day_secs = epoch_secs.getDaySeconds(); - const year_day = epoch_day.calculateYearDay(); - const month_day = year_day.calculateMonthDay(); - - return try std.fmt.allocPrint( - allocator, - "{d:0>4}-{d:0>2}-{d:0>2}T{d:0>2}:{d:0>2}:{d:0>2}.{d:0>3}Z", - .{ - @as(u32, year_day.year), - month_day.month.numeric(), - @as(u32, month_day.day_index) + 1, - day_secs.getHoursIntoDay(), - day_secs.getMinutesIntoHour(), - day_secs.getSecondsIntoMinute(), - ms, - }, - ); -} - -// ============================================================================= -// SessionInfo (listing) -// ============================================================================= - -pub const SessionInfo = struct { - path: []u8, - id: []u8, - cwd: []u8, - created: []u8, // ISO 8601 from header timestamp - modified: []u8, // ISO 8601 from last user/assistant entry, falling back to header, then file mtime - message_count: usize, - - pub fn deinit(self: SessionInfo, alloc: Allocator) void { - alloc.free(self.path); - alloc.free(self.id); - alloc.free(self.cwd); - alloc.free(self.created); - alloc.free(self.modified); - } -}; - -pub fn freeSessionInfos(alloc: Allocator, infos: []SessionInfo) void { - for (infos) |info| info.deinit(alloc); - alloc.free(infos); -} - -// ============================================================================= -// SessionManager -// ============================================================================= - -pub const Error = error{ - NoSessionsFound, - AmbiguousSessionId, - SessionNotFound, - InvalidSessionFile, -} || Allocator.Error || Io.Cancelable; - -pub const SessionManager = struct { - allocator: Allocator, - io: Io, - - /// Absolute path to the per-cwd sessions directory. Lazily created. - session_dir: []u8, - /// Absolute path to the file we *will* write to (computed at init). - /// May not yet exist on disk if `flushed = false`. - session_file: []u8, - - /// Header. Allocated at init for new sessions; reloaded from the file - /// on resume. - header: SessionHeader, - - /// Entries indexed in insertion order. The first entry's `parent_id` - /// is null; each subsequent entry's `parent_id` points to its parent - /// (currently always the previous entry). - entries: std.ArrayList(SessionEntry), - /// id → entry index in `entries`. Used both for parent-id lookups - /// and for collision detection in `newEntryId`. - by_id: std.StringHashMap(usize), - /// id of the most recently appended entry, or null if no entries yet. - /// Borrowed from the entry; do not free. - leaf_id: ?[]const u8, - - /// True once the file exists on disk. False during the "buffered" - /// pre-assistant phase. See module-level docs. - flushed: bool, - /// Number of bytes written to `session_file` so far. Used as the - /// offset for the next positional write. Only meaningful when - /// `flushed = true`. - written_bytes: u64, - - // ---------- Construction ---------- - - /// Create a new session in memory. Allocates a UUIDv7, computes the - /// file path, but does NOT touch the filesystem. The file is created - /// on the first assistant-message flush. - /// - /// `session_dir` is duplicated; the caller retains ownership of the - /// passed slice. - pub fn init( - allocator: Allocator, - io: Io, - session_dir: []const u8, - cwd: []const u8, - ) !SessionManager { - const dir = try allocator.dupe(u8, session_dir); - errdefer allocator.free(dir); - - const id = try newUuidV7(allocator, io); - errdefer allocator.free(id); - - const timestamp = try isoTimestamp(allocator, io); - errdefer allocator.free(timestamp); - - const cwd_copy = try allocator.dupe(u8, cwd); - errdefer allocator.free(cwd_copy); - - const filename = try std.fmt.allocPrint(allocator, "{s}.jsonl", .{id}); - defer allocator.free(filename); - const file_path = try std.fs.path.join(allocator, &.{ dir, filename }); - errdefer allocator.free(file_path); - - return .{ - .allocator = allocator, - .io = io, - .session_dir = dir, - .session_file = file_path, - .header = .{ - .version = CURRENT_VERSION, - .id = id, - .timestamp = timestamp, - .cwd = cwd_copy, - }, - .entries = .empty, - .by_id = std.StringHashMap(usize).init(allocator), - .leaf_id = null, - .flushed = false, - .written_bytes = 0, - }; - } - - /// Open and replay an existing session file. Truncates from the first - /// corrupted line. Runs format migration if needed and rewrites the - /// file once. - pub fn open( - allocator: Allocator, - io: Io, - file_path: []const u8, - ) !SessionManager { - const path_copy = try allocator.dupe(u8, file_path); - errdefer allocator.free(path_copy); - - // The session dir is the file's parent directory. - const dir_path = std.fs.path.dirname(path_copy) orelse "."; - const dir = try allocator.dupe(u8, dir_path); - errdefer allocator.free(dir); - - const bytes = try readWholeFile(allocator, io, path_copy); - defer allocator.free(bytes); - - // Walk line-by-line. The first failure causes a truncation back to - // the start of that line. - var entries: std.ArrayList(SessionEntry) = .empty; - errdefer { - for (entries.items) |e| e.deinit(allocator); - entries.deinit(allocator); - } - var by_id = std.StringHashMap(usize).init(allocator); - errdefer by_id.deinit(); - - var header_opt: ?SessionHeader = null; - errdefer if (header_opt) |h| h.deinit(allocator); - - var cursor: usize = 0; - var valid_bytes: u64 = 0; // length of the file prefix that parses cleanly - var saw_corruption: bool = false; - - while (cursor < bytes.len) { - // Find the next newline (or EOF). - const rest = bytes[cursor..]; - const nl_rel = std.mem.indexOfScalar(u8, rest, '\n'); - const line_end_excl: usize = if (nl_rel) |n| cursor + n else bytes.len; - const line = bytes[cursor..line_end_excl]; - const next_cursor: usize = if (nl_rel != null) line_end_excl + 1 else bytes.len; - - // Allow blank lines silently (just whitespace), but a non-empty - // trimmed line that won't parse triggers truncation. - const trimmed = std.mem.trim(u8, line, " \t\r"); - if (trimmed.len == 0) { - if (nl_rel == null) break; - cursor = next_cursor; - valid_bytes = cursor; - continue; - } - - // If the final line has no trailing newline AND we hit EOF, it - // is presumed truncated mid-write. Treat as corruption. - if (nl_rel == null) { - saw_corruption = true; - break; - } - - const fe = session_mod.parseLine(allocator, line) catch { - saw_corruption = true; - break; - }; - - switch (fe) { - .header => |h| { - if (header_opt != null) { - // Two headers — treat as corruption from this line on. - h.deinit(allocator); - saw_corruption = true; - break; - } - if (entries.items.len != 0) { - // Header arrived after entries — malformed. - h.deinit(allocator); - saw_corruption = true; - break; - } - header_opt = h; - }, - .entry => |e| { - const idx = entries.items.len; - entries.append(allocator, e) catch |err| { - e.deinit(allocator); - return err; - }; - by_id.put(e.base().id, idx) catch |err| { - // Rolling back the append is awkward; in practice - // OOM here is fatal anyway. - return err; - }; - }, - } - - cursor = next_cursor; - valid_bytes = cursor; - } - - // No header at all — refuse to load. - const header = header_opt orelse return error.InvalidSessionFile; - - // Truncate the file if anything beyond `valid_bytes` is corrupt. - if (saw_corruption and valid_bytes < bytes.len) { - try truncateFileTo(io, path_copy, valid_bytes); - } - - var migrated_header = header; - var did_migrate = migrate(allocator, &migrated_header, &entries); - if (elideDanglingToolUses(allocator, &entries)) { - did_migrate = true; - } - // We didn't reassign by_id during migration; rebuild if needed. - if (did_migrate) { - by_id.clearRetainingCapacity(); - for (entries.items, 0..) |e, i| { - try by_id.put(e.base().id, i); - } - try rewriteFile(allocator, io, path_copy, migrated_header, entries.items); - } - - const leaf_id: ?[]const u8 = if (entries.items.len > 0) - entries.items[entries.items.len - 1].base().id - else - null; - - // Compute final file length on disk so future appends use the - // correct offset. - const stat = try statFileForLength(io, path_copy); - - return .{ - .allocator = allocator, - .io = io, - .session_dir = dir, - .session_file = path_copy, - .header = migrated_header, - .entries = entries, - .by_id = by_id, - .leaf_id = leaf_id, - .flushed = true, - .written_bytes = stat, - }; - } - - pub fn deinit(self: *SessionManager) void { - self.header.deinit(self.allocator); - for (self.entries.items) |e| e.deinit(self.allocator); - self.entries.deinit(self.allocator); - self.by_id.deinit(); - self.allocator.free(self.session_dir); - self.allocator.free(self.session_file); - } - - // ---------- Accessors ---------- - - pub fn getCwd(self: *const SessionManager) []const u8 { - return self.header.cwd; - } - - pub fn getSessionId(self: *const SessionManager) []const u8 { - return self.header.id; - } - - pub fn getSessionFile(self: *const SessionManager) []const u8 { - return self.session_file; - } - - pub fn getSessionDir(self: *const SessionManager) []const u8 { - return self.session_dir; - } - - pub fn getLeafId(self: *const SessionManager) ?[]const u8 { - return self.leaf_id; - } - - pub fn getEntry(self: *const SessionManager, id: []const u8) ?*const SessionEntry { - const idx = self.by_id.get(id) orelse return null; - return &self.entries.items[idx]; - } - - pub fn getEntries(self: *const SessionManager) []const SessionEntry { - return self.entries.items; - } - - pub fn isFlushed(self: *const SessionManager) bool { - return self.flushed; - } - - // ---------- Active model resolution ---------- - - /// Determine the active provider/model by walking entries leaf→root - /// and finding the last user-message entry with provider/model - /// stamped. Returns null only when no user message has been appended - /// yet, which is only reachable on a freshly-`init`'d session before - /// the first user prompt (and therefore before any disk flush). - /// - /// Returns borrowed slices owned by the manager; do not free. - pub fn activeModel(self: *const SessionManager) ?struct { provider: []const u8, model: []const u8 } { - var i = self.entries.items.len; - while (i > 0) : (i -= 1) { - const e = self.entries.items[i - 1]; - switch (e) { - .message => |m| { - if (m.message.role == .user) { - if (m.provider) |p| { - if (m.model) |mo| { - return .{ .provider = p, .model = mo }; - } - } - } - }, - } - } - return null; - } - - // ---------- Appending ---------- - - /// Append a message entry. `msg` is consumed (ownership transferred) - /// regardless of success — on error, the message is deinit'd before - /// the error is returned. - /// - /// If `flushed`: writes the new line immediately. - /// If not flushed and `msg.role == .assistant`: writes the header + - /// all buffered entries + the new entry, then sets `flushed`. - /// Otherwise: buffers in memory only. - pub fn appendMessage( - self: *SessionManager, - msg: DiskMessage, - // Top-level stamps on the entry (vs. inside the message). - // Stamped only on user messages. - provider: ?[]const u8, - model: ?[]const u8, - ) ![]const u8 { - // Build the entry up-front, taking ownership of the inputs. - var msg_local = msg; - errdefer msg_local.deinit(self.allocator); - - const id_buf = try self.newEntryId(); - errdefer self.allocator.free(id_buf); - - const timestamp = try isoTimestamp(self.allocator, self.io); - errdefer self.allocator.free(timestamp); - - const parent_id_copy: ?[]const u8 = if (self.leaf_id) |l| try self.allocator.dupe(u8, l) else null; - errdefer if (parent_id_copy) |p| self.allocator.free(p); - - const provider_copy: ?[]const u8 = if (provider) |p| try self.allocator.dupe(u8, p) else null; - errdefer if (provider_copy) |p| self.allocator.free(p); - - const model_copy: ?[]const u8 = if (model) |m| try self.allocator.dupe(u8, m) else null; - errdefer if (model_copy) |m| self.allocator.free(m); - - const entry: SessionEntry = .{ .message = .{ - .base = .{ .id = id_buf, .parent_id = parent_id_copy, .timestamp = timestamp }, - .provider = provider_copy, - .model = model_copy, - .message = msg_local, - } }; - - // The entry now owns msg_local + id_buf + timestamp + parent_id_copy + - // provider/model copies. Cancel the errdefers individually. - // (Zig's errdefer behavior: they only run on error returns; pushing the - // entry into entries.items before any further fallible step means an - // error in by_id.put() would double-free. Instead, do the put first - // against a not-yet-stored id.) - - const idx = self.entries.items.len; - - // Ensure capacity before touching anything. - try self.entries.ensureUnusedCapacity(self.allocator, 1); - try self.by_id.ensureUnusedCapacity(1); - - // Persist BEFORE inserting into the in-memory structures, so that on - // I/O failure we don't have a dangling in-memory entry the caller - // thinks was saved. (Failure leaves the file unchanged for an - // unflushed session, and unchanged-except-for-EOF for a flushed one.) - const is_assistant = entry.message.message.role == .assistant; - if (self.flushed) { - try self.persistEntry(entry); - } else if (is_assistant) { - try self.flushBuffered(entry); - } - // If not flushed and not assistant: nothing to do; the entry will be - // flushed alongside the eventual first assistant entry. - - // Now insert into in-memory structures. All allocations are kept. - self.entries.appendAssumeCapacity(entry); - self.by_id.putAssumeCapacity(entry.base().id, idx); - self.leaf_id = entry.base().id; - return entry.base().id; - } - - /// Returns a freshly allocated 8-character hex id, guaranteed not to - /// collide with any existing entry id in this session. - fn newEntryId(self: *SessionManager) ![]u8 { - const max_tries = 100; - var i: usize = 0; - while (i < max_tries) : (i += 1) { - const buf = try self.allocator.alloc(u8, 8); - errdefer self.allocator.free(buf); - newEntryIdInto(buf[0..8], self.io); - if (!self.by_id.contains(buf)) { - return buf; - } - self.allocator.free(buf); - } - // Fall back to a UUID prefix if 100 retries all collided. With 4 - // random bytes per id and a session with <<2^16 entries, the - // probability of getting here is effectively zero, but we want a - // hard guarantee. - const long = try newUuidV7(self.allocator, self.io); - defer self.allocator.free(long); - const buf = try self.allocator.alloc(u8, 8); - @memcpy(buf, long[0..8]); - return buf; - } - - // ---------- Persistence ---------- - - /// Write the header + all currently-buffered entries + `new_entries` - /// to the file as a single batch. Creates the directory and file. - fn flushBufferedMany(self: *SessionManager, new_entries: []const SessionEntry) !void { - try mkdirP(self.io, self.session_dir); - - const file = try Io.Dir.cwd().createFile(self.io, self.session_file, .{ - .truncate = true, - .read = false, - }); - defer file.close(self.io); - - var offset: u64 = 0; - const header_line = try session_mod.serializeHeader(self.allocator, self.header); - defer self.allocator.free(header_line); - try file.writePositionalAll(self.io, header_line, offset); - offset += header_line.len; - try file.writePositionalAll(self.io, "\n", offset); - offset += 1; - - for (self.entries.items) |e| { - const line = try session_mod.serializeEntry(self.allocator, e); - defer self.allocator.free(line); - try file.writePositionalAll(self.io, line, offset); - offset += line.len; - try file.writePositionalAll(self.io, "\n", offset); - offset += 1; - } - - for (new_entries) |entry| { - const line = try session_mod.serializeEntry(self.allocator, entry); - defer self.allocator.free(line); - try file.writePositionalAll(self.io, line, offset); - offset += line.len; - try file.writePositionalAll(self.io, "\n", offset); - offset += 1; - } - - file.sync(self.io) catch {}; - self.flushed = true; - self.written_bytes = offset; - } - - fn flushBuffered(self: *SessionManager, final_entry: SessionEntry) !void { - try self.flushBufferedMany(&.{final_entry}); - } - - /// Append a single line for `entry` to the open session file. Caller - /// must have already verified `flushed`. - fn persistEntry(self: *SessionManager, entry: SessionEntry) !void { - try self.persistEntries(&.{entry}); - } - - pub fn appendMessagesAtomic( - self: *SessionManager, - messages: []DiskMessage, - providers: []const ?[]const u8, - models: []const ?[]const u8, - ) !void { - std.debug.assert(messages.len == providers.len); - std.debug.assert(messages.len == models.len); - if (messages.len == 0) return; - - const base_len = self.entries.items.len; - try self.entries.ensureUnusedCapacity(self.allocator, messages.len); - try self.by_id.ensureUnusedCapacity(@intCast(messages.len)); - - var entries = try self.allocator.alloc(SessionEntry, messages.len); - defer self.allocator.free(entries); - var built: usize = 0; - errdefer { - for (entries[0..built]) |*e| e.deinit(self.allocator); - } - - var prev_leaf = self.leaf_id; - for (messages, 0..) |msg, i| { - const msg_local = msg; - const id_buf = try self.newEntryId(); - errdefer self.allocator.free(id_buf); - const timestamp = try isoTimestamp(self.allocator, self.io); - errdefer self.allocator.free(timestamp); - const parent_id_copy: ?[]const u8 = if (prev_leaf) |l| try self.allocator.dupe(u8, l) else null; - errdefer if (parent_id_copy) |p| self.allocator.free(p); - const provider_copy: ?[]const u8 = if (providers[i]) |p| try self.allocator.dupe(u8, p) else null; - errdefer if (provider_copy) |p| self.allocator.free(p); - const model_copy: ?[]const u8 = if (models[i]) |m| try self.allocator.dupe(u8, m) else null; - errdefer if (model_copy) |m| self.allocator.free(m); - entries[i] = .{ .message = .{ - .base = .{ .id = id_buf, .parent_id = parent_id_copy, .timestamp = timestamp }, - .provider = provider_copy, - .model = model_copy, - .message = msg_local, - } }; - built += 1; - prev_leaf = entries[i].base().id; - } - - if (self.flushed) { - try self.persistEntries(entries); - } else { - try self.flushBufferedMany(entries); - } - - for (entries, 0..) |entry, i| { - self.entries.appendAssumeCapacity(entry); - self.by_id.putAssumeCapacity(entry.base().id, base_len + i); - } - self.leaf_id = entries[entries.len - 1].base().id; - built = 0; - } - - fn persistEntries(self: *SessionManager, entries: []const SessionEntry) !void { - const file = try Io.Dir.cwd().openFile(self.io, self.session_file, .{ - .mode = .write_only, - }); - defer file.close(self.io); - - var offset = self.written_bytes; - for (entries) |entry| { - const line = try session_mod.serializeEntry(self.allocator, entry); - defer self.allocator.free(line); - try file.writePositionalAll(self.io, line, offset); - offset += line.len; - try file.writePositionalAll(self.io, "\n", offset); - offset += 1; - } - file.sync(self.io) catch {}; - self.written_bytes = offset; - } - - // ============================================================================= - // Conversation rebuild - // ============================================================================= - - /// Build a fresh `Conversation` from the entry log. Caller owns the - /// returned conversation (call `deinit`). - pub fn rebuildConversation(self: *const SessionManager) !conversation_mod.Conversation { - var conv = conversation_mod.Conversation.init(self.allocator); - errdefer conv.deinit(); - - for (self.entries.items) |entry| { - switch (entry) { - .message => |me| try appendMessageToConv(&conv, self.allocator, me.message), - } - } - return conv; - } - - // ============================================================================= - // SessionStore interface - // ============================================================================= - - /// Reconstruct a linear `Conversation` plus an optional dangling - /// trailing user prompt (a user entry with no following assistant — - /// e.g. a crash/quit right after submission). The returned conversation - /// **excludes** that dangling user turn; its text is returned as - /// `dangling_user` (owned by `alloc`) so a resumed agent never - /// auto-sends it. - pub fn loadConversation( - self: *const SessionManager, - alloc: Allocator, - ) !session_store_mod.LoadedSession { - // Detect a dangling trailing user entry: the last message entry is - // a user message and nothing follows it. (Entries are linear; the - // last entry is the leaf.) - const items = self.entries.items; - var dangling: ?[]const u8 = null; - var stop_at: usize = items.len; - if (items.len > 0) { - const last = items[items.len - 1]; - switch (last) { - .message => |me| { - if (me.message.role == .user) { - dangling = try extractUserText(alloc, me.message); - // Only treat it as dangling (and exclude it) when we - // could actually recover prompt text; otherwise keep - // it in the conversation rather than silently drop. - if (dangling != null) stop_at = items.len - 1; - } - }, - } - } - errdefer if (dangling) |d| alloc.free(d); - - var conv = conversation_mod.Conversation.init(alloc); - errdefer conv.deinit(); - for (items[0..stop_at]) |entry| { - switch (entry) { - .message => |me| try appendMessageToConv(&conv, alloc, me.message), - } - } - - return .{ .conversation = conv, .dangling_user = dangling }; - } - - fn appendMessagesVT( - ctx: *anyopaque, - messages: []session_store_mod.DiskMessage, - providers: []const ?[]const u8, - models: []const ?[]const u8, - ) anyerror!void { - const self: *SessionManager = @ptrCast(@alignCast(ctx)); - try self.appendMessagesAtomic(messages, providers, models); - } - - fn loadConversationVT( - ctx: *anyopaque, - alloc: Allocator, - ) anyerror!session_store_mod.LoadedSession { - const self: *const SessionManager = @ptrCast(@alignCast(ctx)); - return self.loadConversation(alloc); - } - - fn sessionIdVT(ctx: *anyopaque) []const u8 { - const self: *const SessionManager = @ptrCast(@alignCast(ctx)); - return self.getSessionId(); - } - - fn activeModelVT(ctx: *anyopaque) ?session_store_mod.ActiveModel { - const self: *const SessionManager = @ptrCast(@alignCast(ctx)); - const am = self.activeModel() orelse return null; - return .{ .provider = am.provider, .model = am.model }; - } - - const store_vtable: session_store_mod.SessionStore.VTable = .{ - .appendMessages = appendMessagesVT, - .loadConversation = loadConversationVT, - .sessionId = sessionIdVT, - .activeModel = activeModelVT, - }; - - /// Wrap this concrete manager as a neutral `SessionStore`. The returned - /// store borrows `self`; `self` must outlive it. - pub fn store(self: *SessionManager) session_store_mod.SessionStore { - return .{ .ptr = self, .vtable = &store_vtable }; - } -}; - -/// Best-effort extraction of plain prompt text from a user `DiskMessage`. -/// Returns null if the message carries no plain text block (e.g. it is a -/// tool-result-only user message, which is never a "dangling prompt"). -/// Caller owns the returned slice. -fn extractUserText(alloc: Allocator, msg: DiskMessage) !?[]const u8 { - for (msg.content) |block| { - if (block == .text) { - return try alloc.dupe(u8, block.text.text); - } - } - return null; -} - -fn appendMessageToConv( - conv: *conversation_mod.Conversation, - allocator: Allocator, - disk_msg: DiskMessage, -) !void { - var content: std.ArrayList(conversation_mod.ContentBlock) = .empty; - errdefer { - for (content.items) |*b| { - var mut = b.*; - mut.deinit(allocator); - } - content.deinit(allocator); - } - try content.ensureTotalCapacity(allocator, disk_msg.content.len); - const sys_mode: conversation_mod.SystemMode = switch (disk_msg.mode) { - .append => .append, - .replace => .replace, - }; - for (disk_msg.content) |db| { - var block = try session_mod.diskContentBlockToInternal(allocator, db); - // System-role text blocks become `.System` blocks carrying the - // message's recorded mode, so the append/replace derivation works - // on the rebuilt conversation exactly as it did when written. - if (disk_msg.role == .system and block == .Text) { - const tb = block.Text; - block = .{ .System = .{ .text = tb, .mode = sys_mode } }; - } - content.appendAssumeCapacity(block); - } - const role: conversation_mod.MessageRole = switch (disk_msg.role) { - .system => .system, - .user => .user, - .assistant => .assistant, - }; - // 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, - }); -} - -// ============================================================================= -// Migration -// ============================================================================= - -/// Future format migrations land here. Returns true if anything changed -/// (which triggers a one-time file rewrite). -fn migrate( - allocator: Allocator, - header: *SessionHeader, - entries: *std.ArrayList(SessionEntry), -) bool { - _ = allocator; - _ = entries; - if (header.version >= CURRENT_VERSION) return false; - // No earlier versions exist yet. When v2 lands, transform v1 entries - // here and bump `header.version`. - return false; -} - -fn elideDanglingToolUses(allocator: Allocator, entries: *std.ArrayList(SessionEntry)) bool { - var needed: std.StringHashMap(void) = .init(allocator); - defer needed.deinit(); - var changed = false; - - var i = entries.items.len; - while (i > 0) { - i -= 1; - const entry = &entries.items[i]; - if (entry.* != .message) continue; - const msg = &entry.message.message; - - if (msg.role == .user) { - for (msg.content) |block| { - if (block == .tool_result) { - needed.put(block.tool_result.tool_use_id, {}) catch {}; - } - } - continue; - } - - if (msg.role != .assistant) continue; - var kept: std.ArrayList(DiskContentBlock) = .empty; - defer kept.deinit(allocator); - var removed = false; - for (msg.content) |block| { - if (block == .tool_use and !needed.contains(block.tool_use.id)) { - block.deinit(allocator); - removed = true; - continue; - } - kept.append(allocator, block) catch unreachable; - } - if (!removed) continue; - allocator.free(msg.content); - msg.content = kept.toOwnedSlice(allocator) catch unreachable; - changed = true; - } - return changed; -} - -// ============================================================================= -// File utilities -// ============================================================================= - -fn readWholeFile(allocator: Allocator, io: Io, path: []const u8) ![]u8 { - const file = Io.Dir.cwd().openFile(io, path, .{ .mode = .read_only }) catch |err| switch (err) { - error.FileNotFound => return error.InvalidSessionFile, - else => return err, - }; - defer file.close(io); - - const len = file.length(io) catch { - // Fall back to a streaming read of a reasonable upper bound. - // Sessions over ~10 MB are out of scope for phase 4. - var list: std.ArrayList(u8) = .empty; - defer list.deinit(allocator); - var chunk: [4096]u8 = undefined; - while (true) { - const n = file.readStreaming(io, &.{&chunk}) catch break; - if (n == 0) break; - try list.appendSlice(allocator, chunk[0..n]); - } - return try list.toOwnedSlice(allocator); - }; - - const buf = try allocator.alloc(u8, @intCast(len)); - errdefer allocator.free(buf); - _ = try file.readPositionalAll(io, buf, 0); - return buf; -} - -fn statFileForLength(io: Io, path: []const u8) !u64 { - const file = try Io.Dir.cwd().openFile(io, path, .{ .mode = .read_only }); - defer file.close(io); - return try file.length(io); -} - -fn truncateFileTo(io: Io, path: []const u8, new_length: u64) !void { - const file = try Io.Dir.cwd().openFile(io, path, .{ .mode = .write_only }); - defer file.close(io); - try file.setLength(io, new_length); - file.sync(io) catch {}; -} - -/// Write a fresh file containing `header` followed by `entries`. Truncates -/// any existing content. Used after a migration rewrites the format. -fn rewriteFile( - allocator: Allocator, - io: Io, - path: []const u8, - header: SessionHeader, - entries: []const SessionEntry, -) !void { - const file = try Io.Dir.cwd().createFile(io, path, .{ - .truncate = true, - .read = false, - }); - defer file.close(io); - - var offset: u64 = 0; - const header_line = try session_mod.serializeHeader(allocator, header); - defer allocator.free(header_line); - try file.writePositionalAll(io, header_line, offset); - offset += header_line.len; - try file.writePositionalAll(io, "\n", offset); - offset += 1; - for (entries) |e| { - const line = try session_mod.serializeEntry(allocator, e); - defer allocator.free(line); - try file.writePositionalAll(io, line, offset); - offset += line.len; - try file.writePositionalAll(io, "\n", offset); - offset += 1; - } - file.sync(io) catch {}; -} - -fn mkdirP(io: Io, path: []const u8) !void { - Io.Dir.cwd().createDirPath(io, path) catch |err| switch (err) { - error.PathAlreadyExists => {}, - else => return err, - }; -} - -// ============================================================================= -// Listing -// ============================================================================= - -/// List sessions in `session_dir`. Returns a slice of `SessionInfo`s -/// sorted by `modified` descending (most recent first). Caller owns the -/// slice and each `SessionInfo`. -/// -/// If the directory does not exist, returns an empty slice (no error). -/// -/// If `on_progress` is non-null, it is invoked after each file is parsed. -pub fn listSessions( - allocator: Allocator, - io: Io, - session_dir: []const u8, - on_progress: ?*const fn (loaded: usize, total: usize) void, -) ![]SessionInfo { - var dir = Io.Dir.cwd().openDir(io, session_dir, .{}) catch |err| switch (err) { - error.FileNotFound => return try allocator.alloc(SessionInfo, 0), - else => return err, - }; - defer dir.close(io); - - var names: std.ArrayList([]u8) = .empty; - defer { - for (names.items) |n| allocator.free(n); - names.deinit(allocator); - } - - var it = dir.iterate(); - while (try it.next(io)) |entry| { - if (entry.kind != .file) continue; - if (!std.mem.endsWith(u8, entry.name, ".jsonl")) continue; - const copy = try allocator.dupe(u8, entry.name); - errdefer allocator.free(copy); - try names.append(allocator, copy); - } - - var infos: std.ArrayList(SessionInfo) = .empty; - errdefer { - for (infos.items) |i| i.deinit(allocator); - infos.deinit(allocator); - } - try infos.ensureTotalCapacity(allocator, names.items.len); - - var loaded: usize = 0; - for (names.items) |name| { - const full = try std.fs.path.join(allocator, &.{ session_dir, name }); - defer allocator.free(full); - const info_opt = buildSessionInfo(allocator, io, full) catch null; - if (info_opt) |info| { - infos.appendAssumeCapacity(info); - } - loaded += 1; - if (on_progress) |cb| cb(loaded, names.items.len); - } - - const slice = try infos.toOwnedSlice(allocator); - std.sort.pdq(SessionInfo, slice, {}, sessionInfoNewerFirst); - return slice; -} - -fn sessionInfoNewerFirst(_: void, a: SessionInfo, b: SessionInfo) bool { - return std.mem.order(u8, a.modified, b.modified) == .gt; -} - -fn buildSessionInfo( - allocator: Allocator, - io: Io, - file_path: []const u8, -) !?SessionInfo { - const bytes = readWholeFile(allocator, io, file_path) catch return null; - defer allocator.free(bytes); - - var header_opt: ?SessionHeader = null; - defer if (header_opt) |h| h.deinit(allocator); - - var message_count: usize = 0; - var last_activity: ?[]u8 = null; - defer if (last_activity) |la| allocator.free(la); - - var lines = std.mem.splitScalar(u8, bytes, '\n'); - while (lines.next()) |line| { - const trimmed = std.mem.trim(u8, line, " \t\r"); - if (trimmed.len == 0) continue; - const fe = session_mod.parseLine(allocator, trimmed) catch break; - switch (fe) { - .header => |h| { - if (header_opt != null) { - h.deinit(allocator); - } else { - header_opt = h; - } - }, - .entry => |e| { - defer e.deinit(allocator); - switch (e) { - .message => |m| { - if (m.message.role == .user or m.message.role == .assistant) { - message_count += 1; - if (last_activity) |la| allocator.free(la); - last_activity = try allocator.dupe(u8, m.base.timestamp); - } - }, - } - }, - } - } - - const header = header_opt orelse return null; - // We will keep header alive through the deinit defer, and dupe its - // fields into the result. Cheap, avoids any ownership shuffling. - - const path = try allocator.dupe(u8, file_path); - errdefer allocator.free(path); - const id = try allocator.dupe(u8, header.id); - errdefer allocator.free(id); - const cwd = try allocator.dupe(u8, header.cwd); - errdefer allocator.free(cwd); - const created = try allocator.dupe(u8, header.timestamp); - errdefer allocator.free(created); - const modified = if (last_activity) |la| blk: { - last_activity = null; - break :blk la; - } else try allocator.dupe(u8, header.timestamp); - - return .{ - .path = path, - .id = id, - .cwd = cwd, - .created = created, - .modified = modified, - .message_count = message_count, - }; -} - -// ============================================================================= -// Recent / resume helpers -// ============================================================================= - -/// Find the most recent session file in `session_dir`. Returns null if -/// none exist. Caller owns the returned path. -pub fn findMostRecentSession(allocator: Allocator, io: Io, session_dir: []const u8) !?[]u8 { - var dir = Io.Dir.cwd().openDir(io, session_dir, .{}) catch |err| switch (err) { - error.FileNotFound => return null, - else => return err, - }; - defer dir.close(io); - - var best_name: ?[]u8 = null; - errdefer if (best_name) |b| allocator.free(b); - - var it = dir.iterate(); - while (try it.next(io)) |entry| { - if (entry.kind != .file) continue; - if (!std.mem.endsWith(u8, entry.name, ".jsonl")) continue; - if (best_name) |b| { - // Lexicographic compare. UUIDv7 filenames sort chronologically. - if (std.mem.order(u8, entry.name, b) == .gt) { - allocator.free(b); - best_name = try allocator.dupe(u8, entry.name); - } - } else { - best_name = try allocator.dupe(u8, entry.name); - } - } - - const name = best_name orelse return null; - defer allocator.free(name); - best_name = null; - return try std.fs.path.join(allocator, &.{ session_dir, name }); -} - -/// Resolve a (possibly abbreviated) session id to a session file path -/// within `session_dir`. Errors if no match or ambiguous prefix. -pub fn resolveSessionId( - allocator: Allocator, - io: Io, - session_dir: []const u8, - id_or_prefix: []const u8, -) ![]u8 { - var dir = Io.Dir.cwd().openDir(io, session_dir, .{}) catch |err| switch (err) { - error.FileNotFound => return error.SessionNotFound, - else => return err, - }; - defer dir.close(io); - - var match: ?[]u8 = null; - errdefer if (match) |m| allocator.free(m); - - var it = dir.iterate(); - while (try it.next(io)) |entry| { - if (entry.kind != .file) continue; - if (!std.mem.endsWith(u8, entry.name, ".jsonl")) continue; - // Strip `.jsonl` for the prefix match. - const stem = entry.name[0 .. entry.name.len - ".jsonl".len]; - if (!std.mem.startsWith(u8, stem, id_or_prefix)) continue; - if (match != null) return error.AmbiguousSessionId; - match = try allocator.dupe(u8, entry.name); - } - - const name = match orelse return error.SessionNotFound; - defer allocator.free(name); - match = null; - return try std.fs.path.join(allocator, &.{ session_dir, name }); -} - -// ============================================================================= -// Tests -// ============================================================================= - -const testing = std.testing; - -test "newUuidV7: produces 36-char hyphenated string with version 7" { - const io = testing.io; - const id = try newUuidV7(testing.allocator, io); - defer testing.allocator.free(id); - try testing.expectEqual(@as(usize, 36), id.len); - // Position 14 is the version nibble — should be '7'. - try testing.expectEqual(@as(u8, '7'), id[14]); - // Hyphens at canonical positions. - try testing.expectEqual(@as(u8, '-'), id[8]); - try testing.expectEqual(@as(u8, '-'), id[13]); - try testing.expectEqual(@as(u8, '-'), id[18]); - try testing.expectEqual(@as(u8, '-'), id[23]); -} - -test "isoTimestamp: well-formed ISO 8601 with millisecond precision" { - const ts = try isoTimestamp(testing.allocator, testing.io); - defer testing.allocator.free(ts); - try testing.expectEqual(@as(usize, 24), ts.len); - try testing.expectEqual(@as(u8, '-'), ts[4]); - try testing.expectEqual(@as(u8, 'T'), ts[10]); - try testing.expectEqual(@as(u8, '.'), ts[19]); - try testing.expectEqual(@as(u8, 'Z'), ts[23]); -} - -// ---- In-memory + filesystem tests (use a tmp dir) ---- - -const TmpSessionDir = struct { - parent: std.testing.TmpDir, - abs_path: []u8, - - fn init(allocator: Allocator) !TmpSessionDir { - var parent = std.testing.tmpDir(.{}); - errdefer parent.cleanup(); - var path_buf: [std.fs.max_path_bytes]u8 = undefined; - const n = try parent.dir.realPath(testing.io, &path_buf); - const abs = try allocator.dupe(u8, path_buf[0..n]); - return .{ .parent = parent, .abs_path = abs }; - } - - fn deinit(self: *TmpSessionDir, allocator: Allocator) void { - allocator.free(self.abs_path); - self.parent.cleanup(); - } -}; - -test "SessionManager.init: does not create file yet" { - const io = testing.io; - - var td = try TmpSessionDir.init(testing.allocator); - defer td.deinit(testing.allocator); - - // Use a non-existent subdirectory inside the tmp dir to also exercise - // lazy directory creation. - const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); - defer testing.allocator.free(sessions); - - var mgr = try SessionManager.init( - testing.allocator, - io, - sessions, - "/some/cwd", - ); - defer mgr.deinit(); - - try testing.expect(!mgr.isFlushed()); - - // The directory should not exist yet. - const stat_err = Io.Dir.cwd().openDir(io, sessions, .{}); - try testing.expectError(error.FileNotFound, stat_err); -} - -test "SessionManager: full flow — buffer, flush on assistant, append, resume" { - 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, - "/proj/foo", - ); - defer mgr.deinit(); - - // System message — non-assistant, should NOT trigger flush. - const sys_blocks = try testing.allocator.alloc(DiskContentBlock, 1); - sys_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "You are helpful.") } }; - _ = try mgr.appendMessage( - .{ .role = .system, .content = sys_blocks }, - null, - null, - ); - try testing.expect(!mgr.isFlushed()); - - // User message — also doesn't flush. - const usr_blocks = try testing.allocator.alloc(DiskContentBlock, 1); - usr_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hi there") } }; - _ = try mgr.appendMessage( - .{ .role = .user, .content = usr_blocks }, - "openai", - "gpt-4o", - ); - try testing.expect(!mgr.isFlushed()); - - // Assistant message — triggers flush. - const a_blocks = try testing.allocator.alloc(DiskContentBlock, 1); - a_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hello") } }; - _ = try mgr.appendMessage( - .{ - .role = .assistant, - .content = a_blocks, - .provider = try testing.allocator.dupe(u8, "openai"), - .model = try testing.allocator.dupe(u8, "gpt-4o"), - .stop_reason = try testing.allocator.dupe(u8, "stop"), - }, - null, - null, - ); - try testing.expect(mgr.isFlushed()); - - // Append another user/assistant round. - const u_two = try testing.allocator.alloc(DiskContentBlock, 1); - u_two[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "what's 2+2?") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u_two }, "openai", "gpt-4o"); - - const a2 = try testing.allocator.alloc(DiskContentBlock, 1); - a2[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "4") } }; - _ = try mgr.appendMessage( - .{ .role = .assistant, .content = a2, .stop_reason = try testing.allocator.dupe(u8, "stop") }, - null, - null, - ); - - try testing.expectEqual(@as(usize, 5), mgr.getEntries().len); - - break :blk try testing.allocator.dupe(u8, mgr.getSessionFile()); - }; - defer testing.allocator.free(session_file); - - // Verify the file exists and is well-formed. - { - const bytes = try readWholeFile(testing.allocator, io, session_file); - defer testing.allocator.free(bytes); - // 1 header + 5 entries + trailing \n on each = 6 newlines. - var nl_count: usize = 0; - for (bytes) |b| if (b == '\n') { - nl_count += 1; - }; - try testing.expectEqual(@as(usize, 6), nl_count); - } - - // Resume. - var resumed = try SessionManager.open(testing.allocator, io, session_file); - defer resumed.deinit(); - try testing.expect(resumed.isFlushed()); - try testing.expectEqual(@as(usize, 5), resumed.getEntries().len); - try testing.expectEqualStrings("/proj/foo", resumed.getCwd()); - - // Continue the conversation. - const u_three = try testing.allocator.alloc(DiskContentBlock, 1); - u_three[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "thanks") } }; - _ = try resumed.appendMessage(.{ .role = .user, .content = u_three }, "openai", "gpt-4o"); - try testing.expectEqual(@as(usize, 6), resumed.getEntries().len); -} - -test "SessionManager: assistant message tags the message metadata and the entry leaf id is the assistant entry" { - 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); - - var mgr = try SessionManager.init(testing.allocator, io, sessions, "/c"); - defer mgr.deinit(); - - const u_blocks = try testing.allocator.alloc(DiskContentBlock, 1); - u_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "ping") } }; - const user_id = try mgr.appendMessage(.{ .role = .user, .content = u_blocks }, "openai", "gpt-4o"); - - const a_blocks = try testing.allocator.alloc(DiskContentBlock, 1); - a_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "pong") } }; - const asst_id = try mgr.appendMessage(.{ .role = .assistant, .content = a_blocks }, null, null); - - // Leaf is the assistant entry. - try testing.expectEqualStrings(asst_id, mgr.getLeafId().?); - // Parent of assistant is the user entry. - const assistant_entry = mgr.getEntry(asst_id).?; - try testing.expectEqualStrings(user_id, assistant_entry.base().parent_id.?); - // User entry's parent is null (no system). - const user_entry = mgr.getEntry(user_id).?; - try testing.expect(user_entry.base().parent_id == null); -} - -test "SessionManager: activeModel is null before any user message, then tracks the latest user stamp" { - 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); - - var mgr = try SessionManager.init(testing.allocator, io, sessions, "/c"); - defer mgr.deinit(); - - // No user messages yet — there is no "active" model on disk yet. - try testing.expect(mgr.activeModel() == null); - - // Stamp a user message with anthropic. - const u_blocks = try testing.allocator.alloc(DiskContentBlock, 1); - u_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hi") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u_blocks }, "anthropic", "claude-sonnet-4-20250514"); - - { - const am = mgr.activeModel().?; - try testing.expectEqualStrings("anthropic", am.provider); - try testing.expectEqualStrings("claude-sonnet-4-20250514", am.model); - } -} - -test "SessionManager: rebuildConversation reconstructs system/user/assistant turn" { - 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); - - var mgr = try SessionManager.init(testing.allocator, io, sessions, "/c"); - defer mgr.deinit(); - - 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 u = try testing.allocator.alloc(DiskContentBlock, 1); - u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hello") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u }, "openai", "gpt-4o"); - - const a = try testing.allocator.alloc(DiskContentBlock, 1); - a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hi!") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, null, null); - - var conv = try mgr.rebuildConversation(); - defer conv.deinit(); - try testing.expectEqual(@as(usize, 3), conv.messages.items.len); - try testing.expectEqual(conversation_mod.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_mod.MessageRole.user, conv.messages.items[1].role); - try testing.expectEqualStrings("hello", conv.messages.items[1].content.items[0].Text.items); - try testing.expectEqual(conversation_mod.MessageRole.assistant, conv.messages.items[2].role); - try testing.expectEqualStrings("hi!", conv.messages.items[2].content.items[0].Text.items); -} - -test "SessionManager: crash recovery truncates corrupted trailing line" { - 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); - - // Build a valid session first. - const session_file: []u8 = blk: { - var mgr = try SessionManager.init(testing.allocator, io, sessions, "/c"); - defer mgr.deinit(); - const u = try testing.allocator.alloc(DiskContentBlock, 1); - u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "ping") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u }, "openai", "gpt-4o"); - const a = try testing.allocator.alloc(DiskContentBlock, 1); - a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "pong") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, null, null); - break :blk try testing.allocator.dupe(u8, mgr.getSessionFile()); - }; - defer testing.allocator.free(session_file); - - // Corrupt the file: append a partial JSON line at the end. - const garbage = "{\"type\":\"message\",\"id\":\"deadbeef\",\"parent"; - { - const file = try Io.Dir.cwd().openFile(io, session_file, .{ .mode = .write_only }); - defer file.close(io); - const len = try file.length(io); - try file.writePositionalAll(io, garbage, len); - } - // Confirm the file got bigger. - { - const f = try Io.Dir.cwd().openFile(io, session_file, .{ .mode = .read_only }); - defer f.close(io); - const corrupted_len = try f.length(io); - try testing.expect(corrupted_len > garbage.len); - } - - // Now resume — the partial line should be truncated. - var resumed = try SessionManager.open(testing.allocator, io, session_file); - defer resumed.deinit(); - try testing.expectEqual(@as(usize, 2), resumed.getEntries().len); - - // And the file on disk should match. - { - const bytes = try readWholeFile(testing.allocator, io, session_file); - defer testing.allocator.free(bytes); - try testing.expect(!std.mem.endsWith(u8, bytes, "parent")); - // Should end with a newline after the assistant entry. - try testing.expectEqual(@as(u8, '\n'), bytes[bytes.len - 1]); - } -} - -test "listSessions: returns most recent first, with counts" { - 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); - - // Create two sessions. - for (0..2) |i| { - var mgr = try SessionManager.init(testing.allocator, io, sessions, "/c"); - defer mgr.deinit(); - const u = try testing.allocator.alloc(DiskContentBlock, 1); - u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "u") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u }, "openai", "gpt-4o"); - const a = try testing.allocator.alloc(DiskContentBlock, 1); - a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, null, null); - // Small sleep so UUIDv7 timestamps differ. - io.sleep(.fromMilliseconds(2), .real) catch {}; - _ = i; - } - - const infos = try listSessions(testing.allocator, io, sessions, null); - defer freeSessionInfos(testing.allocator, infos); - try testing.expectEqual(@as(usize, 2), infos.len); - try testing.expectEqual(@as(usize, 2), infos[0].message_count); - try testing.expect(std.mem.order(u8, infos[0].modified, infos[1].modified) != .lt); -} - -test "findMostRecentSession: picks lexicographically greatest" { - 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); - - // Pre-resolution before any sessions exist → null. - try testing.expect((try findMostRecentSession(testing.allocator, io, sessions)) == null); - - // Create two. - var second_file: ?[]u8 = null; - defer if (second_file) |s| testing.allocator.free(s); - - for (0..2) |i| { - var mgr = try SessionManager.init(testing.allocator, io, sessions, "/c"); - defer mgr.deinit(); - const u = try testing.allocator.alloc(DiskContentBlock, 1); - u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "u") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u }, "openai", "gpt-4o"); - const a = try testing.allocator.alloc(DiskContentBlock, 1); - a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, null, null); - if (i == 1) second_file = try testing.allocator.dupe(u8, mgr.getSessionFile()); - io.sleep(.fromMilliseconds(2), .real) catch {}; - } - - const found = (try findMostRecentSession(testing.allocator, io, sessions)).?; - defer testing.allocator.free(found); - try testing.expectEqualStrings(second_file.?, found); -} - -test "SessionManager: tool-use round-trip — assistant w/ ToolUse, user w/ ToolResult, assistant" { - 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(); - - const u = try testing.allocator.alloc(DiskContentBlock, 1); - u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "list files") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u }, "openai", "gpt-4o"); - - // Assistant emits a ToolUse. - const am1 = try testing.allocator.alloc(DiskContentBlock, 2); - am1[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "checking...") } }; - am1[1] = .{ .tool_use = .{ - .id = try testing.allocator.dupe(u8, "tool_abc"), - .name = try testing.allocator.dupe(u8, "bash"), - .input = try testing.allocator.dupe(u8, "{\"command\":\"ls\"}"), - } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = am1 }, null, null); - - // Tool-result user message. - const tr = try testing.allocator.alloc(DiskContentBlock, 1); - const trp = try testing.allocator.alloc(session_mod.DiskResultPart, 1); - trp[0] = .{ .text = try testing.allocator.dupe(u8, "a.txt\nb.txt") }; - tr[0] = .{ .tool_result = .{ - .tool_use_id = try testing.allocator.dupe(u8, "tool_abc"), - .parts = trp, - } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = tr }, "openai", "gpt-4o"); - - // Final assistant reply. - const a2 = try testing.allocator.alloc(DiskContentBlock, 1); - a2[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "two files: a.txt and b.txt") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a2 }, null, null); - - break :blk try testing.allocator.dupe(u8, mgr.getSessionFile()); - }; - defer testing.allocator.free(session_file); - - // Reopen and verify content blocks survive. - var resumed = try SessionManager.open(testing.allocator, io, session_file); - defer resumed.deinit(); - const entries = resumed.getEntries(); - try testing.expectEqual(@as(usize, 4), entries.len); - - // [1] = assistant with ToolUse - try testing.expectEqual(DiskMessageRole.assistant, entries[1].message.message.role); - try testing.expectEqual(@as(usize, 2), entries[1].message.message.content.len); - try testing.expect(entries[1].message.message.content[1] == .tool_use); - try testing.expectEqualStrings("bash", entries[1].message.message.content[1].tool_use.name); - try testing.expectEqualStrings("{\"command\":\"ls\"}", entries[1].message.message.content[1].tool_use.input); - - // [2] = user with ToolResult, stamped with provider/model. - try testing.expectEqual(DiskMessageRole.user, entries[2].message.message.role); - try testing.expectEqualStrings("openai", entries[2].message.provider.?); - try testing.expect(entries[2].message.message.content[0] == .tool_result); - try testing.expectEqualStrings("tool_abc", entries[2].message.message.content[0].tool_result.tool_use_id); - try testing.expectEqualStrings("a.txt\nb.txt", entries[2].message.message.content[0].tool_result.parts[0].text); - - // Conversation rebuild yields the same shape. - var conv = try resumed.rebuildConversation(); - defer conv.deinit(); - try testing.expectEqual(@as(usize, 4), conv.messages.items.len); - try testing.expect(conv.messages.items[1].content.items[1] == .ToolUse); - try testing.expect(conv.messages.items[2].content.items[0] == .ToolResult); -} - -test "SessionManager: linear chain — each entry's parent_id is the previous entry's id" { - 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); - - var mgr = try SessionManager.init(testing.allocator, io, sessions, "/c"); - defer mgr.deinit(); - - // Three rounds: sys, user, asst, user, asst. - const sys = try testing.allocator.alloc(DiskContentBlock, 1); - sys[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "sys") } }; - _ = try mgr.appendMessage(.{ .role = .system, .content = sys }, null, null); - - const u_one = try testing.allocator.alloc(DiskContentBlock, 1); - u_one[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "u1") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u_one }, "openai", "gpt-4o"); - - const a_one = try testing.allocator.alloc(DiskContentBlock, 1); - a_one[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a1") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a_one }, null, null); - - const u_two = try testing.allocator.alloc(DiskContentBlock, 1); - u_two[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "u2") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u_two }, "openai", "gpt-4o"); - - const a_two = try testing.allocator.alloc(DiskContentBlock, 1); - a_two[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a2") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a_two }, null, null); - - const entries = mgr.getEntries(); - try testing.expectEqual(@as(usize, 5), entries.len); - try testing.expect(entries[0].base().parent_id == null); - for (entries[1..], 1..) |e, i| { - try testing.expectEqualStrings(entries[i - 1].base().id, e.base().parent_id.?); - } -} - -test "resolveSessionId: unique prefix → match, ambiguous → error" { - 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); - - // Create one session. - var mgr = try SessionManager.init(testing.allocator, io, sessions, "/c"); - defer mgr.deinit(); - const u = try testing.allocator.alloc(DiskContentBlock, 1); - u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "u") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u }, "openai", "gpt-4o"); - const a = try testing.allocator.alloc(DiskContentBlock, 1); - a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, null, null); - - const id = mgr.getSessionId(); - const prefix = id[0..8]; - - const resolved = try resolveSessionId(testing.allocator, io, sessions, prefix); - defer testing.allocator.free(resolved); - try testing.expectEqualStrings(mgr.getSessionFile(), resolved); - - 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]); -} - -test "loadConversation: trailing user prompt is split out as dangling, excluded from conversation" { - 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); - - var mgr = try SessionManager.init(testing.allocator, io, sessions, "/c"); - defer mgr.deinit(); - - // A completed user/assistant round, then a dangling user prompt with no - // following assistant (simulating a crash right after submission). - const um1 = try testing.allocator.alloc(DiskContentBlock, 1); - um1[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hi") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = um1 }, "openai", "gpt-4o"); - const am1 = try testing.allocator.alloc(DiskContentBlock, 1); - am1[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hello") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = am1 }, null, null); - const um2 = try testing.allocator.alloc(DiskContentBlock, 1); - um2[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "what's 2+2?") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = um2 }, "openai", "gpt-4o"); - - var loaded = try mgr.loadConversation(testing.allocator); - defer loaded.deinit(testing.allocator); - - // Dangling prompt recovered. - try testing.expect(loaded.dangling_user != null); - try testing.expectEqualStrings("what's 2+2?", loaded.dangling_user.?); - // Conversation excludes the dangling user turn: [user hi, assistant hello]. - try testing.expectEqual(@as(usize, 2), loaded.conversation.messages.items.len); - try testing.expectEqual(conversation_mod.MessageRole.user, loaded.conversation.messages.items[0].role); - try testing.expectEqual(conversation_mod.MessageRole.assistant, loaded.conversation.messages.items[1].role); -} - -test "loadConversation: no dangling prompt when log ends with assistant" { - 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); - - var mgr = try SessionManager.init(testing.allocator, io, sessions, "/c"); - defer mgr.deinit(); - - const um1 = try testing.allocator.alloc(DiskContentBlock, 1); - um1[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hi") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = um1 }, "openai", "gpt-4o"); - const am1 = try testing.allocator.alloc(DiskContentBlock, 1); - am1[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hello") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = am1 }, null, null); - - var loaded = try mgr.loadConversation(testing.allocator); - defer loaded.deinit(testing.allocator); - try testing.expect(loaded.dangling_user == null); - try testing.expectEqual(@as(usize, 2), loaded.conversation.messages.items.len); -} - -test "store(): SessionStore wrapper delegates to the concrete manager" { - 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); - - var mgr = try SessionManager.init(testing.allocator, io, sessions, "/c"); - defer mgr.deinit(); - - const s = mgr.store(); - - // Append a user+assistant batch through the neutral interface. - var messages = try testing.allocator.alloc(DiskMessage, 2); - defer testing.allocator.free(messages); - const uc = try testing.allocator.alloc(DiskContentBlock, 1); - uc[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "ping") } }; - messages[0] = .{ .role = .user, .content = uc }; - const ac = try testing.allocator.alloc(DiskContentBlock, 1); - ac[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "pong") } }; - messages[1] = .{ .role = .assistant, .content = ac }; - - try s.appendMessages(messages, &.{ "openai", null }, &.{ "gpt-4o", null }); - try testing.expectEqual(@as(usize, 2), mgr.getEntries().len); - - const am = s.activeModel().?; - try testing.expectEqualStrings("openai", am.provider); - try testing.expectEqualStrings("gpt-4o", am.model); - try testing.expectEqualStrings(mgr.getSessionId(), s.sessionId()); -} diff --git a/libpanto/src/session_store.zig b/libpanto/src/session_store.zig index b929c66..de25e21 100644 --- a/libpanto/src/session_store.zig +++ b/libpanto/src/session_store.zig @@ -1,79 +1,132 @@ //! `SessionStore`: the neutral persistence seam for the `Agent`. //! -//! Session logging used to be a single concrete type (`SessionManager`, -//! filesystem JSONL). This interface lets a `libpanto` consumer swap in its -//! own backend — e.g. a web service backed by Postgres — without the agent -//! knowing how (or whether) persistence happens. +//! The interface is **asymmetric**: rich on write (audit/provenance-capable), +//! minimal on read (resume-oriented). The store decides how much write-side +//! richness it durably keeps. //! -//! The interface follows the same `{ ptr, vtable }` shape as the other -//! `libpanto` seams (`Tool`/`ToolSource`/`Provider`). It traffics in -//! `DiskMessage` as the neutral in-memory representation: the default -//! backend (`FSJSONLStore`) emits JSONL, but a Postgres backend would map -//! `DiskMessage` to columns and never produce a byte of JSONL. +//! ## Write side (maximalist) //! -//! ## What lives here vs. on the concrete backend +//! `appendMessages` takes `[]PersistentMessage` — the rich, audit-oriented +//! write record. Each carries the in-memory `Message` being appended, its +//! `usage`, the **wire-format** provider identity (`api_style`, `base_url`, +//! `model`, `reasoning` — never CLI config aliases, and never any `api_key` +//! material, not even a hash), and full provenance context (the entire +//! current conversation and the tool set offered for this turn). The library +//! *offers* all of it on every append; a store keeps what it wants. The +//! built-in `FileSystemJSONLStore` deliberately ignores the `conversation` +//! and `tools_available` provenance fields. //! -//! On the interface (every store must do these): -//! - `appendMessages` — the batch-atomic append primitive. A single -//! append is a length-1 batch. -//! - `loadConversation` — reconstruct one linear `Conversation` from the -//! store, plus an optional dangling trailing user prompt (see -//! `LoadedSession`). A store you cannot read is not a valid store. -//! - `sessionId` — opaque id string. -//! - `activeModel` — the provider/model last stamped on a user entry. +//! ## Read side (minimal) //! -//! NOT on the interface (backend-specific, stay as free functions / methods -//! on the concrete type): -//! - filesystem path accessors (`getSessionFile`), -//! - catalog/listing/resume helpers (`listSessions`, -//! `findMostRecentSession`, `resolveSessionId`) — a web backend lists -//! via SQL, not by walking a directory. +//! `load` reconstructs one linear `Conversation`; `list`/`resolve`/`latest` +//! traffic in `SessionInfo` (display/selection metadata) and `Session` +//! (an `info` + a store to proxy to). The read path never reproduces a +//! `PersistentMessage` — a store may not have kept the provenance. +//! +//! ## Store construction +//! +//! Stores own their own (unprescribed) `init`: a Postgres store takes a DSN, +//! the FS store takes a directory. Nothing in the vtable carries an +//! allocator or io — the store captured whatever it needs at its own init. const std = @import("std"); const Allocator = std.mem.Allocator; const session_mod = @import("session.zig"); const conversation_mod = @import("conversation.zig"); +const config_mod = @import("config.zig"); +const tool_source_mod = @import("tool_source.zig"); -// Re-export the disk types so the interface is self-contained: an embedder -// implementing a `SessionStore` imports everything it needs from here. -pub const DiskMessage = session_mod.DiskMessage; -pub const DiskMessageRole = session_mod.DiskMessageRole; -pub const DiskSystemMode = session_mod.DiskSystemMode; -pub const DiskContentBlock = session_mod.DiskContentBlock; -pub const Usage = session_mod.Usage; pub const Conversation = conversation_mod.Conversation; +pub const Message = conversation_mod.Message; +pub const Usage = conversation_mod.Usage; +pub const APIStyle = config_mod.APIStyle; +pub const ReasoningEffort = config_mod.ReasoningEffort; +pub const ToolDecl = tool_source_mod.ToolDecl; + +/// The default filesystem-JSONL backend, re-exported under its +/// interface-facing name. Its concrete constructor (`init`/`open`) and +/// catalog helpers are backend-specific and stay on that module. +pub const FileSystemJSONLStore = @import("file_system_jsonl_store.zig").FileSystemJSONLStore; + +/// Wire-format provider identity. This is the **ground truth** of which +/// endpoint a turn was sent to — never a CLI config alias (aliases get +/// renamed; two keys for one endpoint are indistinguishable on the wire). +/// `reasoning` disambiguates otherwise-identical endpoints. No `api_key` +/// material ever appears here. Aliased from `config` to avoid a module +/// cycle (config must not import session_store). +pub const WireIdentity = config_mod.WireIdentity; + +/// The rich, audit-oriented write record. The library offers all of this on +/// every append; the store keeps what it wants. +pub const PersistentMessage = struct { + /// The in-memory message being appended (carries its own `metadata`). + message: Message, + /// Provider usage for this message (assistant turns), or null. + usage: ?Usage = null, + /// Wire-format provider identity for the turn this message belongs to. + identity: WireIdentity, + /// Full provenance: the entire current conversation at write time. The + /// FS store ignores this; an audit store may content-address it. + conversation: []const Message = &.{}, + /// Full provenance: the tool set offered for this turn. The FS store + /// ignores this; an audit store may content-address it. + tools_available: []const ToolDecl = &.{}, +}; -/// The default filesystem-JSONL backend. Defined in `session_manager.zig`; -/// re-exported here under its interface-facing name. Its concrete -/// constructors (`init`/`open`) and catalog helpers are backend-specific -/// and stay on that module. -pub const FSJSONLStore = @import("session_manager.zig").SessionManager; - -/// The read side's result: a reconstructed linear `Conversation` plus an -/// optional dangling trailing user prompt. -/// -/// `dangling_user` is set when the log ends with a user entry that has no -/// following assistant entry — e.g. a crash or quit right after the prompt -/// was submitted and durably logged but before the model replied. The -/// reconstructed `conversation` **excludes** that dangling turn, so a -/// resumed agent never auto-sends it. Consumers may surface the dangling -/// text (e.g. a TUI prefilling it for editing) or ignore it. -pub const LoadedSession = struct { - conversation: Conversation, - /// Owned by the caller's allocator when present; free it when done. - dangling_user: ?[]const u8 = null, - - pub fn deinit(self: *LoadedSession, alloc: Allocator) void { - self.conversation.deinit(); - if (self.dangling_user) |d| alloc.free(d); +/// Display/selection metadata for one session — pure data, aliased. Used by +/// `panto sessions` and resume pre-selection. The last-used wire identity is +/// updated on append (for resume), never a CLI config alias. +pub const SessionInfo = struct { + id: []const u8, + created: []const u8, + modified: []const u8, + message_count: usize, + /// May be truncated. + last_user_message: []const u8, + /// Last-used wire identity, updated on append. + api_style: APIStyle, + base_url: []const u8, + model: []const u8, + reasoning: ReasoningEffort, + + pub fn deinit(self: SessionInfo, alloc: Allocator) void { + alloc.free(self.id); + alloc.free(self.created); + alloc.free(self.modified); + alloc.free(self.last_user_message); + alloc.free(self.base_url); + alloc.free(self.model); } }; -/// The active provider/model, as last stamped on a user entry. -pub const ActiveModel = struct { - provider: []const u8, - model: []const u8, +/// A session handle: pure data (a `SessionInfo`) plus a store to proxy to. +pub const Session = struct { + info: SessionInfo, + store: SessionStore, + + /// Reconstruct the conversation. The id came from `resolve`/`latest`, so + /// the conversation must exist; a `null` from the store is promoted to + /// an error. + pub fn load(self: Session) !Conversation { + return (try self.store.load(self.info.id)) orelse error.SessionNotFound; + } + + /// Append a batch of messages, proxying to the store. Takes `*Session` + /// for API symmetry and to allow future in-place `info` updates; today + /// it only updates the non-owning `api_style`/`reasoning` last-used + /// fields (the `base_url`/`model` strings stay the owned originals to + /// avoid aliasing borrowed config memory — resume picks the default + /// model rather than matching the stored wire identity, so the stale + /// display strings are harmless). + pub fn append(self: *Session, messages: []PersistentMessage) !void { + try self.store.appendMessages(self.info.id, messages); + if (messages.len > 0) { + const id = messages[messages.len - 1].identity; + self.info.api_style = id.api_style; + self.info.reasoning = id.reasoning; + } + } }; /// A pluggable session-persistence backend. @@ -82,59 +135,54 @@ pub const SessionStore = struct { vtable: *const VTable, pub const VTable = struct { - /// Append a batch of messages atomically. `providers`/`models` are - /// parallel arrays (one per message) carrying the top-level entry - /// stamps (recorded on user entries). A single append is a - /// length-1 batch. - /// - /// Ownership: the store **consumes** each `DiskMessage` on success - /// (takes ownership of its heap allocations). On error the store - /// frees any messages it had already taken and the caller frees the - /// rest — i.e. after this call returns the caller must not free the - /// `DiskMessage`s regardless of outcome. (The `messages` *slice* - /// itself remains the caller's.) - appendMessages: *const fn ( - ctx: *anyopaque, - messages: []DiskMessage, - providers: []const ?[]const u8, - models: []const ?[]const u8, - ) anyerror!void, - - /// Reconstruct one linear `Conversation` from the store, with the - /// dangling trailing user prompt (if any) split out. The returned - /// `LoadedSession` owns its allocations against `alloc`. - loadConversation: *const fn ( - ctx: *anyopaque, - alloc: Allocator, - ) anyerror!LoadedSession, - - /// Opaque session id. Borrowed; lifetime owned by the store. - sessionId: *const fn (ctx: *anyopaque) []const u8, - - /// Provider/model last stamped on a user entry, or null if no user - /// message has been recorded yet. Borrowed slices owned by the - /// store. - activeModel: *const fn (ctx: *anyopaque) ?ActiveModel, + /// Mint an in-memory session handle. Cannot fail: nothing hits the + /// backend until the first `appendMessages` (create-on-demand), so + /// no record exists before the first assistant message. + create: *const fn (ctx: *anyopaque) Session, + + /// List known sessions, newest first. Caller frees via + /// `freeSessionInfos`. + list: *const fn (ctx: *anyopaque) anyerror![]SessionInfo, + + /// Free a slice returned by `list`. + freeSessionInfos: *const fn (ctx: *anyopaque, infos: []SessionInfo) void, + + /// Resolve a (possibly abbreviated) id to a session, or null if no + /// match. + resolve: *const fn (ctx: *anyopaque, id: []const u8) anyerror!?Session, + + /// The most recent session, or null if none exist. + latest: *const fn (ctx: *anyopaque) anyerror!?Session, + + /// Reconstruct one linear `Conversation` for `id`, or null if absent. + /// The returned `Conversation` self-describes its allocator. + load: *const fn (ctx: *anyopaque, id: []const u8) anyerror!?Conversation, + + /// Append a batch atomically. A single append is a length-1 batch. + /// The store reads what it wants off each `PersistentMessage` and + /// is responsible for any de-duplication of provenance. + appendMessages: *const fn (ctx: *anyopaque, session_id: []const u8, messages: []PersistentMessage) anyerror!void, }; - pub fn appendMessages( - self: SessionStore, - messages: []DiskMessage, - providers: []const ?[]const u8, - models: []const ?[]const u8, - ) !void { - return self.vtable.appendMessages(self.ptr, messages, providers, models); + pub fn create(self: SessionStore) Session { + return self.vtable.create(self.ptr); } - - pub fn loadConversation(self: SessionStore, alloc: Allocator) !LoadedSession { - return self.vtable.loadConversation(self.ptr, alloc); + pub fn list(self: SessionStore) ![]SessionInfo { + return self.vtable.list(self.ptr); } - - pub fn sessionId(self: SessionStore) []const u8 { - return self.vtable.sessionId(self.ptr); + pub fn freeSessionInfos(self: SessionStore, infos: []SessionInfo) void { + self.vtable.freeSessionInfos(self.ptr, infos); } - - pub fn activeModel(self: SessionStore) ?ActiveModel { - return self.vtable.activeModel(self.ptr); + pub fn resolve(self: SessionStore, id: []const u8) !?Session { + return self.vtable.resolve(self.ptr, id); + } + pub fn latest(self: SessionStore) !?Session { + return self.vtable.latest(self.ptr); + } + pub fn load(self: SessionStore, id: []const u8) !?Conversation { + return self.vtable.load(self.ptr, id); + } + pub fn appendMessages(self: SessionStore, session_id: []const u8, messages: []PersistentMessage) !void { + return self.vtable.appendMessages(self.ptr, session_id, messages); } }; diff --git a/libpanto/src/turn_persist.zig b/libpanto/src/turn_persist.zig index a8b3faa..0147fbd 100644 --- a/libpanto/src/turn_persist.zig +++ b/libpanto/src/turn_persist.zig @@ -1,147 +1,86 @@ //! Turn → session-log persistence: map in-memory `Conversation` messages -//! to neutral `DiskMessage`s and append them to a `SessionStore`. +//! to rich `PersistentMessage` write records and append them through a +//! `Session` handle. //! -//! 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. +//! This logic 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 `Session`; 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. +//! The write record is **maximalist** (full wire identity + usage + the +//! entire current conversation + the offered tool set). The library offers +//! it all on every append; each store keeps what it wants. `PersistentMessage` +//! borrows the in-memory `Message` directly — no disk conversion happens +//! here; the store does its own serialization. +//! +//! Per-message usage is read directly off `Message.usage`. 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; +const PersistentMessage = session_store.PersistentMessage; +const WireIdentity = session_store.WireIdentity; +const ToolDecl = session_store.ToolDecl; +const Session = session_store.Session; -/// Persist every conversation message at index `>= start_index` to `store` -/// as a single atomic batch. +/// Persist every conversation message at index `>= start_index` through +/// `session` 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). +/// Each `PersistentMessage` borrows the in-memory `Message`, the full +/// current conversation, and `tools` (the offered tool set) — all owned by +/// the caller and valid for the duration of the call. The wire `identity` +/// is stamped on every message; the store decides per-role what to keep +/// (the FS store drops the stamp on system entries). /// /// 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, + session: *Session, conv: *const conversation.Conversation, start_index: usize, - provider: []const u8, - model: []const u8, + identity: WireIdentity, + tools: []const ToolDecl, ) !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 batch: std.ArrayList(PersistentMessage) = .empty; + defer batch.deinit(alloc); + const all_messages = conv.messages.items; 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; - } + while (i < all_messages.len) : (i += 1) { + const msg = all_messages[i]; 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); - }, - } + try batch.append(alloc, .{ + .message = msg, + .usage = msg.usage, + .identity = identity, + .conversation = all_messages, + .tools_available = tools, + }); } - if (batch_messages.items.len == 0) return; - try store.appendMessages(batch_messages.items, batch_providers.items, batch_models.items); + if (batch.items.len == 0) return; + try session.append(batch.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). +/// latest compaction summary onward as fresh entries. pub fn persistCompaction( alloc: Allocator, - store: SessionStore, + session: *Session, conv: *const conversation.Conversation, - provider: []const u8, - model: []const u8, + identity: WireIdentity, + tools: []const ToolDecl, ) !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; + try persistTurn(alloc, session, conv, start, identity, tools); } /// True when the assistant message at `index` contains a ToolUse block -- cgit v1.3