From 65cec367c181b7c457aac901a10154f1086384f0 Mon Sep 17 00:00:00 2001 From: Mux Date: Mon, 24 Aug 2026 13:34:15 -0500 Subject: [PATCH 1/4] fix: wake stream-ends no longer falsely settle active workspace turns Sub-agent progress reports, bash-monitor wakes, and family messages dispatch synthetic user turns inside a child workspace. Their streams carry no workspace-turn correlation metadata, so their stream-end hit the uncorrelated-supersede path and settled the still-running delegated turn as interrupted. The owner saw a false terminal while the child kept working. The supersede interrupt now scans child history between the turn prompt and the uncorrelated stream-end. Only manual (non-synthetic) user input in that window proves the workspace was redirected; synthetic wake continuations keep the handle active so the real terminal report can settle it later. --- src/node/services/taskService.test.ts | 92 ++++++++++++++++++++++++++ src/node/services/taskService.ts | 94 +++++++++++++++++---------- 2 files changed, 152 insertions(+), 34 deletions(-) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index c5b33115f7..2163aab5a8 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -4870,6 +4870,98 @@ describe("TaskService", () => { expect(snapshot?.error).toBeUndefined(); }); + test("mid-turn uncorrelated wake stream-end does not interrupt an active workspace turn", async () => { + // Sub-agent progress reports and bash-monitor wakes dispatch new streams + // inside the child while the watched delegated turn still runs. When such a + // wake stream ends uncorrelated (no workspace-turn muxMetadata) mid-turn, + // the handle must stay active; settling it interrupted reports a false + // terminal to the owner while the child keeps working. + const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest(); + + const prompt = createMuxMessage("turn-prompt", "user", "Summarize the repo", { + muxMetadata: { + type: "workspace-turn-task", + taskHandleId: created.taskId, + ownerWorkspaceId: parentId, + turnId: "turn", + }, + }); + expect((await historyService.appendToHistory(created.workspaceId, prompt)).success).toBe(true); + const wakeOutput = createMuxMessage( + "msg_subagent_wake_stream", + "assistant", + "Sub-agent survey update", + { model: "anthropic:claude-opus-4-6", agentId: "exec", finishReason: "stop" } + ); + expect((await historyService.appendToHistory(created.workspaceId, wakeOutput)).success).toBe( + true + ); + + await handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: created.workspaceId, + messageId: "msg_subagent_wake_stream", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }, + parts: [{ type: "text", text: "Sub-agent survey update" }], + }); + + const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId); + expect(snapshot).toMatchObject({ status: "running", workspaceId: created.workspaceId }); + expect(snapshot?.error).toBeUndefined(); + }); + + test("manual user input still supersedes an active workspace turn on uncorrelated stream-end", async () => { + // Only manual (non-synthetic) user rows between the turn prompt and the + // uncorrelated stream-end prove the workspace was redirected away from the + // delegated turn; the handle must settle interrupted in that case. + const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest(); + + const prompt = createMuxMessage("turn-prompt", "user", "Summarize the repo", { + muxMetadata: { + type: "workspace-turn-task", + taskHandleId: created.taskId, + ownerWorkspaceId: parentId, + turnId: "turn", + }, + }); + expect((await historyService.appendToHistory(created.workspaceId, prompt)).success).toBe(true); + const manualInput = createMuxMessage("manual-input", "user", "Stop that, do something else"); + expect((await historyService.appendToHistory(created.workspaceId, manualInput)).success).toBe( + true + ); + const redirectOutput = createMuxMessage( + "msg_redirect_stream", + "assistant", + "Working on the new request", + { model: "anthropic:claude-opus-4-6", agentId: "exec", finishReason: "stop" } + ); + expect( + (await historyService.appendToHistory(created.workspaceId, redirectOutput)).success + ).toBe(true); + + await handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: created.workspaceId, + messageId: "msg_redirect_stream", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }, + parts: [{ type: "text", text: "Working on the new request" }], + }); + + const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId); + expect(snapshot).toMatchObject({ + status: "interrupted", + error: "Workspace turn superseded by an uncorrelated workspace stream-end", + }); + }); + test("compaction stream-end does not advance a running persistent child toward recovery", async () => { const config = await createTestConfig(rootDir); const projectPath = path.join(rootDir, "repo"); diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 2e462aef2f..488b0be3f9 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -1171,6 +1171,15 @@ function isResetBoundaryMessage(message: MuxMessage): boolean { return message.metadata?.contextBoundaryKind === CONTEXT_BOUNDARY_KINDS.RESET; } +/** + * A human-typed user row in the child workspace. Synthetic rows (sub-agent + * progress reports, bash-monitor wakes, family messages) are also dispatched as + * user turns, but they continue the delegated work instead of redirecting it. + */ +function isManualChildWorkspaceInput(message: MuxMessage): boolean { + return message.role === "user" && message.metadata?.synthetic !== true; +} + function isWorkflowSupersessionMessage(message: MuxMessage): boolean { return isManualUserSupersessionMessage(message) || isResetBoundaryMessage(message); } @@ -11103,39 +11112,6 @@ export class TaskService { }; } - private async isStreamEndBeforeWorkspaceTurnPrompt( - record: WorkspaceTurnTaskHandleRecord, - event: StreamEndEvent - ): Promise { - const historyResult = await this.historyService.getHistoryFromLatestBoundary(event.workspaceId); - if (!historyResult.success) { - log.warn("Could not compare uncorrelated stream-end history for workspace turn", { - workspaceId: event.workspaceId, - handleId: record.handleId, - error: historyResult.error, - }); - return false; - } - - let streamEndIndex = -1; - let promptIndex = -1; - for (const [index, message] of historyResult.data.entries()) { - if (message.id === event.messageId) { - streamEndIndex = index; - } - const metadata = this.getWorkspaceTurnMetadataFromValue(message.metadata?.muxMetadata); - if ( - metadata?.taskHandleId === record.handleId && - metadata.ownerWorkspaceId === record.ownerWorkspaceId && - metadata.turnId === record.turnId - ) { - promptIndex = index; - } - } - - return streamEndIndex !== -1 && promptIndex !== -1 && streamEndIndex < promptIndex; - } - private async interruptWorkspaceTurnFromUncorrelatedStreamEnd( event: StreamEndEvent ): Promise { @@ -11167,7 +11143,34 @@ export class TaskService { return true; } - if (await this.isStreamEndBeforeWorkspaceTurnPrompt(record, event)) { + const historyResult = await this.historyService.getHistoryFromLatestBoundary(event.workspaceId); + if (!historyResult.success) { + log.warn("Could not compare uncorrelated stream-end history for workspace turn", { + workspaceId: event.workspaceId, + handleId: record.handleId, + error: historyResult.error, + }); + return false; + } + + let streamEndIndex = -1; + let promptIndex = -1; + for (const [index, message] of historyResult.data.entries()) { + if (message.id === event.messageId) { + streamEndIndex = index; + } + const metadata = this.getWorkspaceTurnMetadataFromValue(message.metadata?.muxMetadata); + if ( + metadata?.taskHandleId === record.handleId && + metadata.ownerWorkspaceId === record.ownerWorkspaceId && + metadata.turnId === record.turnId + ) { + promptIndex = index; + } + } + + // A stale uncorrelated end that predates the turn's prompt cannot supersede it. + if (streamEndIndex !== -1 && promptIndex !== -1 && streamEndIndex < promptIndex) { log.debug("Ignoring stale uncorrelated stream-end before queued workspace turn prompt", { workspaceId: event.workspaceId, taskHandleId: record.handleId, @@ -11176,6 +11179,29 @@ export class TaskService { return true; } + // Wake continuations (sub-agent progress reports, bash-monitor wakes, family + // messages) dispatch as synthetic user turns whose streams can end without + // the turn's correlation metadata while the delegated turn is still open. + // Such an uncorrelated stream-end continues the delegated work; only manual + // input between the turn prompt and this stream-end proves the workspace was + // redirected away from the delegated turn. Without this guard the handle + // settles interrupted mid-work and the owner sees a false terminal while the + // child keeps streaming. + if (promptIndex !== -1) { + const scanEnd = streamEndIndex === -1 ? historyResult.data.length : streamEndIndex; + const hasManualSupersessionInput = historyResult.data + .slice(promptIndex + 1, scanEnd) + .some(isManualChildWorkspaceInput); + if (!hasManualSupersessionInput) { + log.debug("Ignoring uncorrelated wake stream-end while delegated workspace turn is open", { + workspaceId: event.workspaceId, + taskHandleId: record.handleId, + streamEndMessageId: event.messageId, + }); + return true; + } + } + const error = "Workspace turn superseded by an uncorrelated workspace stream-end"; const next: WorkspaceTurnTaskHandleRecord = { ...record, From 2314bfedb6e2fc268630f561d41adffbd513ffd3 Mon Sep 17 00:00:00 2001 From: Mux Date: Mon, 24 Aug 2026 13:49:37 -0500 Subject: [PATCH 2/4] fix: settle idle wake ends from the event; anchor compaction-preserved turns Codex review found two gaps in the uncorrelated wake guard: - An unconditional ignore could strand the handle as running when the synthetic wake stream was the delegated turn's last activity. The guard now defers only while continuation evidence is live (queued input, auto-retry, monitor wakes, correlated continuations, another active stream) and otherwise settles the handle from the wake event itself. A later correlated end still corrects it via allowTerminalResettle. - After auto-compaction, getHistoryFromLatestBoundary no longer shows the correlated prompt, so promptIndex stayed -1 and post-compaction wakes fell through to the supersede path. The history scan now also anchors on compaction-summary.pendingFollowUp.workspaceTurnMetadata, mirroring AgentSession.inheritOpenWorkspaceTurnMetadata. --- src/node/services/taskService.test.ts | 110 ++++++++++++++++++++++-- src/node/services/taskService.ts | 116 ++++++++++++++++++++++---- 2 files changed, 204 insertions(+), 22 deletions(-) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index 2163aab5a8..ae1395d886 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -4870,13 +4870,15 @@ describe("TaskService", () => { expect(snapshot?.error).toBeUndefined(); }); - test("mid-turn uncorrelated wake stream-end does not interrupt an active workspace turn", async () => { + test("uncorrelated wake stream-end with live continuation evidence keeps the handle active", async () => { // Sub-agent progress reports and bash-monitor wakes dispatch new streams // inside the child while the watched delegated turn still runs. When such a - // wake stream ends uncorrelated (no workspace-turn muxMetadata) mid-turn, - // the handle must stay active; settling it interrupted reports a false - // terminal to the owner while the child keeps working. - const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest(); + // wake stream ends uncorrelated (no workspace-turn muxMetadata) and more + // work is still queued or streaming, settling interrupted would report a + // false terminal while the child keeps working; defer to the real terminal. + const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest({ + hasPendingBashMonitorWakeContinuation: mock(() => true), + }); const prompt = createMuxMessage("turn-prompt", "user", "Summarize the repo", { muxMetadata: { @@ -4914,6 +4916,104 @@ describe("TaskService", () => { expect(snapshot?.error).toBeUndefined(); }); + test("idle uncorrelated wake stream-end settles the delegated turn from the wake output", async () => { + // When a synthetic wake's stream is the delegated turn's last activity and + // nothing else is queued or streaming, that end IS the turn outcome. + // Ignoring it unconditionally would strand the handle as running until + // restart recovery instead of delivering a real report to the owner. + const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest(); + + const prompt = createMuxMessage("turn-prompt", "user", "Summarize the repo", { + muxMetadata: { + type: "workspace-turn-task", + taskHandleId: created.taskId, + ownerWorkspaceId: parentId, + turnId: "turn", + }, + }); + expect((await historyService.appendToHistory(created.workspaceId, prompt)).success).toBe(true); + const wakeOutput = createMuxMessage( + "msg_subagent_wake_stream", + "assistant", + "Sub-agent survey update", + { model: "anthropic:claude-opus-4-6", agentId: "exec", finishReason: "stop" } + ); + expect((await historyService.appendToHistory(created.workspaceId, wakeOutput)).success).toBe( + true + ); + + await handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: created.workspaceId, + messageId: "msg_subagent_wake_stream", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }, + parts: [{ type: "text", text: "Sub-agent survey update" }], + }); + + const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId); + expect(snapshot).toMatchObject({ + status: "completed", + workspaceId: created.workspaceId, + reportMarkdown: "Sub-agent survey update", + }); + }); + + test("compaction-preserved correlation anchors the turn for post-compaction wake ends", async () => { + // Auto-compaction hides the correlated prompt behind a summary boundary; + // the preserved pendingFollowUp correlation must still anchor the turn so + // a later uncorrelated wake end does not supersede it. + const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest(); + + const compactionSummary = createMuxMessage("compaction-summary", "user", "Compacted context", { + muxMetadata: { + type: "compaction-summary", + pendingFollowUp: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + text: "Continue the delegated work", + workspaceTurnMetadata: { + type: "workspace-turn-task", + taskHandleId: created.taskId, + ownerWorkspaceId: parentId, + turnId: "turn", + }, + }, + }, + }); + expect( + (await historyService.appendToHistory(created.workspaceId, compactionSummary)).success + ).toBe(true); + const wakeOutput = createMuxMessage( + "msg_post_compaction_wake", + "assistant", + "Post-compaction wake output", + { model: "anthropic:claude-opus-4-6", agentId: "exec", finishReason: "stop" } + ); + expect((await historyService.appendToHistory(created.workspaceId, wakeOutput)).success).toBe( + true + ); + + await handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: created.workspaceId, + messageId: "msg_post_compaction_wake", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }, + parts: [{ type: "text", text: "Post-compaction wake output" }], + }); + + const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId); + expect(snapshot?.status).not.toBe("interrupted"); + expect(snapshot).toMatchObject({ status: "completed" }); + }); + test("manual user input still supersedes an active workspace turn on uncorrelated stream-end", async () => { // Only manual (non-synthetic) user rows between the turn prompt and the // uncorrelated stream-end prove the workspace was redirected away from the diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 488b0be3f9..08717dc655 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -11159,12 +11159,13 @@ export class TaskService { if (message.id === event.messageId) { streamEndIndex = index; } - const metadata = this.getWorkspaceTurnMetadataFromValue(message.metadata?.muxMetadata); - if ( - metadata?.taskHandleId === record.handleId && - metadata.ownerWorkspaceId === record.ownerWorkspaceId && - metadata.turnId === record.turnId - ) { + // Auto-compaction hides the original correlated prompt behind a summary + // boundary, but the boundary preserves the turn correlation on + // compaction-summary.pendingFollowUp.workspaceTurnMetadata (mirrors + // AgentSession.inheritOpenWorkspaceTurnMetadata). Treat either form as + // this turn's prompt anchor so post-compaction wakes stay guarded. + const anchor = this.getWorkspaceTurnAnchorForRecord(record, message); + if (anchor) { promptIndex = index; } } @@ -11179,29 +11180,111 @@ export class TaskService { return true; } - // Wake continuations (sub-agent progress reports, bash-monitor wakes, family - // messages) dispatch as synthetic user turns whose streams can end without - // the turn's correlation metadata while the delegated turn is still open. - // Such an uncorrelated stream-end continues the delegated work; only manual - // input between the turn prompt and this stream-end proves the workspace was - // redirected away from the delegated turn. Without this guard the handle - // settles interrupted mid-work and the owner sees a false terminal while the - // child keeps streaming. if (promptIndex !== -1) { const scanEnd = streamEndIndex === -1 ? historyResult.data.length : streamEndIndex; const hasManualSupersessionInput = historyResult.data .slice(promptIndex + 1, scanEnd) .some(isManualChildWorkspaceInput); - if (!hasManualSupersessionInput) { - log.debug("Ignoring uncorrelated wake stream-end while delegated workspace turn is open", { + if (hasManualSupersessionInput) { + await this.settleWorkspaceTurnSupersededFromUncorrelatedStreamEnd(record, event); + return true; + } + + // Synthetic wake rows continue the delegated work instead of redirecting + // it. Defer settlement while any continuation evidence is live; a later + // correlated terminal will settle the handle. When the workspace is + // otherwise idle, this uncorrelated wake end IS the delegated turn's last + // activity — settle from it so the owner receives a real outcome instead + // of a handle stranded as running until restart recovery. A later + // correlated end can still correct this via allowTerminalResettle. + if (this.hasLiveUncorrelatedWakeContinuationEvidence(event, record)) { + log.debug("Ignoring uncorrelated wake stream-end with live continuation evidence", { workspaceId: event.workspaceId, taskHandleId: record.handleId, streamEndMessageId: event.messageId, }); return true; } + const terminal = this.buildTerminalWorkspaceTurnRecordFromEvent(record, event, { + queueCutSupersedeEvidence: false, + }); + await this.settleWorkspaceTurn({ + record, + next: terminal, + waiterSettlement: + terminal.status === "completed" + ? { status: "completed", result: this.buildWorkspaceTurnWaitResult(terminal) } + : { status: "error", error: new Error(terminal.error ?? "Workspace turn failed") }, + allowTerminalResettle: true, + }); + return true; + } + + // No anchor in the visible history epoch (e.g. the boundary consumed it and + // preserved no correlation): fail safe to the pre-existing supersede path. + await this.settleWorkspaceTurnSupersededFromUncorrelatedStreamEnd(record, event); + return true; + } + + /** Whether the message anchors this record's delegated turn directly or via compaction preservation. */ + private getWorkspaceTurnAnchorForRecord( + record: WorkspaceTurnTaskHandleRecord, + message: MuxMessage + ): boolean { + const muxMetadata = message.metadata?.muxMetadata; + if (muxMetadata?.type === "workspace-turn-task") { + return ( + muxMetadata.taskHandleId === record.handleId && + muxMetadata.ownerWorkspaceId === record.ownerWorkspaceId && + muxMetadata.turnId === record.turnId + ); } + if (muxMetadata?.type === "compaction-summary") { + const preserved = muxMetadata.pendingFollowUp?.workspaceTurnMetadata; + return ( + preserved != null && + preserved.taskHandleId === record.handleId && + preserved.ownerWorkspaceId === record.ownerWorkspaceId && + preserved.turnId === record.turnId + ); + } + return false; + } + /** + * Live evidence that more work for this delegated turn is still queued or + * streaming after an uncorrelated wake stream-end. + */ + private hasLiveUncorrelatedWakeContinuationEvidence( + event: StreamEndEvent, + record: WorkspaceTurnTaskHandleRecord + ): boolean { + if (this.workspaceService.hasPendingQueuedOrPreparingTurn(event.workspaceId)) { + return true; + } + if (this.workspaceService.hasPendingAutoRetry(event.workspaceId)) { + return true; + } + if (this.workspaceService.hasPendingBashMonitorWakeContinuation(event.workspaceId)) { + return true; + } + const correlation = { + type: "workspace-turn-task" as const, + taskHandleId: record.handleId, + ownerWorkspaceId: record.ownerWorkspaceId, + turnId: record.turnId, + }; + if (this.workspaceService.hasPendingWorkspaceTurnContinuation(event.workspaceId, correlation)) { + return true; + } + const activeStream = this.aiService.getStreamInfo(event.workspaceId); + return activeStream != null && activeStream.messageId !== event.messageId; + } + + private async settleWorkspaceTurnSupersededFromUncorrelatedStreamEnd( + record: WorkspaceTurnTaskHandleRecord, + event: StreamEndEvent + ): Promise { const error = "Workspace turn superseded by an uncorrelated workspace stream-end"; const next: WorkspaceTurnTaskHandleRecord = { ...record, @@ -11215,7 +11298,6 @@ export class TaskService { next, waiterSettlement: { status: "error", error: new Error(error) }, }); - return true; } /** From 06a524b26689a0fcfcd7f6d8d4029c9836d71d40 Mon Sep 17 00:00:00 2001 From: Mux Date: Mon, 24 Aug 2026 13:56:02 -0500 Subject: [PATCH 3/4] fix: settle supersede fallback when wake-end history read fails Second Codex round: returning early on getHistoryFromLatestBoundary failure left the active handle running, and for a normal created workspace the owner's waiter then never settles because this stream-end was the only settlement signal. Restore the pre-existing supersede settlement as the terminal fallback and pin it with a test that injects the failure for every history read. --- src/node/services/taskService.test.ts | 41 +++++++++++++++++++++++++++ src/node/services/taskService.ts | 6 +++- 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index ae1395d886..dba24b9771 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -4916,6 +4916,47 @@ describe("TaskService", () => { expect(snapshot?.error).toBeUndefined(); }); + test("history read failure on uncorrelated wake end still settles the handle as interrupted", async () => { + // Without history we cannot classify the end, but it is the only stream-end + // that can settle the waiter. The supersede fallback preserves the + // pre-existing fail-safe instead of stranding the handle as running. + const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest(); + + const prompt = createMuxMessage("turn-prompt", "user", "Summarize the repo", { + muxMetadata: { + type: "workspace-turn-task", + taskHandleId: created.taskId, + ownerWorkspaceId: parentId, + turnId: "turn", + }, + }); + expect((await historyService.appendToHistory(created.workspaceId, prompt)).success).toBe(true); + + // Inject for every read in this test: handleStreamEnd performs earlier + // history reads that would consume a one-shot mock. + spyOn(historyService, "getHistoryFromLatestBoundary").mockImplementation(() => + Promise.resolve(Err("Failed to read history from boundary: injected test failure")) + ); + + await handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: created.workspaceId, + messageId: "msg_wake_stream", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }, + parts: [{ type: "text", text: "Wake output" }], + }); + + const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId); + expect(snapshot).toMatchObject({ + status: "interrupted", + error: "Workspace turn superseded by an uncorrelated workspace stream-end", + }); + }); + test("idle uncorrelated wake stream-end settles the delegated turn from the wake output", async () => { // When a synthetic wake's stream is the delegated turn's last activity and // nothing else is queued or streaming, that end IS the turn outcome. diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 08717dc655..d4f4865def 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -11150,7 +11150,11 @@ export class TaskService { handleId: record.handleId, error: historyResult.error, }); - return false; + // Terminal fallback: without history we cannot classify this end, and it + // is the only stream-end that can settle the waiter. Staying active would + // strand the owner's wait, so preserve the pre-existing supersede settle. + await this.settleWorkspaceTurnSupersededFromUncorrelatedStreamEnd(record, event); + return true; } let streamEndIndex = -1; From 69541ae9969b9bf070a675288cb66a326d6f61df Mon Sep 17 00:00:00 2001 From: Mux Date: Mon, 24 Aug 2026 14:06:26 -0500 Subject: [PATCH 4/4] fix: deterministic settlement for deferred and provisional wake outcomes Third Codex round, three gaps: - Nested-work blindness: continuation evidence now includes hasActiveWorkspaceTurnDeferredBlockers (descendant tasks, workflows, nested turns), which a continuing reported-agent workspace skips on the parent side of handleStreamEnd. - Immutable completion: uncorrelated wake settlements are stamped provisionalOutcome; settleWorkspaceTurn lets a later strictly- correlated stream-end replace them even when completed. - Vanished continuations: deferral persists the wake end as a deferred message ID (new persistDeferredUncorrelatedWakeEnd, correlation-free sibling of markWorkspaceTurnStreamEndDeferred). Stale recovery reads deferred uncorrelated ends back via buildDeferredWakeEndEventFromHistory, so a continuation that dies without another stream-end settles deterministically. isLiveWorkspaceTurn now counts auto-retry and monitor-wake continuations as runtime activity. --- src/node/services/taskHandleStore.ts | 7 ++ src/node/services/taskService.test.ts | 130 ++++++++++++++++++++++++++ src/node/services/taskService.ts | 125 +++++++++++++++++++++---- 3 files changed, 242 insertions(+), 20 deletions(-) diff --git a/src/node/services/taskHandleStore.ts b/src/node/services/taskHandleStore.ts index 5ac78f9163..67a26475ef 100644 --- a/src/node/services/taskHandleStore.ts +++ b/src/node/services/taskHandleStore.ts @@ -72,6 +72,12 @@ export interface WorkspaceTurnTaskHandleRecord { metadata: StreamEndEvent["metadata"]; }; deferredMessageIds?: string[]; + /** + * True only for a terminal settlement produced from an uncorrelated synthetic + * wake stream-end (no turn correlation existed on that stream). A later + * strictly-correlated stream-end may replace this provisional outcome. + */ + provisionalOutcome?: boolean; error?: string; /** * How the owner workspace's stream-end treats this workspace turn while active. @@ -118,6 +124,7 @@ const WorkspaceTurnTaskHandleRecordSchema = z .passthrough() .optional(), deferredMessageIds: z.array(z.string().min(1)).optional(), + provisionalOutcome: z.boolean().optional(), error: z.string().optional(), attentionPolicy: BackgroundWorkAttentionPolicySchema.optional(), directParentResultDeliveryRequiredAt: z.string().optional(), diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index dba24b9771..3ea7a40495 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -5003,6 +5003,136 @@ describe("TaskService", () => { }); }); + test("correlated terminal replaces a provisional idle-wake completion", async () => { + // The provisional completion from an uncorrelated wake end must stay + // replaceable: a later strictly-correlated stream-end carries the real + // outcome and wins over the synthetic wake report. + const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest(); + + const prompt = createMuxMessage("turn-prompt", "user", "Summarize the repo", { + muxMetadata: { + type: "workspace-turn-task", + taskHandleId: created.taskId, + ownerWorkspaceId: parentId, + turnId: "turn", + }, + }); + expect((await historyService.appendToHistory(created.workspaceId, prompt)).success).toBe(true); + const wakeOutput = createMuxMessage( + "msg_subagent_wake_stream", + "assistant", + "Sub-agent survey update", + { model: "anthropic:claude-opus-4-6", agentId: "exec", finishReason: "stop" } + ); + expect((await historyService.appendToHistory(created.workspaceId, wakeOutput)).success).toBe( + true + ); + + await handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: created.workspaceId, + messageId: "msg_subagent_wake_stream", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }, + parts: [{ type: "text", text: "Sub-agent survey update" }], + }); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId)).toMatchObject({ + status: "completed", + reportMarkdown: "Sub-agent survey update", + }); + + await handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: created.workspaceId, + messageId: "msg_real_final", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata: { + type: "workspace-turn-task", + taskHandleId: created.taskId, + ownerWorkspaceId: parentId, + turnId: "turn", + }, + }, + parts: [{ type: "text", text: "Real final report" }], + }); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId)).toMatchObject({ + status: "completed", + reportMarkdown: "Real final report", + }); + }); + + test("deferred wake end settles once continuation evidence disappears", async () => { + // Continuation signals can vanish without producing another stream-end + // (rejected queued send, abandoned retry, removed monitor wake). The + // persisted deferred marker must stop the handle counting as live so stale + // recovery settles it from the deferred wake output instead of blocking + // the owner until restart. + let monitorWakePending = true; + const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest({ + hasPendingBashMonitorWakeContinuation: mock(() => monitorWakePending), + }); + + const prompt = createMuxMessage("turn-prompt", "user", "Summarize the repo", { + muxMetadata: { + type: "workspace-turn-task", + taskHandleId: created.taskId, + ownerWorkspaceId: parentId, + turnId: "turn", + }, + }); + expect((await historyService.appendToHistory(created.workspaceId, prompt)).success).toBe(true); + const wakeOutput = createMuxMessage( + "msg_subagent_wake_stream", + "assistant", + "Sub-agent survey update", + { model: "anthropic:claude-opus-4-6", agentId: "exec", finishReason: "stop" } + ); + expect((await historyService.appendToHistory(created.workspaceId, wakeOutput)).success).toBe( + true + ); + + await handleTaskServiceStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: created.workspaceId, + messageId: "msg_subagent_wake_stream", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }, + parts: [{ type: "text", text: "Sub-agent survey update" }], + }); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId)).toMatchObject({ + status: "running", + }); + + const store = ( + taskService as unknown as { + taskHandleStore: { + getWorkspaceTurn: ( + ownerWorkspaceId: string, + handleId: string + ) => Promise<{ deferredMessageIds?: string[] } | null>; + }; + } + ).taskHandleStore; + const record = await store.getWorkspaceTurn(parentId, created.taskId); + expect(record?.deferredMessageIds).toContain("msg_subagent_wake_stream"); + + monitorWakePending = false; + expect(await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId)).toMatchObject({ + status: "completed", + reportMarkdown: "Sub-agent survey update", + }); + }); + test("compaction-preserved correlation anchors the turn for post-compaction wake ends", async () => { // Auto-compaction hides the correlated prompt behind a summary boundary; // the preserved pendingFollowUp correlation must still anchor the turn so diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index d4f4865def..a9321a4833 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -7066,15 +7066,20 @@ export class TaskService { "settleWorkspaceTurn requires current record to match workspaceId" ); - // A completed record is immutable; a self-heal-eligible settled record (transient - // error / stale restart interrupt — never an explicit user interrupt) may be - // corrected once by an explicitly allowed resettle, but only when the new settlement - // actually changes the outcome (duplicate stream-end replays must stay idempotent). + // A completed record is immutable — except one settled provisionally + // from an uncorrelated wake stream-end, which a later strictly- + // correlated stream-end must be able to replace with the real outcome. + // A self-heal-eligible settled record (transient error / stale restart + // interrupt — never an explicit user interrupt) may otherwise be + // corrected once by an explicitly allowed resettle, but only when the + // new settlement actually changes the outcome (duplicate stream-end + // replays must stay idempotent). + const currentIsProvisional = current.provisionalOutcome === true; const resettleStaleTerminal = params.allowTerminalResettle === true && this.isTerminalWorkspaceTurnStatus(current.status) && - current.status !== "completed" && - isSelfHealEligibleSettledWorkspaceTurn(current) && + (current.status !== "completed" || currentIsProvisional) && + (currentIsProvisional || isSelfHealEligibleSettledWorkspaceTurn(current)) && (params.next.status !== current.status || params.next.messageId !== current.messageId); if (this.isTerminalWorkspaceTurnStatus(current.status) && !resettleStaleTerminal) { const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(params.record.workspaceId); @@ -7124,6 +7129,7 @@ export class TaskService { nextStatus: nextRecord.status, }); delete nextRecord.terminalAttentionNotifiedAt; + delete nextRecord.provisionalOutcome; } const requiresDirectParentDelivery = this.workspaceTurnRequiresDirectParentDelivery(nextRecord); @@ -9809,7 +9815,9 @@ export class TaskService { const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId); const hasRuntimeActivity = this.aiService.isStreaming(record.workspaceId) || - this.workspaceService.hasPendingQueuedOrPreparingTurn(record.workspaceId); + this.workspaceService.hasPendingQueuedOrPreparingTurn(record.workspaceId) || + this.workspaceService.hasPendingAutoRetry(record.workspaceId) || + this.workspaceService.hasPendingBashMonitorWakeContinuation(record.workspaceId); if (hasRuntimeActivity) { return true; } @@ -10968,6 +10976,30 @@ export class TaskService { }; } + /** + * Terminal event for a deferred UNCORRELATED wake stream-end read back from + * history during stale recovery. The deferred marker replaces the correlation + * check that buildWorkspaceTurnStreamEndEventFromHistory performs. + */ + private buildDeferredWakeEndEventFromHistory( + record: WorkspaceTurnTaskHandleRecord, + message: MuxMessage + ): StreamEndEvent | null { + if (message.role !== "assistant" || message.metadata?.partial === true) { + return null; + } + return { + type: "stream-end", + workspaceId: record.workspaceId, + messageId: message.id, + metadata: { + ...message.metadata, + model: coerceNonEmptyString(message.metadata?.model) ?? record.modelString ?? defaultModel, + }, + parts: message.parts as StreamEndEvent["parts"], + }; + } + private buildTerminalWorkspaceTurnRecordFromEvent( record: WorkspaceTurnTaskHandleRecord, event: StreamEndEvent, @@ -10976,6 +11008,9 @@ export class TaskService { const baseRecord = { ...record }; delete baseRecord.error; delete baseRecord.deferredMessageIds; + // Correlated terminal settlements are final; never inherit the provisional + // marker from an earlier uncorrelated wake settlement. + delete baseRecord.provisionalOutcome; // A "tool-calls" finish on a delegated turn backed by queue-dispatch // evidence is a queue cut: some other queued input (a manual user message, // /compact) dispatched at the tool boundary and superseded the turn @@ -11054,10 +11089,18 @@ export class TaskService { const allowDeferredMessages = !(await this.hasActiveWorkspaceTurnDeferredBlockers(record)); for (const message of historyResult.data.toReversed()) { - if (this.isDeferredWorkspaceTurnMessage(record, message.id) && !allowDeferredMessages) { + const isDeferred = this.isDeferredWorkspaceTurnMessage(record, message.id); + if (isDeferred && !allowDeferredMessages) { continue; } - const event = this.buildWorkspaceTurnStreamEndEventFromHistory(record, message); + // Deferred uncorrelated wake ends carry no correlation to match, but the + // deferred marker itself proves this handler accepted the end as the + // turn's interim last activity. Recover them as terminal events once + // blockers clear; otherwise a continuation that vanishes without another + // stream-end would block the owner until restart. + const event = + this.buildWorkspaceTurnStreamEndEventFromHistory(record, message) ?? + (isDeferred ? this.buildDeferredWakeEndEventFromHistory(record, message) : null); if (event != null) { // History order alone cannot prove a queue cut (a later unrelated user // message is not causal evidence), so stale recovery conservatively @@ -11195,23 +11238,28 @@ export class TaskService { } // Synthetic wake rows continue the delegated work instead of redirecting - // it. Defer settlement while any continuation evidence is live; a later - // correlated terminal will settle the handle. When the workspace is - // otherwise idle, this uncorrelated wake end IS the delegated turn's last - // activity — settle from it so the owner receives a real outcome instead - // of a handle stranded as running until restart recovery. A later - // correlated end can still correct this via allowTerminalResettle. - if (this.hasLiveUncorrelatedWakeContinuationEvidence(event, record)) { - log.debug("Ignoring uncorrelated wake stream-end with live continuation evidence", { + // it. Defer settlement while any continuation evidence is live, and + // persist the end as deferred so the handle stops counting as live once + // that evidence disappears without producing another stream-end (a + // rejected queued send, an abandoned retry, a removed monitor wake). + // Stale recovery then settles from this persisted event. When the + // workspace is otherwise idle, this uncorrelated wake end IS the + // delegated turn's last activity — settle from it provisionally so the + // owner receives a real outcome; a later correlated end can still replace + // it via allowTerminalResettle + provisionalOutcome. + if (await this.hasLiveUncorrelatedWakeContinuationEvidence(event, record)) { + log.debug("Deferring uncorrelated wake stream-end with live continuation evidence", { workspaceId: event.workspaceId, taskHandleId: record.handleId, streamEndMessageId: event.messageId, }); + await this.persistDeferredUncorrelatedWakeEnd(record, event); return true; } const terminal = this.buildTerminalWorkspaceTurnRecordFromEvent(record, event, { queueCutSupersedeEvidence: false, }); + terminal.provisionalOutcome = true; await this.settleWorkspaceTurn({ record, next: terminal, @@ -11257,12 +11305,17 @@ export class TaskService { /** * Live evidence that more work for this delegated turn is still queued or - * streaming after an uncorrelated wake stream-end. + * streaming after an uncorrelated wake stream-end. Includes descendant, + * workflow, and nested-turn blockers because a continuing reported-agent + * workspace skips the parent-side active-descendant scan in handleStreamEnd. */ - private hasLiveUncorrelatedWakeContinuationEvidence( + private async hasLiveUncorrelatedWakeContinuationEvidence( event: StreamEndEvent, record: WorkspaceTurnTaskHandleRecord - ): boolean { + ): Promise { + if (await this.hasActiveWorkspaceTurnDeferredBlockers(record)) { + return true; + } if (this.workspaceService.hasPendingQueuedOrPreparingTurn(event.workspaceId)) { return true; } @@ -11285,6 +11338,36 @@ export class TaskService { return activeStream != null && activeStream.messageId !== event.messageId; } + /** + * Persists an uncorrelated wake stream-end as deferred on the active record. + * markWorkspaceTurnStreamEndDeferred cannot be reused because it keys off + * turn correlation, which these events intentionally lack. The marker makes + * isLiveWorkspaceTurn stop treating the handle as live once blockers clear, + * and lets stale recovery settle from this exact message. + */ + private async persistDeferredUncorrelatedWakeEnd( + record: WorkspaceTurnTaskHandleRecord, + event: StreamEndEvent + ): Promise { + await this.workspaceTurnSettlementLocks.withLock(record.handleId, async () => { + const current = await this.taskHandleStore.getWorkspaceTurn( + record.ownerWorkspaceId, + record.handleId + ); + if (current == null || !isActiveWorkspaceTurnTaskStatus(current.status)) { + return; + } + if ((current.deferredMessageIds ?? []).includes(event.messageId)) { + return; + } + await this.taskHandleStore.upsertWorkspaceTurn({ + ...current, + updatedAt: getIsoNow(), + deferredMessageIds: [...(current.deferredMessageIds ?? []), event.messageId], + }); + }); + } + private async settleWorkspaceTurnSupersededFromUncorrelatedStreamEnd( record: WorkspaceTurnTaskHandleRecord, event: StreamEndEvent @@ -11297,6 +11380,8 @@ export class TaskService { messageId: event.messageId, error, }; + delete next.provisionalOutcome; + delete next.deferredMessageIds; await this.settleWorkspaceTurn({ record, next,