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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
//! Turn → session-log persistence: map in-memory `Conversation` messages
//! to neutral `DiskMessage`s and append them to a `SessionStore`.
//!
//! This logic used to live in the `panto` CLI (`src/session_persist.zig`),
//! driven externally around `runStep`. It now lives in `libpanto` and is
//! called by the `Agent` itself, so every embedder gets persistence for
//! free. The functions here are stateless helpers over a `SessionStore`;
//! the agent owns the store and the conversation.
//!
//! Per-message usage is read directly off `Message.usage` (canonical: both
//! providers stamp it via `addAssistantMessageWithUsage`). There is no
//! separate usage list to thread through.
const std = @import("std");
const Allocator = std.mem.Allocator;
const conversation = @import("conversation.zig");
const session = @import("session.zig");
const session_store = @import("session_store.zig");
const DiskMessage = session_store.DiskMessage;
const DiskContentBlock = session_store.DiskContentBlock;
const SessionStore = session_store.SessionStore;
/// Persist every conversation message at index `>= start_index` to `store`
/// as a single atomic batch.
///
/// Mapping:
/// - assistant → assistant entry with provider/model/stop_reason and the
/// message's own `usage`.
/// - user (incl. ToolResult-only messages) → user entry stamped with
/// provider/model.
/// - system → system entry, verbatim, unstamped (mid-turn system
/// messages aren't produced by pantograph today, but persist harmless).
///
/// An assistant message carrying a ToolUse with no following matching
/// ToolResult is skipped (a dangling tool call from an interrupted turn);
/// persisting it would make the log un-replayable.
///
/// `stop_reason` is `"stop"` for now; the real wire value needs provider
/// plumbing (separate future work).
pub fn persistTurn(
alloc: Allocator,
store: SessionStore,
conv: *const conversation.Conversation,
start_index: usize,
provider: []const u8,
model: []const u8,
) !void {
// Ownership note: the messages we build here are handed to
// `store.appendMessages`, which *consumes* them. We therefore never
// free entries in `batch_messages` ourselves — only the parallel
// metadata arrays and the per-block transient on the error path before
// a message is appended to the batch. This mirrors the original CLI
// `persistTurn` exactly (it transferred ownership to the manager and
// never freed the DiskMessages).
var batch_messages: std.ArrayList(DiskMessage) = .empty;
defer batch_messages.deinit(alloc);
var batch_providers: std.ArrayList(?[]const u8) = .empty;
defer batch_providers.deinit(alloc);
var batch_models: std.ArrayList(?[]const u8) = .empty;
defer batch_models.deinit(alloc);
var i = start_index;
while (i < conv.messages.items.len) : (i += 1) {
const msg = conv.messages.items[i];
const blocks = try alloc.alloc(DiskContentBlock, msg.content.items.len);
var allocated: usize = 0;
errdefer {
for (blocks[0..allocated]) |b| b.deinit(alloc);
alloc.free(blocks);
}
for (msg.content.items) |block| {
blocks[allocated] = try session.contentBlockToDisk(alloc, block);
allocated += 1;
}
if (msg.role == .assistant and hasToolUseWithoutFollowingResults(conv, i)) {
for (blocks[0..allocated]) |b| b.deinit(alloc);
alloc.free(blocks);
continue;
}
switch (msg.role) {
.system => {
// The System block carries the append/replace mode; it
// rides on `DiskMessage.mode`, not on the disk block. Derive
// it from the in-memory message so reconciliation on replay
// reconstructs the same effective prompt.
try batch_messages.append(alloc, .{
.role = .system,
.mode = systemModeOf(msg),
.content = blocks,
});
try batch_providers.append(alloc, null);
try batch_models.append(alloc, null);
},
.user => {
try batch_messages.append(alloc, .{ .role = .user, .content = blocks });
try batch_providers.append(alloc, provider);
try batch_models.append(alloc, model);
},
.assistant => {
try batch_messages.append(alloc, .{
.role = .assistant,
.content = blocks,
.provider = try alloc.dupe(u8, provider),
.model = try alloc.dupe(u8, model),
.stop_reason = try alloc.dupe(u8, "stop"),
.usage = msg.usage,
});
try batch_providers.append(alloc, null);
try batch_models.append(alloc, null);
},
}
}
if (batch_messages.items.len == 0) return;
try store.appendMessages(batch_messages.items, batch_providers.items, batch_models.items);
}
/// Persist a compaction result. The agent rewrote the conversation to
/// `[system..., summary, kept-suffix...]`; persist everything from the
/// latest compaction summary onward as fresh entries. On replay the latest
/// summary resets effective context, so the duplicated suffix is what
/// survives. Usage isn't re-derived for the restated suffix (those turns
/// were already sized when first logged).
pub fn persistCompaction(
alloc: Allocator,
store: SessionStore,
conv: *const conversation.Conversation,
provider: []const u8,
model: []const u8,
) !void {
const start = conversation.latestCompactionIndex(conv.messages.items) orelse return;
try persistTurn(alloc, store, conv, start, provider, model);
}
/// Derive the disk system-mode from a system message's blocks. A `.System`
/// block carries the mode; a `replace` anywhere makes the message a
/// replace. Defaults to `append`.
fn systemModeOf(msg: conversation.Message) session.DiskSystemMode {
for (msg.content.items) |block| {
if (block == .System and block.System.mode == .replace) return .replace;
}
return .append;
}
/// 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;
}
|
