diff --git a/Termind/Application/Agent/AgentBuiltInTools.swift b/Termind/Application/Agent/AgentBuiltInTools.swift index 3bea5315..dce4d63c 100644 --- a/Termind/Application/Agent/AgentBuiltInTools.swift +++ b/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 { diff --git a/Termind/Application/Agent/AgentCommandOutputStream.swift b/Termind/Application/Agent/AgentCommandOutputStream.swift new file mode 100644 index 00000000..28190021 --- /dev/null +++ b/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)) + } +} diff --git a/Termind/Application/Agent/AgentTool.swift b/Termind/Application/Agent/AgentTool.swift index 82a0e45b..26fbc42d 100644 --- a/Termind/Application/Agent/AgentTool.swift +++ b/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 { diff --git a/Termind/Application/Agent/AgentToolExecutor.swift b/Termind/Application/Agent/AgentToolExecutor.swift index d14188b4..bb50481a 100644 --- a/Termind/Application/Agent/AgentToolExecutor.swift +++ b/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)) } ) } diff --git a/Termind/Core/SSH/SSHExecSession.swift b/Termind/Core/SSH/SSHExecSession.swift index ec7c550f..8df80bf6 100644 --- a/Termind/Core/SSH/SSHExecSession.swift +++ b/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 diff --git a/Termind/Domain/Agent/AgentTypes.swift b/Termind/Domain/Agent/AgentTypes.swift index d321443b..86dca59f 100644 --- a/Termind/Domain/Agent/AgentTypes.swift +++ b/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, _, _): diff --git a/Termind/Features/AI/ConversationModel.swift b/Termind/Features/AI/ConversationModel.swift index 4d7a9c31..1906b625 100644 --- a/Termind/Features/AI/ConversationModel.swift +++ b/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 diff --git a/TermindTests/AgentConnectionPoolTests.swift b/TermindTests/AgentConnectionPoolTests.swift index 8e5c3876..20acb6ca 100644 --- a/TermindTests/AgentConnectionPoolTests.swift +++ b/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 diff --git a/TermindTests/AgentToolProgressTests.swift b/TermindTests/AgentToolProgressTests.swift new file mode 100644 index 00000000..4ac146c2 --- /dev/null +++ b/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 { + 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)) } } + } +}