diff options
Diffstat (limited to 'libpanto/src/file_system_jsonl_store.zig')
| -rw-r--r-- | libpanto/src/file_system_jsonl_store.zig | 205 |
1 files changed, 44 insertions, 161 deletions
diff --git a/libpanto/src/file_system_jsonl_store.zig b/libpanto/src/file_system_jsonl_store.zig index 50587f0..5b7f450 100644 --- a/libpanto/src/file_system_jsonl_store.zig +++ b/libpanto/src/file_system_jsonl_store.zig @@ -372,18 +372,15 @@ pub const SessionFile = struct { try truncateFileTo(io, path_copy, valid_bytes); } - var migrated_header = header; - var did_migrate = migrate(allocator, &migrated_header, &entries); + // 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)) { - did_migrate = true; - } - // We didn't reassign by_id during migration; rebuild if needed. - if (did_migrate) { by_id.clearRetainingCapacity(); for (entries.items, 0..) |e, i| { try by_id.put(e.base().id, i); } - try rewriteFile(allocator, io, path_copy, migrated_header, entries.items); + try rewriteFile(allocator, io, path_copy, header, entries.items); } const leaf_id: ?[]const u8 = if (entries.items.len > 0) @@ -400,7 +397,7 @@ pub const SessionFile = struct { .io = io, .session_dir = dir, .session_file = path_copy, - .header = migrated_header, + .header = header, .entries = entries, .by_id = by_id, .leaf_id = leaf_id, @@ -420,131 +417,30 @@ pub const SessionFile = struct { // ---------- Accessors ---------- - pub fn getMetadata(self: *const SessionFile) ?[]const u8 { - return self.header.metadata; - } - - pub fn getSessionId(self: *const SessionFile) []const u8 { - return self.header.id; - } - pub fn getSessionFile(self: *const SessionFile) []const u8 { return self.session_file; } - pub fn getSessionDir(self: *const SessionFile) []const u8 { - return self.session_dir; - } - - pub fn getLeafId(self: *const SessionFile) ?[]const u8 { - return self.leaf_id; - } - - pub fn getEntry(self: *const SessionFile, id: []const u8) ?*const SessionEntry { - const idx = self.by_id.get(id) orelse return null; - return &self.entries.items[idx]; - } - - pub fn getEntries(self: *const SessionFile) []const SessionEntry { - return self.entries.items; - } - pub fn isFlushed(self: *const SessionFile) bool { return self.flushed; } - // ---------- Active model resolution ---------- - - /// Determine the active provider/model by walking entries leaf→root - /// and finding the last user-message entry with provider/model - /// stamped. Returns null only when no user message has been appended - /// yet, which is only reachable on a freshly-`init`'d session before - /// the first user prompt (and therefore before any disk flush). - /// - /// Returns a borrowed stamp owned by the manager; do not free. - pub fn activeStamp(self: *const SessionFile) ?session_mod.WireStamp { - var i = self.entries.items.len; - while (i > 0) : (i -= 1) { - const e = self.entries.items[i - 1]; - switch (e) { - .message => |m| { - if (m.stamp) |st| return st; - }, - } - } - return null; - } - // ---------- Appending ---------- - /// Append a message entry. `msg` is consumed (ownership transferred) - /// regardless of success — on error, the message is deinit'd before - /// the error is returned. - /// - /// If `flushed`: writes the new line immediately. - /// If not flushed and `msg.role == .assistant`: writes the header + - /// all buffered entries + the new entry, then sets `flushed`. - /// Otherwise: buffers in memory only. + /// 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 here into the entry. + // messages. Borrowed; duplicated into the entry. stamp: ?session_mod.WireStamp, ) ![]const u8 { - // Build the entry up-front, taking ownership of the inputs. - var msg_local = msg; - errdefer msg_local.deinit(self.allocator); - - const id_buf = try self.newEntryId(); - errdefer self.allocator.free(id_buf); - - const timestamp = try isoTimestamp(self.allocator, self.io); - errdefer self.allocator.free(timestamp); - - const parent_id_copy: ?[]const u8 = if (self.leaf_id) |l| try self.allocator.dupe(u8, l) else null; - errdefer if (parent_id_copy) |p| self.allocator.free(p); - - const stamp_copy: ?session_mod.WireStamp = if (stamp) |st| try st.dupe(self.allocator) else null; - errdefer if (stamp_copy) |st| st.deinit(self.allocator); - - const entry: SessionEntry = .{ .message = .{ - .base = .{ .id = id_buf, .parent_id = parent_id_copy, .timestamp = timestamp }, - .stamp = stamp_copy, - .message = msg_local, - } }; - - // The entry now owns msg_local + id_buf + timestamp + parent_id_copy + - // provider/model copies. Cancel the errdefers individually. - // (Zig's errdefer behavior: they only run on error returns; pushing the - // entry into entries.items before any further fallible step means an - // error in by_id.put() would double-free. Instead, do the put first - // against a not-yet-stored id.) - - const idx = self.entries.items.len; - - // Ensure capacity before touching anything. - try self.entries.ensureUnusedCapacity(self.allocator, 1); - try self.by_id.ensureUnusedCapacity(1); - - // Persist BEFORE inserting into the in-memory structures, so that on - // I/O failure we don't have a dangling in-memory entry the caller - // thinks was saved. (Failure leaves the file unchanged for an - // unflushed session, and unchanged-except-for-EOF for a flushed one.) - const is_assistant = entry.message.message.role == .assistant; - if (self.flushed) { - try self.persistEntry(entry); - } else if (is_assistant) { - try self.flushBuffered(entry); - } - // If not flushed and not assistant: nothing to do; the entry will be - // flushed alongside the eventual first assistant entry. - - // Now insert into in-memory structures. All allocations are kept. - self.entries.appendAssumeCapacity(entry); - self.by_id.putAssumeCapacity(entry.base().id, idx); - self.leaf_id = entry.base().id; - return entry.base().id; + 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 @@ -616,16 +512,6 @@ pub const SessionFile = struct { self.written_bytes = offset; } - fn flushBuffered(self: *SessionFile, final_entry: SessionEntry) !void { - try self.flushBufferedMany(&.{final_entry}); - } - - /// Append a single line for `entry` to the open session file. Caller - /// must have already verified `flushed`. - fn persistEntry(self: *SessionFile, entry: SessionEntry) !void { - try self.persistEntries(&.{entry}); - } - pub fn appendMessagesAtomic( self: *SessionFile, messages: []StoredMessage, @@ -801,24 +687,9 @@ fn appendMessageToConv( } // ============================================================================= -// Migration +// Entry repair on load // ============================================================================= -/// Future format migrations land here. Returns true if anything changed -/// (which triggers a one-time file rewrite). -fn migrate( - allocator: Allocator, - header: *SessionHeader, - entries: *std.ArrayList(SessionEntry), -) bool { - _ = allocator; - _ = entries; - if (header.version >= CURRENT_VERSION) return false; - // No earlier versions exist yet. When v2 lands, transform v1 entries - // here and bump `header.version`. - return false; -} - fn elideDanglingToolUses(allocator: Allocator, entries: *std.ArrayList(SessionEntry)) bool { var needed: std.StringHashMap(void) = .init(allocator); defer needed.deinit(); @@ -1565,25 +1436,25 @@ test "SessionFile: full flow — buffer, flush on assistant, append, resume" { ); defer mgr.deinit(); - // System message — non-assistant, should NOT trigger flush. + // First append flushes the header + entry to disk immediately. 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()); + try testing.expect(mgr.isFlushed()); - // User message — also doesn't flush. + // User message — appended to the already-flushed file. 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()); + try testing.expect(mgr.isFlushed()); - // Assistant message — triggers flush. + // Assistant message. const a_blocks = try testing.allocator.alloc(StoredContentBlock, 1); a_blocks[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "hello") } }; _ = try mgr.appendMessage( @@ -1608,7 +1479,7 @@ test "SessionFile: full flow — buffer, flush on assistant, append, resume" { oaStamp(), ); - try testing.expectEqual(@as(usize, 5), mgr.getEntries().len); + try testing.expectEqual(@as(usize, 5), mgr.entries.items.len); break :blk try testing.allocator.dupe(u8, mgr.getSessionFile()); }; @@ -1630,14 +1501,14 @@ test "SessionFile: full flow — buffer, flush on assistant, append, resume" { var resumed = try SessionFile.open(testing.allocator, io, session_file); defer resumed.deinit(); try testing.expect(resumed.isFlushed()); - try testing.expectEqual(@as(usize, 5), resumed.getEntries().len); - try testing.expectEqualStrings("{\"cwd\":\"/proj/foo\"}", resumed.getMetadata().?); + 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.getEntries().len); + 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" { @@ -1660,15 +1531,27 @@ test "SessionFile: assistant message tags the message metadata and the entry lea const asst_id = try mgr.appendMessage(.{ .role = .assistant, .content = a_blocks }, null); // Leaf is the assistant entry. - try testing.expectEqualStrings(asst_id, mgr.getLeafId().?); + try testing.expectEqualStrings(asst_id, mgr.leaf_id.?); // Parent of assistant is the user entry. - const assistant_entry = mgr.getEntry(asst_id).?; + 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.getEntry(user_id).?; + 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; @@ -1681,7 +1564,7 @@ test "SessionFile: activeStamp is null before any user message, then tracks the defer mgr.deinit(); // No user messages yet — there is no "active" model on disk yet. - try testing.expect(mgr.activeStamp() == null); + try testing.expect(activeStamp(&mgr) == null); // Stamp a user message with anthropic. const u_blocks = try testing.allocator.alloc(StoredContentBlock, 1); @@ -1689,7 +1572,7 @@ test "SessionFile: activeStamp is null before any user message, then tracks the _ = try mgr.appendMessage(.{ .role = .user, .content = u_blocks }, anStamp()); { - const am = mgr.activeStamp().?; + const am = activeStamp(&mgr).?; try testing.expectEqual(session_mod.APIStyle.anthropic_messages, am.api_style); try testing.expectEqualStrings("claude-sonnet-4-20250514", am.model); } @@ -1770,7 +1653,7 @@ test "SessionFile: crash recovery truncates corrupted trailing line" { // Now resume — the partial line should be truncated. var resumed = try SessionFile.open(testing.allocator, io, session_file); defer resumed.deinit(); - try testing.expectEqual(@as(usize, 2), resumed.getEntries().len); + try testing.expectEqual(@as(usize, 2), resumed.entries.items.len); // And the file on disk should match. { @@ -1895,7 +1778,7 @@ test "SessionFile: tool-use round-trip — assistant w/ ToolUse, user w/ ToolRes // Reopen and verify content blocks survive. var resumed = try SessionFile.open(testing.allocator, io, session_file); defer resumed.deinit(); - const entries = resumed.getEntries(); + const entries = resumed.entries.items; try testing.expectEqual(@as(usize, 4), entries.len); // [1] = assistant with ToolUse @@ -1951,7 +1834,7 @@ test "SessionFile: linear chain — each entry's parent_id is the previous entry a_two[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a2") } }; _ = try mgr.appendMessage(.{ .role = .assistant, .content = a_two }, null); - const entries = mgr.getEntries(); + 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| { @@ -1977,7 +1860,7 @@ test "resolveSessionId: unique prefix → match, ambiguous → error" { a[0] = .{ .text = .{ .text = try testing.allocator.dupe(u8, "a") } }; _ = try mgr.appendMessage(.{ .role = .assistant, .content = a }, null); - const id = mgr.getSessionId(); + const id = mgr.header.id; const prefix = id[0..8]; const resolved = try resolveSessionId(testing.allocator, io, sessions, prefix); |