diff --git a/Termind/Application/Agent/AgentBackgroundCommands.swift b/Termind/Application/Agent/AgentBackgroundCommands.swift index 4e1a9261..9ec5858d 100644 --- a/Termind/Application/Agent/AgentBackgroundCommands.swift +++ b/Termind/Application/Agent/AgentBackgroundCommands.swift @@ -44,6 +44,12 @@ actor AgentBackgroundCommands { let task: Task let session: any SSHExecSessionProtocol var lastAskedAt: ContinuousClock.Instant + /// How many `ssh_wait` calls are blocked on this right now. A command + /// somebody is already watching needs no announcement — they will + /// report the outcome themselves, and hearing it twice is worse than + /// hearing it once. + var waiters = 0 + var hasBeenAnnounced = false } private var entries: [UUID: Entry] = [:] @@ -69,7 +75,8 @@ actor AgentBackgroundCommands { func adopt( hostID: UUID, command: String, stream: AgentCommandOutputStream, task: Task, session: any SSHExecSessionProtocol, - now: ContinuousClock.Instant = .now + now: ContinuousClock.Instant = .now, + announce: (@Sendable (String) -> Void)? = nil ) -> UUID { let handle = UUID() entries[handle] = Entry( @@ -77,9 +84,52 @@ actor AgentBackgroundCommands { session: session, lastAskedAt: now ) startSweeping() + if let announce { + // Unstructured, and nothing awaits it: it lives exactly as long as + // the command does. See `settled` for what happens to a structured + // watcher of a task that ignores cancellation. + Task { [weak self] in + _ = await task.result + await self?.announceCompletion(handle, announce) + } + } return handle } + /// Tells whoever asked that a command nobody was watching has finished. + /// + /// The alternative is silence. A model that takes a handle and never comes + /// back never learns the outcome, and the output is swept ten minutes later + /// having been read by no one — a deploy whose result vanishes because the + /// agent got distracted is the quietest possible failure. + private func announceCompletion( + _ handle: UUID, _ announce: @Sendable (String) -> Void + ) async { + guard var entry = entries[handle], entry.waiters == 0, !entry.hasBeenAnnounced + else { return } + entry.hasBeenAnnounced = true + entries[handle] = entry + guard let result = try? await entry.task.value else { return } + let captured = entry.stream.captured + announce(Self.completionNotice( + handle: handle, command: entry.command, result: result, + standardOut: captured.standardOut, standardError: captured.standardError + )) + } + + static func completionNotice( + handle: UUID, command: String, result: SSHExecResult, + standardOut: String, standardError: String + ) -> String { + var notice = String( + localized: + "The background command `\(command)` has finished with exit code \(result.exitCode). Its handle was \(handle.uuidString)." + ) + if !standardOut.isEmpty { notice += "\n\nstdout:\n\(standardOut)" } + if !standardError.isEmpty { notice += "\n\nstderr:\n\(standardError)" } + return notice + } + /// Waits up to `timeout` for the command behind `handle` to finish. /// /// Returns what it has either way. A wait that expires is not an error and @@ -98,7 +148,12 @@ actor AgentBackgroundCommands { // returned, and its progress gate is shut. if let reporting { entry.stream.redirect(to: reporting) } - guard await Self.settled(entry.task, within: timeout) else { + // Counted across the wait so a completion that lands mid-wait is + // reported by this call rather than announced behind its back. + entries[handle]?.waiters += 1 + let didSettle = await Self.settled(entry.task, within: timeout) + entries[handle]?.waiters -= 1 + guard didSettle else { let captured = entry.stream.captured return AgentCommandOutcome( handle: handle, standardOut: captured.standardOut, diff --git a/Termind/Application/Agent/AgentBuiltInTools.swift b/Termind/Application/Agent/AgentBuiltInTools.swift index 50b7febf..4d64dcc5 100644 --- a/Termind/Application/Agent/AgentBuiltInTools.swift +++ b/Termind/Application/Agent/AgentBuiltInTools.swift @@ -972,7 +972,11 @@ nonisolated enum AgentBuiltInTools: AgentToolSchemaBuilding { await pool.detachExec(hostID: hostID, session: session) let handle = await background.adopt( hostID: hostID, command: command, stream: stream, - task: running, session: session + task: running, session: session, + // If the model never comes back for it, the outcome is + // pushed at the next model boundary instead of being + // swept ten minutes later unread. + announce: context.notify ) let captured = stream.captured return result(invocation, .object([ diff --git a/Termind/Application/Agent/AgentRuntime.swift b/Termind/Application/Agent/AgentRuntime.swift index f855bb65..8262b12b 100644 --- a/Termind/Application/Agent/AgentRuntime.swift +++ b/Termind/Application/Agent/AgentRuntime.swift @@ -44,6 +44,10 @@ actor AgentRuntime { private let channel: AgentEventChannel private var runTask: Task? private var steering: [AgentTranscriptMessage] = [] + /// Things to tell the model that the user is not shown. Same lifetime and + /// same delivery point as steering; a different event, because one is a + /// person talking and the other is a machine reporting. + private var notices: [AgentTranscriptMessage] = [] /// Whether a message queued *now* would still be delivered. /// /// Not the same question as "is there a run task". `complete` awaits its @@ -130,6 +134,22 @@ actor AgentRuntime { return true } + /// Queues something for the model that never appears in the transcript. + /// + /// `false` when there is no run left to tell — a command that finishes + /// after its conversation has gone quiet has nobody to report to, and the + /// model finds out by asking `ssh_wait` or not at all. Deliberately does + /// *not* keep the loop alive on its own: a report is not a request, and an + /// agent that had said its piece should not start talking again because a + /// command it forgot about came back. + @discardableResult + func notify(_ text: String) -> Bool { + guard acceptsSteering else { return false } + guard !text.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty else { return false } + notices.append(AgentTranscriptMessage(role: .user, text: text)) + return true + } + func cancel() { acceptsSteering = false runTask?.cancel() @@ -172,7 +192,8 @@ actor AgentRuntime { tools: tools, approval: approval, hooks: hooks, channel: channel, secretBroker: secretBroker, userInteraction: userInteraction, runID: runID, permissionMode: request.permissionMode, - userIntent: request.prompt + userIntent: request.prompt, + notify: { [weak self] text in Task { await self?.notify(text) } } ), mode: configuration.toolExecution, maximumConcurrency: configuration.maximumToolConcurrency @@ -183,7 +204,10 @@ actor AgentRuntime { for index in 0.. [AgentTranscriptMessage] { + let delivered = notices.map { queued in + var message = queued + message.createdAt = .now + return message + } + notices.removeAll() + for message in delivered { channel.emit(.noticeDelivered(message)) } + return delivered + } + private func deliverSteering() -> [AgentTranscriptMessage] { let delivered = steering.map { queued in var message = queued diff --git a/Termind/Application/Agent/AgentTool.swift b/Termind/Application/Agent/AgentTool.swift index 26fbc42d..06a5ef16 100644 --- a/Termind/Application/Agent/AgentTool.swift +++ b/Termind/Application/Agent/AgentTool.swift @@ -17,6 +17,14 @@ nonisolated struct AgentToolExecutionContext: Sendable { /// 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 } + /// Tell the model something after this call has already returned. + /// + /// For work a tool handed on rather than finished. Unlike `report`, this is + /// not scoped to the call — that is the whole point of it — and unlike a + /// result, it is never shown to the user. Dropped when there is no run left + /// to tell, which is the honest outcome: a command finishing after its + /// conversation went quiet has nobody to report to. + var notify: @Sendable (String) -> Void = { _ in } } nonisolated protocol AgentTool: Sendable { diff --git a/Termind/Application/Agent/AgentToolExecutor.swift b/Termind/Application/Agent/AgentToolExecutor.swift index 6a319dff..860614e5 100644 --- a/Termind/Application/Agent/AgentToolExecutor.swift +++ b/Termind/Application/Agent/AgentToolExecutor.swift @@ -114,6 +114,9 @@ nonisolated struct AgentToolExecutor: Sendable { let runID: UUID let permissionMode: ChatPermissionMode let userIntent: String + /// Deliberately not behind the progress gate: this exists for what a tool + /// says *after* it has returned. + var notify: @Sendable (String) -> Void = { _ in } func plan(_ call: AgentToolCall) async -> AgentPlannedToolCall { let proposed = AgentToolInvocation(runID: runID, call: call) @@ -262,7 +265,8 @@ nonisolated struct AgentToolExecutor: Sendable { }, report: gate.forward { [channel, invocation = ready.invocation] partial in channel.emit(.toolProgress(invocation, partial)) - } + }, + notify: notify ) } diff --git a/Termind/Domain/Agent/AgentTypes.swift b/Termind/Domain/Agent/AgentTypes.swift index 86dca59f..324d05ae 100644 --- a/Termind/Domain/Agent/AgentTypes.swift +++ b/Termind/Domain/Agent/AgentTypes.swift @@ -294,6 +294,15 @@ nonisolated enum AgentEvent: Hashable, Sendable { case messageStarted(AgentTranscriptMessage) case messageDelta(messageID: AgentTranscriptMessage.ID, text: String) case messageFinished(AgentTranscriptMessage) + /// Something the run told the model that the user was not shown. + /// + /// A background command finishing is addressed to the model: it asked for + /// the command and is owed the outcome. It is not addressed to the user, + /// and the transcript has only three roles — so a visible version would be + /// a bubble attributed to a person who did not say it. This lands in the + /// model-facing history only; the user learns about the command from its + /// card. + case noticeDelivered(AgentTranscriptMessage) /// A message the user queued mid-run has entered the model's context. /// /// The one moment at which a steered turn becomes part of the conversation. diff --git a/Termind/Features/AI/ConversationModel.swift b/Termind/Features/AI/ConversationModel.swift index 1906b625..763933a8 100644 --- a/Termind/Features/AI/ConversationModel.swift +++ b/Termind/Features/AI/ConversationModel.swift @@ -752,6 +752,11 @@ final class ConversationModel: Identifiable { )) } refreshRunState() + // The model-facing history only. There is no bubble for it on purpose: + // it is a report the run made to the model, and rendering it would + // attribute to the user words the user never wrote. + case .noticeDelivered(let message): + transcript.append(message) // A queued turn has entered the model's context, so now it is part of // the conversation — in both lists, at the position the run put it, and // no longer in the queue the composer is drawing. diff --git a/TermindTests/AgentBackgroundCommandTests.swift b/TermindTests/AgentBackgroundCommandTests.swift index a8458e1a..56d9adda 100644 --- a/TermindTests/AgentBackgroundCommandTests.swift +++ b/TermindTests/AgentBackgroundCommandTests.swift @@ -214,6 +214,55 @@ struct AgentBackgroundCommandTests { await released.release() } + // MARK: - Telling the model when nobody asked + + /// The quietest possible failure this design could have: a model takes a + /// handle, gets distracted, and the deploy's outcome is swept ten minutes + /// later having been read by no one. + @Test + func aCommandNobodyWaitedOnAnnouncesItself() async throws { + let commands = AgentBackgroundCommands(sweepInterval: nil) + let announced = Sink() + _ = await commands.adopt( + hostID: UUID(), command: "deploy.sh", + stream: AgentCommandOutputStream { _, _ in }, + task: Task { exited(0, stdout: "done") }, session: StubSession(), + announce: { announced.append($0) } + ) + + for _ in 0..<50 where announced.values.isEmpty { + try await Task.sleep(for: .milliseconds(10)) + } + + #expect(announced.values.count == 1) + #expect(announced.values.first?.contains("deploy.sh") == true) + #expect(announced.values.first?.contains("exit code 0") == true) + } + + /// The other half: somebody watching reports it themselves, so announcing + /// it as well would tell the model the same thing twice. + @Test + func aCommandSomebodyIsWaitingOnIsNotAlsoAnnounced() async throws { + let commands = AgentBackgroundCommands(sweepInterval: nil) + let announced = Sink() + let released = AsyncRelease() + let handle = await commands.adopt( + hostID: UUID(), command: "deploy.sh", + stream: AgentCommandOutputStream { _, _ in }, + task: Task { await released.wait(); return exited(0, stdout: "done") }, + session: StubSession(), + announce: { announced.append($0) } + ) + + async let outcome = commands.wait(handle, timeout: .seconds(30)) + try await Task.sleep(for: .milliseconds(20)) + await released.release() + + #expect(try await outcome.exitCode == 0) + try await Task.sleep(for: .milliseconds(50)) + #expect(announced.values.isEmpty, "the waiter already reported it") + } + // MARK: - Shape @Test @MainActor diff --git a/TermindTests/AgentRuntimeTests.swift b/TermindTests/AgentRuntimeTests.swift index d324cf30..cb1a2fa4 100644 --- a/TermindTests/AgentRuntimeTests.swift +++ b/TermindTests/AgentRuntimeTests.swift @@ -361,6 +361,68 @@ struct AgentRuntimeTests { #expect(await cancelled.values == ["hang"]) } + @Test + func aNoticeJoinsTheNextRequestTheRunWasAlreadyGoingToMake() async { + let requests = RequestRecorder() + let gate = AsyncGate() + let runtime = AgentRuntime( + model: RecordingScriptedAgentModel( + turns: [ + [.toolCallSnapshot(id: "1", providerItemID: nil, name: "wait", arguments: "{}"), + .finished(.toolCalls)], + [.textDelta("done"), .finished(.completed)], + ], + requests: requests + ), + tools: [AnyAgentTool(descriptor: AgentToolDescriptor( + name: "wait", summary: "wait", + inputSchema: .object(["type": .string("object")]), + target: .local, safety: .locallyReadOnly + )) { invocation, _ in + await gate.beginAndWait() + return AgentToolResult(callID: invocation.call.id, content: "released") + }], + approval: AgentApprovalBroker(reviewer: nil, manualApproval: { _ in .allow }) + ) + let stream = await runtime.start(AgentRunRequest( + conversationID: UUID(), prompt: "start", permissionMode: .askForApproval + )) + let consumer = Task { for await _ in stream {} } + + await gate.waitUntilStarted() + #expect(await runtime.notify("the deploy finished with exit code 0")) + await gate.release() + await consumer.value + + #expect(await requests.values.last?.messages.contains { + $0.text == "the deploy finished with exit code 0" + } == true) + } + + /// A report is not a request. An agent that has said its piece should not + /// start talking again because a command it forgot about came back — so a + /// notice, unlike steering, does not buy the run another turn. + @Test + func aNoticeAloneDoesNotKeepARunAlive() async { + let requests = RequestRecorder() + let gate = AsyncGate() + let runtime = AgentRuntime( + model: GatedAgentModel(requests: requests, gate: gate), tools: [], + approval: AgentApprovalBroker(reviewer: nil, manualApproval: { _ in .allow }) + ) + let stream = await runtime.start(AgentRunRequest( + conversationID: UUID(), prompt: "start", permissionMode: .askForApproval + )) + let consumer = Task { for await _ in stream {} } + + await gate.waitUntilStarted() + #expect(await runtime.notify("the deploy finished")) + await gate.release() + await consumer.value + + #expect(await requests.values.count == 1, "nothing here was worth another round trip") + } + /// A queue nobody will drain is worse than a refusal: the caller still has /// the words and can put them back in front of the user. @Test diff --git a/TermindTests/AgentSteeringTests.swift b/TermindTests/AgentSteeringTests.swift index dff33cf8..f01ef395 100644 --- a/TermindTests/AgentSteeringTests.swift +++ b/TermindTests/AgentSteeringTests.swift @@ -154,6 +154,24 @@ struct AgentSteeringTests { #expect(conversation.draft == "hello") } + /// A background command finishing is addressed to the model, which asked + /// for it. It is not addressed to the user, and the transcript has only + /// three roles — so a visible version would be a bubble attributed to + /// someone who never said it. + @Test + func aNoticeReachesTheModelWithoutAppearingInTheConversation() { + let conversation = ConversationModel() + let notice = AgentTranscriptMessage( + role: .user, text: "The background command `deploy.sh` has finished." + ) + + conversation.handle(.noticeDelivered(notice)) + + #expect(conversation.transcript == [notice]) + #expect(conversation.messages.isEmpty) + #expect(conversation.timeline.isEmpty) + } + private func startRun(_ conversation: ConversationModel) { conversation.start( runtime: AgentRuntime(