Show a command's output while it is still running

An `ssh_exec` card was blank for as long as the command took. `apt upgrade`
gave you a spinner for four minutes and then four minutes of text at once,
which is the worst possible ordering for the one thing you were watching
for — the line where it went wrong.

Pi has had the answer since `AgentToolUpdateCallback`: a tool may report
what it has so far, `executePreparedToolCall` turns each report into a
`tool_execution_update`, and the TUI redraws the card from it. Ported
here, with one change of address. Pi passes the callback as a fourth
argument to `execute` because it has nowhere else to put it;
`AgentToolExecutionContext` already threads per-call state to every tool,
so `report` goes there and no tool signature changes.

Where the partial goes is the whole design. `toolProgress` is not durable,
on the same grounds as `messageDelta` — history is written from run
snapshots, so a card that fills in as output arrives must not also stream
database writes. And it never reaches the transcript: the model is owed
one result per call, not a running commentary, and a turn that replayed
three partial views of the same command would pay for all three. The card
gets everything; the model gets what the tool finally returned. The card's
state is also left alone, because a call that has something to show has
not therefore finished.

`SSHExecSession` grows an `onOutput` handler called at the four points the
pump already captures a chunk. It runs on the session's serial queue
between two `libssh2` reads, so the contract is that it returns promptly
and never awaits — anything it blocks on stalls every command on that
host. `AgentCommandOutputStream` is the intended shape on the other side:
a lock, an accumulator, and a callback fired outside the lock.

Two details it exists to get right. Reports carry everything captured so
far rather than the newest chunk, because the card replaces its result
wholesale and would otherwise flicker instead of growing. And they are
throttled to one per 150ms — a build printing a thousand lines a second
would otherwise be a thousand SwiftUI invalidations per second to render
text nobody can read at that speed. Throttling drops the report, never the
output: what was captured is still captured, and the finished result
follows moments later regardless. A chunk boundary can also fall inside a
multi-byte character; that decodes as a replacement character and corrects
itself on the next report, and only the live view ever sees it, because
the finished result is decoded once, whole, by the session.

The test for the throttle is worth a note. The first version collected
reports into an actor and awaited them, and raced its own assertions. The
callback's contract is that it may not await, so a collector that has to
be awaited is not testing the thing — it is locked now, like the real one.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Termind/Application/Agent/AgentBuiltInTools.swift

