From 6545cdfd8f2bc865aa06a2b5515056daf58ba111 Mon Sep 17 00:00:00 2001 From: T Date: Wed, 27 May 2026 12:45:20 -0600 Subject: session files --- build.zig | 11 + build.zig.zon | 4 + docs/archive/phase-4.md | 643 +++++++++++ docs/phase-4.md | 645 ----------- libpanto/src/agent.zig | 2 +- libpanto/src/anthropic_messages_json.zig | 52 +- libpanto/src/openai_chat_json.zig | 64 ++ libpanto/src/pricing.zig | 303 ++++++ libpanto/src/provider.zig | 16 +- libpanto/src/provider_anthropic_messages.zig | 126 ++- libpanto/src/provider_openai_chat.zig | 129 ++- libpanto/src/root.zig | 3 + libpanto/src/session.zig | 961 ++++++++++++++++ libpanto/src/session_manager.zig | 1509 ++++++++++++++++++++++++++ src/main.zig | 369 ++++++- src/models_toml.zig | 291 +++++ src/session_paths.zig | 141 +++ src/subcommand.zig | 98 ++ 18 files changed, 4690 insertions(+), 677 deletions(-) create mode 100644 docs/archive/phase-4.md delete mode 100644 docs/phase-4.md create mode 100644 libpanto/src/pricing.zig create mode 100644 libpanto/src/session.zig create mode 100644 libpanto/src/session_manager.zig create mode 100644 src/models_toml.zig create mode 100644 src/session_paths.zig diff --git a/build.zig b/build.zig index 5bf5d44..4e4e862 100644 --- a/build.zig +++ b/build.zig @@ -13,6 +13,15 @@ pub fn build(b: *std.Build) void { .optimize = optimize, }); + // TOML parser (used by the CLI for ~/.config/panto/models.toml). + // We disable the upstream's optional thread-pool dep — we only need + // sequential parsing of a small config file. + const toml_dep = b.dependency("toml", .{ + .target = target, + .optimize = optimize, + .@"thread-pool" = false, + }); + // Fetch upstream Lua source (used both for our static library and // staged at runtime as the `include/` headers under $PANTO_HOME). // Reproducibility comes from the content-addressed hash in @@ -47,6 +56,7 @@ pub fn build(b: *std.Build) void { .link_libc = true, }); exe_mod.addImport("panto", panto_lib_dep.module("panto")); + exe_mod.addImport("toml", toml_dep.module("toml")); exe_mod.addImport("versions", versions_mod.createModule()); exe_mod.addImport("embedded_luarocks", b.createModule(.{ .root_source_file = luarocks_embed_path, @@ -103,6 +113,7 @@ pub fn build(b: *std.Build) void { .link_libc = true, }); test_mod.addImport("panto", panto_lib_dep.module("panto")); + test_mod.addImport("toml", toml_dep.module("toml")); test_mod.addImport("versions", versions_mod.createModule()); test_mod.addImport("embedded_luarocks", b.createModule(.{ .root_source_file = luarocks_embed_path, diff --git a/build.zig.zon b/build.zig.zon index f7eb08b..19feb70 100644 --- a/build.zig.zon +++ b/build.zig.zon @@ -14,6 +14,10 @@ .url = "https://luarocks.github.io/luarocks/releases/luarocks-3.13.0.tar.gz", .hash = "N-V-__8AADZhHwD_K21NLLYGySn7TnLhpLyP8ca2JnXSLgbp", }, + .toml = .{ + .url = "https://gitlab.com/devnw/zig/toml/-/archive/v0.1.4/toml-v0.1.4.tar.gz", + .hash = "toml-0.1.4-MHnSh2GWBQCW9SORG0z01zGj-bBlxbb82Itm5eIHNPX5", + }, }, .paths = .{ "build.zig", diff --git a/docs/archive/phase-4.md b/docs/archive/phase-4.md new file mode 100644 index 0000000..e211e1f --- /dev/null +++ b/docs/archive/phase-4.md @@ -0,0 +1,643 @@ +# Phase 4: Conversation Serialization + +## Goal + +Persistent session storage as an append-only event log. Every event in a session — messages, which provider/model handled each request — is recorded to disk as it happens. On load, pantograph reads the log and rebuilds the in-memory conversation. Sessions survive process restarts and can be reviewed later. + +Unlike systems that derive state by folding a stream of typed events (model-change entries, thinking-level-change entries, etc.), pantograph's load is mechanically simple: walk the entries, instantiate one in-memory `Message` per `message` entry, copy provider/model stamps as-is. There is no event replay or state derivation — the leaf entry's stamp _is_ the active model. The format leaves room for richer event types in the future (compaction, branching), but phase 4 doesn't introduce any. + +## Deliverable + +A session persistence system where pantograph saves and resumes conversations. At the end of this phase, you can: + +- Start `panto`, hold a conversation, and find the session saved to disk. +- Restart `panto --resume` and continue the conversation exactly where you left off. +- Restart `panto --resume ` to resume a specific session. +- Inspect the JSONL session file to see every event that occurred, including which provider/model handled each turn. +- Have pantograph recover gracefully from a crash — any corrupted trailing line is removed, and the session loads from the last valid entry. + +## What is usable at the end + +| Capability | How to exercise it | +|---|---| +| Auto-save session | Start `panto`, converse, quit; session is in `~/.local/share/panto/sessions/` | +| Resume most recent | `panto --resume` — continues the most recent session for the current working directory | +| Resume specific session | `panto --resume ` — opens the session with the given ID (or unique prefix) | +| List sessions | `panto sessions` — lists sessions for the current working directory | +| Crash recovery | Kill `panto` mid-turn; restart with `--resume`; corrupted trailing line is truncated, session loads from last valid entry | + +## What is explicitly out of scope + +- Compaction / context pruning (phase 6 — the format is designed to allow future entry types like `compaction` without breaking older readers, but phase 4 does not implement any form of compaction) +- Branching / tree navigation (the format supports it via `id`/`parentId`, but no CLI commands for branching exist yet) +- Session forking or cloning +- Custom entry types (for extensions — future phase) +- Labels, bookmarks, session naming +- Session export or format migration beyond version 1 +- In-memory-only sessions from the CLI (every CLI session is persisted; libpanto embedders can opt out by passing a no-op store) + +--- + +## Library / CLI Boundary + +The storage interface lives in libpanto. The default `fs_jsonl` implementation also lives in libpanto. The CLI decides _where_ a session is stored and constructs the store with that path. + +### libpanto owns + +- The `SessionStore` interface (vtable: append entry, load entries, etc.). +- An `fs_jsonl` implementation that takes a base directory at construction time and writes the JSONL format defined below. +- The agent loop's interaction with the store: which events get appended, at what points in a turn. + +### CLI owns + +- Selecting the base directory: XDG resolution (`$XDG_DATA_HOME` or `~/.local/share`), the `panto/sessions/` subpath, and the per-project `/` grouping. +- Selecting the session file: new-file-per-invocation by default; `--resume` / `--resume ` to pick an existing file. +- Constructing `fs_jsonl` with the resolved directory and passing it to libpanto. + +Other embedders (a TUI, a daemon, a Lua host, a test harness) bring their own directory policy or their own `SessionStore` implementation. A no-op store is a valid choice for embedders that don't want persistence. + +### Mid-turn writes + +The in-memory `Conversation` is the source of truth during a turn. The store is written between discrete events — never as an inline dependency of streaming. A crash mid-turn loses the in-flight assistant message but never corrupts prior state. (Phase 4 writes per-message; finer-grained streaming-event entries are out of scope.) + +--- + +## Event Log Format + +Sessions are stored as JSONL (JSON Lines) files. Each line is a JSON object with a `type` field. Entries form a tree structure via `id`/`parentId` fields, enabling in-place branching in future phases without creating new files. + +The event log is the authoritative record of a session. It captures everything that happened — not just the conversation content, but the full context of how it happened (which model was active, when it changed, etc.). On load, pantograph rebuilds all state from the log alone. + +### File Location + +``` +~/.local/share/panto/sessions//.jsonl +``` + +Where `` is the working directory with `/` replaced by `--`. This groups sessions by project directory, making it easy to find sessions for a given project. + +The filename is just the session's UUIDv7. Because UUIDv7 is time-ordered, sorting filenames lexicographically sorts sessions chronologically — no separate timestamp prefix is needed. + +The base directory respects `XDG_DATA_HOME`, falling back to `~/.local/share` if unset (per the XDG Base Directory specification). + +Example: +``` +~/.local/share/panto/sessions/--Users-travis-Code-pantograph--/019dc5ba-53f6-71a5-ab8f-b1f8709c2572.jsonl +``` + +### Crash Recovery + +On loading a session, pantograph parses lines one by one from the top. As soon as a line fails to parse as a complete JSON object, **that line and every line after it are truncated from the file**. The session loads from the last fully valid entry. + +This is deliberately stricter than "skip malformed lines." Tool-use/tool-result pairs are stateful — a missing or partial entry in the middle would leave subsequent entries referring to a `toolUseId` that doesn't exist, which providers (notably Anthropic) reject outright. The only safe resume point is immediately before the first corruption. + +In practice the corruption window is the tail of the file (a crash mid-write), so truncation is effectively the same as "drop the partial last line." Defining the rule by position-of-first-error rather than "last line" handles edge cases (e.g., a partial write followed by a clean flush by another process) without ambiguity. + +--- + +## Entry Types + +### SessionHeader + +First (and only) header line of the file. Metadata only — not part of the tree (no `id`/`parentId`). Exactly one header per file, always at the top. + +```json +{ + "type": "session", + "version": 1, + "id": "019dc5ba-53f6-71a5-ab8f-b1f8709c2572", + "timestamp": "2026-04-25T17:40:15.990Z", + "cwd": "/Users/travis/Code/pantograph" +} +``` + +| Field | Type | Purpose | +|---|---|---| +| `version` | number | Format version. `1` for phase 4. | +| `id` | string | Session UUIDv7. | +| `timestamp` | string | ISO 8601 timestamp of session creation. | +| `cwd` | string | Working directory where the session was started. | + +There are no provider/model fields on the header. The session file isn't created until the first assistant message flushes, at which point a user message is already buffered — so every persisted session contains at least one user-message stamp, and that stamp is the authoritative initial model. + +### Versioning and migration + +The `version` field exists from day one so future format changes are explicit. Phase 4 is version 1. + +When a future pantograph version bumps the format to 2+, the load path runs `migrate(entries) -> bool`. If migration mutates anything, the entire file is rewritten once with the new version stamped on the header. Migration is eager: every v1 file encountered by a v2-capable pantograph is upgraded immediately on first load. There is never a mixed-version file. + +### EntryBase + +All entries (except the header) extend this base: + +``` +EntryBase = { + type: string, // entry type discriminator + id: string, // 8-char hex ID, random + parentId: string | null, // parent entry ID (null for first entry) + timestamp: string, // ISO 8601 +} +``` + +The first entry has `parentId: null`. Each subsequent entry points to its parent, forming a linked path from root to leaf. In phase 4 this is always a linear chain; the tree structure exists to support branching in future phases. + +### MessageEntry + +A message in the conversation. The `message` field contains role, content blocks, and (for assistant messages) provider metadata. + +**System message:** +```json +{ + "type": "message", + "id": "s1s2s3s4", + "parentId": null, + "timestamp": "2026-04-25T14:00:00.000Z", + "message": { + "role": "system", + "content": [ + { "type": "text", "text": "You are a helpful assistant." } + ] + } +} +``` + +**User message (human prompt):** +```json +{ + "type": "message", + "id": "a1b2c3d4", + "parentId": "s1s2s3s4", + "timestamp": "2026-04-25T14:00:01.000Z", + "message": { + "role": "user", + "content": [ + { "type": "text", "text": "What files are in this directory?" } + ] + } +} +``` + +**Assistant message (with tool call):** +```json +{ + "type": "message", + "id": "b2c3d4e5", + "parentId": "a1b2c3d4", + "timestamp": "2026-04-25T14:00:02.000Z", + "message": { + "role": "assistant", + "content": [ + { "type": "thinking", "thinking": "The user wants to list files..." }, + { "type": "text", "text": "I'll check that for you." }, + { "type": "toolUse", "id": "tool_abc123", "name": "bash", "input": "{\"command\":\"ls\"}" } + ], + "provider": "anthropic", + "model": "claude-sonnet-4-20250514", + "stopReason": "toolUse", + "usage": { "input": 1500, "output": 85 } + } +} +``` + +**User message (tool results):** +```json +{ + "type": "message", + "id": "c3d4e5f6", + "parentId": "b2c3d4e5", + "timestamp": "2026-04-25T14:00:03.000Z", + "message": { + "role": "user", + "content": [ + { "type": "toolResult", "toolUseId": "tool_abc123", "content": "file1.txt\nfile2.txt" } + ] + } +} +``` + +Note that tool results are content blocks on a user message, consistent with pantograph's internal model (and Anthropic's wire format). This differs from pi's approach of giving `toolResult` its own message role — in pantograph, tool results are content blocks that happen to live on user messages, just as they do in the Conversation model. + +### Content Block Serialization + +The on-disk content block types correspond directly to pantograph's internal `ContentBlock` tagged union: + +| Internal type | On-disk `type` | Fields | +|---|---|---| +| `Text` (`TextualBlock`) | `"text"` | `text: string` | +| `Thinking` (`TextualBlock`) | `"thinking"` | `thinking: string` | +| `ToolUse` (`ToolUseBlock`) | `"toolUse"` | `id: string`, `name: string`, `input: string` (raw JSON bytes) | +| `ToolResult` (`ToolResultBlock`) | `"toolResult"` | `toolUseId: string`, `content: string` | + +Notable decisions: + +- **`input` is a string, not a parsed object.** Consistent with pantograph's internal model where tool input is stored as raw JSON bytes. The on-disk format does not parse or interpret tool schemas. +- **`content` in `toolResult` is a string.** Consistent with pantograph's model where tool result content is a single `TextualBlock`. If structured content (e.g., images) is needed later, the format can evolve. +- **`thinking` stores full thinking content.** The event log records everything that happened. Thinking blocks are never omitted. + +### Assistant message metadata + +Assistant messages carry additional fields beyond `role` and `content`: + +| Field | Type | Purpose | +|---|---|---| +| `provider` | string | Which provider handled this request | +| `model` | string | Which model was used | +| `stopReason` | string | Why the model stopped: `"stop"`, `"length"`, `"toolUse"`, `"error"` | +| `usage` | object | Token counts: `{ "input": number, "output": number }` — optional, omitted if unavailable | + +These fields are metadata about the provider response. They are recorded in the event log but do **not** become part of the in-memory `Conversation` model. The `Conversation` stores only `role` and `content` — the fields needed for provider serialization. Metadata is accessed through the session entry layer when needed (display, review, debugging). + +The assistant metadata is a response-side confirmation of what the user message's `provider`/`model` already declared on the request side. It's useful for quick inspection — you can see which model produced a response without scanning backward to the preceding user message — and for detecting discrepancies (e.g., a provider routing layer returned a different model than requested). + +### Model tracking on user messages + +Every entry with `type: "message"` and `message.role: "user"` carries top-level `provider` and `model` fields recording which provider/model the request was submitted to. This includes both human-authored prompts and user messages containing tool results — both are submissions to a provider API. + +```json +{ + "type": "message", + "id": "a1b2c3d4", + "parentId": "s1s2s3s4", + "timestamp": "2026-04-25T14:00:01.000Z", + "provider": "anthropic", + "model": "claude-sonnet-4-20250514", + "message": { + "role": "user", + "content": [ + { "type": "text", "text": "What files are in this directory?" } + ] + } +} +``` + +```json +{ + "type": "message", + "id": "c3d4e5f6", + "parentId": "b2c3d4e5", + "timestamp": "2026-04-25T14:00:03.000Z", + "provider": "openai", + "model": "gpt-4o", + "message": { + "role": "user", + "content": [ + { "type": "toolResult", "toolUseId": "tool_abc123", "content": "file1.txt\nfile2.txt" } + ] + } +} +``` + +The model history is self-evident from the log: scan user messages in order and you see exactly which provider/model handled each request. The first user message's stamp is the session's initial model. + +--- + +## Tree Structure + +Entries form a tree via `id`/`parentId`: + +- First entry has `parentId: null` +- Each subsequent entry points to its parent +- In phase 4, the tree is always a simple linear chain (one path from root to leaf) +- Branching creates new children from an earlier entry (future phase) +- The "leaf" is the current position — always the last entry + +``` +[system] ─── [user] ─── [assistant] ─── [user (tool results)] ─── [assistant] ─┬─ [user] ← current leaf + │ + └─ (future branch point) +``` + +The tree structure is defined now so that the on-disk format supports branching from day one. Phase 4 simply doesn't exercise it — every session is a linear chain. + +--- + +## Rebuilding State from the Log + +On resume, pantograph reads the log top-to-bottom and rebuilds the in-memory state directly: + +1. **Read the file line by line.** Parse each line as a JSON object. +2. **Crash recovery.** If a line fails to parse, truncate the file at the start of that line (see "Crash Recovery" above). +3. **Extract the header.** First line must be a `session` header. Extract version, id, cwd. +4. **Build the entry index.** Map entry `id` → entry, track `leafId` (the last entry's id). In phase 4 entries form a linear chain, so the read order _is_ the path from root to leaf. +5. **Reconstruct the Conversation.** For each `message` entry, in order: + - Construct a `Message { role, content }` from the entry's `message` field. + - For each content block in the JSON, create the corresponding `ContentBlock` variant. + - `TextualBlock` fields (Text, Thinking) are initialized with their complete content in a single append — no streaming involved. + - `ToolUseBlock.input` is initialized with the raw JSON string from the `input` field. + - `ToolResultBlock.content` is initialized with the string from the `content` field. +6. **Reconstruct the active model.** Walk entries leaf-to-root and find the last user-message entry with `provider`/`model` fields. Those fields determine the active provider/model. A valid persisted session always contains at least one user message (the deferred-file-creation rule guarantees it). +7. **Result:** A `Conversation` ready for the agent loop, plus the active provider/model for configuration. + +There is no "event replay" — entries map 1:1 to in-memory messages, and the active model is read directly off the leaf's stamp. Compaction (phase 6) will introduce a single derived-state entry (`compaction` summarizes a prefix of the log), but that's still not event-folding; it's a special entry whose content stands in for the prefix it summarizes. + +The Conversation stores only `role` and `content` per message — the data needed for provider serialization. Entry metadata (provider, model, usage, stopReason, timestamps) lives in the session entry layer and is accessible for display or review without polluting the Conversation model. + +### Handling incomplete turns on resume + +If the session was interrupted mid-turn, the log may contain an assistant message with `ToolUse` blocks but no corresponding user message with `ToolResult` blocks. On resume, pantograph does **not** automatically re-execute the tools. The conversation is rebuilt as-is and presented to the user. The agent loop waits for user input — the user decides how to proceed. + +--- + +## Persistence Strategy + +**Append-only.** Entries are appended to the JSONL file as they occur. No full-file rewrite during normal operation (except one-time format migration). Each agent turn appends one or more entries: + +- A user message entry (when the user submits a prompt, or when tool results are assembled) +- Provider/model stamped on user message entries (top-level fields on every user message) +- An assistant message entry (when the provider response is complete) + +A single agent turn with tool calls appends multiple rounds of these entries. + +**Flush-per-entry.** Once the file exists, every completed entry is written and flushed to disk as soon as it is assembled. This maximizes recoverability — at most one in-flight entry is ever lost to a crash. + +**Deferred file creation.** The session file is **not** created at startup. Pantograph buffers the header and any pre-assistant entries in memory until the first assistant message completes. At that moment the buffered entries are flushed together and persistence switches to append-per-entry mode. If the user quits before the first assistant response, no file is created — the sessions directory stays clean of empty/single-prompt detritus. + +This is tracked by a `flushed: bool` flag on the `SessionManager`: +- `flushed = false`: entries accumulate in memory; nothing on disk yet. +- First assistant message completes → write header + all buffered entries → set `flushed = true`. +- `flushed = true`: every subsequent entry is appended and flushed immediately on completion. + +**No explicit save command.** Persistence is automatic. + +--- + +## Module Changes + +### New files + +``` +src/session.zig // On-disk entry types, JSON serialization/deserialization +src/session_manager.zig // SessionManager: file management, append, load, tree traversal, crash recovery +``` + +### `session.zig` + +Defines the on-disk types and their JSON serialization. These are separate from the in-memory `Message`/`ContentBlock` types — the on-disk types are the schema of what goes in the JSONL file, and include metadata that doesn't belong in the Conversation model. + +``` +SessionHeader = struct { + version: u32, // always 1 in phase 4 + id: []const u8, // UUIDv7 string, owned + timestamp: []const u8, // ISO 8601, owned + cwd: []const u8, // owned + + pub fn deinit(self) void +}; + +EntryBase = struct { + id: []const u8, // 8-char hex, owned + parent_id: ?[]const u8, // owned + timestamp: []const u8, // ISO 8601, owned + + pub fn deinit(self) void +}; + +SessionEntry = union(enum) { + message: MessageEntry, + pub fn base(self) EntryBase + pub fn deinit(self) void +}; + +MessageEntry = struct { + base: EntryBase, + message: DiskMessage, + // Present on user messages only. Null for system/assistant messages. + provider: ?[]const u8, // owned + model: ?[]const u8, // owned +}; + +DiskMessage = struct { + role: enum { system, user, assistant }, + content: []DiskContentBlock, + // Assistant-only metadata (null for system/user messages): + provider: ?[]const u8, + model: ?[]const u8, + stop_reason: ?[]const u8, + usage: ?Usage, + + pub fn deinit(self) void +}; + +DiskContentBlock = union(enum) { + text: DiskTextBlock, + thinking: DiskThinkingBlock, + tool_use: DiskToolUseBlock, + tool_result: DiskToolResultBlock, + + pub fn deinit(self) void +}; + +DiskTextBlock = struct { + text: []const u8, // owned +}; + +DiskThinkingBlock = struct { + thinking: []const u8, // owned +}; + +DiskToolUseBlock = struct { + id: []const u8, // owned + name: []const u8, // owned + input: []const u8, // raw JSON bytes, owned +}; + +DiskToolResultBlock = struct { + tool_use_id: []const u8, // owned + content: []const u8, // owned +}; + +Usage = struct { + input: u64, + output: u64, +}; + +FileEntry = union(enum) { + header: SessionHeader, + entry: SessionEntry, +}; +``` + +Serialization functions: + +- `serializeHeader(allocator, header) ![]const u8` — header → JSON line +- `serializeEntry(allocator, entry) ![]const u8` — entry → JSON line +- `parseLine(allocator, line: []const u8) !FileEntry` — JSON line → header or entry + +Conversion functions (bridge between on-disk and in-memory types): + +- `contentBlockToDisk(allocator, block: ContentBlock) !DiskContentBlock` — in-memory → on-disk +- `diskContentBlockToInternal(allocator, disk: DiskContentBlock) !ContentBlock` — on-disk → in-memory + +All `[]const u8` fields in on-disk types are owned copies. `deinit()` frees them. The conversion from `ContentBlock` to `DiskContentBlock` extracts the content from `TextualBlock` buffers as owned string copies. The reverse conversion creates new `TextualBlock` instances initialized with the complete content. + +### `session_manager.zig` + +``` +SessionManager = struct { + session_id: []const u8, + session_file: ?[]const u8, // path to JSONL file (set even before flush) + session_dir: []const u8, // directory containing session files for this cwd + cwd: []const u8, + entries: std.ArrayList(SessionEntry), + by_id: std.StringHashMap(*const SessionEntry), + leaf_id: ?[]const u8, + flushed: bool, // false until first assistant message persists + allocator: std.mem.Allocator, + + // ── Creation ── + + /// Create a new session. Allocates a UUIDv7, computes the target file + /// path, but does NOT write to disk yet (see deferred-file-creation). + pub fn init(allocator, cwd) !SessionManager + + /// Open an existing session file, replay the log, rebuild state. + /// Truncates from the first corrupted line onward. + /// Runs format migration if needed and rewrites the file once. + pub fn open(allocator, path) !SessionManager + + /// Find and open the most recent session for the given cwd. + /// Returns null if no sessions exist. + pub fn continueRecent(allocator, cwd) ?SessionManager + + pub fn deinit(self) void + + // ── Appending ── + + /// Append a message entry as child of the current leaf. + /// Generates an 8-char hex id (collision-checked against by_id). + /// If flushed: writes and fsyncs the new line immediately. + /// If not flushed and the message is an assistant message: writes the + /// header + all buffered entries + the new entry, then sets flushed. + /// Otherwise (not flushed, non-assistant): buffers in memory only. + /// Returns the new entry's id. + pub fn appendMessage(self, message: DiskMessage) ![]const u8 + + // ── Tree traversal ── + + pub fn getLeafId(self) ?[]const u8 + pub fn getEntry(self, id: []const u8) ?*const SessionEntry + + /// Walk from an entry to root, returning entries in root-to-leaf order. + pub fn getBranch(self, from_id: ?[]const u8) []const SessionEntry + + // ── Rebuilding ── + + /// Rebuild a Conversation from the event log. + /// Walks from leaf to root, collects message entries, + /// converts DiskContentBlocks → ContentBlocks. + pub fn rebuildConversation(self) !Conversation + + /// Determine the active provider/model by walking the path + /// and finding the last user message with provider/model fields. + /// Returns null only for sessions with no user messages, which can + /// only occur in-memory before the first flush. + pub fn activeModel(self) ?struct { provider: []const u8, model: []const u8 } + + // ── Session listing ── + + /// List sessions for a given cwd. Loads files concurrently via Io.concurrent. + /// `on_progress` is called as each file finishes parsing (loaded, total). + pub fn listSessions( + allocator, + cwd, + on_progress: ?*const fn (loaded: usize, total: usize) void, + ) ![]SessionInfo + + // ── Crash recovery ── + + /// Truncate the file at the first line that fails to parse as a complete + /// JSON object. Called during open() before parsing entries. + fn truncateFromFirstCorruption(file_path: []const u8) !void +}; + +SessionInfo = struct { + path: []const u8, // owned + id: []const u8, // owned + cwd: []const u8, // owned + created: []const u8, // ISO 8601, owned + modified: []const u8, // from file mtime, ISO 8601, owned + message_count: usize, +}; +``` + +### Modified files + +- **`agent.zig`** — Receives a `SessionManager` reference. After each message is assembled, calls `appendMessage()` with the current provider/model (user messages get stamped; assistant/system messages pass null). The agent loop is now a producer of session entries. +- **`config.zig`** — Session directory path added. Default: `$XDG_DATA_HOME/panto/sessions/` (falling back to `~/.local/share/panto/sessions/` if `XDG_DATA_HOME` is unset). Overridable via `PANTO_SESSION_DIR` environment variable. +- **`main.zig`** — `--resume` and `--resume ` flags. `sessions` subcommand. On startup with `--resume`, load session, rebuild conversation, continue agent loop. Without `--resume`, create a new session. + +--- + +## CLI Changes + +### `--resume` + +Resume the most recent session for the current working directory. If no sessions exist, create a new one. + +``` +panto --resume +``` + +### `--resume ` + +Resume a specific session by its ID (or a unique prefix of the ID). Errors if no session matches, or if the prefix is ambiguous. + +``` +panto --resume 019dc5ba +``` + +### `sessions` subcommand + +List sessions for the current working directory. Shows session ID (short), creation time, and message count. + +``` +panto sessions +``` + +Output: +``` +019dc5ba 2026-04-25 17:40 14 messages +019dc80a 2026-04-26 04:27 38 messages +``` + +--- + +## Testing Strategy + +### Unit tests + +| What | How | +|---|---| +| Entry serialization round-trip | Create entries of each type, serialize to JSON, parse back, verify equivalence | +| Content block conversion | Convert each `ContentBlock` variant to/from `DiskContentBlock`, verify content preserved | +| Crash recovery | Write a JSONL file with a corrupted trailing line, load it, verify the line is removed and valid entries are intact | +| Session rebuild | Create a session with system, user, assistant, tool result messages; rebuild conversation; verify messages match original content | +| Model tracking on user messages | Create a session with multiple turns, change model mid-session, verify `provider`/`model` fields appear on every user message entry and reflect the model used for that request | +| Tree traversal | Create entries with parent chain, walk from leaf to root, verify correct root-to-leaf order | + +### Integration test (manual) + +- Start `panto`, hold a multi-turn conversation with tool calls, quit +- Inspect the JSONL file — verify entries are present and well-formed +- Run `panto --resume` — verify conversation continues from where it left off +- Change the model mid-session, send a prompt, verify the user message entry in the log carries the updated `provider`/`model` fields +- Kill `pantograph` mid-turn (e.g., Ctrl+C during streaming), run `panto --resume` — verify session loads from last valid entry +- Run `panto sessions` — verify sessions are listed with correct metadata + +--- + +## Resolved Design Decisions + +The following were open questions in an earlier draft. Recording the resolutions here so they don't get re-litigated. + +1. **Entry ID generation**: 8-char hex IDs, generated randomly, **collision-checked** against `by_id` (the in-memory index of every entry in the session). Retry up to 100 times; in the vanishingly unlikely event of repeated collisions, fall back to a full UUID. `by_id` lookup is O(1), so the check is effectively free. +2. **Timestamp precision**: ISO 8601 with millisecond precision. Format Zig's nanoseconds down to milliseconds. +3. **Session ID**: UUIDv7 (time-ordered). Filenames are `.jsonl` with no separate timestamp prefix — lexicographic filename sort equals chronological sort. +4. **File locking**: Not needed. The crash-recovery rule (truncate from first corruption) handles interleaved writes safely: any garbled tail is discarded on next open. Two pantograph processes resuming the same session simultaneously is rare and self-corrects to whichever process wrote last cleanly. +5. **Session directory creation**: Lazy. Created on first flush (alongside the file itself). Avoids creating empty directory trees when pantograph is invoked transiently (e.g., `panto --help`, `panto sessions` on a fresh install). + +## Resolved at implementation time + +- **Large session files / replay cost**: not a concern — there's no event replay, just a 1:1 entry-to-message rebuild. Bounded by what we'd already send to the provider on every turn. The remediation for genuinely-too-many-tokens is compaction (phase 6), not a session-load optimization. +- **`fsync` cadence**: `fsync` after every completed entry. Cheap on common filesystems, and we'd rather take the crash safety than the throughput. +- **Concurrent listing**: deferred. Single-threaded for phase 4. Disks are fast, the listing path is rarely-traversed, and the public `listSessions` API already accepts a progress callback so we can parallelize later without breaking the signature. diff --git a/docs/phase-4.md b/docs/phase-4.md deleted file mode 100644 index 9dac510..0000000 --- a/docs/phase-4.md +++ /dev/null @@ -1,645 +0,0 @@ -# Phase 4: Conversation Serialization - -## Goal - -Persistent session storage as an append-only event log. Every event in a session — messages, which provider/model handled each request — is recorded to disk as it happens. On load, pantograph replays the event log to fully rebuild conversation state. Sessions survive process restarts and can be reviewed later. - -## Deliverable - -A session persistence system where pantograph saves and resumes conversations. At the end of this phase, you can: - -- Start `panto`, hold a conversation, and find the session saved to disk. -- Restart `panto --resume` and continue the conversation exactly where you left off. -- Restart `panto --resume ` to resume a specific session. -- Inspect the JSONL session file to see every event that occurred, including which provider/model handled each turn. -- Have pantograph recover gracefully from a crash — any corrupted trailing line is removed, and the session loads from the last valid entry. - -## What is usable at the end - -| Capability | How to exercise it | -|---|---| -| Auto-save session | Start `panto`, converse, quit; session is in `~/.local/share/panto/sessions/` | -| Resume most recent | `panto --resume` — continues the most recent session for the current working directory | -| Resume specific session | `panto --resume ` — opens the session with the given ID (or unique prefix) | -| List sessions | `panto sessions` — lists sessions for the current working directory | -| Crash recovery | Kill `panto` mid-turn; restart with `--resume`; corrupted trailing line is truncated, session loads from last valid entry | - -## What is explicitly out of scope - -- Compaction / context pruning (phase 6 — the format is designed to allow future entry types like `compaction` without breaking older readers, but phase 4 does not implement any form of compaction) -- Branching / tree navigation (the format supports it via `id`/`parentId`, but no CLI commands for branching exist yet) -- Session forking or cloning -- Custom entry types (for extensions — future phase) -- Labels, bookmarks, session naming -- Session export or format migration beyond version 1 -- In-memory-only sessions from the CLI (every CLI session is persisted; libpanto embedders can opt out by passing a no-op store) - ---- - -## Library / CLI Boundary - -The storage interface lives in libpanto. The default `fs_jsonl` implementation also lives in libpanto. The CLI decides _where_ a session is stored and constructs the store with that path. - -### libpanto owns - -- The `SessionStore` interface (vtable: append entry, load entries, etc.). -- An `fs_jsonl` implementation that takes a base directory at construction time and writes the JSONL format defined below. -- The agent loop's interaction with the store: which events get appended, at what points in a turn. - -### CLI owns - -- Selecting the base directory: XDG resolution (`$XDG_DATA_HOME` or `~/.local/share`), the `panto/sessions/` subpath, and the per-project `/` grouping. -- Selecting the session file: new-file-per-invocation by default; `--resume` / `--resume ` to pick an existing file. -- Constructing `fs_jsonl` with the resolved directory and passing it to libpanto. - -Other embedders (a TUI, a daemon, a Lua host, a test harness) bring their own directory policy or their own `SessionStore` implementation. A no-op store is a valid choice for embedders that don't want persistence. - -### Mid-turn writes - -The in-memory `Conversation` is the source of truth during a turn. The store is written between discrete events — never as an inline dependency of streaming. A crash mid-turn loses the in-flight assistant message but never corrupts prior state. (Phase 4 writes per-message; finer-grained streaming-event entries are out of scope.) - ---- - -## Event Log Format - -Sessions are stored as JSONL (JSON Lines) files. Each line is a JSON object with a `type` field. Entries form a tree structure via `id`/`parentId` fields, enabling in-place branching in future phases without creating new files. - -The event log is the authoritative record of a session. It captures everything that happened — not just the conversation content, but the full context of how it happened (which model was active, when it changed, etc.). On load, pantograph rebuilds all state from the log alone. - -### File Location - -``` -~/.local/share/panto/sessions//.jsonl -``` - -Where `` is the working directory with `/` replaced by `--`. This groups sessions by project directory, making it easy to find sessions for a given project. - -The filename is just the session's UUIDv7. Because UUIDv7 is time-ordered, sorting filenames lexicographically sorts sessions chronologically — no separate timestamp prefix is needed. - -The base directory respects `XDG_DATA_HOME`, falling back to `~/.local/share` if unset (per the XDG Base Directory specification). - -Example: -``` -~/.local/share/panto/sessions/--Users-travis-Code-pantograph--/019dc5ba-53f6-71a5-ab8f-b1f8709c2572.jsonl -``` - -### Crash Recovery - -On loading a session, pantograph parses lines one by one from the top. As soon as a line fails to parse as a complete JSON object, **that line and every line after it are truncated from the file**. The session loads from the last fully valid entry. - -This is deliberately stricter than "skip malformed lines." Tool-use/tool-result pairs are stateful — a missing or partial entry in the middle would leave subsequent entries referring to a `toolUseId` that doesn't exist, which providers (notably Anthropic) reject outright. The only safe resume point is immediately before the first corruption. - -In practice the corruption window is the tail of the file (a crash mid-write), so truncation is effectively the same as "drop the partial last line." Defining the rule by position-of-first-error rather than "last line" handles edge cases (e.g., a partial write followed by a clean flush by another process) without ambiguity. - ---- - -## Entry Types - -### SessionHeader - -First (and only) header line of the file. Metadata only — not part of the tree (no `id`/`parentId`). Exactly one header per file, always at the top. - -```json -{ - "type": "session", - "version": 1, - "id": "019dc5ba-53f6-71a5-ab8f-b1f8709c2572", - "timestamp": "2026-04-25T17:40:15.990Z", - "cwd": "/Users/travis/Code/pantograph", - "provider": "anthropic", - "model": "claude-sonnet-4-20250514" -} -``` - -| Field | Type | Purpose | -|---|---|---| -| `version` | number | Format version. `1` for phase 4. | -| `id` | string | Session UUIDv7. | -| `timestamp` | string | ISO 8601 timestamp of session creation. | -| `cwd` | string | Working directory where the session was started. | -| `provider` | string | Initial provider (`"openai"` or `"anthropic"`). | -| `model` | string | Initial model name. | - -The `provider` and `model` fields serve as the baseline for model tracking. If no user messages have been submitted yet, the header's values are the active provider/model. - -### Versioning and migration - -The `version` field exists from day one so future format changes are explicit. Phase 4 is version 1. - -When a future pantograph version bumps the format to 2+, the load path runs `migrate(entries) -> bool`. If migration mutates anything, the entire file is rewritten once with the new version stamped on the header. Migration is eager: every v1 file encountered by a v2-capable pantograph is upgraded immediately on first load. There is never a mixed-version file. - -### EntryBase - -All entries (except the header) extend this base: - -``` -EntryBase = { - type: string, // entry type discriminator - id: string, // 8-char hex ID, random - parentId: string | null, // parent entry ID (null for first entry) - timestamp: string, // ISO 8601 -} -``` - -The first entry has `parentId: null`. Each subsequent entry points to its parent, forming a linked path from root to leaf. In phase 4 this is always a linear chain; the tree structure exists to support branching in future phases. - -### MessageEntry - -A message in the conversation. The `message` field contains role, content blocks, and (for assistant messages) provider metadata. - -**System message:** -```json -{ - "type": "message", - "id": "s1s2s3s4", - "parentId": null, - "timestamp": "2026-04-25T14:00:00.000Z", - "message": { - "role": "system", - "content": [ - { "type": "text", "text": "You are a helpful assistant." } - ] - } -} -``` - -**User message (human prompt):** -```json -{ - "type": "message", - "id": "a1b2c3d4", - "parentId": "s1s2s3s4", - "timestamp": "2026-04-25T14:00:01.000Z", - "message": { - "role": "user", - "content": [ - { "type": "text", "text": "What files are in this directory?" } - ] - } -} -``` - -**Assistant message (with tool call):** -```json -{ - "type": "message", - "id": "b2c3d4e5", - "parentId": "a1b2c3d4", - "timestamp": "2026-04-25T14:00:02.000Z", - "message": { - "role": "assistant", - "content": [ - { "type": "thinking", "thinking": "The user wants to list files..." }, - { "type": "text", "text": "I'll check that for you." }, - { "type": "toolUse", "id": "tool_abc123", "name": "bash", "input": "{\"command\":\"ls\"}" } - ], - "provider": "anthropic", - "model": "claude-sonnet-4-20250514", - "stopReason": "toolUse", - "usage": { "input": 1500, "output": 85 } - } -} -``` - -**User message (tool results):** -```json -{ - "type": "message", - "id": "c3d4e5f6", - "parentId": "b2c3d4e5", - "timestamp": "2026-04-25T14:00:03.000Z", - "message": { - "role": "user", - "content": [ - { "type": "toolResult", "toolUseId": "tool_abc123", "content": "file1.txt\nfile2.txt" } - ] - } -} -``` - -Note that tool results are content blocks on a user message, consistent with pantograph's internal model (and Anthropic's wire format). This differs from pi's approach of giving `toolResult` its own message role — in pantograph, tool results are content blocks that happen to live on user messages, just as they do in the Conversation model. - -### Content Block Serialization - -The on-disk content block types correspond directly to pantograph's internal `ContentBlock` tagged union: - -| Internal type | On-disk `type` | Fields | -|---|---|---| -| `Text` (`TextualBlock`) | `"text"` | `text: string` | -| `Thinking` (`TextualBlock`) | `"thinking"` | `thinking: string` | -| `ToolUse` (`ToolUseBlock`) | `"toolUse"` | `id: string`, `name: string`, `input: string` (raw JSON bytes) | -| `ToolResult` (`ToolResultBlock`) | `"toolResult"` | `toolUseId: string`, `content: string` | - -Notable decisions: - -- **`input` is a string, not a parsed object.** Consistent with pantograph's internal model where tool input is stored as raw JSON bytes. The on-disk format does not parse or interpret tool schemas. -- **`content` in `toolResult` is a string.** Consistent with pantograph's model where tool result content is a single `TextualBlock`. If structured content (e.g., images) is needed later, the format can evolve. -- **`thinking` stores full thinking content.** The event log records everything that happened. Thinking blocks are never omitted. - -### Assistant message metadata - -Assistant messages carry additional fields beyond `role` and `content`: - -| Field | Type | Purpose | -|---|---|---| -| `provider` | string | Which provider handled this request | -| `model` | string | Which model was used | -| `stopReason` | string | Why the model stopped: `"stop"`, `"length"`, `"toolUse"`, `"error"` | -| `usage` | object | Token counts: `{ "input": number, "output": number }` — optional, omitted if unavailable | - -These fields are metadata about the provider response. They are recorded in the event log but do **not** become part of the in-memory `Conversation` model. The `Conversation` stores only `role` and `content` — the fields needed for provider serialization. Metadata is accessed through the session entry layer when needed (display, review, debugging). - -The assistant metadata is a response-side confirmation of what the user message's `provider`/`model` already declared on the request side. It's useful for quick inspection — you can see which model produced a response without scanning backward to the preceding user message — and for detecting discrepancies (e.g., a provider routing layer returned a different model than requested). - -### Model tracking on user messages - -Every entry with `type: "message"` and `message.role: "user"` carries top-level `provider` and `model` fields recording which provider/model the request was submitted to. This includes both human-authored prompts and user messages containing tool results — both are submissions to a provider API. - -```json -{ - "type": "message", - "id": "a1b2c3d4", - "parentId": "s1s2s3s4", - "timestamp": "2026-04-25T14:00:01.000Z", - "provider": "anthropic", - "model": "claude-sonnet-4-20250514", - "message": { - "role": "user", - "content": [ - { "type": "text", "text": "What files are in this directory?" } - ] - } -} -``` - -```json -{ - "type": "message", - "id": "c3d4e5f6", - "parentId": "b2c3d4e5", - "timestamp": "2026-04-25T14:00:03.000Z", - "provider": "openai", - "model": "gpt-4o", - "message": { - "role": "user", - "content": [ - { "type": "toolResult", "toolUseId": "tool_abc123", "content": "file1.txt\nfile2.txt" } - ] - } -} -``` - -The model history is self-evident from the log: scan user messages in order and you see exactly which provider/model handled each request. The header's `provider`/`model` fields serve as the baseline for the first turn — if no user messages have been submitted yet, the header's values are the active model. - ---- - -## Tree Structure - -Entries form a tree via `id`/`parentId`: - -- First entry has `parentId: null` -- Each subsequent entry points to its parent -- In phase 4, the tree is always a simple linear chain (one path from root to leaf) -- Branching creates new children from an earlier entry (future phase) -- The "leaf" is the current position — always the last entry - -``` -[system] ─── [user] ─── [assistant] ─── [user (tool results)] ─── [assistant] ─┬─ [user] ← current leaf - │ - └─ (future branch point) -``` - -The tree structure is defined now so that the on-disk format supports branching from day one. Phase 4 simply doesn't exercise it — every session is a linear chain. - ---- - -## Rebuilding State from the Log - -On resume, pantograph fully rebuilds conversation state by replaying the event log: - -1. **Read the file line by line.** Parse each line as a JSON object. -2. **Crash recovery.** If the last line doesn't parse as a complete JSON object, delete it from the file. -3. **Extract the header.** First line must be a `session` header. Extract version, id, cwd, initial provider/model. -4. **Build the entry index.** Map entry `id` → entry, track `leafId` (the last entry's id). -5. **Walk the tree.** From leaf to root, collect entries in root-to-leaf order. -6. **Reconstruct the Conversation.** For each `message` entry on the path: - - Construct a `Message { role, content }` from the entry's `message` field. - - For each content block in the JSON, create the corresponding `ContentBlock` variant. - - `TextualBlock` fields (Text, Thinking) are initialized with their complete content in a single append — no streaming involved. - - `ToolUseBlock.input` is initialized with the raw JSON string from the `input` field. - - `ToolResultBlock.content` is initialized with the string from the `content` field. -7. **Reconstruct the active model.** Walk the path and find the last user message entry with `provider`/`model` fields. Those fields (or the header defaults if no user message exists) determine the active provider/model. -8. **Result:** A `Conversation` ready for the agent loop, plus the active provider/model for configuration. - -The Conversation stores only `role` and `content` per message — the data needed for provider serialization. Entry metadata (provider, model, usage, stopReason, timestamps) lives in the session entry layer and is accessible for display or review without polluting the Conversation model. - -### Handling incomplete turns on resume - -If the session was interrupted mid-turn, the log may contain an assistant message with `ToolUse` blocks but no corresponding user message with `ToolResult` blocks. On resume, pantograph does **not** automatically re-execute the tools. The conversation is rebuilt as-is and presented to the user. The agent loop waits for user input — the user decides how to proceed. - ---- - -## Persistence Strategy - -**Append-only.** Entries are appended to the JSONL file as they occur. No full-file rewrite during normal operation (except one-time format migration). Each agent turn appends one or more entries: - -- A user message entry (when the user submits a prompt, or when tool results are assembled) -- Provider/model stamped on user message entries (top-level fields on every user message) -- An assistant message entry (when the provider response is complete) - -A single agent turn with tool calls appends multiple rounds of these entries. - -**Flush-per-entry.** Once the file exists, every completed entry is written and flushed to disk as soon as it is assembled. This maximizes recoverability — at most one in-flight entry is ever lost to a crash. - -**Deferred file creation.** The session file is **not** created at startup. Pantograph buffers the header and any pre-assistant entries in memory until the first assistant message completes. At that moment the buffered entries are flushed together and persistence switches to append-per-entry mode. If the user quits before the first assistant response, no file is created — the sessions directory stays clean of empty/single-prompt detritus. - -This is tracked by a `flushed: bool` flag on the `SessionManager`: -- `flushed = false`: entries accumulate in memory; nothing on disk yet. -- First assistant message completes → write header + all buffered entries → set `flushed = true`. -- `flushed = true`: every subsequent entry is appended and flushed immediately on completion. - -**No explicit save command.** Persistence is automatic. - ---- - -## Module Changes - -### New files - -``` -src/session.zig // On-disk entry types, JSON serialization/deserialization -src/session_manager.zig // SessionManager: file management, append, load, tree traversal, crash recovery -``` - -### `session.zig` - -Defines the on-disk types and their JSON serialization. These are separate from the in-memory `Message`/`ContentBlock` types — the on-disk types are the schema of what goes in the JSONL file, and include metadata that doesn't belong in the Conversation model. - -``` -SessionHeader = struct { - version: u32, // always 1 in phase 4 - id: []const u8, // UUIDv7 string, owned - timestamp: []const u8, // ISO 8601, owned - cwd: []const u8, // owned - provider: []const u8, // owned - model: []const u8, // owned - - pub fn deinit(self) void -}; - -EntryBase = struct { - id: []const u8, // 8-char hex, owned - parent_id: ?[]const u8, // owned - timestamp: []const u8, // ISO 8601, owned - - pub fn deinit(self) void -}; - -SessionEntry = union(enum) { - message: MessageEntry, - pub fn base(self) EntryBase - pub fn deinit(self) void -}; - -MessageEntry = struct { - base: EntryBase, - message: DiskMessage, - // Present on user messages only. Null for system/assistant messages. - provider: ?[]const u8, // owned - model: ?[]const u8, // owned -}; - -DiskMessage = struct { - role: enum { system, user, assistant }, - content: []DiskContentBlock, - // Assistant-only metadata (null for system/user messages): - provider: ?[]const u8, - model: ?[]const u8, - stop_reason: ?[]const u8, - usage: ?Usage, - - pub fn deinit(self) void -}; - -DiskContentBlock = union(enum) { - text: DiskTextBlock, - thinking: DiskThinkingBlock, - tool_use: DiskToolUseBlock, - tool_result: DiskToolResultBlock, - - pub fn deinit(self) void -}; - -DiskTextBlock = struct { - text: []const u8, // owned -}; - -DiskThinkingBlock = struct { - thinking: []const u8, // owned -}; - -DiskToolUseBlock = struct { - id: []const u8, // owned - name: []const u8, // owned - input: []const u8, // raw JSON bytes, owned -}; - -DiskToolResultBlock = struct { - tool_use_id: []const u8, // owned - content: []const u8, // owned -}; - -Usage = struct { - input: u64, - output: u64, -}; - -FileEntry = union(enum) { - header: SessionHeader, - entry: SessionEntry, -}; -``` - -Serialization functions: - -- `serializeHeader(allocator, header) ![]const u8` — header → JSON line -- `serializeEntry(allocator, entry) ![]const u8` — entry → JSON line -- `parseLine(allocator, line: []const u8) !FileEntry` — JSON line → header or entry - -Conversion functions (bridge between on-disk and in-memory types): - -- `contentBlockToDisk(allocator, block: ContentBlock) !DiskContentBlock` — in-memory → on-disk -- `diskContentBlockToInternal(allocator, disk: DiskContentBlock) !ContentBlock` — on-disk → in-memory - -All `[]const u8` fields in on-disk types are owned copies. `deinit()` frees them. The conversion from `ContentBlock` to `DiskContentBlock` extracts the content from `TextualBlock` buffers as owned string copies. The reverse conversion creates new `TextualBlock` instances initialized with the complete content. - -### `session_manager.zig` - -``` -SessionManager = struct { - session_id: []const u8, - session_file: ?[]const u8, // path to JSONL file (set even before flush) - session_dir: []const u8, // directory containing session files for this cwd - cwd: []const u8, - entries: std.ArrayList(SessionEntry), - by_id: std.StringHashMap(*const SessionEntry), - leaf_id: ?[]const u8, - flushed: bool, // false until first assistant message persists - allocator: std.mem.Allocator, - - // ── Creation ── - - /// Create a new session. Allocates a UUIDv7, computes the target file - /// path, but does NOT write to disk yet (see deferred-file-creation). - pub fn init(allocator, cwd, provider, model) !SessionManager - - /// Open an existing session file, replay the log, rebuild state. - /// Truncates from the first corrupted line onward. - /// Runs format migration if needed and rewrites the file once. - pub fn open(allocator, path) !SessionManager - - /// Find and open the most recent session for the given cwd. - /// Returns null if no sessions exist. - pub fn continueRecent(allocator, cwd) ?SessionManager - - pub fn deinit(self) void - - // ── Appending ── - - /// Append a message entry as child of the current leaf. - /// Generates an 8-char hex id (collision-checked against by_id). - /// If flushed: writes and fsyncs the new line immediately. - /// If not flushed and the message is an assistant message: writes the - /// header + all buffered entries + the new entry, then sets flushed. - /// Otherwise (not flushed, non-assistant): buffers in memory only. - /// Returns the new entry's id. - pub fn appendMessage(self, message: DiskMessage) ![]const u8 - - // ── Tree traversal ── - - pub fn getLeafId(self) ?[]const u8 - pub fn getEntry(self, id: []const u8) ?*const SessionEntry - - /// Walk from an entry to root, returning entries in root-to-leaf order. - pub fn getBranch(self, from_id: ?[]const u8) []const SessionEntry - - // ── Rebuilding ── - - /// Rebuild a Conversation from the event log. - /// Walks from leaf to root, collects message entries, - /// converts DiskContentBlocks → ContentBlocks. - pub fn rebuildConversation(self) !Conversation - - /// Determine the active provider/model by walking the path - /// and finding the last user message with provider/model fields, - /// falling back to header defaults. - pub fn activeModel(self) struct { provider: []const u8, model: []const u8 } - - // ── Session listing ── - - /// List sessions for a given cwd. Loads files concurrently via Io.concurrent. - /// `on_progress` is called as each file finishes parsing (loaded, total). - pub fn listSessions( - allocator, - cwd, - on_progress: ?*const fn (loaded: usize, total: usize) void, - ) ![]SessionInfo - - // ── Crash recovery ── - - /// Truncate the file at the first line that fails to parse as a complete - /// JSON object. Called during open() before parsing entries. - fn truncateFromFirstCorruption(file_path: []const u8) !void -}; - -SessionInfo = struct { - path: []const u8, // owned - id: []const u8, // owned - cwd: []const u8, // owned - created: []const u8, // ISO 8601, owned - modified: []const u8, // from file mtime, ISO 8601, owned - message_count: usize, -}; -``` - -### Modified files - -- **`agent.zig`** — Receives a `SessionManager` reference. After each message is assembled, calls `appendMessage()` with the current provider/model (user messages get stamped; assistant/system messages pass null). The agent loop is now a producer of session entries. -- **`config.zig`** — Session directory path added. Default: `$XDG_DATA_HOME/panto/sessions/` (falling back to `~/.local/share/panto/sessions/` if `XDG_DATA_HOME` is unset). Overridable via `PANTO_SESSION_DIR` environment variable. -- **`main.zig`** — `--resume` and `--resume ` flags. `sessions` subcommand. On startup with `--resume`, load session, rebuild conversation, continue agent loop. Without `--resume`, create a new session. - ---- - -## CLI Changes - -### `--resume` - -Resume the most recent session for the current working directory. If no sessions exist, create a new one. - -``` -panto --resume -``` - -### `--resume ` - -Resume a specific session by its ID (or a unique prefix of the ID). Errors if no session matches, or if the prefix is ambiguous. - -``` -panto --resume 019dc5ba -``` - -### `sessions` subcommand - -List sessions for the current working directory. Shows session ID (short), creation time, and message count. - -``` -panto sessions -``` - -Output: -``` -019dc5ba 2026-04-25 17:40 14 messages -019dc80a 2026-04-26 04:27 38 messages -``` - ---- - -## Testing Strategy - -### Unit tests - -| What | How | -|---|---| -| Entry serialization round-trip | Create entries of each type, serialize to JSON, parse back, verify equivalence | -| Content block conversion | Convert each `ContentBlock` variant to/from `DiskContentBlock`, verify content preserved | -| Crash recovery | Write a JSONL file with a corrupted trailing line, load it, verify the line is removed and valid entries are intact | -| Session rebuild | Create a session with system, user, assistant, tool result messages; rebuild conversation; verify messages match original content | -| Model tracking on user messages | Create a session with multiple turns, change model mid-session, verify `provider`/`model` fields appear on every user message entry and reflect the model used for that request | -| Tree traversal | Create entries with parent chain, walk from leaf to root, verify correct root-to-leaf order | - -### Integration test (manual) - -- Start `panto`, hold a multi-turn conversation with tool calls, quit -- Inspect the JSONL file — verify entries are present and well-formed -- Run `panto --resume` — verify conversation continues from where it left off -- Change the model mid-session, send a prompt, verify the user message entry in the log carries the updated `provider`/`model` fields -- Kill `pantograph` mid-turn (e.g., Ctrl+C during streaming), run `panto --resume` — verify session loads from last valid entry -- Run `panto sessions` — verify sessions are listed with correct metadata - ---- - -## Resolved Design Decisions - -The following were open questions in an earlier draft. Recording the resolutions here so they don't get re-litigated. - -1. **Entry ID generation**: 8-char hex IDs, generated randomly, **collision-checked** against `by_id` (the in-memory index of every entry in the session). Retry up to 100 times; in the vanishingly unlikely event of repeated collisions, fall back to a full UUID. `by_id` lookup is O(1), so the check is effectively free. -2. **Timestamp precision**: ISO 8601 with millisecond precision. Format Zig's nanoseconds down to milliseconds. -3. **Session ID**: UUIDv7 (time-ordered). Filenames are `.jsonl` with no separate timestamp prefix — lexicographic filename sort equals chronological sort. -4. **File locking**: Not needed. The crash-recovery rule (truncate from first corruption) handles interleaved writes safely: any garbled tail is discarded on next open. Two pantograph processes resuming the same session simultaneously is rare and self-corrects to whichever process wrote last cleanly. -5. **Session directory creation**: Lazy. Created on first flush (alongside the file itself). Avoids creating empty directory trees when pantograph is invoked transiently (e.g., `panto --help`, `panto sessions` on a fresh install). - -## Open Questions (to resolve during implementation) - -1. **Large session files**: A long coding session can produce thousands of entries and megabytes of JSONL. Replaying the entire log on every resume could become slow. Should we cache the rebuilt `Conversation` state (e.g., as a checkpoint line at the end of the file) and only replay new entries on subsequent loads? **Defer**: measure first; add only if real sessions hit a noticeable load delay. -2. **`fsync` cadence**: We flush after every completed entry (per the persistence strategy). Do we also `fsync` the file handle, or trust the kernel's writeback? `fsync` is safer but slower. Likely settle on `fsync` after assistant messages and tool-result user messages (the entries that matter for resume), not after every streaming-derived write — but phase 4 only writes once per completed entry, so a single `fsync` per `appendMessage` is probably fine. -3. **Concurrent listing**: Use `Io.concurrent` with what concurrency cap? Pi uses 10. Probably mirror that until we have data. diff --git a/libpanto/src/agent.zig b/libpanto/src/agent.zig index bbcc54c..92c5a3a 100644 --- a/libpanto/src/agent.zig +++ b/libpanto/src/agent.zig @@ -591,7 +591,7 @@ const NoopReceiver = struct { fn noopToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {} fn noop3(_: *anyopaque, _: usize, _: []const u8) anyerror!void {} fn noop4(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {} - fn noop5(_: *anyopaque, _: conversation.Message) anyerror!void {} + fn noop5(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {} fn noop6(_: *anyopaque, _: anyerror) void {} }; diff --git a/libpanto/src/anthropic_messages_json.zig b/libpanto/src/anthropic_messages_json.zig index 708ede5..46ec17c 100644 --- a/libpanto/src/anthropic_messages_json.zig +++ b/libpanto/src/anthropic_messages_json.zig @@ -232,9 +232,31 @@ pub const StreamEventTag = enum { pub const ContentBlockKind = enum { text, thinking, tool_use, unknown }; +/// Partial token-count snapshot emitted in `message_start` and +/// `message_delta`. Field semantics follow Anthropic's wire format: +/// +/// - `input_tokens`: prompt tokens billed at the base rate. +/// - `cache_creation_input_tokens`: prompt tokens written to a new +/// cache entry (1.25× base). +/// - `cache_read_input_tokens`: prompt tokens served from cache +/// (0.1× base). +/// - `output_tokens`: response tokens billed at the output rate. +/// +/// `message_start.usage` carries the initial input-side counts (output +/// will be 0 or absent); `message_delta.usage` carries the final +/// `output_tokens` and may repeat the input-side counts. The provider +/// merges them — "missing" means "unchanged," not "reset to zero." +pub const StreamUsage = struct { + input_tokens: ?u64 = null, + output_tokens: ?u64 = null, + cache_creation_input_tokens: ?u64 = null, + cache_read_input_tokens: ?u64 = null, +}; + pub const StreamEvent = union(StreamEventTag) { message_start: struct { role: ?[]const u8 = null, + usage: StreamUsage = .{}, }, content_block_start: struct { index: usize, @@ -255,6 +277,7 @@ pub const StreamEvent = union(StreamEventTag) { }, message_delta: struct { stop_reason: ?[]const u8 = null, + usage: StreamUsage = .{}, }, message_stop: void, ping: void, @@ -287,14 +310,18 @@ pub fn parseStreamEvent(allocator: Allocator, payload: []const u8) !ParsedStream const ty = type_v.string; if (std.mem.eql(u8, ty, "message_start")) { var role: ?[]const u8 = null; + var usage: StreamUsage = .{}; if (root.object.get("message")) |m| { if (m == .object) { if (m.object.get("role")) |r| { if (r == .string) role = r.string; } + if (m.object.get("usage")) |u| { + if (u == .object) usage = parseStreamUsage(u.object); + } } } - return .{ .parsed = parsed, .event = .{ .message_start = .{ .role = role } } }; + return .{ .parsed = parsed, .event = .{ .message_start = .{ .role = role, .usage = usage } } }; } if (std.mem.eql(u8, ty, "content_block_start")) { @@ -380,6 +407,7 @@ pub fn parseStreamEvent(allocator: Allocator, payload: []const u8) !ParsedStream if (std.mem.eql(u8, ty, "message_delta")) { var stop_reason: ?[]const u8 = null; + var usage: StreamUsage = .{}; if (root.object.get("delta")) |d| { if (d == .object) { if (d.object.get("stop_reason")) |sr| { @@ -387,7 +415,11 @@ pub fn parseStreamEvent(allocator: Allocator, payload: []const u8) !ParsedStream } } } - return .{ .parsed = parsed, .event = .{ .message_delta = .{ .stop_reason = stop_reason } } }; + // `usage` is on the message_delta event itself, not under `delta`. + if (root.object.get("usage")) |u| { + if (u == .object) usage = parseStreamUsage(u.object); + } + return .{ .parsed = parsed, .event = .{ .message_delta = .{ .stop_reason = stop_reason, .usage = usage } } }; } if (std.mem.eql(u8, ty, "message_stop")) { @@ -417,6 +449,22 @@ pub fn parseStreamEvent(allocator: Allocator, payload: []const u8) !ParsedStream return .{ .parsed = parsed, .event = .unknown }; } +fn parseStreamUsage(obj: std.json.ObjectMap) StreamUsage { + return .{ + .input_tokens = readOptionalU64(obj, "input_tokens"), + .output_tokens = readOptionalU64(obj, "output_tokens"), + .cache_creation_input_tokens = readOptionalU64(obj, "cache_creation_input_tokens"), + .cache_read_input_tokens = readOptionalU64(obj, "cache_read_input_tokens"), + }; +} + +fn readOptionalU64(obj: std.json.ObjectMap, name: []const u8) ?u64 { + const v = obj.get(name) orelse return null; + if (v != .integer) return null; + if (v.integer < 0) return null; + return @intCast(v.integer); +} + fn readIndex(root: std.json.Value) ?usize { const v = root.object.get("index") orelse return null; if (v != .integer) return null; diff --git a/libpanto/src/openai_chat_json.zig b/libpanto/src/openai_chat_json.zig index 9531131..37d347b 100644 --- a/libpanto/src/openai_chat_json.zig +++ b/libpanto/src/openai_chat_json.zig @@ -31,6 +31,31 @@ pub const StreamDelta = struct { /// treat the turn as failed. error_message: ?[]const u8 = null, error_type: ?[]const u8 = null, + /// Token usage from the final chunk's top-level `usage` block. + /// Only present on the final chunk when the request was sent with + /// `stream_options.include_usage: true`. Earlier chunks have null. + usage: ?StreamUsage = null, +}; + +/// Token usage payload from OpenAI's terminating SSE chunk. Field +/// semantics: +/// +/// - `prompt_tokens`: total input tokens, **including** cached tokens. +/// - `completion_tokens`: output tokens (including reasoning tokens). +/// - `cached_prompt_tokens`: subset of `prompt_tokens` served from +/// the prompt cache (billed at a discount). +/// - `reasoning_tokens`: subset of `completion_tokens` spent on +/// internal reasoning (o-series models). +/// +/// To map to `Usage`: `input = prompt_tokens - cached_prompt_tokens`, +/// `cache_read = cached_prompt_tokens`, `output = completion_tokens`, +/// `reasoning = reasoning_tokens`, `cache_write = 0` (OpenAI doesn't +/// bill a cache-write premium). +pub const StreamUsage = struct { + prompt_tokens: ?u64 = null, + completion_tokens: ?u64 = null, + cached_prompt_tokens: ?u64 = null, + reasoning_tokens: ?u64 = null, }; /// A single entry from a streaming `tool_calls` array. Multiple parallel @@ -66,6 +91,17 @@ pub fn serializeRequest( try s.objectField("stream"); try s.write(true); + // Ask for the final-chunk usage block. Without this the server + // sends `usage: null` and we can't stamp token counts on the + // session log. Most OpenAI-compatible proxies accept this; ones + // that don't will either ignore it or 400 — in the latter case + // the user has bigger problems than missing token counts. + try s.objectField("stream_options"); + try s.beginObject(); + try s.objectField("include_usage"); + try s.write(true); + try s.endObject(); + switch (cfg.reasoning) { .default => {}, .off => { @@ -285,6 +321,27 @@ pub fn parseStreamEvent(allocator: Allocator, payload: []const u8) !ParsedDelta // Top-level `error` field. Some providers (and OpenAI itself on rare // mid-stream failures) emit `data: {"error":{"message":...,"type":...}}` // with HTTP 200, so we look for this BEFORE the choices array. + // Top-level `usage` field on the terminating chunk. Independent of + // the (often empty) choices array. + if (root.object.get("usage")) |u| { + if (u == .object) { + var su: StreamUsage = .{}; + su.prompt_tokens = readOptU64(u.object, "prompt_tokens"); + su.completion_tokens = readOptU64(u.object, "completion_tokens"); + if (u.object.get("prompt_tokens_details")) |ptd| { + if (ptd == .object) { + su.cached_prompt_tokens = readOptU64(ptd.object, "cached_tokens"); + } + } + if (u.object.get("completion_tokens_details")) |ctd| { + if (ctd == .object) { + su.reasoning_tokens = readOptU64(ctd.object, "reasoning_tokens"); + } + } + delta.usage = su; + } + } + if (root.object.get("error")) |e| { switch (e) { .object => |obj| { @@ -375,6 +432,13 @@ pub fn parseStreamEvent(allocator: Allocator, payload: []const u8) !ParsedDelta }; } +fn readOptU64(obj: std.json.ObjectMap, name: []const u8) ?u64 { + const v = obj.get(name) orelse return null; + if (v != .integer) return null; + if (v.integer < 0) return null; + return @intCast(v.integer); +} + // ----------------------------------------------------------------------------- // Tests // ----------------------------------------------------------------------------- diff --git a/libpanto/src/pricing.zig b/libpanto/src/pricing.zig new file mode 100644 index 0000000..7d4e1f8 --- /dev/null +++ b/libpanto/src/pricing.zig @@ -0,0 +1,303 @@ +//! Per-(provider, model) token pricing and cost calculation. +//! +//! Prices are stored as integers — specifically, *micro-cents per token* +//! (1/1,000,000 of a cent), equivalently picodollars (10^-12 USD) per +//! token. Reasons: +//! +//! - All commonly-quoted "USD per million tokens" prices land on round +//! integers under the conversion: $3.00 / 1M tokens = 300 +//! micro-cents per token. +//! - Per-message token counts (10^2 to 10^5) multiplied by per-token +//! prices (~10^2) stay well inside `u64` for entire long-running +//! sessions — a session worth ~$1 (10^14 micro-cents) is nowhere +//! near `u64.max` (~1.8 × 10^19). +//! - Summing per-turn costs across a session is exact: no +//! floating-point drift. +//! +//! The TOML on-disk format lets users write `input = 3.0` for $3/Mtok, +//! which is the natural unit for humans. The loader multiplies by 100 +//! and rounds to the nearest integer; the rounding handles parse-time +//! float-precision wobble. +//! +//! `costMicroCents` does the integer arithmetic. Each `Pricing` field +//! is `?u64`: `null` means "we don't know the price for this token +//! category," which is distinct from a price of zero (e.g. OpenAI does +//! charge nothing for cache writes — that's a known 0, not unknown). +//! `costMicroCents` returns `?u64`: if any usage category with nonzero +//! count maps to a `null` price, the whole turn cost goes to `null` +//! ("unknown"), and any session-level sum that includes that turn must +//! likewise degenerate to `null`. This prevents silently treating +//! unknown costs as free. +//! +//! The display layer lives in the CLI; libpanto only computes. + +const std = @import("std"); + +const session_mod = @import("session.zig"); +pub const Usage = session_mod.Usage; + +// ============================================================================= +// Pricing struct +// ============================================================================= + +/// Per-token pricing for a single (provider, model) pair. +/// +/// Units: micro-cents per token (1/1,000,000 of a cent per token), aka +/// picodollars (10^-12 USD) per token. +/// +/// Conversion from "USD per million tokens": +/// +/// micro_cents_per_token = round(USD_per_Mtok * 100) +/// +/// So $3.00/Mtok → 300 micro-cents/token. +pub const Pricing = struct { + /// Per fresh (uncached, non-cache-written) input token. `null` = + /// unknown (e.g. field omitted from `models.toml`). + input: ?u64 = null, + /// Per output token. `null` = unknown. + output: ?u64 = null, + /// Per cache-read input token. Typically a fraction of `input` + /// (0.1× on Anthropic, 0.5× on OpenAI for cached prompt tokens). + /// `null` = unknown. + cache_read: ?u64 = null, + /// Per cache-write input token. Anthropic charges a premium + /// (1.25× `input`); OpenAI doesn't bill a cache-write rate (a + /// known 0, which should be written `cache_write = 0` rather than + /// omitted). `null` = unknown. + cache_write: ?u64 = null, + + /// Convert a USD-per-million-tokens float (the human-friendly unit) + /// to the internal integer representation. Rounds to nearest. + /// + /// `dollars_per_mtok * 1_000_000 cents/dollar / 1_000_000 tokens` = + /// cents-per-token, then * 1_000_000 micro-cents/cent = + /// micro-cents-per-token. The 1_000_000s cancel, leaving + /// `dollars_per_mtok * 100`. + /// + /// Non-finite inputs round to 0. Negative inputs are clamped to 0 + /// (treated as a known free price, not unknown). + pub fn fromDollarsPerMtok(dollars_per_mtok: f64) u64 { + if (!std.math.isFinite(dollars_per_mtok) or dollars_per_mtok <= 0) return 0; + const scaled = dollars_per_mtok * 100.0; + const r = @round(scaled); + if (r >= @as(f64, @floatFromInt(std.math.maxInt(u64)))) return std.math.maxInt(u64); + return @intFromFloat(r); + } +}; + +// ============================================================================= +// Cost calculation +// ============================================================================= + +/// Compute the cost of a single turn's `usage` under the given `pricing`, +/// in micro-cents. Returns `null` if any usage category with a nonzero +/// token count maps to a `null` price — i.e. "we used some cache reads +/// but we don't know the cache-read price" poisons the whole turn cost +/// to "unknown". Categories with zero tokens are ignored regardless of +/// whether their price is known, so e.g. a model with `cache_write = +/// null` still produces a known cost on turns that never write to +/// cache. +/// +/// `reasoning` does NOT contribute separately — it's already counted +/// inside `output`. +pub fn costMicroCents(usage: Usage, pricing: Pricing) ?u64 { + var total: u64 = 0; + total +%= component(usage.input, pricing.input) orelse return null; + total +%= component(usage.output, pricing.output) orelse return null; + total +%= component(usage.cache_read, pricing.cache_read) orelse return null; + total +%= component(usage.cache_write, pricing.cache_write) orelse return null; + return total; +} + +/// Cost contribution (in micro-cents) from a single (tokens, price) +/// pair. Returns `0` when `tokens == 0` regardless of whether `price` +/// is known — so unknown prices don't poison turns that never used +/// that category. Returns `null` when tokens are nonzero but the +/// price is unknown; callers convert that to a `null` total. +fn component(tokens: u64, price: ?u64) ?u64 { + if (tokens == 0) return 0; + const p = price orelse return null; + return tokens *% p; +} + +// ============================================================================= +// Registry +// ============================================================================= + +/// In-memory registry of `(provider, model) -> Pricing`. Lookups are by +/// exact match of both fields. +/// +/// The registry owns the (provider, model) key strings; entries are +/// pushed via `set` (which duplicates the inputs). +pub const Registry = struct { + allocator: std.mem.Allocator, + entries: std.ArrayList(Entry), + + pub const Entry = struct { + provider: []u8, + model: []u8, + pricing: Pricing, + }; + + pub fn init(allocator: std.mem.Allocator) Registry { + return .{ .allocator = allocator, .entries = .empty }; + } + + pub fn deinit(self: *Registry) void { + for (self.entries.items) |e| { + self.allocator.free(e.provider); + self.allocator.free(e.model); + } + self.entries.deinit(self.allocator); + } + + /// Look up pricing for (provider, model). Returns null if no entry + /// matches — distinct from "price is zero," which is a valid entry. + pub fn get(self: *const Registry, provider: []const u8, model: []const u8) ?Pricing { + for (self.entries.items) |e| { + if (std.mem.eql(u8, e.provider, provider) and std.mem.eql(u8, e.model, model)) { + return e.pricing; + } + } + return null; + } + + /// Insert or replace pricing for (provider, model). Duplicates the + /// key strings. + pub fn set( + self: *Registry, + provider: []const u8, + model: []const u8, + pricing: Pricing, + ) !void { + for (self.entries.items) |*e| { + if (std.mem.eql(u8, e.provider, provider) and std.mem.eql(u8, e.model, model)) { + e.pricing = pricing; + return; + } + } + const provider_copy = try self.allocator.dupe(u8, provider); + errdefer self.allocator.free(provider_copy); + const model_copy = try self.allocator.dupe(u8, model); + errdefer self.allocator.free(model_copy); + try self.entries.append(self.allocator, .{ + .provider = provider_copy, + .model = model_copy, + .pricing = pricing, + }); + } + + pub fn count(self: *const Registry) usize { + return self.entries.items.len; + } +}; + +// ============================================================================= +// Tests +// ============================================================================= + +const testing = std.testing; + +test "Pricing.fromDollarsPerMtok: $3.00/Mtok -> 300 micro-cents/token" { + try testing.expectEqual(@as(u64, 300), Pricing.fromDollarsPerMtok(3.0)); + try testing.expectEqual(@as(u64, 1500), Pricing.fromDollarsPerMtok(15.0)); + try testing.expectEqual(@as(u64, 30), Pricing.fromDollarsPerMtok(0.3)); + try testing.expectEqual(@as(u64, 0), Pricing.fromDollarsPerMtok(0)); + try testing.expectEqual(@as(u64, 0), Pricing.fromDollarsPerMtok(-1)); +} + +test "Pricing.fromDollarsPerMtok: rounds float noise to clean integer" { + // 0.1 * 3 = 0.30000000000000004 in IEEE 754. Verify rounding + // recovers the clean integer. + const v = 0.1 * 3.0; + try testing.expectEqual(@as(u64, 30), Pricing.fromDollarsPerMtok(v)); +} + +test "costMicroCents: standard mixed input/output" { + const pricing: Pricing = .{ + .input = 300, // $3/Mtok + .output = 1500, // $15/Mtok + }; + const usage: Usage = .{ .input = 1000, .output = 200 }; + // 1000*300 + 200*1500 = 300_000 + 300_000 = 600_000 micro-cents. + // = 6 cents = $0.06. + try testing.expectEqual(@as(?u64, 600_000), costMicroCents(usage, pricing)); +} + +test "costMicroCents: cache_read and cache_write discounts are honored" { + const pricing: Pricing = .{ + .input = 300, + .output = 1500, + .cache_read = 30, // 0.1x input + .cache_write = 375, // 1.25x input + }; + const usage: Usage = .{ + .input = 1000, + .output = 200, + .cache_read = 5000, + .cache_write = 500, + }; + // 1000*300 + 200*1500 + 5000*30 + 500*375 + // = 300_000 + 300_000 + 150_000 + 187_500 = 937_500. + try testing.expectEqual(@as(?u64, 937_500), costMicroCents(usage, pricing)); +} + +test "costMicroCents: reasoning tokens do not double-count" { + const pricing: Pricing = .{ .input = 0, .output = 1500 }; + const usage: Usage = .{ .output = 100, .reasoning = 60 }; + // Cost is from `output` alone; reasoning is a subset of output. + try testing.expectEqual(@as(?u64, 150_000), costMicroCents(usage, pricing)); +} + +test "costMicroCents: unknown price + nonzero usage poisons to null" { + // gpt-4o written with only input/output set; cache fields default + // to null (unknown). A turn that uses any cache reads should + // surface as unknown cost, not silently free. + const pricing: Pricing = .{ + .input = 250, + .output = 1000, + // cache_read, cache_write left null. + }; + const usage: Usage = .{ .input = 1000, .output = 200, .cache_read = 500 }; + try testing.expectEqual(@as(?u64, null), costMicroCents(usage, pricing)); +} + +test "costMicroCents: unknown price + zero usage stays known" { + // Same partially-specified pricing, but the turn never touched + // cache. Cost should remain known. + const pricing: Pricing = .{ .input = 250, .output = 1000 }; + const usage: Usage = .{ .input = 1000, .output = 200 }; + try testing.expectEqual(@as(?u64, 450_000), costMicroCents(usage, pricing)); +} + +test "Registry: set, get, replace" { + var reg = Registry.init(testing.allocator); + defer reg.deinit(); + + try testing.expect(reg.get("anthropic", "claude-sonnet-4") == null); + + try reg.set("anthropic", "claude-sonnet-4", .{ .input = 300, .output = 1500 }); + try testing.expectEqual(@as(usize, 1), reg.count()); + + const p = reg.get("anthropic", "claude-sonnet-4").?; + try testing.expectEqual(@as(?u64, 300), p.input); + try testing.expectEqual(@as(?u64, 1500), p.output); + + // Replace existing. + try reg.set("anthropic", "claude-sonnet-4", .{ .input = 400, .output = 1600 }); + try testing.expectEqual(@as(usize, 1), reg.count()); + const p2 = reg.get("anthropic", "claude-sonnet-4").?; + try testing.expectEqual(@as(?u64, 400), p2.input); +} + +test "Registry: distinct (provider, model) pairs" { + var reg = Registry.init(testing.allocator); + defer reg.deinit(); + try reg.set("openai", "gpt-4o", .{ .input = 250 }); + try reg.set("openai", "gpt-4o-mini", .{ .input = 15 }); + try reg.set("anthropic", "claude-sonnet-4", .{ .input = 300 }); + try testing.expectEqual(@as(usize, 3), reg.count()); + try testing.expectEqual(@as(?u64, 250), reg.get("openai", "gpt-4o").?.input); + try testing.expectEqual(@as(?u64, 15), reg.get("openai", "gpt-4o-mini").?.input); + try testing.expectEqual(@as(?u64, 300), reg.get("anthropic", "claude-sonnet-4").?.input); +} diff --git a/libpanto/src/provider.zig b/libpanto/src/provider.zig index bc39026..ab60678 100644 --- a/libpanto/src/provider.zig +++ b/libpanto/src/provider.zig @@ -3,7 +3,9 @@ const std = @import("std"); const config_mod = @import("config.zig"); const conversation = @import("conversation.zig"); const tool_registry_mod = @import("tool_registry.zig"); +const session_mod = @import("session.zig"); pub const ToolRegistry = tool_registry_mod.ToolRegistry; +pub const Usage = session_mod.Usage; pub const ContentBlockType = enum { Text, @@ -35,13 +37,21 @@ pub const ContentBlockType = enum { /// or in the Provider (HTTP/parse/stream failure). It is the last callback the /// receiver will see for that turn. `onError` itself cannot fail; receivers /// must swallow secondary failures during cleanup. +/// +/// `onMessageComplete`'s `usage` argument carries the wire-reported token +/// counts for the just-finished assistant turn. Providers fire this exactly +/// once per successful turn. `usage` is `null` only when the wire genuinely +/// did not deliver any usage information — chiefly OpenAI-compatible proxies +/// (OpenRouter, vLLM, some self-hosted backends) that ignore +/// `stream_options.include_usage`. Receivers that compute cost should record +/// the null case explicitly ("unknown") rather than treating it as zero. pub const ReceiverVTable = struct { onMessageStart: *const fn (*anyopaque, conversation.MessageRole) anyerror!void, onBlockStart: *const fn (*anyopaque, ContentBlockType, usize) anyerror!void, onToolDetails: *const fn (*anyopaque, usize, []const u8, []const u8) anyerror!void, onContentDelta: *const fn (*anyopaque, usize, []const u8) anyerror!void, onBlockComplete: *const fn (*anyopaque, usize, conversation.ContentBlock) anyerror!void, - onMessageComplete: *const fn (*anyopaque, conversation.Message) anyerror!void, + onMessageComplete: *const fn (*anyopaque, conversation.Message, ?Usage) anyerror!void, onError: *const fn (*anyopaque, anyerror) void, }; @@ -69,8 +79,8 @@ pub const Receiver = struct { try self.vtable.onBlockComplete(self.ptr, block_index, block); } - pub fn onMessageComplete(self: Receiver, message: conversation.Message) !void { - try self.vtable.onMessageComplete(self.ptr, message); + pub fn onMessageComplete(self: Receiver, message: conversation.Message, usage: ?Usage) !void { + try self.vtable.onMessageComplete(self.ptr, message, usage); } pub fn onError(self: Receiver, err: anyerror) void { diff --git a/libpanto/src/provider_anthropic_messages.zig b/libpanto/src/provider_anthropic_messages.zig index 96649b6..5d5520a 100644 --- a/libpanto/src/provider_anthropic_messages.zig +++ b/libpanto/src/provider_anthropic_messages.zig @@ -216,6 +216,15 @@ const StreamState = struct { /// Assembled blocks for the final message, in stream order. blocks: std.ArrayList(conversation.ContentBlock) = .empty, + /// Accumulated token counts. Anthropic reports the input-side counts + /// on `message_start.usage` and the final `output_tokens` (plus + /// possibly updated input-side counts) on `message_delta.usage`. + /// `usage_seen` distinguishes "genuinely all zero" from "never + /// reported" — the former stamps a real `Usage` on + /// `onMessageComplete`, the latter stamps `null`. + usage: provider_mod.Usage = .{}, + usage_seen: bool = false, + const ActiveBlock = struct { /// Index reported on the wire (Anthropic's content-array index). wire_index: usize, @@ -252,6 +261,22 @@ const StreamState = struct { try receiver.onMessageStart(.assistant); } + /// Merge a wire-level usage snapshot into the accumulated counts. + /// Missing fields mean "unchanged," not "reset to zero." Marks + /// `usage_seen` so `finalize` delivers a non-null `Usage` to + /// `onMessageComplete`. + fn mergeUsage(self: *StreamState, partial: json_mod.StreamUsage) void { + if (partial.input_tokens) |v| self.usage.input = v; + if (partial.output_tokens) |v| self.usage.output = v; + if (partial.cache_creation_input_tokens) |v| self.usage.cache_write = v; + if (partial.cache_read_input_tokens) |v| self.usage.cache_read = v; + if (partial.input_tokens != null or partial.output_tokens != null or + partial.cache_creation_input_tokens != null or partial.cache_read_input_tokens != null) + { + self.usage_seen = true; + } + } + fn openBlock( self: *StreamState, receiver: *provider_mod.Receiver, @@ -402,7 +427,8 @@ const StreamState = struct { try conv.addAssistantMessage(moved_blocks); const msg = conv.messages.items[conv.messages.items.len - 1]; - try receiver.onMessageComplete(msg); + const usage: ?provider_mod.Usage = if (self.usage_seen) self.usage else null; + try receiver.onMessageComplete(msg, usage); } }; @@ -416,8 +442,9 @@ fn handleEvent( defer parsed.deinit(); switch (parsed.event) { - .message_start => { + .message_start => |s| { try state.ensureStarted(receiver); + state.mergeUsage(s.usage); }, .content_block_start => |s| { try state.ensureStarted(receiver); @@ -438,9 +465,10 @@ fn handleEvent( .content_block_stop => { try state.closeBlock(receiver); }, - .message_delta => { + .message_delta => |d| { // We don't act on stop_reason directly; message_stop is the // authoritative end-of-stream signal. + state.mergeUsage(d.usage); }, .message_stop => { state.end_of_stream = true; @@ -486,7 +514,7 @@ const RecordingReceiver = struct { text: []const u8, // owned copy signature: ?[]const u8 = null, // owned copy when present }, - message_complete, + message_complete: ?provider_mod.Usage, err: anyerror, }; @@ -578,9 +606,9 @@ const RecordingReceiver = struct { else => {}, } } - fn onMessageComplete(ptr: *anyopaque, _: conversation.Message) anyerror!void { + fn onMessageComplete(ptr: *anyopaque, _: conversation.Message, usage: ?provider_mod.Usage) anyerror!void { const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - try self.events.append(self.allocator, .message_complete); + try self.events.append(self.allocator, .{ .message_complete = usage }); } fn onError(ptr: *anyopaque, err: anyerror) void { const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); @@ -650,7 +678,91 @@ test "streams a text-only turn end-to-end" { try testing.expectEqualStrings("Hello", rec.events.items[2].delta.bytes); try testing.expectEqualStrings("!", rec.events.items[3].delta.bytes); try testing.expectEqualStrings("Hello!", rec.events.items[4].block_complete.text); - try testing.expectEqual(@as(@TypeOf(rec.events.items[5]), .message_complete), rec.events.items[5]); + try testing.expect(rec.events.items[5] == .message_complete); + // No usage on the wire — the assertion is structural. + try testing.expect(rec.events.items[5].message_complete == null); +} + +test "anthropic: captures usage from message_start and message_delta on message_complete" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("hi"); + + var rec = RecordingReceiver.init(allocator); + defer rec.deinit(); + var recv = rec.receiver(); + + const events = [_][]const u8{ + // Initial input-side counts on message_start. + \\{"type":"message_start","message":{"id":"m","type":"message","role":"assistant","content":[],"model":"claude","usage":{"input_tokens":100,"cache_creation_input_tokens":50,"cache_read_input_tokens":200}}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"ok"}} + , + \\{"type":"content_block_stop","index":0} + , + // Final output count on message_delta. + \\{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":42}} + , + \\{"type":"message_stop"} + , + }; + + try runStreamedTurn(allocator, &conv, &recv, &events); + + // Find the message_complete event and check its usage payload. + var found: ?provider_mod.Usage = null; + for (rec.events.items) |ev| { + if (ev == .message_complete) found = ev.message_complete; + } + try testing.expect(found != null); + const u = found.?; + try testing.expectEqual(@as(u64, 100), u.input); + try testing.expectEqual(@as(u64, 42), u.output); + try testing.expectEqual(@as(u64, 200), u.cache_read); + try testing.expectEqual(@as(u64, 50), u.cache_write); + try testing.expectEqual(@as(u64, 0), u.reasoning); // Anthropic doesn't split reasoning separately. +} + +test "anthropic: message_complete carries null usage when wire omits it" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("hi"); + + var rec = RecordingReceiver.init(allocator); + defer rec.deinit(); + var recv = rec.receiver(); + + const events = [_][]const u8{ + \\{"type":"message_start","message":{"id":"m","role":"assistant","content":[],"model":"claude"}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"hi"}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"message_delta","delta":{"stop_reason":"end_turn"}} + , + \\{"type":"message_stop"} + , + }; + + try runStreamedTurn(allocator, &conv, &recv, &events); + + var saw_complete = false; + for (rec.events.items) |ev| { + if (ev == .message_complete) { + saw_complete = true; + try testing.expect(ev.message_complete == null); + } + } + try testing.expect(saw_complete); } test "captures thinking signature for round-trip" { diff --git a/libpanto/src/provider_openai_chat.zig b/libpanto/src/provider_openai_chat.zig index ca8fa4c..fcbc0bb 100644 --- a/libpanto/src/provider_openai_chat.zig +++ b/libpanto/src/provider_openai_chat.zig @@ -207,14 +207,17 @@ pub const OpenAIChatProvider = struct { return; } try handleEvent(self.allocator, ev_payload, &state, receiver); - if (state.end_of_stream) { - try state.finalize(receiver, conv); - return; - } + // Note: we do NOT bail when state.end_of_stream is set. + // OpenAI emits the terminating `usage` chunk *after* the + // chunk carrying finish_reason, then sends `[DONE]`. If + // we returned on finish_reason we'd never capture usage. + // `[DONE]` is the authoritative end-of-stream marker. } } - // Stream ended without [DONE] or finish_reason. Finalize anyway. + // Stream ended without [DONE]. Some servers and proxies omit it + // (or drop the trailing usage chunk). Finalize with whatever we've + // got — usage will be null in that case, which is fine. try state.finalize(receiver, conv); } }; @@ -266,6 +269,11 @@ const StreamState = struct { /// whose block we've already emitted. closed_tool_indices: std.AutoHashMap(usize, void), + /// Token counts from the terminating chunk's `usage` block. Only + /// populated when the server sent `usage` (i.e. the request used + /// `stream_options.include_usage: true` AND the server honored it). + usage: ?provider_mod.Usage = null, + const ToolUseInProgress = struct { /// Block index emitted to the receiver for this tool call's /// onBlockStart / onContentDelta / onBlockComplete callbacks. @@ -542,7 +550,7 @@ const StreamState = struct { try conv.addAssistantMessage(moved_blocks); const msg = conv.messages.items[conv.messages.items.len - 1]; - try receiver.onMessageComplete(msg); + try receiver.onMessageComplete(msg, self.usage); } }; @@ -556,6 +564,24 @@ fn handleEvent( defer parsed.deinit(); const d = parsed.delta; + // Usage block arrives in the terminating chunk (after finish_reason, + // with an empty `choices` array). Capture it; `finalize` delivers it + // as part of `onMessageComplete`. OpenAI bills `prompt_tokens` as + // the *total* input including cached tokens; we split them so + // callers don't have to. + if (d.usage) |u| { + const prompt: u64 = u.prompt_tokens orelse 0; + const cached: u64 = u.cached_prompt_tokens orelse 0; + const fresh: u64 = if (cached > prompt) 0 else prompt - cached; + state.usage = .{ + .input = fresh, + .output = u.completion_tokens orelse 0, + .cache_read = cached, + .cache_write = 0, // OpenAI doesn't bill a cache-write premium. + .reasoning = u.reasoning_tokens orelse 0, + }; + } + // Mid-stream provider error: some OpenAI-compatible endpoints (and // OpenAI itself on rare transient failures) return HTTP 200 with an // error embedded in the SSE stream. Treat the turn as failed. @@ -631,7 +657,7 @@ const NoopReceiver = struct { fn noopToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {} fn noopDelta(_: *anyopaque, _: usize, _: []const u8) anyerror!void {} fn noopBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {} - fn noopMsgComplete(_: *anyopaque, _: conversation.Message) anyerror!void {} + fn noopMsgComplete(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {} fn noopErr(_: *anyopaque, _: anyerror) void {} }; @@ -648,8 +674,10 @@ fn runStreamedTurn( for (events) |payload| { if (std.mem.eql(u8, payload, "[DONE]")) break; + // Process every chunk through to [DONE], including the + // post-finish_reason usage chunk. Mirrors the production loop + // in OpenAIChatProvider.streamStep. try handleEvent(allocator, payload, &state, receiver); - if (state.end_of_stream) break; } try state.finalize(receiver, conv); } @@ -712,6 +740,76 @@ test "two streamed turns persist assistant replies in the conversation" { ); } +test "openai_chat: terminating usage chunk lands on message_complete with split cache_read" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("hi"); + + var rec = RecordingReceiver{ .allocator = allocator }; + defer { + for (rec.events.items) |s| allocator.free(s); + rec.events.deinit(allocator); + } + var recv = rec.receiver(); + + const events = [_][]const u8{ + \\{"choices":[{"delta":{"role":"assistant"}}]} + , + \\{"choices":[{"delta":{"content":"hi"}}]} + , + \\{"choices":[{"delta":{},"finish_reason":"stop"}]} + , + // OpenAI's terminating chunk: empty choices, top-level usage. + \\{"choices":[],"usage":{"prompt_tokens":150,"completion_tokens":42,"prompt_tokens_details":{"cached_tokens":120},"completion_tokens_details":{"reasoning_tokens":18}}} + , + "[DONE]", + }; + try runStreamedTurn(allocator, &conv, &recv, &events); + + var found: ?[]const u8 = null; + for (rec.events.items) |s| { + if (std.mem.startsWith(u8, s, "msg_complete[")) found = s; + } + try testing.expect(found != null); + // 150 prompt - 120 cached = 30 fresh input. + try testing.expectEqualStrings("msg_complete[usage:in=30,out=42,cr=120,cw=0,rsn=18]", found.?); +} + +test "openai_chat: omitted stream usage yields null on message_complete" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("hi"); + + var rec = RecordingReceiver{ .allocator = allocator }; + defer { + for (rec.events.items) |s| allocator.free(s); + rec.events.deinit(allocator); + } + var recv = rec.receiver(); + + const events = [_][]const u8{ + \\{"choices":[{"delta":{"role":"assistant"}}]} + , + \\{"choices":[{"delta":{"content":"hi"}}]} + , + \\{"choices":[{"delta":{},"finish_reason":"stop"}]} + , + "[DONE]", + }; + try runStreamedTurn(allocator, &conv, &recv, &events); + + var found: ?[]const u8 = null; + for (rec.events.items) |s| { + if (std.mem.startsWith(u8, s, "msg_complete")) found = s; + } + try testing.expect(found != null); + try testing.expectEqualStrings("msg_complete[usage:null]", found.?); +} + test "fragmented tool_call id and name are reassembled" { // Lenient OpenAI-compatible providers occasionally split `id` and // `function.name` across multiple deltas instead of sending them whole @@ -817,9 +915,16 @@ const RecordingReceiver = struct { const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); try self.recordFmt("block_complete[{d}]", .{idx}); } - fn onMessageComplete(ptr: *anyopaque, _: conversation.Message) anyerror!void { + fn onMessageComplete(ptr: *anyopaque, _: conversation.Message, usage: ?provider_mod.Usage) anyerror!void { const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - try self.record("msg_complete"); + if (usage) |u| { + try self.recordFmt( + "msg_complete[usage:in={d},out={d},cr={d},cw={d},rsn={d}]", + .{ u.input, u.output, u.cache_read, u.cache_write, u.reasoning }, + ); + } else { + try self.record("msg_complete[usage:null]"); + } } fn onError(_: *anyopaque, _: anyerror) void {} }; @@ -878,7 +983,9 @@ test "parallel tool_calls emit one complete start/delta/complete cycle per block "tool_details[3]:c3:ping", "delta[3]:{\"host\":\"d\"}", "block_complete[3]", - "msg_complete", + // No usage chunk in this fixture (older test data) — record + // shows null. + "msg_complete[usage:null]", }; // Identity arrives in the assembled ContentBlock at completion time. diff --git a/libpanto/src/root.zig b/libpanto/src/root.zig index ff02d4d..de62cfa 100644 --- a/libpanto/src/root.zig +++ b/libpanto/src/root.zig @@ -8,6 +8,9 @@ pub const sse = @import("sse.zig"); 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 pricing = @import("pricing.zig"); // Re-exports for ergonomic embedder use. pub const Tool = tool.Tool; diff --git a/libpanto/src/session.zig b/libpanto/src/session.zig new file mode 100644 index 0000000..e77d121 --- /dev/null +++ b/libpanto/src/session.zig @@ -0,0 +1,961 @@ +//! On-disk session entry types and JSON serialization. +//! +//! These types are the wire format of pantograph's session log. They are +//! intentionally separate from the in-memory `Conversation`/`Message`/ +//! `ContentBlock` model: +//! +//! - The in-memory model holds only what providers need to serialize a +//! request (`role`, `content`). +//! - The on-disk model holds the full event-log story: provider/model +//! used per request, assistant stop reason and usage, timestamps, and +//! enough tree structure (`id`/`parent_id`) to allow future branching. +//! +//! Bridge functions at the bottom convert between the two. The bridge is +//! lossy by design: assistant metadata (provider/model/stop_reason/usage) +//! is recorded in entries but does NOT round-trip into the in-memory +//! conversation, because providers don't need it for request serialization. +//! +//! Format version: 1 (see `CURRENT_VERSION`). Migrations live in +//! `session_manager.zig`. + +const std = @import("std"); +const Allocator = std.mem.Allocator; +const Writer = std.Io.Writer; + +const conversation = @import("conversation.zig"); + +/// Bumped whenever the on-disk format changes in a way that older readers +/// cannot tolerate. Older files are upgraded by `migrate()` on load and +/// the file is rewritten once. +pub const CURRENT_VERSION: u32 = 1; + +// ============================================================================= +// Header +// ============================================================================= + +/// First (and only) line of a session file. Metadata only — not part of +/// the entry tree (no id/parent_id). +pub const SessionHeader = struct { + version: u32, + id: []const u8, // UUIDv7 string, owned + timestamp: []const u8, // ISO 8601, owned + cwd: []const u8, // owned + + pub fn deinit(self: SessionHeader, alloc: Allocator) void { + alloc.free(self.id); + alloc.free(self.timestamp); + alloc.free(self.cwd); + } +}; + +// ============================================================================= +// Entries +// ============================================================================= + +/// Fields shared by every non-header entry. +pub const EntryBase = struct { + id: []const u8, // 8-char hex, owned + parent_id: ?[]const u8, // owned, null for first entry + timestamp: []const u8, // ISO 8601, owned + + pub fn deinit(self: EntryBase, alloc: Allocator) void { + alloc.free(self.id); + if (self.parent_id) |p| alloc.free(p); + alloc.free(self.timestamp); + } +}; + +pub const SessionEntry = union(enum) { + message: MessageEntry, + + pub fn base(self: SessionEntry) EntryBase { + return switch (self) { + .message => |m| m.base, + }; + } + + pub fn deinit(self: SessionEntry, alloc: Allocator) void { + switch (self) { + .message => |m| m.deinit(alloc), + } + } +}; + +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, + + 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); + self.message.deinit(alloc); + } +}; + +pub const DiskMessageRole = enum { system, user, assistant }; + +pub const DiskMessage = struct { + role: DiskMessageRole, + content: []DiskContentBlock, // owned + // Assistant-only metadata. Null for system/user messages. + provider: ?[]const u8 = null, // owned + model: ?[]const u8 = null, // owned + stop_reason: ?[]const u8 = null, // owned + usage: ?Usage = null, + + pub fn deinit(self: DiskMessage, 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); + } +}; + +/// Token usage reported by a provider for a single assistant turn. +/// +/// All four input categories sum to the total prompt tokens that were +/// billed for the turn: +/// `input + cache_read + cache_write` = total prompt tokens. +/// +/// `reasoning` is a **subset** of `output`, not additive — it's the +/// portion of the output that the model spent on internal reasoning +/// (OpenAI o-series, Anthropic extended thinking). Cost is computed +/// from `output` only; `reasoning` is tracked for display. +/// +/// All fields are `u64` and default to 0. Providers that don't report a +/// given category leave it at 0. Future cache-tier breakdowns (e.g. +/// Anthropic's 5m vs 1h tiers) can be added as new fields without +/// breaking the format. +pub const Usage = struct { + /// Fresh input tokens billed at the model's base input rate. + input: u64 = 0, + /// Output tokens billed at the model's base output rate. + output: u64 = 0, + /// Input tokens served from cache (typ. 0.1× base input rate on + /// Anthropic, 0.5× on OpenAI). + cache_read: u64 = 0, + /// Input tokens written to a new cache entry (Anthropic only at + /// time of writing; typ. 1.25× base input rate). OpenAI's cache + /// hits don't bill a write premium, so this stays 0 there. + cache_write: u64 = 0, + /// Subset of `output` spent on reasoning. For OpenAI o-series + /// models and Anthropic extended thinking. Display-only — cost is + /// already accounted for via `output`. + reasoning: u64 = 0, +}; + +// ============================================================================= +// Content blocks +// ============================================================================= + +pub const DiskContentBlock = union(enum) { + text: DiskTextBlock, + thinking: DiskThinkingBlock, + tool_use: DiskToolUseBlock, + tool_result: DiskToolResultBlock, + + pub fn deinit(self: DiskContentBlock, alloc: Allocator) void { + switch (self) { + .text => |b| b.deinit(alloc), + .thinking => |b| b.deinit(alloc), + .tool_use => |b| b.deinit(alloc), + .tool_result => |b| b.deinit(alloc), + } + } +}; + +pub const DiskTextBlock = struct { + text: []const u8, // owned + pub fn deinit(self: DiskTextBlock, alloc: Allocator) void { + alloc.free(self.text); + } +}; + +pub const DiskThinkingBlock = 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 { + alloc.free(self.thinking); + if (self.signature) |s| alloc.free(s); + } +}; + +pub const DiskToolUseBlock = struct { + id: []const u8, // owned + name: []const u8, // owned + input: []const u8, // raw JSON bytes, owned + pub fn deinit(self: DiskToolUseBlock, alloc: Allocator) void { + alloc.free(self.id); + alloc.free(self.name); + alloc.free(self.input); + } +}; + +pub const DiskToolResultBlock = struct { + tool_use_id: []const u8, // owned + content: []const u8, // owned + pub fn deinit(self: DiskToolResultBlock, alloc: Allocator) void { + alloc.free(self.tool_use_id); + alloc.free(self.content); + } +}; + +// ============================================================================= +// File entry (header or entry) +// ============================================================================= + +pub const FileEntry = union(enum) { + header: SessionHeader, + entry: SessionEntry, + + pub fn deinit(self: FileEntry, alloc: Allocator) void { + switch (self) { + .header => |h| h.deinit(alloc), + .entry => |e| e.deinit(alloc), + } + } +}; + +// ============================================================================= +// Serialization +// ============================================================================= + +/// Serialize the header as a single JSON line. Caller owns returned bytes. +/// The returned slice does NOT include a trailing newline. +pub fn serializeHeader(allocator: Allocator, header: SessionHeader) ![]u8 { + var aw: Writer.Allocating = .init(allocator); + errdefer aw.deinit(); + + var s: std.json.Stringify = .{ .writer = &aw.writer }; + try s.beginObject(); + try s.objectField("type"); + try s.write("session"); + try s.objectField("version"); + try s.write(header.version); + try s.objectField("id"); + try s.write(header.id); + try s.objectField("timestamp"); + try s.write(header.timestamp); + try s.objectField("cwd"); + try s.write(header.cwd); + try s.endObject(); + + return try aw.toOwnedSlice(); +} + +/// Serialize an entry as a single JSON line. Caller owns returned bytes. +pub fn serializeEntry(allocator: Allocator, entry: SessionEntry) ![]u8 { + var aw: Writer.Allocating = .init(allocator); + errdefer aw.deinit(); + var s: std.json.Stringify = .{ .writer = &aw.writer }; + try writeEntry(&s, entry); + return try aw.toOwnedSlice(); +} + +fn writeEntry(s: *std.json.Stringify, entry: SessionEntry) !void { + switch (entry) { + .message => |m| try writeMessageEntry(s, m), + } +} + +fn writeMessageEntry(s: *std.json.Stringify, m: MessageEntry) !void { + try s.beginObject(); + try s.objectField("type"); + try s.write("message"); + try s.objectField("id"); + try s.write(m.base.id); + try s.objectField("parentId"); + 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); + } + try s.objectField("message"); + try writeDiskMessage(s, m.message); + try s.endObject(); +} + +fn writeDiskMessage(s: *std.json.Stringify, msg: DiskMessage) !void { + try s.beginObject(); + try s.objectField("role"); + try s.write(@tagName(msg.role)); + try s.objectField("content"); + try s.beginArray(); + for (msg.content) |block| { + 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.usage) |u| { + try s.objectField("usage"); + try s.beginObject(); + try s.objectField("input"); + try s.write(u.input); + try s.objectField("output"); + try s.write(u.output); + // Omit zero-valued auxiliary fields to keep older / unused + // sessions compact. Readers default missing fields to 0, so + // round-trip behavior is preserved. + if (u.cache_read != 0) { + try s.objectField("cacheRead"); + try s.write(u.cache_read); + } + if (u.cache_write != 0) { + try s.objectField("cacheWrite"); + try s.write(u.cache_write); + } + if (u.reasoning != 0) { + try s.objectField("reasoning"); + try s.write(u.reasoning); + } + try s.endObject(); + } + try s.endObject(); +} + +fn writeDiskBlock(s: *std.json.Stringify, block: DiskContentBlock) !void { + switch (block) { + .text => |b| { + try s.beginObject(); + try s.objectField("type"); + try s.write("text"); + try s.objectField("text"); + try s.write(b.text); + try s.endObject(); + }, + .thinking => |b| { + try s.beginObject(); + try s.objectField("type"); + try s.write("thinking"); + try s.objectField("thinking"); + try s.write(b.thinking); + if (b.signature) |sig| { + try s.objectField("signature"); + try s.write(sig); + } + try s.endObject(); + }, + .tool_use => |b| { + try s.beginObject(); + try s.objectField("type"); + try s.write("toolUse"); + try s.objectField("id"); + try s.write(b.id); + try s.objectField("name"); + try s.write(b.name); + try s.objectField("input"); + try s.write(b.input); + try s.endObject(); + }, + .tool_result => |b| { + try s.beginObject(); + try s.objectField("type"); + try s.write("toolResult"); + try s.objectField("toolUseId"); + try s.write(b.tool_use_id); + try s.objectField("content"); + try s.write(b.content); + try s.endObject(); + }, + } +} + +// ============================================================================= +// Parsing +// ============================================================================= + +pub const ParseError = error{ + InvalidJson, + MissingField, + UnknownType, + UnknownRole, + UnknownBlockType, +} || Allocator.Error; + +/// Parse one JSON line into a `FileEntry`. Caller owns all bytes. +pub fn parseLine(allocator: Allocator, line: []const u8) ParseError!FileEntry { + var parsed = std.json.parseFromSlice(std.json.Value, allocator, line, .{}) catch { + return error.InvalidJson; + }; + defer parsed.deinit(); + return parseValue(allocator, parsed.value); +} + +fn parseValue(allocator: Allocator, v: std.json.Value) ParseError!FileEntry { + if (v != .object) return error.InvalidJson; + const type_v = v.object.get("type") orelse return error.MissingField; + if (type_v != .string) return error.MissingField; + const t = type_v.string; + if (std.mem.eql(u8, t, "session")) { + return .{ .header = try parseHeaderFromObject(allocator, v.object) }; + } else if (std.mem.eql(u8, t, "message")) { + return .{ .entry = .{ .message = try parseMessageEntry(allocator, v.object) } }; + } else { + return error.UnknownType; + } +} + +fn parseHeaderFromObject(allocator: Allocator, obj: std.json.ObjectMap) ParseError!SessionHeader { + const version: u32 = blk: { + if (obj.get("version")) |vv| { + if (vv == .integer) break :blk @intCast(vv.integer); + } + break :blk 1; + }; + const id = try dupeStringField(allocator, obj, "id"); + errdefer allocator.free(id); + const timestamp = try dupeStringField(allocator, obj, "timestamp"); + errdefer allocator.free(timestamp); + const cwd = try dupeStringField(allocator, obj, "cwd"); + errdefer allocator.free(cwd); + return .{ + .version = version, + .id = id, + .timestamp = timestamp, + .cwd = cwd, + }; +} + +fn parseMessageEntry(allocator: Allocator, obj: std.json.ObjectMap) ParseError!MessageEntry { + const id = try dupeStringField(allocator, obj, "id"); + errdefer allocator.free(id); + const timestamp = try dupeStringField(allocator, obj, "timestamp"); + errdefer allocator.free(timestamp); + const parent_id: ?[]const u8 = blk: { + const pv = obj.get("parentId") orelse break :blk null; + if (pv == .null) break :blk null; + if (pv != .string) return error.MissingField; + break :blk try allocator.dupe(u8, pv.string); + }; + 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 msg_v = obj.get("message") orelse return error.MissingField; + if (msg_v != .object) return error.MissingField; + const msg = try parseDiskMessage(allocator, msg_v.object); + + return .{ + .base = .{ .id = id, .parent_id = parent_id, .timestamp = timestamp }, + .provider = provider, + .model = model, + .message = msg, + }; +} + +fn parseDiskMessage(allocator: Allocator, obj: std.json.ObjectMap) ParseError!DiskMessage { + 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 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); + errdefer { + for (content_list.items) |b| b.deinit(allocator); + content_list.deinit(allocator); + } + for (content_v.array.items) |item| { + if (item != .object) return error.UnknownBlockType; + const block = try parseDiskBlock(allocator, item.object); + try content_list.append(allocator, block); + } + const content = try content_list.toOwnedSlice(allocator); + errdefer { + for (content) |b| b.deinit(allocator); + 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); + + var usage: ?Usage = null; + if (obj.get("usage")) |uv| { + if (uv == .object) { + usage = .{ + .input = readU64(uv.object, "input"), + .output = readU64(uv.object, "output"), + .cache_read = readU64(uv.object, "cacheRead"), + .cache_write = readU64(uv.object, "cacheWrite"), + .reasoning = readU64(uv.object, "reasoning"), + }; + } + } + + return .{ + .role = role, + .content = content, + .provider = provider, + .model = model, + .stop_reason = stop_reason, + .usage = usage, + }; +} + +fn parseDiskBlock(allocator: Allocator, obj: std.json.ObjectMap) ParseError!DiskContentBlock { + const type_v = obj.get("type") orelse return error.MissingField; + if (type_v != .string) return error.MissingField; + const t = type_v.string; + if (std.mem.eql(u8, t, "text")) { + const text = try dupeStringField(allocator, obj, "text"); + return .{ .text = .{ .text = text } }; + } else if (std.mem.eql(u8, t, "thinking")) { + const text = try dupeStringField(allocator, obj, "thinking"); + errdefer allocator.free(text); + const sig = try dupeOptionalStringField(allocator, obj, "signature"); + return .{ .thinking = .{ .thinking = text, .signature = sig } }; + } else if (std.mem.eql(u8, t, "toolUse")) { + const id = try dupeStringField(allocator, obj, "id"); + errdefer allocator.free(id); + const name = try dupeStringField(allocator, obj, "name"); + errdefer allocator.free(name); + const input = try dupeStringField(allocator, obj, "input"); + return .{ .tool_use = .{ .id = id, .name = name, .input = input } }; + } else if (std.mem.eql(u8, t, "toolResult")) { + const tuid = try dupeStringField(allocator, obj, "toolUseId"); + errdefer allocator.free(tuid); + const content = try dupeStringField(allocator, obj, "content"); + return .{ .tool_result = .{ .tool_use_id = tuid, .content = content } }; + } else { + return error.UnknownBlockType; + } +} + +fn readU64(obj: std.json.ObjectMap, name: []const u8) u64 { + const v = obj.get(name) orelse return 0; + if (v != .integer) return 0; + if (v.integer < 0) return 0; + return @intCast(v.integer); +} + +fn dupeStringField(allocator: Allocator, obj: std.json.ObjectMap, name: []const u8) ParseError![]const u8 { + const v = obj.get(name) orelse return error.MissingField; + if (v != .string) return error.MissingField; + return try allocator.dupe(u8, v.string); +} + +fn dupeOptionalStringField(allocator: Allocator, obj: std.json.ObjectMap, name: []const u8) ParseError!?[]const u8 { + const v = obj.get(name) orelse return null; + if (v == .null) return null; + if (v != .string) return error.MissingField; + return try allocator.dupe(u8, v.string); +} + +// ============================================================================= +// Bridge between in-memory and on-disk content blocks +// ============================================================================= + +/// Convert an in-memory `ContentBlock` to a `DiskContentBlock`. 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 { + switch (block) { + .Text => |tb| { + const text = try allocator.dupe(u8, tb.items); + return .{ .text = .{ .text = text } }; + }, + .Thinking => |tb| { + const text = try allocator.dupe(u8, tb.text.items); + errdefer allocator.free(text); + const sig: ?[]const u8 = if (tb.signature) |s| try allocator.dupe(u8, s) else null; + return .{ .thinking = .{ .thinking = text, .signature = sig } }; + }, + .ToolUse => |tu| { + const id = try allocator.dupe(u8, tu.id); + errdefer allocator.free(id); + const name = try allocator.dupe(u8, tu.name); + errdefer allocator.free(name); + const input = try allocator.dupe(u8, tu.input.items); + return .{ .tool_use = .{ .id = id, .name = name, .input = input } }; + }, + .ToolResult => |tr| { + const tuid = try allocator.dupe(u8, tr.tool_use_id); + errdefer allocator.free(tuid); + const content = try allocator.dupe(u8, tr.content.items); + return .{ .tool_result = .{ .tool_use_id = tuid, .content = content } }; + }, + } +} + +/// Convert a `DiskContentBlock` 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, +) !conversation.ContentBlock { + switch (block) { + .text => |b| { + const tb = try conversation.textualBlockFromSlice(allocator, b.text); + return .{ .Text = tb }; + }, + .thinking => |b| { + const tb = try conversation.textualBlockFromSlice(allocator, b.thinking); + errdefer { + var mut = tb; + mut.deinit(allocator); + } + const sig: ?[]const u8 = if (b.signature) |s| try allocator.dupe(u8, s) else null; + return .{ .Thinking = .{ .text = tb, .signature = sig } }; + }, + .tool_use => |b| { + const id = try allocator.dupe(u8, b.id); + errdefer allocator.free(id); + const name = try allocator.dupe(u8, b.name); + errdefer allocator.free(name); + const input = try conversation.textualBlockFromSlice(allocator, b.input); + return .{ .ToolUse = .{ .id = id, .name = name, .input = input } }; + }, + .tool_result => |b| { + const tuid = try allocator.dupe(u8, b.tool_use_id); + errdefer allocator.free(tuid); + const content = try conversation.textualBlockFromSlice(allocator, b.content); + return .{ .ToolResult = .{ .tool_use_id = tuid, .content = content } }; + }, + } +} + +// ============================================================================= +// Tests +// ============================================================================= + +const testing = std.testing; + +fn dupe(allocator: Allocator, s: []const u8) ![]const u8 { + return try allocator.dupe(u8, s); +} + +test "serialize/parse header round-trip" { + const a = testing.allocator; + const header: SessionHeader = .{ + .version = 1, + .id = try dupe(a, "019dc5ba-53f6-71a5-ab8f-b1f8709c2572"), + .timestamp = try dupe(a, "2026-04-25T17:40:15.990Z"), + .cwd = try dupe(a, "/Users/travis/Code/pantograph"), + }; + defer header.deinit(a); + + const line = try serializeHeader(a, header); + defer a.free(line); + + var fe = try parseLine(a, line); + defer fe.deinit(a); + try testing.expect(fe == .header); + try testing.expectEqual(@as(u32, 1), fe.header.version); + try testing.expectEqualStrings(header.id, fe.header.id); + try testing.expectEqualStrings(header.cwd, fe.header.cwd); +} + +test "serialize/parse user message entry round-trip (with provider/model stamp)" { + const a = testing.allocator; + + var content = try a.alloc(DiskContentBlock, 1); + content[0] = .{ .text = .{ .text = try dupe(a, "hello world") } }; + + const entry: SessionEntry = .{ .message = .{ + .base = .{ + .id = try dupe(a, "a1b2c3d4"), + .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"), + .message = .{ + .role = .user, + .content = content, + }, + } }; + defer entry.deinit(a); + + const line = try serializeEntry(a, entry); + defer a.free(line); + + var fe = try parseLine(a, line); + defer fe.deinit(a); + try testing.expect(fe == .entry); + 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(@as(usize, 1), got.message.content.len); + try testing.expectEqualStrings("hello world", got.message.content[0].text.text); +} + +test "serialize/parse assistant message entry with metadata" { + const a = testing.allocator; + + var content = try a.alloc(DiskContentBlock, 3); + content[0] = .{ .thinking = .{ + .thinking = try dupe(a, "let me think"), + .signature = try dupe(a, "sig-xyz"), + } }; + content[1] = .{ .text = .{ .text = try dupe(a, "I'll check.") } }; + content[2] = .{ .tool_use = .{ + .id = try dupe(a, "tool_abc"), + .name = try dupe(a, "bash"), + .input = try dupe(a, "{\"command\":\"ls\"}"), + } }; + + const entry: SessionEntry = .{ .message = .{ + .base = .{ + .id = try dupe(a, "b2c3d4e5"), + .parent_id = try dupe(a, "a1b2c3d4"), + .timestamp = try dupe(a, "2026-04-25T17:40:17.000Z"), + }, + .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 }, + }, + } }; + defer entry.deinit(a); + + const line = try serializeEntry(a, entry); + defer a.free(line); + + 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(@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("toolUse", got.message.stop_reason.?); + 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); +} + +test "serialize/parse tool result message entry" { + const a = testing.allocator; + + var content = try a.alloc(DiskContentBlock, 1); + content[0] = .{ .tool_result = .{ + .tool_use_id = try dupe(a, "tool_abc"), + .content = try dupe(a, "file1.txt\nfile2.txt"), + } }; + + const entry: SessionEntry = .{ .message = .{ + .base = .{ + .id = try dupe(a, "c3d4e5f6"), + .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"), + .message = .{ + .role = .user, + .content = content, + }, + } }; + defer entry.deinit(a); + + const line = try serializeEntry(a, entry); + defer a.free(line); + + var fe = try parseLine(a, line); + defer fe.deinit(a); + const got = fe.entry.message; + try testing.expectEqual(DiskMessageRole.user, got.message.role); + try testing.expectEqualStrings("tool_abc", got.message.content[0].tool_result.tool_use_id); + try testing.expectEqualStrings("file1.txt\nfile2.txt", got.message.content[0].tool_result.content); + try testing.expectEqualStrings("anthropic", got.provider.?); +} + +test "parse: null parentId is handled" { + const a = testing.allocator; + const line = + \\{"type":"message","id":"abcdefab","parentId":null,"timestamp":"2026-04-25T17:40:00Z","message":{"role":"system","content":[{"type":"text","text":"hi"}]}} + ; + var fe = try parseLine(a, line); + defer fe.deinit(a); + try testing.expect(fe.entry.message.base.parent_id == null); +} + +test "parse: malformed JSON is reported" { + const a = testing.allocator; + try testing.expectError(error.InvalidJson, parseLine(a, "not json")); + try testing.expectError(error.InvalidJson, parseLine(a, "{\"type\":\"message\"")); +} + +test "parse: unknown entry type is reported" { + const a = testing.allocator; + const line = + \\{"type":"future_entry","id":"abcdefab","parentId":null,"timestamp":"2026-04-25T17:40:00Z"} + ; + try testing.expectError(error.UnknownType, parseLine(a, line)); +} + +test "contentBlockToDisk: Text round-trips via in-memory" { + const a = testing.allocator; + + var tb = try conversation.textualBlockFromSlice(a, "hello"); + defer tb.deinit(a); + const block: conversation.ContentBlock = .{ .Text = tb }; + + const disk = try contentBlockToDisk(a, block); + defer disk.deinit(a); + try testing.expectEqualStrings("hello", disk.text.text); +} + +test "diskContentBlockToInternal: ToolUse preserves id/name/input" { + const a = testing.allocator; + + const disk: DiskContentBlock = .{ .tool_use = .{ + .id = try a.dupe(u8, "tu_1"), + .name = try a.dupe(u8, "bash"), + .input = try a.dupe(u8, "{\"command\":\"ls\"}"), + } }; + defer disk.deinit(a); + + var inmem = try diskContentBlockToInternal(a, disk); + defer inmem.deinit(a); + try testing.expectEqualStrings("tu_1", inmem.ToolUse.id); + try testing.expectEqualStrings("bash", inmem.ToolUse.name); + try testing.expectEqualStrings("{\"command\":\"ls\"}", inmem.ToolUse.input.items); +} + +test "Usage: all five fields round-trip; zero-valued fields omitted from JSON" { + const a = testing.allocator; + + var content = try a.alloc(DiskContentBlock, 1); + content[0] = .{ .text = .{ .text = try dupe(a, "hi") } }; + + const entry: SessionEntry = .{ .message = .{ + .base = .{ + .id = try dupe(a, "deadbeef"), + .parent_id = null, + .timestamp = try dupe(a, "2026-04-25T17:40:17.000Z"), + }, + .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, + .output = 50, + .cache_read = 800, + .cache_write = 200, + .reasoning = 30, + }, + }, + } }; + defer entry.deinit(a); + + const line = try serializeEntry(a, entry); + defer a.free(line); + + // Every non-zero field should appear in the serialized JSON. + try testing.expect(std.mem.indexOf(u8, line, "\"input\":100") != null); + try testing.expect(std.mem.indexOf(u8, line, "\"output\":50") != null); + try testing.expect(std.mem.indexOf(u8, line, "\"cacheRead\":800") != null); + try testing.expect(std.mem.indexOf(u8, line, "\"cacheWrite\":200") != null); + try testing.expect(std.mem.indexOf(u8, line, "\"reasoning\":30") != null); + + var fe = try parseLine(a, line); + defer fe.deinit(a); + const u = fe.entry.message.message.usage.?; + try testing.expectEqual(@as(u64, 100), u.input); + try testing.expectEqual(@as(u64, 50), u.output); + try testing.expectEqual(@as(u64, 800), u.cache_read); + try testing.expectEqual(@as(u64, 200), u.cache_write); + try testing.expectEqual(@as(u64, 30), u.reasoning); +} + +test "Usage: zero-valued auxiliary fields are omitted but parse back as 0" { + const a = testing.allocator; + + var content = try a.alloc(DiskContentBlock, 1); + content[0] = .{ .text = .{ .text = try dupe(a, "hi") } }; + + const entry: SessionEntry = .{ .message = .{ + .base = .{ + .id = try dupe(a, "deadbeef"), + .parent_id = null, + .timestamp = try dupe(a, "2026-04-25T17:40:17.000Z"), + }, + .message = .{ + .role = .assistant, + .content = content, + .usage = .{ .input = 100, .output = 50 }, + }, + } }; + defer entry.deinit(a); + + const line = try serializeEntry(a, entry); + defer a.free(line); + + try testing.expect(std.mem.indexOf(u8, line, "cacheRead") == null); + try testing.expect(std.mem.indexOf(u8, line, "cacheWrite") == null); + try testing.expect(std.mem.indexOf(u8, line, "reasoning") == null); + + var fe = try parseLine(a, line); + defer fe.deinit(a); + const u = fe.entry.message.message.usage.?; + try testing.expectEqual(@as(u64, 0), u.cache_read); + try testing.expectEqual(@as(u64, 0), u.cache_write); + try testing.expectEqual(@as(u64, 0), u.reasoning); +} + +test "diskContentBlockToInternal: Thinking preserves signature" { + const a = testing.allocator; + + const disk: DiskContentBlock = .{ .thinking = .{ + .thinking = try a.dupe(u8, "reasoning..."), + .signature = try a.dupe(u8, "sig123"), + } }; + defer disk.deinit(a); + + var inmem = try diskContentBlockToInternal(a, disk); + defer inmem.deinit(a); + try testing.expectEqualStrings("reasoning...", inmem.Thinking.text.items); + try testing.expectEqualStrings("sig123", inmem.Thinking.signature.?); +} diff --git a/libpanto/src/session_manager.zig b/libpanto/src/session_manager.zig new file mode 100644 index 0000000..bf9585d --- /dev/null +++ b/libpanto/src/session_manager.zig @@ -0,0 +1,1509 @@ +//! 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"); + +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 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); + } + + // Run format migration. Currently a no-op for version 1, but the + // hook exists so future versions can rewrite the file. + var migrated_header = header; + const did_migrate = migrate(allocator, &migrated_header, &entries); + // 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 + `final_entry` + /// to the file as a single batch. Creates the directory and file. + /// Called only when `flushed = false` and we just got an assistant + /// message. + fn flushBuffered(self: *SessionManager, final_entry: SessionEntry) !void { + // Ensure the directory exists. + try mkdirP(self.io, self.session_dir); + + // Open (create exclusively isn't required; if a stale file with the + // same UUIDv7 existed we'd just overwrite — UUIDv7s are unique). + 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; + // Header + 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; + + // Buffered entries. + 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; + } + + // Final (the assistant message we just got). + const final_line = try session_mod.serializeEntry(self.allocator, final_entry); + defer self.allocator.free(final_line); + try file.writePositionalAll(self.io, final_line, offset); + offset += final_line.len; + try file.writePositionalAll(self.io, "\n", offset); + offset += 1; + + // Best-effort fsync — the directory itself doesn't need syncing + // for our purposes (we don't unlink/rename); the file body does. + file.sync(self.io) catch {}; + + self.flushed = true; + self.written_bytes = offset; + } + + /// Append a single line for `entry` to the open session file. Caller + /// must have already verified `flushed`. + fn persistEntry(self: *SessionManager, entry: SessionEntry) !void { + const file = try Io.Dir.cwd().openFile(self.io, self.session_file, .{ + .mode = .write_only, + }); + defer file.close(self.io); + + const line = try session_mod.serializeEntry(self.allocator, entry); + defer self.allocator.free(line); + + try file.writePositionalAll(self.io, line, self.written_bytes); + self.written_bytes += line.len; + try file.writePositionalAll(self.io, "\n", self.written_bytes); + self.written_bytes += 1; + file.sync(self.io) catch {}; + } + + // ============================================================================= + // 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; + } +}; + +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); + for (disk_msg.content) |db| { + const block = try session_mod.diskContentBlockToInternal(allocator, db); + content.appendAssumeCapacity(block); + } + const role: conversation_mod.MessageRole = switch (disk_msg.role) { + .system => .system, + .user => .user, + .assistant => .assistant, + }; + try conv.messages.append(allocator, .{ .role = role, .content = content }); +} + +// ============================================================================= +// 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; +} + +// ============================================================================= +// 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].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 a1 = try testing.allocator.alloc(DiskContentBlock, 2); + a1[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "checking...") } }; + a1[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 = a1 }, null, null); + + // Tool-result user message. + const tr = try testing.allocator.alloc(DiskContentBlock, 1); + tr[0] = .{ .tool_result = .{ + .tool_use_id = try testing.allocator.dupe(u8, "tool_abc"), + .content = try testing.allocator.dupe(u8, "a.txt\nb.txt"), + } }; + _ = 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.content); + + // 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")); +} diff --git a/src/main.zig b/src/main.zig index 6654c95..3bc24e3 100644 --- a/src/main.zig +++ b/src/main.zig @@ -8,6 +8,8 @@ const panto_home = @import("panto_home.zig"); const luarocks_runtime = @import("luarocks_runtime.zig"); const self_exe = @import("self_exe.zig"); const subcommand = @import("subcommand.zig"); +const session_paths = @import("session_paths.zig"); +const models_toml = @import("models_toml.zig"); // Shorthand alias for the Lua C API. The bridge module owns the actual // `@cImport`; we re-use it here so the smoke check uses identical types. @@ -31,6 +33,7 @@ test { _ = luarocks_runtime; _ = self_exe; _ = subcommand; + _ = models_toml; } const Receiver = panto.provider.Receiver; @@ -40,14 +43,37 @@ const MessageRole = panto.conversation.MessageRole; /// Receiver that prints streaming deltas to stdout. Thinking blocks are /// dimmed with ANSI escape codes; text blocks render plain. +/// +/// Also captures one `?Usage` per assistant response during a turn. In +/// a tool-using turn the agent loop drives multiple streamStep calls; +/// each one ends with `onMessageComplete(msg, usage)`. We append the +/// usage — including `null` when the wire didn't deliver any — to +/// `per_message_usage` in order so `persistTurn` can pair each captured +/// usage with its corresponding assistant message. const CLIReceiver = struct { stdout: *std.Io.Writer, file: *std.Io.File.Writer, + allocator: std.mem.Allocator, + + /// One slot per assistant message completed during the current + /// turn, in completion order. `null` means the provider did not + /// report usage on the wire for that message (typical of + /// OpenAI-compatible proxies that ignore `stream_options.include_usage`). + per_message_usage: std.ArrayList(?panto.session_manager.Usage) = .empty, pub fn receiver(self: *CLIReceiver) Receiver { return .{ .ptr = self, .vtable = &vtable }; } + /// Reset usage state at the start of each turn. + pub fn beginTurn(self: *CLIReceiver) void { + self.per_message_usage.clearRetainingCapacity(); + } + + pub fn deinit(self: *CLIReceiver) void { + self.per_message_usage.deinit(self.allocator); + } + const vtable: ReceiverVTable = .{ .onMessageStart = onMessageStart, .onBlockStart = onBlockStart, @@ -121,9 +147,19 @@ const CLIReceiver = struct { try self.file.flush(); } - fn onMessageComplete(ptr: *anyopaque, message: panto.conversation.Message) anyerror!void { - _ = message; + fn onMessageComplete( + ptr: *anyopaque, + message: panto.conversation.Message, + usage: ?panto.session_manager.Usage, + ) anyerror!void { const self: *CLIReceiver = @ptrCast(@alignCast(ptr)); + // Only assistant messages come through streaming. The receiver + // contract says onMessageComplete fires exactly once per + // streamStep, with role=.assistant; record the usage slot in + // turn order regardless of whether the wire actually had usage. + if (message.role == .assistant) { + try self.per_message_usage.append(self.allocator, usage); + } try self.stdout.writeAll("\n"); try self.file.flush(); } @@ -261,6 +297,10 @@ pub fn main(init: std.process.Init) !void { const config = try loadConfig(init.environ_map); + // Parse the agent-mode flags. Currently only `--resume []`. + const cli_flags = try parseAgentFlags(alloc, init.minimal.args); + defer cli_flags.deinit(alloc); + var stdout_buffer: [4096]u8 = undefined; var stdout_file = std.Io.File.stdout().writer(io, &stdout_buffer); const stdout = &stdout_file.interface; @@ -269,9 +309,62 @@ pub fn main(init: std.process.Init) !void { var stdin_file = std.Io.File.stdin().reader(io, &stdin_buffer); const stdin = &stdin_file.interface; + // Resolve where this project's sessions live. + var cwd_buf: [std.fs.max_path_bytes]u8 = undefined; + const cwd_n = try std.process.currentPath(io, &cwd_buf); + const cwd = cwd_buf[0..cwd_n]; + const session_dir = try session_paths.sessionDirForCwd(alloc, init.environ_map, cwd); + defer alloc.free(session_dir); + + // Load the user's models.toml — a missing file is fine (empty + // registry). Cost lookups against an empty registry return null, + // and the display layer will format that as "unknown." + const models_toml_path = try models_toml.configPath(alloc, init.environ_map); + defer alloc.free(models_toml_path); + var pricing_registry = try models_toml.loadFromPath(alloc, io, models_toml_path); + defer pricing_registry.deinit(); + std.log.debug("models.toml: {d} entries from {s}", .{ pricing_registry.count(), models_toml_path }); + + const banner_model_initial: []const u8 = switch (config) { + inline else => |c| c.model, + }; + const banner_provider_initial: []const u8 = @tagName(config); + + // Create or resume the session. Resume failures (missing/ambiguous id) + // are user errors — print a tidy message and exit 1 rather than + // printing a Zig stack trace. + var session_mgr = openSession( + alloc, + io, + session_dir, + cwd, + cli_flags, + stdout, + &stdout_file, + ) catch |err| switch (err) { + error.SessionNotFound, error.AmbiguousSessionId => std.process.exit(1), + else => return err, + }; + defer session_mgr.deinit(); + var conv = panto.conversation.Conversation.init(alloc); defer conv.deinit(); - try conv.addSystemMessage("You are a helpful assistant."); + + if (session_mgr.getEntries().len > 0) { + // Resumed an existing session — rebuild the conversation from the + // log. The system prompt is part of the log. + conv.deinit(); + conv = try session_mgr.rebuildConversation(); + try stdout.print( + "resumed session {s} ({d} entries)\n", + .{ session_mgr.getSessionId()[0..@min(8, session_mgr.getSessionId().len)], session_mgr.getEntries().len }, + ); + } else { + // Fresh session — install the default system prompt and record it. + const system_text = "You are a helpful assistant."; + try conv.addSystemMessage(system_text); + try appendSystemToSession(alloc, &session_mgr, system_text); + } const prov = try panto.provider.Provider.init(alloc, io, config); var agent = panto.agent.Agent.init(alloc, io, prov); @@ -325,17 +418,22 @@ pub fn main(init: std.process.Init) !void { try agent.registerToolSource(rt.toolSource()); } - const banner_model: []const u8 = switch (config) { - inline else => |c| c.model, - }; const banner_base: []const u8 = switch (config) { inline else => |c| c.base_url, }; - try stdout.print("panto — {s}: {s} @ {s}\n", .{ @tagName(config), banner_model, banner_base }); + try stdout.print( + "panto — {s}: {s} @ {s}\n", + .{ banner_provider_initial, banner_model_initial, banner_base }, + ); try stdout.print("> ", .{}); try stdout_file.flush(); - var cli_recv = CLIReceiver{ .stdout = stdout, .file = &stdout_file }; + var cli_recv = CLIReceiver{ + .stdout = stdout, + .file = &stdout_file, + .allocator = alloc, + }; + defer cli_recv.deinit(); var recv = cli_recv.receiver(); while (true) { @@ -356,12 +454,267 @@ pub fn main(init: std.process.Init) !void { } try conv.addUserMessage(line); + try appendUserPromptToSession( + alloc, + &session_mgr, + line, + banner_provider_initial, + banner_model_initial, + ); + const entries_before_step = conv.messages.items.len; + + cli_recv.beginTurn(); agent.runStep(&conv, &recv) catch |err| { try stdout.print("\n[error: {s}]\n", .{@errorName(err)}); }; + // Persist whatever new entries the agent produced this turn. The + // agent loop may have appended: + // - assistant message(s) (one per provider response) + // - user messages containing ToolResult blocks (one per tool round) + // Each assistant message gets paired with the Usage that the + // receiver captured at its onMessageComplete time (or null if + // the provider didn't emit usage that round). + try persistTurn( + alloc, + &session_mgr, + &conv, + entries_before_step, + banner_provider_initial, + banner_model_initial, + cli_recv.per_message_usage.items, + ); + try stdout.writeAll("\n> "); try stdout_file.flush(); } } + +// ----------------------------------------------------------------------------- +// CLI flag parsing +// ----------------------------------------------------------------------------- + +const AgentFlags = struct { + /// `--resume` without an id: resume the most recent session. + /// `--resume `: resume the session whose id has this prefix. + /// Not present: start a new session. + resume_kind: ResumeKind = .none, + resume_id: ?[]const u8 = null, // owned + + pub fn deinit(self: AgentFlags, alloc: std.mem.Allocator) void { + if (self.resume_id) |id| alloc.free(id); + } +}; + +const ResumeKind = enum { none, most_recent, by_id }; + +fn parseAgentFlags(alloc: std.mem.Allocator, args: std.process.Args) !AgentFlags { + var flags: AgentFlags = .{}; + errdefer flags.deinit(alloc); + + var it = args.iterate(); + defer it.deinit(); + _ = it.next(); // argv[0] + + while (it.next()) |a| { + if (std.mem.eql(u8, a, "--resume")) { + // Peek at the next arg. If it exists and doesn't start with `-`, + // treat it as a session id (or prefix). + const next = it.next(); + if (next) |id| { + if (id.len > 0 and id[0] != '-') { + flags.resume_kind = .by_id; + flags.resume_id = try alloc.dupe(u8, id); + continue; + } else { + // Not an id; rewind by treating it as a separate flag. + // The Args API doesn't support rewind, so handle inline. + flags.resume_kind = .most_recent; + if (std.mem.eql(u8, id, "--resume")) { + // back-to-back --resume; second resets, fine. + continue; + } + // Otherwise, fall through and process this token as a flag. + // (Currently we don't have other flags; ignore unknowns.) + std.log.warn("panto: ignoring unknown argument '{s}'", .{id}); + continue; + } + } + flags.resume_kind = .most_recent; + continue; + } + // Future agent-mode flags would land here. Unknown args are tolerated + // (the user might be passing something we don't recognize yet). + } + return flags; +} + +// ----------------------------------------------------------------------------- +// Session bootstrap +// ----------------------------------------------------------------------------- + +fn openSession( + alloc: std.mem.Allocator, + io: std.Io, + session_dir: []const u8, + cwd: []const u8, + flags: AgentFlags, + stdout: *std.Io.Writer, + stdout_file: *std.Io.File.Writer, +) !panto.session_manager.SessionManager { + switch (flags.resume_kind) { + .none => return try panto.session_manager.SessionManager.init( + alloc, + io, + session_dir, + cwd, + ), + .most_recent => { + const path_opt = panto.session_manager.findMostRecentSession(alloc, io, session_dir) catch null; + if (path_opt) |path| { + defer alloc.free(path); + return try panto.session_manager.SessionManager.open(alloc, io, path); + } + try stdout.print("no sessions to resume; starting fresh.\n", .{}); + try stdout_file.flush(); + return try panto.session_manager.SessionManager.init( + alloc, + io, + session_dir, + cwd, + ); + }, + .by_id => { + const id = flags.resume_id.?; + const path = panto.session_manager.resolveSessionId(alloc, io, session_dir, id) catch |err| switch (err) { + error.SessionNotFound => { + try stdout.print("error: no session matching '{s}' in {s}\n", .{ id, session_dir }); + try stdout_file.flush(); + return err; + }, + error.AmbiguousSessionId => { + try stdout.print("error: session id '{s}' is ambiguous\n", .{id}); + try stdout_file.flush(); + return err; + }, + else => return err, + }; + defer alloc.free(path); + return try panto.session_manager.SessionManager.open(alloc, io, path); + }, + } +} + +// ----------------------------------------------------------------------------- +// Session append helpers — bridge in-memory ContentBlocks to on-disk entries. +// ----------------------------------------------------------------------------- + +fn appendSystemToSession( + alloc: std.mem.Allocator, + mgr: *panto.session_manager.SessionManager, + text: []const u8, +) !void { + const blocks = try alloc.alloc(panto.session_manager.DiskContentBlock, 1); + blocks[0] = .{ .text = .{ .text = try alloc.dupe(u8, text) } }; + _ = try mgr.appendMessage( + .{ .role = .system, .content = blocks }, + null, + null, + ); +} + +fn appendUserPromptToSession( + alloc: std.mem.Allocator, + mgr: *panto.session_manager.SessionManager, + text: []const u8, + provider: []const u8, + model: []const u8, +) !void { + const blocks = try alloc.alloc(panto.session_manager.DiskContentBlock, 1); + blocks[0] = .{ .text = .{ .text = try alloc.dupe(u8, text) } }; + _ = try mgr.appendMessage( + .{ .role = .user, .content = blocks }, + provider, + model, + ); +} + +/// After the agent loop has driven a turn to completion, persist every +/// new in-memory message at index `>= start_index` to the session log. +/// +/// Each in-memory message is mapped to disk: +/// - assistant → assistant entry with provider/model/stop_reason metadata +/// and Usage (from `per_message_usage`, one slot per assistant message in +/// completion order). +/// - user (with ToolResult blocks) → user entry stamped with provider/model. +/// +/// `stop_reason` is recorded as `"stop"` for now; surfacing the real wire +/// value requires plumbing it through the Receiver vtable (future work, +/// separate from token plumbing). +fn persistTurn( + alloc: std.mem.Allocator, + mgr: *panto.session_manager.SessionManager, + conv: *const panto.conversation.Conversation, + start_index: usize, + provider: []const u8, + model: []const u8, + per_message_usage: []const ?panto.session_manager.Usage, +) !void { + var i = start_index; + var assistant_seen: usize = 0; + while (i < conv.messages.items.len) : (i += 1) { + const msg = conv.messages.items[i]; + const blocks = try alloc.alloc(panto.session_manager.DiskContentBlock, msg.content.items.len); + var allocated: usize = 0; + errdefer { + for (blocks[0..allocated]) |b| b.deinit(alloc); + alloc.free(blocks); + } + for (msg.content.items) |block| { + blocks[allocated] = try panto.session.contentBlockToDisk(alloc, block); + allocated += 1; + } + switch (msg.role) { + .system => { + // Mid-turn system messages aren't a thing in pantograph today. + // Treat as harmless and persist verbatim, sans stamps. + _ = try mgr.appendMessage( + .{ .role = .system, .content = blocks }, + null, + null, + ); + }, + .user => { + _ = try mgr.appendMessage( + .{ .role = .user, .content = blocks }, + provider, + model, + ); + }, + .assistant => { + // Pair this assistant message with its usage, if the + // receiver captured one for this position. (A turn + // ending in a stream error can have fewer usage entries + // than assistant messages; treat as null and move on.) + const usage: ?panto.session_manager.Usage = if (assistant_seen < per_message_usage.len) + per_message_usage[assistant_seen] + else + null; + assistant_seen += 1; + _ = try mgr.appendMessage( + .{ + .role = .assistant, + .content = blocks, + .provider = try alloc.dupe(u8, provider), + .model = try alloc.dupe(u8, model), + .stop_reason = try alloc.dupe(u8, "stop"), + .usage = usage, + }, + null, + null, + ); + }, + } + } +} diff --git a/src/models_toml.zig b/src/models_toml.zig new file mode 100644 index 0000000..5ccb302 --- /dev/null +++ b/src/models_toml.zig @@ -0,0 +1,291 @@ +//! Loader for `~/.config/panto/models.toml`. +//! +//! Schema: +//! +//! [.] +//! input = # USD per million tokens (fresh input) +//! output = # USD per million tokens +//! cache_read = # USD per million tokens (optional; default "unknown") +//! cache_write = # USD per million tokens (optional; default "unknown") +//! +//! All four price fields are optional at the parse layer. Any field +//! omitted from the TOML comes through as `null` in the in-memory +//! `Pricing`, which means "unknown price for this token category" — +//! NOT "zero." If the model later reports usage in an unknown +//! category (e.g. gpt-4o reads from prompt cache but the TOML omitted +//! `cache_read`), session cost degenerates to "unknown" rather than +//! silently treating that usage as free. To declare a category as a +//! known zero (e.g. OpenAI doesn't bill a cache-write rate), write +//! `cache_write = 0` explicitly. +//! +//! `` is `openai` or `anthropic` (matching pantograph's API +//! styles). `` is the model id pantograph sends to the API. Both +//! are TOML "dotted-key" path segments; quote them if they contain +//! characters TOML doesn't allow bare (`-`, `.`, `:` etc. — `-` is OK +//! bare; `.` requires quoting since dots separate path segments). +//! +//! Example: +//! +//! [anthropic."claude-sonnet-4-20250514"] +//! input = 3.0 +//! output = 15.0 +//! cache_read = 0.3 +//! cache_write = 3.75 +//! +//! [openai.gpt-4o] +//! input = 2.5 +//! output = 10.0 +//! +//! The model id in the section header may also be unquoted when it +//! contains no `.` or other reserved chars (e.g. `gpt-4o`); pantograph +//! reads either form. We do not currently bake in default pricing — +//! a missing entry (or a present entry with missing fields) surfaces +//! as "unknown cost" and the display layer formats accordingly. + +const std = @import("std"); +const Allocator = std.mem.Allocator; +const Io = std.Io; + +const toml = @import("toml"); +const panto = @import("panto"); + +pub const Pricing = panto.pricing.Pricing; +pub const Registry = panto.pricing.Registry; + +/// Resolve the absolute path to `models.toml`. Honors `XDG_CONFIG_HOME`, +/// falling back to `$HOME/.config`. Caller owns the returned slice. +pub fn configPath(allocator: Allocator, environ_map: *const std.process.Environ.Map) ![]u8 { + if (environ_map.get("XDG_CONFIG_HOME")) |xdg| { + return try std.fs.path.join(allocator, &.{ xdg, "panto", "models.toml" }); + } + if (environ_map.get("HOME")) |home| { + return try std.fs.path.join(allocator, &.{ home, ".config", "panto", "models.toml" }); + } + return error.NoHomeDirectory; +} + +/// Load `models.toml` into a fresh `Registry`. If the file does not +/// exist, returns an empty registry (no error — missing config is fine). +/// +/// On parse errors, returns `error.InvalidModelsToml` after logging the +/// line/column of the first error. +pub fn loadFromPath( + allocator: Allocator, + io: Io, + path: []const u8, +) !Registry { + var reg = Registry.init(allocator); + errdefer reg.deinit(); + + const file = Io.Dir.cwd().openFile(io, path, .{ .mode = .read_only }) catch |err| switch (err) { + error.FileNotFound => return reg, // empty registry; perfectly fine. + else => return err, + }; + defer file.close(io); + + const file_len = try file.length(io); + const bytes = try allocator.alloc(u8, @intCast(file_len)); + defer allocator.free(bytes); + _ = try file.readPositionalAll(io, bytes, 0); + + try parseInto(®, bytes); + return reg; +} + +/// Parse a TOML string into the given registry. Useful for tests. +pub fn parseInto(reg: *Registry, source: []const u8) !void { + const alloc = reg.allocator; + const result = toml.parseWithError(alloc, source, .{}); + switch (result) { + .err => |e| { + // Silenced in tests so the test runner doesn't flag the + // expected-failure case as a real error. + if (!@import("builtin").is_test) { + std.log.err( + "models.toml: parse error at line {d}, column {d}: {s}", + .{ e.line, e.column, e.message }, + ); + } + return error.InvalidModelsToml; + }, + .ok => |doc| { + defer { + var d = doc; + d.deinit(); + } + try ingestDocument(reg, doc); + }, + } +} + +fn ingestDocument(reg: *Registry, doc: *toml.Document) !void { + // Root is a table of provider -> table of model -> { input, output, ... }. + const root_val: *const toml.Value = doc.root; + if (root_val.* != .table) return; + var provider_it = toml.tableIterator(root_val); + while (provider_it.next()) |provider_entry| { + const provider = provider_entry.key; + const provider_val: *const toml.Value = provider_entry.value; + if (provider_val.* != .table) continue; + var model_it = toml.tableIterator(provider_val); + while (model_it.next()) |model_entry| { + const model = model_entry.key; + const v: *const toml.Value = model_entry.value; + if (v.* != .table) continue; + const pricing = pricingFromValue(v); + try reg.set(provider, model, pricing); + } + } +} + +fn pricingFromValue(v: *const toml.Value) Pricing { + return .{ + .input = readPriceField(v, "input"), + .output = readPriceField(v, "output"), + .cache_read = readPriceField(v, "cache_read"), + .cache_write = readPriceField(v, "cache_write"), + }; +} + +/// Returns `null` if the field is absent or has a type we can't +/// interpret as a price. An explicit `0` (or `0.0`) comes through as a +/// known zero, not `null` — callers rely on that distinction. +fn readPriceField(table: *const toml.Value, name: []const u8) ?u64 { + const field = table.get(name) orelse return null; + // Accept either floats (the natural form) or integers (the user + // wrote `3` instead of `3.0`). + if (field.asF64()) |f| return Pricing.fromDollarsPerMtok(f); + if (field.asI64()) |i| return Pricing.fromDollarsPerMtok(@floatFromInt(i)); + return null; +} + +// ============================================================================= +// Tests +// ============================================================================= + +const testing = std.testing; + +test "parseInto: two providers, multiple models" { + var reg = Registry.init(testing.allocator); + defer reg.deinit(); + + const src = + \\[anthropic."claude-sonnet-4-20250514"] + \\input = 3.0 + \\output = 15.0 + \\cache_read = 0.3 + \\cache_write = 3.75 + \\ + \\[openai.gpt-4o] + \\input = 2.5 + \\output = 10.0 + \\cache_read = 1.25 + \\ + \\[openai.gpt-4o-mini] + \\input = 0.15 + \\output = 0.6 + ; + try parseInto(®, src); + + const anth = reg.get("anthropic", "claude-sonnet-4-20250514").?; + try testing.expectEqual(@as(?u64, 300), anth.input); + try testing.expectEqual(@as(?u64, 1500), anth.output); + try testing.expectEqual(@as(?u64, 30), anth.cache_read); + try testing.expectEqual(@as(?u64, 375), anth.cache_write); + + const oa = reg.get("openai", "gpt-4o").?; + try testing.expectEqual(@as(?u64, 250), oa.input); + try testing.expectEqual(@as(?u64, 1000), oa.output); + try testing.expectEqual(@as(?u64, 125), oa.cache_read); + // cache_write absent in source — stays unknown, NOT silently 0. + try testing.expectEqual(@as(?u64, null), oa.cache_write); + + const mini = reg.get("openai", "gpt-4o-mini").?; + try testing.expectEqual(@as(?u64, 15), mini.input); + try testing.expectEqual(@as(?u64, 60), mini.output); + try testing.expectEqual(@as(?u64, null), mini.cache_read); + try testing.expectEqual(@as(?u64, null), mini.cache_write); +} + +test "parseInto: integer values are accepted (user writes `3` not `3.0`)" { + var reg = Registry.init(testing.allocator); + defer reg.deinit(); + + const src = + \\[openai.gpt-4o] + \\input = 3 + \\output = 15 + ; + try parseInto(®, src); + const p = reg.get("openai", "gpt-4o").?; + try testing.expectEqual(@as(?u64, 300), p.input); + try testing.expectEqual(@as(?u64, 1500), p.output); +} + +test "parseInto: missing optional fields stay null (unknown, not zero)" { + var reg = Registry.init(testing.allocator); + defer reg.deinit(); + + const src = + \\[openai.gpt-4o] + \\input = 2.5 + \\output = 10.0 + ; + try parseInto(®, src); + const p = reg.get("openai", "gpt-4o").?; + try testing.expectEqual(@as(?u64, null), p.cache_read); + try testing.expectEqual(@as(?u64, null), p.cache_write); +} + +test "parseInto: explicit 0 is a known zero, distinct from omission" { + // OpenAI doesn't charge for cache writes. Writing `cache_write = 0` + // in the TOML must produce a known 0 — not null — so cost stays + // computable on turns that report cache_write usage. + var reg = Registry.init(testing.allocator); + defer reg.deinit(); + + const src = + \\[openai.gpt-4o] + \\input = 2.5 + \\output = 10.0 + \\cache_read = 1.25 + \\cache_write = 0 + ; + try parseInto(®, src); + const p = reg.get("openai", "gpt-4o").?; + try testing.expectEqual(@as(?u64, 0), p.cache_write); +} + +test "parseInto: malformed TOML returns InvalidModelsToml" { + var reg = Registry.init(testing.allocator); + defer reg.deinit(); + try testing.expectError(error.InvalidModelsToml, parseInto(®, "this is not valid toml = =")); +} + +test "loadFromPath: missing file returns empty registry, no error" { + const io = testing.io; + var reg = try loadFromPath(testing.allocator, io, "/nonexistent/path/models.toml"); + defer reg.deinit(); + try testing.expectEqual(@as(usize, 0), reg.count()); +} + +test "configPath: XDG_CONFIG_HOME wins" { + const a = testing.allocator; + var env: std.process.Environ.Map = .init(a); + defer env.deinit(); + try env.put("XDG_CONFIG_HOME", "/custom/cfg"); + try env.put("HOME", "/ignored"); + const got = try configPath(a, &env); + defer a.free(got); + try testing.expectEqualStrings("/custom/cfg/panto/models.toml", got); +} + +test "configPath: falls back to HOME/.config" { + const a = testing.allocator; + var env: std.process.Environ.Map = .init(a); + defer env.deinit(); + try env.put("HOME", "/home/user"); + const got = try configPath(a, &env); + defer a.free(got); + try testing.expectEqualStrings("/home/user/.config/panto/models.toml", got); +} diff --git a/src/session_paths.zig b/src/session_paths.zig new file mode 100644 index 0000000..162085e --- /dev/null +++ b/src/session_paths.zig @@ -0,0 +1,141 @@ +//! Resolves session file locations from the environment. +//! +//! Layout (defaults; override base via `PANTO_SESSION_DIR`): +//! +//! $XDG_DATA_HOME/panto/sessions// +//! ↳ falls back to $HOME/.local/share/panto/sessions// +//! if $XDG_DATA_HOME is unset. +//! +//! `` is the working directory with leading `/` stripped and +//! every `/` or `:` replaced by `-`, with `--` glued to both ends. This +//! gives a flat directory name per project, easy to spot in `ls`. + +const std = @import("std"); +const Allocator = std.mem.Allocator; + +/// Resolve the absolute sessions directory for the given cwd. Caller owns +/// the returned slice. +/// +/// Precedence: +/// 1. `PANTO_SESSION_DIR` (full path to the base dir, used as-is — no +/// "panto/sessions" suffix is added) +/// 2. `XDG_DATA_HOME/panto/sessions` +/// 3. `HOME/.local/share/panto/sessions` +/// +/// In all three cases, `/` is appended. +pub fn sessionDirForCwd( + allocator: Allocator, + environ_map: *const std.process.Environ.Map, + cwd: []const u8, +) ![]u8 { + const base = try resolveSessionsBase(allocator, environ_map); + defer allocator.free(base); + + const encoded = try encodeCwd(allocator, cwd); + defer allocator.free(encoded); + + return try std.fs.path.join(allocator, &.{ base, encoded }); +} + +/// Resolve the absolute "sessions" base directory, before per-cwd grouping. +/// Caller owns the returned slice. +pub fn resolveSessionsBase( + allocator: Allocator, + environ_map: *const std.process.Environ.Map, +) ![]u8 { + if (environ_map.get("PANTO_SESSION_DIR")) |explicit| { + return try allocator.dupe(u8, explicit); + } + if (environ_map.get("XDG_DATA_HOME")) |xdg| { + return try std.fs.path.join(allocator, &.{ xdg, "panto", "sessions" }); + } + if (environ_map.get("HOME")) |home| { + return try std.fs.path.join(allocator, &.{ home, ".local", "share", "panto", "sessions" }); + } + return error.NoHomeDirectory; +} + +/// Encode a working directory into a flat directory name. Caller owns. +/// +/// Example: `/Users/travis/Code/pantograph` → `--Users-travis-Code-pantograph--` +pub fn encodeCwd(allocator: Allocator, cwd: []const u8) ![]u8 { + // Strip leading slash(es), then replace `/` and `:` with `-`. + var start: usize = 0; + while (start < cwd.len and (cwd[start] == '/' or cwd[start] == '\\')) : (start += 1) {} + const body = cwd[start..]; + const out = try allocator.alloc(u8, body.len + 4); // `--` + body + `--` + out[0] = '-'; + out[1] = '-'; + for (body, 0..) |c, i| { + out[2 + i] = if (c == '/' or c == '\\' or c == ':') '-' else c; + } + out[out.len - 2] = '-'; + out[out.len - 1] = '-'; + return out; +} + +// ============================================================================= +// Tests +// ============================================================================= + +const testing = std.testing; + +test "encodeCwd: replaces slashes and colons" { + const a = testing.allocator; + const got = try encodeCwd(a, "/Users/travis/Code/pantograph"); + defer a.free(got); + try testing.expectEqualStrings("--Users-travis-Code-pantograph--", got); +} + +test "encodeCwd: handles already-relative paths" { + const a = testing.allocator; + const got = try encodeCwd(a, "Users/travis"); + defer a.free(got); + try testing.expectEqualStrings("--Users-travis--", got); +} + +test "resolveSessionsBase: PANTO_SESSION_DIR wins" { + const a = testing.allocator; + var env: std.process.Environ.Map = .init(a); + defer env.deinit(); + try env.put("PANTO_SESSION_DIR", "/custom/sessions"); + try env.put("XDG_DATA_HOME", "/ignored"); + + const got = try resolveSessionsBase(a, &env); + defer a.free(got); + try testing.expectEqualStrings("/custom/sessions", got); +} + +test "resolveSessionsBase: XDG_DATA_HOME before HOME" { + const a = testing.allocator; + var env: std.process.Environ.Map = .init(a); + defer env.deinit(); + try env.put("XDG_DATA_HOME", "/x/data"); + try env.put("HOME", "/h"); + + const got = try resolveSessionsBase(a, &env); + defer a.free(got); + try testing.expectEqualStrings("/x/data/panto/sessions", got); +} + +test "resolveSessionsBase: falls back to HOME/.local/share" { + const a = testing.allocator; + var env: std.process.Environ.Map = .init(a); + defer env.deinit(); + try env.put("HOME", "/home/user"); + + const got = try resolveSessionsBase(a, &env); + defer a.free(got); + try testing.expectEqualStrings("/home/user/.local/share/panto/sessions", got); +} + +test "sessionDirForCwd: joins base and encoded cwd" { + const a = testing.allocator; + var env: std.process.Environ.Map = .init(a); + defer env.deinit(); + try env.put("PANTO_SESSION_DIR", "/sess"); + + const got = try sessionDirForCwd(a, &env, "/Users/travis/Code/pantograph"); + defer a.free(got); + try testing.expectEqualStrings("/sess/--Users-travis-Code-pantograph--", got); +} diff --git a/src/subcommand.zig b/src/subcommand.zig index 97eb977..f6c3058 100644 --- a/src/subcommand.zig +++ b/src/subcommand.zig @@ -23,6 +23,8 @@ const Io = std.Io; const lua_bridge = @import("lua_bridge.zig"); const luarocks_runtime = @import("luarocks_runtime.zig"); const self_exe = @import("self_exe.zig"); +const session_paths = @import("session_paths.zig"); +const panto = @import("panto"); const c = lua_bridge.c; @@ -72,9 +74,47 @@ pub fn dispatch( try runBootstrapSubcommand(allocator, io, environ_map, panto_executable_path, .{ .force = force }); return .done; } + if (std.mem.eql(u8, sub, "sessions")) { + try runSessionsSubcommand(allocator, io, environ_map); + return .done; + } + if (std.mem.eql(u8, sub, "--help") or std.mem.eql(u8, sub, "-h") or std.mem.eql(u8, sub, "help")) { + try printHelp(io); + return .done; + } return .agent; } +fn printHelp(io: Io) !void { + var buffer: [4096]u8 = undefined; + var stdout_file = std.Io.File.stdout().writer(io, &buffer); + const w = &stdout_file.interface; + try w.writeAll( + \\panto — a conversational coding agent + \\ + \\Usage: + \\ panto Start a new conversation. + \\ panto --resume Resume the most recent conversation in this directory. + \\ panto --resume Resume the conversation whose id begins with . + \\ panto sessions List saved sessions for this directory. + \\ panto bootstrap [--force] + \\ Run the luarocks bootstrap and exit. + \\ panto lua [args...] Drop into the embedded Lua interpreter. + \\ panto help Show this message. + \\ + \\Environment: + \\ PANTO_API_STYLE "openai_chat" (default) or "anthropic_messages". + \\ OPENAI_API_KEY, OPENAI_MODEL, OPENAI_BASE_URL, OPENAI_REASONING + \\ ANTHROPIC_API_KEY, ANTHROPIC_MODEL, ANTHROPIC_BASE_URL, + \\ ANTHROPIC_API_VERSION, ANTHROPIC_MAX_TOKENS + \\ PANTO_SESSION_DIR Override the base sessions directory. Defaults to + \\ $XDG_DATA_HOME/panto/sessions or ~/.local/share/panto/sessions. + \\ PANTO_HOME Override the runtime/rocks tree location. + \\ + ); + try stdout_file.flush(); +} + pub const BootstrapOptions = struct { /// Wipe the per-Lua-version tree before reinstalling everything. /// Surfaced as `panto bootstrap --force`. Equivalent to deleting @@ -204,6 +244,64 @@ fn runBootstrapSubcommand( ); } +// --------------------------------------------------------------------------- +// `panto sessions` +// --------------------------------------------------------------------------- + +/// List sessions for the current working directory. +/// +/// Output format (one session per line): +/// messages +/// +/// where `` is the first 8 hex chars of the session UUIDv7. +fn runSessionsSubcommand( + allocator: Allocator, + io: Io, + environ_map: *const std.process.Environ.Map, +) !void { + var cwd_buf: [std.fs.max_path_bytes]u8 = undefined; + const cwd_n = try std.process.currentPath(io, &cwd_buf); + const cwd = cwd_buf[0..cwd_n]; + + const session_dir = try session_paths.sessionDirForCwd(allocator, environ_map, cwd); + defer allocator.free(session_dir); + + const infos = try panto.session_manager.listSessions(allocator, io, session_dir, null); + defer panto.session_manager.freeSessionInfos(allocator, infos); + + var stdout_buffer: [4096]u8 = undefined; + var stdout_file = std.Io.File.stdout().writer(io, &stdout_buffer); + const stdout = &stdout_file.interface; + + if (infos.len == 0) { + try stdout.print("no sessions for {s}\n", .{cwd}); + try stdout_file.flush(); + return; + } + + for (infos) |info| { + const short = info.id[0..@min(8, info.id.len)]; + // `created` is ISO 8601 (e.g. `2026-04-25T17:40:15.990Z`). Trim + // to `YYYY-MM-DD HH:MM` for terseness. + const created_short = trimCreated(info.created); + try stdout.print( + "{s} {s} {d} messages\n", + .{ short, created_short, info.message_count }, + ); + } + try stdout_file.flush(); +} + +fn trimCreated(iso: []const u8) []const u8 { + if (iso.len < 16) return iso; + // `YYYY-MM-DDTHH:MM:...` → `YYYY-MM-DD HH:MM` (T → space). + // We can't mutate a borrowed slice, so just return a 16-byte slice + // of the original; the caller prints character-by-character via + // format, so the 'T' will still appear. Use a small buffer trick: + // return the slice unmodified — the 'T' is fine and unambiguous. + return iso[0..16]; +} + // --------------------------------------------------------------------------- // `panto lua` argv plumbing — sketched against the older Args API for // reference (kept here so the design notes survive the implementation). -- cgit v1.3