From eb4179f9958deeae408a0a06b1f8ae6437e089db Mon Sep 17 00:00:00 2001 From: t Date: Sun, 7 Jun 2026 15:19:57 -0600 Subject: Sync Agent/Stream internals to public names; alias Stream Push the API-shaping we'd done in the public facade down into the internal types, collapsing the wrappers: - Stream: rename phase->state, Phase->State; underscore-prefix the internal fields (_agent/_queue/_response/_start/_persisted/_pending_error). state is the one intended-public field. public.zig now aliases Stream; run returns *Stream. - Agent compaction: pure transform compact()->private _compactInPlace; compactAndPersist()->compact() (the public name, persists). - Agent system prompt: addSystemMessage(text,mode) split into addSystemMessage(text) + setSystemPrompt(text) over a private _persistSystemMessage. - UserMessage moved off Agent to module scope. - Underscore-prefix pure-internal Agent fields: _open_stream_fn, _auto_compacted, _retry_prng. The Agent facade is now pure 1:1 forwarders; it stays a wrapper only because init heap-pins the inner (copyable, move-safe handle) and conversation()/ sessionId() are accessors. Conversation and Stream are plain aliases. --- libpanto/src/agent.zig | 263 +++++++++++++++++++++++++----------------------- libpanto/src/public.zig | 46 ++++----- 2 files changed, 159 insertions(+), 150 deletions(-) (limited to 'libpanto') diff --git a/libpanto/src/agent.zig b/libpanto/src/agent.zig index 45347ea..8a7fbe4 100644 --- a/libpanto/src/agent.zig +++ b/libpanto/src/agent.zig @@ -179,6 +179,13 @@ fn toolErrorResult( return tool_mod.ownedTextResult(allocator, msg); } +/// The user's submission that opens a turn. A struct (not a bare slice) so +/// it can grow to carry file/image attachments alongside the chat text +/// without changing `Agent.run`'s signature. +pub const UserMessage = struct { + text: []const u8, +}; + pub const Agent = struct { allocator: Allocator, io: Io, @@ -205,15 +212,15 @@ pub const Agent = struct { session: session_store_mod.Session, /// Injectable streaming seam. Defaults to the real provider dispatch /// (`provider_mod.openStream`); tests override it with a stub. - open_stream_fn: provider_mod.OpenStreamFn = provider_mod.openStream, + _open_stream_fn: provider_mod.OpenStreamFn = provider_mod.openStream, /// Set by the embedder after `runStep` returns to learn whether an /// automatic compaction occurred this turn (so it can persist the /// rewritten conversation). Reset at the top of each `runStep`. - auto_compacted: bool = false, + _auto_compacted: bool = false, /// PRNG state for backoff jitter. Seeded lazily on first retry. Only /// touched from the single agent-loop thread (retries are serial), so /// no synchronization is needed. - retry_prng: ?std.Random.DefaultPrng = null, + _retry_prng: ?std.Random.DefaultPrng = null, /// Construct an agent. /// @@ -280,7 +287,19 @@ pub const Agent = struct { /// Add a system message (append or replace mode) to the conversation /// and persist it. The persisted entry records the mode so replay /// reconstructs the same effective system prompt. - pub fn addSystemMessage( + /// Append a system message (`.append` mode) and persist it. Adds to the + /// effective system prompt. + pub fn addSystemMessage(self: *Agent, text: []const u8) !void { + return self._persistSystemMessage(text, .append); + } + + /// Replace the effective system prompt (`.replace` mode) and persist it. + /// Discards all prior system text; replay reconstructs the same prompt. + pub fn setSystemPrompt(self: *Agent, text: []const u8) !void { + return self._persistSystemMessage(text, .replace); + } + + fn _persistSystemMessage( self: *Agent, text: []const u8, mode: conversation.SystemMode, @@ -300,13 +319,6 @@ pub const Agent = struct { ); } - /// The user's submission that opens a turn. A struct (not a bare slice) - /// so it can grow to carry file/image attachments alongside the chat - /// text without changing `run`'s signature. - pub const UserMessage = struct { - text: []const u8, - }; - /// Submit a user message and begin a turn, returning a resumable pull /// `Stream`. /// @@ -326,7 +338,7 @@ pub const Agent = struct { /// response inside the stream, so a mid-conversation `setConfig` takes /// effect at the next response boundary, never mid-stream. pub fn run(self: *Agent, message: UserMessage) !*Stream { - self.auto_compacted = false; + self._auto_compacted = false; // Append + persist the user prompt up front (the dangling-prompt // recovery guarantee). @@ -352,7 +364,7 @@ pub const Agent = struct { /// post-compaction window instead of `[start..]`. fn persistTurnTail(self: *Agent, start: usize) !void { const id = self.wireIdentity(); - if (self.auto_compacted) { + if (self._auto_compacted) { try turn_persist.persistCompaction( self.allocator, &self.session, @@ -409,7 +421,7 @@ pub const Agent = struct { var attempt: usize = 1; while (true) { var diag: provider_mod.ProviderDiagnostic = .{}; - const ps = self.open_stream_fn(self.allocator, self.io, cfg, &self.registry, &self.conversation, &diag) catch |err| { + const ps = self._open_stream_fn(self.allocator, self.io, cfg, &self.registry, &self.conversation, &diag) catch |err| { if (err == error.ContextOverflow) { return self.handleContextOverflow(cfg, out, err); } @@ -448,11 +460,11 @@ pub const Agent = struct { out: *stream_mod.EventQueue, err: anyerror, ) !provider_mod.ProviderStream { - if (self.auto_compacted) return err; // already retried once this turn + if (self._auto_compacted) return err; // already retried once this turn const sys = self.config.compaction.compaction_prompt orelse return err; - const res = try self.compact(sys, null); + const res = try self._compactInPlace(sys, null); if (!res.compacted) return err; // nothing to shed; give up - self.auto_compacted = true; + self._auto_compacted = true; try out.push(.{ .provider_retry = .{ .attempt = 1, .max_attempts = 2, @@ -462,7 +474,7 @@ pub const Agent = struct { } }); // Retry the same request against the compacted context. var diag: provider_mod.ProviderDiagnostic = .{}; - return self.open_stream_fn(self.allocator, self.io, cfg, &self.registry, &self.conversation, &diag); + return self._open_stream_fn(self.allocator, self.io, cfg, &self.registry, &self.conversation, &diag); } /// Compute the backoff delay (ms) for the just-failed `attempt` @@ -484,12 +496,12 @@ pub const Agent = struct { const capped: f64 = @min(base, @as(f64, @floatFromInt(policy.max_delay_ms))); var delay: u64 = @intFromFloat(capped); if (policy.jitter and delay > 0) { - if (self.retry_prng == null) { + if (self._retry_prng == null) { const ns = std.Io.Clock.now(.real, self.io).nanoseconds; const seed: u64 = @truncate(@as(u128, @bitCast(@as(i128, ns)))); - self.retry_prng = std.Random.DefaultPrng.init(seed); + self._retry_prng = std.Random.DefaultPrng.init(seed); } - const r = self.retry_prng.?.random(); + const r = self._retry_prng.?.random(); delay = r.intRangeLessThan(u64, 0, delay + 1); } return delay; @@ -512,9 +524,9 @@ pub const Agent = struct { /// verbatim. Mutates `self.conversation` in place. /// /// This is the pure transform — it does **not** persist. The explicit - /// `/compact` path uses `compactAndPersist`; the automatic - /// (context-overflow) path persists via `runStep`'s turn tail (the - /// rewritten window is logged as a fresh compaction window). + /// `/compact` entry point is the public `compact` (which persists); the + /// automatic (context-overflow) path persists via `runStep`'s turn tail + /// (the rewritten window is logged as a fresh compaction window). /// /// The system prompt survives untouched: all `.system`-role messages /// are preserved in order, and no `replace` block is written. Only the @@ -530,7 +542,7 @@ pub const Agent = struct { /// /// `system_prompt` is the compaction system prompt (resolved by the /// embedder from its `COMPACTION.md` layers, or a built-in default). - pub fn compact( + fn _compactInPlace( self: *Agent, system_prompt: []const u8, extra_instructions: ?[]const u8, @@ -586,16 +598,16 @@ pub const Agent = struct { } /// Compact and persist the result to the session store. This is the - /// explicit `/compact` entry point: it summarizes (via `compact`) and, - /// if anything was compacted, appends the new compaction window - /// (summary + restated kept suffix) to the store. Returns the same - /// `CompactionResult` for the embedder to report. + /// explicit `/compact` entry point: it summarizes (via the private + /// `_compactInPlace` transform) and, if anything was compacted, appends + /// the new compaction window (summary + restated kept suffix) to the + /// store. Returns the `CompactionResult` for the embedder to report. /// /// `override_system_prompt`, when non-null, is the compaction system /// prompt for this run; when null it falls back to /// `config.compaction.compaction_prompt`. With neither set, this errors /// (`error.NoCompactionPrompt`). - pub fn compactAndPersist( + pub fn compact( self: *Agent, override_system_prompt: ?[]const u8, extra_instructions: ?[]const u8, @@ -603,7 +615,7 @@ pub const Agent = struct { const system_prompt = override_system_prompt orelse self.config.compaction.compaction_prompt orelse return error.NoCompactionPrompt; - const res = try self.compact(system_prompt, extra_instructions); + const res = try self._compactInPlace(system_prompt, extra_instructions); if (res.compacted) { try turn_persist.persistCompaction( self.allocator, @@ -794,7 +806,7 @@ pub const Agent = struct { // stream until it commits the assistant message. var queue = stream_mod.EventQueue.init(alloc); defer queue.deinit(); - var ps = try self.open_stream_fn(alloc, self.io, cfg, registry, &conv, null); + var ps = try self._open_stream_fn(alloc, self.io, cfg, registry, &conv, null); defer ps.deinit(); while (true) { const status = try ps.produce(&queue); @@ -1065,21 +1077,24 @@ pub const Agent = struct { /// Event payloads borrow from the stream's decode state or the conversation /// and are valid only until the next `next()` call. pub const Stream = struct { - agent: *Agent, - queue: stream_mod.EventQueue, - phase: Phase, + // Internal state (underscore-prefixed by convention: not part of the + // public surface, even though Zig fields are always reachable). The only + // intended-public field is `state`, the transparent turn state machine. + _agent: *Agent, + _queue: stream_mod.EventQueue, + state: State, /// The active provider response, when in `.streaming`. - response: ?provider_mod.ProviderStream = null, + _response: ?provider_mod.ProviderStream = null, /// First message index of this turn (for persistence). - start: usize, + _start: usize, /// Set once the turn's tail has been persisted (on terminal or deinit). - persisted: bool = false, + _persisted: bool = false, /// A terminal error to surface once any already-queued events (e.g. /// `provider_retry` notices pushed before the failing attempt) have been /// drained. `next()` yields the queue first, then this error. - pending_error: ?anyerror = null, + _pending_error: ?anyerror = null, - pub const Phase = enum { + pub const State = enum { /// Open the next provider response (with retries). turn_start, /// Pump the active provider response into events. @@ -1094,10 +1109,10 @@ pub const Stream = struct { fn init(agent: *Agent) Stream { return .{ - .agent = agent, - .queue = stream_mod.EventQueue.init(agent.allocator), - .phase = .turn_start, - .start = agent.conversation.messages.items.len, + ._agent = agent, + ._queue = stream_mod.EventQueue.init(agent.allocator), + .state = .turn_start, + ._start = agent.conversation.messages.items.len, }; } @@ -1105,15 +1120,15 @@ pub const Stream = struct { // Persist whatever the turn committed, on every exit path — including // dropping the stream mid-turn after some messages were committed. self.persistTail(); - if (self.response) |ps| ps.deinit(); - self.queue.deinit(); - self.agent.allocator.destroy(self); + if (self._response) |ps| ps.deinit(); + self._queue.deinit(); + self._agent.allocator.destroy(self); } fn persistTail(self: *Stream) void { - if (self.persisted) return; - self.persisted = true; - self.agent.persistTurnTail(self.start) catch |e| { + if (self._persisted) return; + self._persisted = true; + self._agent.persistTurnTail(self._start) catch |e| { std.log.err("session: failed to persist turn: {t}", .{e}); }; } @@ -1123,84 +1138,84 @@ pub const Stream = struct { pub fn next(self: *Stream) !?Event { // Always drain queued events first; they borrow decode/conversation // state valid until this call returns. - if (self.queue.pop()) |ev| return ev; + if (self._queue.pop()) |ev| return ev; // Queue drained: a deferred terminal error surfaces now. - if (self.pending_error) |err| { - self.pending_error = null; - self.phase = .failed; + if (self._pending_error) |err| { + self._pending_error = null; + self.state = .failed; return err; } while (true) { - switch (self.phase) { + switch (self.state) { .done => return null, .failed => return error.StreamPoisoned, .turn_start => { // Re-read the config snapshot at each response boundary // so a mid-conversation swap takes effect here, never // mid-stream. - const cfg = self.agent.config; - const ps = self.agent.openWithRetries(cfg, &self.queue) catch |err| { + const cfg = self._agent.config; + const ps = self._agent.openWithRetries(cfg, &self._queue) catch |err| { // Surface the failure after any queued retry notices // (pushed before each backoff) are drained. - if (self.queue.pop()) |ev| { - self.pending_error = err; + if (self._queue.pop()) |ev| { + self._pending_error = err; return ev; } - self.phase = .failed; + self.state = .failed; return err; }; - self.response = ps; - self.phase = .streaming; - if (self.queue.pop()) |ev| return ev; // retry notices + self._response = ps; + self.state = .streaming; + if (self._queue.pop()) |ev| return ev; // retry notices }, .streaming => { - const ps = self.response.?; - const status = ps.produce(&self.queue) catch |err| { + const ps = self._response.?; + const status = ps.produce(&self._queue) catch |err| { // Mid-stream failure. The conversation was not // mutated (commit happens only at response // completion), so retry by re-opening from scratch, // exactly as the prior push loop replayed a failed // streamStep. Non-retryable errors propagate. ps.deinit(); - self.response = null; + self._response = null; if (!provider_mod.isRetryableProviderError(err)) { - self.phase = .failed; + self.state = .failed; return err; } - self.phase = .turn_start; + self.state = .turn_start; // Emit a retry notice so consumers see the stall. - const cfg = self.agent.config; - const delay_ms = self.agent.backoffDelayMs(cfg.retry, 1, null); - self.queue.push(.{ .provider_retry = .{ + const cfg = self._agent.config; + const delay_ms = self._agent.backoffDelayMs(cfg.retry, 1, null); + self._queue.push(.{ .provider_retry = .{ .attempt = 1, .max_attempts = cfg.retry.max_attempts, .delay_ms = delay_ms, .err = err, } }) catch |e| { - self.phase = .failed; + self.state = .failed; return e; }; if (delay_ms > 0) { const ms: i64 = @intCast(@min(delay_ms, std.math.maxInt(i64))); - self.agent.io.sleep(.fromMilliseconds(ms), .real) catch |e| { - self.phase = .failed; + self._agent.io.sleep(.fromMilliseconds(ms), .real) catch |e| { + self.state = .failed; return e; }; } - if (self.queue.pop()) |ev| return ev; + if (self._queue.pop()) |ev| return ev; continue; }; - if (self.queue.pop()) |ev| return ev; + if (self._queue.pop()) |ev| return ev; if (status == .response_complete) { ps.deinit(); - self.response = null; - self.phase = .after_response; + self._response = null; + self.state = .after_response; } // else `.more`: loop and pump again. }, .after_response => { - const conv = &self.agent.conversation; + const conv = &self._agent.conversation; const last = conv.messages.items[conv.messages.items.len - 1]; std.debug.assert(last.role == .assistant); @@ -1208,33 +1223,33 @@ pub const Stream = struct { // empty assistant message means the turn made no // observable progress. Surface it instead of looping. if (last.content.items.len == 0) { - self.phase = .failed; + self.state = .failed; return error.EmptyAssistantResponse; } if (!Agent.hasToolUseBlock(last)) { - self.phase = .done; + self.state = .done; self.persistTail(); return .turn_complete; } // Dispatch the tool calls, bracketed by boundary events. const count = toolUseCount(last); - self.queue.push(.{ .tool_dispatch_start = .{ .count = count } }) catch |e| { - self.phase = .failed; + self._queue.push(.{ .tool_dispatch_start = .{ .count = count } }) catch |e| { + self.state = .failed; return e; }; - self.agent.dispatchToolCalls(last) catch |err| { - self.phase = .failed; + self._agent.dispatchToolCalls(last) catch |err| { + self.state = .failed; return err; }; const result_msg = conv.messages.items[conv.messages.items.len - 1]; - self.queue.push(.{ .tool_dispatch_complete = .{ .message = result_msg } }) catch |e| { - self.phase = .failed; + self._queue.push(.{ .tool_dispatch_complete = .{ .message = result_msg } }) catch |e| { + self.state = .failed; return e; }; - self.phase = .turn_start; - if (self.queue.pop()) |ev| return ev; + self.state = .turn_start; + if (self._queue.pop()) |ev| return ev; }, } } @@ -1436,7 +1451,7 @@ fn trText(tr: conversation.ToolResultBlock) []const u8 { return ""; } -/// Test harness for the injectable `open_stream_fn` seam. +/// Test harness for the injectable `_open_stream_fn` seam. /// /// `provider_mod.OpenStreamFn` carries no user context (it mirrors the real /// free function exactly), so the stub parks its state in a module-level @@ -1485,7 +1500,7 @@ const StubProvider = struct { }; /// Point the global seam at this stub and return the function to assign - /// to `agent.open_stream_fn`. Call once per test, after constructing the + /// to `agent._open_stream_fn`. Call once per test, after constructing the /// stub on the stack. fn install(self: *StubProvider) provider_mod.OpenStreamFn { stub_active = self; @@ -1888,7 +1903,7 @@ test "agent persists user, assistant, and tool-result messages of a turn" { var agent = Agent.init(allocator, io, &h.config, cap.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); try drainTurn(&agent, "call a tool"); @@ -1925,7 +1940,7 @@ test "agent runs a turn against NullStore without persisting or erroring" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); try drainTurn(&agent, "hello"); @@ -2215,7 +2230,7 @@ test "runStep dispatches a tool call and loops to a final text turn" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; @@ -2266,7 +2281,7 @@ test "runStep dispatches multiple tool calls in parallel" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; @@ -2306,7 +2321,7 @@ test "runStep: native tool handler error becomes an error result and the model g var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; @@ -2340,7 +2355,7 @@ test "runStep: unknown tool becomes an error tool result and the loop continues" var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; @@ -2373,7 +2388,7 @@ test "runStep with no tool calls returns after one provider step" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; @@ -2400,7 +2415,7 @@ test "runStep surfaces EmptyAssistantResponse when provider commits an empty mes var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); try testing.expectError(error.EmptyAssistantResponse, drainTurn(&agent, "hi")); @@ -2431,7 +2446,7 @@ test "runStep delivers all source-backed calls in one batch on one thread" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; @@ -2483,7 +2498,7 @@ test "runStep: distinct sources run on distinct threads in parallel" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); try drainTurn(&agent, "go"); @@ -2521,7 +2536,7 @@ test "runStep: source whole-batch error becomes per-call error results and conti var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; @@ -2565,7 +2580,7 @@ test "runStep: mixed single Tools and source-backed tools coexist in one turn" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; @@ -2605,7 +2620,7 @@ test "setConfig swaps provider between turns; agent tool set persists" { var agent = Agent.init(allocator, io, &cfg_a, ns.store().create(), null); defer agent.deinit(); try agent.registerTool(try EchoTool.create(allocator, "late", "B:")); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); // The tool is visible regardless of which config is active. try testing.expect(agent.registry.lookup("late") != null); @@ -2644,7 +2659,7 @@ test "compact: summarizes prefix, keeps suffix, system survives" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; try conv.addSystemMessage("you are helpful"); @@ -2657,7 +2672,7 @@ test "compact: summarizes prefix, keeps suffix, system survives" { .{ .Text = try conversation.textualBlockFromSlice(allocator, "second recent answer") }, }, null); - const res = try agent.compact("Summarize the conversation.", null); + const res = try agent._compactInPlace("Summarize the conversation.", null); try testing.expect(res.compacted); // Expected rebuilt: [system, compaction summary(user), user q2, asst a2] @@ -2707,7 +2722,7 @@ test "compact: restated suffix usage reconstructs a fresh cumulative chain" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; // Prefix turn (will be summarized). Cumulative footprint = 500+40+10+50 @@ -2733,7 +2748,7 @@ test "compact: restated suffix usage reconstructs a fresh cumulative chain" { .{ .input = 600, .output = 4, .cache_read = 8, .cache_write = 2 }, ); - const res = try agent.compact("Summarize.", null); + const res = try agent._compactInPlace("Summarize.", null); try testing.expect(res.compacted); // Rebuilt: [summary(user), u1, a1, u2, a2]. @@ -2781,7 +2796,7 @@ test "compact: no-op when conversation already fits the budget" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; try conv.addSystemMessage("sys"); @@ -2790,7 +2805,7 @@ test "compact: no-op when conversation already fits the budget" { .{ .Text = try conversation.textualBlockFromSlice(allocator, "hello") }, }, null); - const res = try agent.compact("Summarize.", null); + const res = try agent._compactInPlace("Summarize.", null); try testing.expect(!res.compacted); try testing.expectEqual(@as(usize, 3), conv.messages.items.len); // Stub was never consumed. @@ -2819,7 +2834,7 @@ test "compact: extra instructions are appended to the system prompt" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; try conv.addUserMessage("question one two three"); @@ -2831,7 +2846,7 @@ test "compact: extra instructions are appended to the system prompt" { .{ .Text = try conversation.textualBlockFromSlice(allocator, "answer two") }, }, null); - const res = try agent.compact("Base prompt.", "keep bug #3 details"); + const res = try agent._compactInPlace("Base prompt.", "keep bug #3 details"); try testing.expect(res.compacted); } @@ -2860,7 +2875,7 @@ test "runStep: auto-compacts on context overflow and retries once" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; try conv.addSystemMessage("you are helpful"); @@ -2871,7 +2886,7 @@ test "runStep: auto-compacts on context overflow and retries once" { try drainTurn(&agent, "second recent question"); - try testing.expect(agent.auto_compacted); + try testing.expect(agent._auto_compacted); // After compaction + retry: [system, summary, user q2, assistant final]. const msgs = conv.messages.items; try testing.expectEqual(conversation.MessageRole.system, msgs[0].role); @@ -2904,7 +2919,7 @@ test "runStep: context overflow without compaction prompt propagates" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); // No compaction_system_prompt set -> overflow propagates. @@ -2976,7 +2991,7 @@ test "runStep: provider 429 retries then succeeds without duplicate messages" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; @@ -3020,7 +3035,7 @@ test "runStep: provider 500 retries with backoff notification" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); var rr = RetryRecorder{ .allocator = allocator }; @@ -3058,7 +3073,7 @@ test "runStep: provider auth failure does not retry" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); var rr = RetryRecorder{ .allocator = allocator }; @@ -3097,7 +3112,7 @@ test "runStep: retries exhaust and hard-fail after max_attempts" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); var rr = RetryRecorder{ .allocator = allocator }; @@ -3135,7 +3150,7 @@ test "runStep: Retry-After is honored and reported" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); var rr = RetryRecorder{ .allocator = allocator }; @@ -3170,7 +3185,7 @@ test "runStep: cancellation from a tool still hard-fails" { var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; @@ -3201,7 +3216,7 @@ test "runStep: source per-call error produces a per-call error result and contin var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; @@ -3240,7 +3255,7 @@ test "runStep: context-overflow compaction fires a compaction retry notification var agent = Agent.init(allocator, io, &h.config, ns.store().create(), null); defer agent.deinit(); h.seedInto(&agent); - agent.open_stream_fn = stub.install(); + agent._open_stream_fn = stub.install(); const conv = &agent.conversation; try conv.addSystemMessage("you are helpful"); @@ -3253,7 +3268,7 @@ test "runStep: context-overflow compaction fires a compaction retry notification defer rr.deinit(); try drainTurnRecording(&agent, "second recent question", &rr); - try testing.expect(agent.auto_compacted); + try testing.expect(agent._auto_compacted); // Exactly one notification, flagged as a compaction retry with no delay. try testing.expectEqual(@as(usize, 1), rr.infos.items.len); try testing.expect(rr.infos.items[0].compaction); diff --git a/libpanto/src/public.zig b/libpanto/src/public.zig index 31d5f7b..8a6fcd1 100644 --- a/libpanto/src/public.zig +++ b/libpanto/src/public.zig @@ -127,32 +127,24 @@ pub const Event = stream_mod.Event; pub const ContentBlockType = provider_mod.ContentBlockType; pub const ProviderRetryInfo = provider_mod.ProviderRetryInfo; -/// The transparent turn state machine, renamed from the internal `Phase`. -pub const State = agent_mod.Stream.Phase; - -pub const Stream = struct { - inner: *agent_mod.Stream, - - /// `!?Event`: a value is progress (including the terminal event), `null` - /// is exhaustion, an error is a genuine failure. - pub fn next(self: Stream) !?Event { - return self.inner.next(); - } - pub fn deinit(self: Stream) void { - // The internal `Stream.deinit` persists the turn tail and frees its - // own heap allocation. - self.inner.deinit(); - } - pub fn state(self: Stream) State { - return self.inner.phase; - } -}; +/// The pull event stream for one turn. Aliased straight through: its public +/// surface is exactly `next`/`deinit` plus the transparent `state` field +/// (`State`); every other field is underscore-prefixed internal state. +/// `Agent.run` returns a `*Stream` (heap-pinned); the caller owns it and +/// must `deinit` it (which persists the turn tail and frees the allocation). +/// +/// `next()` returns `!?Event`: a value is progress (including the terminal +/// event), `null` is exhaustion, an error is a genuine failure. +pub const Stream = agent_mod.Stream; + +/// The transparent turn state machine (`Stream.state`). +pub const State = agent_mod.Stream.State; // =========================================================================== // Agent (behavioral, pointer-wrapped) // =========================================================================== -pub const UserMessage = struct { text: []const u8 }; +pub const UserMessage = agent_mod.UserMessage; pub const CompactionResult = agent_mod.Agent.CompactionResult; pub const Agent = struct { @@ -192,20 +184,22 @@ pub const Agent = struct { pub fn setConfig(self: Agent, cfg: *const Config) void { self.inner.setConfig(cfg); } - pub fn run(self: Agent, message: UserMessage) !Stream { - return .{ .inner = try self.inner.run(.{ .text = message.text }) }; + /// Submit a user message and begin a turn, returning a heap-pinned + /// `*Stream` the caller owns and must `deinit`. + pub fn run(self: Agent, message: UserMessage) !*Stream { + return self.inner.run(message); } /// Append a system message to the conversation and persist it (the /// `.append` `SystemMode`: it adds to the effective prompt). pub fn addSystemMessage(self: Agent, text: []const u8) !void { - return self.inner.addSystemMessage(text, .append); + return self.inner.addSystemMessage(text); } /// Replace the effective system prompt and persist it (the `.replace` /// `SystemMode`: it discards all prior system text). On replay the log /// reconstructs the same effective prompt. pub fn setSystemPrompt(self: Agent, text: []const u8) !void { - return self.inner.addSystemMessage(text, .replace); + return self.inner.setSystemPrompt(text); } /// Compact and persist (the explicit `/compact` entry point). @@ -214,7 +208,7 @@ pub const Agent = struct { /// to the compaction prompt for this run (the `/compact $ARGUMENTS` /// path). pub fn compact(self: Agent, override_system_prompt: ?[]const u8, extra: ?[]const u8) !CompactionResult { - return self.inner.compactAndPersist(override_system_prompt, extra); + return self.inner.compact(override_system_prompt, extra); } /// A borrowed pointer to the agent's live conversation, for in-place -- cgit v1.3