@@ -490,18 +490,19 @@ actor AgentConnectionPool {
     func execute(
         _ command: String, workingDirectory: String?, stdin: Data?,
         timeout: TimeInterval, outputLimit: Int, hostID: UUID,
-        configuration override: ConfigurationResolving? = nil
+        configuration override: ConfigurationResolving? = nil,
+        onOutput: SSHExecOutputHandler? = nil
     ) async throws -> SSHExecResult {
         do {
             return try await exec(hostID: hostID, configuration: override).execute(
                 command, workingDirectory: workingDirectory, stdin: stdin,
-                timeout: timeout, outputLimit: outputLimit
+                timeout: timeout, outputLimit: outputLimit, onOutput: onOutput
             )
         } catch SSHExecSessionError.channelUnavailable {
             await disconnect(hostID: hostID, kind: .exec)
             return try await exec(hostID: hostID, configuration: override).execute(
                 command, workingDirectory: workingDirectory, stdin: stdin,
-                timeout: timeout, outputLimit: outputLimit
+                timeout: timeout, outputLimit: outputLimit, onOutput: onOutput
             )
         }
     }
@@ -776,6 +777,15 @@ nonisolated enum AgentBuiltInTools: AgentToolSchemaBuilding {
                 let interactiveConfiguration: AgentConnectionPool.ConfigurationResolving = { _ in
                     try await authentication.configuration()
                 }
+                // Live output, so a command that takes a minute is a card
+                // filling in rather than a spinner. Reported, never returned:
+                // the model is owed one result per call and gets it below.
+                let stream = AgentCommandOutputStream { [report = context.report] out, error in
+                    report(result(invocation, .object([
+                        "status": .string("running"),
+                        "stdout": .string(out), "stderr": .string(error),
+                    ])))
+                }
                 let output: SSHExecResult
                 while true {
                     do {
@@ -783,7 +793,8 @@ nonisolated enum AgentBuiltInTools: AgentToolSchemaBuilding {
                             command, workingDirectory: arguments.optionalString("cwd"), stdin: input,
                             timeout: TimeInterval(arguments.optionalInt("timeout_seconds") ?? 60),
                             outputLimit: 256 * 1_024, hostID: hostID,
-                            configuration: interactiveConfiguration
+                            configuration: interactiveConfiguration,
+                            onOutput: stream.handler
                         )
                         break
                     } catch let error as TermindSSHError where error.shouldRequestAuthentication {

Termind/Application/Agent/AgentCommandOutputStream.swift

@@ -0,0 +1,74 @@
+import Foundation
+
+/// Turns the chunks a running command emits into progress reports.
+///
+/// Locked rather than actor-isolated, because it is written from `libssh2`'s own
+/// serial queue between two reads — see `SSHExecOutputHandler`, which may not
+/// await. The callback fires outside the lock, so a slow reader cannot stall the
+/// command it is reading.
+///
+/// Reports are throttled. A build printing a thousand lines a second would
+/// otherwise raise a thousand events, each one a SwiftUI invalidation, to show
+/// text nobody can read at that speed. The final result is not throttled and
+/// arrives moments later either way, so the only thing a dropped report costs is
+/// an intermediate frame.
+nonisolated final class AgentCommandOutputStream: @unchecked Sendable {
+    private let lock = NSLock()
+    private var standardOut = Data()
+    private var standardError = Data()
+    private var lastReportedAt: ContinuousClock.Instant?
+    private let interval: Duration
+    private let limit: Int
+    private let report: @Sendable (String, String) -> Void
+
+    init(
+        interval: Duration = .milliseconds(150),
+        limit: Int = 256 * 1_024,
+        report: @escaping @Sendable (String, String) -> Void
+    ) {
+        self.interval = interval
+        self.limit = limit
+        self.report = report
+    }
+
+    /// The handler to hand to a session. Retains this stream for as long as the
+    /// command runs, which is exactly its useful life.
+    var handler: SSHExecOutputHandler {
+        { [self] chunk, isStandardError in append(chunk, isStandardError: isStandardError) }
+    }
+
+    /// Everything captured so far, whether or not a report was due.
+    var captured: (standardOut: String, standardError: String) {
+        lock.withLock { (decoded(standardOut), decoded(standardError)) }
+    }
+
+    private func append(_ chunk: Data, isStandardError: Bool) {
+        let due: (String, String)? = lock.withLock {
+            if isStandardError {
+                Self.appendCapped(chunk, to: &standardError, limit: limit)
+            } else {
+                Self.appendCapped(chunk, to: &standardOut, limit: limit)
+            }
+            let now = ContinuousClock.now
+            if let lastReportedAt, now - lastReportedAt < interval { return nil }
+            lastReportedAt = now
+            return (decoded(standardOut), decoded(standardError))
+        }
+        guard let due else { return }
+        report(due.0, due.1)
+    }
+
+    /// A chunk boundary can fall inside a multi-byte character, which decodes as
+    /// a replacement character and corrects itself on the next report. Only the
+    /// live view ever sees it: the finished result is decoded once, whole, by
+    /// the session.
+    private func decoded(_ data: Data) -> String { String(decoding: data, as: UTF8.self) }
+
+    /// The same head-keeping rule the session's own capture uses, so the live
+    /// view and the finished result are never showing different ends of the
+    /// same output.
+    private static func appendCapped(_ chunk: Data, to buffer: inout Data, limit: Int) {
+        guard buffer.count < limit else { return }
+        buffer.append(chunk.prefix(limit - buffer.count))
+    }
+}

Termind/Application/Agent/AgentTool.swift

@@ -7,6 +7,16 @@ nonisolated struct AgentToolExecutionContext: Sendable {
     var userInteraction: any AgentUserInteractionHandling
     var preflightMetadata: [String: AgentJSONValue]
     var authorize: @Sendable (AgentApprovalRequest) async -> AgentApprovalDecision
+    /// What the call has produced so far.
+    ///
+    /// Pi's `onUpdate`, which it passes as a fourth argument to `execute`
+    /// because it has no per-call context object to put it in. This is
+    /// addressed to the reader and not to the model: the transcript gets the
+    /// value the tool finally returns, never a partial one, so a tool may
+    /// report as often as it likes without inflating anything the model pays
+    /// for. Defaults to discarding, so a caller assembling a context by hand —
+    /// a test, a preview — need not care.
+    var report: @Sendable (AgentToolResult) -> Void = { _ in }
 }
 
 nonisolated protocol AgentTool: Sendable {

Termind/Application/Agent/AgentToolExecutor.swift

@@ -231,6 +231,9 @@ nonisolated struct AgentToolExecutor: Sendable {
             preflightMetadata: ready.executionMetadata,
             authorize: { [approval, permissionMode] request in
                 await approval.authorize(request, mode: permissionMode)
+            },
+            report: { [channel, invocation = ready.invocation] partial in
+                channel.emit(.toolProgress(invocation, partial))
             }
         )
     }

Termind/Core/SSH/SSHExecSession.swift

@@ -40,11 +40,34 @@ nonisolated protocol SSHExecSessionProtocol: Sendable {
     func connect(using configuration: SSHConnectionConfiguration) async throws
     func execute(
         _ command: String, workingDirectory: String?, stdin: Data?,
-        timeout: TimeInterval, outputLimit: Int
+        timeout: TimeInterval, outputLimit: Int, onOutput: SSHExecOutputHandler?
     ) async throws -> SSHExecResult
     func disconnect() async
 }
 
+/// Called with each chunk the command produces, and which stream it came from —
+/// `false` for stdout, `true` for stderr.
+///
+/// It runs on the session's own serial queue, between two `libssh2` reads, so it
+/// must return promptly and must never await: anything it blocks on stalls every
+/// command on that host. Appending to a lock-guarded buffer is the intended
+/// shape.
+typealias SSHExecOutputHandler = @Sendable (Data, Bool) -> Void
+
+nonisolated extension SSHExecSessionProtocol {
+    /// For the callers that only want the finished result — which is most of
+    /// them, and all of the ones that predate streaming.
+    func execute(
+        _ command: String, workingDirectory: String?, stdin: Data?,
+        timeout: TimeInterval, outputLimit: Int
+    ) async throws -> SSHExecResult {
+        try await execute(
+            command, workingDirectory: workingDirectory, stdin: stdin,
+            timeout: timeout, outputLimit: outputLimit, onOutput: nil
+        )
+    }
+}
+
 actor SSHExecSession: SSHExecSessionProtocol {
     private nonisolated let worker = SSHExecWorker()
 
@@ -57,13 +80,14 @@ actor SSHExecSession: SSHExecSessionProtocol {
     /// once, and the worker's queue would serialize them anyway.
     func execute(
         _ command: String, workingDirectory: String?, stdin: Data?,
-        timeout: TimeInterval, outputLimit: Int
+        timeout: TimeInterval, outputLimit: Int, onOutput: SSHExecOutputHandler?
     ) async throws -> SSHExecResult {
         let cancellation = TransferCancellationToken()
         return try await withTaskCancellationHandler {
             try await worker.execute(
                 command, workingDirectory: workingDirectory, stdin: stdin,
-                timeout: timeout, outputLimit: outputLimit, cancellation: cancellation
+                timeout: timeout, outputLimit: outputLimit, cancellation: cancellation,
+                onOutput: onOutput
             )
         } onCancel: {
             cancellation.cancel()
@@ -104,13 +128,15 @@ private nonisolated final class SSHExecWorker: @unchecked Sendable {
 
     func execute(
         _ command: String, workingDirectory: String?, stdin: Data?,
-        timeout: TimeInterval, outputLimit: Int, cancellation: TransferCancellationToken
+        timeout: TimeInterval, outputLimit: Int, cancellation: TransferCancellationToken,
+        onOutput: SSHExecOutputHandler?
     ) async throws -> SSHExecResult {
         try await perform {
             do {
                 return try self.executeOnQueue(
                     command, workingDirectory: workingDirectory, stdin: stdin,
-                    timeout: timeout, outputLimit: outputLimit, cancellation: cancellation
+                    timeout: timeout, outputLimit: outputLimit, cancellation: cancellation,
+                    onOutput: onOutput
                 )
             } catch {
                 // Still on the queue, and the channel defer has already run, so
@@ -124,7 +150,8 @@ private nonisolated final class SSHExecWorker: @unchecked Sendable {
 
     private func executeOnQueue(
         _ command: String, workingDirectory: String?, stdin: Data?,
-        timeout: TimeInterval, outputLimit: Int, cancellation: TransferCancellationToken
+        timeout: TimeInterval, outputLimit: Int, cancellation: TransferCancellationToken,
+        onOutput: SSHExecOutputHandler?
     ) throws -> SSHExecResult {
         guard !command.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty else {
             throw TermindSSHError.library("The command is empty.")
@@ -216,6 +243,7 @@ private nonisolated final class SSHExecWorker: @unchecked Sendable {
                     if count > 0 {
                         progressed = true
                         let chunk = Data(bytes: buffer, count: count)
+                        onOutput?(chunk, stream != 0)
                         if stream == 0 {
                             Self.appendCapped(
                                 chunk, to: &standardOut, limit: outputLimit,
@@ -251,6 +279,7 @@ private nonisolated final class SSHExecWorker: @unchecked Sendable {
                 let count = libssh2_channel_read_ex(channel, stream, &buffer, buffer.count)
                 guard count > 0 else { break }
                 let chunk = Data(bytes: buffer, count: count)
+                onOutput?(chunk, stream != 0)
                 if stream == 0 {
                     Self.appendCapped(
                         chunk, to: &standardOut, limit: outputLimit, truncated: &outTruncated

Termind/Domain/Agent/AgentTypes.swift

@@ -304,6 +304,14 @@ nonisolated enum AgentEvent: Hashable, Sendable {
     case steeringDelivered(AgentTranscriptMessage)
     case toolProposed(AgentToolInvocation, AgentToolDescriptor)
     case toolStarted(AgentToolInvocation)
+    /// What a call has produced so far, for a reader watching it happen.
+    ///
+    /// Deliberately absent from `isDurable` and from the model-facing
+    /// transcript, on the same grounds as `messageDelta`: history is written
+    /// from run snapshots, so a card that fills in as output arrives must not
+    /// also stream database writes, and the model is owed one result per call
+    /// rather than a running commentary.
+    case toolProgress(AgentToolInvocation, AgentToolResult)
     case toolFinished(AgentToolInvocation, AgentToolResult)
     case approvalRequested(AgentApprovalRequest)
     case reviewStarted(AgentApprovalRequest)
@@ -317,7 +325,7 @@ nonisolated enum AgentEvent: Hashable, Sendable {
         case .runStarted(let id):
             id
         case .toolProposed(let invocation, _), .toolStarted(let invocation),
-            .toolFinished(let invocation, _):
+            .toolProgress(let invocation, _), .toolFinished(let invocation, _):
             invocation.runID
         case .approvalRequested(let request), .reviewStarted(let request),
             .reviewFinished(let request, _), .reviewFailed(let request, _, _):

Termind/Features/AI/ConversationModel.swift

@@ -782,6 +782,14 @@ final class ConversationModel: Identifiable {
         case .toolProposed(let invocation, let descriptor):
             toolCards.append(AgentToolCard(invocation: invocation, descriptor: descriptor))
         case .toolStarted(let invocation): setToolState(invocation.id, .running)
+        // The card only, and the card's state is left alone: this is a call
+        // still running, showing its work. Nothing here reaches `transcript` —
+        // the model is owed one result per call, and it gets that from
+        // `toolFinished`.
+        case .toolProgress(let invocation, let partial):
+            if let index = toolCards.firstIndex(where: { $0.id == invocation.id }) {
+                toolCards[index].result = partial
+            }
         case .toolFinished(let invocation, let result):
             if let index = toolCards.firstIndex(where: { $0.id == invocation.id }) {
                 toolCards[index].result = result

TermindTests/AgentConnectionPoolTests.swift

@@ -301,10 +301,11 @@ private actor StubExecSession: SSHExecSessionProtocol {
 
     func execute(
         _ command: String, workingDirectory: String?, stdin: Data?,
-        timeout: TimeInterval, outputLimit: Int
+        timeout: TimeInterval, outputLimit: Int, onOutput: SSHExecOutputHandler?
     ) async throws -> SSHExecResult {
         executeCount += 1
         if !failures.isEmpty { throw failures.removeFirst() }
+        onOutput?(Data(command.utf8), false)
         return SSHExecResult(
             stdout: command, stderr: "", exitCode: 0, signal: nil,
             stdoutTruncated: false, stderrTruncated: false

TermindTests/AgentToolProgressTests.swift

@@ -0,0 +1,166 @@
+import Foundation
+import Testing
+
+@testable import Termind
+
+private let emptySchema = AgentJSONValue.object([
+    "type": .string("object"), "properties": .object([:]),
+    "required": .array([]), "additionalProperties": .bool(false),
+])
+
+private actor RequestLog {
+    var values: [AgentModelRequest] = []
+    func append(_ value: AgentModelRequest) { values.append(value) }
+}
+
+private struct TwoTurnModel: AgentModelStreaming {
+    let requests: RequestLog
+    let counter = Counter()
+
+    func stream(
+        _ request: AgentModelRequest
+    ) -> AsyncThrowingStream<AgentModelStreamEvent, any Error> {
+        AsyncThrowingStream { continuation in
+            Task {
+                await requests.append(request)
+                if await counter.next() == 0 {
+                    continuation.yield(.toolCallSnapshot(
+                        id: "1", providerItemID: nil, name: "slow", arguments: "{}"
+                    ))
+                    continuation.yield(.finished(.toolCalls))
+                } else {
+                    continuation.yield(.textDelta("done"))
+                    continuation.yield(.finished(.completed))
+                }
+                continuation.finish()
+            }
+        }
+    }
+
+    actor Counter {
+        var value = 0
+        func next() -> Int { defer { value += 1 }; return value }
+    }
+}
+
+/// Pi's `onUpdate`, ported: a tool may say what it has so far, and the loop
+/// neither waits for it nor tells the model about it.
+@Suite
+struct AgentToolProgressTests {
+
+    @Test
+    func partialsReachTheCardAndOnlyTheFinalResultReachesTheModel() async {
+        let requests = RequestLog()
+        let tool = AnyAgentTool(descriptor: AgentToolDescriptor(
+            name: "slow", summary: "slow", inputSchema: emptySchema,
+            target: .local, safety: .locallyReadOnly, concurrency: .parallel
+        )) { invocation, context in
+            context.report(AgentToolResult(callID: invocation.call.id, content: "10 of 20"))
+            context.report(AgentToolResult(callID: invocation.call.id, content: "19 of 20"))
+            return AgentToolResult(callID: invocation.call.id, content: "20 of 20")
+        }
+        let runtime = AgentRuntime(
+            model: TwoTurnModel(requests: requests), tools: [tool],
+            approval: AgentApprovalBroker(reviewer: nil, manualApproval: { _ in .allow })
+        )
+
+        var progress: [String] = []
+        var finished: [String] = []
+        for await event in await runtime.start(AgentRunRequest(
+            conversationID: UUID(), prompt: "go", permissionMode: .askForApproval
+        )) {
+            if case .toolProgress(_, let partial) = event { progress.append(partial.content) }
+            if case .toolFinished(_, let result) = event { finished.append(result.content) }
+        }
+
+        #expect(progress == ["10 of 20", "19 of 20"])
+        #expect(finished == ["20 of 20"])
+
+        // The model is owed one result per call, not a running commentary.
+        let replayed = await requests.values.last?.messages.filter { $0.role == .tool } ?? []
+        #expect(replayed.map(\.text) == ["20 of 20"])
+    }
+
+    @MainActor
+    @Test
+    func aProgressReportFillsTheCardWithoutFinishingIt() {
+        let conversation = ConversationModel()
+        let invocation = AgentToolInvocation(
+            runID: UUID(), call: AgentToolCall(id: "1", name: "slow", arguments: .object([:]))
+        )
+        let descriptor = AgentToolDescriptor(
+            name: "slow", summary: "slow", inputSchema: emptySchema,
+            target: .local, safety: .locallyReadOnly
+        )
+        conversation.handle(.toolProposed(invocation, descriptor))
+        conversation.handle(.toolStarted(invocation))
+
+        conversation.handle(.toolProgress(
+            invocation, AgentToolResult(callID: "1", content: "half way")
+        ))
+
+        #expect(conversation.toolCards.count == 1)
+        #expect(conversation.toolCards[0].result?.content == "half way")
+        // Still running: a card that showed "Completed" the moment it had
+        // something to show would be lying about the command.
+        #expect(conversation.toolCards[0].state == .running)
+        #expect(conversation.transcript.isEmpty)
+    }
+}
+
+/// The buffer between `libssh2`'s queue and the events the card is drawn from.
+@Suite
+struct AgentCommandOutputStreamTests {
+
+    @Test
+    func reportsCarryEverythingSoFarRatherThanTheLatestChunk() async {
+        let reports = Reports()
+        let stream = AgentCommandOutputStream(interval: .zero) { out, error in
+            reports.append(out, error)
+        }
+
+        stream.handler(Data("one\n".utf8), false)
+        stream.handler(Data("two\n".utf8), false)
+        stream.handler(Data("oops\n".utf8), true)
+
+        // The card replaces its result wholesale, so a report carrying only the
+        // newest chunk would make the output flicker rather than grow.
+        #expect(reports.values.map(\.0) == ["one\n", "one\ntwo\n", "one\ntwo\n"])
+        #expect(reports.values.map(\.1) == ["", "", "oops\n"])
+        #expect(stream.captured.standardOut == "one\ntwo\n")
+        #expect(stream.captured.standardError == "oops\n")
+    }
+
+    @Test
+    func chunksInsideTheIntervalAreCapturedButNotReported() async {
+        let reports = Reports()
+        let stream = AgentCommandOutputStream(interval: .seconds(60)) { out, error in
+            reports.append(out, error)
+        }
+
+        stream.handler(Data("first".utf8), false)
+        stream.handler(Data(" second".utf8), false)
+
+        #expect(reports.values.count == 1, "the second chunk is inside the window")
+        // Throttling a report must never drop the output it was reporting.
+        #expect(stream.captured.standardOut == "first second")
+    }
+
+    @Test
+    func captureStopsAtTheLimitTheSessionItselfUses() {
+        let stream = AgentCommandOutputStream(interval: .zero, limit: 4) { _, _ in }
+        stream.handler(Data("abc".utf8), false)
+        stream.handler(Data("def".utf8), false)
+        #expect(stream.captured.standardOut == "abcd")
+    }
+
+    /// Locked rather than an actor: the callback under test may not await, so
+    /// a collector that had to be awaited would be testing something else — and
+    /// would race the assertions, which is how this test first failed.
+    private nonisolated final class Reports: @unchecked Sendable {
+        private let lock = NSLock()
+        private var stored: [(String, String)] = []
+        var values: [(String, String)] { lock.withLock { stored } }
+        func append(_ out: String, _ error: String) { lock.withLock { stored.append((out, error)) } }
+    }
+}