diff options
Diffstat (limited to 'libpanto/src/file_system_jsonl_store.zig')
| -rw-r--r-- | libpanto/src/file_system_jsonl_store.zig | 2103 |
1 files changed, 0 insertions, 2103 deletions
diff --git a/libpanto/src/file_system_jsonl_store.zig b/libpanto/src/file_system_jsonl_store.zig deleted file mode 100644 index 9bc9436..0000000 --- a/libpanto/src/file_system_jsonl_store.zig +++ /dev/null @@ -1,2103 +0,0 @@ -//! Session lifecycle: create, open, replay, append. -//! -//! Backed by an append-only JSONL file on disk. The on-disk types live in -//! `session.zig`. This module owns: -//! -//! - Path resolution (sessions dir is supplied by the caller; we own the -//! filename and writes within it). -//! - The in-memory entry index (`by_id` map + leaf pointer). -//! - Deferred file creation: the file is not written until the first -//! assistant message persists. Until that point, all entries are -//! buffered in memory. -//! - Append semantics: once flushed, every completed entry is written -//! and synced to disk immediately. -//! - Crash recovery: on open, the file is parsed line-by-line; the first -//! line that fails to parse causes everything from that line onward to -//! be truncated from the file. -//! - One-time format migration when a future version reads a v1 file -//! (currently a no-op; the hook is in place). -//! - Rebuilding a `Conversation` from the entry tree, plus determining -//! the active provider/model. -//! -//! The library-vs-CLI boundary: callers pass an absolute path to the -//! per-cwd sessions directory. We compute the per-session filename -//! ourselves (`<uuidv7>.jsonl`) and lazily mkdir the directory on the -//! first flush. The CLI owns XDG resolution, encoded-cwd grouping, and -//! the `--resume` flag plumbing. - -const std = @import("std"); -const Allocator = std.mem.Allocator; -const Io = std.Io; - -const session_mod = @import("session.zig"); -const conversation_mod = @import("conversation.zig"); -const session_store_mod = @import("session_store.zig"); -const turn_persist = @import("turn_persist.zig"); - -pub const SessionHeader = session_mod.SessionHeader; -pub const SessionEntry = session_mod.SessionEntry; -pub const MessageEntry = session_mod.MessageEntry; -pub const StoredMessage = session_mod.StoredMessage; -pub const StoredMessageRole = session_mod.StoredMessageRole; -pub const StoredSystemMode = session_mod.StoredSystemMode; -pub const StoredContentBlock = session_mod.StoredContentBlock; -pub const Usage = session_mod.Usage; -pub const CURRENT_VERSION = session_mod.CURRENT_VERSION; - -// ============================================================================= -// IDs and timestamps -// ============================================================================= - -/// Generate a UUIDv7 (RFC 9562 §5.7). Returns a 36-character canonical -/// hex string with hyphens. Caller owns. -/// -/// Layout: -/// - 48 bits: unix_ts_ms (big-endian) -/// - 4 bits: version (7) -/// - 12 bits: random -/// - 2 bits: variant (10) -/// - 62 bits: random -pub fn newUuidV7(allocator: Allocator, io: Io) ![]u8 { - const ts = Io.Timestamp.now(io, .real); - const now_ms: u64 = @intCast(@max(ts.toMilliseconds(), 0)); - - var rand_bytes: [10]u8 = undefined; - io.random(&rand_bytes); - - var b: [16]u8 = undefined; - // Timestamp (48 bits, big-endian). - b[0] = @intCast((now_ms >> 40) & 0xFF); - b[1] = @intCast((now_ms >> 32) & 0xFF); - b[2] = @intCast((now_ms >> 24) & 0xFF); - b[3] = @intCast((now_ms >> 16) & 0xFF); - b[4] = @intCast((now_ms >> 8) & 0xFF); - b[5] = @intCast(now_ms & 0xFF); - // Version (4 high bits = 0x7) + 12 bits random. - b[6] = 0x70 | (rand_bytes[0] & 0x0F); - b[7] = rand_bytes[1]; - // Variant (2 high bits = 10) + 62 bits random. - b[8] = 0x80 | (rand_bytes[2] & 0x3F); - b[9] = rand_bytes[3]; - b[10] = rand_bytes[4]; - b[11] = rand_bytes[5]; - b[12] = rand_bytes[6]; - b[13] = rand_bytes[7]; - b[14] = rand_bytes[8]; - b[15] = rand_bytes[9]; - - return try std.fmt.allocPrint( - allocator, - "{x:0>2}{x:0>2}{x:0>2}{x:0>2}-{x:0>2}{x:0>2}-{x:0>2}{x:0>2}-{x:0>2}{x:0>2}-{x:0>2}{x:0>2}{x:0>2}{x:0>2}{x:0>2}{x:0>2}", - .{ b[0], b[1], b[2], b[3], b[4], b[5], b[6], b[7], b[8], b[9], b[10], b[11], b[12], b[13], b[14], b[15] }, - ); -} - -/// Generate a fresh 8-character hex entry id. Caller owns. -fn newEntryIdInto(buf: []u8, io: Io) void { - std.debug.assert(buf.len == 8); - var bytes: [4]u8 = undefined; - io.random(&bytes); - _ = std.fmt.bufPrint(buf, "{x:0>2}{x:0>2}{x:0>2}{x:0>2}", .{ bytes[0], bytes[1], bytes[2], bytes[3] }) catch unreachable; -} - -/// Format `now` as an ISO 8601 UTC string with millisecond precision. -/// Example: `2026-04-25T17:40:15.990Z`. Caller owns. -pub fn isoTimestamp(allocator: Allocator, io: Io) ![]u8 { - const ts = Io.Timestamp.now(io, .real); - const ms_total: i64 = ts.toMilliseconds(); - const seconds_total: i64 = @divTrunc(ms_total, 1000); - const ms: u64 = @intCast(@mod(ms_total, 1000)); - - const epoch_secs = std.time.epoch.EpochSeconds{ .secs = @intCast(seconds_total) }; - const epoch_day = epoch_secs.getEpochDay(); - const day_secs = epoch_secs.getDaySeconds(); - const year_day = epoch_day.calculateYearDay(); - const month_day = year_day.calculateMonthDay(); - - return try std.fmt.allocPrint( - allocator, - "{d:0>4}-{d:0>2}-{d:0>2}T{d:0>2}:{d:0>2}:{d:0>2}.{d:0>3}Z", - .{ - @as(u32, year_day.year), - month_day.month.numeric(), - @as(u32, month_day.day_index) + 1, - day_secs.getHoursIntoDay(), - day_secs.getMinutesIntoHour(), - day_secs.getSecondsIntoMinute(), - ms, - }, - ); -} - -// ============================================================================= -// FileInfo (per-file listing scan result) -// ============================================================================= - -/// Internal scan result for one session file. Carries the file `path` (the -/// catalog needs it to open/resolve) plus everything the public -/// `session_store.SessionInfo` needs. -const FileInfo = struct { - path: []u8, - id: []u8, - created: []u8, // ISO 8601 from header timestamp - modified: []u8, // ISO 8601 from last activity, falling back to header - message_count: usize, - last_user_message: []u8, - stamp: ?session_mod.WireStamp, // last-used wire identity - - pub fn deinit(self: FileInfo, alloc: Allocator) void { - alloc.free(self.path); - alloc.free(self.id); - alloc.free(self.created); - alloc.free(self.modified); - alloc.free(self.last_user_message); - if (self.stamp) |s| s.deinit(alloc); - } -}; - -// ============================================================================= -// SessionFile -// ============================================================================= - -pub const Error = error{ - NoSessionsFound, - AmbiguousSessionId, - SessionNotFound, - InvalidSessionFile, -} || Allocator.Error || Io.Cancelable; - -pub const SessionFile = struct { - allocator: Allocator, - io: Io, - - /// Absolute path to the per-cwd sessions directory. Lazily created. - session_dir: []u8, - /// Absolute path to the file we *will* write to (computed at init). - /// May not yet exist on disk if `flushed = false`. - session_file: []u8, - - /// Header. Allocated at init for new sessions; reloaded from the file - /// on resume. - header: SessionHeader, - - /// Entries indexed in insertion order. The first entry's `parent_id` - /// is null; each subsequent entry's `parent_id` points to its parent - /// (currently always the previous entry). - entries: std.ArrayList(SessionEntry), - /// id → entry index in `entries`. Used both for parent-id lookups - /// and for collision detection in `newEntryId`. - by_id: std.StringHashMap(usize), - /// id of the most recently appended entry, or null if no entries yet. - /// Borrowed from the entry; do not free. - leaf_id: ?[]const u8, - - /// True once the file exists on disk. False during the "buffered" - /// pre-assistant phase. See module-level docs. - flushed: bool, - /// Number of bytes written to `session_file` so far. Used as the - /// offset for the next positional write. Only meaningful when - /// `flushed = true`. - written_bytes: u64, - - // ---------- Construction ---------- - - /// Create a new session in memory. Allocates a UUIDv7, computes the - /// file path, but does NOT touch the filesystem. The file is created - /// on the first assistant-message flush. - /// - /// `session_dir` is duplicated; the caller retains ownership of the - /// passed slice. - pub fn init( - allocator: Allocator, - io: Io, - session_dir: []const u8, - metadata: ?[]const u8, - ) !SessionFile { - const id = try newUuidV7(allocator, io); - defer allocator.free(id); - return initWithId(allocator, io, session_dir, metadata, id); - } - - /// Like `init`, but uses a caller-supplied session id (duped here) - /// rather than minting a fresh UUIDv7. Used by the catalog when a - /// `Session` handle was minted (with its id) before the first append. - pub fn initWithId( - allocator: Allocator, - io: Io, - session_dir: []const u8, - metadata: ?[]const u8, - session_id: []const u8, - ) !SessionFile { - const dir = try allocator.dupe(u8, session_dir); - errdefer allocator.free(dir); - - const id = try allocator.dupe(u8, session_id); - errdefer allocator.free(id); - - const timestamp = try isoTimestamp(allocator, io); - errdefer allocator.free(timestamp); - - const metadata_copy: ?[]const u8 = if (metadata) |m| try allocator.dupe(u8, m) else null; - errdefer if (metadata_copy) |m| allocator.free(m); - - 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, - .metadata = metadata_copy, - }, - .entries = .empty, - .by_id = std.StringHashMap(usize).init(allocator), - .leaf_id = null, - .flushed = false, - .written_bytes = 0, - }; - } - - /// Open and replay an existing session file. Truncates from the first - /// corrupted line. Runs format migration if needed and rewrites the - /// file once. - pub fn open( - allocator: Allocator, - io: Io, - file_path: []const u8, - ) !SessionFile { - const path_copy = try allocator.dupe(u8, file_path); - errdefer allocator.free(path_copy); - - // The session dir is the file's parent directory. - const dir_path = std.fs.path.dirname(path_copy) orelse "."; - const dir = try allocator.dupe(u8, dir_path); - errdefer allocator.free(dir); - - const bytes = try readWholeFile(allocator, io, path_copy); - defer allocator.free(bytes); - - // Walk line-by-line. The first failure causes a truncation back to - // the start of that line. - var entries: std.ArrayList(SessionEntry) = .empty; - errdefer { - for (entries.items) |e| e.deinit(allocator); - entries.deinit(allocator); - } - var by_id = std.StringHashMap(usize).init(allocator); - errdefer by_id.deinit(); - - var header_opt: ?SessionHeader = null; - errdefer if (header_opt) |h| h.deinit(allocator); - - var cursor: usize = 0; - var valid_bytes: u64 = 0; // length of the file prefix that parses cleanly - var saw_corruption: bool = false; - - while (cursor < bytes.len) { - // Find the next newline (or EOF). - const rest = bytes[cursor..]; - const nl_rel = std.mem.indexOfScalar(u8, rest, '\n'); - const line_end_excl: usize = if (nl_rel) |n| cursor + n else bytes.len; - const line = bytes[cursor..line_end_excl]; - const next_cursor: usize = if (nl_rel != null) line_end_excl + 1 else bytes.len; - - // Allow blank lines silently (just whitespace), but a non-empty - // trimmed line that won't parse triggers truncation. - const trimmed = std.mem.trim(u8, line, " \t\r"); - if (trimmed.len == 0) { - if (nl_rel == null) break; - cursor = next_cursor; - valid_bytes = cursor; - continue; - } - - // If the final line has no trailing newline AND we hit EOF, it - // is presumed truncated mid-write. Treat as corruption. - if (nl_rel == null) { - saw_corruption = true; - break; - } - - const fe = session_mod.parseLine(allocator, line) catch { - saw_corruption = true; - break; - }; - - switch (fe) { - .header => |h| { - if (header_opt != null) { - // Two headers — treat as corruption from this line on. - h.deinit(allocator); - saw_corruption = true; - break; - } - if (entries.items.len != 0) { - // Header arrived after entries — malformed. - h.deinit(allocator); - saw_corruption = true; - break; - } - header_opt = h; - }, - .entry => |e| { - const idx = entries.items.len; - entries.append(allocator, e) catch |err| { - e.deinit(allocator); - return err; - }; - by_id.put(e.base().id, idx) catch |err| { - // Rolling back the append is awkward; in practice - // OOM here is fatal anyway. - return err; - }; - }, - } - - cursor = next_cursor; - valid_bytes = cursor; - } - - // No header at all — refuse to load. - const header = header_opt orelse return error.InvalidSessionFile; - - // Truncate the file if anything beyond `valid_bytes` is corrupt. - if (saw_corruption and valid_bytes < bytes.len) { - try truncateFileTo(io, path_copy, valid_bytes); - } - - // Drop assistant tool_use blocks left dangling (no matching - // tool_result) by an interrupted prior turn. If anything changed, - // rebuild the id index and rewrite the file once. - if (elideDanglingToolUses(allocator, &entries)) { - by_id.clearRetainingCapacity(); - for (entries.items, 0..) |e, i| { - try by_id.put(e.base().id, i); - } - try rewriteFile(allocator, io, path_copy, 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 = header, - .entries = entries, - .by_id = by_id, - .leaf_id = leaf_id, - .flushed = true, - .written_bytes = stat, - }; - } - - pub fn deinit(self: *SessionFile) void { - self.header.deinit(self.allocator); - for (self.entries.items) |e| e.deinit(self.allocator); - self.entries.deinit(self.allocator); - self.by_id.deinit(); - self.allocator.free(self.session_dir); - self.allocator.free(self.session_file); - } - - // ---------- Accessors ---------- - - pub fn getSessionFile(self: *const SessionFile) []const u8 { - return self.session_file; - } - - pub fn isFlushed(self: *const SessionFile) bool { - return self.flushed; - } - - // ---------- Appending ---------- - - /// Append a single message, returning its entry id. A thin wrapper over - /// `appendMessagesAtomic` (which handles `len == 1`); used by the tests. - /// `msg` is consumed (ownership transferred). - pub fn appendMessage( - self: *SessionFile, - msg: StoredMessage, - // Wire-format provider identity for the entry. Null on system - // messages. Borrowed; duplicated into the entry. - stamp: ?session_mod.WireStamp, - ) ![]const u8 { - var msgs = [_]StoredMessage{msg}; - const stamps = [_]?session_mod.WireStamp{stamp}; - try self.appendMessagesAtomic(&msgs, &stamps); - return self.leaf_id.?; - } - - /// Returns a freshly allocated 8-character hex id, guaranteed not to - /// collide with any existing entry id in this session. - fn newEntryId(self: *SessionFile) ![]u8 { - const max_tries = 100; - var i: usize = 0; - while (i < max_tries) : (i += 1) { - const buf = try self.allocator.alloc(u8, 8); - errdefer self.allocator.free(buf); - newEntryIdInto(buf[0..8], self.io); - if (!self.by_id.contains(buf)) { - return buf; - } - self.allocator.free(buf); - } - // Fall back to a UUID prefix if 100 retries all collided. With 4 - // random bytes per id and a session with <<2^16 entries, the - // probability of getting here is effectively zero, but we want a - // hard guarantee. - const long = try newUuidV7(self.allocator, self.io); - defer self.allocator.free(long); - const buf = try self.allocator.alloc(u8, 8); - @memcpy(buf, long[0..8]); - return buf; - } - - // ---------- Persistence ---------- - - /// Write the header + all currently-buffered entries + `new_entries` - /// to the file as a single batch. Creates the directory and file. - fn flushBufferedMany(self: *SessionFile, new_entries: []const SessionEntry) !void { - try mkdirP(self.io, self.session_dir); - - const file = try Io.Dir.cwd().createFile(self.io, self.session_file, .{ - .truncate = true, - .read = false, - }); - defer file.close(self.io); - - var offset: u64 = 0; - const header_line = try session_mod.serializeHeader(self.allocator, self.header); - defer self.allocator.free(header_line); - try file.writePositionalAll(self.io, header_line, offset); - offset += header_line.len; - try file.writePositionalAll(self.io, "\n", offset); - offset += 1; - - for (self.entries.items) |e| { - const line = try session_mod.serializeEntry(self.allocator, e); - defer self.allocator.free(line); - try file.writePositionalAll(self.io, line, offset); - offset += line.len; - try file.writePositionalAll(self.io, "\n", offset); - offset += 1; - } - - for (new_entries) |entry| { - const line = try session_mod.serializeEntry(self.allocator, entry); - defer self.allocator.free(line); - try file.writePositionalAll(self.io, line, offset); - offset += line.len; - try file.writePositionalAll(self.io, "\n", offset); - offset += 1; - } - - file.sync(self.io) catch {}; - self.flushed = true; - self.written_bytes = offset; - } - - pub fn appendMessagesAtomic( - self: *SessionFile, - messages: []StoredMessage, - stamps: []const ?session_mod.WireStamp, - ) !void { - std.debug.assert(messages.len == stamps.len); - if (messages.len == 0) return; - - const base_len = self.entries.items.len; - try self.entries.ensureUnusedCapacity(self.allocator, messages.len); - try self.by_id.ensureUnusedCapacity(@intCast(messages.len)); - - var entries = try self.allocator.alloc(SessionEntry, messages.len); - defer self.allocator.free(entries); - var built: usize = 0; - errdefer { - for (entries[0..built]) |*e| e.deinit(self.allocator); - } - - var prev_leaf = self.leaf_id; - for (messages, 0..) |msg, i| { - const msg_local = msg; - const id_buf = try self.newEntryId(); - errdefer self.allocator.free(id_buf); - const timestamp = try isoTimestamp(self.allocator, self.io); - errdefer self.allocator.free(timestamp); - const parent_id_copy: ?[]const u8 = if (prev_leaf) |l| try self.allocator.dupe(u8, l) else null; - errdefer if (parent_id_copy) |p| self.allocator.free(p); - const stamp_copy: ?session_mod.WireStamp = if (stamps[i]) |st| try st.dupe(self.allocator) else null; - errdefer if (stamp_copy) |st| st.deinit(self.allocator); - entries[i] = .{ .message = .{ - .base = .{ .id = id_buf, .parent_id = parent_id_copy, .timestamp = timestamp }, - .stamp = stamp_copy, - .message = msg_local, - } }; - built += 1; - prev_leaf = entries[i].base().id; - } - - if (self.flushed) { - try self.persistEntries(entries); - } else { - // File is created on the first assistant message (see module docs). - for (messages) |m| if (m.role == .assistant) { - try self.flushBufferedMany(entries); - break; - }; - } - - for (entries, 0..) |entry, i| { - self.entries.appendAssumeCapacity(entry); - self.by_id.putAssumeCapacity(entry.base().id, base_len + i); - } - self.leaf_id = entries[entries.len - 1].base().id; - built = 0; - } - - fn persistEntries(self: *SessionFile, entries: []const SessionEntry) !void { - const file = try Io.Dir.cwd().openFile(self.io, self.session_file, .{ - .mode = .write_only, - }); - defer file.close(self.io); - - var offset = self.written_bytes; - for (entries) |entry| { - const line = try session_mod.serializeEntry(self.allocator, entry); - defer self.allocator.free(line); - try file.writePositionalAll(self.io, line, offset); - offset += line.len; - try file.writePositionalAll(self.io, "\n", offset); - offset += 1; - } - file.sync(self.io) catch {}; - self.written_bytes = offset; - } - - // ============================================================================= - // Conversation rebuild - // ============================================================================= - - /// Build a fresh `Conversation` from the entry log. Caller owns the - /// returned conversation (call `deinit`). - pub fn rebuildConversation(self: *const SessionFile) !conversation_mod.Conversation { - var conv = conversation_mod.Conversation.init(self.allocator); - errdefer conv.deinit(); - - for (self.entries.items) |entry| { - switch (entry) { - .message => |me| try appendMessageToConv(&conv, self.allocator, me.message, me.stamp), - } - } - return conv; - } -}; - -/// Best-effort extraction of plain prompt text from a user `StoredMessage`. -/// Used to populate `SessionInfo.last_user_message`. Returns null if the -/// message carries no plain text block. Caller owns the returned slice. -fn extractUserText(alloc: Allocator, msg: StoredMessage) !?[]u8 { - for (msg.content) |block| { - if (block == .text) { - return try alloc.dupe(u8, block.text.text); - } - } - return null; -} - -fn appendMessageToConv( - conv: *conversation_mod.Conversation, - allocator: Allocator, - disk_msg: StoredMessage, - stamp: ?session_mod.WireStamp, -) !void { - var content: std.ArrayList(conversation_mod.ContentBlock) = .empty; - errdefer { - for (content.items) |*b| { - var mut = b.*; - mut.deinit(allocator); - } - content.deinit(allocator); - } - try content.ensureTotalCapacity(allocator, disk_msg.content.len); - const sys_mode: conversation_mod.SystemMode = switch (disk_msg.mode) { - .append => .append, - .replace => .replace, - }; - for (disk_msg.content) |db| { - var block = try session_mod.diskContentBlockToInternal(allocator, db); - // System-role text blocks become `.System` blocks carrying the - // message's recorded mode, so the append/replace derivation works - // on the rebuilt conversation exactly as it did when written. - if (disk_msg.role == .system and block == .Text) { - const tb = block.Text; - block = .{ .System = .{ .text = tb, .mode = sys_mode } }; - } - if (block == .Thinking and stamp != null) { - block.Thinking.signature_origin = try conversation_mod.SignatureOrigin.init( - allocator, - stamp.?.api_style, - stamp.?.base_url, - stamp.?.model, - ); - } - content.appendAssumeCapacity(block); - } - const role: conversation_mod.MessageRole = switch (disk_msg.role) { - .system => .system, - .user => .user, - .assistant => .assistant, - }; - // Reconstruct the per-message producing identity from the wire stamp, so - // a later in-session compaction preserves it (rather than re-stamping the - // restated turn with the compaction model). Borrowed `stamp` slices are - // duped into the conversation allocator. - const identity: ?session_store_mod.WireIdentity = if (stamp) |st| - try conversation_mod.dupeWireIdentity(allocator, .{ - .api_style = st.api_style, - .base_url = st.base_url, - .model = st.model, - .reasoning = st.reasoning, - .thinking = st.thinking, - .effort = st.effort, - .thinking_budget_tokens = st.thinking_budget_tokens, - .thinking_interleaved = st.thinking_interleaved, - }) - else - null; - errdefer if (identity) |id| conversation_mod.freeWireIdentity(allocator, id); - // Carry the recorded usage forward so compaction can size the retention - // window after a session is reopened (it's null for user/system). - try conv.messages.append(allocator, .{ - .role = role, - .content = content, - .usage = disk_msg.usage, - .identity = identity, - }); -} - -// ============================================================================= -// Entry repair on load -// ============================================================================= - -fn elideDanglingToolUses(allocator: Allocator, entries: *std.ArrayList(SessionEntry)) bool { - var needed: std.StringHashMap(void) = .init(allocator); - defer needed.deinit(); - var changed = false; - - var i = entries.items.len; - while (i > 0) { - i -= 1; - const entry = &entries.items[i]; - if (entry.* != .message) continue; - const msg = &entry.message.message; - - if (msg.role == .user) { - for (msg.content) |block| { - if (block == .tool_result) { - needed.put(block.tool_result.tool_use_id, {}) catch {}; - } - } - continue; - } - - if (msg.role != .assistant) continue; - var kept: std.ArrayList(StoredContentBlock) = .empty; - defer kept.deinit(allocator); - var removed = false; - for (msg.content) |block| { - if (block == .tool_use and !needed.contains(block.tool_use.id)) { - block.deinit(allocator); - removed = true; - continue; - } - kept.append(allocator, block) catch unreachable; - } - if (!removed) continue; - allocator.free(msg.content); - msg.content = kept.toOwnedSlice(allocator) catch unreachable; - changed = true; - } - return changed; -} - -// ============================================================================= -// File utilities -// ============================================================================= - -fn readWholeFile(allocator: Allocator, io: Io, path: []const u8) ![]u8 { - const file = Io.Dir.cwd().openFile(io, path, .{ .mode = .read_only }) catch |err| switch (err) { - error.FileNotFound => return error.InvalidSessionFile, - else => return err, - }; - defer file.close(io); - - const len = file.length(io) catch { - // Fall back to a streaming read of a reasonable upper bound. - // Sessions over ~10 MB are out of scope for phase 4. - var list: std.ArrayList(u8) = .empty; - defer list.deinit(allocator); - var chunk: [4096]u8 = undefined; - while (true) { - const n = file.readStreaming(io, &.{&chunk}) catch break; - if (n == 0) break; - try list.appendSlice(allocator, chunk[0..n]); - } - return try list.toOwnedSlice(allocator); - }; - - const buf = try allocator.alloc(u8, @intCast(len)); - errdefer allocator.free(buf); - _ = try file.readPositionalAll(io, buf, 0); - return buf; -} - -fn statFileForLength(io: Io, path: []const u8) !u64 { - const file = try Io.Dir.cwd().openFile(io, path, .{ .mode = .read_only }); - defer file.close(io); - return try file.length(io); -} - -fn truncateFileTo(io: Io, path: []const u8, new_length: u64) !void { - const file = try Io.Dir.cwd().openFile(io, path, .{ .mode = .write_only }); - defer file.close(io); - try file.setLength(io, new_length); - file.sync(io) catch {}; -} - -/// Write a fresh file containing `header` followed by `entries`. Truncates -/// any existing content. Used after a migration rewrites the format. -fn rewriteFile( - allocator: Allocator, - io: Io, - path: []const u8, - header: SessionHeader, - entries: []const SessionEntry, -) !void { - const file = try Io.Dir.cwd().createFile(io, path, .{ - .truncate = true, - .read = false, - }); - defer file.close(io); - - var offset: u64 = 0; - const header_line = try session_mod.serializeHeader(allocator, header); - defer allocator.free(header_line); - try file.writePositionalAll(io, header_line, offset); - offset += header_line.len; - try file.writePositionalAll(io, "\n", offset); - offset += 1; - for (entries) |e| { - const line = try session_mod.serializeEntry(allocator, e); - defer allocator.free(line); - try file.writePositionalAll(io, line, offset); - offset += line.len; - try file.writePositionalAll(io, "\n", offset); - offset += 1; - } - file.sync(io) catch {}; -} - -fn mkdirP(io: Io, path: []const u8) !void { - Io.Dir.cwd().createDirPath(io, path) catch |err| switch (err) { - error.PathAlreadyExists => {}, - else => return err, - }; -} - -// ============================================================================= -// Listing -// ============================================================================= - -/// List sessions in `session_dir`. Returns a slice of `SessionInfo`s -/// sorted by `modified` descending (most recent first). Caller owns the -/// slice and each `SessionInfo`. -/// -/// If the directory does not exist, returns an empty slice (no error). -/// -/// If `on_progress` is non-null, it is invoked after each file is parsed. -pub fn listSessions( - allocator: Allocator, - io: Io, - session_dir: []const u8, - on_progress: ?*const fn (loaded: usize, total: usize) void, -) ![]FileInfo { - var dir = Io.Dir.cwd().openDir(io, session_dir, .{}) catch |err| switch (err) { - error.FileNotFound => return try allocator.alloc(FileInfo, 0), - else => return err, - }; - defer dir.close(io); - - var names: std.ArrayList([]u8) = .empty; - defer { - for (names.items) |n| allocator.free(n); - names.deinit(allocator); - } - - var it = dir.iterate(); - while (try it.next(io)) |entry| { - if (entry.kind != .file) continue; - if (!std.mem.endsWith(u8, entry.name, ".jsonl")) continue; - const copy = try allocator.dupe(u8, entry.name); - errdefer allocator.free(copy); - try names.append(allocator, copy); - } - - var infos: std.ArrayList(FileInfo) = .empty; - errdefer { - for (infos.items) |i| i.deinit(allocator); - infos.deinit(allocator); - } - try infos.ensureTotalCapacity(allocator, names.items.len); - - var loaded: usize = 0; - for (names.items) |name| { - const full = try std.fs.path.join(allocator, &.{ session_dir, name }); - defer allocator.free(full); - const info_opt = buildFileInfo(allocator, io, full) catch null; - if (info_opt) |info| { - infos.appendAssumeCapacity(info); - } - loaded += 1; - if (on_progress) |cb| cb(loaded, names.items.len); - } - - const slice = try infos.toOwnedSlice(allocator); - std.sort.pdq(FileInfo, slice, {}, fileInfoNewerFirst); - return slice; -} - -fn fileInfoNewerFirst(_: void, a: FileInfo, b: FileInfo) bool { - return std.mem.order(u8, a.modified, b.modified) == .gt; -} - -fn buildFileInfo( - allocator: Allocator, - io: Io, - file_path: []const u8, -) !?FileInfo { - const bytes = readWholeFile(allocator, io, file_path) catch return null; - defer allocator.free(bytes); - - var header_opt: ?SessionHeader = null; - defer if (header_opt) |h| h.deinit(allocator); - - var message_count: usize = 0; - var last_activity: ?[]u8 = null; - defer if (last_activity) |la| allocator.free(la); - var last_user: ?[]u8 = null; - defer if (last_user) |lu| allocator.free(lu); - var last_stamp: ?session_mod.WireStamp = null; - defer if (last_stamp) |st| st.deinit(allocator); - - var lines = std.mem.splitScalar(u8, bytes, '\n'); - while (lines.next()) |line| { - const trimmed = std.mem.trim(u8, line, " \t\r"); - if (trimmed.len == 0) continue; - const fe = session_mod.parseLine(allocator, trimmed) catch break; - switch (fe) { - .header => |h| { - if (header_opt != null) { - h.deinit(allocator); - } else { - header_opt = h; - } - }, - .entry => |e| { - defer e.deinit(allocator); - switch (e) { - .message => |m| { - if (m.message.role == .user or m.message.role == .assistant) { - message_count += 1; - if (last_activity) |la| allocator.free(la); - last_activity = try allocator.dupe(u8, m.base.timestamp); - } - if (m.stamp) |st| { - if (last_stamp) |old| old.deinit(allocator); - last_stamp = try st.dupe(allocator); - } - if (m.message.role == .user) { - if (try extractUserText(allocator, m.message)) |ut| { - if (last_user) |lu| allocator.free(lu); - last_user = ut; - } - } - }, - } - }, - } - } - - const header = header_opt orelse return null; - - const path = try allocator.dupe(u8, file_path); - errdefer allocator.free(path); - const id = try allocator.dupe(u8, header.id); - errdefer allocator.free(id); - const created = try allocator.dupe(u8, header.timestamp); - errdefer allocator.free(created); - const modified = if (last_activity) |la| blk: { - last_activity = null; - break :blk la; - } else try allocator.dupe(u8, header.timestamp); - errdefer allocator.free(modified); - const last_user_message = if (last_user) |lu| blk: { - last_user = null; - break :blk lu; - } else try allocator.dupe(u8, ""); - errdefer allocator.free(last_user_message); - const stamp_out = if (last_stamp) |st| blk: { - last_stamp = null; - break :blk st; - } else null; - - return .{ - .path = path, - .id = id, - .created = created, - .modified = modified, - .message_count = message_count, - .last_user_message = last_user_message, - .stamp = stamp_out, - }; -} - -// ============================================================================= -// Recent / resume helpers -// ============================================================================= - -/// Resolve a (possibly abbreviated) session id to a session file path -/// within `session_dir`. Errors if no match or ambiguous prefix. -pub fn resolveSessionId( - allocator: Allocator, - io: Io, - session_dir: []const u8, - id_or_prefix: []const u8, -) ![]u8 { - var dir = Io.Dir.cwd().openDir(io, session_dir, .{}) catch |err| switch (err) { - error.FileNotFound => return error.SessionNotFound, - else => return err, - }; - defer dir.close(io); - - var match: ?[]u8 = null; - errdefer if (match) |m| allocator.free(m); - - var it = dir.iterate(); - while (try it.next(io)) |entry| { - if (entry.kind != .file) continue; - if (!std.mem.endsWith(u8, entry.name, ".jsonl")) continue; - // Strip `.jsonl` for the prefix match. - const stem = entry.name[0 .. entry.name.len - ".jsonl".len]; - if (!std.mem.startsWith(u8, stem, id_or_prefix)) continue; - if (match != null) return error.AmbiguousSessionId; - match = try allocator.dupe(u8, entry.name); - } - - const name = match orelse return error.SessionNotFound; - defer allocator.free(name); - match = null; - return try std.fs.path.join(allocator, &.{ session_dir, name }); -} - -// ============================================================================= -// FileSystemJSONLStore — the directory-backed catalog (SessionStore impl) -// ============================================================================= - -/// A directory-backed `SessionStore`: each session is one `<id>.jsonl` file -/// under `dir`. The catalog mints `Session` handles, lists/resolves files, -/// loads conversations, and routes appends to the right `SessionFile`. -/// -/// Open `SessionFile`s are cached by id for the catalog's lifetime so the -/// buffered-until-first-assistant write discipline survives across the -/// separate user-prompt and assistant-turn appends of a single turn. -/// -/// `dir` is the already-resolved sessions directory (the panto CLI derives -/// the per-cwd grouping; the store itself is cwd-agnostic). Optional -/// `metadata` is stamped into new session headers for display/provenance only. -pub const FileSystemJSONLStore = struct { - allocator: Allocator, - io: Io, - dir: []u8, // owned: the sessions directory - metadata: ?[]u8, // owned: recorded in new session headers - open: std.StringHashMap(*SessionFile), - - pub fn init(allocator: Allocator, io: Io, dir: []const u8) !FileSystemJSONLStore { - return initWithMetadata(allocator, io, dir, null); - } - - pub fn initWithMetadata(allocator: Allocator, io: Io, dir: []const u8, metadata: ?[]const u8) !FileSystemJSONLStore { - const dir_copy = try allocator.dupe(u8, dir); - errdefer allocator.free(dir_copy); - const metadata_copy: ?[]u8 = if (metadata) |m| try allocator.dupe(u8, m) else null; - return .{ - .allocator = allocator, - .io = io, - .dir = dir_copy, - .metadata = metadata_copy, - .open = std.StringHashMap(*SessionFile).init(allocator), - }; - } - - pub fn deinit(self: *FileSystemJSONLStore) void { - var it = self.open.iterator(); - while (it.next()) |e| { - e.value_ptr.*.deinit(); - self.allocator.destroy(e.value_ptr.*); - self.allocator.free(e.key_ptr.*); - } - self.open.deinit(); - self.allocator.free(self.dir); - if (self.metadata) |m| self.allocator.free(m); - } - - /// Borrow (opening if needed) the `SessionFile` for `id`, caching it. - /// Returns null if the file does not exist on disk and `create_missing` - /// is false. - fn fileFor(self: *FileSystemJSONLStore, id: []const u8, create_missing: bool) !?*SessionFile { - if (self.open.get(id)) |sf| return sf; - // Locate the file by exact id. - const path = try std.fs.path.join(self.allocator, &.{ self.dir, id }); - defer self.allocator.free(path); - const full = try std.fmt.allocPrint(self.allocator, "{s}.jsonl", .{path}); - defer self.allocator.free(full); - - const exists = blk: { - Io.Dir.cwd().access(self.io, full, .{}) catch break :blk false; - break :blk true; - }; - if (!exists and !create_missing) return null; - - const sf = try self.allocator.create(SessionFile); - errdefer self.allocator.destroy(sf); - sf.* = if (exists) - try SessionFile.open(self.allocator, self.io, full) - else - try SessionFile.initWithId(self.allocator, self.io, self.dir, self.metadata, id); - errdefer sf.deinit(); - - const key = try self.allocator.dupe(u8, id); - errdefer self.allocator.free(key); - try self.open.put(key, sf); - return sf; - } - - fn infoFromFileInfo(self: *FileSystemJSONLStore, fi: FileInfo) !session_store_mod.SessionInfo { - const id = try self.allocator.dupe(u8, fi.id); - errdefer self.allocator.free(id); - const created = try self.allocator.dupe(u8, fi.created); - errdefer self.allocator.free(created); - const modified = try self.allocator.dupe(u8, fi.modified); - errdefer self.allocator.free(modified); - const last_user = try self.allocator.dupe(u8, fi.last_user_message); - errdefer self.allocator.free(last_user); - const base_url = try self.allocator.dupe(u8, if (fi.stamp) |s| s.base_url else ""); - errdefer self.allocator.free(base_url); - const model = try self.allocator.dupe(u8, if (fi.stamp) |s| s.model else ""); - return .{ - .id = id, - .created = created, - .modified = modified, - .message_count = fi.message_count, - .last_user_message = last_user, - .api_style = if (fi.stamp) |s| s.api_style else .openai_chat, - .base_url = base_url, - .model = model, - .reasoning = if (fi.stamp) |s| s.reasoning else .default, - }; - } - - // ---------- vtable ---------- - - fn createVT(ctx: *anyopaque) session_store_mod.Session { - const self: *FileSystemJSONLStore = @ptrCast(@alignCast(ctx)); - // Mint a fresh id; nothing hits disk until the first append. - const id = newUuidV7(self.allocator, self.io) catch ""; - // The SessionFile is created lazily on first append via fileFor. - const info: session_store_mod.SessionInfo = .{ - .id = id, - .created = self.allocator.dupe(u8, "") catch "", - .modified = self.allocator.dupe(u8, "") catch "", - .message_count = 0, - .last_user_message = self.allocator.dupe(u8, "") catch "", - .api_style = .openai_chat, - .base_url = self.allocator.dupe(u8, "") catch "", - .model = self.allocator.dupe(u8, "") catch "", - .reasoning = .default, - }; - return .{ .info = info, .store = self.store() }; - } - - fn listVT(ctx: *anyopaque) anyerror![]session_store_mod.SessionInfo { - const self: *FileSystemJSONLStore = @ptrCast(@alignCast(ctx)); - const fis = try listSessions(self.allocator, self.io, self.dir, null); - defer { - for (fis) |fi| fi.deinit(self.allocator); - self.allocator.free(fis); - } - var out = try self.allocator.alloc(session_store_mod.SessionInfo, fis.len); - var built: usize = 0; - errdefer { - for (out[0..built]) |i| i.deinit(self.allocator); - self.allocator.free(out); - } - for (fis, 0..) |fi, i| { - out[i] = try self.infoFromFileInfo(fi); - built += 1; - } - return out; - } - - fn freeSessionInfosVT(ctx: *anyopaque, infos: []session_store_mod.SessionInfo) void { - const self: *FileSystemJSONLStore = @ptrCast(@alignCast(ctx)); - for (infos) |i| i.deinit(self.allocator); - self.allocator.free(infos); - } - - fn resolveVT(ctx: *anyopaque, id_or_prefix: []const u8) anyerror!?session_store_mod.Session { - const self: *FileSystemJSONLStore = @ptrCast(@alignCast(ctx)); - const path = resolveSessionId(self.allocator, self.io, self.dir, id_or_prefix) catch |err| switch (err) { - error.SessionNotFound => return null, - else => return err, - }; - defer self.allocator.free(path); - return try self.sessionFromPath(path); - } - - fn latestVT(ctx: *anyopaque) anyerror!?session_store_mod.Session { - const self: *FileSystemJSONLStore = @ptrCast(@alignCast(ctx)); - // Most-recently-*modified* wins, matching the `list()` sort order — - // not newest-created (lexicographic UUIDv7 filename). - const fis = try listSessions(self.allocator, self.io, self.dir, null); - defer { - for (fis) |fi| fi.deinit(self.allocator); - self.allocator.free(fis); - } - if (fis.len == 0) return null; - const info = try self.infoFromFileInfo(fis[0]); - return .{ .info = info, .store = self.store() }; - } - - fn sessionFromPath(self: *FileSystemJSONLStore, path: []const u8) !?session_store_mod.Session { - const fi = (try buildFileInfo(self.allocator, self.io, path)) orelse return null; - defer fi.deinit(self.allocator); - const info = try self.infoFromFileInfo(fi); - return .{ .info = info, .store = self.store() }; - } - - fn loadVT(ctx: *anyopaque, id: []const u8) anyerror!?conversation_mod.Conversation { - const self: *FileSystemJSONLStore = @ptrCast(@alignCast(ctx)); - const sf = (try self.fileFor(id, false)) orelse return null; - return try sf.rebuildConversation(); - } - - fn appendMessagesVT( - ctx: *anyopaque, - session_id: []const u8, - messages: []session_store_mod.PersistentMessage, - ) anyerror!void { - const self: *FileSystemJSONLStore = @ptrCast(@alignCast(ctx)); - if (messages.len == 0) return; - const sf = (try self.fileFor(session_id, true)).?; - - // Convert each rich PersistentMessage to a StoredMessage + wire stamp. - // The FS store deliberately ignores the `conversation` and - // `tools_available` provenance fields. - var stored = try self.allocator.alloc(StoredMessage, messages.len); - var stamps = try self.allocator.alloc(?session_mod.WireStamp, messages.len); - defer self.allocator.free(stored); - defer self.allocator.free(stamps); - var built: usize = 0; - errdefer for (stored[0..built]) |sm| sm.deinit(self.allocator); - - for (messages, 0..) |pm, i| { - stored[i] = try persistentToStored(self.allocator, pm); - // System entries carry no wire stamp; user/assistant do. - stamps[i] = if (pm.message.role == .system) null else .{ - .api_style = pm.identity.api_style, - .base_url = pm.identity.base_url, - .model = pm.identity.model, - .reasoning = pm.identity.reasoning, - .thinking = pm.identity.thinking, - .effort = pm.identity.effort, - .thinking_budget_tokens = pm.identity.thinking_budget_tokens, - .thinking_interleaved = pm.identity.thinking_interleaved, - }; - built += 1; - } - - // appendMessagesAtomic consumes the StoredMessages; it dupes stamps. - try sf.appendMessagesAtomic(stored, stamps); - } - - const store_vtable: session_store_mod.SessionStore.VTable = .{ - .create = createVT, - .list = listVT, - .freeSessionInfos = freeSessionInfosVT, - .resolve = resolveVT, - .latest = latestVT, - .load = loadVT, - .appendMessages = appendMessagesVT, - }; - - /// Wrap this catalog as a neutral `SessionStore`. The handle borrows - /// `self`; `self` must outlive it. - pub fn store(self: *FileSystemJSONLStore) session_store_mod.SessionStore { - return .{ .ptr = self, .vtable = &store_vtable }; - } -}; - -/// Convert a rich in-memory `PersistentMessage` to the on-disk -/// `StoredMessage`. Strings are duplicated; the source is untouched. -fn persistentToStored( - alloc: Allocator, - pm: session_store_mod.PersistentMessage, -) !StoredMessage { - const msg = pm.message; - const blocks = try alloc.alloc(session_mod.StoredContentBlock, msg.content.items.len); - var allocated: usize = 0; - errdefer { - for (blocks[0..allocated]) |b| b.deinit(alloc); - alloc.free(blocks); - } - for (msg.content.items) |block| { - blocks[allocated] = try session_mod.contentBlockToDisk(alloc, block); - allocated += 1; - } - const mode: session_mod.StoredSystemMode = blk: { - for (msg.content.items) |block| { - if (block == .System and block.System.mode == .replace) break :blk .replace; - } - break :blk .append; - }; - const stop_reason: ?[]const u8 = if (msg.role == .assistant) try alloc.dupe(u8, "stop") else null; - errdefer if (stop_reason) |s| alloc.free(s); - const metadata: ?[]const u8 = if (msg.metadata) |m| try alloc.dupe(u8, m) else null; - const role: session_mod.StoredMessageRole = switch (msg.role) { - .system => .system, - .user => .user, - .assistant => .assistant, - }; - return .{ - .role = role, - .content = blocks, - .mode = mode, - .stop_reason = stop_reason, - .usage = pm.usage, - .metadata = metadata, - }; -} - -// ============================================================================= -// Tests -// ============================================================================= - -const testing = std.testing; - -/// Borrowed wire stamps for tests (no allocation; the manager dupes them). -fn oaStamp() session_mod.WireStamp { - return .{ .api_style = .openai_chat, .base_url = "https://api.openai.com/v1", .model = "gpt-4o" }; -} -fn anStamp() session_mod.WireStamp { - return .{ .api_style = .anthropic_messages, .base_url = "https://api.anthropic.com", .model = "claude-sonnet-4-20250514" }; -} - -/// Test helper: append a single-text user message (see the `addUserMessage` -/// signature change to a block slice). -fn addUserText(conv: *conversation_mod.Conversation, text: []const u8) !void { - const tb = try conversation_mod.textualBlockFromSlice(conv.allocator, text); - var block: conversation_mod.ContentBlock = .{ .Text = tb }; - errdefer block.deinit(conv.allocator); - try conv.addUserMessage(&.{block}); -} - -test "newUuidV7: produces 36-char hyphenated string with version 7" { - const io = testing.io; - const id = try newUuidV7(testing.allocator, io); - defer testing.allocator.free(id); - try testing.expectEqual(@as(usize, 36), id.len); - // Position 14 is the version nibble — should be '7'. - try testing.expectEqual(@as(u8, '7'), id[14]); - // Hyphens at canonical positions. - try testing.expectEqual(@as(u8, '-'), id[8]); - try testing.expectEqual(@as(u8, '-'), id[13]); - try testing.expectEqual(@as(u8, '-'), id[18]); - try testing.expectEqual(@as(u8, '-'), id[23]); -} - -test "isoTimestamp: well-formed ISO 8601 with millisecond precision" { - const ts = try isoTimestamp(testing.allocator, testing.io); - defer testing.allocator.free(ts); - try testing.expectEqual(@as(usize, 24), ts.len); - try testing.expectEqual(@as(u8, '-'), ts[4]); - try testing.expectEqual(@as(u8, 'T'), ts[10]); - try testing.expectEqual(@as(u8, '.'), ts[19]); - try testing.expectEqual(@as(u8, 'Z'), ts[23]); -} - -// ---- In-memory + filesystem tests (use a tmp dir) ---- - -const TmpSessionDir = struct { - parent: std.testing.TmpDir, - abs_path: []u8, - - fn init(allocator: Allocator) !TmpSessionDir { - var parent = std.testing.tmpDir(.{}); - errdefer parent.cleanup(); - var path_buf: [std.fs.max_path_bytes]u8 = undefined; - const n = try parent.dir.realPath(testing.io, &path_buf); - const abs = try allocator.dupe(u8, path_buf[0..n]); - return .{ .parent = parent, .abs_path = abs }; - } - - fn deinit(self: *TmpSessionDir, allocator: Allocator) void { - allocator.free(self.abs_path); - self.parent.cleanup(); - } -}; - -test "SessionFile.init: does not create file yet" { - const io = testing.io; - - var td = try TmpSessionDir.init(testing.allocator); - defer td.deinit(testing.allocator); - - // Use a non-existent subdirectory inside the tmp dir to also exercise - // lazy directory creation. - const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); - defer testing.allocator.free(sessions); - - var mgr = try SessionFile.init( - testing.allocator, - io, - sessions, - "{\"cwd\":\"/some/cwd\"}", - ); - defer mgr.deinit(); - - try testing.expect(!mgr.isFlushed()); - - // The directory should not exist yet. - const stat_err = Io.Dir.cwd().openDir(io, sessions, .{}); - try testing.expectError(error.FileNotFound, stat_err); -} - -test "SessionFile: full flow — buffer, flush on assistant, append, resume" { - const io = testing.io; - - var td = try TmpSessionDir.init(testing.allocator); - defer td.deinit(testing.allocator); - - const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); - defer testing.allocator.free(sessions); - - const session_file: []u8 = blk: { - var mgr = try SessionFile.init( - testing.allocator, - io, - sessions, - "{\"cwd\":\"/proj/foo\"}", - ); - defer mgr.deinit(); - - // System message buffers in memory — nothing on disk yet. - const sys_blocks = try testing.allocator.alloc(StoredContentBlock, 1); - sys_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "You are helpful.") } }; - _ = try mgr.appendMessage( - .{ .role = .system, .content = sys_blocks }, - null, - ); - try testing.expect(!mgr.isFlushed()); - - // User message — still buffered. - const usr_blocks = try testing.allocator.alloc(StoredContentBlock, 1); - usr_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hi there") } }; - _ = try mgr.appendMessage( - .{ .role = .user, .content = usr_blocks }, - oaStamp(), - ); - try testing.expect(!mgr.isFlushed()); - - // Assistant message — first flush: header + all buffered entries. - const a_blocks = try testing.allocator.alloc(StoredContentBlock, 1); - a_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hello") } }; - _ = try mgr.appendMessage( - .{ - .role = .assistant, - .content = a_blocks, - .stop_reason = try testing.allocator.dupe(u8, "stop"), - }, - oaStamp(), - ); - try testing.expect(mgr.isFlushed()); - - // Append another user/assistant round. - const u_two = try testing.allocator.alloc(StoredContentBlock, 1); - u_two[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "what's 2+2?") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u_two }, oaStamp()); - - const a2 = try testing.allocator.alloc(StoredContentBlock, 1); - a2[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "4") } }; - _ = try mgr.appendMessage( - .{ .role = .assistant, .content = a2, .stop_reason = try testing.allocator.dupe(u8, "stop") }, - oaStamp(), - ); - - try testing.expectEqual(@as(usize, 5), mgr.entries.items.len); - - break :blk try testing.allocator.dupe(u8, mgr.getSessionFile()); - }; - defer testing.allocator.free(session_file); - - // Verify the file exists and is well-formed. - { - const bytes = try readWholeFile(testing.allocator, io, session_file); - defer testing.allocator.free(bytes); - // 1 header + 5 entries + trailing \n on each = 6 newlines. - var nl_count: usize = 0; - for (bytes) |b| if (b == '\n') { - nl_count += 1; - }; - try testing.expectEqual(@as(usize, 6), nl_count); - } - - // Resume. - var resumed = try SessionFile.open(testing.allocator, io, session_file); - defer resumed.deinit(); - try testing.expect(resumed.isFlushed()); - try testing.expectEqual(@as(usize, 5), resumed.entries.items.len); - try testing.expectEqualStrings("{\"cwd\":\"/proj/foo\"}", resumed.header.metadata.?); - - // Continue the conversation. - const u_three = try testing.allocator.alloc(StoredContentBlock, 1); - u_three[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "thanks") } }; - _ = try resumed.appendMessage(.{ .role = .user, .content = u_three }, oaStamp()); - try testing.expectEqual(@as(usize, 6), resumed.entries.items.len); -} - -test "SessionFile: assistant message tags the message metadata and the entry leaf id is the assistant entry" { - const io = testing.io; - - var td = try TmpSessionDir.init(testing.allocator); - defer td.deinit(testing.allocator); - const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); - defer testing.allocator.free(sessions); - - var mgr = try SessionFile.init(testing.allocator, io, sessions, null); - defer mgr.deinit(); - - const u_blocks = try testing.allocator.alloc(StoredContentBlock, 1); - u_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "ping") } }; - const user_id = try mgr.appendMessage(.{ .role = .user, .content = u_blocks }, oaStamp()); - - const a_blocks = try testing.allocator.alloc(StoredContentBlock, 1); - a_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "pong") } }; - const asst_id = try mgr.appendMessage(.{ .role = .assistant, .content = a_blocks }, null); - - // Leaf is the assistant entry. - try testing.expectEqualStrings(asst_id, mgr.leaf_id.?); - // Parent of assistant is the user entry. - const assistant_entry = mgr.entries.items[mgr.by_id.get(asst_id).?]; - try testing.expectEqualStrings(user_id, assistant_entry.base().parent_id.?); - // User entry's parent is null (no system). - const user_entry = mgr.entries.items[mgr.by_id.get(user_id).?]; - try testing.expect(user_entry.base().parent_id == null); -} - -/// Test helper: the active provider/model stamp — the last entry carrying a -/// wire stamp, walking leaf→root. Null when no stamped message exists yet. -fn activeStamp(sf: *const SessionFile) ?session_mod.WireStamp { - var i = sf.entries.items.len; - while (i > 0) : (i -= 1) { - switch (sf.entries.items[i - 1]) { - .message => |m| if (m.stamp) |st| return st, - } - } - return null; -} - -test "SessionFile: activeStamp is null before any user message, then tracks the latest user stamp" { - const io = testing.io; - - var td = try TmpSessionDir.init(testing.allocator); - defer td.deinit(testing.allocator); - const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); - defer testing.allocator.free(sessions); - - var mgr = try SessionFile.init(testing.allocator, io, sessions, null); - defer mgr.deinit(); - - // No user messages yet — there is no "active" model on disk yet. - try testing.expect(activeStamp(&mgr) == null); - - // Stamp a user message with anthropic. - const u_blocks = try testing.allocator.alloc(StoredContentBlock, 1); - u_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hi") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u_blocks }, anStamp()); - - { - const am = activeStamp(&mgr).?; - try testing.expectEqual(session_mod.APIStyle.anthropic_messages, am.api_style); - try testing.expectEqualStrings("claude-sonnet-4-20250514", am.model); - } -} - -test "SessionFile: rebuildConversation reconstructs system/user/assistant turn" { - const io = testing.io; - - var td = try TmpSessionDir.init(testing.allocator); - defer td.deinit(testing.allocator); - const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); - defer testing.allocator.free(sessions); - - var mgr = try SessionFile.init(testing.allocator, io, sessions, null); - defer mgr.deinit(); - - const sys = try testing.allocator.alloc(StoredContentBlock, 1); - sys[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "you are helpful") } }; - _ = try mgr.appendMessage(.{ .role = .system, .content = sys }, null); - - const u = try testing.allocator.alloc(StoredContentBlock, 1); - u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hello") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u }, oaStamp()); - - const a = try testing.allocator.alloc(StoredContentBlock, 1); - a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hi!") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, null); - - var conv = try mgr.rebuildConversation(); - defer conv.deinit(); - try testing.expectEqual(@as(usize, 3), conv.messages.items.len); - try testing.expectEqual(conversation_mod.MessageRole.system, conv.messages.items[0].role); - try testing.expectEqualStrings("you are helpful", conv.messages.items[0].content.items[0].System.text.items); - try testing.expectEqual(conversation_mod.MessageRole.user, conv.messages.items[1].role); - try testing.expectEqualStrings("hello", conv.messages.items[1].content.items[0].Text.items); - try testing.expectEqual(conversation_mod.MessageRole.assistant, conv.messages.items[2].role); - try testing.expectEqualStrings("hi!", conv.messages.items[2].content.items[0].Text.items); -} - -test "SessionFile: crash recovery truncates corrupted trailing line" { - const io = testing.io; - - var td = try TmpSessionDir.init(testing.allocator); - defer td.deinit(testing.allocator); - const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); - defer testing.allocator.free(sessions); - - // Build a valid session first. - const session_file: []u8 = blk: { - var mgr = try SessionFile.init(testing.allocator, io, sessions, null); - defer mgr.deinit(); - const u = try testing.allocator.alloc(StoredContentBlock, 1); - u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "ping") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u }, oaStamp()); - const a = try testing.allocator.alloc(StoredContentBlock, 1); - a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "pong") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, null); - break :blk try testing.allocator.dupe(u8, mgr.getSessionFile()); - }; - defer testing.allocator.free(session_file); - - // Corrupt the file: append a partial JSON line at the end. - const garbage = "{\"type\":\"message\",\"id\":\"deadbeef\",\"parent"; - { - const file = try Io.Dir.cwd().openFile(io, session_file, .{ .mode = .write_only }); - defer file.close(io); - const len = try file.length(io); - try file.writePositionalAll(io, garbage, len); - } - // Confirm the file got bigger. - { - const f = try Io.Dir.cwd().openFile(io, session_file, .{ .mode = .read_only }); - defer f.close(io); - const corrupted_len = try f.length(io); - try testing.expect(corrupted_len > garbage.len); - } - - // Now resume — the partial line should be truncated. - var resumed = try SessionFile.open(testing.allocator, io, session_file); - defer resumed.deinit(); - try testing.expectEqual(@as(usize, 2), resumed.entries.items.len); - - // And the file on disk should match. - { - const bytes = try readWholeFile(testing.allocator, io, session_file); - defer testing.allocator.free(bytes); - try testing.expect(!std.mem.endsWith(u8, bytes, "parent")); - // Should end with a newline after the assistant entry. - try testing.expectEqual(@as(u8, '\n'), bytes[bytes.len - 1]); - } -} - -test "listSessions: returns most recent first, with counts" { - const io = testing.io; - - var td = try TmpSessionDir.init(testing.allocator); - defer td.deinit(testing.allocator); - const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); - defer testing.allocator.free(sessions); - - // Create two sessions. - for (0..2) |i| { - var mgr = try SessionFile.init(testing.allocator, io, sessions, null); - defer mgr.deinit(); - const u = try testing.allocator.alloc(StoredContentBlock, 1); - u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "u") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u }, oaStamp()); - const a = try testing.allocator.alloc(StoredContentBlock, 1); - a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, null); - // Small sleep so UUIDv7 timestamps differ. - io.sleep(.fromMilliseconds(2), .real) catch {}; - _ = i; - } - - const infos = try listSessions(testing.allocator, io, sessions, null); - defer { - for (infos) |fi| fi.deinit(testing.allocator); - testing.allocator.free(infos); - } - try testing.expectEqual(@as(usize, 2), infos.len); - try testing.expectEqual(@as(usize, 2), infos[0].message_count); - try testing.expect(std.mem.order(u8, infos[0].modified, infos[1].modified) != .lt); -} - -test "latest: picks most-recently-modified, not newest-created" { - const io = testing.io; - - var td = try TmpSessionDir.init(testing.allocator); - defer td.deinit(testing.allocator); - const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); - defer testing.allocator.free(sessions); - - var catalog = try FileSystemJSONLStore.init(testing.allocator, io, sessions); - defer catalog.deinit(); - const st = catalog.store(); - - // Before any sessions exist → null. - try testing.expect((try st.latest()) == null); - - // Create session A, then session B (newer id), then append to A again - // so A is the most recently *modified*. - var first_id: ?[]u8 = null; - defer if (first_id) |s| testing.allocator.free(s); - - for (0..2) |i| { - var mgr = try SessionFile.init(testing.allocator, io, sessions, null); - defer mgr.deinit(); - const u = try testing.allocator.alloc(StoredContentBlock, 1); - u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "u") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u }, oaStamp()); - const a = try testing.allocator.alloc(StoredContentBlock, 1); - a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, oaStamp()); - if (i == 0) first_id = try testing.allocator.dupe(u8, mgr.header.id); - io.sleep(.fromMilliseconds(2), .real) catch {}; - } - { - const path = try resolveSessionId(testing.allocator, io, sessions, first_id.?); - defer testing.allocator.free(path); - var mgr = try SessionFile.open(testing.allocator, io, path); - defer mgr.deinit(); - const a = try testing.allocator.alloc(StoredContentBlock, 1); - a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, oaStamp()); - } - - var latest = (try st.latest()).?; - defer latest.info.deinit(testing.allocator); - try testing.expectEqualStrings(first_id.?, latest.info.id); -} - -test "SessionFile: tool-use round-trip — assistant w/ ToolUse, user w/ ToolResult, assistant" { - const io = testing.io; - var td = try TmpSessionDir.init(testing.allocator); - defer td.deinit(testing.allocator); - const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); - defer testing.allocator.free(sessions); - - const session_file: []u8 = blk: { - var mgr = try SessionFile.init(testing.allocator, io, sessions, null); - defer mgr.deinit(); - - const u = try testing.allocator.alloc(StoredContentBlock, 1); - u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "list files") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u }, oaStamp()); - - // Assistant emits a ToolUse. - const am1 = try testing.allocator.alloc(StoredContentBlock, 2); - am1[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "checking...") } }; - am1[1] = .{ .tool_use = .{ - .id = try testing.allocator.dupe(u8, "tool_abc"), - .name = try testing.allocator.dupe(u8, "bash"), - .input = try testing.allocator.dupe(u8, "{\"command\":\"ls\"}"), - } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = am1 }, null); - - // Tool-result user message. - const tr = try testing.allocator.alloc(StoredContentBlock, 1); - const trp = try testing.allocator.alloc(session_mod.StoredResultPart, 1); - trp[0] = .{ .text = try testing.allocator.dupe(u8, "a.txt\nb.txt") }; - tr[0] = .{ .tool_result = .{ - .tool_use_id = try testing.allocator.dupe(u8, "tool_abc"), - .parts = trp, - } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = tr }, oaStamp()); - - // Final assistant reply. - const a2 = try testing.allocator.alloc(StoredContentBlock, 1); - a2[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "two files: a.txt and b.txt") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a2 }, null); - - break :blk try testing.allocator.dupe(u8, mgr.getSessionFile()); - }; - defer testing.allocator.free(session_file); - - // Reopen and verify content blocks survive. - var resumed = try SessionFile.open(testing.allocator, io, session_file); - defer resumed.deinit(); - const entries = resumed.entries.items; - try testing.expectEqual(@as(usize, 4), entries.len); - - // [1] = assistant with ToolUse - try testing.expectEqual(StoredMessageRole.assistant, entries[1].message.message.role); - try testing.expectEqual(@as(usize, 2), entries[1].message.message.content.len); - try testing.expect(entries[1].message.message.content[1] == .tool_use); - try testing.expectEqualStrings("bash", entries[1].message.message.content[1].tool_use.name); - try testing.expectEqualStrings("{\"command\":\"ls\"}", entries[1].message.message.content[1].tool_use.input); - - // [2] = user with ToolResult, stamped with wire identity. - try testing.expectEqual(StoredMessageRole.user, entries[2].message.message.role); - try testing.expectEqual(session_mod.APIStyle.openai_chat, entries[2].message.stamp.?.api_style); - try testing.expect(entries[2].message.message.content[0] == .tool_result); - try testing.expectEqualStrings("tool_abc", entries[2].message.message.content[0].tool_result.tool_use_id); - try testing.expectEqualStrings("a.txt\nb.txt", entries[2].message.message.content[0].tool_result.parts[0].text); - - // Conversation rebuild yields the same shape. - var conv = try resumed.rebuildConversation(); - defer conv.deinit(); - try testing.expectEqual(@as(usize, 4), conv.messages.items.len); - try testing.expect(conv.messages.items[1].content.items[1] == .ToolUse); - try testing.expect(conv.messages.items[2].content.items[0] == .ToolResult); -} - -test "SessionFile: linear chain — each entry's parent_id is the previous entry's id" { - const io = testing.io; - var td = try TmpSessionDir.init(testing.allocator); - defer td.deinit(testing.allocator); - const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); - defer testing.allocator.free(sessions); - - var mgr = try SessionFile.init(testing.allocator, io, sessions, null); - defer mgr.deinit(); - - // Three rounds: sys, user, asst, user, asst. - const sys = try testing.allocator.alloc(StoredContentBlock, 1); - sys[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "sys") } }; - _ = try mgr.appendMessage(.{ .role = .system, .content = sys }, null); - - const u_one = try testing.allocator.alloc(StoredContentBlock, 1); - u_one[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "u1") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u_one }, oaStamp()); - - const a_one = try testing.allocator.alloc(StoredContentBlock, 1); - a_one[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a1") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a_one }, null); - - const u_two = try testing.allocator.alloc(StoredContentBlock, 1); - u_two[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "u2") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u_two }, oaStamp()); - - const a_two = try testing.allocator.alloc(StoredContentBlock, 1); - a_two[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a2") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a_two }, null); - - const entries = mgr.entries.items; - try testing.expectEqual(@as(usize, 5), entries.len); - try testing.expect(entries[0].base().parent_id == null); - for (entries[1..], 1..) |e, i| { - try testing.expectEqualStrings(entries[i - 1].base().id, e.base().parent_id.?); - } -} - -test "resolveSessionId: unique prefix → match, ambiguous → error" { - const io = testing.io; - - var td = try TmpSessionDir.init(testing.allocator); - defer td.deinit(testing.allocator); - const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); - defer testing.allocator.free(sessions); - - // Create one session. - var mgr = try SessionFile.init(testing.allocator, io, sessions, null); - defer mgr.deinit(); - const u = try testing.allocator.alloc(StoredContentBlock, 1); - u[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "u") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = u }, oaStamp()); - const a = try testing.allocator.alloc(StoredContentBlock, 1); - a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, null); - - const id = mgr.header.id; - const prefix = id[0..8]; - - const resolved = try resolveSessionId(testing.allocator, io, sessions, prefix); - defer testing.allocator.free(resolved); - try testing.expectEqualStrings(mgr.getSessionFile(), resolved); - - try testing.expectError(error.SessionNotFound, resolveSessionId(testing.allocator, io, sessions, "ffffffff")); -} - -test "compaction summary round-trips through persist + resume + rebuild" { - const io = testing.io; - - var td = try TmpSessionDir.init(testing.allocator); - defer td.deinit(testing.allocator); - const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); - defer testing.allocator.free(sessions); - - const session_file: []u8 = blk: { - var mgr = try SessionFile.init(testing.allocator, io, sessions, null); - defer mgr.deinit(); - - // System + an old turn that will be superseded. - const sys = try testing.allocator.alloc(StoredContentBlock, 1); - sys[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "you are helpful") } }; - _ = try mgr.appendMessage(.{ .role = .system, .content = sys }, null); - - const uo = try testing.allocator.alloc(StoredContentBlock, 1); - uo[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "old q") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = uo }, oaStamp()); - const ao = try testing.allocator.alloc(StoredContentBlock, 1); - ao[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "old a") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = ao }, null); - - // Compaction: summary message + duplicated kept suffix. - const cs = try testing.allocator.alloc(StoredContentBlock, 1); - cs[0] = .{ .compaction_summary = .{ .text = try testing.allocator.dupe(u8, "SUMMARY") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = cs }, oaStamp()); - - const ur = try testing.allocator.alloc(StoredContentBlock, 1); - ur[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "recent q") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = ur }, oaStamp()); - - break :blk try testing.allocator.dupe(u8, mgr.getSessionFile()); - }; - defer testing.allocator.free(session_file); - - var resumed = try SessionFile.open(testing.allocator, io, session_file); - defer resumed.deinit(); - var conv = try resumed.rebuildConversation(); - defer conv.deinit(); - - // The compaction summary block survived as a CompactionSummary. - const anchor = conversation_mod.latestCompactionIndex(conv.messages.items).?; - try testing.expectEqualStrings( - "SUMMARY", - conv.messages.items[anchor].content.items[0].CompactionSummary.text.items, - ); - - // The active window is [summary, recent q]. - const window = conversation_mod.activeMessageWindow(conv.messages.items); - try testing.expectEqual(@as(usize, 2), window.len); - try testing.expectEqualStrings("recent q", window[1].content.items[0].Text.items); - - // System prompt survives (derived independently). - var sys_blocks = try conversation_mod.effectiveSystemBlocks(testing.allocator, conv.messages.items); - defer sys_blocks.deinit(testing.allocator); - try testing.expectEqual(@as(usize, 1), sys_blocks.items.len); - try testing.expectEqualStrings("you are helpful", sys_blocks.items[0]); -} - -test "loadConversation: trailing user prompt is split out as dangling, excluded from conversation" { - const io = testing.io; - - var td = try TmpSessionDir.init(testing.allocator); - defer td.deinit(testing.allocator); - const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); - defer testing.allocator.free(sessions); - - var mgr = try SessionFile.init(testing.allocator, io, sessions, null); - defer mgr.deinit(); - - // A completed user/assistant round, then a trailing user prompt with no - // following assistant. The dangling-prompt recovery feature was dropped - // in R2: the trailing user message simply round-trips into the rebuilt - // conversation like any other. - const um1 = try testing.allocator.alloc(StoredContentBlock, 1); - um1[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hi") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = um1 }, oaStamp()); - const am1 = try testing.allocator.alloc(StoredContentBlock, 1); - am1[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hello") } }; - _ = try mgr.appendMessage(.{ .role = .assistant, .content = am1 }, null); - const um2 = try testing.allocator.alloc(StoredContentBlock, 1); - um2[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "what's 2+2?") } }; - _ = try mgr.appendMessage(.{ .role = .user, .content = um2 }, oaStamp()); - - var conv = try mgr.rebuildConversation(); - defer conv.deinit(); - // All three messages are present (dangling recovery dropped). - try testing.expectEqual(@as(usize, 3), conv.messages.items.len); - try testing.expectEqual(conversation_mod.MessageRole.user, conv.messages.items[2].role); -} - -test "FileSystemJSONLStore catalog: create → append → load round-trips" { - const io = testing.io; - - var td = try TmpSessionDir.init(testing.allocator); - defer td.deinit(testing.allocator); - const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); - defer testing.allocator.free(sessions); - - var catalog = try FileSystemJSONLStore.init(testing.allocator, io, sessions); - defer catalog.deinit(); - const store = catalog.store(); - - var sess = store.create(); - defer sess.info.deinit(testing.allocator); - - // Build a user + assistant PersistentMessage batch (borrows in-memory - // messages owned here). - var conv = conversation_mod.Conversation.init(testing.allocator); - defer conv.deinit(); - try addUserText(&conv, "ping"); - try conv.addAssistantMessage(&.{}, null); - - const id: session_store_mod.WireIdentity = .{ .api_style = .openai_chat, .base_url = "u", .model = "m" }; - var batch = [_]session_store_mod.PersistentMessage{ - .{ .message = conv.messages.items[0], .identity = id }, - .{ .message = conv.messages.items[1], .identity = id }, - }; - try sess.append(&batch); - - // The session's last-used api_style updated after append. - try testing.expectEqual(session_store_mod.APIStyle.openai_chat, sess.info.api_style); - - // Load it back by id. - var loaded = (try store.load(sess.info.id)).?; - defer loaded.deinit(); - try testing.expectEqual(@as(usize, 2), loaded.messages.items.len); - try testing.expectEqual(conversation_mod.MessageRole.user, loaded.messages.items[0].role); -} - -test "FileSystemJSONLStore load restores thinking signature origin from message 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 catalog = try FileSystemJSONLStore.init(testing.allocator, io, sessions); - defer catalog.deinit(); - const store = catalog.store(); - - var sess = store.create(); - defer sess.info.deinit(testing.allocator); - - var conv = conversation_mod.Conversation.init(testing.allocator); - defer conv.deinit(); - try addUserText(&conv, "ping"); - try conv.addAssistantMessage(&.{ - .{ .Thinking = .{ - .text = try conversation_mod.textualBlockFromSlice(testing.allocator, "thinking..."), - .signature = try testing.allocator.dupe(u8, "sig123"), - } }, - .{ .Text = try conversation_mod.textualBlockFromSlice(testing.allocator, "pong") }, - }, null); - - const id: session_store_mod.WireIdentity = .{ - .api_style = .anthropic_messages, - .base_url = "https://api.anthropic.com", - .model = "claude-sonnet-4-20250514", - }; - var batch = [_]session_store_mod.PersistentMessage{ - .{ .message = conv.messages.items[0], .identity = id }, - .{ .message = conv.messages.items[1], .identity = id }, - }; - try sess.append(&batch); - - var loaded = (try store.load(sess.info.id)).?; - defer loaded.deinit(); - const thinking = loaded.messages.items[1].content.items[0].Thinking; - try testing.expect(thinking.signature_origin != null); - try testing.expect(thinking.signature_origin.?.matches( - .anthropic_messages, - "https://api.anthropic.com", - "claude-sonnet-4-20250514", - )); -} - -test "persistTurn: a message's own identity overrides the uniform persist identity" { - // Regression for the compaction re-stamping bug: a kept-verbatim turn - // produced by model A must persist with A's wire identity even when the - // turn is persisted under model B (the compaction model). persistTurn - // stamps the uniform identity only on messages that don't already carry - // one; a message with its own identity keeps it. - const io = testing.io; - - var td = try TmpSessionDir.init(testing.allocator); - defer td.deinit(testing.allocator); - const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" }); - defer testing.allocator.free(sessions); - - var catalog = try FileSystemJSONLStore.init(testing.allocator, io, sessions); - defer catalog.deinit(); - const store = catalog.store(); - var sess = store.create(); - defer sess.info.deinit(testing.allocator); - - var conv = conversation_mod.Conversation.init(testing.allocator); - defer conv.deinit(); - // User turn carries no identity (will be stamped with the persist id). - try addUserText(&conv, "ping"); - // Assistant turn was produced by model A — stamp its identity directly, - // as a live turn / a reloaded turn would have. - try conv.addAssistantMessage(&.{ - .{ .Thinking = .{ - .text = try conversation_mod.textualBlockFromSlice(testing.allocator, "thinking..."), - .signature = try testing.allocator.dupe(u8, "sig123"), - } }, - .{ .Text = try conversation_mod.textualBlockFromSlice(testing.allocator, "pong") }, - }, null); - conv.messages.items[1].identity = try conversation_mod.dupeWireIdentity(testing.allocator, .{ - .api_style = .anthropic_messages, - .base_url = "https://api.anthropic.com", - .model = "claude-sonnet-4-20250514", - }); - - // Persist under model B (a different, "compaction" identity). - const persist_id: session_store_mod.WireIdentity = .{ - .api_style = .openai_chat, - .base_url = "https://api.openai.com/v1", - .model = "gpt-4o", - }; - try turn_persist.persistTurn(testing.allocator, &sess, &conv, 0, persist_id, &.{}); - - // Reload: the user turn took model B's identity; the assistant turn kept - // model A's — so its thinking signature replays to A, not B. - var loaded = (try store.load(sess.info.id)).?; - defer loaded.deinit(); - - const asst_origin = loaded.messages.items[1].content.items[0].Thinking.signature_origin.?; - try testing.expect(asst_origin.matches(.anthropic_messages, "https://api.anthropic.com", "claude-sonnet-4-20250514")); - try testing.expect(!asst_origin.matches(.openai_chat, "https://api.openai.com/v1", "gpt-4o")); - - const user_id = loaded.messages.items[0].identity.?; - try testing.expectEqual(session_store_mod.APIStyle.openai_chat, user_id.api_style); - try testing.expectEqualStrings("gpt-4o", user_id.model); -} |