Revert "Tell the model when a background command nobody watched has finished"
This reverts commit f671913. The problem it addressed is real — a model
that takes a handle and never comes back never learns the outcome — but
the fix worked in a narrower band than the argument for it suggested, and
cost more than the band was worth.
It fired only when the command finished *and* nobody was in `ssh_wait`
*and* the run was still alive. That third condition is the one that
undoes it: a model which starts a deploy, tells the user it is running,
and ends its turn has no run left to be told, so `notify` was dropped —
in exactly the case where forgetting is most likely. What remained was
the narrow case of a model still working on something else in the same
run, which is also the case where it is most likely to remember.
Against that: up to 256 KB of output pushed into the context at a
boundary nothing asked for; a `user`-role message the user never wrote,
replayed in history forever, because the transcript has only three roles
and a report has no seat of its own; a waiter count, a per-command
watcher, an event and a queue. And a defect — `announceCompletion` marked
the entry announced without removing it, so a later `ssh_wait` returned
the same outcome a second time.
The two-bad-options problem it was really solving is that a model told
"still running" can either block a long `ssh_wait` — which is the turn
occupancy this whole line of work exists to remove — or poll in short
ones, at a full round trip each. The answer to that is to make a long
wait interruptible rather than to push at the model, which is what
replaces this. Cutting a *wait* short costs nothing at all: the command
keeps running, its session stays, its handle stays valid. That is the
same reasoning that made an interrupt wrong for `ssh_exec`, where it
would have cost a half-finished remote command, and right here.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
f671913c98
commit
fed5f39caf
10 files changed
+6
-255
No files matched your search
@@ -44,12 +44,6 @@ actor AgentBackgroundCommands {
|
||||
let task: Task<SSHExecResult, any Error>
|
||||
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] = [:]
|
||||
@@ -75,8 +69,7 @@ actor AgentBackgroundCommands {
|
||||
func adopt(
|
||||
hostID: UUID, command: String, stream: AgentCommandOutputStream,
|
||||
task: Task<SSHExecResult, any Error>, session: any SSHExecSessionProtocol,
|
||||
now: ContinuousClock.Instant = .now,
|
||||
announce: (@Sendable (String) -> Void)? = nil
|
||||
now: ContinuousClock.Instant = .now
|
||||
) -> UUID {
|
||||
let handle = UUID()
|
||||
entries[handle] = Entry(
|
||||
@@ -84,52 +77,9 @@ 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
|
||||
@@ -148,12 +98,7 @@ actor AgentBackgroundCommands {
|
||||
// returned, and its progress gate is shut.
|
||||
if let reporting { entry.stream.redirect(to: reporting) }
|
||||
|
||||
// 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 {
|
||||
guard await Self.settled(entry.task, within: timeout) else {
|
||||
let captured = entry.stream.captured
|
||||
return AgentCommandOutcome(
|
||||
handle: handle, standardOut: captured.standardOut,
|
||||
|
||||
@@ -972,11 +972,7 @@ 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,
|
||||
// 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
|
||||
task: running, session: session
|
||||
)
|
||||
let captured = stream.captured
|
||||
return result(invocation, .object([
|
||||
|
||||
@@ -44,10 +44,6 @@ actor AgentRuntime {
|
||||
private let channel: AgentEventChannel
|
||||
private var runTask: Task<Void, Never>?
|
||||
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
|
||||
@@ -134,22 +130,6 @@ 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()
|
||||
@@ -192,8 +172,7 @@ actor AgentRuntime {
|
||||
tools: tools, approval: approval, hooks: hooks, channel: channel,
|
||||
secretBroker: secretBroker, userInteraction: userInteraction,
|
||||
runID: runID, permissionMode: request.permissionMode,
|
||||
userIntent: request.prompt,
|
||||
notify: { [weak self] text in Task { await self?.notify(text) } }
|
||||
userIntent: request.prompt
|
||||
),
|
||||
mode: configuration.toolExecution,
|
||||
maximumConcurrency: configuration.maximumToolConcurrency
|
||||
@@ -204,10 +183,7 @@ actor AgentRuntime {
|
||||
for index in 0..<configuration.maxTurns {
|
||||
try Task.checkCancellation()
|
||||
channel.emit(.turnStarted(index: index))
|
||||
if !notices.isEmpty || !steering.isEmpty {
|
||||
// Reports first, then instructions: what happened while the
|
||||
// model was away is context for what it is being asked now.
|
||||
snapshot.messages.append(contentsOf: deliverNotices())
|
||||
if !steering.isEmpty {
|
||||
snapshot.messages.append(contentsOf: deliverSteering())
|
||||
await persist(snapshot)
|
||||
}
|
||||
@@ -300,17 +276,6 @@ actor AgentRuntime {
|
||||
/// and a message stamped while the previous turn was still streaming sorts
|
||||
/// above the reply the user was reading when they typed it — which is not
|
||||
/// where the model saw it.
|
||||
private func deliverNotices() -> [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
|
||||
|
||||
@@ -17,14 +17,6 @@ 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 {
|
||||
|
||||
@@ -114,9 +114,6 @@ 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)
|
||||
@@ -265,8 +262,7 @@ nonisolated struct AgentToolExecutor: Sendable {
|
||||
},
|
||||
report: gate.forward { [channel, invocation = ready.invocation] partial in
|
||||
channel.emit(.toolProgress(invocation, partial))
|
||||
},
|
||||
notify: notify
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -294,15 +294,6 @@ 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.
|
||||
|
||||
@@ -752,11 +752,6 @@ 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.
|
||||
|
||||
@@ -214,55 +214,6 @@ 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
|
||||
|
||||
@@ -361,68 +361,6 @@ 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
|
||||
|
||||
@@ -154,24 +154,6 @@ 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(
|
||||
|
||||
Reference in new issue
Block a user