diff --git a/packages/control-plane/src/session/durable-object.ts b/packages/control-plane/src/session/durable-object.ts index 9828e75ba4..01916ffcc5 100644 --- a/packages/control-plane/src/session/durable-object.ts +++ b/packages/control-plane/src/session/durable-object.ts @@ -869,6 +869,7 @@ export class SessionDO extends DurableObject { messageCreatedAt, terminalMessageCompletedAt: completedAt, }), + (error, completedAt) => this.messageQueue.failUnfinishedMessages(error, completedAt), this.statusService, (timestamp) => this.lifecycleManager.updateLastActivity(timestamp), () => this.lifecycleManager.scheduleInactivityCheck(), diff --git a/packages/control-plane/src/session/message-queue.test.ts b/packages/control-plane/src/session/message-queue.test.ts index e5fae69152..0fc7d1cf16 100644 --- a/packages/control-plane/src/session/message-queue.test.ts +++ b/packages/control-plane/src/session/message-queue.test.ts @@ -1346,4 +1346,56 @@ describe("SessionMessageQueue", () => { expect(h.repository.updateParticipantCoalesce).not.toHaveBeenCalled(); }); }); + + describe("failUnfinishedMessages", () => { + it("settles a prompt that never left pending and tells its caller why", () => { + const h = buildQueue(); + // A boot failure fails the prompt before it is ever dispatched, so it is + // still "pending" — failStuckProcessingMessage only ever sees "processing". + h.repository.listUnfinishedMessages.mockReturnValue([ + createMessage({ id: "msg-boot", status: "pending" }), + ]); + + const failed = h.queue.failUnfinishedMessages("git sync failed for owner/repo", 5000); + + expect(failed).toBe(1); + expect(h.repository.recordMessageCompletion).toHaveBeenCalledWith( + expect.objectContaining({ + messageId: "msg-boot", + success: false, + error: "git sync failed for owner/repo", + }), + 5000, + "pending" + ); + expect(h.callbackService.notifyComplete).toHaveBeenCalledWith( + "msg-boot", + false, + "git sync failed for owner/repo" + ); + }); + + it("settles a processing prompt against its own status", () => { + const h = buildQueue(); + h.repository.listUnfinishedMessages.mockReturnValue([ + createMessage({ id: "msg-running", status: "processing" }), + ]); + + h.queue.failUnfinishedMessages("sandbox died", 5000); + + expect(h.repository.recordMessageCompletion).toHaveBeenCalledWith( + expect.anything(), + 5000, + "processing" + ); + }); + + it("stays quiet when there is nothing in flight to settle", () => { + const h = buildQueue(); + h.repository.listUnfinishedMessages.mockReturnValue([]); + + expect(h.queue.failUnfinishedMessages("sandbox died", 5000)).toBe(0); + expect(h.callbackService.notifyComplete).not.toHaveBeenCalled(); + }); + }); }); diff --git a/packages/control-plane/src/session/message-queue.ts b/packages/control-plane/src/session/message-queue.ts index 4daeb60fb9..6d8245e051 100644 --- a/packages/control-plane/src/session/message-queue.ts +++ b/packages/control-plane/src/session/message-queue.ts @@ -531,6 +531,27 @@ export class SessionMessageQueue { await this.sessionStatus.reconcileAfterExecution(false); } + /** + * Settle every in-flight prompt because the sandbox reported it cannot continue. + * + * A boot failure fails the prompt before it is ever dispatched, so it is still + * `pending` — `failStuckProcessingMessage` only ever sees `processing` — and + * only `execution_complete` otherwise notifies the originating client. Without + * this the caller that asked for the work is told nothing at all. + */ + failUnfinishedMessages(error: string, completedAt: number): number { + let failed = 0; + for (const message of this.messageRepository.listUnfinishedMessages()) { + const expectedStatus = message.status === "processing" ? "processing" : "pending"; + if (this.failMessage(message, error, completedAt, expectedStatus)) failed++; + } + if (failed > 0) { + this.messenger.broadcast({ type: "processing_status", isProcessing: false }); + this.broadcastPromptQueue(); + } + return failed; + } + private failMessage( message: { id: string; created_at: number }, error: string, diff --git a/packages/control-plane/src/session/sandbox-events.test.ts b/packages/control-plane/src/session/sandbox-events.test.ts index a0ccebda23..1152938c7b 100644 --- a/packages/control-plane/src/session/sandbox-events.test.ts +++ b/packages/control-plane/src/session/sandbox-events.test.ts @@ -70,7 +70,11 @@ function createProcessor() { const diffService = { pinBaselines: vi.fn() }; const triggerSnapshot = vi.fn(async (_reason: string) => {}); const projectTerminalMessage = vi.fn(async () => {}); - const statusService = { reconcileAfterExecution: vi.fn(async (_success: boolean) => {}) }; + const failUnfinishedMessages = vi.fn((_error: string, _completedAt: number) => 0); + const statusService = { + reconcileAfterExecution: vi.fn(async (_success: boolean) => {}), + transition: vi.fn(async (_status: string) => true), + }; const scheduleInactivityCheck = vi.fn(async () => {}); const processMessageQueue = vi.fn(async () => {}); const broadcastPromptQueue = vi.fn(); @@ -100,6 +104,7 @@ function createProcessor() { applySessionTitleUpdate, triggerSnapshot, projectTerminalMessage, + failUnfinishedMessages, statusService as unknown as SessionStatusService, updateLastActivity, scheduleInactivityCheck, @@ -118,6 +123,7 @@ function createProcessor() { diffService, triggerSnapshot, projectTerminalMessage, + failUnfinishedMessages, statusService, scheduleInactivityCheck, processMessageQueue, @@ -939,4 +945,39 @@ describe("SessionSandboxEventProcessor", () => { expect(h.wsManager.send).not.toHaveBeenCalled(); }); }); + + it("settles in-flight prompts and drives the session to failed on a fatal error", async () => { + const h = createProcessor(); + + await h.processor.processSandboxEvent({ + type: "error", + error: "git sync failed for owner/repo", + sandboxId: "sandbox-1", + timestamp: 4000, + fatal: true, + } as never); + + // Settling is what notifies whoever asked for the work; before this a boot + // failure reached them as silence and the session sat "active" forever. + expect(h.failUnfinishedMessages).toHaveBeenCalledWith( + "git sync failed for owner/repo", + expect.any(Number) + ); + expect(h.statusService.transition).toHaveBeenCalledWith("failed"); + }); + + it("leaves in-flight prompts and status alone for a non-fatal error event", async () => { + const h = createProcessor(); + + await h.processor.processSandboxEvent({ + type: "error", + error: "tool blew up", + sandboxId: "sandbox-1", + timestamp: 4000, + messageId: "msg-1", + } as never); + + expect(h.failUnfinishedMessages).not.toHaveBeenCalled(); + expect(h.statusService.transition).not.toHaveBeenCalled(); + }); }); diff --git a/packages/control-plane/src/session/sandbox-events.ts b/packages/control-plane/src/session/sandbox-events.ts index 86ea172137..c666343cf2 100644 --- a/packages/control-plane/src/session/sandbox-events.ts +++ b/packages/control-plane/src/session/sandbox-events.ts @@ -61,6 +61,8 @@ export class SessionSandboxEventProcessor { messageCreatedAt: number, completedAt: number ) => Promise, + /** Settles in-flight prompts and notifies their caller; owned by the message queue. */ + private readonly failUnfinishedMessages: (error: string, completedAt: number) => number, private readonly statusService: SessionStatusService, private readonly updateLastActivity: (timestamp: number) => void, private readonly scheduleInactivityCheck: () => Promise, @@ -293,6 +295,18 @@ export class SessionSandboxEventProcessor { this.messenger.broadcast({ type: "sandbox_event", event }); + // A fatal error is the sandbox saying it cannot continue. Only + // execution_complete otherwise moves status or notifies the caller, so + // without this a boot failure — a repository that will not clone, an image + // that will not start — left the session sitting "active" with no messages + // forever, and whoever asked for the work hearing nothing at all. Settle + // in-flight prompts so the reason reaches them, and transition after persist + // and broadcast so it is on record first. + if (event.type === "error" && event.fatal) { + this.failUnfinishedMessages(event.error, now); + await this.statusService.transition("failed"); + } + if (CRITICAL_EVENT_TYPES.has(event.type)) { this.sendAck(ackId); } diff --git a/packages/shared/src/types/sandbox-events.ts b/packages/shared/src/types/sandbox-events.ts index 2d29628177..15e5bdedfc 100644 --- a/packages/shared/src/types/sandbox-events.ts +++ b/packages/shared/src/types/sandbox-events.ts @@ -101,9 +101,18 @@ export const sandboxEventSchema = z.discriminatedUnion("type", [ status: gitSyncStatusSchema, sha: z.string().optional(), }), - messageSandboxEventBaseSchema.extend({ + sandboxEventBaseSchema.extend({ type: z.literal("error"), error: z.string(), + // Message-scoped when the bridge reports a failure mid-execution, absent + // for a fatal boot error — that happens before any message exists, which + // is why requiring messageId here hard-rejected exactly the case the + // event most needs to carry. processSandboxEvent already falls back to the + // processing message when it is absent. + messageId: z.string().optional(), + // The sandbox cannot continue. Drives the session to "failed" and settles + // in-flight prompts; a plain error stays advisory and leaves status alone. + fatal: z.boolean().optional(), isSubtask: z.boolean().optional(), childSessionId: z.string().optional(), taskCallId: z.string().optional(), diff --git a/packages/web/src/lib/timeline-items.ts b/packages/web/src/lib/timeline-items.ts index b23d48fb86..28dce20705 100644 --- a/packages/web/src/lib/timeline-items.ts +++ b/packages/web/src/lib/timeline-items.ts @@ -119,6 +119,10 @@ export function buildTimelineItems(events: SandboxEvent[]): TimelineItem[] { for (const event of deduped) { if (!("isSubtask" in event) || !event.isSubtask || !("taskCallId" in event)) continue; if (!event.taskCallId) continue; + // An `error` event carries an optional messageId — a fatal boot error + // happens before any message exists — so it cannot be keyed to a parent + // Task call. Leave it at the top level rather than mis-nesting it. + if (!event.messageId) continue; const key = taskKey(event.messageId, event.taskCallId); if (!tasks.has(key)) continue;