diff --git a/src/agent.zig b/src/agent.zig
index 648f1ad4b..2693a4c64 100644
--- a/src/agent.zig
+++ b/src/agent.zig
@@ -23,8 +23,6 @@ const no_local_tools = @import("no_local_tools.zig"); // #330: --no-local-tools
 const models_cache = @import("models_cache.zig");
 const keys_cli = @import("keys_cli.zig");
 const run_budget_mod = @import("run_budget.zig");
-const protocol_seq = @import("protocol_seq.zig"); // #330: monotonic `seq` on every --json event
-const tick_gate = @import("tick_gate.zig"); // #tui-tick: child ticks wait for a foreground line boundary
 
 // prompt_ui (agent_prompt.zig) owns the width-budgeted status line (#209,
 // 600-line goal); prompt() is member-aliased back onto Agent below.
@@ -96,6 +94,9 @@ pub const Agent = struct {
     label: []const u8,
     out: ?*Io.Writer,
     in: ?*Io.Reader = null, // stdin, root only — backs the ask_user tool
+    /// Frontend event sink (#422). Null = resolved per emission from the
+    /// process mode (engine_sink.forAgent); set to inject a custom frontend.
+    sink: ?@import("engine_sink.zig").EngineSink = null,
     registry: ?*mcp.Registry = null,
     approvals: ?*approvals_mod.Approvals = null, // shared bash-approval state, set by main()
     tracer: ?*trace.Tracer = null, // shared JSONL event trace, set by main()
@@ -206,77 +207,13 @@ pub const Agent = struct {
     // of Agent's own methods.
     pub const prompt = prompt_ui.prompt;
 
-    pub fn say(self: *Agent, comptime fmt: []const u8, args: anytype) !void {
-        // stdout is a strict JSONL transport in --json mode. Human-facing
-        // notices are represented by their structured terminal/error events;
-        // never leak an unframed line that breaks SDK parsers.
-        if (main_mod.json_mode and !self.sub) return;
-        if (self.out) |w| {
-            try w.print(fmt, args);
-            try w.flush();
-            // The root just ended a row: anything a child offered mid-stream
-            // may land now (#tui-tick).
-            if (!self.sub and !main_mod.json_mode and comptime endsLine(fmt)) _ = tick_gate.setLineStart(true);
-        } else {
-            // A pool-thread child has no writer: its activity line goes to
-            // stderr THROUGH the gate, so it lands at a line boundary the root
-            // has published rather than mid-row (#tui-tick).
-            var buf: [tick_gate.slot_bytes]u8 = undefined;
-            var sink = Io.Writer.fixed(&buf);
-            const fit = if (sink.print("  [{s}] " ++ fmt, .{self.label} ++ args)) |_| true else |_| false;
-            var line = sink.buffered();
-            // Over-long (an uncapped provider error) means the fixed sink cut
-            // the text and ate the trailing newline. The gate cannot repair
-            // that — the cut exactly fills a slot, so its own guard never fires
-            // — and a line that does not end its row splices the next worker
-            // line onto it mid-column, which is the reported artifact. End it.
-            if (!fit or line.len == 0 or line[line.len - 1] != '\n') {
-                // usize, not @min's narrowed comptime-derived type: at + 1 == buf.len.
-                const at: usize = @min(line.len, buf.len - 1); // append, or overwrite the last byte
-                buf[at] = '\n';
-                line = buf[0 .. at + 1];
-            }
-            tick_gate.workerLine(line);
-        }
-    }
-
-    /// Comptime: does this say() format end a terminal row?
-    fn endsLine(comptime fmt: []const u8) bool {
-        return fmt.len > 0 and fmt[fmt.len - 1] == '\n';
-    }
+    // say()/sayApiError() human-facing lines and emit() — the structured
+    // --json JSONL writer — live in agent_output.zig (#422, 600-line cap).
+    // Member-aliased so `self.say(...)`/`self.emit(...)` resolve unchanged.
+    pub const say = @import("agent_output.zig").say;
+    pub const sayApiError = @import("agent_output.zig").sayApiError;
+    pub const emit = @import("agent_output.zig").emit;
 
-    /// 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 {
-        self.last_api_error = std.fmt.allocPrint(self.arena, fmt, args) catch null;
-        if (self.last_api_error) |m| if (@import("retry_hint.zig").humanizeRetrySeconds(m)) |h| return self.say("{s} (~{s})\n", .{ m, h.buf[0..h.len] });
-        try self.say(fmt ++ "\n", args);
-    }
-
-    /// Emit one structured JSONL event to stdout (--json mode). `ev` is any
-    /// struct/anonymous struct; field names become JSON keys (a std.json.Value
-    /// field, e.g. tool input, serializes correctly). Best-effort.
-    ///
-    /// #330: in --json mode the event is stamped with a monotonic `seq` so a
-    /// supervisor that loses the stream can say exactly where it stopped. The
-    /// counter is bumped inside the same lock that serializes stdout, which is
-    /// what makes the sequence gap-free rather than merely increasing.
-    pub fn emit(self: *Agent, ev: anytype) void {
-        const w = self.out orelse return;
-        // --json: the GUI stream is shared with pool-thread subagent emits
-        // (guiEmit), so serialize + flush under the lock — a raw line must never
-        // land mid-buffer and two writers must never interleave on stdout.
-        if (main_mod.json_mode) main_mod.g_gui_mu.lockUncancelable(self.io);
-        defer if (main_mod.json_mode) main_mod.g_gui_mu.unlock(self.io);
-        if (main_mod.json_mode) {
-            protocol_seq.writeEvent(w, ev) catch return;
-        } else {
-            var s: std.json.Stringify = .{ .writer = w };
-            s.write(ev) catch return;
-        }
-        w.writeByte('\n') catch return;
-        w.flush() catch return;
-    }
     pub fn systemPrompt(self: *const Agent) []const u8 {
         if (self.review_mode) return self.sys_override orelse self.sys_normal;
         if (self.sub) return self.sys_override orelse prompts.sub_system_prompt;
@@ -493,12 +430,12 @@ pub const Agent = struct {
     // spirit from arpagon/pi-animations, MIT), persists in settings.json.
     pub var g_spin_stop: std.atomic.Value(bool) = .init(true);
     pub var g_spin_future: ?Io.Future(void) = null;
-    pub const spinnerTask = @import("agent_stream.zig").spinnerTask;
-    pub const spinnerStart = @import("agent_stream.zig").spinnerStart;
-    pub const spinnerStop = @import("agent_stream.zig").spinnerStop;
-    pub const streamThinking = @import("agent_stream.zig").streamThinking;
-    pub const closeThinkingBlock = @import("agent_stream.zig").closeThinkingBlock;
-    pub const toggleThinkingFold = @import("agent_stream.zig").toggleThinkingFold;
+    pub const spinnerTask = @import("agent_stream_render.zig").spinnerTask;
+    pub const spinnerStart = @import("agent_stream_render.zig").spinnerStart;
+    pub const spinnerStop = @import("agent_stream_render.zig").spinnerStop;
+    pub const streamThinking = @import("agent_stream_render.zig").streamThinking;
+    pub const closeThinkingBlock = @import("agent_stream_render.zig").closeThinkingBlock;
+    pub const toggleThinkingFold = @import("agent_stream_render.zig").toggleThinkingFold;
     pub const postStream = @import("agent_stream.zig").postStream;
     pub const postStreamWithClient = @import("agent_stream.zig").postStreamWithClient;
     pub const printDelta = @import("agent_stream.zig").printDelta;
@@ -538,6 +475,7 @@ pub const Agent = struct {
     pub const drainSteerStdin = @import("agent_interrupt.zig").drainSteerStdin;
     pub const drainStdin = @import("agent_interrupt.zig").drainStdin;
     pub const rawNonblockStdin = @import("agent_interrupt.zig").rawNonblockStdin;
+    pub const restoreStdin = @import("agent_interrupt.zig").restoreStdin;
     pub const sleepInterruptible = @import("agent_interrupt.zig").sleepInterruptible;
     pub const ssePayload = @import("agent_interrupt.zig").ssePayload;
     pub const sseIndex = @import("agent_interrupt.zig").sseIndex;
diff --git a/src/agent_interrupt.zig b/src/agent_interrupt.zig
index 53f568664..9d406bdfb 100644
--- a/src/agent_interrupt.zig
+++ b/src/agent_interrupt.zig
@@ -206,6 +206,12 @@ pub fn rawNonblockStdin() ?tty.RawState {
     return tty.enterRaw(false);
 }
 
+/// Undo rawNonblockStdin. Lives here so the transport loop needs no terminal
+/// import of its own (#422: engine files never import term.zig).
+pub fn restoreStdin(orig: tty.RawState) void {
+    tty.restore(orig);
+}
+
 /// Sleep `ms` watching stdin for Esc (when the root is on a TTY), so the
 /// user can cancel a retry backoff instead of waiting it out.
 pub fn sleepInterruptible(self: *Agent, ms: u64) error{Interrupted}!void {
diff --git a/src/agent_output.zig b/src/agent_output.zig
new file mode 100644
index 000000000..9f9263f86
--- /dev/null
+++ b/src/agent_output.zig
@@ -0,0 +1,87 @@
+//! Root-agent output plumbing, split off agent.zig (600-line cap, #422):
+//! say() — human-facing lines routed by mode/thread, sayApiError() — the
+//! remembered API-error notice, and emit() — the single structured JSONL
+//! writer for --json mode. Member-aliased back onto Agent so call sites
+//! (`self.say(...)`, `self.emit(...)`) resolve unchanged.
+
+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 protocol_seq = @import("protocol_seq.zig"); // #330: monotonic `seq` on every --json event
+const tick_gate = @import("tick_gate.zig"); // #tui-tick: child ticks wait for a foreground line boundary
+
+pub fn say(self: *Agent, comptime fmt: []const u8, args: anytype) !void {
+    // stdout is a strict JSONL transport in --json mode. Human-facing
+    // notices are represented by their structured terminal/error events;
+    // never leak an unframed line that breaks SDK parsers.
+    if (main_mod.json_mode and !self.sub) return;
+    if (self.out) |w| {
+        try w.print(fmt, args);
+        try w.flush();
+        // The root just ended a row: anything a child offered mid-stream
+        // may land now (#tui-tick).
+        if (!self.sub and !main_mod.json_mode and comptime endsLine(fmt)) _ = tick_gate.setLineStart(true);
+    } else {
+        // A pool-thread child has no writer: its activity line goes to
+        // stderr THROUGH the gate, so it lands at a line boundary the root
+        // has published rather than mid-row (#tui-tick).
+        var buf: [tick_gate.slot_bytes]u8 = undefined;
+        var sink = Io.Writer.fixed(&buf);
+        const fit = if (sink.print("  [{s}] " ++ fmt, .{self.label} ++ args)) |_| true else |_| false;
+        var line = sink.buffered();
+        // Over-long (an uncapped provider error) means the fixed sink cut
+        // the text and ate the trailing newline. The gate cannot repair
+        // that — the cut exactly fills a slot, so its own guard never fires
+        // — and a line that does not end its row splices the next worker
+        // line onto it mid-column, which is the reported artifact. End it.
+        if (!fit or line.len == 0 or line[line.len - 1] != '\n') {
+            // usize, not @min's narrowed comptime-derived type: at + 1 == buf.len.
+            const at: usize = @min(line.len, buf.len - 1); // append, or overwrite the last byte
+            buf[at] = '\n';
+            line = buf[0 .. at + 1];
+        }
+        tick_gate.workerLine(line);
+    }
+}
+
+/// Comptime: does this say() format end a terminal row?
+fn endsLine(comptime fmt: []const u8) bool {
+    return fmt.len > 0 and fmt[fmt.len - 1] == '\n';
+}
+
+/// 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 {
+    self.last_api_error = std.fmt.allocPrint(self.arena, fmt, args) catch null;
+    if (self.last_api_error) |m| if (@import("retry_hint.zig").humanizeRetrySeconds(m)) |h| return self.say("{s} (~{s})\n", .{ m, h.buf[0..h.len] });
+    try self.say(fmt ++ "\n", args);
+}
+
+/// Emit one structured JSONL event to stdout (--json mode). `ev` is any
+/// struct/anonymous struct; field names become JSON keys (a std.json.Value
+/// field, e.g. tool input, serializes correctly). Best-effort.
+///
+/// #330: in --json mode the event is stamped with a monotonic `seq` so a
+/// supervisor that loses the stream can say exactly where it stopped. The
+/// counter is bumped inside the same lock that serializes stdout, which is
+/// what makes the sequence gap-free rather than merely increasing.
+pub fn emit(self: *Agent, ev: anytype) void {
+    const w = self.out orelse return;
+    // --json: the GUI stream is shared with pool-thread subagent emits
+    // (guiEmit), so serialize + flush under the lock — a raw line must never
+    // land mid-buffer and two writers must never interleave on stdout.
+    if (main_mod.json_mode) main_mod.g_gui_mu.lockUncancelable(self.io);
+    defer if (main_mod.json_mode) main_mod.g_gui_mu.unlock(self.io);
+    if (main_mod.json_mode) {
+        protocol_seq.writeEvent(w, ev) catch return;
+    } else {
+        var s: std.json.Stringify = .{ .writer = w };
+        s.write(ev) catch return;
+    }
+    w.writeByte('\n') catch return;
+    w.flush() catch return;
+}
diff --git a/src/agent_stream.zig b/src/agent_stream.zig
index 7bc5eebb7..5d579718e 100644
--- a/src/agent_stream.zig
+++ b/src/agent_stream.zig
@@ -1,16 +1,13 @@
-//! The live streaming path: the thinking spinner (spinnerTask/Start/Stop,
-//! an animated indicator while the model is silent), the live dimmed
-//! "Thinking" reasoning block (streamThinking/closeThinkingBlock/
-//! toggleThinkingFold), and postStream itself — the root agent's
-//! streaming POST, racing send/receive/read against stall watchdogs,
-//! printing text deltas (printDelta) as they arrive. The highest-
+//! The live streaming path: postStream — the root agent's streaming POST,
+//! racing send/receive/read against stall watchdogs — and printDelta, which
+//! turns SSE lines into the semantic deltas a frontend renders. The highest-
 //! entanglement piece of the Agent struct (#123, 600-line goal); extracted
 //! last, after agent_request/agent_steps/agent_argstream/agent_render/
 //! agent_interrupt so it can sibling-import them directly.
 //!
-//! Agent.g_spin_stop/Agent.g_spin_future are struct-level `pub var`s that stay
-//! declared directly inside the Agent struct in main.zig (never alias a
-//! `var`) — reached here as `Agent.Agent.g_spin_stop`/`Agent.Agent.g_spin_future`.
+//! #422: this file draws nothing. The spinner and the live "Thinking" block
+//! live in agent_stream_render.zig (reached through Agent member aliases);
+//! term.zig/agent_render.zig/ansi.zig/anim.zig must never be imported here.
 
 const std = @import("std");
 const Io = std.Io;
@@ -20,16 +17,7 @@ const main_mod = @import("main.zig");
 const agent_mod = @import("agent.zig");
 const Agent = agent_mod.Agent;
 
-const style = &@import("ansi.zig").style;
-
-const anim = @import("anim.zig");
-const tick_gate = @import("tick_gate.zig"); // #tui-tick: child ticks wait for a foreground line boundary
-
-const terminal = @import("term.zig");
-const tty = terminal.tty;
-const termCols = terminal.termCols;
-const termRows = terminal.termRows;
-const advanceThinkingRows = terminal.advanceThinkingRows;
+const engine_sink = @import("engine_sink.zig"); // #422: every emission goes through the sink
 
 const reasoningDelta = @import("title.zig").reasoningDelta;
 const stream_tests = @import("agent_stream_tests.zig");
@@ -45,134 +33,14 @@ const streamLineTask = http.streamLineTask;
 const streamStallWatch = http.streamStallWatch;
 const watchdogError = http.watchdogError;
 
-// escPressed/drainSteerStdin/rawNonblockStdin/ssePayload live in
+// escPressed/drainSteerStdin/rawNonblockStdin/restoreStdin/ssePayload live in
 // agent_interrupt.zig; reached through the Agent struct's member aliases.
 const escPressed = Agent.escPressed;
 const drainSteerStdin = Agent.drainSteerStdin;
 const rawNonblockStdin = Agent.rawNonblockStdin;
+const restoreStdin = Agent.restoreStdin;
 const ssePayload = Agent.ssePayload;
 
-pub fn spinnerTask(io: Io) void {
-    var i: usize = 0;
-    var buf: [512]u8 = undefined;
-    var w = Io.File.stdout().writer(io, &buf);
-    while (!Agent.g_spin_stop.load(.acquire)) {
-        if (main_mod.g_steer_visible.load(.acquire)) {
-            io.sleep(.fromMilliseconds(20), .awake) catch break;
-            continue;
-        }
-        // Clear-then-draw each frame: animations may vary in width.
-        w.interface.writeAll("\r\x1b[2K\x1b[?7l") catch return; // ?7l: autowrap off so a wide spinner truncates instead of wrapping in a narrow window (the "goes on and on" bug)
-        anim.anims[anim.g_anim_current].frame(&w.interface, i) catch return;
-        w.interface.writeAll("\x1b[?7h") catch return; // restore autowrap
-        w.interface.flush() catch return;
-        i += 1;
-        const frame_ticks = @max(@as(usize, 1), @as(usize, anim.anims[anim.g_anim_current].frame_ms) / 20);
-        var t: usize = 0;
-        while (t < frame_ticks and !Agent.g_spin_stop.load(.acquire)) : (t += 1) {
-            if (main_mod.g_steer_visible.load(.acquire)) break;
-            io.sleep(.fromMilliseconds(20), .awake) catch break;
-        }
-    }
-    if (!main_mod.g_steer_visible.load(.acquire)) {
-        w.interface.writeAll("\x1b[?7h\r\x1b[2K") catch return; // restore autowrap + clear
-        w.interface.flush() catch {};
-    }
-}
-
-pub fn spinnerStart(self: *Agent) void {
-    if (self.sub or main_mod.json_mode or !main_mod.use_color or self.out == null) return;
-    if (anim.g_anim_off) return;
-    if (Agent.g_spin_future != null) return;
-    anim.selectSpinner(self.io);
-    Agent.g_spin_stop.store(false, .release);
-    Agent.g_spin_future = self.io.concurrent(spinnerTask, .{self.io}) catch blk: {
-        Agent.g_spin_stop.store(true, .release); // no spare concurrency: skip quietly
-        break :blk null;
-    };
-}
-
-pub fn spinnerStop(self: *Agent) void {
-    if (self.sub) return; // root-only state — subs run on pool threads
-    if (Agent.g_spin_future) |*f| {
-        Agent.g_spin_stop.store(true, .release);
-        f.await(self.io);
-        Agent.g_spin_future = null;
-    }
-}
-
-/// Stream a chunk of the model's reasoning into a live, dimmed "Thinking"
-/// block in the terminal, opening the block (and handing the line off from
-/// the spinner) on the first chunk. Gated by /thinking; when off the block is
-/// never opened and the spinner stands in for it. We track the block's
-/// on-screen height as it streams so closeThinkingBlock can collapse it to a
-/// one-line summary when the answer starts (#75).
-pub fn streamThinking(self: *Agent, chunk: []const u8) void {
-    const w = self.out orelse return;
-    if (!self.thinking_open) {
-        self.spinnerStop();
-        w.print("{s}▼ Thinking{s}\n{s}", .{ style.dim, style.reset, style.dim }) catch return;
-        self.thinking_open = true;
-        main_mod.g_thinking_open = true;
-        self.thinking_rows = 1; // the header newline already moved us down one line
-        self.thinking_col = 0;
-        self.thinking_overflow = false;
-        // The block owns every row below this one and collapses them by cursor
-        // math — a child's tick printed inside it would be erased with the
-        // block (or shift the erase onto real output). Hold until it closes.
-        tick_gate.hold();
-    }
-    self.thinking_text.appendSlice(self.gpa, chunk) catch {};
-    if (self.thinking_folded) return; // folded: buffer only, don't draw the live block
-    w.writeAll(chunk) catch return;
-    w.flush() catch return;
-    advanceThinkingRows(&self.thinking_rows, &self.thinking_col, termCols(), chunk);
-    if (self.thinking_rows + 1 >= termRows()) self.thinking_overflow = true;
-}
-
-/// Close an open "Thinking" block. If it still fits on screen, collapse it in
-/// place to a one-line "Thought" summary (#75); if it has scrolled off
-/// (overflow) leave the reasoning and just append the summary, so we never
-/// erase the user's earlier output. Runs on the reasoning->answer transition
-/// and at stream end.
-pub fn closeThinkingBlock(self: *Agent) void {
-    if (!self.thinking_open) return;
-    self.thinking_open = false;
-    main_mod.g_thinking_open = false;
-    self.thinking_folded = false;
-    const w = self.out orelse return;
-    if (!self.thinking_overflow and self.thinking_rows >= 1 and main_mod.use_color) {
-        w.print("\x1b[{d}F\x1b[0J{s}✓ Thought{s}\n\n", .{ self.thinking_rows, style.dim, style.reset }) catch return;
-    } else {
-        w.print("{s}\n{s}✓ Thought{s}\n\n", .{ style.reset, style.dim, style.reset }) catch return;
-    }
-    w.flush() catch return;
-    _ = tick_gate.setLineStart(true); // both branches end at column 0 — held ticks land here (#tui-tick)
-}
-
-/// Ctrl-T: fold/unfold the live "Thinking" block in place (#92/#85). Only
-/// acts on an open, on-screen block; folding erases it to a one-line marker,
-/// unfolding re-streams the buffered reasoning. Cursor math mirrors
-/// closeThinkingBlock (erase `thinking_rows` lines up, clear to end).
-pub fn toggleThinkingFold(self: *Agent) void {
-    if (!self.thinking_open or self.thinking_overflow or !main_mod.use_color) return;
-    const w = self.out orelse return;
-    if (!self.thinking_folded) {
-        w.print("\x1b[{d}F\x1b[0J{s}▶ Thinking (folded · ^T){s}\n", .{ self.thinking_rows, style.dim, style.reset }) catch return;
-        self.thinking_folded = true;
-        self.thinking_rows = 1;
-        self.thinking_col = 0;
-    } else {
-        w.print("\x1b[1F\x1b[0J{s}▼ Thinking{s}\n{s}", .{ style.dim, style.reset, style.dim }) catch return;
-        self.thinking_folded = false;
-        self.thinking_rows = 1;
-        self.thinking_col = 0;
-        w.writeAll(self.thinking_text.items) catch return;
-        advanceThinkingRows(&self.thinking_rows, &self.thinking_col, termCols(), self.thinking_text.items);
-    }
-    w.flush() catch return;
-}
-
 pub fn postStream(self: *Agent, body: []const u8) ![]u8 {
     return postStreamWithClient(self, self.client, body);
 }
@@ -182,8 +50,15 @@ pub fn postStream(self: *Agent, body: []const u8) ![]u8 {
 /// keep-alive cannot poison the WS→SSE handoff and every fallback retry dials
 /// from a clean pool.
 pub fn postStreamWithClient(self: *Agent, client: *std.http.Client, body: []const u8) ![]u8 {
-    self.spinnerStart();
-    defer self.spinnerStop();
+    const sink = engine_sink.forAgent(self);
+    sink.emit(self.io, .stream_begin);
+    // Every exit path — success, interrupt, transport error — tears down the
+    // live-stream presentation (spinner, a reasoning-only turn's open block).
+    defer sink.emit(self.io, .stream_finished);
+    // Per-stream frontend bookkeeping still lives on the Agent in #422 slice 1
+    // (TuiSink wraps it); the resets are state hygiene, not emissions, and run
+    // for every mode exactly as before — emitArgText can dirty the markdown
+    // state even when this stream's deltas go to the wire.
     self.thinking_open = false; // fresh "Thinking" block state per request
     main_mod.g_thinking_open = false;
     self.thinking_rows = 0;
@@ -191,7 +66,6 @@ pub fn postStreamWithClient(self: *Agent, client: *std.http.Client, body: []cons
     self.thinking_folded = false;
     self.thinking_text.clearRetainingCapacity();
     self.thinking_overflow = false;
-    defer self.closeThinkingBlock(); // close a reasoning-only turn's block
     const gpa = self.gpa;
     const provider = self.provider;
     const bearer = switch (provider.auth) {
@@ -226,11 +100,10 @@ pub fn postStreamWithClient(self: *Agent, client: *std.http.Client, body: []cons
     // the terminal scroll its own scrollback — native scroll wins, and Ctrl-T
     // still folds the live Thinking block (escPressed).
     const watch_esc = !self.sub and self.in != null and main_mod.use_color and !main_mod.json_mode;
-    var orig_tio: ?tty.RawState = null;
-    if (watch_esc) orig_tio = rawNonblockStdin();
+    const orig_tio = if (watch_esc) rawNonblockStdin() else null;
     defer if (orig_tio) |o| {
         _ = drainSteerStdin(true);
-        tty.restore(o);
+        restoreStdin(o);
     };
     var req = try client.request(.POST, try std.Uri.parse(provider.url), .{
         .redirect_behavior = .unhandled,
@@ -359,11 +232,7 @@ pub fn postStreamWithClient(self: *Agent, client: *std.http.Client, body: []cons
                 // flush the partial, poison, and end the turn as StreamStalled (never a
                 // hang, never a mislabeled StreamDropped or user Esc).
                 if (saw_done) break :stream;
-                self.flushStreamTail();
-                if (!main_mod.json_mode) if (self.out) |o| {
-                    o.writeAll("\n⚠ stream stalled — ending turn\n") catch {};
-                    o.flush() catch {};
-                };
+                sink.emit(self.io, .{ .stream_aborted = .stalled });
                 if (req.connection) |conn| conn.closing = true;
                 return error.StreamStalled;
             };
@@ -376,7 +245,7 @@ pub fn postStreamWithClient(self: *Agent, client: *std.http.Client, body: []cons
                 _ = r.line catch |e| {
                     if (saw_done and readErrIsClose(e)) break :stream;
                     if (got_body and readErrIsClose(e)) {
-                        noteDropped(self);
+                        sink.emit(self.io, .{ .stream_aborted = .dropped });
                         return error.StreamDropped;
                     } // #133
                     return e;
@@ -392,22 +261,18 @@ pub fn postStreamWithClient(self: *Agent, client: *std.http.Client, body: []cons
                 .line => |r| _ = r catch |e| {
                     if (saw_done and readErrIsClose(e)) break :stream; // #134/#135: post-completion close/reset is success, not a retryable flake
                     if (got_body and readErrIsClose(e)) {
-                        noteDropped(self);
+                        sink.emit(self.io, .{ .stream_aborted = .dropped });
                         return error.StreamDropped;
                     } // #133: closed before the terminal event
                     return e;
                 },
                 .stall => |w| {
-                    self.flushStreamTail();
-                    if (req.connection) |conn| conn.closing = true;
                     // A user Esc is a deliberate cancel; a `.deadline` is a dead or
                     // idle stream (silent past this read's budget) — end the turn as
                     // error.StreamStalled so it is never recorded as "[response
-                    // interrupted by user]" (#134). The notice below is deadline-only.
-                    if (w == .deadline and !main_mod.json_mode) if (self.out) |o| {
-                        o.writeAll("\n⚠ stream stalled — ending turn\n") catch {};
-                        o.flush() catch {};
-                    };
+                    // interrupted by user]" (#134). Only the deadline gets a notice.
+                    sink.emit(self.io, .{ .stream_aborted = if (w == .deadline) .stalled else .interrupted });
+                    if (req.connection) |conn| conn.closing = true;
                     return watchdogError(w, error.StreamStalled);
                 },
             }
@@ -421,7 +286,7 @@ pub fn postStreamWithClient(self: *Agent, client: *std.http.Client, body: []cons
         self.printDelta(line.writer.buffered());
         if (main_mod.g_thinking_fold_request) {
             main_mod.g_thinking_fold_request = false;
-            self.toggleThinkingFold();
+            sink.emit(self.io, .thinking_fold_toggle);
         }
         // Logical stream terminator: once the provider's final event ([DONE] /
         // response.completed / message_stop) lands in `full` the response is
@@ -438,7 +303,7 @@ pub fn postStreamWithClient(self: *Agent, client: *std.http.Client, body: []cons
         }
         line.clearRetainingCapacity();
         if ((orig_tio != null and escPressed(true)) or (self.sub and Agent.esc_cancel.load(.acquire))) {
-            self.flushStreamTail();
+            sink.emit(self.io, .{ .stream_aborted = .interrupted });
             // Mark the connection closing so req.deinit() tears it down
             // instead of draining the rest of the stream (which would
             // block until the model finished generating anyway). Subs get
@@ -449,12 +314,7 @@ pub fn postStreamWithClient(self: *Agent, client: *std.http.Client, body: []cons
         if (!more) break;
         reader.toss(1);
     }
-    self.flushStreamTail(); // render any held partial markdown line
-    if (!main_mod.json_mode and self.streamed_text) if (self.out) |w| {
-        w.writeAll("\n") catch {};
-        w.flush() catch {};
-        _ = tick_gate.setLineStart(true); // answer is off the wire: release held child ticks (#tui-tick)
-    };
+    sink.emit(self.io, .{ .stream_complete = .{ .streamed_text = self.streamed_text } });
     return full.toOwnedSlice();
 }
 
@@ -476,18 +336,6 @@ fn openaiComplete(raw_line: []const u8) bool {
     return std.mem.indexOf(u8, payload, "\"finish_reason\":\"") != null;
 }
 
-/// The provider closed/reset the stream before its terminal event landed — the
-/// harness is ending the turn, not the user (#133). Flush the partial and tell
-/// the user (TTY/plain), so a Moonshot-style mid-reasoning drop can never pass
-/// silently as a completed answer.
-fn noteDropped(self: *Agent) void {
-    self.flushStreamTail();
-    if (!main_mod.json_mode) if (self.out) |o| {
-        o.writeAll("\n⚠ connection dropped — response ended early\n") catch {};
-        o.flush() catch {};
-    };
-}
-
 /// True if this SSE line is the provider's terminal event — after it no more
 /// content comes, so postStream can stop instead of waiting for the socket to
 /// close (#134/#135). Precise: matches the `[DONE]` sentinel, a structural
@@ -533,10 +381,11 @@ test "openaiComplete (#133): finish_reason marks completion, deltas do not" {
     try stream_tests.openaiCompletion(openaiComplete);
 }
 
-/// Print the user-visible text from one SSE line, if any. Best-effort:
-/// parse failures are ignored (the buffered body is parsed afterwards).
+/// Extract the user-visible content from one SSE line and dispatch it as
+/// typed events through the sink. Best-effort: parse failures are ignored
+/// (the buffered body is parsed afterwards).
 pub fn printDelta(self: *Agent, raw_line: []const u8) void {
-    const w = self.out orelse return;
+    if (self.out == null) return; // pool-thread subagents have no frontend writer: skip entirely (capture included), as ever
     const payload = ssePayload(raw_line) orelse return;
     const parsed = std.json.parseFromSlice(Value, self.gpa, payload, .{}) catch return;
     defer parsed.deinit();
@@ -572,29 +421,13 @@ pub fn printDelta(self: *Agent, raw_line: []const u8) void {
         },
     };
     // Reasoning/thinking deltas: deepseek streams reasoning_content, anthropic
-    // a thinking_delta, codex a summary delta. JSON clients get a `reasoning`
-    // event; on a TTY we stream it into a live, dimmed "Thinking" block when
-    // /thinking is enabled, otherwise the spinner stands in for it.
+    // a thinking_delta, codex a summary delta. Presentation is the sink's call:
+    // the wire's `reasoning` event, or the live "Thinking" block / spinner.
+    const sink = engine_sink.forAgent(self);
     const reasoning = reasoningDelta(self.provider.kind, obj);
-    if (reasoning.len != 0) {
-        if (main_mod.json_mode) {
-            self.emit(.{ .type = "reasoning", .text = reasoning });
-        } else if (self.show_thinking and !self.sub and !self.stream_quiet and main_mod.use_color) {
-            self.streamThinking(reasoning);
-        }
-    }
+    if (reasoning.len != 0) sink.emit(self.io, .{ .reasoning_delta = .{ .text = reasoning } });
     if (text.len == 0) return;
-    if (self.thinking_open) self.closeThinkingBlock(); // reasoning → answer transition
-    self.spinnerStop(); // first visible byte: clear the thinking line
     self.streamed_text = true;
     self.partial_text.appendSlice(self.arena, text) catch {}; // Esc-interrupt capture
-    if (main_mod.json_mode) {
-        self.emit(.{ .type = "text", .text = text });
-    } else if (main_mod.use_color) {
-        self.streamMarkdown(text);
-    } else {
-        w.writeAll(text) catch return;
-        w.flush() catch return;
-        if (!self.sub) _ = tick_gate.setLineStart(text[text.len - 1] == '\n'); // #tui-tick
-    }
+    sink.emit(self.io, .{ .text_delta = .{ .text = text } });
 }
diff --git a/src/agent_stream_render.zig b/src/agent_stream_render.zig
new file mode 100644
index 000000000..46655412e
--- /dev/null
+++ b/src/agent_stream_render.zig
@@ -0,0 +1,150 @@
+//! TUI-side live-stream rendering (#422): the thinking spinner
+//! (spinnerTask/Start/Stop, an animated indicator while the model is silent)
+//! and the live dimmed "Thinking" reasoning block (streamThinking/
+//! closeThinkingBlock/toggleThinkingFold), moved verbatim out of
+//! agent_stream.zig so the transport loop owns no terminal drawing. Driven by
+//! TuiSink (engine_sink.zig); agent_ws.zig still reaches the spinner through
+//! its Agent member aliases. Frontend territory: term.zig/ansi.zig/anim.zig
+//! imports live here, never in engine files.
+//!
+//! Agent.g_spin_stop/Agent.g_spin_future are struct-level `pub var`s that stay
+//! declared directly inside the Agent struct (never alias a `var`) — reached
+//! here as `Agent.g_spin_stop`/`Agent.g_spin_future`.
+
+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 style = &@import("ansi.zig").style;
+
+const anim = @import("anim.zig");
+const tick_gate = @import("tick_gate.zig"); // #tui-tick: child ticks wait for a foreground line boundary
+
+const terminal = @import("term.zig");
+const termCols = terminal.termCols;
+const termRows = terminal.termRows;
+const advanceThinkingRows = terminal.advanceThinkingRows;
+
+pub fn spinnerTask(io: Io) void {
+    var i: usize = 0;
+    var buf: [512]u8 = undefined;
+    var w = Io.File.stdout().writer(io, &buf);
+    while (!Agent.g_spin_stop.load(.acquire)) {
+        if (main_mod.g_steer_visible.load(.acquire)) {
+            io.sleep(.fromMilliseconds(20), .awake) catch break;
+            continue;
+        }
+        // Clear-then-draw each frame: animations may vary in width.
+        w.interface.writeAll("\r\x1b[2K\x1b[?7l") catch return; // ?7l: autowrap off so a wide spinner truncates instead of wrapping in a narrow window (the "goes on and on" bug)
+        anim.anims[anim.g_anim_current].frame(&w.interface, i) catch return;
+        w.interface.writeAll("\x1b[?7h") catch return; // restore autowrap
+        w.interface.flush() catch return;
+        i += 1;
+        const frame_ticks = @max(@as(usize, 1), @as(usize, anim.anims[anim.g_anim_current].frame_ms) / 20);
+        var t: usize = 0;
+        while (t < frame_ticks and !Agent.g_spin_stop.load(.acquire)) : (t += 1) {
+            if (main_mod.g_steer_visible.load(.acquire)) break;
+            io.sleep(.fromMilliseconds(20), .awake) catch break;
+        }
+    }
+    if (!main_mod.g_steer_visible.load(.acquire)) {
+        w.interface.writeAll("\x1b[?7h\r\x1b[2K") catch return; // restore autowrap + clear
+        w.interface.flush() catch {};
+    }
+}
+
+pub fn spinnerStart(self: *Agent) void {
+    if (self.sub or main_mod.json_mode or !main_mod.use_color or self.out == null) return;
+    if (anim.g_anim_off) return;
+    if (Agent.g_spin_future != null) return;
+    anim.selectSpinner(self.io);
+    Agent.g_spin_stop.store(false, .release);
+    Agent.g_spin_future = self.io.concurrent(spinnerTask, .{self.io}) catch blk: {
+        Agent.g_spin_stop.store(true, .release); // no spare concurrency: skip quietly
+        break :blk null;
+    };
+}
+
+pub fn spinnerStop(self: *Agent) void {
+    if (self.sub) return; // root-only state — subs run on pool threads
+    if (Agent.g_spin_future) |*f| {
+        Agent.g_spin_stop.store(true, .release);
+        f.await(self.io);
+        Agent.g_spin_future = null;
+    }
+}
+
+/// Stream a chunk of the model's reasoning into a live, dimmed "Thinking"
+/// block in the terminal, opening the block (and handing the line off from
+/// the spinner) on the first chunk. Gated by /thinking; when off the block is
+/// never opened and the spinner stands in for it. We track the block's
+/// on-screen height as it streams so closeThinkingBlock can collapse it to a
+/// one-line summary when the answer starts (#75).
+pub fn streamThinking(self: *Agent, chunk: []const u8) void {
+    const w = self.out orelse return;
+    if (!self.thinking_open) {
+        self.spinnerStop();
+        w.print("{s}▼ Thinking{s}\n{s}", .{ style.dim, style.reset, style.dim }) catch return;
+        self.thinking_open = true;
+        main_mod.g_thinking_open = true;
+        self.thinking_rows = 1; // the header newline already moved us down one line
+        self.thinking_col = 0;
+        self.thinking_overflow = false;
+        // The block owns every row below this one and collapses them by cursor
+        // math — a child's tick printed inside it would be erased with the
+        // block (or shift the erase onto real output). Hold until it closes.
+        tick_gate.hold();
+    }
+    self.thinking_text.appendSlice(self.gpa, chunk) catch {};
+    if (self.thinking_folded) return; // folded: buffer only, don't draw the live block
+    w.writeAll(chunk) catch return;
+    w.flush() catch return;
+    advanceThinkingRows(&self.thinking_rows, &self.thinking_col, termCols(), chunk);
+    if (self.thinking_rows + 1 >= termRows()) self.thinking_overflow = true;
+}
+
+/// Close an open "Thinking" block. If it still fits on screen, collapse it in
+/// place to a one-line "Thought" summary (#75); if it has scrolled off
+/// (overflow) leave the reasoning and just append the summary, so we never
+/// erase the user's earlier output. Runs on the reasoning->answer transition
+/// and at stream end.
+pub fn closeThinkingBlock(self: *Agent) void {
+    if (!self.thinking_open) return;
+    self.thinking_open = false;
+    main_mod.g_thinking_open = false;
+    self.thinking_folded = false;
+    const w = self.out orelse return;
+    if (!self.thinking_overflow and self.thinking_rows >= 1 and main_mod.use_color) {
+        w.print("\x1b[{d}F\x1b[0J{s}✓ Thought{s}\n\n", .{ self.thinking_rows, style.dim, style.reset }) catch return;
+    } else {
+        w.print("{s}\n{s}✓ Thought{s}\n\n", .{ style.reset, style.dim, style.reset }) catch return;
+    }
+    w.flush() catch return;
+    _ = tick_gate.setLineStart(true); // both branches end at column 0 — held ticks land here (#tui-tick)
+}
+
+/// Ctrl-T: fold/unfold the live "Thinking" block in place (#92/#85). Only
+/// acts on an open, on-screen block; folding erases it to a one-line marker,
+/// unfolding re-streams the buffered reasoning. Cursor math mirrors
+/// closeThinkingBlock (erase `thinking_rows` lines up, clear to end).
+pub fn toggleThinkingFold(self: *Agent) void {
+    if (!self.thinking_open or self.thinking_overflow or !main_mod.use_color) return;
+    const w = self.out orelse return;
+    if (!self.thinking_folded) {
+        w.print("\x1b[{d}F\x1b[0J{s}▶ Thinking (folded · ^T){s}\n", .{ self.thinking_rows, style.dim, style.reset }) catch return;
+        self.thinking_folded = true;
+        self.thinking_rows = 1;
+        self.thinking_col = 0;
+    } else {
+        w.print("\x1b[1F\x1b[0J{s}▼ Thinking{s}\n{s}", .{ style.dim, style.reset, style.dim }) catch return;
+        self.thinking_folded = false;
+        self.thinking_rows = 1;
+        self.thinking_col = 0;
+        w.writeAll(self.thinking_text.items) catch return;
+        advanceThinkingRows(&self.thinking_rows, &self.thinking_col, termCols(), self.thinking_text.items);
+    }
+    w.flush() catch return;
+}
diff --git a/src/engine_events.zig b/src/engine_events.zig
new file mode 100644
index 000000000..145d71a07
--- /dev/null
+++ b/src/engine_events.zig
@@ -0,0 +1,157 @@
+//! The engine's internal event vocabulary (#422 slice 1): one tagged union of
+//! every distinct output emission the live streaming path (agent_stream.zig)
+//! produces, expressed as semantic content — no ANSI, no pre-rendered text.
+//! Frontends never see engine internals: a sink (engine_sink.zig) receives
+//! these events and renders them (TUI) or serializes them (--json wire).
+//!
+//! Growth rules for the vocabulary:
+//! - New engine output = a new variant (or a new field on a payload struct),
+//!   never a pre-rendered string where structure exists. Rendering choices
+//!   (color, spinners, wording of notices) belong to sinks.
+//! - `durable` classifies each variant: durable events are the protocol
+//!   stream a wire/log sink persists and are what reserve sequence ids;
+//!   everything else is a presentation pulse that rides at the current
+//!   position. Promoting a pulse to the wire is an externally visible shape
+//!   change and is gated behind a schema_version bump (epic #422 rule 1).
+//! - Payloads are structs so fields can grow without call-site churn.
+
+const std = @import("std");
+const protocol_seq = @import("protocol_seq.zig");
+
+/// Position of an event in a session's event log. `sequence` alone is
+/// meaningless across restarts: `generation` increments whenever the log
+/// restarts (process restart, resume — wired in a later #422 slice), and
+/// `sequence` is monotonic within a generation (protocol_seq.zig, the one
+/// counter the --json wire already stamps).
+pub const Cursor = struct {
+    generation: u64,
+    sequence: u64,
+};
+
+/// Why a delta stream stopped before the provider's terminal event.
+pub const StreamAbort = enum {
+    /// The user cancelled (Esc) — a deliberate stop, rendered silently.
+    interrupted,
+    /// The stream went silent past its watchdog budget, or the read path lost
+    /// its watchdog (pool exhaustion) — the harness is ending the turn.
+    stalled,
+    /// The provider closed/reset the socket before its terminal event (#133).
+    dropped,
+};
+
+/// A streamed content chunk. `text` is never empty — emitters drop empty
+/// deltas before dispatch.
+pub const Delta = 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) {
+    /// A streaming model request is in flight; nothing has arrived yet. The
+    /// TUI answers with the thinking spinner and a fresh per-stream render
+    /// state; the wire has no shape for it.
+    stream_begin,
+    /// A chunk of the model's reasoning ("thinking") text. Wire: the
+    /// existing `reasoning` event. TUI: the live dimmed Thinking block,
+    /// gated by /thinking.
+    reasoning_delta: Delta,
+    /// A chunk of visible answer text. Wire: the existing `text` event.
+    /// TUI: streamed markdown (or raw bytes off-color). The first one also
+    /// ends the reasoning presentation (block close, spinner stop).
+    text_delta: Delta,
+    /// The user asked to fold/unfold the live reasoning view (Ctrl-T, #92).
+    /// Presentation-only; a headless frontend ignores it. TRANSITIONAL
+    /// (Phase 1b): this is frontend INPUT round-tripping engine-ward through
+    /// a global (g_thinking_fold_request) and coming back out — when input
+    /// inversion lands it leaves this union (a frontend-owned command, not an
+    /// engine event), so plan for removal, not extension.
+    thinking_fold_toggle,
+    /// The delta stream was cut before its terminal event; the payload says
+    /// how. Sinks flush any held partial output and may surface a notice for
+    /// the non-deliberate reasons.
+    stream_aborted: StreamAbort,
+    /// The delta stream ended normally (terminal event seen). streamed_text:
+    /// at least one text_delta was emitted live, so the TUI ends the answer
+    /// line.
+    stream_complete: struct { streamed_text: bool },
+    /// The streaming transport call is fully over — emitted on EVERY exit
+    /// path, after stream_complete/stream_aborted when those fired. Sinks
+    /// tear down live-stream presentation (spinner, an open reasoning-only
+    /// Thinking block).
+    stream_finished,
+};
+
+/// 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.
+pub fn durable(ev: EngineEvent) bool {
+    return switch (ev) {
+        .reasoning_delta, .text_delta => true,
+        else => false,
+    };
+}
+
+// Generation 1 is the first life of this process's event log; restart/resume
+// wiring bumps it in a later #422 slice (serve attach, #420).
+var g_generation: std.atomic.Value(u64) = .init(1);
+
+pub fn generation() u64 {
+    return g_generation.load(.monotonic);
+}
+
+/// The session's event log restarted: later sequences are a fresh line of
+/// history, not a continuation the old cursor can index into.
+pub fn bumpGeneration() u64 {
+    return g_generation.fetchAdd(1, .monotonic) + 1;
+}
+
+/// Stamp a Cursor at the emission boundary. `reserve` draws a fresh id from
+/// protocol_seq (durable events on a durable sink; in --json mode the caller
+/// holds the stdout lock so reservation and wire order can never diverge —
+/// an injected durable sink outside --json currently reserves unlocked, see
+/// the engine_sink.zig header note); otherwise the event observes the last
+/// reserved position without advancing it.
+pub fn stamp(reserve: bool) Cursor {
+    return .{
+        .generation = generation(),
+        .sequence = if (reserve) protocol_seq.next() else protocol_seq.current(),
+    };
+}
+
+test "every variant constructs; only the wire deltas are durable" {
+    const wire: [2]EngineEvent = .{
+        .{ .reasoning_delta = .{ .text = "why" } },
+        .{ .text_delta = .{ .text = "hi" } },
+    };
+    for (wire) |ev| try std.testing.expect(durable(ev));
+    const pulses: [5]EngineEvent = .{
+        .stream_begin,
+        .thinking_fold_toggle,
+        .{ .stream_aborted = .stalled },
+        .{ .stream_complete = .{ .streamed_text = true } },
+        .stream_finished,
+    };
+    for (pulses) |ev| try std.testing.expect(!durable(ev));
+}
+
+test "stamp: reserving draws fresh monotonic ids; observing never advances" {
+    protocol_seq.resetForTest();
+    defer protocol_seq.resetForTest();
+    const a = stamp(true);
+    const b = stamp(true);
+    try std.testing.expectEqual(@as(u64, 1), a.sequence);
+    try std.testing.expectEqual(a.sequence + 1, b.sequence);
+    // A pulse rides at the last reserved position and reserves nothing.
+    const o = stamp(false);
+    try std.testing.expectEqual(b.sequence, o.sequence);
+    try std.testing.expectEqual(b.sequence, protocol_seq.current());
+    try std.testing.expectEqual(generation(), o.generation);
+}
+
+test "generation only moves forward, one restart at a time" {
+    const before = generation();
+    try std.testing.expect(before >= 1);
+    try std.testing.expectEqual(before + 1, bumpGeneration());
+    try std.testing.expectEqual(before + 1, generation());
+}
diff --git a/src/engine_sink.zig b/src/engine_sink.zig
new file mode 100644
index 000000000..b22f04e7f
--- /dev/null
+++ b/src/engine_sink.zig
@@ -0,0 +1,283 @@
+//! The EngineSink boundary (#422 slice 1): engine code emits typed events
+//! (engine_events.zig) and a sink turns them into a frontend's output — it
+//! receives events, nothing else. Two impls:
+//!
+//! - TuiSink: today's terminal rendering, verbatim. Its helpers live in
+//!   frontend territory (agent_stream_render.zig, agent_render.zig via Agent
+//!   aliases); slice 1 keeps the render state (thinking_*/md_* fields) on the
+//!   Agent it wraps, so the ctx pointer is that state handle. Known slice-1
+//!   debt against the strict-sink rule ("sinks render from events only"):
+//!   besides that drawing bookkeeping, the reasoning gate still back-reads
+//!   Agent policy (show_thinking/sub/stream_quiet — see tuiEmit). A later
+//!   slice moves visibility into the event payload or makes it a sink-owned
+//!   preference so a transport-split sink needs no reads into the Agent.
+//! - 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.
+//!
+//! Events are stamped with a {generation, sequence} Cursor at the dispatch
+//! boundary. In --json mode (the only durable sink today), durable events
+//! reserve their id INSIDE the same lock that serializes --json stdout,
+//! keeping the wire's numbering gap-free and ordered against pool-thread
+//! guiEmit writers. NOTE the lock condition is keyed to json_mode, not to
+//! vt.durable: an injected durable sink outside --json reserves WITHOUT the
+//! lock. When serve/attach (#420) adds one, key the lock to the sink (with a
+//! test seam) or it inherits exactly the reorder race this lock prevents.
+
+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 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 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.
+pub const Stamped = struct {
+    cursor: engine_events.Cursor,
+    event: EngineEvent,
+};
+
+pub const VTable = struct {
+    emit: *const fn (ctx: *anyopaque, ev: Stamped) void,
+    /// A durable sink persists/forwards the protocol stream: durable events
+    /// reserve a fresh sequence id at dispatch. Presentation-only sinks
+    /// observe the current position instead, so an interactive session never
+    /// advances the persisted event_seq the --json wire owns.
+    durable: bool,
+};
+
+pub const EngineSink = struct {
+    ctx: *anyopaque,
+    vt: *const VTable,
+
+    /// 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.
+    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) {
+            // Reserving outside the lock could put a smaller seq on the wire
+            // AFTER a pool-thread guiEmit line took a larger one.
+            main_mod.g_gui_mu.lockUncancelable(io);
+            defer main_mod.g_gui_mu.unlock(io);
+            self.vt.emit(self.ctx, .{ .cursor = engine_events.stamp(true), .event = ev });
+            return;
+        }
+        self.vt.emit(self.ctx, .{ .cursor = engine_events.stamp(reserve), .event = ev });
+    }
+};
+
+/// The agent's sink: an injected one (tests, future frontends) or the
+/// process-mode default — the wire in --json mode, the terminal otherwise.
+pub fn forAgent(a: *Agent) EngineSink {
+    if (a.sink) |s| return s;
+    return if (main_mod.json_mode) jsonSink(a) else tuiSink(a);
+}
+
+pub fn tuiSink(a: *Agent) EngineSink {
+    return .{ .ctx = a, .vt = &tui_vtable };
+}
+
+pub fn jsonSink(a: *Agent) EngineSink {
+    return .{ .ctx = a, .vt = &json_vtable };
+}
+
+const tui_vtable: VTable = .{ .emit = tuiEmit, .durable = false };
+const json_vtable: VTable = .{ .emit = jsonEmit, .durable = true };
+
+/// Today's interactive rendering, relocated behind the event contract. Every
+/// branch is the old inline agent_stream.zig code path, gate for gate.
+fn tuiEmit(ctx: *anyopaque, ev: Stamped) void {
+    const a: *Agent = @ptrCast(@alignCast(ctx));
+    switch (ev.event) {
+        .stream_begin => render.spinnerStart(a),
+        // Reasoning streams into the live dimmed "Thinking" block when
+        // /thinking is on for a live, colored root turn; otherwise the
+        // spinner stands in for it. TRANSITIONAL (slice-1 debt, see header):
+        // this gate back-reads Agent policy to decide WHAT to render — a
+        // wire-split sink cannot, so it must move into the event payload or
+        // become a sink-owned preference before Phase 2.
+        .reasoning_delta => |d| if (a.show_thinking and !a.sub and !a.stream_quiet and main_mod.use_color)
+            render.streamThinking(a, d.text),
+        .text_delta => |d| {
+            if (a.thinking_open) render.closeThinkingBlock(a); // reasoning -> answer transition
+            render.spinnerStop(a); // first visible byte: clear the thinking line
+            if (main_mod.use_color) {
+                a.streamMarkdown(d.text);
+            } else if (a.out) |w| {
+                w.writeAll(d.text) catch return;
+                w.flush() catch return;
+                if (!a.sub) _ = tick_gate.setLineStart(d.text[d.text.len - 1] == '\n'); // #tui-tick
+            }
+        },
+        .thinking_fold_toggle => render.toggleThinkingFold(a),
+        .stream_aborted => |reason| {
+            a.flushStreamTail(); // render any held partial markdown line
+            switch (reason) {
+                .interrupted => {}, // a deliberate Esc needs no notice
+                .stalled => notice(a, "\n⚠ stream stalled — ending turn\n"),
+                .dropped => notice(a, "\n⚠ connection dropped — response ended early\n"),
+            }
+        },
+        .stream_complete => |c| {
+            a.flushStreamTail(); // render any held partial markdown line
+            if (c.streamed_text) if (a.out) |w| {
+                w.writeAll("\n") catch {};
+                w.flush() catch {};
+                _ = tick_gate.setLineStart(true); // answer is off the wire: release held child ticks (#tui-tick)
+            };
+        },
+        .stream_finished => {
+            render.closeThinkingBlock(a); // a reasoning-only turn still closes its block
+            render.spinnerStop(a);
+        },
+    }
+}
+
+fn notice(a: *Agent, text: []const u8) void {
+    const w = a.out orelse return;
+    w.writeAll(text) catch return;
+    w.flush() catch {};
+}
+
+/// The existing --json wire. Only the durable events have a shape; giving a
+/// pulse one is an externally visible change (schema_version gate). Dispatch
+/// already holds the stdout lock for these writes in --json mode.
+fn jsonEmit(ctx: *anyopaque, ev: Stamped) void {
+    const a: *Agent = @ptrCast(@alignCast(ctx));
+    const w = a.out orelse return;
+    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: emitArgText streams tool-arg prose
+        // through streamMarkdown whenever use_color is on — --json on a TTY
+        // included — so md_buf/md_table can hold bytes even here. Yes, that
+        // interleaves non-JSONL text into the wire exactly as before;
+        // making --json drop the tail (or never dirty md state) is a
+        // deliberate future wire change, not slice-1 fallout.
+        .stream_aborted, .stream_complete => a.flushStreamTail(),
+        else => {},
+    }
+}
+
+fn jsonLine(w: *Io.Writer, cursor: engine_events.Cursor, payload: anytype) void {
+    protocol_seq.writeEventStamped(w, cursor.sequence, payload) catch return;
+    w.writeByte('\n') catch return;
+    w.flush() catch return;
+}
+
+test "dispatch preserves emission order and stamps at the boundary" {
+    // emit(undefined, ...) is sound only while json_mode is false (no lock
+    // taken): pin it so a leaky earlier test can never turn this into UB.
+    const saved_json = main_mod.json_mode;
+    main_mod.json_mode = false;
+    defer main_mod.json_mode = saved_json;
+    protocol_seq.resetForTest();
+    defer protocol_seq.resetForTest();
+    var rec: std.ArrayList(Stamped) = .empty;
+    defer rec.deinit(std.testing.allocator);
+    const vt: VTable = .{ .emit = recordEmit, .durable = true };
+    const s: EngineSink = .{ .ctx = &rec, .vt = &vt };
+    s.emit(undefined, .stream_begin);
+    s.emit(undefined, .{ .reasoning_delta = .{ .text = "think" } });
+    s.emit(undefined, .{ .text_delta = .{ .text = "hi" } });
+    s.emit(undefined, .{ .stream_complete = .{ .streamed_text = true } });
+    s.emit(undefined, .stream_finished);
+    try std.testing.expectEqual(@as(usize, 5), rec.items.len);
+    const want_tags: [5]std.meta.Tag(EngineEvent) = .{
+        .stream_begin, .reasoning_delta, .text_delta, .stream_complete, .stream_finished,
+    };
+    for (want_tags, rec.items) |tag, got| try std.testing.expectEqual(tag, std.meta.activeTag(got.event));
+    // Durable deltas reserved fresh ids; pulses ride at the last reserved
+    // position — the wire's numbering shows no gap for them.
+    const want_seq: [5]u64 = .{ 0, 1, 2, 2, 2 };
+    for (want_seq, rec.items) |seq, got| try std.testing.expectEqual(seq, got.cursor.sequence);
+    for (rec.items) |got| try std.testing.expectEqual(engine_events.generation(), got.cursor.generation);
+}
+
+test "a presentation sink never reserves sequence ids" {
+    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 rec: std.ArrayList(Stamped) = .empty;
+    defer rec.deinit(std.testing.allocator);
+    const vt: VTable = .{ .emit = recordEmit, .durable = false };
+    const s: EngineSink = .{ .ctx = &rec, .vt = &vt };
+    s.emit(undefined, .{ .text_delta = .{ .text = "hi" } });
+    try std.testing.expectEqual(@as(u64, 0), rec.items[0].cursor.sequence);
+    try std.testing.expectEqual(@as(u64, 0), protocol_seq.current());
+}
+
+fn recordEmit(ctx: *anyopaque, ev: Stamped) void {
+    const rec: *std.ArrayList(Stamped) = @ptrCast(@alignCast(ctx));
+    rec.append(std.testing.allocator, ev) catch @panic("OOM");
+}
+
+test "JsonSink writes today's wire lines byte-for-byte" {
+    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: Agent = .{
+        .gpa = std.testing.allocator,
+        .arena = std.testing.allocator,
+        .io = undefined,
+        .client = undefined,
+        .provider = undefined,
+        .messages = undefined,
+        .sub = false,
+        .label = "test",
+        .out = &aw.writer,
+    };
+    const s = jsonSink(&a);
+    s.emit(undefined, .{ .reasoning_delta = .{ .text = "why" } });
+    s.emit(undefined, .{ .text_delta = .{ .text = "hi\n" } });
+    s.emit(undefined, .stream_begin); // pulses have no wire shape
+    // End-of-stream flushes the held render tail (old-path parity); with
+    // clean md state that adds no bytes and emits no wire line.
+    s.emit(undefined, .{ .stream_complete = .{ .streamed_text = true } });
+    try std.testing.expectEqualStrings(
+        "{\"seq\":1,\"type\":\"reasoning\",\"text\":\"why\"}\n{\"seq\":2,\"type\":\"text\",\"text\":\"hi\\n\"}\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,
+    };
+    const s = tuiSink(&a);
+    s.emit(undefined, .{ .text_delta = .{ .text = "plain\n" } });
+    try std.testing.expectEqualStrings("plain\n", aw.writer.buffered());
+    // Normal end after streamed text: the separating newline, as before.
+    s.emit(undefined, .{ .stream_complete = .{ .streamed_text = true } });
+    try std.testing.expectEqualStrings("plain\n\n", aw.writer.buffered());
+}
diff --git a/src/protocol_seq.zig b/src/protocol_seq.zig
index acdf04af6..d54d688f8 100644
--- a/src/protocol_seq.zig
+++ b/src/protocol_seq.zig
@@ -48,10 +48,18 @@ pub fn resetForTest() void {
 /// Does NOT write the trailing newline — the caller owns line framing (and,
 /// for stdout, the lock that keeps two writers from interleaving).
 pub fn writeEvent(w: *Io.Writer, ev: anytype) !void {
+    try writeEventStamped(w, next(), ev);
+}
+
+/// Same wire shape with a caller-reserved id: the #422 emission boundary
+/// (engine_sink.zig) stamps an event's Cursor first — inside the stdout lock —
+/// and the sink serializes with that exact sequence, so the stamp and the wire
+/// can never drift. Never bumps the counter itself.
+pub fn writeEventStamped(w: *Io.Writer, seq_id: u64, ev: anytype) !void {
     var s: std.json.Stringify = .{ .writer = w };
     try s.beginObject();
     try s.objectField("seq");
-    try s.write(next());
+    try s.write(seq_id);
     inline for (comptime std.meta.fieldNames(@TypeOf(ev))) |name| {
         comptime {
             if (std.mem.eql(u8, name, "seq"))
@@ -137,3 +145,20 @@ test "seqOf reads the envelope off a prefix and refuses to guess" {
     try std.testing.expect(seqOf("{\"type\":\"tool_result\",\"seq\":5}") == null);
     try std.testing.expect(seqOf(" {\"seq\":5}") == null); // callers trim first
 }
+
+test "writeEventStamped uses the caller's id and never bumps the counter" {
+    resetForTest();
+    defer resetForTest();
+    var buf: [256]u8 = undefined;
+    var w: Io.Writer = .fixed(&buf);
+    try writeEventStamped(&w, 7, .{ .type = "text", .text = "hi" });
+    const line = w.buffered();
+    try std.testing.expect(std.mem.startsWith(u8, line, "{\"seq\":7,\"type\":\"text\""));
+    try std.testing.expectEqual(@as(u64, 7), seqOf(line).?);
+    // The stamped write reserved nothing: the next writeEvent still takes id 1.
+    try std.testing.expectEqual(@as(u64, 0), current());
+    var buf2: [256]u8 = undefined;
+    var w2: Io.Writer = .fixed(&buf2);
+    try writeEvent(&w2, .{ .type = "text" });
+    try std.testing.expectEqual(@as(u64, 1), seqOf(w2.buffered()).?);
+}
diff --git a/src/test_hooks.zig b/src/test_hooks.zig
index 7ac859363..cf8aeac5e 100644
--- a/src/test_hooks.zig
+++ b/src/test_hooks.zig
@@ -125,6 +125,11 @@ const snapshots_tests = @import("snapshots_tests.zig");
 // calls from oauth.zig and the catalog writers, so its own tests need the hook.
 const credential_store = @import("credential_store.zig");
 
+// #422: the engine event vocabulary + the sink boundary that dispatches it.
+// Production reaches both only through CALLS, so their tests need the hook.
+const engine_events = @import("engine_events.zig");
+const engine_sink = @import("engine_sink.zig");
+
 test {
     _ = learn_holdout;
     _ = learn_receipt;
@@ -162,6 +167,8 @@ test {
     _ = playbook_reflect;
     _ = shutdown_trace;
     _ = credential_store;
+    _ = engine_events;
+    _ = engine_sink;
     _ = escalation;
     _ = escalation_tests;
     _ = edit_contract;