diff --git a/src/node/services/workspaceTurnManager.test.ts b/src/node/services/workspaceTurnManager.test.ts index 20588945ad..0ec48e2e06 100644 --- a/src/node/services/workspaceTurnManager.test.ts +++ b/src/node/services/workspaceTurnManager.test.ts @@ -18,12 +18,13 @@ import { Ok, Err, type Result } from "@/common/types/result"; import { DEFAULT_TASK_SETTINGS } from "@/common/types/tasks"; import type { SendMessageError } from "@/common/types/errors"; import type { ErrorEvent, StreamEndEvent } from "@/common/types/stream"; -import { createMuxMessage } from "@/common/types/message"; +import { createMuxMessage, type MuxMessageMetadata } from "@/common/types/message"; import type { WorkspaceMetadata } from "@/common/types/workspace"; import type { AIService } from "@/node/services/aiService"; import type { WorkspaceHost, BackgroundableForegroundWaiter, + QueueCutAttributionSnapshot, WorkspaceTurnManagerHost, } from "@/node/services/taskWorkspaceSeam"; import type { InitStateManager } from "@/node/services/initStateManager"; @@ -362,6 +363,63 @@ describe("WorkspaceTurnManager", () => { }; } + async function finalizeWorkspaceTurnStreamEndForTest( + taskService: WorkspaceTurnManager, + event: StreamEndEvent + ): Promise { + const internal = taskService as unknown as { + captureQueueCutAttributionSnapshot: (workspaceId: string) => QueueCutAttributionSnapshot; + finalizeWorkspaceTurnFromStreamEnd: ( + event: StreamEndEvent, + queueCutSnapshot: QueueCutAttributionSnapshot + ) => Promise; + }; + return await internal.finalizeWorkspaceTurnFromStreamEnd( + event, + internal.captureQueueCutAttributionSnapshot(event.workspaceId) + ); + } + + async function appendUncorrelatedWakeHistory(params: { + historyService: HistoryService; + parentId: string; + taskId: string; + workspaceId: string; + inputSynthetic: boolean; + }): Promise { + const prompt = createMuxMessage("turn-prompt", "user", "Summarize the repo", { + muxMetadata: workspaceTurnMuxMetadata(params.parentId, params.taskId), + }); + expect((await params.historyService.appendToHistory(params.workspaceId, prompt)).success).toBe( + true + ); + const input = createMuxMessage("wake-input", "user", "Continue after a wake", { + synthetic: params.inputSynthetic, + }); + expect((await params.historyService.appendToHistory(params.workspaceId, input)).success).toBe( + true + ); + const output = createMuxMessage("wake-output", "assistant", "Wake result", { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }); + expect((await params.historyService.appendToHistory(params.workspaceId, output)).success).toBe( + true + ); + return { + type: "stream-end", + workspaceId: params.workspaceId, + messageId: output.id, + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }, + parts: [{ type: "text", text: "Wake result" }], + }; + } + async function createWorkspaceLifecycleHarness( options: { archived?: boolean; @@ -3837,6 +3895,157 @@ describe("WorkspaceTurnManager", () => { expect(snapshot).toMatchObject({ status: "running", workspaceId: created.workspaceId }); }); + test("uncorrelated synthetic wake end leaves the active handle running", async () => { + const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest(); + const event = await appendUncorrelatedWakeHistory({ + historyService, + parentId, + taskId: created.taskId, + workspaceId: created.workspaceId, + inputSynthetic: true, + }); + + expect(await finalizeWorkspaceTurnStreamEndForTest(taskService, event)).toBe(true); + expect(await workspaceTurnSnapshot(taskService, parentId, created.taskId)).toMatchObject({ + status: "running", + }); + + const correlatedFinal: StreamEndEvent = { + ...event, + messageId: "real-final", + metadata: { + ...event.metadata, + muxMetadata: workspaceTurnMuxMetadata(parentId, created.taskId), + }, + parts: [{ type: "text", text: "Real final result" }], + }; + expect(await finalizeWorkspaceTurnStreamEndForTest(taskService, correlatedFinal)).toBe(true); + expect(await workspaceTurnSnapshot(taskService, parentId, created.taskId)).toMatchObject({ + status: "completed", + messageId: "real-final", + reportMarkdown: "Real final result", + }); + }); + + test("compaction-preserved turn anchor ignores an uncorrelated synthetic wake end", async () => { + 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: workspaceTurnMuxMetadata(parentId, created.taskId), + }, + }, + }); + expect( + (await historyService.appendToHistory(created.workspaceId, compactionSummary)).success + ).toBe(true); + const wakeInput = createMuxMessage("wake-input", "user", "Continue after a wake", { + synthetic: true, + }); + expect((await historyService.appendToHistory(created.workspaceId, wakeInput)).success).toBe( + true + ); + const wakeOutput = createMuxMessage("wake-output", "assistant", "Wake result", { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }); + expect((await historyService.appendToHistory(created.workspaceId, wakeOutput)).success).toBe( + true + ); + + expect( + await finalizeWorkspaceTurnStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: created.workspaceId, + messageId: wakeOutput.id, + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }, + parts: [{ type: "text", text: "Wake result" }], + }) + ).toBe(true); + expect(await workspaceTurnSnapshot(taskService, parentId, created.taskId)).toMatchObject({ + status: "running", + }); + }); + + test("manual user input still supersedes an active workspace turn on uncorrelated end", async () => { + const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest(); + const event = await appendUncorrelatedWakeHistory({ + historyService, + parentId, + taskId: created.taskId, + workspaceId: created.workspaceId, + inputSynthetic: false, + }); + + expect(await finalizeWorkspaceTurnStreamEndForTest(taskService, event)).toBe(true); + expect(await workspaceTurnSnapshot(taskService, parentId, created.taskId)).toMatchObject({ + status: "interrupted", + messageId: "wake-output", + error: "Workspace turn superseded by an uncorrelated workspace stream-end", + }); + }); + + test("malformed user-triggered auto-compaction still supersedes an active workspace turn", async () => { + const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest(); + const prompt = createMuxMessage("turn-prompt", "user", "Summarize the repo", { + muxMetadata: workspaceTurnMuxMetadata(parentId, created.taskId), + }); + expect((await historyService.appendToHistory(created.workspaceId, prompt)).success).toBe(true); + const compactionRequest = createMuxMessage( + "auto-compaction", + "user", + "Compacting before a new user prompt", + { + synthetic: true, + muxMetadata: { + type: "compaction-request", + rawCommand: "/compact", + parsed: null, + source: "auto-compaction", + } as unknown as MuxMessageMetadata, + } + ); + expect( + (await historyService.appendToHistory(created.workspaceId, compactionRequest)).success + ).toBe(true); + const wakeOutput = createMuxMessage("wake-output", "assistant", "Wake result", { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }); + expect((await historyService.appendToHistory(created.workspaceId, wakeOutput)).success).toBe( + true + ); + + expect( + await finalizeWorkspaceTurnStreamEndForTest(taskService, { + type: "stream-end", + workspaceId: created.workspaceId, + messageId: wakeOutput.id, + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + }, + parts: [{ type: "text", text: "Wake result" }], + }) + ).toBe(true); + expect(await workspaceTurnSnapshot(taskService, parentId, created.taskId)).toMatchObject({ + status: "interrupted", + messageId: wakeOutput.id, + error: "Workspace turn superseded by an uncorrelated workspace stream-end", + }); + }); + test("getWorkspaceTurnSnapshot recovers stale completed handles from matching history", async () => { const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest(); const appendResult = await historyService.appendToHistory( diff --git a/src/node/services/workspaceTurnManager.ts b/src/node/services/workspaceTurnManager.ts index b74f3b9d6e..0b4f7918c0 100644 --- a/src/node/services/workspaceTurnManager.ts +++ b/src/node/services/workspaceTurnManager.ts @@ -52,6 +52,7 @@ import { } from "@/common/types/backgroundWorkAttention"; import { createMuxMessage, + getCompactionFollowUpContent, parseWorkspaceTurnTaskCorrelation, type MuxMessage, type MuxMessageMetadata, @@ -281,6 +282,22 @@ const WORKSPACE_TURN_STALE_RESTART_ERROR = "Workspace turn interrupted after res const WORKSPACE_TURN_SUPERSEDED_BY_NEW_INPUT_ERROR = "Workspace turn superseded by new input in the target workspace; the workspace continues under that input and this delegated turn will not report"; +/** A human-authored child input that redirects the delegated turn. */ +function isManualChildWorkspaceInput(message: MuxMessage): boolean { + if (message.role !== "user") { + return false; + } + if (message.metadata?.synthetic !== true) { + return true; + } + const muxMetadata = message.metadata.muxMetadata; + return ( + muxMetadata?.type === "compaction-request" && + muxMetadata.source === "auto-compaction" && + getCompactionFollowUpContent(muxMetadata)?.dispatchOptions?.source !== "internal-resume" + ); +} + /** * Reason prefix persisted when the owner's OWN follow-up turn (task * kind="workspace", mode="existing", tool-end dispatch) cut its active @@ -4217,37 +4234,27 @@ export class WorkspaceTurnManager { }; } - private async isStreamEndBeforeWorkspaceTurnPrompt( + private isWorkspaceTurnAnchorForRecord( 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; + 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 + ); } - - 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; - } + if (muxMetadata?.type === "compaction-summary") { + const preserved = muxMetadata.pendingFollowUp?.workspaceTurnMetadata; + return ( + preserved?.taskHandleId === record.handleId && + preserved.ownerWorkspaceId === record.ownerWorkspaceId && + preserved.turnId === record.turnId + ); } - - return streamEndIndex !== -1 && promptIndex !== -1 && streamEndIndex < promptIndex; + return false; } private async interruptWorkspaceTurnFromUncorrelatedStreamEnd( @@ -4281,7 +4288,33 @@ export class WorkspaceTurnManager { 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, + }); + await this.settleWorkspaceTurnSupersededFromUncorrelatedStreamEnd(record, event); + return true; + } + + let streamEndIndex = -1; + let turnAnchorIndex = -1; + for (const [index, message] of historyResult.data.entries()) { + if (message.id === event.messageId) { + streamEndIndex = index; + } + if (this.isWorkspaceTurnAnchorForRecord(record, message)) { + turnAnchorIndex = index; + } + } + + if (streamEndIndex === -1 || turnAnchorIndex === -1) { + await this.settleWorkspaceTurnSupersededFromUncorrelatedStreamEnd(record, event); + return true; + } + if (streamEndIndex < turnAnchorIndex) { log.debug("Ignoring stale uncorrelated stream-end before queued workspace turn prompt", { workspaceId: event.workspaceId, taskHandleId: record.handleId, @@ -4290,6 +4323,19 @@ export class WorkspaceTurnManager { return true; } + const hasManualSupersessionInput = historyResult.data + .slice(turnAnchorIndex + 1, streamEndIndex) + .some(isManualChildWorkspaceInput); + if (hasManualSupersessionInput) { + await this.settleWorkspaceTurnSupersededFromUncorrelatedStreamEnd(record, event); + } + return true; + } + + private async settleWorkspaceTurnSupersededFromUncorrelatedStreamEnd( + record: WorkspaceTurnTaskHandleRecord, + event: StreamEndEvent + ): Promise { const error = "Workspace turn superseded by an uncorrelated workspace stream-end"; const next: WorkspaceTurnTaskHandleRecord = { ...record, @@ -4298,12 +4344,12 @@ export class WorkspaceTurnManager { messageId: event.messageId, error, }; + delete next.deferredMessageIds; await this.settleWorkspaceTurn({ record, next, waiterSettlement: { status: "error", error: new Error(error) }, }); - return true; } /**