Merge nucleic/eager-glass-civet-f3ph into dev

Sources/NucleicCore/AppStore.swift

@@ -1590,8 +1590,8 @@ public final class AppStore: ConflictArbiter {
         // shared listener so Build/Run spawns under nash can report. Everything stays lazy — the
         // listener binds no port until a spawn actually registers.
         Task {
-            await HostShellEventListener.shared.install { [weak self] origin, call in
-                await self?.observeAppScopedShellEvent(origin: origin, call: call)
+            await HostShellEventListener.shared.install { [weak self] origin, calls in
+                await self?.observeAppScopedShellEvents(origin: origin, calls: calls)
             }
         }
         defaultsChangeObserver = NotificationObserverToken(NotificationCenter.default.addObserver(
@@ -5793,19 +5793,43 @@ public final class AppStore: ConflictArbiter {
     ///
     /// While the shims and nash coexist, exec events dedupe naturally by feed. Purely observational
     /// for the non-git kinds.
-    public func observeShellEvent(sessionID: SessionID, call: MCPApprovalServer.ShellReportCall) async {
+    ///
+    /// Delivered one nash batch at a time (docs/NASH_STREAM_PERF_PLAN.md §P1): the viewer's feed
+    /// takes the whole batch in a single append, and only then is each event routed onward. Order
+    /// within the batch is nash's own, which is what keeps an `exec-start` ahead of the `exec` that
+    /// retires it.
+    public func observeShellEvents(
+        sessionID: SessionID, calls: [MCPApprovalServer.ShellReportCall]
+    ) async {
         // The Nash Viewer's feed, recorded first and for *every* kind — the routing below is lossy
         // by design (it forwards exec onto the git/command feeds and reduces the rest to a log
         // line), whereas the viewer's promise is the complete emission stream for this session.
-        nashEvents.record(sessionID: sessionID, call: call, at: now())
+        nashEvents.record(sessionID: sessionID, calls: calls, at: now())
+        for call in calls {
+            await routeShellEvent(sessionID: sessionID, call: call)
+        }
+    }
 
+    /// The per-event half of ``observeShellEvents(sessionID:calls:)``: everything downstream of the
+    /// viewer's feed, for one event. Split out so the feed can take the batch whole.
+    ///
+    /// The `CMDTRACE` lines here are at `.debug` rather than `.notice`
+    /// (docs/NASH_STREAM_PERF_PLAN.md §P1.3): nash emits one per pipe link, per redirect and per
+    /// `cd`, so a build storm writes thousands of them a minute per agent — a volume at which
+    /// os_log's public-interpolation formatting is itself measurable, and at which the trail is
+    /// unreadable anyway. The durable feed (`nash_event`) is the record that survives; these are the
+    /// live tail, and `log stream --level debug` still shows them. `.policy` keeps `.error`: it is
+    /// rare and never routine.
+    private func routeShellEvent(
+        sessionID: SessionID, call: MCPApprovalServer.ShellReportCall
+    ) async {
         switch call.kind {
         case .execStart(let argv, let cwd, _, _):
             // Announcement only (docs/NASH.md §5.1) — the command hasn't returned, so there is no
             // outcome to route: no git/gh op to record, nothing to classify for the command feed,
             // no residue to capture. The Nash Viewer's in-flight row above is the whole point of
             // it; here it is a trace line, and the `exec` that follows does the real work.
-            commandLog.notice(
+            commandLog.debug(
                 "CMDTRACE session=\(sessionID.rawValue, privacy: .public) src=nash kind=exec-start argv=\(argv.joined(separator: " "), privacy: .public) cwd=\(cwd, privacy: .public)")
         case .exec(let argv, let cwd, let exitCode, let durationMs, _):
             guard let head = argv.first else { return }
@@ -5824,16 +5848,16 @@ public final class AppStore: ConflictArbiter {
                     durationMs: durationMs, stdout: "", stderr: "", truncated: false,
                     source: .nash, sessionID: sessionID.rawValue))
         case .cd(let from, let to):
-            commandLog.notice(
+            commandLog.debug(
                 "CMDTRACE session=\(sessionID.rawValue, privacy: .public) src=nash kind=cd from=\(from, privacy: .public) to=\(to, privacy: .public)")
         case .export(let op, let names):
             // Names only — values are never reported by nash (docs/NASH.md §5.4).
-            commandLog.notice(
+            commandLog.debug(
                 "CMDTRACE session=\(sessionID.rawValue, privacy: .public) src=nash kind=\(op, privacy: .public) names=\(names.joined(separator: ","), privacy: .public)")
         case .redirect(let op, _, let target, let bytes, _, _, _):
             // Data-flow observation (docs/NASH.md §5.2). Feed rows that render the preview are a
             // follow-up (§9.5); for now the operator/target/size is logged.
-            commandLog.notice(
+            commandLog.debug(
                 "CMDTRACE session=\(sessionID.rawValue, privacy: .public) src=nash kind=redirect op=\(op, privacy: .public) target=\(target ?? "-", privacy: .public) bytes=\(bytes, privacy: .public)")
             // A write redirection is a file edit the agent made with raw shell. The approval gate
             // already locked the targets it could name statically; this is the ground-truth half —
@@ -5843,15 +5867,19 @@ public final class AppStore: ConflictArbiter {
                 await claimShellWrite(sessionID: sessionID, path: target)
             }
         case .cmdsub(let bytes, _, _, _):
-            commandLog.notice(
+            commandLog.debug(
                 "CMDTRACE session=\(sessionID.rawValue, privacy: .public) src=nash kind=cmdsub bytes=\(bytes, privacy: .public)")
         case .pipe(_, let fromIndex, let toIndex, _, _, let bytes, _, _, _):
-            commandLog.notice(
+            commandLog.debug(
                 "CMDTRACE session=\(sessionID.rawValue, privacy: .public) src=nash kind=pipe link=\(fromIndex, privacy: .public)->\(toIndex, privacy: .public) bytes=\(bytes, privacy: .public)")
         case .fallback(let reason, _):
+            // Rare and structural — nash gave up on an input and re-execed bash. Stays at `.notice`
+            // for the same reason `.policy` stays at `.error`: it is never part of the flood.
             commandLog.notice(
                 "CMDTRACE session=\(sessionID.rawValue, privacy: .public) src=nash kind=fallback reason=\(reason, privacy: .public)")
         case .dropped(let count):
+            // Back-pressure: the stream lost events. Also rare, and worth seeing without a debug
+            // filter, since it bounds how much the rest of this trail can be trusted.
             commandLog.notice(
                 "CMDTRACE session=\(sessionID.rawValue, privacy: .public) src=nash kind=dropped count=\(count, privacy: .public)")
         case .policy(let lever, let value, let action, _):
@@ -5862,25 +5890,30 @@ public final class AppStore: ConflictArbiter {
         }
     }
 
-    /// The app-scoped twin of ``observeShellEvent(sessionID:call:)`` (docs/NASH.md §7.7.3): shell
+    /// The app-scoped twin of ``observeShellEvents(sessionID:calls:)`` (docs/NASH.md §7.7.3): shell
     /// events from host spawns with no session — Build/Run today — delivered through
     /// ``HostShellEventListener`` and attributed by a **synthetic origin id**
     /// (`buildrun:<projectID>`) rather than a fake session. They join the Nash Viewer's feed under
     /// that id; none of the session-only semantics apply (git/gh convergence feeds a session's
     /// locks/autoship, residue capture targets a session's worktree — neither exists here), so the
     /// rest is the CMDTRACE log, same as the session path's non-exec kinds.
-    public func observeAppScopedShellEvent(
-        origin: String, call: MCPApprovalServer.ShellReportCall
+    ///
+    /// Batch-shaped for the same reason as the session path (docs/NASH_STREAM_PERF_PLAN.md §P1): a
+    /// Build & Run posts one body per 200 events, and the feed takes it in one append.
+    public func observeAppScopedShellEvents(
+        origin: String, calls: [MCPApprovalServer.ShellReportCall]
     ) async {
         nashEvents.record(
-            sessionID: SessionID(rawValue: origin), call: call,
+            sessionID: SessionID(rawValue: origin), calls: calls,
             environment: .app(
                 origin: origin, label: origin.hasPrefix("buildrun:") ? "Build & Run" : "App"),
             at: now())
-        if case .exec(let argv, let cwd, let exitCode, _, _) = call.kind {
-            commandLog.notice(
-                "CMDTRACE origin=\(origin, privacy: .public) src=nash kind=exec argv=\(argv.joined(separator: " "), privacy: .public) cwd=\(cwd, privacy: .public) exit=\(exitCode, privacy: .public)"
-            )
+        for call in calls {
+            if case .exec(let argv, let cwd, let exitCode, _, _) = call.kind {
+                commandLog.debug(
+                    "CMDTRACE origin=\(origin, privacy: .public) src=nash kind=exec argv=\(argv.joined(separator: " "), privacy: .public) cwd=\(cwd, privacy: .public) exit=\(exitCode, privacy: .public)"
+                )
+            }
         }
     }
 

Sources/NucleicCore/Claude/ClaudeCodeBackend.swift

@@ -923,14 +923,14 @@ public actor ClaudeCodeBackend: AgentBackend {
                 // The nash agent shell's exec gate reports to `/shell-event` under the same
                 // gate/token (docs/NASH.md §6) — ground-truth exec/cd/export/fallback events for
                 // the feeds. Same fire-and-forget contract.
-                await server.registerShellReport(token: token) { [weak self] call in
-                    Task { await self?.handleShellReport(call) }
+                await server.registerShellReport(token: token) { [weak self] calls in
+                    Task { await self?.handleShellReport(calls) }
                 }
             } else if run.container != nil {
                 // Non-Control sandbox runs do not install the git/gh/command shims, but their
                 // default `linux_container` children can still reuse this control socket for nash.
-                await server.registerShellReport(token: token) { [weak self] call in
-                    Task { await self?.handleShellReport(call) }
+                await server.registerShellReport(token: token) { [weak self] calls in
+                    Task { await self?.handleShellReport(calls) }
                 }
             }
             let mcpConfig = server.mcpConfigJSON(host: controlHost, port: port, token: token)
@@ -2850,10 +2850,8 @@ public actor ClaudeCodeBackend: AgentBackend {
         guard let manager = macVMManager, let sessionID, let conflictCoordinator else { return }
         let batches = await manager.drainShellEvents(name: vmName)
         for batch in batches {
-            for call in MCPApprovalServer.parseShellBatch(batch) {
-                await conflictCoordinator.observeShellEvent(
-                    sessionID: sessionID, call: call.attributed(to: environment))
-            }
+            let calls = MCPApprovalServer.parseShellBatch(batch).map { $0.attributed(to: environment) }
+            await conflictCoordinator.observeShellEvents(sessionID: sessionID, calls: calls)
         }
     }
 
@@ -3229,8 +3227,8 @@ public actor ClaudeCodeBackend: AgentBackend {
                 if spec.controlSocketHostPath != nil, let server = runServer,
                     let token = serverToken
                 {
-                    await server.registerShellReport(token: token) { [weak self] call in
-                        Task { await self?.handleShellReport(call) }
+                    await server.registerShellReport(token: token) { [weak self] calls in
+                        Task { await self?.handleShellReport(calls) }
                     }
                     env = CommandInterceptor.hookEnv(
                         host: "127.0.0.1", port: ContainerSpec.controlBridgePort,
@@ -3820,9 +3818,9 @@ public actor ClaudeCodeBackend: AgentBackend {
     /// `ConflictCoordinator` — the same seam as `handleCommandReport`. `AppStore` owns the
     /// semantics: converge exec events onto the command/git/gh feeds and log shell state changes
     /// (docs/NASH.md §9). Best-effort: dropped when there's no coordinator (pipeline tests).
-    private func handleShellReport(_ call: MCPApprovalServer.ShellReportCall) async {
+    private func handleShellReport(_ calls: [MCPApprovalServer.ShellReportCall]) async {
         guard let sessionID, let conflictCoordinator else { return }
-        await conflictCoordinator.observeShellEvent(sessionID: sessionID, call: call)
+        await conflictCoordinator.observeShellEvents(sessionID: sessionID, calls: calls)
     }
 
     /// NDJSON user-message envelope (ADAPTERS §1.3 — pinned by the M0 live capture).

Sources/NucleicCore/Claude/MCPApprovalServer.swift

@@ -788,7 +788,13 @@ public actor MCPApprovalServer {
     /// As ``GitReportHandler``, for the non-git command interceptor (`POST /command-event`).
     public typealias CommandReportHandler = @Sendable (CommandReportCall) -> Void
     /// As ``GitReportHandler``, for the nash agent shell's exec gate (`POST /shell-event`).
-    public typealias ShellReportHandler = @Sendable (ShellReportCall) -> Void
+    ///
+    /// Takes the **whole batch** rather than one call at a time (docs/NASH_STREAM_PERF_PLAN.md §P1):
+    /// nash posts up to 200 events per body, and a build storm posts thousands per minute across
+    /// every concurrent agent. Delivering per event made each one its own hop to the main actor —
+    /// and, because those hops were unstructured `Task`s, let a batch's rows land out of the order
+    /// nash emitted them (an `exec` overtaking its own `exec-start`). One call per batch fixes both.
+    public typealias ShellReportHandler = @Sendable ([ShellReportCall]) -> Void
 
     /// The dedicated POST path the `git` interceptor shim reports to — bearer-token gated like
     /// the MCP route, but a plain JSON POST rather than JSON-RPC.
@@ -3386,9 +3392,11 @@ public actor MCPApprovalServer {
 
     /// Handle a `POST /shell-event` from the nash agent shell (docs/NASH.md §6.1): bearer-token
     /// gated like `/git-event`. Body is a `shell-batch` (`{type, source:"nash", sessionId, shellId,
-    /// events:[…]}`), fanned out to one ``ShellReportCall`` per event. Fire-and-forget: the handler
-    /// returns immediately. Always 202s a known token (even on a malformed body) so nash never
-    /// retries or stalls; 401s an unknown one (which nash treats as "spool instead").
+    /// events:[…]}`), decoded to one ``ShellReportCall`` per event and delivered to the handler as
+    /// **one batch** — the ingest cost is per delivery, not per event, so a 200-event body must not
+    /// become 200 hops (docs/NASH_STREAM_PERF_PLAN.md §P1). Fire-and-forget: the handler returns
+    /// immediately. Always 202s a known token (even on a malformed body) so nash never retries or
+    /// stalls; 401s an unknown one (which nash treats as "spool instead").
     private func handleShellEvent(_ request: HTTPRequest) -> HTTPResponse {
         let authorization = request.headers["authorization"] ?? ""
         let token = authorization.hasPrefix("Bearer ") ? String(authorization.dropFirst(7)) : ""
@@ -3398,7 +3406,8 @@ public actor MCPApprovalServer {
                 body: Data(#"{"error":"invalid bearer token"}"#.utf8))
         }
         if let message = try? JSONValue(parsing: request.body) {
-            for call in Self.parseShellBatch(message) { handler(call) }
+            let calls = Self.parseShellBatch(message)
+            if !calls.isEmpty { handler(calls) }
         }
         return HTTPResponse(status: 202, statusText: "Accepted", contentType: nil, body: Data())
     }

Sources/NucleicCore/Codex/CodexAppServerBackend.swift

@@ -280,8 +280,8 @@ public actor CodexAppServerBackend: AgentBackend {
                     await server.registerCommandReport(token: token) { [weak self] call in
                         Task { await self?.handleCommandReport(call) }
                     }
-                    await server.registerShellReport(token: token) { [weak self] call in
-                        Task { await self?.handleShellReport(call) }
+                    await server.registerShellReport(token: token) { [weak self] calls in
+                        Task { await self?.handleShellReport(calls) }
                     }
                     await platformToolRuntime.configurePlatformToolRuntime(for: run) {
                         [weak self] kind, nativeType in
@@ -554,9 +554,9 @@ public actor CodexAppServerBackend: AgentBackend {
     }
 
     /// Forward a nash agent-shell event (`/shell-event`) for the feeds (docs/NASH.md §9).
-    private func handleShellReport(_ call: MCPApprovalServer.ShellReportCall) async {
+    private func handleShellReport(_ calls: [MCPApprovalServer.ShellReportCall]) async {
         guard let sessionID, let conflictCoordinator else { return }
-        await conflictCoordinator.observeShellEvent(sessionID: sessionID, call: call)
+        await conflictCoordinator.observeShellEvents(sessionID: sessionID, calls: calls)
     }
 
     private func interruptCurrentTurn() async {

Sources/NucleicCore/Codex/CodexExecBackend.swift

@@ -279,8 +279,8 @@ public actor CodexExecBackend: AgentBackend {
                         await server.registerCommandReport(token: token) { [weak self] call in
                             Task { await self?.handleCommandReport(call) }
                         }
-                        await server.registerShellReport(token: token) { [weak self] call in
-                            Task { await self?.handleShellReport(call) }
+                        await server.registerShellReport(token: token) { [weak self] calls in
+                            Task { await self?.handleShellReport(calls) }
                         }
                         env.merge(
                             CommandInterceptor.hookEnv(
@@ -433,9 +433,9 @@ public actor CodexExecBackend: AgentBackend {
         await conflictCoordinator.observeCommand(sessionID: sessionID, call: call)
     }
 
-    private func handleShellReport(_ call: MCPApprovalServer.ShellReportCall) async {
+    private func handleShellReport(_ calls: [MCPApprovalServer.ShellReportCall]) async {
         guard let sessionID, let conflictCoordinator else { return }
-        await conflictCoordinator.observeShellEvent(sessionID: sessionID, call: call)
+        await conflictCoordinator.observeShellEvents(sessionID: sessionID, calls: calls)
     }
 }
 

Sources/NucleicCore/Conflict.swift

@@ -218,10 +218,16 @@ public protocol ConflictArbiter: AnyObject, Sendable {
     /// shim or bash-tracer report). The arbiter classifies it and records it into the command
     /// activity feed + structured log. No-op outside Nucleic Control.
     @MainActor func observeCommand(sessionID: SessionID, call: MCPApprovalServer.CommandReportCall) async
-    /// A ground-truth event the nash agent shell observed for `sessionID` (an exec-gate report; the
-    /// `POST /shell-event` route). The arbiter converges exec events onto the command/git/gh feeds
-    /// (docs/NASH.md §9) and records shell state changes. No-op outside Nucleic Control.
-    @MainActor func observeShellEvent(sessionID: SessionID, call: MCPApprovalServer.ShellReportCall) async
+    /// One nash batch of ground-truth events observed for `sessionID` (the `POST /shell-event`
+    /// route, in the order nash emitted them). The arbiter converges exec events onto the
+    /// command/git/gh feeds (docs/NASH.md §9) and records shell state changes. No-op outside
+    /// Nucleic Control.
+    ///
+    /// Delivered a batch at a time because that is nash's own unit — up to 200 events per post —
+    /// and the arbiter's per-delivery costs (the actor hop, the viewer's ring append and its
+    /// observation invalidation) are what a build storm multiplies (docs/NASH_STREAM_PERF_PLAN.md
+    /// §P1). ``observeShellEvent(sessionID:call:)`` remains for single-event callers.
+    @MainActor func observeShellEvents(sessionID: SessionID, calls: [MCPApprovalServer.ShellReportCall]) async
     /// Try to claim a host-run slot for `sessionID` before a `host_exec` command runs on the shared
     /// host (HOST_EXEC §concurrency). Registers the command as running and returns `.clear` when no
     /// *similar* command is already running in another session (the fuzzy
@@ -247,6 +253,17 @@ public protocol ConflictArbiter: AnyObject, Sendable {
     @MainActor func dequeueHostCommandWait(sessionID: SessionID) async
 }
 
+extension ConflictArbiter {
+    /// One nash event on its own — the same delivery as ``observeShellEvents(sessionID:calls:)``
+    /// with a batch of one. For callers that genuinely have a single event (tests, a replay); the
+    /// live `/shell-event` route always delivers whole batches.
+    @MainActor public func observeShellEvent(
+        sessionID: SessionID, call: MCPApprovalServer.ShellReportCall
+    ) async {
+        await observeShellEvents(sessionID: sessionID, calls: [call])
+    }
+}
+
 /// Pure, deterministic conflict detection — the model-free core (mirrors
 /// `HeuristicTriage`). File-path overlap is the strong signal; a task-intent token
 /// overlap is a secondary net for when the agent can't predict the files up front.

Sources/NucleicCore/ConflictCoordinator.swift

@@ -91,9 +91,18 @@ public actor ConflictCoordinator {
         await arbiter?.observeCommand(sessionID: sessionID, call: call)
     }
 
-    /// Forward a nash agent-shell event to the arbiter. Best-effort, as above.
+    /// Forward one nash batch — every event of a `/shell-event` post, in order — to the arbiter.
+    /// Best-effort, as above.
+    public func observeShellEvents(
+        sessionID: SessionID, calls: [MCPApprovalServer.ShellReportCall]
+    ) async {
+        guard !calls.isEmpty else { return }
+        await arbiter?.observeShellEvents(sessionID: sessionID, calls: calls)
+    }
+
+    /// Forward a single nash agent-shell event to the arbiter. Best-effort, as above.
     public func observeShellEvent(sessionID: SessionID, call: MCPApprovalServer.ShellReportCall) async {
-        await arbiter?.observeShellEvent(sessionID: sessionID, call: call)
+        await arbiter?.observeShellEvents(sessionID: sessionID, calls: [call])
     }
 
     /// Try to claim a host-run slot before a `host_exec` command runs (HOST_EXEC §concurrency).

Sources/NucleicCore/Grok/GrokACPBackend.swift

@@ -294,8 +294,8 @@ public actor ACPBackend: AgentBackend {
                     await server.registerCommandReport(token: token) { [weak self] call in
                         Task { await self?.handleCommandReport(call) }
                     }
-                    await server.registerShellReport(token: token) { [weak self] call in
-                        Task { await self?.handleShellReport(call) }
+                    await server.registerShellReport(token: token) { [weak self] calls in
+                        Task { await self?.handleShellReport(calls) }
                     }
                     await nucleicToolRuntime.configurePlatformToolRuntime(for: run) {
                         [weak self] kind, nativeType in
@@ -571,9 +571,9 @@ public actor ACPBackend: AgentBackend {
     }
 
     /// Forward a nash agent-shell event (`/shell-event`) for the feeds (docs/NASH.md §9).
-    private func handleShellReport(_ call: MCPApprovalServer.ShellReportCall) async {
+    private func handleShellReport(_ calls: [MCPApprovalServer.ShellReportCall]) async {
         guard let sessionID, let conflictCoordinator else { return }
-        await conflictCoordinator.observeShellEvent(sessionID: sessionID, call: call)
+        await conflictCoordinator.observeShellEvents(sessionID: sessionID, calls: calls)
     }
 
     /// Cooperatively cancel the in-flight turn: ACP `session/cancel` (a notification); the agent

Sources/NucleicCore/HostShellEventListener.swift

@@ -22,8 +22,10 @@ import Foundation
 public actor HostShellEventListener {
     public static let shared = HostShellEventListener()
 
-    /// Delivery of one decoded shell event, tagged with the registering spawn's origin id.
-    public typealias Sink = @Sendable (String, MCPApprovalServer.ShellReportCall) async -> Void
+    /// Delivery of one decoded shell **batch**, tagged with the registering spawn's origin id.
+    /// Batch-shaped like the session route (docs/NASH_STREAM_PERF_PLAN.md §P1) — the per-delivery
+    /// cost is the actor hop and the viewer's ring append, so a 200-event post pays it once.
+    public typealias Sink = @Sendable (String, [MCPApprovalServer.ShellReportCall]) async -> Void
 
     private let server: MCPApprovalServer
     private var boundPort: UInt16?
@@ -54,8 +56,8 @@ public actor HostShellEventListener {
         guard let port = boundPort, port != 0 else { return nil }
         let token = UUID().uuidString
         origins[token] = origin
-        await server.registerShellReport(token: token) { [weak self] call in
-            Task { await self?.deliver(token: token, call: call) }
+        await server.registerShellReport(token: token) { [weak self] calls in
+            Task { await self?.deliver(token: token, calls: calls) }
         }
         return ("http://127.0.0.1:\(port)\(MCPApprovalServer.shellEventPath)", token)
     }
@@ -66,8 +68,8 @@ public actor HostShellEventListener {
         await server.unregister(token: token)
     }
 
-    private func deliver(token: String, call: MCPApprovalServer.ShellReportCall) async {
+    private func deliver(token: String, calls: [MCPApprovalServer.ShellReportCall]) async {
         guard let origin = origins[token], let sink else { return }
-        await sink(origin, call)
+        await sink(origin, calls)
     }
 }

Sources/NucleicCore/NashEventLog.swift

@@ -480,38 +480,76 @@ public final class NashEventLog {
     /// The one emission that does not simply append is a command's completion: when nash already
     /// announced that command as running (docs/NASH.md §5.1), the finished row **replaces** the
     /// in-flight one where it sits rather than landing as a second row for the same command — see
-    /// ``retireRunningRow(sessionID:call:event:)``.
+    /// ``append(_:sessionID:call:to:)``.
     @discardableResult
     public func record(
         sessionID: SessionID, call: MCPApprovalServer.ShellReportCall,
         environment: NashEnvironment? = nil, at: Date = Date()
     ) -> NashEvent {
-        // A running row is stamped with when its command *started*, not when we heard about it:
-        // nash waits out its announce threshold and then the batch waits out a flush, so a row
-        // clocked from arrival would show a build as a second younger than it is — and would keep
-        // that error for as long as the row counts up. The completion inherits this stamp, so the
-        // finished row also sits at the time the command began.
-        var at = at
-        if case .execStart(_, _, let elapsedMs, _) = call.kind {
-            at -= Double(elapsedMs) / 1000
+        record(sessionID: sessionID, calls: [call], environment: environment, at: at)[0]
+    }
+
+    /// Record one nash **batch** — every event of a `/shell-event` post, in the order nash emitted
+    /// them. Returns the rows as they landed, one per call.
+    ///
+    /// This is the shape the live route delivers (docs/NASH_STREAM_PERF_PLAN.md §P1): a batch of up
+    /// to 200 events costs *one* in-place mutation of the session's ring, one trim, and therefore
+    /// one `@Observable` invalidation, rather than paying all three per event. The ring is mutated
+    /// through ``withRing(_:_:)`` so appending never copies the 1000-row buffer.
+    @discardableResult
+    public func record(
+        sessionID: SessionID, calls: [MCPApprovalServer.ShellReportCall],
+        environment: NashEnvironment? = nil, at: Date = Date()
+    ) -> [NashEvent] {
+        guard !calls.isEmpty else { return [] }
+        var landed: [NashEvent] = []
+        landed.reserveCapacity(calls.count)
+        // Trimming is deferred to the end of the batch so the ring is walked once; the in-flight
+        // rows the trim invalidates are pruned after, outside the ring's exclusive access.
+        var trimmedFrom: UInt64?
+        withRing(sessionID) { bucket in
+            for call in calls {
+                // A running row is stamped with when its command *started*, not when we heard about
+                // it: nash waits out its announce threshold and then the batch waits out a flush, so
+                // a row clocked from arrival would show a build as a second younger than it is — and
+                // would keep that error for as long as the row counts up. The completion inherits
+                // this stamp, so the finished row also sits at the time the command began.
+                var at = at
+                if case .execStart(_, _, let elapsedMs, _) = call.kind {
+                    at -= Double(elapsedMs) / 1000
+                }
+                let event = Self.project(
+                    seq: nextSeq, at: at, sessionID: sessionID,
+                    environment: environment ?? call.environment ?? .agent,
+                    shellID: call.shellID ?? "", kind: call.kind)
+                landed.append(append(event, sessionID: sessionID, call: call, to: &bucket))
+            }
+            if bucket.count > capacity {
+                bucket.removeFirst(bucket.count - capacity)
+                trimmedFrom = bucket.first?.seq ?? 0
+            }
         }
-        let event = Self.project(
-            seq: nextSeq, at: at, sessionID: sessionID,
-            environment: environment ?? call.environment ?? .agent, shellID: call.shellID ?? "",
-            kind: call.kind)
+        if let trimmedFrom { pruneRunningRows(sessionID: sessionID, below: trimmedFrom) }
+        return landed
+    }
+
+    /// Land one projected row in `bucket` — appending it, or folding it into the in-flight row it
+    /// completes — and do the bookkeeping that follows. Returns the row as it landed.
+    ///
+    /// Takes the ring `inout` rather than reaching for `eventsBySession`, because its caller is
+    /// already holding that property under exclusive access (see ``withRing(_:_:)``); everything
+    /// else it touches is a different property.
+    private func append(
+        _ event: NashEvent, sessionID: SessionID, call: MCPApprovalServer.ShellReportCall,
+        to bucket: inout [NashEvent]
+    ) -> NashEvent {
         if case .exec = call.kind,
-            let landed = retireRunningRow(sessionID: sessionID, call: call, event: event)
+            let landed = retireRunningRow(sessionID: sessionID, call: call, event: event, in: &bucket)
         {
             return landed
         }
         nextSeq += 1
-        var bucket = eventsBySession[sessionID] ?? []
         bucket.append(event)
-        if bucket.count > capacity {
-            bucket.removeFirst(bucket.count - capacity)
-            pruneRunningRows(sessionID: sessionID, below: bucket.first?.seq ?? 0)
-        }
-        eventsBySession[sessionID] = bucket
         environmentsBySession[sessionID, default: []].insert(event.environment)
         // An in-flight row is a *promise* of a row, not one of its own: it is not counted (its
         // completion is the emission nash made), not written to the durable feed (a running row
@@ -530,6 +568,18 @@ public final class NashEventLog {
         return event
     }
 
+    /// Mutate one session's ring **in place**.
+    ///
+    /// `&eventsBySession[sessionID, default: []]` threads `_modify` accessors all the way down — the
+    /// `@Observable` property's and the dictionary subscript's — so the body writes straight into
+    /// the stored buffer: no copy-on-write of the 1000-row array, and exactly one
+    /// willSet/didSet pair for the whole body however many rows it appends. Reading or writing
+    /// `eventsBySession` from inside `body` would overlap that exclusive access, so the body is
+    /// handed the ring and must use it.
+    private func withRing<R>(_ sessionID: SessionID, _ body: (inout [NashEvent]) -> R) -> R {
+        body(&eventsBySession[sessionID, default: []])
+    }
+
     /// Identity of one command run in flight: nash's per-shell gate token, scoped by the shell that
     /// issued it (the token counter restarts in every nash process, exactly like the pipeline one).
     private struct RunningKey: Hashable {
@@ -559,20 +609,22 @@ public final class NashEventLog {
     ///
     /// A row the ring has already trimmed away is *not* resurrected: the key is dropped and the
     /// completion appends like any other, because a row nobody can see is not one to update.
+    ///
+    /// `bucket` is the session's ring, held `inout` by the batch that is recording into it — the
+    /// row this folds into may well have been appended by an earlier call of the same batch.
     private func retireRunningRow(
-        sessionID: SessionID, call: MCPApprovalServer.ShellReportCall, event: NashEvent
+        sessionID: SessionID, call: MCPApprovalServer.ShellReportCall, event: NashEvent,
+        in bucket: inout [NashEvent]
     ) -> NashEvent? {
         guard let key = runningKey(sessionID: sessionID, call: call),
             let seq = runningRows.removeValue(forKey: key),
-            var bucket = eventsBySession[key.sessionID],
             // Newest-last, and a running row is by definition recent, so this finds it immediately.
             let index = bucket.lastIndex(where: { $0.seq == seq })
         else { return nil }
         let landed = bucket[index].completed(by: event)
         bucket[index] = landed
-        eventsBySession[key.sessionID] = bucket
-        indexPipelineStage(landed, sessionID: key.sessionID)
-        totalsBySession[key.sessionID, default: [:]][landed.kind, default: 0] += 1
+        indexPipelineStage(landed, sessionID: sessionID)
+        totalsBySession[sessionID, default: [:]][landed.kind, default: 0] += 1
         enqueue(landed)
         return landed
     }

Tests/NucleicCoreTests/ConflictDetectorTests.swift

@@ -271,7 +271,7 @@ private final class StubArbiter: ConflictArbiter {
     func observeGitOp(sessionID: SessionID, argv: [String], exitCode: Int) async {}
     func observeGhOp(sessionID: SessionID, argv: [String], exitCode: Int) async {}
     func observeCommand(sessionID: SessionID, call: MCPApprovalServer.CommandReportCall) async {}
-    func observeShellEvent(sessionID: SessionID, call: MCPApprovalServer.ShellReportCall) async {}
+    func observeShellEvents(sessionID: SessionID, calls: [MCPApprovalServer.ShellReportCall]) async {}
     func beginHostCommand(sessionID: SessionID, command: String, override: Bool) async -> HostCommandClearance {
         .clear
     }

Tests/NucleicCoreTests/HostShellEventListenerTests.swift

@@ -39,9 +39,9 @@ import Testing
         #expect(await listener.register(origin: "buildrun:p1") == nil)
 
         // A tiny async inbox so the test can await delivery without polling shared state.
-        let inbox = AsyncStream.makeStream(of: (String, MCPApprovalServer.ShellReportCall).self)
-        await listener.install { origin, call in
-            inbox.continuation.yield((origin, call))
+        let inbox = AsyncStream.makeStream(of: (String, [MCPApprovalServer.ShellReportCall]).self)
+        await listener.install { origin, calls in
+            inbox.continuation.yield((origin, calls))
         }
 
         let registration = try #require(await listener.register(origin: "buildrun:p1"))
@@ -57,8 +57,9 @@ import Testing
         var iterator = inbox.stream.makeAsyncIterator()
         let delivered = try #require(await iterator.next())
         #expect(delivered.0 == "buildrun:p1")
+        #expect(delivered.1.count == 1)
         #expect(
-            delivered.1.kind
+            delivered.1.first?.kind
                 == .exec(
                     argv: ["swift", "build"], cwd: "/proj", exitCode: 0, durationMs: 5,
                     pipeline: nil))
@@ -71,9 +72,9 @@ import Testing
 
     @Test func concurrentSpawnsKeepDistinctOrigins() async throws {
         let listener = HostShellEventListener()
-        let inbox = AsyncStream.makeStream(of: (String, MCPApprovalServer.ShellReportCall).self)
-        await listener.install { origin, call in
-            inbox.continuation.yield((origin, call))
+        let inbox = AsyncStream.makeStream(of: (String, [MCPApprovalServer.ShellReportCall]).self)
+        await listener.install { origin, calls in
+            inbox.continuation.yield((origin, calls))
         }
         let a = try #require(await listener.register(origin: "buildrun:alpha"))
         let b = try #require(await listener.register(origin: "buildrun:beta"))

Tests/NucleicCoreTests/NashEventLogTests.swift

@@ -304,6 +304,66 @@ import Testing
         #expect(log.stages(for: session, pipeline: .init(shellID: "sh-1", pipelineID: 1)).isEmpty)
         #expect(log.stages(for: session, pipeline: .init(shellID: "sh-1", pipelineID: 3))[0] != nil)
     }
+
+    /// A whole nash batch lands in one pass (docs/NASH_STREAM_PERF_PLAN.md §P1) and has to be
+    /// indistinguishable from the same events recorded one at a time: order preserved, the trim
+    /// applied once at the end, and a completion still folding into the announcement it retires —
+    /// including when both arrive in the *same* batch, which is what a 500 ms flush makes common.
+    @MainActor
+    @Test func recordsAWholeBatchInOnePass() {
+        let log = NashEventLog(capacity: 3)
+        func call(_ kind: MCPApprovalServer.ShellReportCall.Kind, token: Int? = nil)
+            -> MCPApprovalServer.ShellReportCall
+        {
+            .init(kind: kind, sessionID: session.rawValue, shellID: "sh-1", execToken: token)
+        }
+        let landed = log.record(
+            sessionID: session,
+            calls: [
+                call(.cd(from: "/", to: "/a")),
+                call(.execStart(argv: ["make"], cwd: "/a", elapsedMs: 450, pipeline: nil), token: 7),
+                call(.cd(from: "/a", to: "/b")),
+                call(
+                    .exec(argv: ["make"], cwd: "/a", exitCode: 0, durationMs: 900, pipeline: nil),
+                    token: 7),
+                call(.dropped(count: 2)),
+            ], at: at)
+
+        // Five calls in, five rows back — but the completion folded into the announcement, so the
+        // ring holds four, and the trim to `capacity` ran once over the finished batch.
+        #expect(landed.count == 5)
+        #expect(landed[1].id == landed[3].id)
+        #expect(log.events(for: session).map(\.summary) == ["make", "cd /b", "dropped 2 events under back-pressure"])
+        #expect(log.events(for: session).map(\.isRunning) == [false, false, false])
+        #expect(log.events(for: session)[0].detail == "in /a · 900ms")
+        // The trimmed `cd /a` still counts, and the announcement still counts only once.
+        #expect(log.total(for: session, kind: .cd) == 2)
+        #expect(log.total(for: session, kind: .exec) == 1)
+        #expect(log.total(for: session, kind: .dropped) == 1)
+    }
+
+    /// An announcement whose completion arrives in a *later* batch still finds its row — the ring
+    /// the batch mutates is the same one the previous batch appended to.
+    @MainActor
+    @Test func batchesJoinAcrossDeliveries() {
+        let log = NashEventLog()
+        log.record(
+            sessionID: session,
+            calls: [
+                .init(
+                    kind: .execStart(argv: ["make"], cwd: "/w", elapsedMs: 450, pipeline: nil),
+                    sessionID: session.rawValue, shellID: "sh-1", execToken: 7)
+            ], at: at)
+        log.record(
+            sessionID: session,
+            calls: [
+                .init(
+                    kind: .exec(argv: ["make"], cwd: "/w", exitCode: 2, durationMs: 9000, pipeline: nil),
+                    sessionID: session.rawValue, shellID: "sh-1", execToken: 7)
+            ], at: at + 9)
+        #expect(log.events(for: session).count == 1)
+        #expect(log.events(for: session)[0].detail == "in /w · exit 2 · 9.0s")
+    }
 }
 
 /// The durable half of the feed (docs/NASH.md §9.4): rows written through `NashEventPersisting`

docs/NASH.md

@@ -545,6 +545,14 @@ toText, bytes, truncated, preview, hash}` via mechanism (b). This surfaces what
 transcript never shows today — the intermediate data between stages (`curl … | sh`
 becomes *visible*).
 
+**How the copy runs** (docs/NASH_STREAM_PERF_PLAN.md §P2): only the captured prefix passes
+through userspace. Once `CAP` bytes are mirrored the copier switches to `splice(2)` on Linux —
+pipe-to-pipe inside the kernel, with the byte count coming back from the same call that moves
+them — and falls back to a large-buffer read/write loop on macOS or wherever the kernel refuses
+the splice. Both ends of a tapped link are grown with `F_SETPIPE_SZ` where that is allowed, and
+the copier threads are pooled per process rather than spawned per link, since thread creation was
+most of what a tapped pipeline cost to *set up*.
+
 `fromText`/`toText` name the commands either side of the link, rendered from their AST
 when the tee is spawned. They are carried on the link rather than inferred from the
 stages' `exec` events for two reasons: a stage that is a compound command
@@ -568,11 +576,23 @@ double event volume for no new signal.
 ### 5.4 Caps, redaction, volume
 
 - Per-event preview cap (64 KiB), per-command total capture cap (256 KiB), per-batch cap
-  (1 MiB); beyond caps → counts/hashes only, `truncated: true`.
+  (1 MiB); beyond caps → counts/hashes only, `truncated: true`. The per-batch cap is spent in
+  arrival order, so a pipeline storm loses its *tail* of previews and never an event: every link
+  still reports its byte count and its hash.
+- Hashes cover the **captured prefix**, uniformly across the three data-flow kinds. A pipe's
+  full stream is never buffered, so it never could cover more; making the file and cmdsub kinds
+  agree keeps the marker's meaning single ("these previews carried the same bytes") and keeps
+  the hash off the paths a command waits on.
 - Redaction runs *inside nash before bytes leave the guest*: the same posture as
   [OBSERVABILITY_AND_TESTING.md](OBSERVABILITY_AND_TESTING.md) — pattern-based masking of
   obvious credential shapes (bearer/PAT/AWS-style tokens, `PRIVATE KEY` blocks) in previews;
   env vars are **never** captured wholesale (exec events carry argv, not environment).
+- **Where that work runs** (docs/NASH_STREAM_PERF_PLAN.md §P3): a command's own path pays only
+  for *capture* — the bytes, plus one `stat` to fix which range of a redirected file this command
+  wrote. Reading that range back, redacting, base64-ing and hashing all happen on the flusher
+  thread as the batch is serialized. The range is measured at exit rather than at flush precisely
+  so that `echo a > f; echo b >> f` still reports one line each: the reading is deferred, the
+  measurement never is.
 - Environment mutations are **names only** (locked decision): `export`/`declare`/`unset`
   produce events carrying the variable name (`export AWS_SECRET_ACCESS_KEY`) and never the
   value — the name alone is the signal that a credential-shaped variable was set, with zero
@@ -705,6 +725,12 @@ is — the event model is the same, the liveness is a property of the transport.
 Auth is the existing per-session bearer token (`NUCLEIC_HOOK_TOKEN`); unknown tokens get
 401 and nash stops posting (spools) for that process.
 
+The connection is **kept open** across batches (docs/NASH_STREAM_PERF_PLAN.md §P4.1): a shell
+posts every 500 ms for as long as it lives, and the host's route already answers `Connection:
+keep-alive` with a `Content-Length`. nash reads each response to its end, so the socket stays
+framed for the next batch; a connection the host has since closed costs one retry on a fresh one,
+after which the ordinary spool fallback applies.
+
 ---
 
 ## 7. Forcing nash as the default shell — per surface
@@ -1074,7 +1100,14 @@ not scheduled work in this plan.
 ## 9. Host-side integration (Swift)
 
 Follows the existing report pipeline shape exactly (shim → route → struct → backend →
-coordinator → AppStore → feed):
+coordinator → AppStore → feed).
+
+**The unit of delivery is the batch, not the event** (docs/NASH_STREAM_PERF_PLAN.md §P1). One
+POST carries up to 200 events, and every per-delivery cost the host pays — the unstructured
+`Task`, the hop to the main actor, the viewer's ring append and its `@Observable` invalidation —
+is paid once for the whole batch. Delivering per event also silently reordered a batch, because
+unstructured `Task`s do not run in creation order: an `exec` could land ahead of the `exec-start`
+it retires. Handlers therefore take `[ShellReportCall]`, in nash's own order.
 
 1. **Route**: `shellEventPath = "/shell-event"` in `MCPApprovalServer.swift` (beside
    `gitEventPath:649`); dispatch beside lines 1752-1760; `handleShellEvent` parses
@@ -1084,9 +1117,9 @@ coordinator → AppStore → feed):
 2. **Registration**: `registerShellReport(token:handler:)` beside
    `registerCommandReport` (1410-1425); wired in all four backends
    (`ClaudeCodeBackend.swift:2721-2744` and Codex/Grok twins).
-3. **Coordinator**: `ConflictCoordinator.observeShellEvent(sessionID:call:)` forwarding to
+3. **Coordinator**: `ConflictCoordinator.observeShellEvents(sessionID:calls:)` forwarding to
    `AppStore`.
-4. **AppStore**: `observeShellEvent` —
+4. **AppStore**: `observeShellEvents` —
    - `exec` events flow into the existing `observeCommand` path (`AppStore.swift:4666`)
      via `CommandSummary.classify(argv:)`, tagged `source: .nash`, preserving the NVRSION
      residue-capture backstop (4679-4689) and `CMDTRACE` logging;

docs/NASH_STREAM_PERF_PLAN.md

@@ -1,7 +1,11 @@
 # Nash stream performance: findings & optimization plan
 
-*Investigated 2026-07-29. Companion to [NASH.md](NASH.md) (§5 data-flow taps, §6 transport,
-§9 host ingest) and [MAIN_THREAD_PERFORMANCE_PLAN.md](MAIN_THREAD_PERFORMANCE_PLAN.md).*
+*Investigated 2026-07-29; P1–P4 and the regression guard implemented 2026-07-29. Companion to
+[NASH.md](NASH.md) (§5 data-flow taps, §6 transport, §9 host ingest) and
+[MAIN_THREAD_PERFORMANCE_PLAN.md](MAIN_THREAD_PERFORMANCE_PLAN.md).*
+
+**Status: all four phases shipped.** What landed, and where it deviates from the plan below, is
+recorded in [§4 Outcome](#4-outcome).
 
 Nash's observation stream was suspected of significantly affecting performance. This doc
 records what was measured, where the cost actually is, and a prioritized plan to reduce it.
@@ -143,3 +147,71 @@ Extend `shell/corpus/overhead.py`:
 
 - `on_cmdsub` coverage gap for substitutions in assignments (`x=$(…)`).
 - Brush's ~+35% bare-exec baseline vs bash (spawn-path work, independent of observation).
+
+---
+
+## 4. Outcome
+
+### What shipped
+
+**P1 — batch-preserving host ingest.** `ShellReportHandler` takes `[ShellReportCall]`;
+`handleShellEvent` delivers one parsed batch per POST, through one `Task` and one MainActor hop,
+to `ConflictArbiter.observeShellEvents(sessionID:calls:)` →
+`AppStore.observeShellEvents`. `NashEventLog.record(sessionID:calls:…)` lands the whole batch in
+a single in-place mutation of the session ring (`withRing`, which threads `_modify` through the
+`@Observable` property and the dictionary subscript), with one trim and one invalidation for the
+batch instead of per event — and no copy-on-write of the 1000-row buffer, which the old
+read-modify-write did on *every* event. `HostShellEventListener`'s app-scoped sink and the VM
+spool drain are batch-shaped too. Per-event `CMDTRACE` lines dropped to `.debug`; `fallback`,
+`dropped` and `policy` — the rare ones — kept their levels.
+
+**P2 — cheaper tee.** After the 64 KiB prefix, a tapped link is moved with `splice(2)` on Linux
+(pipe-to-pipe in the kernel, byte count from the same call), falling back to a 128 KiB
+read/write loop on macOS or on `EINVAL`/`ENOSYS`/`EPERM`. Both pipes are grown with
+`F_SETPIPE_SZ` where permitted, and copier threads are pooled per process (≤4 idle, 512 KiB
+stacks) instead of spawned per link — with a pid check so a `fork` can never hand a job to a
+worker that only exists in the parent.
+
+**P3 — encoding off the hot paths.** Data-flow events carry a `Capture` — raw bytes, or a
+measured-but-unread file range — and `Serialize` does the redaction, base64 and FNV hash, which
+means all of it runs on the flusher thread. Binary hex previews are table-driven.
+
+**P4 — transport.** One connection, kept open across every batch of a shell's life
+(`Connection: keep-alive`, responses drained to their `Content-Length` so the socket stays
+framed, one retry on a connection the host has since closed). Plus NASH.md §5.4's per-batch
+preview budget, which was specified but never implemented: 1 MiB of payload per batch, spent in
+arrival order, beyond which events keep their counts and hashes and give up their previews.
+
+**Regression guard.** `overhead.py` reports CPU (rusage) beside wall for every case, adds
+`--stream-mb` throughput cases (one and two links) gated on CPU, defaults `--bash` to
+`/usr/bin/bash.real`, and refuses a baseline that reports `NUCLEIC_NASH=1` (probed with a
+scrubbed environment, since the harness itself usually runs *under* nash).
+
+### Measured (aarch64 Linux container, release build, observed → spool)
+
+| Case | before | after |
+|---|---|---|
+| 256 MB through `a\|b` | +0.2% wall / **+11.9% CPU** | −22.4% wall / **−11.3% CPU** |
+| 256 MB through `a\|b\|c` | +29.8% wall / **+100.9% CPU** | −35.8% wall / **+11.4% CPU** |
+| 300 tiny pipelines | 0.220 s wall / 0.328 s CPU | 0.209 s wall / 0.309 s CPU |
+
+The two-link stream case — the one that doubled CPU — now costs about a tenth of that, and both
+stream cases run *faster than bash* in wall time, because the tee's extra pipe adds pipelining
+the kernel now moves for free. The pipeline-storm case improves ~5% from the thread pool; most of
+what remains there is brush's spawn baseline, which is the separate workstream noted above.
+
+### Deviations from the plan
+
+- **P3.2 measures at exit, reads at flush.** Moving the redirect read-back wholesale to the
+  flusher is wrong, not merely stale: `echo a > f; echo b >> f` flushes as one batch, and by then
+  the truncating write's read-back would report both lines. The command path keeps one `stat`
+  (which also answers "is this a regular file at all"), fixing the range; everything
+  else — open, read, hash, redact, encode — moved. Guarded by
+  `deferred_readback_reads_the_measured_range`.
+- **Hashes are uniformly over the captured prefix.** `cmdsub` used to hash its *entire* output,
+  which meant `x=$(cat 32MB)` hashed 32 MB on the interpreter thread. Carrying those bytes to the
+  flusher to preserve that would have cost more than the hash. Pipes already hashed the prefix;
+  files never exceed it. NASH.md §5.4 now says so.
+- **P1.4 (UI coalescing) not done, and not needed yet.** Batching already reduced the viewer's
+  invalidations to one per POST — at most two per second per shell, which SwiftUI absorbs
+  without a rate limiter. Worth revisiting only if a profile says otherwise.

package-lock.json

@@ -0,0 +1,6 @@
+{
+  "name": "nucleic",
+  "lockfileVersion": 3,
+  "requires": true,
+  "packages": {}
+}

shell/corpus/overhead.py

@@ -1,22 +1,44 @@
 #!/usr/bin/env python3
-"""Corpus wall-clock overhead harness for nash (docs/NASH.md §11, M2/M3 gate: <3%).
+"""Corpus overhead harness for nash (docs/NASH.md §11, M2/M3 gate: <3%).
 
 Replays the corpus under bash and nash in alternating rounds and compares total
-wall-clock per shell. Only the shell subprocess is timed (fixture seeding and
-filesystem snapshots are outside the clock). Rounds alternate shell order so
-cache/thermal drift cancels; the reported figure uses the median round total.
+wall-clock **and CPU time** per shell. Only the shell subprocess is measured
+(fixture seeding and filesystem snapshots are outside both clocks). Rounds
+alternate shell order so cache/thermal drift cancels; the reported figures use
+the median round total.
+
+Two measurements, because they answer different questions
+(docs/NASH_STREAM_PERF_PLAN.md §2):
+
+  * **wall** is what a single command waits for — the M3 gate.
+  * **cpu** (user+sys of the shell and everything it spawned, via `rusage`) is
+    what nash actually spends. The pipe tee runs on its own thread, so on idle
+    hardware it can add CPU while *costing no wall time at all* — and a
+    wall-only harness would report that as free. It is not: the deployed box
+    runs many agents at once, where that CPU comes out of everyone's clock.
+
+Beyond the corpus, `--stream-mb` runs a **throughput** case — hundreds of MB
+across one and two pipe links — which is the shape the tee is optimized for and
+the one the corpus (a few KB per command) cannot see.
 
 Usage: overhead.py [--nash PATH] [--bash PATH] [--corpus PATH] [--rounds N]
-                   [--observe-spool DIR] [--gate PCT] [--json PATH]
+                   [--observe-spool DIR] [--gate PCT] [--stream-mb MB]
+                   [--stream-gate PCT] [--json PATH] [--allow-nash-baseline]
 
 --observe-spool enables nash observation (spool transport) so the measured
 configuration is the deployed one; the env is set identically for bash, where
 it is inert.
+
+The baseline shell must be a *real* bash: on a Nucleic-managed box `/bin/bash`
+IS nash (docs/NASH.md §7), so the default baseline is `/usr/bin/bash.real` and
+a baseline that reports `NUCLEIC_NASH=1` is refused outright — benchmarking
+nash against itself reports ~0% overhead and means nothing.
 """
 
 import argparse
 import json
 import os
+import resource
 import shutil
 import statistics
 import subprocess
@@ -26,12 +48,30 @@ import time
 
 from replay import FIXTURES, TIMEOUT_S, seed
 
+#: Where a Nucleic-managed image keeps the real bash after the nash divert.
+DEFAULT_BASH = "/usr/bin/bash.real"
 
-def run_timed(shell, cmd, extra_env):
-    parent = tempfile.mkdtemp(prefix="nash-overhead-")
-    workdir = os.path.join(parent, "workspace")
-    os.makedirs(workdir)
-    seed(workdir)
+
+def child_cpu():
+    """User+sys seconds of every child this process has reaped."""
+    usage = resource.getrusage(resource.RUSAGE_CHILDREN)
+    return usage.ru_utime + usage.ru_stime
+
+
+def run_timed(shell, cmd, extra_env, workdir=None):
+    """Run one command, returning (wall_seconds, cpu_seconds).
+
+    CPU comes from the delta of `RUSAGE_CHILDREN`, which only counts children
+    already reaped — `subprocess.run` waits, and the harness runs one child at a
+    time, so the delta is exactly this command's shell and its descendants
+    (including nash's tee and flusher threads).
+    """
+    parent = None
+    if workdir is None:
+        parent = tempfile.mkdtemp(prefix="nash-overhead-")
+        workdir = os.path.join(parent, "workspace")
+        os.makedirs(workdir)
+        seed(workdir)
     env = {
         "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
         "HOME": workdir,
@@ -42,6 +82,7 @@ def run_timed(shell, cmd, extra_env):
     }
     if extra_env:
         env.update(extra_env)
+    cpu_before = child_cpu()
     start = time.monotonic()
     try:
         subprocess.run(
@@ -54,24 +95,136 @@ def run_timed(shell, cmd, extra_env):
     except subprocess.TimeoutExpired:
         pass
     elapsed = time.monotonic() - start
-    shutil.rmtree(parent, ignore_errors=True)
-    return elapsed
+    cpu = child_cpu() - cpu_before
+    if parent:
+        shutil.rmtree(parent, ignore_errors=True)
+    return elapsed, cpu
+
+
+def is_nash(shell):
+    """Whether `shell` is nash wearing another name (docs/NASH.md §2).
+
+    Probed with a scrubbed environment on purpose: the harness itself is very
+    likely running *under* nash, which exports `NUCLEIC_NASH=1` to everything it
+    spawns — inheriting that would make every shell look like nash. Only a shell
+    that sets the variable for itself answers 1 here.
+    """
+    try:
+        out = subprocess.run(
+            [shell, "-c", 'printf %s "${NUCLEIC_NASH-}"'],
+            capture_output=True,
+            timeout=30,
+            text=True,
+            env={"PATH": "/usr/bin:/bin"},
+        )
+    except (OSError, subprocess.SubprocessError):
+        return False
+    return out.stdout.strip() == "1"
+
+
+def stream_cases(megabytes):
+    """Throughput scripts: the same payload across one link and across two.
+
+    `/dev/zero → /dev/null` deliberately: the point is the cost of *carrying*
+    bytes across a tapped link, so neither end should be doing work of its own.
+    """
+    count = megabytes * 1024 * 1024
+    return [
+        (f"1 link ({megabytes} MB)", f"head -c {count} /dev/zero | cat > /dev/null"),
+        (
+            f"2 links ({megabytes} MB)",
+            f"head -c {count} /dev/zero | cat | cat > /dev/null",
+        ),
+    ]
+
+
+def measure(shell, cmds, extra_env):
+    """Total (wall, cpu) for one pass over `cmds`."""
+    wall = cpu = 0.0
+    for cmd in cmds:
+        w, c = run_timed(shell, cmd, extra_env)
+        wall += w
+        cpu += c
+    return wall, cpu
+
+
+def percent(nash, bash):
+    """nash's cost over bash's, in percent. Infinite-safe for a zero baseline."""
+    return 100.0 * (nash / bash - 1.0) if bash > 0 else float("nan")
+
+
+def report(label, bash_vals, nash_vals, gate=None):
+    """Print (and return) one comparison's medians and percentages."""
+    bash_wall = statistics.median(w for w, _ in bash_vals)
+    bash_cpu = statistics.median(c for _, c in bash_vals)
+    nash_wall = statistics.median(w for w, _ in nash_vals)
+    nash_cpu = statistics.median(c for _, c in nash_vals)
+    result = {
+        "bash_wall_s": bash_wall,
+        "bash_cpu_s": bash_cpu,
+        "nash_wall_s": nash_wall,
+        "nash_cpu_s": nash_cpu,
+        "wall_percent": percent(nash_wall, bash_wall),
+        "cpu_percent": percent(nash_cpu, bash_cpu),
+    }
+    print(f"\n{label}")
+    print(f"  bash: {bash_wall:.3f}s wall / {bash_cpu:.3f}s cpu")
+    print(f"  nash: {nash_wall:.3f}s wall / {nash_cpu:.3f}s cpu")
+    suffix = f"   (gate: <{gate:.1f}%)" if gate is not None else ""
+    print(
+        f"  overhead: {result['wall_percent']:+.2f}% wall / "
+        f"{result['cpu_percent']:+.2f}% cpu{suffix}"
+    )
+    return result
 
 
 def main():
     ap = argparse.ArgumentParser()
     ap.add_argument("--nash", default=os.environ.get("NASH_BIN", "nash"))
-    ap.add_argument("--bash", default="/bin/bash")
+    ap.add_argument(
+        "--bash",
+        default=DEFAULT_BASH,
+        help=f"baseline shell — must not be nash (default {DEFAULT_BASH})",
+    )
     ap.add_argument(
         "--corpus",
         default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "corpus.jsonl"),
     )
     ap.add_argument("--rounds", type=int, default=5)
     ap.add_argument("--observe-spool", help="enable nash observation, spooling to this dir")
-    ap.add_argument("--gate", type=float, default=3.0, help="max overhead percent")
+    ap.add_argument("--gate", type=float, default=3.0, help="max corpus wall overhead percent")
+    ap.add_argument(
+        "--stream-mb",
+        type=int,
+        default=256,
+        help="payload per throughput case (0 disables the stream cases)",
+    )
+    ap.add_argument(
+        "--stream-gate",
+        type=float,
+        default=50.0,
+        help="max stream CPU overhead percent (the tee's own cost)",
+    )
+    ap.add_argument(
+        "--allow-nash-baseline",
+        action="store_true",
+        help="benchmark against a nash baseline anyway (produces meaningless numbers)",
+    )
     ap.add_argument("--json", help="also write results to this path")
     args = ap.parse_args()
 
+    # A baseline that is itself nash makes every number here ~0% and hides whatever
+    # regressed. On a Nucleic box that is the *default* state of /bin/bash, so this is
+    # a refusal rather than a warning (docs/NASH_STREAM_PERF_PLAN.md §regression guard).
+    if is_nash(args.bash) and not args.allow_nash_baseline:
+        print(
+            f"error: baseline shell {args.bash} reports NUCLEIC_NASH=1 — it IS nash.\n"
+            f"       Point --bash at the real bash ({DEFAULT_BASH} on a Nucleic image),\n"
+            "       or pass --allow-nash-baseline if you really mean to compare nash to nash.",
+            file=sys.stderr,
+        )
+        return 2
+
     observe_env = None
     if args.observe_spool:
         os.makedirs(args.observe_spool, exist_ok=True)
@@ -82,6 +235,7 @@ def main():
 
     with open(args.corpus) as f:
         cmds = [json.loads(line)["cmd"] for line in f if line.strip()]
+    streams = stream_cases(args.stream_mb) if args.stream_mb > 0 else []
 
     # Warm-up: one untimed pass per shell (page cache, binary load).
     for shell in (args.bash, args.nash):
@@ -89,22 +243,33 @@ def main():
             run_timed(shell, cmd, observe_env)
 
     totals = {"bash": [], "nash": []}
+    stream_totals = {name: {"bash": [], "nash": []} for name, _ in streams}
     for round_no in range(args.rounds):
         order = [("bash", args.bash), ("nash", args.nash)]
         if round_no % 2:
             order.reverse()
         for name, shell in order:
-            total = sum(run_timed(shell, cmd, observe_env) for cmd in cmds)
-            totals[name].append(total)
-            print(f"round {round_no + 1} {name}: {total:.3f}s", flush=True)
+            wall, cpu = measure(shell, cmds, observe_env)
+            totals[name].append((wall, cpu))
+            print(f"round {round_no + 1} {name}: {wall:.3f}s wall / {cpu:.3f}s cpu", flush=True)
+            for case, script in streams:
+                stream_totals[case][name].append(run_timed(shell, script, observe_env))
 
-    bash_med = statistics.median(totals["bash"])
-    nash_med = statistics.median(totals["nash"])
-    overhead = 100.0 * (nash_med / bash_med - 1.0)
+    corpus = report(
+        f"corpus ({len(cmds)} cmds x {args.rounds} rounds)",
+        totals["bash"],
+        totals["nash"],
+        gate=args.gate,
+    )
+    stream_results = {
+        case: report(f"stream {case}", vals["bash"], vals["nash"], gate=args.stream_gate)
+        for case, vals in stream_totals.items()
+    }
 
-    print(f"\ncorpus: {len(cmds)} cmds x {args.rounds} rounds")
-    print(f"bash median: {bash_med:.3f}s   nash median: {nash_med:.3f}s")
-    print(f"overhead: {overhead:+.2f}%   (gate: <{args.gate:.1f}%)")
+    corpus_ok = corpus["wall_percent"] < args.gate
+    # The stream cases are gated on CPU: the tee's cost is a copier thread, which idle
+    # hardware hides from wall-clock entirely.
+    stream_ok = all(r["cpu_percent"] < args.stream_gate for r in stream_results.values())
 
     if args.json:
         with open(args.json, "w") as f:
@@ -112,20 +277,30 @@ def main():
                 {
                     "cmds": len(cmds),
                     "rounds": args.rounds,
-                    "totals": totals,
-                    "bash_median_s": bash_med,
-                    "nash_median_s": nash_med,
-                    "overhead_percent": overhead,
+                    "totals": {k: [{"wall_s": w, "cpu_s": c} for w, c in v] for k, v in totals.items()},
+                    "corpus": corpus,
+                    "streams": stream_results,
+                    "stream_mb": args.stream_mb,
                     "gate_percent": args.gate,
+                    "stream_gate_percent": args.stream_gate,
                     "observed": bool(args.observe_spool),
+                    "bash": args.bash,
+                    "nash": args.nash,
+                    # Kept for readers of the pre-CPU schema.
+                    "bash_median_s": corpus["bash_wall_s"],
+                    "nash_median_s": corpus["nash_wall_s"],
+                    "overhead_percent": corpus["wall_percent"],
                 },
                 f,
                 indent=1,
             )
 
-    gate = overhead < args.gate
-    print(f"M3 overhead gate (<{args.gate:.1f}%): {'PASS' if gate else 'FAIL'}")
-    return 0 if gate else 1
+    print(f"\nM3 overhead gate (corpus wall <{args.gate:.1f}%): {'PASS' if corpus_ok else 'FAIL'}")
+    if streams:
+        print(
+            f"stream gate (cpu <{args.stream_gate:.1f}%): {'PASS' if stream_ok else 'FAIL'}"
+        )
+    return 0 if corpus_ok and stream_ok else 1
 
 
 if __name__ == "__main__":

shell/nash-observe/src/lib.rs

@@ -18,6 +18,7 @@ use std::sync::mpsc;
 use std::sync::{Mutex, OnceLock};
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
+use serde::ser::SerializeMap;
 use serde::Serialize;
 
 const FLUSH_MAX_EVENTS: usize = 200;
@@ -35,11 +36,144 @@ const IO_TIMEOUT: Duration = Duration::from_millis(1500);
 const EXIT_FLUSH_TIMEOUT: Duration = Duration::from_millis(2000);
 /// Per-redirection / per-cmdsub preview cap in bytes (docs/NASH.md §5.4).
 const PREVIEW_CAP: usize = 64 * 1024;
+/// Captured payload one batch may carry, in raw bytes (docs/NASH.md §5.4's per-batch cap).
+/// Beyond it the events keep their counts and hashes and lose their previews — a pipeline storm
+/// can otherwise put 200 × 64 KiB of preview in one body, which the host then has to parse,
+/// base64-decode and hold (docs/NASH_STREAM_PERF_PLAN.md §P4.2).
+const BATCH_PREVIEW_BUDGET: usize = 1024 * 1024;
 
 // ---------------------------------------------------------------------------
 // Event model (docs/NASH.md §6.1)
 // ---------------------------------------------------------------------------
 
+/// The captured half of a data-flow event (`redirect` / `cmdsub` / `pipe`) — everything the wire
+/// calls `bytes`, `truncated`, `hash` and `previewB64`.
+///
+/// It exists so that **none of that encoding happens where the bytes were captured**
+/// (docs/NASH_STREAM_PERF_PLAN.md §P3). The interpreter thread, the tee thread and the command's
+/// own exit path hand over raw bytes; redaction, base64 and the FNV hash run inside
+/// [`Serialize`] — which is only ever called from the flusher thread, off every path a command
+/// waits on. The wire shape is unchanged.
+enum Capture {
+    /// Nothing was captured: a special file (`/dev/null`, a tty, a socket), or a target that
+    /// could not be read. Counts as zero, and carries no hash — an empty hash is how the host
+    /// tells "no payload" from "a payload that happened to be empty".
+    None,
+    /// A file range that has not been read back yet; the flusher opens and reads it
+    /// (docs/NASH_STREAM_PERF_PLAN.md §P3.2). `offset`/`total` are fixed at the moment the
+    /// command exited, so a later command appending to the same file cannot widen this event's
+    /// range — only the *reading* is deferred, never the measurement.
+    Deferred {
+        path: PathBuf,
+        offset: u64,
+        total: u64,
+    },
+    /// Bytes in hand, raw and already capped to [`PREVIEW_CAP`].
+    Bytes {
+        bytes: u64,
+        truncated: bool,
+        data: Vec<u8>,
+    },
+    /// A preview dropped by the per-batch budget (docs/NASH.md §5.4): counts and hash survive,
+    /// and `truncated` is true because the preview no longer stands for the bytes.
+    Counted { bytes: u64, hash: String },
+}
+
+impl Capture {
+    /// Bytes in hand, capped to the preview cap.
+    fn from_bytes(total: u64, data: &[u8]) -> Self {
+        let capped = data.len().min(PREVIEW_CAP);
+        Self::Bytes {
+            bytes: total,
+            truncated: total > capped as u64,
+            data: data[..capped].to_vec(),
+        }
+    }
+
+    /// The payload this will preview, in raw bytes — what the per-batch budget is spent on.
+    const fn preview_len(&self) -> usize {
+        match self {
+            Self::Bytes { data, .. } => data.len(),
+            // The budget is spent after the deferred reads are resolved, so this arm is only
+            // reached for a range that could not be read; charge its bounded worst case anyway
+            // rather than letting an unresolved capture spend nothing.
+            Self::Deferred { total, .. } => {
+                if *total > PREVIEW_CAP as u64 {
+                    PREVIEW_CAP
+                } else {
+                    *total as usize
+                }
+            }
+            Self::None | Self::Counted { .. } => 0,
+        }
+    }
+
+    /// Read back a deferred range (flusher thread only). A range that has since become
+    /// unreadable degrades to [`Capture::None`] — the event still reports what it observed.
+    fn resolve(&mut self) {
+        let Self::Deferred {
+            path,
+            offset,
+            total,
+        } = self
+        else {
+            return;
+        };
+        *self = read_range(path, *offset, *total).map_or(Self::None, |(total, data)| Self::Bytes {
+            bytes: total,
+            truncated: total > data.len() as u64,
+            data,
+        });
+    }
+
+    /// Drop the preview, keeping the counts and the hash (the per-batch budget).
+    fn count_only(&mut self) {
+        if let Self::Bytes { bytes, data, .. } = self {
+            *self = Self::Counted {
+                bytes: *bytes,
+                hash: fnv1a_hex(data),
+            };
+        }
+    }
+}
+
+impl Serialize for Capture {
+    /// Flattened into its event, so the four fields sit where they always have. This is where the
+    /// hashing, redaction and base64 of every preview nash sends actually happen.
+    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        let mut map = serializer.serialize_map(Some(4))?;
+        match self {
+            // A range left unresolved would mean the flusher skipped it; report it as the
+            // metadata-only event it effectively is rather than inventing a payload.
+            Self::None | Self::Deferred { .. } => {
+                map.serialize_entry("bytes", &0u64)?;
+                map.serialize_entry("truncated", &false)?;
+                map.serialize_entry("hash", "")?;
+                map.serialize_entry("previewB64", "")?;
+            }
+            Self::Bytes {
+                bytes,
+                truncated,
+                data,
+            } => {
+                map.serialize_entry("bytes", bytes)?;
+                map.serialize_entry("truncated", truncated)?;
+                // Over the captured prefix: the full stream is never buffered for a pipe, and this
+                // is a did-these-bytes-repeat marker rather than an integrity claim (NASH.md §5.2).
+                map.serialize_entry("hash", &fnv1a_hex(data))?;
+                map.serialize_entry("previewB64", &preview_b64(data))?;
+            }
+            Self::Counted { bytes, hash } => {
+                map.serialize_entry("bytes", bytes)?;
+                map.serialize_entry("truncated", &true)?;
+                map.serialize_entry("hash", hash)?;
+                map.serialize_entry("previewB64", "")?;
+            }
+        }
+        map.end()
+    }
+}
+
 #[derive(Serialize)]
 #[serde(tag = "kind", rename_all = "camelCase")]
 enum Event {
@@ -104,19 +238,17 @@ enum Event {
         op: String,
         fd: u32,
         target: Option<String>,
-        bytes: u64,
-        truncated: bool,
-        hash: String,
-        preview_b64: String,
+        /// Usually [`Capture::Deferred`] when it goes into the channel and [`Capture::Bytes`] by
+        /// the time it is serialized — the file read happens on the flusher (§P3.2).
+        #[serde(flatten)]
+        capture: Capture,
     },
     #[serde(rename = "cmdsub", rename_all = "camelCase")]
     Cmdsub {
         seq: u64,
         ts: u64,
-        bytes: u64,
-        truncated: bool,
-        hash: String,
-        preview_b64: String,
+        #[serde(flatten)]
+        capture: Capture,
     },
     #[serde(rename = "pipe", rename_all = "camelCase")]
     Pipe {
@@ -128,12 +260,15 @@ enum Event {
         /// The commands either side of this link, as written — brush-core renders them from the
         /// AST at spawn, so they name a compound stage correctly and arrive with the very first
         /// link rather than waiting on the stages to exit.
+        ///
+        /// Stage text is source, so it can carry a literal credential the same way argv can;
+        /// masking runs at serialization time like every other outbound string (NASH.md §5.4).
+        #[serde(serialize_with = "serialize_redacted")]
         from_text: String,
+        #[serde(serialize_with = "serialize_redacted")]
         to_text: String,
-        bytes: u64,
-        truncated: bool,
-        hash: String,
-        preview_b64: String,
+        #[serde(flatten)]
+        capture: Capture,
     },
     #[serde(rename = "dropped", rename_all = "camelCase")]
     Dropped { seq: u64, ts: u64, count: u64 },
@@ -406,26 +541,48 @@ fn looks_secret(word: &str) -> bool {
     false
 }
 
-/// The captured (bytes-total, capped-preview, truncated) for one redirect record.
-fn read_back(record: &brush_core::gate::RedirectRecord) -> Option<(u64, Vec<u8>, bool)> {
+/// What one redirect record captured, measured on the command's own path but **not yet read**
+/// (docs/NASH_STREAM_PERF_PLAN.md §P3.2).
+///
+/// The split is deliberate. The *range* — is this a regular file at all, and which bytes did this
+/// command put there — is only true at the moment the command exited: `echo a > f; echo b >> f`
+/// would otherwise have the truncating write report both lines, because by the time a batch
+/// flushes the file has grown. So the exit path pays one `stat` and nothing else; opening the
+/// file, reading up to 64 KiB of it, hashing, redacting and base64-encoding all move to the
+/// flusher, which is where the bytes were always going anyway.
+fn measure_readback(record: &brush_core::gate::RedirectRecord) -> Capture {
     use brush_core::gate::RedirectReadback;
     if let Some(inline) = &record.inline {
+        // Heredoc / here-string: the body was known at setup, so there is nothing to read.
         let full = inline.as_bytes();
-        let capped = full.len().min(PREVIEW_CAP);
-        return Some((full.len() as u64, full[..capped].to_vec(), full.len() > capped));
+        return Capture::from_bytes(full.len() as u64, full);
     }
-    let path = record.path.as_ref()?;
+    let Some(path) = record.path.as_ref() else {
+        return Capture::None;
+    };
     // Only read back regular files; skip /dev/null, ttys, fifos, sockets, devices.
-    let meta = std::fs::metadata(path).ok()?;
+    let Ok(meta) = std::fs::metadata(path) else {
+        return Capture::None;
+    };
     if !meta.is_file() {
-        return None;
+        return Capture::None;
     }
     let size = meta.len();
     let (offset, total) = match record.readback {
         RedirectReadback::Append => (record.size_before, size.saturating_sub(record.size_before)),
         RedirectReadback::Truncate | RedirectReadback::Input => (0, size),
-        RedirectReadback::Inline => return None,
+        RedirectReadback::Inline => return Capture::None,
     };
+    Capture::Deferred {
+        path: path.clone(),
+        offset,
+        total,
+    }
+}
+
+/// Read a measured range back, capped to [`PREVIEW_CAP`]. Returns the range's full length (which
+/// may exceed what was read) and the bytes. Runs on the flusher thread only.
+fn read_range(path: &Path, offset: u64, total: u64) -> Option<(u64, Vec<u8>)> {
     let mut file = std::fs::File::open(path).ok()?;
     if offset > 0 {
         use std::io::Seek;
@@ -435,7 +592,7 @@ fn read_back(record: &brush_core::gate::RedirectRecord) -> Option<(u64, Vec<u8>,
     let mut buf = vec![0u8; want];
     let n = std::io::Read::read(&mut file, &mut buf).unwrap_or(0);
     buf.truncate(n);
-    Some((total, buf, total > n as u64))
+    Some((total, buf))
 }
 
 /// Resolve a redirection target against the directory its command ran in, so the
@@ -471,11 +628,24 @@ fn preview_b64(data: &[u8]) -> String {
         let redacted = redact(&String::from_utf8_lossy(data));
         b64(redacted.as_bytes())
     } else {
-        let hex: String = data.iter().take(1024).map(|b| format!("{b:02x}")).collect();
-        b64(format!("<binary> {hex}").as_bytes())
+        let mut hex = String::with_capacity(1024 * 2 + 9);
+        hex.push_str("<binary> ");
+        for byte in data.iter().take(1024) {
+            hex.push(HEX[usize::from(byte >> 4)] as char);
+            hex.push(HEX[usize::from(byte & 0xf)] as char);
+        }
+        b64(hex.as_bytes())
     }
 }
 
+const HEX: &[u8; 16] = b"0123456789abcdef";
+
+/// Mask a string on its way out (docs/NASH.md §5.4). Used where redaction is deferred to
+/// serialization — the flusher thread — rather than done where the string was captured.
+fn serialize_redacted<S: serde::Serializer>(text: &str, serializer: S) -> Result<S::Ok, S::Error> {
+    serializer.serialize_str(&redact(text))
+}
+
 // ---------------------------------------------------------------------------
 // Gate implementation (allow-all; docs/NASH.md §4.2)
 // ---------------------------------------------------------------------------
@@ -570,7 +740,8 @@ impl brush_core::gate::Gate for RecordingGate {
                 pipeline: pending.pipeline.map(PipelineRef::from),
             });
 
-            // Data-flow read-back for this command's redirects (docs/NASH.md §5.2).
+            // Data-flow read-back for this command's redirects (docs/NASH.md §5.2). Only the
+            // *range* is settled here (one stat); the read and the encoding ride to the flusher.
             for record in &pending.redirects {
                 // The host resolves a redirect target against the session's worktree to decide
                 // which file was edited (and which lock that is), so report the path absolutely:
@@ -580,22 +751,6 @@ impl brush_core::gate::Gate for RecordingGate {
                     .path
                     .as_ref()
                     .map(|p| absolute_path(p, &pending.cwd).to_string_lossy().into_owned());
-                let Some((total, data, truncated)) = read_back(record) else {
-                    // Special file / unreadable — emit metadata only.
-                    obs.send(Event::Redirect {
-                        seq: obs.seq(),
-                        ts,
-                        cmd_seq: exec_seq,
-                        op: record.op.clone(),
-                        fd: record.fd,
-                        target,
-                        bytes: 0,
-                        truncated: false,
-                        hash: String::new(),
-                        preview_b64: String::new(),
-                    });
-                    continue;
-                };
                 obs.send(Event::Redirect {
                     seq: obs.seq(),
                     ts,
@@ -603,10 +758,7 @@ impl brush_core::gate::Gate for RecordingGate {
                     op: record.op.clone(),
                     fd: record.fd,
                     target,
-                    bytes: total,
-                    truncated,
-                    hash: fnv1a_hex(&data),
-                    preview_b64: preview_b64(&data),
+                    capture: measure_readback(record),
                 });
             }
         }));
@@ -616,14 +768,13 @@ impl brush_core::gate::Gate for RecordingGate {
         let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
             let Some(obs) = OBSERVER.get() else { return };
             let full = output.as_bytes();
-            let capped = full.len().min(PREVIEW_CAP);
+            // Copies the capped prefix and nothing else: this runs on the interpreter thread, in
+            // the middle of the expansion the shell is waiting on, and `$(cat big-file)` is a
+            // substitution whose *whole* result used to be hashed here (§P3.1).
             obs.send(Event::Cmdsub {
                 seq: obs.seq(),
                 ts: now_millis(),
-                bytes: full.len() as u64,
-                truncated: full.len() > capped,
-                hash: fnv1a_hex(full),
-                preview_b64: preview_b64(&full[..capped]),
+                capture: Capture::from_bytes(full.len() as u64, full),
             });
         }));
     }
@@ -632,21 +783,21 @@ impl brush_core::gate::Gate for RecordingGate {
         let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
             let Some(obs) = OBSERVER.get() else { return };
             let capped = ev.captured.len().min(PREVIEW_CAP);
+            let mut data = ev.captured;
+            data.truncate(capped);
             obs.send(Event::Pipe {
                 seq: obs.seq(),
                 ts: now_millis(),
                 pipeline_id: ev.pipeline_id,
                 from_index: ev.from_index as u64,
                 to_index: ev.to_index as u64,
-                // Stage text is source, so it can carry a literal credential the same way argv
-                // can — every outbound string goes through the same masking (docs/NASH.md §5.4).
-                from_text: redact(&ev.from_text),
-                to_text: redact(&ev.to_text),
-                bytes: ev.total_bytes,
-                truncated: ev.truncated || ev.captured.len() > capped,
-                // Hash over the captured prefix (the full stream isn't buffered).
-                hash: fnv1a_hex(&ev.captured),
-                preview_b64: preview_b64(&ev.captured[..capped]),
+                from_text: ev.from_text,
+                to_text: ev.to_text,
+                capture: Capture::Bytes {
+                    bytes: ev.total_bytes,
+                    truncated: ev.truncated || ev.total_bytes > capped as u64,
+                    data,
+                },
             });
         }));
     }
@@ -660,8 +811,12 @@ impl brush_core::gate::Gate for RecordingGate {
 static UNAUTHORIZED: AtomicBool = AtomicBool::new(false);
 
 fn http_request(path: &str, token: Option<&str>, body: &[u8]) -> Vec<u8> {
+    // Keep-alive, deliberately (docs/NASH_STREAM_PERF_PLAN.md §P4.1): a shell posts a batch every
+    // 500 ms for as long as it lives, and `Connection: close` made every one of them a fresh
+    // connect + accept + teardown on both sides. The host's route already answers with
+    // `Connection: keep-alive` and a `Content-Length`, which is what makes reuse framable.
     let mut req = format!(
-        "POST {path} HTTP/1.1\r\nHost: nucleic\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n",
+        "POST {path} HTTP/1.1\r\nHost: nucleic\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: keep-alive\r\n",
         body.len()
     );
     if let Some(token) = token {
@@ -673,55 +828,185 @@ fn http_request(path: &str, token: Option<&str>, body: &[u8]) -> Vec<u8> {
     bytes
 }
 
-fn read_status<R: Read>(mut stream: R) -> Option<u16> {
-    let mut buf = [0u8; 64];
-    let n = stream.read(&mut buf).ok()?;
-    let text = String::from_utf8_lossy(&buf[..n]);
-    let status = text.split_whitespace().nth(1)?;
-    status.parse().ok()
+/// The outcome of one post on a live connection.
+enum Posted {
+    /// The host took the batch and the connection can carry another.
+    Ok,
+    /// The host took the batch but the connection cannot be reused (no `Content-Length` to frame
+    /// the next response by, or the host asked to close).
+    OkAndClose,
+    /// The batch was not delivered; the connection is dropped. `retry` is false when a fresh
+    /// connection would not help (an outright refusal), true for an I/O failure — most often a
+    /// kept-open socket the host has since closed, which is nobody's fault and worth one retry.
+    Failed { retry: bool },
 }
 
-fn check_status(status: Option<u16>) -> Result<(), ()> {
-    match status {
-        Some(code) if (200..300).contains(&code) => Ok(()),
-        Some(401) => {
-            UNAUTHORIZED.store(true, Ordering::Relaxed);
-            Err(())
+/// Anything the transport can post over: the unix socket and the TCP fallback both qualify.
+trait Wire: Read + Write {}
+impl<T: Read + Write> Wire for T {}
+
+/// A connection kept open across batches (docs/NASH_STREAM_PERF_PLAN.md §P4.1). Lives on the
+/// flusher thread, which is the only thing that posts, so it needs no locking.
+#[derive(Default)]
+struct Connection {
+    /// The live connection and the request path the transport it came from wants.
+    open: Option<(Box<dyn Wire>, String)>,
+}
+
+impl Connection {
+    /// Post one batch, reconnecting once if the kept-open connection turned out to be dead.
+    fn post(&mut self, cfg: &Config, body: &[u8]) -> Result<(), ()> {
+        for _ in 0..2 {
+            if self.open.is_none() {
+                self.open = connect(cfg);
+            }
+            let Some((stream, path)) = self.open.as_mut() else {
+                return Err(()); // nothing to connect to; the caller spools
+            };
+            let request = http_request(path, cfg.token.as_deref(), body);
+            match post_on(stream.as_mut(), &request) {
+                Posted::Ok => return Ok(()),
+                Posted::OkAndClose => {
+                    self.open = None;
+                    return Ok(());
+                }
+                Posted::Failed { retry } => {
+                    self.open = None;
+                    if !retry {
+                        return Err(());
+                    }
+                }
+            }
         }
-        // Treat unreadable/other responses as delivered-at-best-effort: the
-        // host 202s known tokens, so anything else is not retryable.
-        Some(_) | None => Err(()),
+        Err(())
     }
 }
 
-fn post_unix(socket: &Path, token: Option<&str>, body: &[u8]) -> Result<(), ()> {
-    let stream = std::os::unix::net::UnixStream::connect(socket).map_err(|_| ())?;
+/// Open the configured transport: the unix socket first, then the TCP hook.
+fn connect(cfg: &Config) -> Option<(Box<dyn Wire>, String)> {
+    if let Some(socket) = &cfg.socket {
+        if let Ok(stream) = std::os::unix::net::UnixStream::connect(socket) {
+            let _ = stream.set_write_timeout(Some(IO_TIMEOUT));
+            let _ = stream.set_read_timeout(Some(IO_TIMEOUT));
+            return Some((Box::new(stream), "/shell-event".to_string()));
+        }
+    }
+    let url = cfg.hook_url.as_ref()?;
+    let addr = std::net::ToSocketAddrs::to_socket_addrs(&(url.host.as_str(), url.port))
+        .ok()?
+        .next()?;
+    let stream = std::net::TcpStream::connect_timeout(&addr, IO_TIMEOUT).ok()?;
     let _ = stream.set_write_timeout(Some(IO_TIMEOUT));
     let _ = stream.set_read_timeout(Some(IO_TIMEOUT));
-    let mut stream = stream;
-    stream
-        .write_all(&http_request("/shell-event", token, body))
-        .map_err(|_| ())?;
-    check_status(read_status(&mut stream))
+    // Batches are one write each and latency is not what this path optimizes, but a delayed ACK
+    // waiting on Nagle would hold the *response* — and with it the next batch — for milliseconds.
+    let _ = stream.set_nodelay(true);
+    Some((Box::new(stream), url.path.clone()))
 }
 
-fn post_tcp(url: &HookUrl, token: Option<&str>, body: &[u8]) -> Result<(), ()> {
-    let addr = (url.host.as_str(), url.port);
-    let stream = std::net::TcpStream::connect_timeout(
-        &std::net::ToSocketAddrs::to_socket_addrs(&addr)
-            .ok()
-            .and_then(|mut a| a.next())
-            .ok_or(())?,
-        IO_TIMEOUT,
-    )
-    .map_err(|_| ())?;
-    let _ = stream.set_write_timeout(Some(IO_TIMEOUT));
-    let _ = stream.set_read_timeout(Some(IO_TIMEOUT));
-    let mut stream = stream;
-    stream
-        .write_all(&http_request(&url.path, token, body))
-        .map_err(|_| ())?;
-    check_status(read_status(&mut stream))
+/// Write one request and read its whole response, so the connection is left framed for the next.
+fn post_on(stream: &mut dyn Wire, request: &[u8]) -> Posted {
+    if stream.write_all(request).is_err() {
+        return Posted::Failed { retry: true };
+    }
+    let Some(response) = read_response(stream) else {
+        return Posted::Failed { retry: true };
+    };
+    match response.status {
+        401 => {
+            UNAUTHORIZED.store(true, Ordering::Relaxed);
+            Posted::Failed { retry: false }
+        }
+        code if (200..300).contains(&code) => {
+            if response.reusable {
+                Posted::Ok
+            } else {
+                Posted::OkAndClose
+            }
+        }
+        // The host 202s known tokens, so anything else is not retryable.
+        _ => Posted::Failed { retry: false },
+    }
+}
+
+struct Response {
+    status: u16,
+    /// Whether the body was fully consumed and the peer means to keep the connection.
+    reusable: bool,
+}
+
+/// Read a response's head, then drain exactly its `Content-Length` body — the whole point being
+/// that the next response starts where this one ends. A response we cannot frame (no length,
+/// chunked, `Connection: close`) is still reported, but its connection is not reused.
+fn read_response(stream: &mut dyn Read) -> Option<Response> {
+    let mut buf = Vec::with_capacity(256);
+    let mut chunk = [0u8; 512];
+    let head_end = loop {
+        if let Some(at) = find_headers_end(&buf) {
+            break at;
+        }
+        // A response head this long is not one of ours; give up rather than read forever.
+        if buf.len() > 8192 {
+            return None;
+        }
+        match stream.read(&mut chunk) {
+            Ok(0) => return None,
+            Ok(n) => buf.extend_from_slice(&chunk[..n]),
+            Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => (),
+            Err(_) => return None,
+        }
+    };
+    let head = String::from_utf8_lossy(&buf[..head_end]);
+    let mut lines = head.split("\r\n");
+    let status: u16 = lines.next()?.split_whitespace().nth(1)?.parse().ok()?;
+    let mut length: Option<usize> = None;
+    let mut close = false;
+    for line in lines {
+        let Some((name, value)) = line.split_once(':') else {
+            continue;
+        };
+        match name.trim().to_ascii_lowercase().as_str() {
+            "content-length" => length = value.trim().parse().ok(),
+            "connection" => close = value.trim().eq_ignore_ascii_case("close"),
+            _ => (),
+        }
+    }
+    let Some(length) = length else {
+        return Some(Response {
+            status,
+            reusable: false,
+        });
+    };
+    // Drain the body so the socket is positioned at the next response.
+    let mut remaining = length.saturating_sub(buf.len() - (head_end + 4));
+    while remaining > 0 {
+        let want = remaining.min(chunk.len());
+        match stream.read(&mut chunk[..want]) {
+            Ok(0) => {
+                return Some(Response {
+                    status,
+                    reusable: false,
+                })
+            }
+            Ok(n) => remaining -= n,
+            Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => (),
+            Err(_) => {
+                return Some(Response {
+                    status,
+                    reusable: false,
+                })
+            }
+        }
+    }
+    Some(Response {
+        status,
+        reusable: !close,
+    })
+}
+
+/// Offset of the `\r\n\r\n` that ends a response head.
+fn find_headers_end(buf: &[u8]) -> Option<usize> {
+    buf.windows(4).position(|w| w == b"\r\n\r\n")
 }
 
 fn spool(dir: &Path, shell_id: &str, body: &[u8]) {
@@ -733,19 +1018,9 @@ fn spool(dir: &Path, shell_id: &str, body: &[u8]) {
     }
 }
 
-fn deliver(cfg: &Config, shell_id: &str, body: &[u8]) {
-    let token = cfg.token.as_deref();
-    if !UNAUTHORIZED.load(Ordering::Relaxed) {
-        if let Some(socket) = &cfg.socket {
-            if post_unix(socket, token, body).is_ok() {
-                return;
-            }
-        }
-        if let Some(url) = &cfg.hook_url {
-            if post_tcp(url, token, body).is_ok() {
-                return;
-            }
-        }
+fn deliver(cfg: &Config, shell_id: &str, conn: &mut Connection, body: &[u8]) {
+    if !UNAUTHORIZED.load(Ordering::Relaxed) && conn.post(cfg, body).is_ok() {
+        return;
     }
     if let Some(dir) = &cfg.spool {
         spool(dir, shell_id, body);
@@ -806,14 +1081,46 @@ fn sweep_running() -> Sweep {
     sweep
 }
 
+/// The captured half of an event, for the flusher's two passes over a batch.
+fn event_capture(event: &mut Event) -> Option<&mut Capture> {
+    match event {
+        Event::Redirect { capture, .. }
+        | Event::Cmdsub { capture, .. }
+        | Event::Pipe { capture, .. } => Some(capture),
+        _ => None,
+    }
+}
+
+/// Settle a batch's payloads before it is serialized — the flusher-thread half of §P3/§P4.2.
+///
+/// Two passes, in this order: read back every deferred file range, then spend the per-batch
+/// preview budget over what that leaves. Events keep their place and their counts either way; only
+/// previews are ever given up, oldest kept first, so a storm of huge pipes loses its tail rather
+/// than the whole batch losing its head.
+fn settle(events: &mut [Event]) {
+    let mut spent = 0usize;
+    for event in events {
+        let Some(capture) = event_capture(event) else {
+            continue;
+        };
+        capture.resolve();
+        spent = spent.saturating_add(capture.preview_len());
+        if spent > BATCH_PREVIEW_BUDGET {
+            capture.count_only();
+        }
+    }
+}
+
 fn flusher(cfg: Config, shell_id: String, parent_shell: Option<String>, rx: mpsc::Receiver<Msg>) {
     let mut buf: Vec<Event> = Vec::new();
     let mut first_at: Option<Instant> = None;
+    let mut conn = Connection::default();
 
-    let flush = |buf: &mut Vec<Event>| {
+    let flush = |buf: &mut Vec<Event>, conn: &mut Connection| {
         if buf.is_empty() {
             return;
         }
+        settle(buf);
         let batch = Batch {
             batch_type: "shell-batch",
             source: "nash",
@@ -826,7 +1133,7 @@ fn flusher(cfg: Config, shell_id: String, parent_shell: Option<String>, rx: mpsc
             events: buf,
         };
         if let Ok(body) = serde_json::to_vec(&batch) {
-            deliver(&cfg, &shell_id, &body);
+            deliver(&cfg, &shell_id, conn, &body);
         }
         buf.clear();
     };
@@ -842,7 +1149,7 @@ fn flusher(cfg: Config, shell_id: String, parent_shell: Option<String>, rx: mpsc
             }
             buf.extend(sweep.announce);
             if buf.len() >= FLUSH_MAX_EVENTS {
-                flush(&mut buf);
+                flush(&mut buf, &mut conn);
                 first_at = None;
             }
         }
@@ -863,12 +1170,12 @@ fn flusher(cfg: Config, shell_id: String, parent_shell: Option<String>, rx: mpsc
                 }
                 buf.push(ev);
                 if buf.len() >= FLUSH_MAX_EVENTS {
-                    flush(&mut buf);
+                    flush(&mut buf, &mut conn);
                     first_at = None;
                 }
             }
             Ok(Msg::FlushSync(ack)) => {
-                flush(&mut buf);
+                flush(&mut buf, &mut conn);
                 first_at = None;
                 let _ = ack.send(());
             }
@@ -878,12 +1185,12 @@ fn flusher(cfg: Config, shell_id: String, parent_shell: Option<String>, rx: mpsc
             Err(mpsc::RecvTimeoutError::Timeout) => {
                 // A tick is just "go re-sweep"; the batch is not due yet.
                 if !ticking {
-                    flush(&mut buf);
+                    flush(&mut buf, &mut conn);
                     first_at = None;
                 }
             }
             Err(mpsc::RecvTimeoutError::Disconnected) => {
-                flush(&mut buf);
+                flush(&mut buf, &mut conn);
                 return;
             }
         }
@@ -976,3 +1283,179 @@ pub fn report_fallback(reason: &str, input: &str) {
     });
     flush_sync();
 }
+
+// ---------------------------------------------------------------------------
+// Tests
+// ---------------------------------------------------------------------------
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Serialize one event the way the flusher does, as JSON.
+    fn json(event: &Event) -> serde_json::Value {
+        serde_json::to_value(event).expect("event serializes")
+    }
+
+    fn pipe(bytes: u64, data: Vec<u8>) -> Event {
+        Event::Pipe {
+            seq: 1,
+            ts: 1,
+            pipeline_id: 1,
+            from_index: 0,
+            to_index: 1,
+            from_text: "a".to_string(),
+            to_text: "b".to_string(),
+            capture: Capture::Bytes {
+                bytes,
+                truncated: bytes > data.len() as u64,
+                data,
+            },
+        }
+    }
+
+    /// The four payload fields sit where they always have, and an empty hash still means "nothing
+    /// was captured" rather than "the hash of nothing" — the host reads it that way.
+    #[test]
+    fn capture_serializes_the_wire_shape() {
+        let event = json(&pipe(5, b"hello".to_vec()));
+        assert_eq!(event["kind"], "pipe");
+        assert_eq!(event["bytes"], 5);
+        assert_eq!(event["truncated"], false);
+        assert_eq!(event["previewB64"], b64(b"hello"));
+        assert_eq!(event["hash"], fnv1a_hex(b"hello"));
+
+        let empty = json(&Event::Cmdsub {
+            seq: 1,
+            ts: 1,
+            capture: Capture::None,
+        });
+        assert_eq!(empty["bytes"], 0);
+        assert_eq!(empty["hash"], "");
+        assert_eq!(empty["previewB64"], "");
+    }
+
+    /// Redaction and base64 run at serialization time now, so they must still actually run —
+    /// on the preview *and* on the stage labels a link carries.
+    #[test]
+    fn serialization_redacts_previews_and_stage_text() {
+        let secret = "ghp_0123456789abcdefghij";
+        let event = json(&Event::Pipe {
+            seq: 1,
+            ts: 1,
+            pipeline_id: 1,
+            from_index: 0,
+            to_index: 1,
+            from_text: format!("curl -H {secret}"),
+            to_text: "sh".to_string(),
+            capture: Capture::from_bytes(secret.len() as u64, secret.as_bytes()),
+        });
+        assert!(!event["fromText"].as_str().unwrap().contains(secret));
+        assert_eq!(event["fromText"], "curl -H «redacted»");
+        assert_eq!(event["previewB64"], b64("«redacted»".as_bytes()));
+    }
+
+    /// A capture is capped to the preview cap where it is taken, and says so.
+    #[test]
+    fn capture_caps_and_marks_truncation() {
+        let big = vec![b'x'; PREVIEW_CAP + 10];
+        let Capture::Bytes {
+            bytes,
+            truncated,
+            data,
+        } = Capture::from_bytes(big.len() as u64, &big)
+        else {
+            panic!("expected captured bytes");
+        };
+        assert_eq!(bytes, big.len() as u64);
+        assert_eq!(data.len(), PREVIEW_CAP);
+        assert!(truncated);
+    }
+
+    /// The per-batch budget (docs/NASH.md §5.4): once a batch has carried its allowance of
+    /// payload, later events keep their counts and hashes and give up their previews.
+    #[test]
+    fn batch_preview_budget_drops_previews_not_events() {
+        let payload = vec![b'z'; PREVIEW_CAP];
+        let over = BATCH_PREVIEW_BUDGET / PREVIEW_CAP + 2;
+        let mut events: Vec<Event> = (0..over)
+            .map(|_| pipe(PREVIEW_CAP as u64, payload.clone()))
+            .collect();
+        settle(&mut events);
+
+        let rendered: Vec<serde_json::Value> = events.iter().map(json).collect();
+        // Every event survives — only previews are given up, and the earliest keep theirs.
+        assert_eq!(rendered.len(), over);
+        assert_ne!(rendered[0]["previewB64"], "");
+        let last = rendered.last().unwrap();
+        assert_eq!(last["previewB64"], "");
+        assert_eq!(last["bytes"], PREVIEW_CAP);
+        assert_eq!(last["hash"], fnv1a_hex(&payload));
+        assert_eq!(last["truncated"], true);
+        let kept = rendered
+            .iter()
+            .filter(|e| e["previewB64"] != "")
+            .count();
+        assert_eq!(kept, BATCH_PREVIEW_BUDGET / PREVIEW_CAP);
+    }
+
+    /// A redirect's bytes are read by the flusher, from the range measured when the command
+    /// exited — an append that lands afterwards must not widen it.
+    #[test]
+    fn deferred_readback_reads_the_measured_range() {
+        let dir = std::env::temp_dir().join(format!("nash-observe-{}", std::process::id()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let path = dir.join("out.txt");
+        std::fs::write(&path, b"first\n").unwrap();
+
+        let mut event = Event::Redirect {
+            seq: 1,
+            ts: 1,
+            cmd_seq: 1,
+            op: ">".to_string(),
+            fd: 1,
+            target: Some(path.to_string_lossy().into_owned()),
+            capture: Capture::Deferred {
+                path: path.clone(),
+                offset: 0,
+                total: 6,
+            },
+        };
+        // The file grows before the batch flushes; the event still reports what its command wrote.
+        std::fs::write(&path, b"first\nsecond\n").unwrap();
+        settle(std::slice::from_mut(&mut event));
+
+        let rendered = json(&event);
+        assert_eq!(rendered["bytes"], 6);
+        assert_eq!(rendered["previewB64"], b64(b"first\n"));
+        assert_eq!(rendered["truncated"], false);
+
+        // A target that has since vanished degrades to metadata only rather than failing.
+        std::fs::remove_file(&path).unwrap();
+        let mut gone = Event::Redirect {
+            seq: 2,
+            ts: 1,
+            cmd_seq: 1,
+            op: ">".to_string(),
+            fd: 1,
+            target: None,
+            capture: Capture::Deferred {
+                path,
+                offset: 0,
+                total: 6,
+            },
+        };
+        settle(std::slice::from_mut(&mut gone));
+        assert_eq!(json(&gone)["bytes"], 0);
+        assert_eq!(json(&gone)["hash"], "");
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
+    /// Binary payloads are previewed as a hex head (now table-driven, §P3.3) — same bytes as the
+    /// `format!`-per-byte encoding it replaced.
+    #[test]
+    fn binary_previews_as_hex() {
+        let preview = preview_b64(&[0xff, 0x00, 0x0a]);
+        assert_eq!(preview, b64(b"<binary> ff000a"));
+    }
+}

shell/nash/tests/observe.rs

@@ -48,6 +48,9 @@ fn scratch_dir(prefix: &str) -> std::path::PathBuf {
 struct Sink {
     dir: std::path::PathBuf,
     rx: mpsc::Receiver<(Option<String>, serde_json::Value)>,
+    /// Connections accepted so far — how the keep-alive test tells one reused connection from
+    /// one per batch (docs/NASH_STREAM_PERF_PLAN.md §P4.1).
+    connections: std::sync::Arc<AtomicU32>,
 }
 
 impl Sink {
@@ -59,37 +62,52 @@ impl Sink {
         let listener = UnixListener::bind(&socket_path)
             .unwrap_or_else(|e| panic!("bind {} ({} bytes): {e}", socket_path.display(), socket_path.as_os_str().len()));
         let (tx, rx) = mpsc::channel();
+        let connections = std::sync::Arc::new(AtomicU32::new(0));
+        let accepted = std::sync::Arc::clone(&connections);
         std::thread::spawn(move || {
             for stream in listener.incoming() {
                 let Ok(mut stream) = stream else { break };
+                accepted.fetch_add(1, Ordering::Relaxed);
                 let tx = tx.clone();
                 std::thread::spawn(move || {
                     let mut buf = Vec::new();
                     let mut chunk = [0u8; 4096];
-                    // Read until headers + declared body length are complete.
+                    // Read until headers + declared body length are complete, answer, and keep
+                    // reading: nash reuses one connection for every batch of a shell's life
+                    // (docs/NASH_STREAM_PERF_PLAN.md §P4.1), exactly as the host's route does.
                     loop {
-                        match stream.read(&mut chunk) {
-                            Ok(0) => break,
-                            Ok(n) => {
-                                buf.extend_from_slice(&chunk[..n]);
-                                if let Some(body) = full_body(&buf) {
-                                    let auth = header(&buf, "authorization");
-                                    if let Ok(json) = serde_json::from_slice(body) {
-                                        let _ = tx.send((auth, json));
-                                    }
-                                    let _ = stream.write_all(
-                                        b"HTTP/1.1 202 Accepted\r\nContent-Length: 0\r\n\r\n",
-                                    );
-                                    break;
+                        if let Some(request) = request_len(&buf) {
+                            let auth = header(&buf, "authorization");
+                            if let Some(body) = full_body(&buf) {
+                                if let Ok(json) = serde_json::from_slice(body) {
+                                    let _ = tx.send((auth, json));
                                 }
                             }
+                            if stream
+                                .write_all(
+                                    b"HTTP/1.1 202 Accepted\r\nContent-Length: 0\r\nConnection: keep-alive\r\n\r\n",
+                                )
+                                .is_err()
+                            {
+                                break;
+                            }
+                            buf.drain(..request);
+                            continue;
+                        }
+                        match stream.read(&mut chunk) {
+                            Ok(0) => break,
+                            Ok(n) => buf.extend_from_slice(&chunk[..n]),
                             Err(_) => break,
                         }
                     }
                 });
             }
         });
-        Self { dir, rx }
+        Self {
+            dir,
+            rx,
+            connections,
+        }
     }
 
     fn socket(&self) -> std::path::PathBuf {
@@ -148,6 +166,14 @@ impl Drop for Sink {
     }
 }
 
+/// Total bytes of the first complete request in `buf` (head + body), or None while it is still
+/// arriving — what lets one connection carry the next batch after this one is answered.
+fn request_len(buf: &[u8]) -> Option<usize> {
+    let headers_end = buf.windows(4).position(|w| w == b"\r\n\r\n")? + 4;
+    let len: usize = header(buf, "content-length")?.parse().ok()?;
+    (buf.len() >= headers_end + len).then_some(headers_end + len)
+}
+
 fn full_body(buf: &[u8]) -> Option<&[u8]> {
     let headers_end = buf.windows(4).position(|w| w == b"\r\n\r\n")? + 4;
     let len: usize = header(buf, "content-length")?.parse().ok()?;
@@ -869,3 +895,53 @@ fn capture_off_silences_nash_without_a_policy() {
         "capture=off with no policy posts nothing"
     );
 }
+
+/// A shell posts a batch every 500 ms for as long as it lives, so the connection is opened once
+/// and kept (docs/NASH_STREAM_PERF_PLAN.md §P4.1) rather than reconnected per batch. Two commands
+/// either side of a flush window are two batches; one connection has to carry both.
+#[test]
+fn one_connection_carries_every_batch() {
+    let sink = Sink::start();
+    let status = run_nash(&sink, "echo one; sleep 0.9; echo two");
+    assert_eq!(status.code(), Some(0));
+
+    let events = sink.events_until(|evs| {
+        find(evs, "exec", |e| e["argv"][0] == "echo" && e["argv"][1] == "two").is_some()
+    });
+    // Both commands were reported…
+    assert!(find(&events, "exec", |e| e["argv"][1] == "one").is_some());
+    assert!(find(&events, "exec", |e| e["argv"][1] == "two").is_some());
+    // …and the shell only ever dialled the sink once.
+    assert_eq!(
+        sink.connections.load(Ordering::Relaxed),
+        1,
+        "batches must share one kept-open connection"
+    );
+}
+
+/// The budget's end-to-end half (docs/NASH.md §5.4): a burst of fat pipes in one batch keeps every
+/// event and every byte count, and gives up only the previews past the allowance.
+#[test]
+fn batch_preview_budget_keeps_counts_and_drops_previews() {
+    let sink = Sink::start();
+    // 24 links × 64 KiB of payload each — comfortably past the 1 MiB per-batch preview budget,
+    // and fast enough that they all land in one 500 ms flush.
+    let status = run_nash(
+        &sink,
+        "for i in $(seq 1 24); do head -c 65536 /dev/zero | cat > /dev/null; done",
+    );
+    assert_eq!(status.code(), Some(0));
+
+    let events = sink.events_until(|evs| evs.iter().filter(|e| e["kind"] == "pipe").count() >= 24);
+    let pipes: Vec<_> = events.iter().filter(|e| e["kind"] == "pipe").collect();
+    assert!(pipes.len() >= 24, "every link is still reported: {}", pipes.len());
+    // Nothing loses its count or its fingerprint…
+    for pipe in &pipes {
+        assert_eq!(pipe["bytes"], 65536);
+        assert_ne!(pipe["hash"], "");
+    }
+    // …and the ones past the budget are the ones without a preview.
+    let dropped = pipes.iter().filter(|p| p["previewB64"] == "").count();
+    assert!(dropped > 0, "the budget dropped no previews");
+    assert!(dropped < pipes.len(), "the budget dropped every preview");
+}

third_party/brush/brush-core/src/interp.rs

@@ -1999,71 +1999,291 @@ fn nash_stage_text(pipeline: &ast::Pipeline, index: usize) -> String {
     text
 }
 
-// nash: spawn a copier that moves bytes from `src` (producer's output) to `dst`
-// (consumer's input), mirroring a bounded prefix, then reports the link to the
-// gate (docs/NASH.md §5.3). The synchronous copy preserves pipe backpressure;
-// EOF on `src` or EPIPE on `dst` (consumer gone, e.g. `yes | head`) ends it and
-// closes `dst` so the consumer sees EOF.
+// nash: bytes moved per userspace copy on the portable tee path. 128 KiB rather than the 32 KiB
+// this started with: the same stream costs a quarter of the read/write pairs, and a pipe link's
+// tee cost is almost entirely syscalls and the context switches around them
+// (docs/NASH_STREAM_PERF_PLAN.md §P2.2).
+const NASH_TEE_BUF: usize = 128 * 1024;
+
+// nash: how large both ends of a tapped link are grown to, where the platform allows it. A tee
+// puts *two* pipes where the shell asked for one, so a producer that outruns the default 64 KiB
+// buffer pays for it twice; growing them cuts the wakeups on both sides.
+#[cfg(target_os = "linux")]
+const NASH_TEE_PIPE_SIZE: libc::c_int = 256 * 1024;
+
+// nash: bytes asked of one `splice` call. The kernel moves at most a pipe-buffer's worth per call
+// regardless, so this only has to be comfortably larger than the pipe.
+#[cfg(target_os = "linux")]
+const NASH_TEE_SPLICE_CHUNK: usize = 1024 * 1024;
+
+// nash: how many finished tee threads stay parked waiting for the next pipeline. Thread creation
+// is the bulk of the ~0.13 ms a tapped pipeline costs to *set up*, and shell-heavy builds run
+// pipelines in the thousands (docs/NASH_STREAM_PERF_PLAN.md §P2.3). A shell runs its pipelines
+// mostly one at a time, so a small pool covers the common case; a burst of concurrent pipelines
+// simply spawns beyond it and lets the extra workers retire when they finish.
+const NASH_TEE_POOL_MAX_IDLE: usize = 4;
+
+// nash: stack for a tee worker. It copies through a heap buffer and hands the gate an already-built
+// event, so its own frames are shallow — a quarter of the 2 MiB default is ample and maps less.
+const NASH_TEE_STACK: usize = 512 * 1024;
+
+// nash: one link waiting to be copied — what a tee worker is handed.
+struct NashTeeJob {
+    src: std::io::PipeReader,
+    dst: std::io::PipeWriter,
+    pipeline_id: u64,
+    from_index: usize,
+    from_text: String,
+    to_text: String,
+}
+
+// nash: senders of the tee workers currently parked, newest last.
+static NASH_TEE_POOL: std::sync::Mutex<Vec<std::sync::mpsc::Sender<NashTeeJob>>> =
+    std::sync::Mutex::new(Vec::new());
+
+// nash: the process the parked workers belong to. A `fork` copies the pool's *memory* but none of
+// its threads, so a child that inherited it would hand jobs to workers that do not exist — and a
+// dropped job silently breaks the pipeline it was tapping. Checked (and reset) on every dispatch.
+static NASH_TEE_POOL_PID: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
+
+// nash: take a parked worker, if this process has one to spare.
+//
+// `try_lock`, never `lock`: this runs on the shell's own pipeline-setup path, so waiting on the
+// pool would put observation in front of a command — and after a `fork` the mutex can be left
+// locked by a thread that no longer exists. Failing to take a worker just means spawning one.
+fn nash_tee_take_idle() -> Option<std::sync::mpsc::Sender<NashTeeJob>> {
+    let mut pool = NASH_TEE_POOL.try_lock().ok()?;
+    let pid = std::process::id();
+    if NASH_TEE_POOL_PID.swap(pid, std::sync::atomic::Ordering::Relaxed) != pid {
+        pool.clear(); // inherited across a fork: those workers are not in this process
+    }
+    pool.pop()
+}
+
+// nash: park a worker that just finished a link. Returns whether it was taken — a worker the pool
+// has no room for (or cannot reach) simply exits.
+fn nash_tee_park(tx: &std::sync::mpsc::Sender<NashTeeJob>) -> bool {
+    let Ok(mut pool) = NASH_TEE_POOL.try_lock() else {
+        return false;
+    };
+    if pool.len() >= NASH_TEE_POOL_MAX_IDLE {
+        return false;
+    }
+    pool.push(tx.clone());
+    true
+}
+
+// nash: a parked worker's loop — copy a link, park again, wait for the next one.
+fn nash_tee_worker(
+    rx: &std::sync::mpsc::Receiver<NashTeeJob>,
+    parked: &std::sync::mpsc::Sender<NashTeeJob>,
+) {
+    while let Ok(job) = rx.recv() {
+        nash_run_pipe_tee(job);
+        if !nash_tee_park(parked) {
+            return;
+        }
+    }
+}
+
+// nash: hand a copier the link `src` (producer's output) → `dst` (consumer's input): it mirrors a
+// bounded prefix, then reports the link to the gate (docs/NASH.md §5.3). The synchronous copy
+// preserves pipe backpressure; EOF on `src` or EPIPE on `dst` (consumer gone, e.g. `yes | head`)
+// ends it and closes `dst` so the consumer sees EOF.
 // `to_index` is always `from_index + 1` — a link joins adjacent stages — so it is derived rather
 // than passed; `from_text`/`to_text` are the two stages already rendered by `nash_stage_text`.
 fn nash_spawn_pipe_tee(
-    mut src: std::io::PipeReader,
-    mut dst: std::io::PipeWriter,
+    src: std::io::PipeReader,
+    dst: std::io::PipeWriter,
     pipeline_id: u64,
     from_index: usize,
     from_text: String,
     to_text: String,
 ) {
-    std::thread::Builder::new()
+    let mut job = NashTeeJob {
+        src,
+        dst,
+        pipeline_id,
+        from_index,
+        from_text,
+        to_text,
+    };
+    // A worker that retired between being parked and being handed this job returns it, so try the
+    // next one down rather than losing the link.
+    while let Some(tx) = nash_tee_take_idle() {
+        match tx.send(job) {
+            Ok(()) => return,
+            Err(std::sync::mpsc::SendError(returned)) => job = returned,
+        }
+    }
+    let (tx, rx) = std::sync::mpsc::channel::<NashTeeJob>();
+    let parked = tx.clone();
+    // If the spawn fails there is nobody to copy this link and the job is dropped, closing both
+    // ends — the same outcome an unspawnable tee thread has always had, in a shell that is out of
+    // threads either way.
+    if std::thread::Builder::new()
         .name("nash-pipe-tee".into())
-        .spawn(move || {
-            use std::io::{Read, Write};
-            let mut buf = [0u8; 32 * 1024];
-            let mut captured: Vec<u8> = Vec::new();
-            let mut total: u64 = 0;
-            let mut truncated = false;
-            loop {
-                let n = match src.read(&mut buf) {
-                    Ok(0) => break,             // producer closed → done
-                    Ok(n) => n,
-                    Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
-                    Err(_) => break,
-                };
-                total += n as u64;
-                if captured.len() < NASH_PIPE_CAPTURE_CAP {
-                    let room = NASH_PIPE_CAPTURE_CAP - captured.len();
-                    let take = room.min(n);
-                    captured.extend_from_slice(&buf[..take]);
-                    if take < n {
-                        truncated = true;
-                    }
-                } else {
-                    truncated = true;
-                }
-                // Pass through; if the consumer went away, stop (its EOF is enough).
-                if dst.write_all(&buf[..n]).is_err() {
-                    break;
-                }
-            }
-            // Report BEFORE closing `dst`: dropping it unblocks the consumer, which
-            // lets the shell reach its exit-flush — so the event must already be
-            // enqueued to avoid a lost-event race on short-lived shells.
-            crate::gate::gate().on_pipe(crate::gate::PipeEvent {
-                pipeline_id,
-                from_index,
-                to_index: from_index + 1,
-                from_text,
-                to_text,
-                total_bytes: total,
-                captured,
-                truncated,
-            });
-            // Now close the write end so the consumer sees EOF.
-            drop(dst);
-        })
-        .ok();
+        .stack_size(NASH_TEE_STACK)
+        .spawn(move || nash_tee_worker(&rx, &parked))
+        .is_ok()
+    {
+        let _ = tx.send(job);
+    }
 }
 
+// nash: copy one link through to its consumer, then report it.
+fn nash_run_pipe_tee(job: NashTeeJob) {
+    let NashTeeJob {
+        mut src,
+        mut dst,
+        pipeline_id,
+        from_index,
+        from_text,
+        to_text,
+    } = job;
+    nash_grow_pipe(&src);
+    nash_grow_pipe(&dst);
+
+    let mut captured: Vec<u8> = Vec::new();
+    let mut total: u64 = 0;
+    // Only the prefix is ever copied through userspace; the rest of the stream — which is all of
+    // it, for the transfers that actually cost something — is handed to the kernel.
+    if nash_tee_prefix(&mut src, &mut dst, &mut captured, &mut total) {
+        nash_tee_passthrough(&mut src, &mut dst, &mut total);
+    }
+
+    // Report BEFORE closing `dst`: dropping it unblocks the consumer, which
+    // lets the shell reach its exit-flush — so the event must already be
+    // enqueued to avoid a lost-event race on short-lived shells.
+    crate::gate::gate().on_pipe(crate::gate::PipeEvent {
+        pipeline_id,
+        from_index,
+        to_index: from_index + 1,
+        from_text,
+        to_text,
+        total_bytes: total,
+        truncated: total > captured.len() as u64,
+        captured,
+    });
+    // Now close the write end so the consumer sees EOF.
+    drop(dst);
+}
+
+// nash: copy until the capture cap is reached, mirroring what goes past. Returns whether the link
+// is still open (false on EOF, or once either end gives up).
+fn nash_tee_prefix(
+    src: &mut std::io::PipeReader,
+    dst: &mut std::io::PipeWriter,
+    captured: &mut Vec<u8>,
+    total: &mut u64,
+) -> bool {
+    use std::io::{Read, Write};
+    let mut buf = vec![0u8; NASH_TEE_BUF];
+    while captured.len() < NASH_PIPE_CAPTURE_CAP {
+        let n = match src.read(&mut buf) {
+            Ok(0) => return false, // producer closed → done
+            Ok(n) => n,
+            Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
+            Err(_) => return false,
+        };
+        *total += n as u64;
+        let room = NASH_PIPE_CAPTURE_CAP - captured.len();
+        captured.extend_from_slice(&buf[..room.min(n)]);
+        // Pass through; if the consumer went away, stop (its EOF is enough).
+        if dst.write_all(&buf[..n]).is_err() {
+            return false;
+        }
+    }
+    true
+}
+
+// nash: the portable passthrough — the same read/write loop as the prefix phase, minus the
+// capture. Used on platforms without `splice`, and as the fallback when the kernel refuses it.
+fn nash_tee_copy(src: &mut std::io::PipeReader, dst: &mut std::io::PipeWriter, total: &mut u64) {
+    use std::io::{Read, Write};
+    let mut buf = vec![0u8; NASH_TEE_BUF];
+    loop {
+        let n = match src.read(&mut buf) {
+            Ok(0) => return,
+            Ok(n) => n,
+            Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
+            Err(_) => return,
+        };
+        *total += n as u64;
+        if dst.write_all(&buf[..n]).is_err() {
+            return;
+        }
+    }
+}
+
+// nash: move the rest of the link pipe-to-pipe inside the kernel (docs/NASH_STREAM_PERF_PLAN.md
+// §P2.1). Past the captured prefix the tee has nothing to look at, so there is no reason for the
+// bytes to enter this process at all — and the byte count comes back from the same call that
+// moves them.
+#[cfg(target_os = "linux")]
+fn nash_tee_passthrough(
+    src: &mut std::io::PipeReader,
+    dst: &mut std::io::PipeWriter,
+    total: &mut u64,
+) {
+    use std::os::fd::AsRawFd;
+    let (in_fd, out_fd) = (src.as_raw_fd(), dst.as_raw_fd());
+    loop {
+        // SAFETY: both descriptors are pipes owned by this job for the duration of the call, and
+        // the offsets are null because pipes have none.
+        let moved = unsafe {
+            libc::splice(
+                in_fd,
+                std::ptr::null_mut(),
+                out_fd,
+                std::ptr::null_mut(),
+                NASH_TEE_SPLICE_CHUNK,
+                libc::SPLICE_F_MOVE,
+            )
+        };
+        if moved == 0 {
+            return; // producer closed → done
+        }
+        if moved > 0 {
+            *total += u64::try_from(moved).unwrap_or(0);
+            continue;
+        }
+        let err = std::io::Error::last_os_error();
+        match err.raw_os_error() {
+            Some(libc::EINTR) => (),
+            // A kernel (or a seccomp policy) that will not splice these descriptors: finish the
+            // link the portable way rather than dropping it.
+            Some(libc::EINVAL | libc::ENOSYS | libc::EPERM) => {
+                nash_tee_copy(src, dst, total);
+                return;
+            }
+            // Consumer gone (EPIPE) or a link that broke: its EOF is enough.
+            _ => return,
+        }
+    }
+}
+
+#[cfg(not(target_os = "linux"))]
+fn nash_tee_passthrough(
+    src: &mut std::io::PipeReader,
+    dst: &mut std::io::PipeWriter,
+    total: &mut u64,
+) {
+    nash_tee_copy(src, dst, total);
+}
+
+// nash: grow a tapped pipe's buffer, best-effort. The kernel caps unprivileged growth at
+// `/proc/sys/fs/pipe-max-size` and simply refuses anything larger, which costs one failed syscall
+// per link and leaves the default in place.
+#[cfg(target_os = "linux")]
+fn nash_grow_pipe<F: std::os::fd::AsRawFd>(pipe: &F) {
+    // SAFETY: `fcntl` with `F_SETPIPE_SZ` takes an int and only reads the descriptor's pipe size.
+    let _ = unsafe { libc::fcntl(pipe.as_raw_fd(), libc::F_SETPIPE_SZ, NASH_TEE_PIPE_SIZE) };
+}
+
+#[cfg(not(target_os = "linux"))]
+fn nash_grow_pipe<F>(_pipe: &F) {}
+
 // nash: record a regular-file redirect for post-run read-back (docs/NASH.md §5.2a).
 // The child still gets the real fd; this only notes what to read back afterward.
 fn nash_record_file_redirect(