From 02daacdd08d5efe7f1df051070a4f3d71aa8f2a7 Mon Sep 17 00:00:00 2001 From: Rach Pradhan <54503978+justrach@users.noreply.github.com> Date: Thu, 6 Aug 2026 12:39:05 +0800 Subject: [PATCH 1/4] =?UTF-8?q?feat(engine):=20sayText=20=E2=80=94=20say()?= =?UTF-8?q?=20for=20a=20line=20a=20sink=20already=20rendered=20(#422)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Slice 1c needs the notices out of agent_tools.zig, and every one of them went through say(), whose routing is not incidental: --json swallows a root's line, a pool-thread child has no writer and goes through the tick gate with a "[label] " prefix and a repair for the cut newline, and a format ending in \n releases held child ticks. A sink holds bytes, not a comptime format, so it cannot call say(). This is the same function with the line-ending test moved from the format to the last byte, and errors swallowed because an emit path has nowhere to return them. say() is left alone: its worker branch formats prefix and payload in ONE print into a fixed slot, and re-deriving that through a temporary would change where an over-long line gets cut. Co-Authored-By: Codegraff --- src/agent_output.zig | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/src/agent_output.zig b/src/agent_output.zig index 9f9263f8..7c4c757c 100644 --- a/src/agent_output.zig +++ b/src/agent_output.zig @@ -53,6 +53,32 @@ fn endsLine(comptime fmt: []const u8) bool { return fmt.len > 0 and fmt[fmt.len - 1] == '\n'; } +/// say() for text that is already rendered (#422: a sink holds the bytes, not +/// a comptime format). Same routing, same worker-line framing, same tick-gate +/// rule — only the line-ending test moves from the format to the last byte. +/// Errors are swallowed: an event sink's emit path has nowhere to return them. +pub fn sayText(self: *Agent, text: []const u8) void { + if (main_mod.json_mode and !self.sub) return; + if (self.out) |w| { + w.print("{s}", .{text}) catch return; + w.flush() catch return; + if (!self.sub and !main_mod.json_mode and text.len > 0 and text[text.len - 1] == '\n') _ = tick_gate.setLineStart(true); + return; + } + // A pool-thread child has no writer: same fixed slot, same label prefix, + // and the same repair when an over-long line lost its newline to the cut. + var buf: [tick_gate.slot_bytes]u8 = undefined; + var sink = Io.Writer.fixed(&buf); + const fit = if (sink.print(" [{s}] {s}", .{ self.label, text })) |_| true else |_| false; + var line = sink.buffered(); + if (!fit or line.len == 0 or line[line.len - 1] != '\n') { + const at: usize = @min(line.len, buf.len - 1); + buf[at] = '\n'; + line = buf[0 .. at + 1]; + } + tick_gate.workerLine(line); +} + /// Remember the formatted message for the --json `error` event, then print /// like say() + a #398 duration hint; last_api_error keeps provider words. pub fn sayApiError(self: *Agent, comptime fmt: []const u8, args: anytype) !void { From 850038250acfff43b56fee297248790874cdc6d2 Mon Sep 17 00:00:00 2001 From: Rach Pradhan <54503978+justrach@users.noreply.github.com> Date: Thu, 6 Aug 2026 12:39:14 +0800 Subject: [PATCH 2/4] feat(engine): the tool-execution cluster gets a vocabulary and both sinks (#422) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Eleven variants for the moments agent_tools.zig has been drawing inline: the call bracket (announced/started) and its closing bracket (result/ finished), a pre-run refusal, the parallel-batch tallies, and the meta-tool notices. The pairs exist because one moment is TWO wire lines and each line owns a sequence id — collapsing them would reserve one id and write two. durable() now reads the payload, not just the tag, and it has to: whether a tool moment reaches the wire depends on which tool it is (ask_user's bracket goes out as the `ask_user` event; a meta tool's result never had a wire shape). JsonSink is now defined as "write exactly the durable events" with a panic on the else arm, so a future durable variant cannot reach the wire without a shape and burn its id on nothing (#330). The terminal half lives in agent_tool_render.zig — the one file in the cluster that reaches the palette, like agent_stream_render.zig for the stream. Every function there is the old code path gate for gate, with the two byte-preserving caps (160 arg bytes, 100 preview bytes) named rather than inlined. Co-Authored-By: Codegraff --- src/agent_tool_render.zig | 247 ++++++++++++++++++++++++++++++++++++++ src/engine_events.zig | 135 +++++++++++++++++++++ src/engine_sink.zig | 217 +++++++++++++++++++++++---------- 3 files changed, 534 insertions(+), 65 deletions(-) create mode 100644 src/agent_tool_render.zig diff --git a/src/agent_tool_render.zig b/src/agent_tool_render.zig new file mode 100644 index 00000000..f602d038 --- /dev/null +++ b/src/agent_tool_render.zig @@ -0,0 +1,247 @@ +//! TuiSink's half of the tool-execution cluster (#422 slice 1c): the terminal +//! rendering for the tool events engine_events.zig defines — the ⚙ call line, +//! the compact ✓/✗/⊘ result line, the parallel-batch tallies, and the +//! meta-tool notices. Frontend territory, like agent_stream_render.zig: this +//! is the ONLY file in the tool cluster that reaches the terminal palette, and +//! engine_sink.zig is its only caller. +//! +//! Every function here is the old inline agent_tools.zig code path, gate for +//! gate and byte for byte, with two deliberate mechanical differences: +//! - text is rendered whole and handed to agent_output.sayText, which is +//! say()'s runtime-string sibling (same routing, same worker-line framing); +//! - write errors are swallowed rather than propagated, because a sink's emit +//! path returns void. A caller that used to abort its batch on a broken +//! stdout now continues into the next failing write instead. + +const std = @import("std"); +const Io = std.Io; + +const main_mod = @import("main.zig"); +const agent_mod = @import("agent.zig"); +const Agent = agent_mod.Agent; +const agent_output = @import("agent_output.zig"); +const sayText = agent_output.sayText; +const util = @import("util.zig"); // tests only: repeatBytes for the arg-cap case + +const ansi = @import("ansi.zig"); +const style = &ansi.style; + +const engine_events = @import("engine_events.zig"); +const ToolInvocation = engine_events.ToolInvocation; +const ToolOutcome = engine_events.ToolOutcome; +const BatchOutcome = engine_events.BatchOutcome; + +/// How much of a call's JSON arguments the ⚙ line shows. A byte cap, not a +/// character one: the cut may split a UTF-8 sequence, exactly as before. +const arg_preview_bytes: usize = 160; +/// How much of a result's first line the ✓ line shows. +const result_preview_bytes: usize = 100; + +/// The ⚙ announcement line. Suppressed when the call's prose already streamed +/// live out of its arguments — the line would just repeat what was read. +pub fn toolUseLine(a: *Agent, t: ToolInvocation) void { + if (t.arg_streamed) return; + var args: Io.Writer.Allocating = .init(a.gpa); + defer args.deinit(); + var s: std.json.Stringify = .{ .writer = &args.writer }; + s.write(t.input) catch return; + const full = args.writer.buffered(); + const shown = if (full.len > arg_preview_bytes) full[0..arg_preview_bytes] else full; + var line: Io.Writer.Allocating = .init(a.gpa); + defer line.deinit(); + line.writer.print("{s}⚙{s} {s}{s} {s}{s}{s}{s}\n", .{ + style.dim, style.reset, style.accent, t.name, + style.dim, shown, + if (full.len > arg_preview_bytes) "…" else "", + style.reset, + }) catch return; + sayText(a, line.writer.buffered()); +} + +/// Compact result feedback for one finished tool call: a green ✓ / red ✗ / +/// yellow ⊘ and a one-line preview of what it returned. Root only (subagents +/// have no writer); meta tools render their own UX, so skip them. This one +/// writes straight to the writer — it never ended a tick-gate row. +pub fn toolResultLine(a: *Agent, r: ToolOutcome) void { + const w = a.out orelse return; + if (r.meta) return; + const all = std.mem.trim(u8, r.text, " \t\r\n"); + var preview = all; + if (std.mem.indexOfScalar(u8, preview, '\n')) |nl| preview = preview[0..nl]; + preview = std.mem.trim(u8, preview, " \t\r"); + const shown = if (preview.len > result_preview_bytes) preview[0..result_preview_bytes] else preview; + const truncated = shown.len < all.len; // more content (extra lines or >100 chars) + const mark = if (r.cancelled) "⊘" else if (r.is_error) "✗" else "✓"; + const mc = if (r.cancelled) style.yellow else if (r.is_error) style.red else style.green; + var tbuf: [24]u8 = undefined; + const timing = if (main_mod.show_timing and r.ms > 0) + (std.fmt.bufPrint(&tbuf, " ({d}ms)", .{r.ms}) catch "") + else + ""; + w.print(" {s}{s}{s}{s}{s}{s} {s}{s}{s}{s}\n", .{ + mc, mark, style.reset, style.dim, timing, style.reset, + style.dim, shown, + if (truncated) "…" else "", + style.reset, + }) catch return; + w.flush() catch return; +} + +// The batch tallies fit a fixed slot with room to spare: the widest line is +// three usize decimals plus ~50 bytes of wording and two short style runs. +const notice_buf_bytes: usize = 192; + +pub fn parallelBatchStarted(a: *Agent, count: usize) void { + if (a.sub) return; // a child's fan-out is the root's line to draw, not its own + var buf: [notice_buf_bytes]u8 = undefined; + const line = std.fmt.bufPrint(&buf, " {s}↯ running {d} tools in parallel{s}\n", .{ style.dim, count, style.reset }) catch return; + sayText(a, line); +} + +pub fn parallelBatchFinished(a: *Agent, o: BatchOutcome) void { + if (a.sub) return; + var buf: [notice_buf_bytes]u8 = undefined; + const line = std.fmt.bufPrint(&buf, " {s}↯ parallel tools finished: {d} completed, {d} failed, {d} cancelled{s}\n", .{ style.dim, o.done, o.failed, o.cancelled, style.reset }) catch return; + sayText(a, line); +} + +/// #318: the checklist isn't settled, so the completion parks. The escapes are +/// the literal bytes the old call site spelled out (⏸, 🎯 below). +pub fn completionDeferred(a: *Agent) void { + if (a.sub) return; + sayText(a, "\xe2\x8f\xb8 completion deferred \xe2\x80\x94 the standing goal's checklist isn't settled\n"); +} + +pub fn goalCompleted(a: *Agent) void { + sayText(a, "\xf0\x9f\x8e\xaf standing goal complete\n"); +} + +/// A meta tool's own user-facing text, one line, exactly as it came. +pub fn toolTextLine(a: *Agent, text: []const u8) void { + if (a.sub) return; + var line: Io.Writer.Allocating = .init(a.gpa); + defer line.deinit(); + line.writer.print("{s}\n", .{text}) catch return; + sayText(a, line.writer.buffered()); +} + +fn testAgent(w: *Io.Writer) Agent { + return .{ + .gpa = std.testing.allocator, + .arena = std.testing.allocator, + .io = undefined, + .client = undefined, + .provider = undefined, + .messages = undefined, + .sub = false, + .label = "test", + .out = w, + }; +} + +test "the ⚙ line reproduces the old inline announcement, cap and ellipsis included" { + const saved = ansi.style; + ansi.style = .{}; // no color: assert the text, not the palette + defer ansi.style = saved; + const saved_json = main_mod.json_mode; // sayText's root gate reads it + main_mod.json_mode = false; + defer main_mod.json_mode = saved_json; + var arena_state = std.heap.ArenaAllocator.init(std.testing.allocator); + defer arena_state.deinit(); + var aw: Io.Writer.Allocating = .init(std.testing.allocator); + defer aw.deinit(); + var a = testAgent(&aw.writer); + + const input = try std.json.parseFromSliceLeaky(std.json.Value, arena_state.allocator(), "{\"path\":\"fixture.txt\"}", .{}); + toolUseLine(&a, .{ .name = "read_file", .input = input }); + try std.testing.expectEqualStrings("⚙ read_file {\"path\":\"fixture.txt\"}\n", aw.writer.buffered()); + + // Prose that already streamed live is not announced a second time. + aw.clearRetainingCapacity(); + toolUseLine(&a, .{ .name = "attempt_completion", .input = input, .arg_streamed = true }); + try std.testing.expectEqualStrings("", aw.writer.buffered()); + + // Over the cap: exactly arg_preview_bytes of JSON, then the ellipsis. + aw.clearRetainingCapacity(); + const long = try std.fmt.allocPrint(arena_state.allocator(), "{{\"q\":\"{s}\"}}", .{&util.repeatBytes("x", 400)}); + const long_input = try std.json.parseFromSliceLeaky(std.json.Value, arena_state.allocator(), long, .{}); + toolUseLine(&a, .{ .name = "codedb", .input = long_input }); + const line = aw.writer.buffered(); + try std.testing.expect(std.mem.startsWith(u8, line, "⚙ codedb {\"q\":\"xxx")); + try std.testing.expect(std.mem.endsWith(u8, line, "…\n")); + try std.testing.expectEqual(arg_preview_bytes, line.len - "⚙ codedb ".len - "…\n".len); +} + +test "the result line marks success, failure and cancellation and previews one line" { + const saved = ansi.style; + ansi.style = .{}; + defer ansi.style = saved; + const saved_timing = main_mod.show_timing; + main_mod.show_timing = false; + defer main_mod.show_timing = saved_timing; + var aw: Io.Writer.Allocating = .init(std.testing.allocator); + defer aw.deinit(); + var a = testAgent(&aw.writer); + + toolResultLine(&a, .{ .name = "read_file", .text = "line one\nline two\n", .is_error = false }); + try std.testing.expectEqualStrings(" ✓ line one…\n", aw.writer.buffered()); + + aw.clearRetainingCapacity(); + toolResultLine(&a, .{ .name = "bash", .text = "boom", .is_error = true }); + try std.testing.expectEqualStrings(" ✗ boom\n", aw.writer.buffered()); + + aw.clearRetainingCapacity(); + toolResultLine(&a, .{ .name = "bash", .text = "stopped", .is_error = true, .cancelled = true }); + try std.testing.expectEqualStrings(" ⊘ stopped\n", aw.writer.buffered()); + + // Meta tools draw their own UX; the ✓ line stays out of their way. + aw.clearRetainingCapacity(); + toolResultLine(&a, .{ .name = "todo_write", .text = "todos", .is_error = false, .meta = true }); + try std.testing.expectEqualStrings("", aw.writer.buffered()); + + // --timing adds the measured duration between the mark and the preview. + aw.clearRetainingCapacity(); + main_mod.show_timing = true; + toolResultLine(&a, .{ .name = "bash", .text = "ok", .is_error = false, .ms = 42 }); + try std.testing.expectEqualStrings(" ✓ (42ms) ok\n", aw.writer.buffered()); +} + +test "batch tallies and meta notices render the old wording verbatim" { + const saved = ansi.style; + ansi.style = .{}; + defer ansi.style = saved; + const saved_json = main_mod.json_mode; // sayText's root gate reads it + main_mod.json_mode = false; + defer main_mod.json_mode = saved_json; + var aw: Io.Writer.Allocating = .init(std.testing.allocator); + defer aw.deinit(); + var a = testAgent(&aw.writer); + + parallelBatchStarted(&a, 3); + try std.testing.expectEqualStrings(" ↯ running 3 tools in parallel\n", aw.writer.buffered()); + + aw.clearRetainingCapacity(); + parallelBatchFinished(&a, .{ .done = 2, .failed = 1, .cancelled = 0 }); + try std.testing.expectEqualStrings(" ↯ parallel tools finished: 2 completed, 1 failed, 0 cancelled\n", aw.writer.buffered()); + + aw.clearRetainingCapacity(); + completionDeferred(&a); + try std.testing.expectEqualStrings("⏸ completion deferred — the standing goal's checklist isn't settled\n", aw.writer.buffered()); + + aw.clearRetainingCapacity(); + goalCompleted(&a); + try std.testing.expectEqualStrings("🎯 standing goal complete\n", aw.writer.buffered()); + + aw.clearRetainingCapacity(); + toolTextLine(&a, "completion recorded"); + try std.testing.expectEqualStrings("completion recorded\n", aw.writer.buffered()); + + // A subagent announces none of the root's batch/goal lines. + aw.clearRetainingCapacity(); + a.sub = true; + parallelBatchStarted(&a, 3); + parallelBatchFinished(&a, .{ .done = 1, .failed = 0, .cancelled = 0 }); + completionDeferred(&a); + toolTextLine(&a, "x"); + try std.testing.expectEqualStrings("", aw.writer.buffered()); +} diff --git a/src/engine_events.zig b/src/engine_events.zig index d54bc0ca..0b94edbe 100644 --- a/src/engine_events.zig +++ b/src/engine_events.zig @@ -59,6 +59,50 @@ pub const TransportAbort = struct { turn_ending: bool, }; +/// One tool call as the engine hands it to a frontend (#422 slice 1c). +/// `ask_user` is called out on its own because a --json client learns of +/// that moment through its own `ask_user` event, never as a tool-call pair. +/// `arg_streamed` says the call's prose already streamed out of its +/// still-in-flight arguments (agent_argstream.zig), so an announcement line +/// would repeat what the reader just watched arrive. +pub const ToolInvocation = struct { + name: []const u8, + input: std.json.Value, + ask_user: bool = false, + arg_streamed: bool = false, +}; + +/// One finished tool call. `text` is the model-facing result the engine +/// already capped and previewed; `ms` is its measured wall clock. `meta` +/// marks a meta tool (schema.isMetaName) — those own their UX and, except +/// for ask_user, never carried a result on the wire. +pub const ToolOutcome = struct { + name: []const u8, + text: []const u8, + is_error: bool, + cancelled: bool = false, + ms: i64 = 0, + meta: bool = false, + ask_user: bool = false, +}; + +/// A tool call the harness refused before it ran: the verifier boundary, a +/// stale eval, the --max-tool-calls budget, the dedupe rule, or review mode. +pub const ToolRejection = struct { + name: []const u8, + input: std.json.Value, + reason: []const u8, + message: []const u8, +}; + +/// How a parallel tool batch ended (#266): the tallies, not a rendered line. +pub const BatchOutcome = struct { done: usize, failed: usize, cancelled: usize }; + +/// Text whose layout belongs to the meta tool that produced it (a completion +/// answer, a rendered todo list) — carried whole because the structure lives +/// in that tool's own module, not in this vocabulary. +pub const ToolText = struct { text: []const u8 }; + /// Everything the streaming path tells a frontend. Each doc comment states /// the emission site's contract, not how any one sink draws it. pub const EngineEvent = union(enum) { @@ -106,14 +150,63 @@ pub const EngineEvent = union(enum) { /// tear down live-stream presentation (spinner, an open reasoning-only /// Thinking block). stream_finished, + + // ── The tool-execution cluster (slice 1c) ──────────────────────────── + /// A tool call cleared the gates and is about to run. Wire: the existing + /// `tool_call` event. TUI: the ⚙ announcement line. + tool_call_announced: ToolInvocation, + /// The same call, now dispatched — the bracket a supervisor times against. + /// Wire: the existing `tool_call_started` event. TUI: nothing (the ⚙ line + /// already said it). + tool_call_started: ToolInvocation, + /// A tool call returned. Wire: the existing `tool_result` event. TUI: the + /// compact ✓/✗/⊘ line with a one-line preview. + tool_result: ToolOutcome, + /// The same call's closing bracket, carrying the outcome and duration + /// rather than the text. Wire: `tool_call_finished`. TUI: nothing. + tool_call_finished: ToolOutcome, + /// A tool call the harness refused before running it. Wire: the existing + /// `tool_rejected` event; the TUI has never drawn one (the refusal reaches + /// the user as the model's next answer). + tool_rejected: ToolRejection, + /// A batch of external tool calls is fanning out across the pool. + /// Presentation-only: the wire brackets each call individually. + parallel_batch_started: struct { count: usize }, + /// That batch is joined; the payload is the tally (#266 — a cancelled + /// batch used to just look "running" and then fail). + parallel_batch_finished: BatchOutcome, + /// attempt_completion was refused because the standing goal's checklist + /// is not settled (#318). Presentation-only; the model gets the refusal + /// as its tool result. + completion_deferred, + /// A standing --goal was retired by an accepted attempt_completion. + goal_completed, + /// The answer attempt_completion carried, surfaced because it did NOT + /// stream live out of the call's arguments. + completion_text: ToolText, + /// todo_write applied; the payload is the list as goal_todo rendered it. + todo_list_updated: ToolText, }; /// Durable events are the protocol stream: what the --json wire emits today /// and what a future event log persists. Only they reserve sequence ids — /// presentation pulses must not open gaps in the wire's numbering. +/// +/// This reads the PAYLOAD, not just the tag, and it has to: whether a tool +/// moment reaches the wire depends on which tool it is. A durable sink writes +/// exactly the events this returns true for (engine_sink.jsonEmit asserts it), +/// so a reserved id can never burn with no line behind it (#330). pub fn durable(ev: EngineEvent) bool { return switch (ev) { .reasoning_delta, .text_delta => true, + // Every tool's call bracket is on the wire except ask_user's: a + // --json client is handed that moment as its own `ask_user` event, + // and answers it on stdin. + .tool_call_announced, .tool_call_started => |t| !t.ask_user, + // Meta tools render their own UX and never carried a wire result — + // except ask_user, whose typed reply IS the result. + .tool_result, .tool_call_finished => |r| !r.meta or r.ask_user, + .tool_rejected => true, else => false, }; } @@ -187,6 +280,48 @@ test "slice 1b: tool-arg prose and transport aborts are presentation pulses" { for (pulses) |ev| try std.testing.expect(!durable(ev)); } +test "slice 1c: the tool bracket is durable per TOOL, not per tag" { + const call: ToolInvocation = .{ .name = "bash", .input = .null }; + const ask: ToolInvocation = .{ .name = "ask_user", .input = .null, .ask_user = true }; + // An ordinary tool's bracket is the wire's tool_call/tool_call_started… + try std.testing.expect(durable(.{ .tool_call_announced = call })); + try std.testing.expect(durable(.{ .tool_call_started = call })); + // …while ask_user's reaches a --json client as the `ask_user` event, so + // neither half may reserve an id the wire will not spend (#330). + try std.testing.expect(!durable(.{ .tool_call_announced = ask })); + try std.testing.expect(!durable(.{ .tool_call_started = ask })); + // arg_streamed is a TUI-only suppression: the wire still carries the call. + const streamed: ToolInvocation = .{ .name = "bash", .input = .null, .arg_streamed = true }; + try std.testing.expect(durable(.{ .tool_call_announced = streamed })); + + const ext: ToolOutcome = .{ .name = "bash", .text = "ok", .is_error = false }; + const meta: ToolOutcome = .{ .name = "todo_write", .text = "ok", .is_error = false, .meta = true }; + const answer: ToolOutcome = .{ .name = "ask_user", .text = "yes", .is_error = false, .meta = true, .ask_user = true }; + try std.testing.expect(durable(.{ .tool_result = ext })); + try std.testing.expect(durable(.{ .tool_call_finished = ext })); + try std.testing.expect(!durable(.{ .tool_result = meta })); + try std.testing.expect(!durable(.{ .tool_call_finished = meta })); + try std.testing.expect(durable(.{ .tool_result = answer })); + try std.testing.expect(durable(.{ .tool_call_finished = answer })); + + // A refusal has always been a wire-only moment. + try std.testing.expect(durable(.{ .tool_rejected = .{ .name = "bash", .input = .null, .reason = "budget", .message = "no" } })); +} + +test "slice 1c: the tool-cluster notices are presentation pulses" { + // None of these ever appeared on the --json wire, so promoting one is a + // schema_version event rather than refactor fallout. + const pulses: [6]EngineEvent = .{ + .{ .parallel_batch_started = .{ .count = 3 } }, + .{ .parallel_batch_finished = .{ .done = 2, .failed = 1, .cancelled = 0 } }, + .completion_deferred, + .goal_completed, + .{ .completion_text = .{ .text = "done" } }, + .{ .todo_list_updated = .{ .text = "todos" } }, + }; + for (pulses) |ev| try std.testing.expect(!durable(ev)); +} + test "generation only moves forward, one restart at a time" { const before = generation(); try std.testing.expect(before >= 1); diff --git a/src/engine_sink.zig b/src/engine_sink.zig index fa3290e1..6ed6131d 100644 --- a/src/engine_sink.zig +++ b/src/engine_sink.zig @@ -36,6 +36,7 @@ const engine_events = @import("engine_events.zig"); const EngineEvent = engine_events.EngineEvent; const protocol_seq = @import("protocol_seq.zig"); const render = @import("agent_stream_render.zig"); +const tool_render = @import("agent_tool_render.zig"); // slice 1c: the tool cluster's terminal half const tick_gate = @import("tick_gate.zig"); // #tui-tick: child ticks wait for a foreground line boundary /// An event plus its position, as delivered to a sink. @@ -171,6 +172,18 @@ fn tuiEmit(ctx: *anyopaque, ev: Stamped) void { render.closeThinkingBlock(a); // a reasoning-only turn still closes its block render.spinnerStop(a); }, + // The tool cluster (slice 1c). Its drawing lives in agent_tool_render, + // the one file down here that still reaches the palette; the moments + // the terminal never drew (the dispatch/close brackets, refusals) are + // silent rather than absent from the vocabulary. + .tool_call_announced => |t| tool_render.toolUseLine(a, t), + .tool_result => |r| tool_render.toolResultLine(a, r), + .tool_call_started, .tool_call_finished, .tool_rejected => {}, + .parallel_batch_started => |b| tool_render.parallelBatchStarted(a, b.count), + .parallel_batch_finished => |b| tool_render.parallelBatchFinished(a, b), + .completion_deferred => tool_render.completionDeferred(a), + .goal_completed => tool_render.goalCompleted(a), + .completion_text, .todo_list_updated => |t| tool_render.toolTextLine(a, t.text), } } @@ -186,19 +199,35 @@ fn notice(a: *Agent, text: []const u8) void { fn jsonEmit(ctx: *anyopaque, ev: Stamped) void { const a: *Agent = @ptrCast(@alignCast(ctx)); const w = a.out orelse return; + // The invariant that keeps #330's numbering gap-free: this sink writes a + // line for EXACTLY the events engine_events.durable() claims, so a + // reserved id can never end up with nothing behind it. Payload-dependent + // durability (ask_user's bracket, a meta tool's result) is decided there, + // once, rather than re-derived per branch below. + // + // Stream end/abort still flushes the held render tail, as the old inline + // path did in EVERY mode. (Slice 1b correction to this note: tool-arg + // prose CANNOT dirty md state here — argLiveDelta has always gated --json + // off, and this sink drops .tool_arg_delta — so with .text_delta going to + // the wire the tail is clean and the flush adds no bytes today. It stays + // because removing a wire-visible behavior, however latent, is a + // deliberate schema-gated change, not refactor fallout.) + if (!engine_events.durable(ev.event)) return switch (ev.event) { + .stream_aborted, .stream_complete => a.flushStreamTail(), + else => {}, + }; switch (ev.event) { .reasoning_delta => |d| jsonLine(w, ev.cursor, .{ .type = "reasoning", .text = d.text }), .text_delta => |d| jsonLine(w, ev.cursor, .{ .type = "text", .text = d.text }), - // Stream end/abort still flushes the held render tail, as the old - // inline path did in EVERY mode. (Slice 1b correction to this note: - // tool-arg prose CANNOT dirty md state here — argLiveDelta has always - // gated --json off, and this sink drops .tool_arg_delta — so with - // .text_delta going to the wire the tail is clean and the flush adds - // no bytes today. It stays because removing a wire-visible behavior, - // however latent, is a deliberate schema-gated change, not refactor - // fallout.) - .stream_aborted, .stream_complete => a.flushStreamTail(), - else => {}, + .tool_call_announced => |t| jsonLine(w, ev.cursor, .{ .type = "tool_call", .name = t.name, .input = t.input }), + .tool_call_started => |t| jsonLine(w, ev.cursor, .{ .type = "tool_call_started", .name = t.name, .input = t.input }), + .tool_result => |r| jsonLine(w, ev.cursor, .{ .type = "tool_result", .name = r.name, .is_error = r.is_error, .text = r.text }), + .tool_call_finished => |r| jsonLine(w, ev.cursor, .{ .type = "tool_call_finished", .name = r.name, .is_error = r.is_error, .ms = r.ms }), + .tool_rejected => |r| jsonLine(w, ev.cursor, .{ .type = "tool_rejected", .name = r.name, .reason = r.reason, .input = r.input, .message = r.message }), + // Unreachable: durable() gated every other tag out above. Kept as a + // hard stop so a NEW durable variant cannot silently reach the wire + // without a shape — that would burn its id on nothing (#330). + else => @panic("engine_sink: durable event with no wire shape"), } } @@ -252,6 +281,23 @@ test "a presentation sink never reserves sequence ids" { try std.testing.expectEqual(@as(u64, 0), protocol_seq.current()); } +/// The Agent shape every sink test renders through: allocator-backed, rooted +/// (not a subagent), writing into the caller's buffer. `io` stays undefined — +/// dispatch only touches it under the --json lock, which these tests pin off. +fn testAgent(w: *Io.Writer) Agent { + return .{ + .gpa = std.testing.allocator, + .arena = std.testing.allocator, + .io = undefined, + .client = undefined, + .provider = undefined, + .messages = undefined, + .sub = false, + .label = "test", + .out = w, + }; +} + fn recordEmit(ctx: *anyopaque, ev: Stamped) void { const rec: *std.ArrayList(Stamped) = @ptrCast(@alignCast(ctx)); rec.append(std.testing.allocator, ev) catch @panic("OOM"); @@ -265,17 +311,7 @@ test "JsonSink writes today's wire lines byte-for-byte" { defer protocol_seq.resetForTest(); var aw: Io.Writer.Allocating = .init(std.testing.allocator); defer aw.deinit(); - var a: Agent = .{ - .gpa = std.testing.allocator, - .arena = std.testing.allocator, - .io = undefined, - .client = undefined, - .provider = undefined, - .messages = undefined, - .sub = false, - .label = "test", - .out = &aw.writer, - }; + var a = testAgent(&aw.writer); const s = jsonSink(&a); s.emit(undefined, .{ .reasoning_delta = .{ .text = "why" } }); s.emit(undefined, .{ .text_delta = .{ .text = "hi\n" } }); @@ -295,17 +331,7 @@ test "TuiSink streams tool-arg prose raw and never ends the answer line (slice 1 defer main_mod.use_color = saved_color; var aw: Io.Writer.Allocating = .init(std.testing.allocator); defer aw.deinit(); - var a: Agent = .{ - .gpa = std.testing.allocator, - .arena = std.testing.allocator, - .io = undefined, - .client = undefined, - .provider = undefined, - .messages = undefined, - .sub = false, - .label = "test", - .out = &aw.writer, - }; + var a = testAgent(&aw.writer); const s = tuiSink(&a); s.emit(undefined, .{ .tool_arg_delta = .{ .text = "answer " } }); s.emit(undefined, .{ .tool_arg_delta = .{ .text = "prose" } }); @@ -317,17 +343,7 @@ test "TuiSink streams tool-arg prose raw and never ends the answer line (slice 1 test "TuiSink transport-abort notices reproduce the old inline lines (slice 1b)" { var aw: Io.Writer.Allocating = .init(std.testing.allocator); defer aw.deinit(); - var a: Agent = .{ - .gpa = std.testing.allocator, - .arena = std.testing.allocator, - .io = undefined, - .client = undefined, - .provider = undefined, - .messages = undefined, - .sub = false, - .label = "test", - .out = &aw.writer, - }; + var a = testAgent(&aw.writer); const s = tuiSink(&a); const cases = [_]struct { ev: engine_events.TransportAbort, want: []const u8 }{ .{ .ev = .{ .reason = .stalled, .turn_ending = false }, .want = "\n⚠ stream stalled\n" }, @@ -352,17 +368,7 @@ test "JsonSink stays silent for moments the wire never carried (slice 1b)" { defer protocol_seq.resetForTest(); var aw: Io.Writer.Allocating = .init(std.testing.allocator); defer aw.deinit(); - var a: Agent = .{ - .gpa = std.testing.allocator, - .arena = std.testing.allocator, - .io = undefined, - .client = undefined, - .provider = undefined, - .messages = undefined, - .sub = false, - .label = "test", - .out = &aw.writer, - }; + var a = testAgent(&aw.writer); const s = jsonSink(&a); s.emit(undefined, .{ .tool_arg_delta = .{ .text = "prose" } }); s.emit(undefined, .{ .transport_aborted = .{ .reason = .stalled, .turn_ending = true } }); @@ -373,20 +379,101 @@ test "JsonSink stays silent for moments the wire never carried (slice 1b)" { try std.testing.expectEqual(@as(u64, 0), protocol_seq.current()); } +test "JsonSink writes the tool bracket byte-for-byte, in wire order (slice 1c)" { + const saved_json = main_mod.json_mode; // pin: see the dispatch-order test + main_mod.json_mode = false; + defer main_mod.json_mode = saved_json; + protocol_seq.resetForTest(); + defer protocol_seq.resetForTest(); + var arena_state = std.heap.ArenaAllocator.init(std.testing.allocator); + defer arena_state.deinit(); + var aw: Io.Writer.Allocating = .init(std.testing.allocator); + defer aw.deinit(); + var a = testAgent(&aw.writer); + const s = jsonSink(&a); + + const input = try std.json.parseFromSliceLeaky(std.json.Value, arena_state.allocator(), "{\"path\":\"fixture.txt\"}", .{}); + const call: engine_events.ToolInvocation = .{ .name = "read_file", .input = input }; + const done: engine_events.ToolOutcome = .{ .name = "read_file", .text = "hi", .is_error = false, .ms = 7 }; + s.emit(undefined, .{ .tool_call_announced = call }); + s.emit(undefined, .{ .tool_call_started = call }); + s.emit(undefined, .{ .tool_result = done }); + s.emit(undefined, .{ .tool_call_finished = done }); + s.emit(undefined, .{ .tool_rejected = .{ .name = "bash", .input = input, .reason = "budget", .message = "no" } }); + try std.testing.expectEqualStrings( + "{\"seq\":1,\"type\":\"tool_call\",\"name\":\"read_file\",\"input\":{\"path\":\"fixture.txt\"}}\n" ++ + "{\"seq\":2,\"type\":\"tool_call_started\",\"name\":\"read_file\",\"input\":{\"path\":\"fixture.txt\"}}\n" ++ + "{\"seq\":3,\"type\":\"tool_result\",\"name\":\"read_file\",\"is_error\":false,\"text\":\"hi\"}\n" ++ + "{\"seq\":4,\"type\":\"tool_call_finished\",\"name\":\"read_file\",\"is_error\":false,\"ms\":7}\n" ++ + "{\"seq\":5,\"type\":\"tool_rejected\",\"name\":\"bash\",\"reason\":\"budget\",\"input\":{\"path\":\"fixture.txt\"},\"message\":\"no\"}\n", + aw.writer.buffered(), + ); +} + +test "JsonSink drops ask_user's bracket and a meta result without burning ids (slice 1c)" { + const saved_json = main_mod.json_mode; // pin: see the dispatch-order test + main_mod.json_mode = false; + defer main_mod.json_mode = saved_json; + protocol_seq.resetForTest(); + defer protocol_seq.resetForTest(); + var aw: Io.Writer.Allocating = .init(std.testing.allocator); + defer aw.deinit(); + var a = testAgent(&aw.writer); + const s = jsonSink(&a); + + const ask: engine_events.ToolInvocation = .{ .name = "ask_user", .input = .null, .ask_user = true }; + s.emit(undefined, .{ .tool_call_announced = ask }); // the wire's `ask_user` event carries this moment + s.emit(undefined, .{ .tool_call_started = ask }); + s.emit(undefined, .{ .tool_result = .{ .name = "todo_write", .text = "todos", .is_error = false, .meta = true } }); + s.emit(undefined, .{ .parallel_batch_started = .{ .count = 2 } }); // TUI-only notices + s.emit(undefined, .completion_deferred); + s.emit(undefined, .{ .completion_text = .{ .text = "done" } }); + // No line AND no id spent: the wire's numbering stays gap-free (#330). + try std.testing.expectEqualStrings("", aw.writer.buffered()); + try std.testing.expectEqual(@as(u64, 0), protocol_seq.current()); + + // ask_user's own RESULT is on the wire, though — the typed reply is it. + s.emit(undefined, .{ .tool_result = .{ .name = "ask_user", .text = "yes", .is_error = false, .meta = true, .ask_user = true } }); + try std.testing.expectEqualStrings( + "{\"seq\":1,\"type\":\"tool_result\",\"name\":\"ask_user\",\"is_error\":false,\"text\":\"yes\"}\n", + aw.writer.buffered(), + ); +} + +test "TuiSink draws the ⚙ and ✓ lines and nothing for the brackets (slice 1c)" { + const saved_color = main_mod.use_color; + main_mod.use_color = false; + defer main_mod.use_color = saved_color; + const saved_json = main_mod.json_mode; // sayText's root gate reads it + main_mod.json_mode = false; + defer main_mod.json_mode = saved_json; + const ansi = @import("ansi.zig"); + const saved_style = ansi.style; + ansi.style = .{}; // assert the text, not the palette + defer ansi.style = saved_style; + var arena_state = std.heap.ArenaAllocator.init(std.testing.allocator); + defer arena_state.deinit(); + var aw: Io.Writer.Allocating = .init(std.testing.allocator); + defer aw.deinit(); + var a = testAgent(&aw.writer); + const s = tuiSink(&a); + + const input = try std.json.parseFromSliceLeaky(std.json.Value, arena_state.allocator(), "{\"path\":\"fixture.txt\"}", .{}); + const call: engine_events.ToolInvocation = .{ .name = "read_file", .input = input }; + const done: engine_events.ToolOutcome = .{ .name = "read_file", .text = "line one\nline two\n", .is_error = false }; + s.emit(undefined, .{ .tool_call_announced = call }); + s.emit(undefined, .{ .tool_call_started = call }); // silent: the ⚙ line already said it + s.emit(undefined, .{ .tool_result = done }); + s.emit(undefined, .{ .tool_call_finished = done }); // silent + s.emit(undefined, .{ .tool_rejected = .{ .name = "bash", .input = input, .reason = "budget", .message = "no" } }); // silent + // Exactly the two lines the eval golden's tool turn shows. + try std.testing.expectEqualStrings("⚙ read_file {\"path\":\"fixture.txt\"}\n ✓ line one…\n", aw.writer.buffered()); +} + test "TuiSink renders a plain text delta exactly as the no-color TTY did" { var aw: Io.Writer.Allocating = .init(std.testing.allocator); defer aw.deinit(); - var a: Agent = .{ - .gpa = std.testing.allocator, - .arena = std.testing.allocator, - .io = undefined, - .client = undefined, - .provider = undefined, - .messages = undefined, - .sub = false, - .label = "test", - .out = &aw.writer, - }; + var a = testAgent(&aw.writer); const s = tuiSink(&a); s.emit(undefined, .{ .text_delta = .{ .text = "plain\n" } }); try std.testing.expectEqualStrings("plain\n", aw.writer.buffered()); From 53ffb49103e79d7c9c42853412e9ac6fbdaea9d0 Mon Sep 17 00:00:00 2001 From: Rach Pradhan <54503978+justrach@users.noreply.github.com> Date: Thu, 6 Aug 2026 12:39:28 +0800 Subject: [PATCH 3/4] =?UTF-8?q?feat(engine):=20agent=5Ftools=20emits=20typ?= =?UTF-8?q?ed=20events=20=E2=80=94=20the=20ansi=20import=20is=20gone=20(#4?= =?UTF-8?q?22)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The engine half keeps only what it owns: which moments happened, and the bookkeeping around them (the batch tally, argStreamedFully's dedup state, isMetaName). Every json/tty fork it used to carry is now a sink's choice, including the two suppressions that used to look symmetric but are not — the wire skips ask_user's bracket, the terminal skips prose that already streamed live. The `!self.sub` guards move into the renderer with the drawing they gated; goal_completed deliberately keeps none, matching the old call site (goalActive() is false for a subagent, so it is root-only anyway). sayToolResult keeps its out == null early-out at the emit site: that one is not presentation, it is "there is no frontend attached". One accepted behavior delta: sayToolUse's writes used to propagate and abort the batch on a failed stdout; a sink's emit returns void, so they are swallowed now, like every other converted emission since slice 1a. term.zig stays imported. Its use here is raw/nonblocking stdin for the Esc watcher, which is frontend INPUT and belongs to #430 — the same carve-out #429 already makes for agent_ws.zig. Proven by the eval harness: all 9 golden files byte identical, including the permission-prompt PTY golden whose ⚙/prompt/✓ interleaving is exactly what this touches. exec.zig, tools.zig and edit_verify.zig needed no change (every write there builds the tool RESULT, never the terminal), and agent_tool_gate.zig's lines are all prompt-block text, held for #430. Co-Authored-By: Codegraff --- src/agent_tools.zig | 124 ++++++++++++++++++++------------------------ 1 file changed, 57 insertions(+), 67 deletions(-) diff --git a/src/agent_tools.zig b/src/agent_tools.zig index c34d45e1..d567580a 100644 --- a/src/agent_tools.zig +++ b/src/agent_tools.zig @@ -21,8 +21,12 @@ const AnswerRequest = tools_mod.AnswerRequest; const answerParseError = tools_mod.answerParseError; const parseAnswerRequest = tools_mod.parseAnswerRequest; -const ansi = @import("ansi.zig"); -const style = &ansi.style; +// #422 slice 1c: every emission here leaves as a typed event; the terminal +// palette lives in agent_tool_render.zig, behind the sink. `term.zig` stays +// because raw/nonblocking stdin for the Esc watcher below is frontend INPUT, +// which belongs to the input-inversion issue (#430), not to this one. +const engine_events = @import("engine_events.zig"); +const engine_sink = @import("engine_sink.zig"); const terminal = @import("term.zig"); const tty = terminal.tty; @@ -128,8 +132,8 @@ pub fn runTools(self: *Agent, calls: []const ToolCall) ![]ExecResult { } if (ext_idx.items.len > 0) { - if (ext_idx.items.len > 1 and !self.sub) { - try self.say(" {s}↯ running {d} tools in parallel{s}\n", .{ style.dim, ext_idx.items.len, style.reset }); + if (ext_idx.items.len > 1) { + engine_sink.forAgent(self).emit(self.io, .{ .parallel_batch_started = .{ .count = ext_idx.items.len } }); } const ctx: ToolCtx = .{ .gpa = self.gpa, @@ -189,15 +193,13 @@ pub fn runTools(self: *Agent, calls: []const ToolCall) ![]ExecResult { brief_diversity.noteSiblingBatch(self.arena, self.tracer, calls, ext_idx.items, results); // #382 // #266: a cancelled parallel batch used to just look "running" and then // failed — one terminal line says what completed, failed, and cancelled. - if (ext_idx.items.len > 1 and !self.sub) { - var done: usize = 0; - var failed: usize = 0; - var cancelled: usize = 0; + if (ext_idx.items.len > 1) { + var tally: engine_events.BatchOutcome = .{ .done = 0, .failed = 0, .cancelled = 0 }; for (ext_idx.items) |i| { const r = results[i]; - if (r.cancelled) cancelled += 1 else if (r.is_error) failed += 1 else done += 1; + if (r.cancelled) tally.cancelled += 1 else if (r.is_error) tally.failed += 1 else tally.done += 1; } - try self.say(" {s}↯ parallel tools finished: {d} completed, {d} failed, {d} cancelled{s}\n", .{ style.dim, done, failed, cancelled, style.reset }); + engine_sink.forAgent(self).emit(self.io, .{ .parallel_batch_finished = tally }); } } // Show a compact ✓/✗ + preview for each non-meta call (no-op for subs). @@ -257,9 +259,16 @@ pub fn toolDedupeKey(self: *Agent, call: ToolCall) ![]const u8 { return key; } +/// A refusal that happened before the tool ran. The terminal has never drawn +/// one (the user learns of it through the model's next answer), so only a +/// wire sink gives this moment a shape. pub fn emitToolRejected(self: *Agent, call: ToolCall, reason: []const u8, message: []const u8) void { - if (!main_mod.json_mode) return; - self.emit(.{ .type = "tool_rejected", .name = call.name, .reason = reason, .input = call.input, .message = message }); + engine_sink.forAgent(self).emit(self.io, .{ .tool_rejected = .{ + .name = call.name, + .input = call.input, + .reason = reason, + .message = message, + } }); } /// #225: clock_sleep meta tool — root-only, feature-flagged (main.zig @@ -306,7 +315,7 @@ pub fn handleMeta(self: *Agent, call: ToolCall) !ExecResult { } if (try goal_state.completionGate(self.arena, self)) |refusal| { goal_state.noteCompletionRefused(self); // arm the double-check (across turns) and mark the turn as worked (#318) - if (!self.sub) try self.say("\xe2\x8f\xb8 completion deferred \xe2\x80\x94 the standing goal's checklist isn't settled\n", .{}); + engine_sink.forAgent(self).emit(self.io, .completion_deferred); return .{ .text = refusal, .is_error = true }; } const result = if (tools_mod.json_args.object(call.input)) |o| (tools_mod.json_args.str(o, "result") orelse "") else ""; @@ -315,12 +324,12 @@ pub fn handleMeta(self: *Agent, call: ToolCall) !ExecResult { // A --goal standing objective is exempt: the completion is recorded above, the steering stays, and only /goal clear|pause| retires it. if (goal_state.retireOnCompletion(self, util.unixMs(self.io))) { if (self.tracer) |t| t.note("goal", "completed via attempt_completion"); - try self.say("\xf0\x9f\x8e\xaf standing goal complete\n", .{}); + engine_sink.forAgent(self).emit(self.io, .goal_completed); } else if (goal_state.goalActive(self)) { if (self.tracer) |t| t.note("goal", "completion; standing goal retained"); } // Skip the re-print only when the result streamed live in full. - if (!self.sub and !self.argStreamedFully(call)) try self.say("{s}\n", .{result}); + if (!self.argStreamedFully(call)) engine_sink.forAgent(self).emit(self.io, .{ .completion_text = .{ .text = result } }); return .{ .text = "completion recorded", .is_error = false }; } if (std.mem.eql(u8, call.name, "eval")) { @@ -331,7 +340,7 @@ pub fn handleMeta(self: *Agent, call: ToolCall) !ExecResult { // Epoch-scoped replace, keeping omitted completed items; a write with // no usable items is rejected untouched (#318). goal_todo owns the rule. const r = try goal_todo.applyTodoWrite(self, if (tools_mod.json_args.object(call.input)) |o| o.get("todos") else null); - if (!self.sub and !r.rejected) try self.say("{s}\n", .{r.text}); + if (!r.rejected) engine_sink.forAgent(self).emit(self.io, .{ .todo_list_updated = .{ .text = r.text } }); return .{ .text = r.text, .is_error = r.rejected }; } if (std.mem.eql(u8, call.name, "clock_sleep")) { @@ -424,61 +433,42 @@ pub fn emitAskUser(self: *Agent, call_id: []const u8, question: []const u8, inpu try w.flush(); } +/// The call's announcement moment, as one pair of events: the bracket a +/// --json supervisor times against, and the ⚙ line the terminal draws for the +/// first of the two. Which of them a frontend surfaces (the wire skips +/// ask_user, the terminal skips prose that already streamed) is the sink's +/// call — engine_events.durable() decides the wire half so no sequence id is +/// ever reserved for a line the wire drops. pub fn sayToolUse(self: *Agent, call: ToolCall) !void { - if (main_mod.json_mode) { - if (std.mem.eql(u8, call.name, "ask_user")) return; - self.emit(.{ .type = "tool_call", .name = call.name, .input = call.input }); - self.emit(.{ .type = "tool_call_started", .name = call.name, .input = call.input }); - return; - } - // The ⚙ line would just repeat prose that already streamed live. - if (self.argStreamedFully(call)) return; - var aw: Io.Writer.Allocating = .init(self.gpa); - defer aw.deinit(); - var s: std.json.Stringify = .{ .writer = &aw.writer }; - try s.write(call.input); - const full = aw.writer.buffered(); - const shown = if (full.len > 160) full[0..160] else full; - try self.say("{s}⚙{s} {s}{s} {s}{s}{s}{s}\n", .{ - style.dim, style.reset, style.accent, call.name, - style.dim, shown, - if (full.len > 160) "…" else "", - style.reset, - }); + const ev: engine_events.ToolInvocation = .{ + .name = call.name, + .input = call.input, + .ask_user = std.mem.eql(u8, call.name, "ask_user"), + .arg_streamed = self.argStreamedFully(call), + }; + const sink = engine_sink.forAgent(self); + sink.emit(self.io, .{ .tool_call_announced = ev }); + sink.emit(self.io, .{ .tool_call_started = ev }); } -/// Compact result feedback for one finished tool call: a green ✓ / red ✗ -/// and a one-line preview of what it returned. Root only (subagents have -/// no writer); meta tools render their own UX, so skip them. +/// The same bracket, closing: the result a frontend shows and the finish +/// event carrying its outcome and duration. Meta tools render their own UX, +/// so both sinks drop them (ask_user excepted on the wire, where the typed +/// reply IS the result). pub fn sayToolResult(self: *Agent, name: []const u8, r: ExecResult) void { - const w = self.out orelse return; - if (main_mod.json_mode) { - if (isMetaName(name) and !std.mem.eql(u8, name, "ask_user")) return; - self.emit(.{ .type = "tool_result", .name = name, .is_error = r.is_error, .text = r.text }); - self.emit(.{ .type = "tool_call_finished", .name = name, .is_error = r.is_error, .ms = r.ms }); - return; - } - if (isMetaName(name)) return; - const all = std.mem.trim(u8, r.text, " \t\r\n"); - var preview = all; - if (std.mem.indexOfScalar(u8, preview, '\n')) |nl| preview = preview[0..nl]; - preview = std.mem.trim(u8, preview, " \t\r"); - const shown = if (preview.len > 100) preview[0..100] else preview; - const truncated = shown.len < all.len; // more content (extra lines or >100 chars) - const mark = if (r.cancelled) "⊘" else if (r.is_error) "✗" else "✓"; - const mc = if (r.cancelled) style.yellow else if (r.is_error) style.red else style.green; - var tbuf: [24]u8 = undefined; - const timing = if (main_mod.show_timing and r.ms > 0) - (std.fmt.bufPrint(&tbuf, " ({d}ms)", .{r.ms}) catch "") - else - ""; - w.print(" {s}{s}{s}{s}{s}{s} {s}{s}{s}{s}\n", .{ - mc, mark, style.reset, style.dim, timing, style.reset, - style.dim, shown, - if (truncated) "…" else "", - style.reset, - }) catch return; - w.flush() catch return; + if (self.out == null) return; // no frontend attached: nothing to tell, as ever + const ev: engine_events.ToolOutcome = .{ + .name = name, + .text = r.text, + .is_error = r.is_error, + .cancelled = r.cancelled, + .ms = r.ms, + .meta = isMetaName(name), + .ask_user = std.mem.eql(u8, name, "ask_user"), + }; + const sink = engine_sink.forAgent(self); + sink.emit(self.io, .{ .tool_result = ev }); + sink.emit(self.io, .{ .tool_call_finished = ev }); } test "parseClockSleepMs: valid ms passes through, missing/negative/non-integer reject, over-cap clamps (#225)" { From ad622e84a97855756157daee90dfc74afc65b020 Mon Sep 17 00:00:00 2001 From: Rach Pradhan <54503978+justrach@users.noreply.github.com> Date: Thu, 6 Aug 2026 13:13:08 +0800 Subject: [PATCH 4/4] fix(engine): a frontendless agent's wire sink reserves no sequence ids (#330) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit EngineSink.emit stamps BEFORE dispatch, so a durable sink handed an event it will drop burns a #330 sequence id on nothing. jsonEmit drops everything when out == null, and slice 1c moved two emitters (sayToolUse, emitToolRejected) ahead of any writer check — so in `graff --json` every subagent tool call advanced the SHARED counter by two with no wire line, and in `graff acp` (json_mode on, root.out nulled) the root did the same for a whole session, inflating the persisted event_seq. A supervisor reads that jump as lost data, which is the one thing protocol_seq promises cannot happen. Fixed where the rule can be stated once instead of re-derived per call site: jsonSink hands a NON-durable vtable to an agent with no writer. Same emitter, same (absent) output, no reservation — and no g_gui_mu round trip per pool-thread tool call either. Not fixed by guarding the two emitters, which was the obvious move and is wrong: on main sayToolUse's TUI branch went through say(), whose out == null path is the pool-thread WORKER LINE, so an early return there would have deleted every subagent's ⚙ line from the terminal. sayToolResult's own guard stays: main had it too. Also, per review of the slice: the four `!self.sub` gates go back to their emit sites. Who may announce a fan-out or a meta notice is engine policy about who owns the terminal, not a drawing decision; keeping it in agent_tool_render meant the renderer back-read Agent state (widening the debt engine_sink.zig's header says to shrink) and, worse, meant a future serve/attach sink would start receiving subagent tallies the engine never used to produce, with no field in the payload to tell them apart. Two test tightenings from the same review: the cap assertions now pin the literals 160/100 the pre-#422 inline path spelled out rather than the new constants (a test that compares the code to itself cannot guard a conversion), and the 100-byte result cap gets its first exercise. Proven: new engine_sink test asserts protocol_seq.current() == 0 after a full tool bracket + rejection + text delta on a writerless agent (it read 6 before). 1003/1003 tests pass and all 9 eval goldens are byte identical. Co-Authored-By: Codegraff --- src/agent_tool_render.zig | 36 ++++++++++++++---------- src/agent_tools.zig | 19 ++++++++----- src/engine_sink.zig | 59 ++++++++++++++++++++++++++++++++++----- 3 files changed, 86 insertions(+), 28 deletions(-) diff --git a/src/agent_tool_render.zig b/src/agent_tool_render.zig index f602d038..1c651876 100644 --- a/src/agent_tool_render.zig +++ b/src/agent_tool_render.zig @@ -12,6 +12,11 @@ //! - write errors are swallowed rather than propagated, because a sink's emit //! path returns void. A caller that used to abort its batch on a broken //! stdout now continues into the next failing write instead. +//! +//! WHO may announce a moment stays engine policy: the `!self.sub` gates on the +//! batch tallies and the meta notices remain at their emit sites, so nothing +//! here back-reads Agent policy to decide whether to draw. The only Agent +//! reads below are drawing handles — the writer and the allocator. const std = @import("std"); const Io = std.Io; @@ -21,7 +26,7 @@ const agent_mod = @import("agent.zig"); const Agent = agent_mod.Agent; const agent_output = @import("agent_output.zig"); const sayText = agent_output.sayText; -const util = @import("util.zig"); // tests only: repeatBytes for the arg-cap case +const util = @import("util.zig"); // tests only: repeatBytes for the two cap cases const ansi = @import("ansi.zig"); const style = &ansi.style; @@ -92,14 +97,12 @@ pub fn toolResultLine(a: *Agent, r: ToolOutcome) void { const notice_buf_bytes: usize = 192; pub fn parallelBatchStarted(a: *Agent, count: usize) void { - if (a.sub) return; // a child's fan-out is the root's line to draw, not its own var buf: [notice_buf_bytes]u8 = undefined; const line = std.fmt.bufPrint(&buf, " {s}↯ running {d} tools in parallel{s}\n", .{ style.dim, count, style.reset }) catch return; sayText(a, line); } pub fn parallelBatchFinished(a: *Agent, o: BatchOutcome) void { - if (a.sub) return; var buf: [notice_buf_bytes]u8 = undefined; const line = std.fmt.bufPrint(&buf, " {s}↯ parallel tools finished: {d} completed, {d} failed, {d} cancelled{s}\n", .{ style.dim, o.done, o.failed, o.cancelled, style.reset }) catch return; sayText(a, line); @@ -108,7 +111,6 @@ pub fn parallelBatchFinished(a: *Agent, o: BatchOutcome) void { /// #318: the checklist isn't settled, so the completion parks. The escapes are /// the literal bytes the old call site spelled out (⏸, 🎯 below). pub fn completionDeferred(a: *Agent) void { - if (a.sub) return; sayText(a, "\xe2\x8f\xb8 completion deferred \xe2\x80\x94 the standing goal's checklist isn't settled\n"); } @@ -118,7 +120,6 @@ pub fn goalCompleted(a: *Agent) void { /// A meta tool's own user-facing text, one line, exactly as it came. pub fn toolTextLine(a: *Agent, text: []const u8) void { - if (a.sub) return; var line: Io.Writer.Allocating = .init(a.gpa); defer line.deinit(); line.writer.print("{s}\n", .{text}) catch return; @@ -169,7 +170,9 @@ test "the ⚙ line reproduces the old inline announcement, cap and ellipsis incl const line = aw.writer.buffered(); try std.testing.expect(std.mem.startsWith(u8, line, "⚙ codedb {\"q\":\"xxx")); try std.testing.expect(std.mem.endsWith(u8, line, "…\n")); - try std.testing.expectEqual(arg_preview_bytes, line.len - "⚙ codedb ".len - "…\n".len); + // The literal, not arg_preview_bytes: this asserts the conversion kept the + // pre-#422 inline cap, and asserting the constant against itself would not. + try std.testing.expectEqual(@as(usize, 160), line.len - "⚙ codedb ".len - "…\n".len); } test "the result line marks success, failure and cancellation and previews one line" { @@ -199,6 +202,15 @@ test "the result line marks success, failure and cancellation and previews one l toolResultLine(&a, .{ .name = "todo_write", .text = "todos", .is_error = false, .meta = true }); try std.testing.expectEqualStrings("", aw.writer.buffered()); + // Over the cap: exactly 100 bytes of the first line, then the ellipsis — + // the literal the pre-#422 inline path spelled out, not result_preview_bytes. + aw.clearRetainingCapacity(); + const long = util.repeatBytes("z", 250); + toolResultLine(&a, .{ .name = "bash", .text = &long, .is_error = false }); + const rline = aw.writer.buffered(); + try std.testing.expect(std.mem.endsWith(u8, rline, "…\n")); + try std.testing.expectEqual(@as(usize, 100), rline.len - " ✓ ".len - "…\n".len); + // --timing adds the measured duration between the mark and the preview. aw.clearRetainingCapacity(); main_mod.show_timing = true; @@ -236,12 +248,8 @@ test "batch tallies and meta notices render the old wording verbatim" { toolTextLine(&a, "completion recorded"); try std.testing.expectEqualStrings("completion recorded\n", aw.writer.buffered()); - // A subagent announces none of the root's batch/goal lines. - aw.clearRetainingCapacity(); - a.sub = true; - parallelBatchStarted(&a, 3); - parallelBatchFinished(&a, .{ .done = 1, .failed = 0, .cancelled = 0 }); - completionDeferred(&a); - toolTextLine(&a, "x"); - try std.testing.expectEqualStrings("", aw.writer.buffered()); + // Who may announce these is engine policy, not a drawing decision: the + // `!self.sub` gates live at the emit sites (agent_tools.runTools / + // handleMeta), so a subagent never produces the moment at all and this + // file needs no read into the Agent to suppress it. } diff --git a/src/agent_tools.zig b/src/agent_tools.zig index d567580a..aa7f07a0 100644 --- a/src/agent_tools.zig +++ b/src/agent_tools.zig @@ -132,7 +132,11 @@ pub fn runTools(self: *Agent, calls: []const ToolCall) ![]ExecResult { } if (ext_idx.items.len > 0) { - if (ext_idx.items.len > 1) { + // A child's fan-out is the root's to announce, so the moment is not + // produced at all for a subagent: engine policy about who owns the + // terminal, kept at the emit site rather than re-derived by a sink + // that would need to back-read the Agent to know (#422 slice-1 rule). + if (ext_idx.items.len > 1 and !self.sub) { engine_sink.forAgent(self).emit(self.io, .{ .parallel_batch_started = .{ .count = ext_idx.items.len } }); } const ctx: ToolCtx = .{ @@ -193,7 +197,7 @@ pub fn runTools(self: *Agent, calls: []const ToolCall) ![]ExecResult { brief_diversity.noteSiblingBatch(self.arena, self.tracer, calls, ext_idx.items, results); // #382 // #266: a cancelled parallel batch used to just look "running" and then // failed — one terminal line says what completed, failed, and cancelled. - if (ext_idx.items.len > 1) { + if (ext_idx.items.len > 1 and !self.sub) { // root's line to draw, as above var tally: engine_events.BatchOutcome = .{ .done = 0, .failed = 0, .cancelled = 0 }; for (ext_idx.items) |i| { const r = results[i]; @@ -315,7 +319,7 @@ pub fn handleMeta(self: *Agent, call: ToolCall) !ExecResult { } if (try goal_state.completionGate(self.arena, self)) |refusal| { goal_state.noteCompletionRefused(self); // arm the double-check (across turns) and mark the turn as worked (#318) - engine_sink.forAgent(self).emit(self.io, .completion_deferred); + if (!self.sub) engine_sink.forAgent(self).emit(self.io, .completion_deferred); // root-only notice, as ever return .{ .text = refusal, .is_error = true }; } const result = if (tools_mod.json_args.object(call.input)) |o| (tools_mod.json_args.str(o, "result") orelse "") else ""; @@ -329,7 +333,7 @@ pub fn handleMeta(self: *Agent, call: ToolCall) !ExecResult { if (self.tracer) |t| t.note("goal", "completion; standing goal retained"); } // Skip the re-print only when the result streamed live in full. - if (!self.argStreamedFully(call)) engine_sink.forAgent(self).emit(self.io, .{ .completion_text = .{ .text = result } }); + if (!self.sub and !self.argStreamedFully(call)) engine_sink.forAgent(self).emit(self.io, .{ .completion_text = .{ .text = result } }); return .{ .text = "completion recorded", .is_error = false }; } if (std.mem.eql(u8, call.name, "eval")) { @@ -340,7 +344,7 @@ pub fn handleMeta(self: *Agent, call: ToolCall) !ExecResult { // Epoch-scoped replace, keeping omitted completed items; a write with // no usable items is rejected untouched (#318). goal_todo owns the rule. const r = try goal_todo.applyTodoWrite(self, if (tools_mod.json_args.object(call.input)) |o| o.get("todos") else null); - if (!r.rejected) engine_sink.forAgent(self).emit(self.io, .{ .todo_list_updated = .{ .text = r.text } }); + if (!self.sub and !r.rejected) engine_sink.forAgent(self).emit(self.io, .{ .todo_list_updated = .{ .text = r.text } }); return .{ .text = r.text, .is_error = r.rejected }; } if (std.mem.eql(u8, call.name, "clock_sleep")) { @@ -437,8 +441,9 @@ pub fn emitAskUser(self: *Agent, call_id: []const u8, question: []const u8, inpu /// --json supervisor times against, and the ⚙ line the terminal draws for the /// first of the two. Which of them a frontend surfaces (the wire skips /// ask_user, the terminal skips prose that already streamed) is the sink's -/// call — engine_events.durable() decides the wire half so no sequence id is -/// ever reserved for a line the wire drops. +/// call — engine_events.durable() decides the wire half, and jsonSink is +/// non-durable for an agent with no writer, so no sequence id is ever +/// reserved for a line the wire drops (#330). pub fn sayToolUse(self: *Agent, call: ToolCall) !void { const ev: engine_events.ToolInvocation = .{ .name = call.name, diff --git a/src/engine_sink.zig b/src/engine_sink.zig index 6ed6131d..da07f6ef 100644 --- a/src/engine_sink.zig +++ b/src/engine_sink.zig @@ -14,7 +14,12 @@ //! - JsonSink: the existing --json wire lines for these events, //! byte-identical to the old inline emits. The ONE translation point from //! internal type to wire shape: any change here is externally visible and -//! gated behind a schema_version bump. +//! gated behind a schema_version bump. Not yet the ONLY producer of those +//! shapes, though: subagent_run.zig's guiEmit writes `tool_call`/ +//! `tool_result` rows with an extra `id` (schema_protocol.zig) straight to +//! stdout, bypassing this sink. Routing them through it needs an optional +//! `id` on ToolInvocation/ToolOutcome — decide that when #429 takes the +//! subagent cluster, while these structs are still cheap to change. //! //! Events are stamped with a {generation, sequence} Cursor at the dispatch //! boundary. In --json mode (the only durable sink today), durable events @@ -61,11 +66,11 @@ pub const EngineSink = struct { /// The emission boundary: stamp, then hand off. `io` backs the stdout /// lock, taken only when a durable sink reserves AND json_mode is on — /// a global, so `io` may be passed undefined only where json_mode is - /// known false (TUI dispatch; the tests here pin it). Durable emitters - /// must not emit durable events a sink will drop (jsonEmit returns on - /// out == null): the reserved id would burn with no wire line, opening - /// a seq gap (#330). Today printDelta guarantees that by returning - /// early when out == null, before any durable emission. + /// known false (TUI dispatch; the tests here pin it). A durable sink must + /// never be handed an event it will drop: the reserved id would burn with + /// no wire line, opening a seq gap (#330). jsonSink enforces that at + /// construction — an agent with no writer gets a non-durable vtable — so + /// emitters need no `out == null` guard of their own. pub fn emit(self: EngineSink, io: Io, ev: EngineEvent) void { const reserve = self.vt.durable and engine_events.durable(ev); if (reserve and main_mod.json_mode) { @@ -91,12 +96,21 @@ pub fn tuiSink(a: *Agent) EngineSink { return .{ .ctx = a, .vt = &tui_vtable }; } +/// The --json wire, for an agent that HAS one. A frontendless agent — a +/// pool-thread subagent (subagent_run.zig builds every child with +/// `.out = null`) or the ACP root (acp.zig nulls it while json_mode is on) — +/// gets the same emitter but a NON-durable vtable: jsonEmit drops every line +/// for it, and a durable sink would have reserved a sequence id for each of +/// those dropped lines, opening exactly the gap #330 promises cannot exist. +/// Structural, so no emitter has to re-derive the rule per call site. pub fn jsonSink(a: *Agent) EngineSink { - return .{ .ctx = a, .vt = &json_vtable }; + return .{ .ctx = a, .vt = if (a.out == null) &json_dropped_vtable else &json_vtable }; } const tui_vtable: VTable = .{ .emit = tuiEmit, .durable = false }; const json_vtable: VTable = .{ .emit = jsonEmit, .durable = true }; +/// Same writer, nothing to write to: see jsonSink. +const json_dropped_vtable: VTable = .{ .emit = jsonEmit, .durable = false }; /// Today's interactive rendering, relocated behind the event contract. Every /// branch is the old inline agent_stream.zig code path, gate for gate. @@ -440,6 +454,37 @@ test "JsonSink drops ask_user's bracket and a meta result without burning ids (s ); } +test "a frontendless agent's wire sink burns no ids for the lines it cannot write (#330)" { + const saved_json = main_mod.json_mode; // pin: see the dispatch-order test + main_mod.json_mode = false; + defer main_mod.json_mode = saved_json; + protocol_seq.resetForTest(); + defer protocol_seq.resetForTest(); + var arena_state = std.heap.ArenaAllocator.init(std.testing.allocator); + defer arena_state.deinit(); + var aw: Io.Writer.Allocating = .init(std.testing.allocator); + defer aw.deinit(); + var a = testAgent(&aw.writer); + a.out = null; // a pool-thread subagent, or the ACP root: nowhere to write + const s = jsonSink(&a); + try std.testing.expect(!s.vt.durable); // the guarantee is structural, not per-call-site + + const input = try std.json.parseFromSliceLeaky(std.json.Value, arena_state.allocator(), "{\"path\":\"f\"}", .{}); + const call: engine_events.ToolInvocation = .{ .name = "read_file", .input = input }; + const done: engine_events.ToolOutcome = .{ .name = "read_file", .text = "hi", .is_error = false }; + s.emit(undefined, .{ .tool_call_announced = call }); + s.emit(undefined, .{ .tool_call_started = call }); + s.emit(undefined, .{ .tool_result = done }); + s.emit(undefined, .{ .tool_call_finished = done }); + s.emit(undefined, .{ .tool_rejected = .{ .name = "bash", .input = input, .reason = "budget", .message = "no" } }); + s.emit(undefined, .{ .text_delta = .{ .text = "hi" } }); + // Every one of those would have been a wire line for a rooted agent. With + // no writer there is no line, so there must be no id either: a supervisor + // reads a gap as lost data. + try std.testing.expectEqualStrings("", aw.writer.buffered()); + try std.testing.expectEqual(@as(u64, 0), protocol_seq.current()); +} + test "TuiSink draws the ⚙ and ✓ lines and nothing for the brackets (slice 1c)" { const saved_color = main_mod.use_color; main_mod.use_color = false;