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 c5b33115f7..3ea7a40495 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -4870,6 +4870,369 @@ describe("TaskService", () => { expect(snapshot?.error).toBeUndefined(); }); + 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) 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: { + 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("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. + // 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("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 + // 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 + // 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..a9321a4833 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); } @@ -7057,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); @@ -7115,6 +7129,7 @@ export class TaskService { nextStatus: nextRecord.status, }); delete nextRecord.terminalAttentionNotifiedAt; + delete nextRecord.provisionalOutcome; } const requiresDirectParentDelivery = this.workspaceTurnRequiresDirectParentDelivery(nextRecord); @@ -9800,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; } @@ -10959,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, @@ -10967,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 @@ -11045,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 @@ -11103,39 +11155,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 +11186,39 @@ 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, + }); + // 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; + let promptIndex = -1; + for (const [index, message] of historyResult.data.entries()) { + if (message.id === event.messageId) { + streamEndIndex = index; + } + // 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; + } + } + + // 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 +11227,151 @@ export class TaskService { return true; } + if (promptIndex !== -1) { + const scanEnd = streamEndIndex === -1 ? historyResult.data.length : streamEndIndex; + const hasManualSupersessionInput = historyResult.data + .slice(promptIndex + 1, scanEnd) + .some(isManualChildWorkspaceInput); + 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, 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, + 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. Includes descendant, + * workflow, and nested-turn blockers because a continuing reported-agent + * workspace skips the parent-side active-descendant scan in handleStreamEnd. + */ + private async hasLiveUncorrelatedWakeContinuationEvidence( + event: StreamEndEvent, + record: WorkspaceTurnTaskHandleRecord + ): Promise { + if (await this.hasActiveWorkspaceTurnDeferredBlockers(record)) { + return true; + } + 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; + } + + /** + * 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 + ): Promise { const error = "Workspace turn superseded by an uncorrelated workspace stream-end"; const next: WorkspaceTurnTaskHandleRecord = { ...record, @@ -11184,12 +11380,13 @@ export class TaskService { messageId: event.messageId, error, }; + delete next.provisionalOutcome; + delete next.deferredMessageIds; await this.settleWorkspaceTurn({ record, next, waiterSettlement: { status: "error", error: new Error(error) }, }); - return true; } /**