From 201b8fd32276f8ab05633cbc318910441a4b8496 Mon Sep 17 00:00:00 2001 From: Mux Date: Tue, 1 Sep 2026 15:11:31 -0500 Subject: [PATCH 1/5] =?UTF-8?q?=F0=9F=A4=96=20fix:=20correlate=20workspace?= =?UTF-8?q?-turn=20liveness?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Use exact stream and queued-turn correlation before slot accounting. --- .../services/workspaceTurnManager.test.ts | 105 +++++++++++++++--- src/node/services/workspaceTurnManager.ts | 14 ++- 2 files changed, 101 insertions(+), 18 deletions(-) diff --git a/src/node/services/workspaceTurnManager.test.ts b/src/node/services/workspaceTurnManager.test.ts index 20588945ad..f1e6b033b1 100644 --- a/src/node/services/workspaceTurnManager.test.ts +++ b/src/node/services/workspaceTurnManager.test.ts @@ -21,6 +21,7 @@ import type { ErrorEvent, StreamEndEvent } from "@/common/types/stream"; import { createMuxMessage } from "@/common/types/message"; import type { WorkspaceMetadata } from "@/common/types/workspace"; import type { AIService } from "@/node/services/aiService"; +import type { StreamManager } from "@/node/services/streamManager"; import type { WorkspaceHost, BackgroundableForegroundWaiter, @@ -250,12 +251,25 @@ function createWorkspaceTurnManagerHost( }; } +function activeStreamInfo(muxMetadata: unknown, messageId: string) { + return { + messageId, + model: "test-model", + historySequence: 0, + startTime: 0, + parts: [], + toolCompletionTimestamps: new Map(), + muxMetadata, + }; +} + function createWorkspaceTurnManagerHarness( config: Config, overrides?: { aiService?: AIService; workspaceService?: WorkspaceHost; initStateManager?: InitStateManager; + streamManager?: StreamManager; } ): { historyService: HistoryService; @@ -284,7 +298,8 @@ function createWorkspaceTurnManagerHarness( workspaceService, initStateManager, taskHost, - terminalAttentionStore + terminalAttentionStore, + overrides?.streamManager ); return { @@ -322,6 +337,7 @@ describe("WorkspaceTurnManager", () => { getQueueCutCutter?: ReturnType; hasPendingAutoRetry?: ReturnType; waitForPendingStreamErrorRecoveryDecision?: ReturnType; + streamManager?: StreamManager; } = {} ) { const config = await createTestConfig(rootDir); @@ -336,6 +352,7 @@ describe("WorkspaceTurnManager", () => { const { historyService, taskService, taskHost } = createWorkspaceTurnManagerHarness(config, { aiService: aiMocks.aiService, workspaceService: workspaceMocks.workspaceService, + streamManager: options.streamManager, }); const created = await taskService.createWorkspaceTurn({ @@ -3560,7 +3577,7 @@ describe("WorkspaceTurnManager", () => { expect(aiMocks.stopStream).not.toHaveBeenCalled(); }); - test("createWorkspaceTurn reserves a slot before queueing a manually busy existing workspace", async () => { + test("createWorkspaceTurn repairs stale handles before parallel-slot accounting", async () => { const config = await createTestConfig(rootDir); stubStableIds(config, ["queuedhandle", "queuedturn"]); const { parentId, projectPath } = await saveLocalParentWorkspace(config, rootDir); @@ -3591,13 +3608,27 @@ describe("WorkspaceTurnManager", () => { const workspaceMocks = createWorkspaceServiceMocks({ sendMessage, isBusyForMessage: mock((workspaceId: string) => workspaceId === "childworkspace"), + hasPendingQueuedOrPreparingTurn: mock( + (workspaceId: string) => workspaceId === "otherworkspace" + ), }); const aiMocks = createAIServiceMocks(config, { isStreaming: mock((workspaceId: string) => workspaceId === "otherworkspace"), }); + const streamManager = { + getStreamInfo: mock((workspaceId: string) => + workspaceId === "otherworkspace" + ? activeStreamInfo( + workspaceTurnMuxMetadata(parentId, "wst_unrelated", "unrelatedturn"), + "unrelated-message" + ) + : undefined + ), + } as unknown as StreamManager; const { taskService } = createWorkspaceTurnManagerHarness(config, { aiService: aiMocks.aiService, workspaceService: workspaceMocks.workspaceService, + streamManager, }); const taskHandleStore = (taskService as unknown as { taskHandleStore: TaskHandleStore }) .taskHandleStore; @@ -3610,6 +3641,7 @@ describe("WorkspaceTurnManager", () => { createdWorkspace: true, }) ); + // The target has unrelated activity. It must not reserve a slot for this stale handle. await taskHandleStore.upsertWorkspaceTurn( workspaceTurnRecord(parentId, "otherworkspace", "wst_other", "running", { turnId: "otherturn", @@ -3626,10 +3658,14 @@ describe("WorkspaceTurnManager", () => { workspace: { mode: "existing", workspaceId: "childworkspace" }, }); - expect(result.success).toBe(false); - if (result.success) return; - expect(result.error).toContain("maxParallelAgentTasks exceeded"); - expect(sendMessage).not.toHaveBeenCalled(); + expect(result.success).toBe(true); + if (!result.success) return; + expect(result.data.taskId).toBe("wst_queuedhandle"); + expect(sendMessage).toHaveBeenCalledTimes(1); + expect(await workspaceTurnSnapshot(taskService, parentId, "wst_other")).toMatchObject({ + status: "interrupted", + error: "Workspace turn interrupted after restart", + }); }); test("createWorkspaceTurn counts active workspace turns across all owners", async () => { @@ -3715,9 +3751,19 @@ describe("WorkspaceTurnManager", () => { ); return cfg; }); - const isStreaming = mock((workspaceId: string) => workspaceId === reawakenedTaskId); - const { aiService } = createAIServiceMocks(config, { isStreaming }); - const { taskService, taskHost } = createWorkspaceTurnManagerHarness(config, { aiService }); + const streamManager = { + getStreamInfo: mock((workspaceId: string) => + workspaceId === reawakenedTaskId + ? activeStreamInfo( + workspaceTurnMuxMetadata(parentId, "wst_reawakened_quota", "turn-reawakened-quota"), + "reawakened-message" + ) + : undefined + ), + } as unknown as StreamManager; + const { taskService, taskHost } = createWorkspaceTurnManagerHarness(config, { + streamManager, + }); const taskHandleStore = (taskService as unknown as { taskHandleStore: TaskHandleStore }) .taskHandleStore; await taskHandleStore.upsertWorkspaceTurn( @@ -3760,13 +3806,34 @@ describe("WorkspaceTurnManager", () => { }); }); - test("active workspace turn count keeps startup-retrying handles live", async () => { - const hasPendingQueuedOrPreparingTurn = mock( - (workspaceId: string) => workspaceId === "childworkspace" - ); + test("active workspace turn count keeps correlated continuation activity live", async () => { + let ownerWorkspaceId = ""; + let activity: "stream" | "queued" | "auto-retry" | "monitor-wake" = "stream"; + const hasPendingWorkspaceTurnContinuation = mock( + (workspaceId: string, metadata: ReturnType) => + activity === "queued" && + workspaceId === "childworkspace" && + metadata.taskHandleId === "wst_handle" && + metadata.ownerWorkspaceId === ownerWorkspaceId && + metadata.turnId === "turn" + ); + const streamManager = { + getStreamInfo: mock((workspaceId: string) => + activity === "stream" && workspaceId === "childworkspace" + ? activeStreamInfo( + workspaceTurnMuxMetadata(ownerWorkspaceId, "wst_handle", "turn"), + "correlated-message" + ) + : undefined + ), + } as unknown as StreamManager; const { parentId, taskService } = await startWorkspaceTurnForTest({ - hasPendingQueuedOrPreparingTurn, + hasPendingWorkspaceTurnContinuation, + hasPendingAutoRetry: mock(() => activity === "auto-retry"), + hasPendingBashMonitorWakeContinuation: mock(() => activity === "monitor-wake"), + streamManager, }); + ownerWorkspaceId = parentId; const internal = taskService as unknown as { activeWorkspaceTurnHandleByWorkspaceId: Map< string, @@ -3776,8 +3843,14 @@ describe("WorkspaceTurnManager", () => { }; internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); - expect(await internal.countActiveWorkspaceTurns()).toBe(1); - expect(hasPendingQueuedOrPreparingTurn).toHaveBeenCalledWith("childworkspace"); + for (const nextActivity of ["stream", "queued", "auto-retry", "monitor-wake"] as const) { + activity = nextActivity; + expect(await internal.countActiveWorkspaceTurns()).toBe(1); + } + expect(hasPendingWorkspaceTurnContinuation).toHaveBeenCalledWith( + "childworkspace", + workspaceTurnMuxMetadata(parentId, "wst_handle", "turn") + ); const snapshot = await workspaceTurnSnapshot(taskService, parentId); expect(snapshot).toMatchObject({ status: "running", workspaceId: "childworkspace" }); diff --git a/src/node/services/workspaceTurnManager.ts b/src/node/services/workspaceTurnManager.ts index b74f3b9d6e..caf594fcbf 100644 --- a/src/node/services/workspaceTurnManager.ts +++ b/src/node/services/workspaceTurnManager.ts @@ -3802,9 +3802,19 @@ export class WorkspaceTurnManager { private async isLiveWorkspaceTurn(record: WorkspaceTurnTaskHandleRecord): Promise { const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId); + const activeStreamCorrelation = parseWorkspaceTurnTaskCorrelation( + this.streamManager?.getStreamInfo(record.workspaceId)?.muxMetadata + ); const hasRuntimeActivity = - this.aiService.isStreaming(record.workspaceId) || - this.workspaceService.hasPendingQueuedOrPreparingTurn(record.workspaceId); + (activeStreamCorrelation?.taskHandleId === record.handleId && + activeStreamCorrelation.ownerWorkspaceId === record.ownerWorkspaceId && + activeStreamCorrelation.turnId === record.turnId) || + this.workspaceService.hasPendingWorkspaceTurnContinuation( + record.workspaceId, + this.buildWorkspaceTurnMuxMetadata(record) + ) || + this.workspaceService.hasPendingAutoRetry(record.workspaceId) || + this.workspaceService.hasPendingBashMonitorWakeContinuation(record.workspaceId); if (hasRuntimeActivity) { return true; } From 4d2a44786d185d0f9b187a296c70e702b5837109 Mon Sep 17 00:00:00 2001 From: Mux Date: Tue, 1 Sep 2026 15:16:53 -0500 Subject: [PATCH 2/5] =?UTF-8?q?=F0=9F=A4=96=20tests:=20model=20exact=20wor?= =?UTF-8?q?kspace-turn=20continuations?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Keep busy-workspace admission tests aligned with exact handle liveness. --- src/node/services/workspaceTurnManager.test.ts | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/src/node/services/workspaceTurnManager.test.ts b/src/node/services/workspaceTurnManager.test.ts index f1e6b033b1..91426aebfc 100644 --- a/src/node/services/workspaceTurnManager.test.ts +++ b/src/node/services/workspaceTurnManager.test.ts @@ -3493,6 +3493,10 @@ describe("WorkspaceTurnManager", () => { (workspaceId: string, handleId: string) => workspaceId === "childworkspace" && handleId === "wst_secondhandle" ), + hasPendingWorkspaceTurnContinuation: mock( + (workspaceId: string, metadata: ReturnType) => + workspaceId === "childworkspace" && metadata.taskHandleId === "wst_secondhandle" + ), isBusyForMessage, hasQueuedMessages, }); @@ -4209,12 +4213,13 @@ describe("WorkspaceTurnManager", () => { }); test("mode=existing tool-end follow-up reports the same-owner turn it may supersede", async () => { - const hasPendingQueuedOrPreparingTurn = mock( - (workspaceId: string) => workspaceId === "childworkspace" + const hasPendingWorkspaceTurnContinuation = mock( + (workspaceId: string, metadata: ReturnType) => + workspaceId === "childworkspace" && metadata.taskHandleId === "wst_handle2" ); const { parentId, taskService, workspaceMocks } = await startWorkspaceTurnForTest({ stableIds: ["handle", "turn", "handle2", "turn2", "handle3", "turn3"], - hasPendingQueuedOrPreparingTurn, + hasPendingWorkspaceTurnContinuation, }); workspaceMocks.isBusyForMessage.mockImplementation( (workspaceId: string) => workspaceId === "childworkspace" @@ -4269,12 +4274,13 @@ describe("WorkspaceTurnManager", () => { // queued, C supersedes B (not A) at B's first boundary — and B's own // settlement wake is suppressed, so C's announcement is the only place // B's interruption can surface. - const hasPendingQueuedOrPreparingTurn = mock( - (workspaceId: string) => workspaceId === "childworkspace" + const hasPendingWorkspaceTurnContinuation = mock( + (workspaceId: string, metadata: ReturnType) => + workspaceId === "childworkspace" && metadata.taskHandleId === "wst_handle2" ); const { parentId, taskService, workspaceMocks } = await startWorkspaceTurnForTest({ stableIds: ["handle", "turn", "handle2", "turn2", "handle3", "turn3"], - hasPendingQueuedOrPreparingTurn, + hasPendingWorkspaceTurnContinuation, }); workspaceMocks.isBusyForMessage.mockImplementation( (workspaceId: string) => workspaceId === "childworkspace" From 3c1d37bb14428ab32797320c1d548b80c3df64d0 Mon Sep 17 00:00:00 2001 From: Mux Date: Tue, 1 Sep 2026 15:33:22 -0500 Subject: [PATCH 3/5] =?UTF-8?q?=F0=9F=A4=96=20fix:=20reject=20unrelated=20?= =?UTF-8?q?workspace-turn=20activity?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../services/workspaceTurnManager.test.ts | 18 +++++- src/node/services/workspaceTurnManager.ts | 62 +++++++++++++++---- 2 files changed, 67 insertions(+), 13 deletions(-) diff --git a/src/node/services/workspaceTurnManager.test.ts b/src/node/services/workspaceTurnManager.test.ts index 91426aebfc..d8b0c5f3ce 100644 --- a/src/node/services/workspaceTurnManager.test.ts +++ b/src/node/services/workspaceTurnManager.test.ts @@ -3652,8 +3652,10 @@ describe("WorkspaceTurnManager", () => { createdAt, updatedAt: createdAt, createdWorkspace: true, + deferredMessageIds: ["assistant-deferred"], }) ); + markWorkspaceTurnActive(taskService, "otherworkspace", "wst_other", parentId); const result = await taskService.createWorkspaceTurn({ ownerWorkspaceId: parentId, @@ -5638,20 +5640,34 @@ describe("WorkspaceTurnManager", () => { }, ]) { test(scenario.name, async () => { + let ownerWorkspaceId = ""; let retryDecisionAwaited = false; const pending = mock( (workspaceId: string) => retryDecisionAwaited && workspaceId === "childworkspace" ); + const hasPendingWorkspaceTurnContinuation = mock( + (workspaceId: string, metadata: ReturnType) => + scenario.pending === "queued" && + retryDecisionAwaited && + workspaceId === "childworkspace" && + metadata.taskHandleId === "wst_handle" && + metadata.ownerWorkspaceId === ownerWorkspaceId && + metadata.turnId === "turn" + ); const waitForPendingStreamErrorRecoveryDecision = mock((): Promise => { retryDecisionAwaited = true; return Promise.resolve(); }); const { parentId, taskService } = await startWorkspaceTurnForTest({ ...(scenario.pending === "queued" - ? { hasPendingQueuedOrPreparingTurn: pending } + ? { + hasPendingQueuedOrPreparingTurn: pending, + hasPendingWorkspaceTurnContinuation, + } : { hasPendingAutoRetry: pending }), waitForPendingStreamErrorRecoveryDecision, }); + ownerWorkspaceId = parentId; await taskService.finalizeWorkspaceTurnFromStreamError(scenario.event); diff --git a/src/node/services/workspaceTurnManager.ts b/src/node/services/workspaceTurnManager.ts index caf594fcbf..c85d398043 100644 --- a/src/node/services/workspaceTurnManager.ts +++ b/src/node/services/workspaceTurnManager.ts @@ -3800,25 +3800,59 @@ export class WorkspaceTurnManager { return (await this.listActiveWorkspaceTurnTaskIdsForOwner(record.workspaceId)).length > 0; } - private async isLiveWorkspaceTurn(record: WorkspaceTurnTaskHandleRecord): Promise { - const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId); + private getWorkspaceTurnRuntimeActivity(record: WorkspaceTurnTaskHandleRecord): { + hasAnyActivity: boolean; + hasCorrelatedActivity: boolean; + } { const activeStreamCorrelation = parseWorkspaceTurnTaskCorrelation( this.streamManager?.getStreamInfo(record.workspaceId)?.muxMetadata ); - const hasRuntimeActivity = - (activeStreamCorrelation?.taskHandleId === record.handleId && - activeStreamCorrelation.ownerWorkspaceId === record.ownerWorkspaceId && - activeStreamCorrelation.turnId === record.turnId) || + const hasActiveStream = + this.aiService.isStreaming(record.workspaceId) || activeStreamCorrelation != null; + const hasPendingQueuedOrPreparingTurn = this.workspaceService.hasPendingQueuedOrPreparingTurn( + record.workspaceId + ); + const hasCorrelatedStream = + hasActiveStream && + activeStreamCorrelation?.taskHandleId === record.handleId && + activeStreamCorrelation.ownerWorkspaceId === record.ownerWorkspaceId && + activeStreamCorrelation.turnId === record.turnId; + const hasCorrelatedQueuedOrPreparingTurn = this.workspaceService.hasPendingWorkspaceTurnContinuation( record.workspaceId, this.buildWorkspaceTurnMuxMetadata(record) - ) || - this.workspaceService.hasPendingAutoRetry(record.workspaceId) || - this.workspaceService.hasPendingBashMonitorWakeContinuation(record.workspaceId); - if (hasRuntimeActivity) { + ); + const hasPendingAutoRetry = this.workspaceService.hasPendingAutoRetry(record.workspaceId); + const hasPendingBashMonitorWake = this.workspaceService.hasPendingBashMonitorWakeContinuation( + record.workspaceId + ); + + return { + hasAnyActivity: + hasActiveStream || + hasPendingQueuedOrPreparingTurn || + hasCorrelatedQueuedOrPreparingTurn || + hasPendingAutoRetry || + hasPendingBashMonitorWake, + hasCorrelatedActivity: + hasCorrelatedStream || + hasCorrelatedQueuedOrPreparingTurn || + hasPendingAutoRetry || + hasPendingBashMonitorWake, + }; + } + + private async isLiveWorkspaceTurn(record: WorkspaceTurnTaskHandleRecord): Promise { + const runtimeActivity = this.getWorkspaceTurnRuntimeActivity(record); + if (runtimeActivity.hasCorrelatedActivity) { return true; } + // Unrelated target activity cannot prove that this persisted handle still owns the workspace. + if (runtimeActivity.hasAnyActivity) { + return false; + } + const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId); const isActiveHandle = active?.handleId === record.handleId && active.ownerWorkspaceId === record.ownerWorkspaceId; if (!isActiveHandle) { @@ -3853,11 +3887,15 @@ export class WorkspaceTurnManager { } // Same-process deferred stream-ends can be observed before the final assistant message is - // readable from history. Keep the handle alive in that narrow window; after restart the active - // map is empty, so unrecoverable deferred handles still settle terminally instead of leaking. + // readable from history. Keep the handle alive only while it still owns the runtime activity. + const runtimeActivity = this.getWorkspaceTurnRuntimeActivity(record); + if (runtimeActivity.hasCorrelatedActivity) { + return; + } const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId); if ( (record.deferredMessageIds?.length ?? 0) > 0 && + !runtimeActivity.hasAnyActivity && active?.handleId === record.handleId && active.ownerWorkspaceId === record.ownerWorkspaceId ) { From 3f22e1209f277ba1657ec204bd422c3a72098797 Mon Sep 17 00:00:00 2001 From: Mux Date: Tue, 1 Sep 2026 16:16:16 -0500 Subject: [PATCH 4/5] =?UTF-8?q?=F0=9F=A4=96=20fix:=20preserve=20correlated?= =?UTF-8?q?=20workspace=20activity?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../services/workspaceTurnManager.test.ts | 85 ++++++++++++++++--- src/node/services/workspaceTurnManager.ts | 66 +++++++++++--- 2 files changed, 128 insertions(+), 23 deletions(-) diff --git a/src/node/services/workspaceTurnManager.test.ts b/src/node/services/workspaceTurnManager.test.ts index d8b0c5f3ce..7f0e95f4e5 100644 --- a/src/node/services/workspaceTurnManager.test.ts +++ b/src/node/services/workspaceTurnManager.test.ts @@ -334,6 +334,7 @@ describe("WorkspaceTurnManager", () => { hasPendingQueuedOrPreparingTurn?: ReturnType; hasPendingBashMonitorWakeContinuation?: ReturnType; hasPendingWorkspaceTurnContinuation?: ReturnType; + hasQueuedWorkspaceTurn?: ReturnType; getQueueCutCutter?: ReturnType; hasPendingAutoRetry?: ReturnType; waitForPendingStreamErrorRecoveryDecision?: ReturnType; @@ -3652,6 +3653,7 @@ describe("WorkspaceTurnManager", () => { createdAt, updatedAt: createdAt, createdWorkspace: true, + disposableWorkspace: true, deferredMessageIds: ["assistant-deferred"], }) ); @@ -3671,7 +3673,9 @@ describe("WorkspaceTurnManager", () => { expect(await workspaceTurnSnapshot(taskService, parentId, "wst_other")).toMatchObject({ status: "interrupted", error: "Workspace turn interrupted after restart", + disposableWorkspace: false, }); + expect(workspaceMocks.remove).not.toHaveBeenCalled(); }); test("createWorkspaceTurn counts active workspace turns across all owners", async () => { @@ -3814,7 +3818,13 @@ describe("WorkspaceTurnManager", () => { test("active workspace turn count keeps correlated continuation activity live", async () => { let ownerWorkspaceId = ""; - let activity: "stream" | "queued" | "auto-retry" | "monitor-wake" = "stream"; + let activity: + | "stream" + | "compaction" + | "queued" + | "queued-behind" + | "auto-retry" + | "monitor-wake" = "stream"; const hasPendingWorkspaceTurnContinuation = mock( (workspaceId: string, metadata: ReturnType) => activity === "queued" && @@ -3823,18 +3833,51 @@ describe("WorkspaceTurnManager", () => { metadata.ownerWorkspaceId === ownerWorkspaceId && metadata.turnId === "turn" ); + const hasQueuedWorkspaceTurn = mock( + (workspaceId: string, handleId: string) => + activity === "queued-behind" && + workspaceId === "childworkspace" && + handleId === "wst_handle" + ); const streamManager = { - getStreamInfo: mock((workspaceId: string) => - activity === "stream" && workspaceId === "childworkspace" - ? activeStreamInfo( - workspaceTurnMuxMetadata(ownerWorkspaceId, "wst_handle", "turn"), - "correlated-message" - ) - : undefined - ), + getStreamInfo: mock((workspaceId: string) => { + if (workspaceId !== "childworkspace") { + return undefined; + } + if (activity === "stream") { + return activeStreamInfo( + workspaceTurnMuxMetadata(ownerWorkspaceId, "wst_handle", "turn"), + "correlated-message" + ); + } + if (activity === "compaction") { + return activeStreamInfo( + { + type: "compaction-request", + rawCommand: "/compact", + source: "auto-compaction", + parsed: { + followUpContent: { + text: "Continue", + model: "anthropic:claude-opus-4-6", + agentId: "exec", + workspaceTurnMetadata: workspaceTurnMuxMetadata( + ownerWorkspaceId, + "wst_handle", + "turn" + ), + }, + }, + }, + "compaction-message" + ); + } + return undefined; + }), } as unknown as StreamManager; const { parentId, taskService } = await startWorkspaceTurnForTest({ hasPendingWorkspaceTurnContinuation, + hasQueuedWorkspaceTurn, hasPendingAutoRetry: mock(() => activity === "auto-retry"), hasPendingBashMonitorWakeContinuation: mock(() => activity === "monitor-wake"), streamManager, @@ -3849,7 +3892,14 @@ describe("WorkspaceTurnManager", () => { }; internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); - for (const nextActivity of ["stream", "queued", "auto-retry", "monitor-wake"] as const) { + for (const nextActivity of [ + "stream", + "compaction", + "queued", + "queued-behind", + "auto-retry", + "monitor-wake", + ] as const) { activity = nextActivity; expect(await internal.countActiveWorkspaceTurns()).toBe(1); } @@ -3857,12 +3907,27 @@ describe("WorkspaceTurnManager", () => { "childworkspace", workspaceTurnMuxMetadata(parentId, "wst_handle", "turn") ); + expect(hasQueuedWorkspaceTurn).toHaveBeenCalledWith("childworkspace", "wst_handle"); const snapshot = await workspaceTurnSnapshot(taskService, parentId); expect(snapshot).toMatchObject({ status: "running", workspaceId: "childworkspace" }); expect(snapshot?.error).toBeUndefined(); }); + test("active workspace turn count preserves mock streams without StreamInfo", async () => { + const isStreaming = mock((workspaceId: string) => workspaceId === "childworkspace"); + const { parentId, taskService } = await startWorkspaceTurnForTest({ isStreaming }); + const internal = taskService as unknown as { + countActiveWorkspaceTurns: () => Promise; + }; + + expect(await internal.countActiveWorkspaceTurns()).toBe(1); + expect(await workspaceTurnSnapshot(taskService, parentId)).toMatchObject({ + status: "running", + workspaceId: "childworkspace", + }); + }); + test("getWorkspaceTurnSnapshot settles stale active handles before returning", async () => { const { parentId, taskService } = await startWorkspaceTurnForTest(); const internal = taskService as unknown as { diff --git a/src/node/services/workspaceTurnManager.ts b/src/node/services/workspaceTurnManager.ts index c85d398043..71fb75baad 100644 --- a/src/node/services/workspaceTurnManager.ts +++ b/src/node/services/workspaceTurnManager.ts @@ -3800,15 +3800,41 @@ export class WorkspaceTurnManager { return (await this.listActiveWorkspaceTurnTaskIdsForOwner(record.workspaceId)).length > 0; } + private getRuntimeWorkspaceTurnMetadataFromValue( + value: unknown + ): { taskHandleId: string; ownerWorkspaceId: string; turnId: string } | undefined { + const direct = parseWorkspaceTurnTaskCorrelation(value); + if (direct != null) { + return direct; + } + if (value == null || typeof value !== "object") { + return undefined; + } + const compaction = value as { + type?: unknown; + parsed?: { followUpContent?: { workspaceTurnMetadata?: unknown } }; + }; + if (compaction.type !== "compaction-request") { + return undefined; + } + return ( + parseWorkspaceTurnTaskCorrelation( + compaction.parsed?.followUpContent?.workspaceTurnMetadata + ) ?? undefined + ); + } + private getWorkspaceTurnRuntimeActivity(record: WorkspaceTurnTaskHandleRecord): { hasAnyActivity: boolean; hasCorrelatedActivity: boolean; + hasUncorrelatedActivity: boolean; } { - const activeStreamCorrelation = parseWorkspaceTurnTaskCorrelation( - this.streamManager?.getStreamInfo(record.workspaceId)?.muxMetadata + const activeStreamInfo = this.streamManager?.getStreamInfo(record.workspaceId); + const activeStreamCorrelation = this.getRuntimeWorkspaceTurnMetadataFromValue( + activeStreamInfo?.muxMetadata ); const hasActiveStream = - this.aiService.isStreaming(record.workspaceId) || activeStreamCorrelation != null; + this.aiService.isStreaming(record.workspaceId) || activeStreamInfo != null; const hasPendingQueuedOrPreparingTurn = this.workspaceService.hasPendingQueuedOrPreparingTurn( record.workspaceId ); @@ -3821,7 +3847,7 @@ export class WorkspaceTurnManager { this.workspaceService.hasPendingWorkspaceTurnContinuation( record.workspaceId, this.buildWorkspaceTurnMuxMetadata(record) - ); + ) || this.workspaceService.hasQueuedWorkspaceTurn(record.workspaceId, record.handleId); const hasPendingAutoRetry = this.workspaceService.hasPendingAutoRetry(record.workspaceId); const hasPendingBashMonitorWake = this.workspaceService.hasPendingBashMonitorWakeContinuation( record.workspaceId @@ -3839,6 +3865,11 @@ export class WorkspaceTurnManager { hasCorrelatedQueuedOrPreparingTurn || hasPendingAutoRetry || hasPendingBashMonitorWake, + // A missing StreamInfo is ambiguous because MockAiStreamPlayer reports only through + // AIService.isStreaming. Preserve the active-map fallback for that test/runtime path. + hasUncorrelatedActivity: + (hasActiveStream && activeStreamInfo != null && !hasCorrelatedStream) || + (hasPendingQueuedOrPreparingTurn && !hasCorrelatedQueuedOrPreparingTurn), }; } @@ -3847,8 +3878,9 @@ export class WorkspaceTurnManager { if (runtimeActivity.hasCorrelatedActivity) { return true; } - // Unrelated target activity cannot prove that this persisted handle still owns the workspace. - if (runtimeActivity.hasAnyActivity) { + // Only positive evidence of unrelated activity can invalidate the active-map fallback. + // Mock streams can report busy without exposing StreamInfo correlation. + if (runtimeActivity.hasUncorrelatedActivity) { return false; } @@ -3873,29 +3905,35 @@ export class WorkspaceTurnManager { if (!isActiveWorkspaceTurnTaskStatus(record.status)) { return; } + const runtimeActivity = this.getWorkspaceTurnRuntimeActivity(record); + const disposableOwnershipTransferred = + record.disposableWorkspace && runtimeActivity.hasUncorrelatedActivity; const recovered = await this.recoverTerminalWorkspaceTurnFromHistory(record); if (recovered != null) { + const next = disposableOwnershipTransferred + ? { ...recovered, disposableWorkspace: false } + : recovered; await this.settleWorkspaceTurn({ record, - next: recovered, + next, waiterSettlement: - recovered.status === "completed" - ? { status: "completed", result: this.buildWorkspaceTurnWaitResult(recovered) } - : { status: "error", error: new Error(recovered.error ?? "Workspace turn failed") }, + next.status === "completed" + ? { status: "completed", result: this.buildWorkspaceTurnWaitResult(next) } + : { status: "error", error: new Error(next.error ?? "Workspace turn failed") }, + disposableOwnershipTransferred, }); return; } // Same-process deferred stream-ends can be observed before the final assistant message is - // readable from history. Keep the handle alive only while it still owns the runtime activity. - const runtimeActivity = this.getWorkspaceTurnRuntimeActivity(record); + // readable from history. Keep the handle alive unless unrelated activity owns the runtime. if (runtimeActivity.hasCorrelatedActivity) { return; } const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId); if ( (record.deferredMessageIds?.length ?? 0) > 0 && - !runtimeActivity.hasAnyActivity && + !runtimeActivity.hasUncorrelatedActivity && active?.handleId === record.handleId && active.ownerWorkspaceId === record.ownerWorkspaceId ) { @@ -3907,6 +3945,7 @@ export class WorkspaceTurnManager { status: "interrupted", updatedAt: getIsoNow(), error: WORKSPACE_TURN_STALE_RESTART_ERROR, + ...(disposableOwnershipTransferred ? { disposableWorkspace: false } : {}), }; await this.settleWorkspaceTurn({ record, @@ -3915,6 +3954,7 @@ export class WorkspaceTurnManager { status: "error", error: new Error(WORKSPACE_TURN_STALE_RESTART_ERROR), }, + disposableOwnershipTransferred, }); } From 3f190cc89743d25892d00581892a48586a9f8848 Mon Sep 17 00:00:00 2001 From: Mux Date: Tue, 1 Sep 2026 17:16:41 -0500 Subject: [PATCH 5/5] =?UTF-8?q?=F0=9F=A4=96=20fix:=20close=20workspace-tur?= =?UTF-8?q?n=20liveness=20races?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Correlate compaction, retry, wake, queue, and send-preflight activity. Carry workspace ownership evidence through stale settlement. --- _Generated with `xum` • Model: `openai:gpt-5.6-sol` • Thinking: `xhigh` • Cost: `$0.19`_ --- .../services/workspaceTurnManager.test.ts | 186 ++++++++++++++---- src/node/services/workspaceTurnManager.ts | 142 +++++++++---- 2 files changed, 257 insertions(+), 71 deletions(-) diff --git a/src/node/services/workspaceTurnManager.test.ts b/src/node/services/workspaceTurnManager.test.ts index 7f0e95f4e5..87e5be7514 100644 --- a/src/node/services/workspaceTurnManager.test.ts +++ b/src/node/services/workspaceTurnManager.test.ts @@ -455,10 +455,14 @@ describe("WorkspaceTurnManager", () => { taskService as unknown as { activeWorkspaceTurnHandleByWorkspaceId: Map< string, - { handleId: string; ownerWorkspaceId: string } + { handleId: string; ownerWorkspaceId: string; accepted: boolean } >; } - ).activeWorkspaceTurnHandleByWorkspaceId.set(workspaceId, { handleId, ownerWorkspaceId }); + ).activeWorkspaceTurnHandleByWorkspaceId.set(workspaceId, { + handleId, + ownerWorkspaceId, + accepted: true, + }); } test("workspace lifecycle archives only parent-owned created workspace turns", async () => { @@ -3582,7 +3586,97 @@ describe("WorkspaceTurnManager", () => { expect(aiMocks.stopStream).not.toHaveBeenCalled(); }); - test("createWorkspaceTurn repairs stale handles before parallel-slot accounting", async () => { + test("createWorkspaceTurn keeps queued handles live during send preflight", async () => { + const config = await createTestConfig(rootDir); + stubStableIds(config, ["firsthandle", "firstturn", "secondhandle", "secondturn"]); + const { parentId, projectPath } = await saveLocalParentWorkspace(config, rootDir); + const createWorkspace = makeWorkspaceTurnCreateMock(config, projectPath); + const secondSendStarted = Promise.withResolvers(); + const releaseSecondSend = Promise.withResolvers(); + let sendCallCount = 0; + let queueContainsSecondHandle = false; + const sendMessage = mock( + async (..._args: unknown[]): Promise> => { + sendCallCount += 1; + if (sendCallCount === 2) { + secondSendStarted.resolve(); + await releaseSecondSend.promise; + queueContainsSecondHandle = true; + } + return Ok(undefined); + } + ); + const busyWorkspaceIds = new Set(); + const workspaceMocks = createWorkspaceServiceMocks({ + create: createWorkspace, + sendMessage, + isBusyForMessage: mock((workspaceId: string) => busyWorkspaceIds.has(workspaceId)), + hasQueuedMessages: mock((workspaceId: string) => busyWorkspaceIds.has(workspaceId)), + hasQueuedWorkspaceTurn: mock( + (workspaceId: string, handleId: string) => + queueContainsSecondHandle && + workspaceId === "childworkspace" && + handleId === "wst_secondhandle" + ), + }); + const aiMocks = createAIServiceMocks(config, { + isStreaming: mock((workspaceId: string) => busyWorkspaceIds.has(workspaceId)), + }); + const streamManager = { + getStreamInfo: mock((workspaceId: string) => + busyWorkspaceIds.has(workspaceId) + ? activeStreamInfo( + workspaceTurnMuxMetadata(parentId, "wst_firsthandle", "firstturn"), + "first-message" + ) + : undefined + ), + } as unknown as StreamManager; + const { taskService } = createWorkspaceTurnManagerHarness(config, { + aiService: aiMocks.aiService, + workspaceService: workspaceMocks.workspaceService, + streamManager, + }); + + const first = await taskService.createWorkspaceTurn({ + ownerWorkspaceId: parentId, + prompt: "First prompt", + title: "Workspace turn", + workspace: { mode: "new" }, + }); + expect(first.success).toBe(true); + busyWorkspaceIds.add("childworkspace"); + + const secondPromise = taskService.createWorkspaceTurn({ + ownerWorkspaceId: parentId, + prompt: "Queued prompt", + title: "Follow-up", + workspace: { mode: "existing", workspaceId: "childworkspace" }, + }); + await secondSendStarted.promise; + + let snapshot: Awaited>; + try { + snapshot = await workspaceTurnSnapshot(taskService, parentId, "wst_secondhandle"); + } finally { + releaseSecondSend.resolve(); + } + const second = await secondPromise; + + expect(second.success).toBe(true); + expect(snapshot).toMatchObject({ + handleId: "wst_secondhandle", + status: "queued", + workspaceId: "childworkspace", + }); + expect(await workspaceTurnSnapshot(taskService, parentId, "wst_secondhandle")).toMatchObject({ + handleId: "wst_secondhandle", + status: "queued", + workspaceId: "childworkspace", + }); + }); + + test("createWorkspaceTurn preserves ownership evidence while releasing stale capacity", async () => { const config = await createTestConfig(rootDir); stubStableIds(config, ["queuedhandle", "queuedturn"]); const { parentId, projectPath } = await saveLocalParentWorkspace(config, rootDir); @@ -3613,25 +3707,21 @@ describe("WorkspaceTurnManager", () => { const workspaceMocks = createWorkspaceServiceMocks({ sendMessage, isBusyForMessage: mock((workspaceId: string) => workspaceId === "childworkspace"), - hasPendingQueuedOrPreparingTurn: mock( - (workspaceId: string) => workspaceId === "otherworkspace" - ), - }); - const aiMocks = createAIServiceMocks(config, { - isStreaming: mock((workspaceId: string) => workspaceId === "otherworkspace"), }); + let unrelatedActivityObserved = false; const streamManager = { - getStreamInfo: mock((workspaceId: string) => - workspaceId === "otherworkspace" - ? activeStreamInfo( - workspaceTurnMuxMetadata(parentId, "wst_unrelated", "unrelatedturn"), - "unrelated-message" - ) - : undefined - ), + getStreamInfo: mock((workspaceId: string) => { + if (workspaceId !== "otherworkspace" || unrelatedActivityObserved) { + return undefined; + } + unrelatedActivityObserved = true; + return activeStreamInfo( + workspaceTurnMuxMetadata(parentId, "wst_unrelated", "unrelatedturn"), + "unrelated-message" + ); + }), } as unknown as StreamManager; const { taskService } = createWorkspaceTurnManagerHarness(config, { - aiService: aiMocks.aiService, workspaceService: workspaceMocks.workspaceService, streamManager, }); @@ -3646,7 +3736,8 @@ describe("WorkspaceTurnManager", () => { createdWorkspace: true, }) ); - // The target has unrelated activity. It must not reserve a slot for this stale handle. + // The unrelated activity disappears before settlement. The first observation still + // transfers disposable ownership and releases this stale handle's task slot. await taskHandleStore.upsertWorkspaceTurn( workspaceTurnRecord(parentId, "otherworkspace", "wst_other", "running", { turnId: "otherturn", @@ -3820,7 +3911,8 @@ describe("WorkspaceTurnManager", () => { let ownerWorkspaceId = ""; let activity: | "stream" - | "compaction" + | "compaction-direct" + | "compaction-inherited" | "queued" | "queued-behind" | "auto-retry" @@ -3844,13 +3936,11 @@ describe("WorkspaceTurnManager", () => { if (workspaceId !== "childworkspace") { return undefined; } + const correlation = workspaceTurnMuxMetadata(ownerWorkspaceId, "wst_handle", "turn"); if (activity === "stream") { - return activeStreamInfo( - workspaceTurnMuxMetadata(ownerWorkspaceId, "wst_handle", "turn"), - "correlated-message" - ); + return activeStreamInfo(correlation, "correlated-message"); } - if (activity === "compaction") { + if (activity === "compaction-direct" || activity === "compaction-inherited") { return activeStreamInfo( { type: "compaction-request", @@ -3861,11 +3951,9 @@ describe("WorkspaceTurnManager", () => { text: "Continue", model: "anthropic:claude-opus-4-6", agentId: "exec", - workspaceTurnMetadata: workspaceTurnMuxMetadata( - ownerWorkspaceId, - "wst_handle", - "turn" - ), + ...(activity === "compaction-direct" + ? { muxMetadata: correlation } + : { workspaceTurnMetadata: correlation }), }, }, }, @@ -3886,21 +3974,25 @@ describe("WorkspaceTurnManager", () => { const internal = taskService as unknown as { activeWorkspaceTurnHandleByWorkspaceId: Map< string, - { handleId: string; ownerWorkspaceId: string } + { handleId: string; ownerWorkspaceId: string; accepted: boolean } >; countActiveWorkspaceTurns: () => Promise; }; - internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); for (const nextActivity of [ "stream", - "compaction", + "compaction-direct", + "compaction-inherited", "queued", "queued-behind", "auto-retry", "monitor-wake", ] as const) { activity = nextActivity; + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + if (nextActivity === "auto-retry" || nextActivity === "monitor-wake") { + markWorkspaceTurnActive(taskService, "childworkspace", "wst_handle", parentId); + } expect(await internal.countActiveWorkspaceTurns()).toBe(1); } expect(hasPendingWorkspaceTurnContinuation).toHaveBeenCalledWith( @@ -3914,6 +4006,34 @@ describe("WorkspaceTurnManager", () => { expect(snapshot?.error).toBeUndefined(); }); + for (const continuation of ["auto-retry", "monitor-wake"] as const) { + test( + "active workspace turn count rejects unrelated " + continuation + " activity", + async () => { + const { parentId, taskService, created } = await startWorkspaceTurnForTest({ + hasPendingAutoRetry: mock( + (workspaceId: string) => + continuation === "auto-retry" && workspaceId === "childworkspace" + ), + hasPendingBashMonitorWakeContinuation: mock( + (workspaceId: string) => + continuation === "monitor-wake" && workspaceId === "childworkspace" + ), + }); + markWorkspaceTurnActive(taskService, created.workspaceId, "wst_other", "other-owner"); + const internal = taskService as unknown as { + countActiveWorkspaceTurns: () => Promise; + }; + + expect(await internal.countActiveWorkspaceTurns()).toBe(0); + expect(await workspaceTurnSnapshot(taskService, parentId)).toMatchObject({ + status: "interrupted", + error: "Workspace turn interrupted after restart", + }); + } + ); + } + test("active workspace turn count preserves mock streams without StreamInfo", async () => { const isStreaming = mock((workspaceId: string) => workspaceId === "childworkspace"); const { parentId, taskService } = await startWorkspaceTurnForTest({ isStreaming }); diff --git a/src/node/services/workspaceTurnManager.ts b/src/node/services/workspaceTurnManager.ts index 71fb75baad..9dd2c76ae7 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, @@ -296,6 +297,17 @@ const WORKSPACE_TURN_SUPERSEDED_BY_NEW_INPUT_ERROR = const WORKSPACE_TURN_SUPERSEDED_BY_OWNER_FOLLOW_UP_ERROR_PREFIX = "Workspace turn superseded by follow-up turn "; +interface WorkspaceTurnRuntimeActivity { + hasAnyActivity: boolean; + hasCorrelatedActivity: boolean; + hasUncorrelatedActivity: boolean; +} + +interface WorkspaceTurnLiveness { + isLive: boolean; + runtimeActivity: WorkspaceTurnRuntimeActivity; +} + function buildOwnerFollowUpSupersededError(successorHandleId: string): string { return ( `${WORKSPACE_TURN_SUPERSEDED_BY_OWNER_FOLLOW_UP_ERROR_PREFIX}${successorHandleId} from the ` + @@ -442,6 +454,7 @@ export class WorkspaceTurnManager { string, { handleId: string; ownerWorkspaceId: string; accepted: boolean } >(); + private readonly workspaceTurnCreationReservationsByWorkspaceId = new Map>(); private lastWorkspaceTurnCreatedAtMs = 0; private readonly taskHandleStore: TaskHandleStore; @@ -473,6 +486,22 @@ export class WorkspaceTurnManager { return this.activeWorkspaceTurnHandleByWorkspaceId.get(workspaceId); } + private reserveWorkspaceTurnCreation(workspaceId: string, handleId: string) { + const reservations = + this.workspaceTurnCreationReservationsByWorkspaceId.get(workspaceId) ?? new Set(); + reservations.add(handleId); + this.workspaceTurnCreationReservationsByWorkspaceId.set(workspaceId, reservations); + + return { + [Symbol.dispose]: () => { + reservations.delete(handleId); + if (reservations.size === 0) { + this.workspaceTurnCreationReservationsByWorkspaceId.delete(workspaceId); + } + }, + }; + } + async markWorkspaceTurnBackgroundWorkNotifyOnTerminal( taskId: string, ownerWorkspaceId: string @@ -1298,6 +1327,10 @@ export class WorkspaceTurnManager { // mutex → lifecycle edge of the global lock order (task-tree → this.mutex → // workspaceLifecycleLocks; see the workspaceLifecycleLocks declaration), with sorted keys // preventing lifecycle-key cycles between concurrent owner/target pairs. + // Keep the persisted handle live while sendMessage completes pricing, settings, and queue + // admission. The queue does not expose correlation until that preflight finishes. + using _creationReservation = this.reserveWorkspaceTurnCreation(targetWorkspaceId, handleId); + const isArchivedInConfig = (workspaceId: string): boolean => { const entry = findWorkspaceEntry(this.config.loadConfigOrDefault(), workspaceId); return ( @@ -2573,12 +2606,15 @@ export class WorkspaceTurnManager { } } - if ( - isActiveWorkspaceTurnTaskStatus(record.status) && - !(await this.isLiveWorkspaceTurn(record)) - ) { - await this.settleStaleWorkspaceTurn(record); - return await this.taskHandleStore.getWorkspaceTurn(record.ownerWorkspaceId, record.handleId); + if (isActiveWorkspaceTurnTaskStatus(record.status)) { + const liveness = await this.getWorkspaceTurnLiveness(record); + if (!liveness.isLive) { + await this.settleStaleWorkspaceTurn(record, liveness.runtimeActivity); + return await this.taskHandleStore.getWorkspaceTurn( + record.ownerWorkspaceId, + record.handleId + ); + } } if ( @@ -3761,8 +3797,9 @@ export class WorkspaceTurnManager { if (record.workspaceId !== workspaceId || !this.isActiveWorkspaceTurn(record)) { continue; } - if (!(await this.isLiveWorkspaceTurn(record))) { - await this.settleStaleWorkspaceTurn(record); + const liveness = await this.getWorkspaceTurnLiveness(record); + if (!liveness.isLive) { + await this.settleStaleWorkspaceTurn(record, liveness.runtimeActivity); continue; } return record; @@ -3810,25 +3847,21 @@ export class WorkspaceTurnManager { if (value == null || typeof value !== "object") { return undefined; } - const compaction = value as { - type?: unknown; - parsed?: { followUpContent?: { workspaceTurnMetadata?: unknown } }; - }; + const compaction = value as MuxMessageMetadata; if (compaction.type !== "compaction-request") { return undefined; } + const followUpContent = getCompactionFollowUpContent(compaction); return ( - parseWorkspaceTurnTaskCorrelation( - compaction.parsed?.followUpContent?.workspaceTurnMetadata - ) ?? undefined + parseWorkspaceTurnTaskCorrelation(followUpContent?.muxMetadata) ?? + parseWorkspaceTurnTaskCorrelation(followUpContent?.workspaceTurnMetadata) ?? + undefined ); } - private getWorkspaceTurnRuntimeActivity(record: WorkspaceTurnTaskHandleRecord): { - hasAnyActivity: boolean; - hasCorrelatedActivity: boolean; - hasUncorrelatedActivity: boolean; - } { + private getWorkspaceTurnRuntimeActivity( + record: WorkspaceTurnTaskHandleRecord + ): WorkspaceTurnRuntimeActivity { const activeStreamInfo = this.streamManager?.getStreamInfo(record.workspaceId); const activeStreamCorrelation = this.getRuntimeWorkspaceTurnMetadataFromValue( activeStreamInfo?.muxMetadata @@ -3848,66 +3881,99 @@ export class WorkspaceTurnManager { record.workspaceId, this.buildWorkspaceTurnMuxMetadata(record) ) || this.workspaceService.hasQueuedWorkspaceTurn(record.workspaceId, record.handleId); + const creationReservations = this.workspaceTurnCreationReservationsByWorkspaceId.get( + record.workspaceId + ); + const hasCorrelatedCreationReservation = creationReservations?.has(record.handleId) === true; + const hasUncorrelatedCreationReservation = + creationReservations != null && + creationReservations.size > 0 && + !hasCorrelatedCreationReservation; + const activeRegistration = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId); + const hasActiveRegistration = + activeRegistration?.handleId === record.handleId && + activeRegistration.ownerWorkspaceId === record.ownerWorkspaceId; const hasPendingAutoRetry = this.workspaceService.hasPendingAutoRetry(record.workspaceId); const hasPendingBashMonitorWake = this.workspaceService.hasPendingBashMonitorWakeContinuation( record.workspaceId ); + const hasCorrelatedRetryOrWake = + hasActiveRegistration && (hasPendingAutoRetry || hasPendingBashMonitorWake); + const hasUncorrelatedRetryOrWake = + !hasActiveRegistration && (hasPendingAutoRetry || hasPendingBashMonitorWake); return { hasAnyActivity: hasActiveStream || hasPendingQueuedOrPreparingTurn || hasCorrelatedQueuedOrPreparingTurn || + creationReservations != null || hasPendingAutoRetry || hasPendingBashMonitorWake, hasCorrelatedActivity: hasCorrelatedStream || hasCorrelatedQueuedOrPreparingTurn || - hasPendingAutoRetry || - hasPendingBashMonitorWake, + hasCorrelatedCreationReservation || + hasCorrelatedRetryOrWake, // A missing StreamInfo is ambiguous because MockAiStreamPlayer reports only through // AIService.isStreaming. Preserve the active-map fallback for that test/runtime path. hasUncorrelatedActivity: (hasActiveStream && activeStreamInfo != null && !hasCorrelatedStream) || - (hasPendingQueuedOrPreparingTurn && !hasCorrelatedQueuedOrPreparingTurn), + (hasPendingQueuedOrPreparingTurn && !hasCorrelatedQueuedOrPreparingTurn) || + hasUncorrelatedCreationReservation || + hasUncorrelatedRetryOrWake, }; } - private async isLiveWorkspaceTurn(record: WorkspaceTurnTaskHandleRecord): Promise { + private async getWorkspaceTurnLiveness( + record: WorkspaceTurnTaskHandleRecord + ): Promise { const runtimeActivity = this.getWorkspaceTurnRuntimeActivity(record); if (runtimeActivity.hasCorrelatedActivity) { - return true; + return { isLive: true, runtimeActivity }; } // Only positive evidence of unrelated activity can invalidate the active-map fallback. // Mock streams can report busy without exposing StreamInfo correlation. if (runtimeActivity.hasUncorrelatedActivity) { - return false; + return { isLive: false, runtimeActivity }; } const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId); const isActiveHandle = active?.handleId === record.handleId && active.ownerWorkspaceId === record.ownerWorkspaceId; if (!isActiveHandle) { - return false; + return { isLive: false, runtimeActivity }; } if ((record.deferredMessageIds?.length ?? 0) === 0) { - return true; + return { isLive: true, runtimeActivity }; } // A deferred workspace-turn stream-end was waiting for background work. Once there is no // live stream/queued retry and no active descendant/workflow/nested turn left, the in-memory // handle is stale and should be recovered from the deferred history instead of blocking forever. - return await this.hasActiveWorkspaceTurnDeferredBlockers(record); + return { + isLive: await this.hasActiveWorkspaceTurnDeferredBlockers(record), + runtimeActivity, + }; } - private async settleStaleWorkspaceTurn(record: WorkspaceTurnTaskHandleRecord): Promise { + private async settleStaleWorkspaceTurn( + record: WorkspaceTurnTaskHandleRecord, + observedRuntimeActivity: WorkspaceTurnRuntimeActivity + ): Promise { if (!isActiveWorkspaceTurnTaskStatus(record.status)) { return; } const runtimeActivity = this.getWorkspaceTurnRuntimeActivity(record); + if (runtimeActivity.hasCorrelatedActivity) { + return; + } + // Preserve the ownership evidence that made the caller classify this handle as stale. + // The unrelated stream or queue entry can finish before settlement acquires its lock. const disposableOwnershipTransferred = - record.disposableWorkspace && runtimeActivity.hasUncorrelatedActivity; + record.disposableWorkspace && + (observedRuntimeActivity.hasUncorrelatedActivity || runtimeActivity.hasUncorrelatedActivity); const recovered = await this.recoverTerminalWorkspaceTurnFromHistory(record); if (recovered != null) { const next = disposableOwnershipTransferred @@ -3927,12 +3993,10 @@ export class WorkspaceTurnManager { // Same-process deferred stream-ends can be observed before the final assistant message is // readable from history. Keep the handle alive unless unrelated activity owns the runtime. - if (runtimeActivity.hasCorrelatedActivity) { - return; - } const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId); if ( (record.deferredMessageIds?.length ?? 0) > 0 && + !observedRuntimeActivity.hasUncorrelatedActivity && !runtimeActivity.hasUncorrelatedActivity && active?.handleId === record.handleId && active.ownerWorkspaceId === record.ownerWorkspaceId @@ -3973,8 +4037,9 @@ export class WorkspaceTurnManager { if (!this.isActiveWorkspaceTurn(record)) { continue; } - if (!(await this.isLiveWorkspaceTurn(record))) { - await this.settleStaleWorkspaceTurn(record); + const liveness = await this.getWorkspaceTurnLiveness(record); + if (!liveness.isLive) { + await this.settleStaleWorkspaceTurn(record, liveness.runtimeActivity); continue; } if (record.status === "queued") { @@ -3999,8 +4064,9 @@ export class WorkspaceTurnManager { const taskIds: string[] = []; for (const record of records) { if (isActiveWorkspaceTurnTaskStatus(record.status)) { - if (!(await this.isLiveWorkspaceTurn(record))) { - await this.settleStaleWorkspaceTurn(record); + const liveness = await this.getWorkspaceTurnLiveness(record); + if (!liveness.isLive) { + await this.settleStaleWorkspaceTurn(record, liveness.runtimeActivity); continue; } taskIds.push(record.handleId);