1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
//! Turn → session-log persistence: map in-memory `Conversation` messages
//! to rich `PersistentMessage` write records and append them through a
//! `Session` handle.
//!
//! This logic lives in `libpanto` and is called by the `Agent` itself, so
//! every embedder gets persistence for free. The functions here are
//! stateless helpers over a `Session`; the agent owns the store and the
//! conversation.
//!
//! The write record is **maximalist** (full wire identity + usage + the
//! entire current conversation + the offered tool set). The library offers
//! it all on every append; each store keeps what it wants. `PersistentMessage`
//! borrows the in-memory `Message` directly — no disk conversion happens
//! here; the store does its own serialization.
//!
//! Per-message usage is read directly off `Message.usage`.
const std = @import("std");
const Allocator = std.mem.Allocator;
const conversation = @import("conversation.zig");
const session_store = @import("session_store.zig");
const PersistentMessage = session_store.PersistentMessage;
const WireIdentity = session_store.WireIdentity;
const ToolDecl = session_store.ToolDecl;
const Session = session_store.Session;
/// Persist every conversation message at index `>= start_index` through
/// `session` as a single atomic batch.
///
/// Each `PersistentMessage` borrows the in-memory `Message`, the full
/// current conversation, and `tools` (the offered tool set) — all owned by
/// the caller and valid for the duration of the call. The wire `identity`
/// is stamped on every message; the store decides per-role what to keep
/// (the FS store drops the stamp on system entries).
///
/// An assistant message carrying a ToolUse with no following matching
/// ToolResult is skipped (a dangling tool call from an interrupted turn);
/// persisting it would make the log un-replayable.
pub fn persistTurn(
alloc: Allocator,
session: *Session,
conv: *conversation.Conversation,
start_index: usize,
identity: WireIdentity,
tools: []const ToolDecl,
) !void {
return persistRange(alloc, session, conv, start_index, conv.messages.items.len, identity, tools);
}
/// Persist conversation messages in `[start_index, end_index)`. Like
/// `persistTurn` but with an explicit upper bound, so an incremental flush can
/// exclude a trailing not-yet-coherent message (a dangling tool call awaiting
/// its results) while still committing everything before it.
pub fn persistRange(
alloc: Allocator,
session: *Session,
conv: *conversation.Conversation,
start_index: usize,
end_index: usize,
identity: WireIdentity,
tools: []const ToolDecl,
) !void {
var batch: std.ArrayList(PersistentMessage) = .empty;
defer batch.deinit(alloc);
const all_messages = conv.messages.items;
var i = start_index;
while (i < end_index) : (i += 1) {
const msg = &all_messages[i];
if (msg.role == .assistant and hasToolUseWithoutFollowingResults(conv, i)) {
continue;
}
// Stamp the producing identity once, on first persist. A message that
// already carries one (e.g. a kept-verbatim turn carried through
// compaction, or one rebuilt from disk) keeps it — only the messages
// freshly produced under `identity` are stamped now. This is what
// makes the stamp survive a later compaction onto a different model.
if (msg.identity == null) {
msg.identity = try conversation.dupeWireIdentity(alloc, identity);
}
try batch.append(alloc, .{
.message = msg.*,
.usage = msg.usage,
.identity = msg.identity.?,
.conversation = all_messages,
.tools_available = tools,
});
}
if (batch.items.len == 0) return;
try session.append(batch.items);
}
/// Persist a compaction result. The agent rewrote the conversation to
/// `[system..., summary, kept-suffix...]`; persist everything from the
/// latest compaction summary onward as fresh entries.
pub fn persistCompaction(
alloc: Allocator,
session: *Session,
conv: *conversation.Conversation,
identity: WireIdentity,
tools: []const ToolDecl,
) !void {
const start = conversation.latestCompactionIndex(conv.messages.items) orelse return;
try persistTurn(alloc, session, conv, start, identity, tools);
}
/// True when the assistant message at `index` contains a ToolUse block
/// with no matching ToolResult in the immediately following user message
/// — i.e. a dangling tool call (e.g. a turn cut short by an error).
pub fn hasToolUseWithoutFollowingResults(
conv: *const conversation.Conversation,
index: usize,
) bool {
const msg = conv.messages.items[index];
for (msg.content.items) |block| {
if (block != .ToolUse) continue;
if (index + 1 >= conv.messages.items.len) return true;
const next = conv.messages.items[index + 1];
if (next.role != .user) return true;
var found = false;
for (next.content.items) |next_block| {
if (next_block == .ToolResult and std.mem.eql(u8, next_block.ToolResult.tool_use_id, block.ToolUse.id)) {
found = true;
break;
}
}
if (!found) return true;
}
return false;
}
|
