From 0e3c0ed1290c6a994e112d2c7ffb0dbc4b4d3fcf Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Tue, 27 Oct 2026 10:35:56 +0000 Subject: [PATCH] =?UTF-8?q?=F0=9F=A4=96=20refactor:=20move=20the=20stop=20?= =?UTF-8?q?latch=20into=20TaskStopLatch?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move the stop epochs, the stop-in-progress latches, the queue rechecks a held latch owes and the stop records (Phase A capture, Phase B cleanup, Phase C release) from TaskService into a plain TaskStopLatch class in taskStopLatch.ts. TaskService builds it with call-time closures; attempt, send and stream-end state stays on TaskService and is reached through them. Seam methods (bump/getWorkspaceStopEpoch, latchWorkspaceStopsInProgress, isWorkspaceStopInProgress, releaseRetainedStopLatches) and the spied runWorkspaceStopCleanup and waitForWorkspaceStopsToRelease stay on TaskService as one-line facades. recordWorkspaceTurnSettled/Superseded and workspaceUserStopEpochs stay on TaskService. Tests that cast a moved field change only the cast target. No behavior change: moved code is verbatim, statement order is kept and no await is added. Signed-off-by: Thomas Kosiewski --- .../taskService.attemptAdmission.test.ts | 84 +-- .../services/taskService.reawakening.test.ts | 32 +- .../taskService.reportDecisionHold.test.ts | 4 +- .../taskService.settlementReceipts.test.ts | 8 +- src/node/services/taskService.ts | 558 +++-------------- src/node/services/taskStopLatch.test.ts | 499 +++++++++++++++ src/node/services/taskStopLatch.ts | 567 ++++++++++++++++++ .../workspaceService.removalCascade.test.ts | 5 +- 8 files changed, 1236 insertions(+), 521 deletions(-) create mode 100644 src/node/services/taskStopLatch.test.ts create mode 100644 src/node/services/taskStopLatch.ts diff --git a/src/node/services/taskService.attemptAdmission.test.ts b/src/node/services/taskService.attemptAdmission.test.ts index e052c162671..fd872203f2b 100644 --- a/src/node/services/taskService.attemptAdmission.test.ts +++ b/src/node/services/taskService.attemptAdmission.test.ts @@ -90,18 +90,20 @@ interface Internals { string, Set<{ state: string; turnId?: symbol; attemptId: string; attempt?: unknown }> >; - workspaceStopRecords: Map< - string, - { - attemptId: string | undefined; - ownedAttempt: unknown; - capturedTurns: Set; - pendingAdmissions: Set; - cleanupInFlight: number; - capturedExecutionId: string | undefined; - executionSettled: boolean; - } - >; + taskStopLatch: { + workspaceStopRecords: Map< + string, + { + attemptId: string | undefined; + ownedAttempt: unknown; + capturedTurns: Set; + pendingAdmissions: Set; + cleanupInFlight: number; + capturedExecutionId: string | undefined; + executionSettled: boolean; + } + >; + }; currentAttemptIdByTaskId: Map; resumeFailureSettlementByTaskId: Map; markTaskLaunchFailed: (taskId: string, message: string) => Promise; @@ -1316,8 +1318,10 @@ describe("TaskService attempt identity and send admission (G1)", () => { if (race === "pending-before-decision") { // The pending obligation counted as live: a stop record captured it and cannot release // (nor settle) until it is dispositioned. - await waitForCondition(() => svc.workspaceStopRecords.has(taskId)); - expect(svc.workspaceStopRecords.get(taskId)!.pendingAdmissions.size).toBe(1); + await waitForCondition(() => svc.taskStopLatch.workspaceStopRecords.has(taskId)); + expect(svc.taskStopLatch.workspaceStopRecords.get(taskId)!.pendingAdmissions.size).toBe( + 1 + ); expect(pending!.admissionStale()).toBe(true); pending!.onDisposed("refused"); } @@ -1474,7 +1478,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { // ...so a Stop has nothing to wait for and releases its latch. await raceWithTimeout(taskService.terminateAllDescendantAgentTasks(rootId), 2_000); expect(taskService.isWorkspaceStopInProgress(taskId)).toBe(false); - expect(svc.workspaceStopRecords.has(taskId)).toBe(false); + expect(svc.taskStopLatch.workspaceStopRecords.has(taskId)).toBe(false); }); describe("a reawaken whose manual send/resume fails settles its owned attempt (#4310)", () => { @@ -1893,7 +1897,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { const stop = taskService.terminateAllDescendantAgentTasks(rootId); await settle(); await settle(); - expect(svc.workspaceStopRecords.has(taskId)).toBe(false); + expect(svc.taskStopLatch.workspaceStopRecords.has(taskId)).toBe(false); expect(taskService.isWorkspaceStopInProgress(taskId)).toBe(false); expect(svc.ownedAttemptByTaskId.has(taskId)).toBe(false); // "during-write": nothing is committed yet; "after-durable": the fresh id is already the @@ -1909,8 +1913,8 @@ describe("TaskService attempt identity and send admission (G1)", () => { expect(fresh).not.toBe(previous); // Ownership was installed before the mutex was released; only now can Phase A capture, // and it captures the fresh attempt OWNED. - await waitForCondition(() => svc.workspaceStopRecords.has(taskId)); - expect(svc.workspaceStopRecords.get(taskId)).toMatchObject({ + await waitForCondition(() => svc.taskStopLatch.workspaceStopRecords.has(taskId)); + expect(svc.taskStopLatch.workspaceStopRecords.get(taskId)).toMatchObject({ attemptId: fresh, ownedAttempt: svc.ownedAttemptByTaskId.get(taskId), }); @@ -1922,7 +1926,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { cleanupGate.resolve(); await stop; await waitForStopRelease(taskService, taskId); - expect(svc.workspaceStopRecords.has(taskId)).toBe(false); + expect(svc.taskStopLatch.workspaceStopRecords.has(taskId)).toBe(false); expect(entryOf(config, taskId)).toMatchObject({ taskStatus: "interrupted", taskAttemptId: fresh, @@ -1964,8 +1968,8 @@ describe("TaskService attempt identity and send admission (G1)", () => { expect(gapSend.admissionStale()).toBe(false); // The Stop captures the pending obligation and cannot release until it is dispositioned. const stop = taskService.terminateAllDescendantAgentTasks(rootId); - await waitForCondition(() => svc.workspaceStopRecords.has(taskId)); - expect(svc.workspaceStopRecords.get(taskId)!.pendingAdmissions.size).toBe(1); + await waitForCondition(() => svc.taskStopLatch.workspaceStopRecords.has(taskId)); + expect(svc.taskStopLatch.workspaceStopRecords.get(taskId)!.pendingAdmissions.size).toBe(1); expect(gapSend.admissionStale()).toBe(true); await settle(); await settle(); @@ -2009,7 +2013,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { const reawaken = taskService.markInterruptedTaskRunning(taskId); // Outside the critical section: a real Stop runs fully to completion. await taskService.terminateAllDescendantAgentTasks(rootId); - expect(svc.workspaceStopRecords.has(taskId)).toBe(false); + expect(svc.taskStopLatch.workspaceStopRecords.has(taskId)).toBe(false); gate.resolve(); expect(await reawaken).toBe(false); expect(entryOf(config, taskId)).toMatchObject({ @@ -2342,7 +2346,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { if (racing === "pending-outlives") racingToken?.onDisposed("refused"); else settleRacingTurn(spawnedId); await waitForStopRelease(taskService, spawnedId); - expect(svc.workspaceStopRecords.has(spawnedId)).toBe(false); + expect(svc.taskStopLatch.workspaceStopRecords.has(spawnedId)).toBe(false); expect(svc.attemptSettlementByTaskId.get(spawnedId)).toMatchObject({ phase: "settled", source: "stop-settled", @@ -2424,7 +2428,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { }); expect(entryOf(config, spawnedId)?.taskLaunchError).toBeUndefined(); expect(taskService.isWorkspaceStopInProgress(spawnedId)).toBe(false); - expect(svc.workspaceStopRecords.has(spawnedId)).toBe(false); + expect(svc.taskStopLatch.workspaceStopRecords.has(spawnedId)).toBe(false); // This process still owns A; idle under it, the launch failure settles exactly A. expect(svc.ownedAttemptByTaskId.get(spawnedId)?.attemptId).toBe(mintedAttemptId); expect(svc.attemptSettlementByTaskId.get(spawnedId)).toMatchObject({ @@ -2527,7 +2531,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { }); expect(entryOf(config, spawnedId)?.taskLaunchError).toBeUndefined(); expect(taskService.isWorkspaceStopInProgress(spawnedId)).toBe(false); - expect(svc.workspaceStopRecords.has(spawnedId)).toBe(false); + expect(svc.taskStopLatch.workspaceStopRecords.has(spawnedId)).toBe(false); expect(svc.ownedAttemptByTaskId.get(spawnedId)?.attemptId).toBe(mintedAttemptId); if (liveness === "idle") { expect(svc.attemptSettlementByTaskId.get(spawnedId)).toMatchObject({ @@ -2570,7 +2574,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { attemptId: mintedAttemptId, phase: "closing", }); - expect(svc.workspaceStopRecords.has(spawnedId)).toBe(false); + expect(svc.taskStopLatch.workspaceStopRecords.has(spawnedId)).toBe(false); expect(taskService.isWorkspaceStopInProgress(spawnedId)).toBe(false); expect(entryOf(config, spawnedId)).toMatchObject({ taskStatus: "running", @@ -2640,7 +2644,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { // The captured owner outlived the bound: the latch holds for A's turn, and B's row — // written after our marker — is untouched (no deletion, no second marker). expect(rollbackSpy).not.toHaveBeenCalled(); - expect(svc.workspaceStopRecords.get(spawnedId)).toMatchObject({ + expect(svc.taskStopLatch.workspaceStopRecords.get(spawnedId)).toMatchObject({ attemptId: mintedAttemptId, }); expect(entryOf(config, spawnedId)).toMatchObject({ @@ -2656,14 +2660,14 @@ describe("TaskService attempt identity and send admission (G1)", () => { // promise settles and hands its in-flight count back; the record cannot release while // that cleanup is still counted, so wait for it before checking that only A's turn holds. await waitForCondition( - () => svc.workspaceStopRecords.get(spawnedId)?.cleanupInFlight === 0 + () => svc.taskStopLatch.workspaceStopRecords.get(spawnedId)?.cleanupInFlight === 0 ); expect(taskService.isWorkspaceStopInProgress(spawnedId)).toBe(true); // Only A's turn settling releases the record and settles A; B's row stays as it was. liveTurn = undefined; host.settleTurn(spawnedId, racingTurn); expect(taskService.isWorkspaceStopInProgress(spawnedId)).toBe(false); - expect(svc.workspaceStopRecords.has(spawnedId)).toBe(false); + expect(svc.taskStopLatch.workspaceStopRecords.has(spawnedId)).toBe(false); expect(svc.attemptSettlementByTaskId.get(spawnedId)).toMatchObject({ attemptId: mintedAttemptId, phase: "settled", @@ -2884,7 +2888,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { await taskService.terminateAllDescendantAgentTasks(rootId); // Phase B ran (one stopStream) but the record is retained by the pending obligation. expect(stopStream).toHaveBeenCalledTimes(1); - const record = svc.workspaceStopRecords.get(taskId)!; + const record = svc.taskStopLatch.workspaceStopRecords.get(taskId)!; expect(record.pendingAdmissions.size).toBe(1); expect(record.capturedTurns.size).toBe(0); expect(await taskService.readAttemptOutcome(taskId, requesting)).toEqual({ @@ -2942,7 +2946,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { token.onAdmitted(first); activeTurn = first; await taskService.terminateAllDescendantAgentTasks(rootId); - const record = svc.workspaceStopRecords.get(taskId)!; + const record = svc.taskStopLatch.workspaceStopRecords.get(taskId)!; expect(record.capturedTurns.has(first)).toBe(true); // The coordinator replaced T1 with T2 without T1 ever going idle. host.supersedeTurn(taskId, first, second); @@ -3222,7 +3226,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { expect(reactivatedEntry?.taskExecutionStatus).toBe("running"); await taskService.terminateAllDescendantAgentTasks(rootId); - const record = svc.workspaceStopRecords.get(taskId)!; + const record = svc.taskStopLatch.workspaceStopRecords.get(taskId)!; expect(record.capturedExecutionId).toBe(handleId); // The shortened deadlines let the cascade return while its cleanup (handle interrupt + // stream stop) is still in flight; ownership stays with that cleanup, so wait for it. @@ -3312,7 +3316,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { await settle(); await settle(); // Phase A blocks on the mutex: nothing captured, nothing latched, no owner yet. - expect(svc.workspaceStopRecords.has(taskId)).toBe(false); + expect(svc.taskStopLatch.workspaceStopRecords.has(taskId)).toBe(false); expect(svc.ownedAttemptByTaskId.has(taskId)).toBe(false); gate.resolve(); // createWorkspaceTurn runs after the section, behind the cascade's Phase A in the mutex @@ -3322,8 +3326,8 @@ describe("TaskService attempt identity and send admission (G1)", () => { success: false, error: { code: "send_failed" }, }); - await waitForCondition(() => svc.workspaceStopRecords.has(taskId)); - expect(svc.workspaceStopRecords.get(taskId)).toMatchObject({ + await waitForCondition(() => svc.taskStopLatch.workspaceStopRecords.has(taskId)); + expect(svc.taskStopLatch.workspaceStopRecords.get(taskId)).toMatchObject({ attemptId: fresh, ownedAttempt: svc.ownedAttemptByTaskId.get(taskId), }); @@ -3354,7 +3358,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { const { taskService } = createHarness(config, { aiService }); const svc = internals(taskService); const stop = taskService.terminateAllDescendantAgentTasks(rootId); - await waitForCondition(() => svc.workspaceStopRecords.has(taskId)); + await waitForCondition(() => svc.taskStopLatch.workspaceStopRecords.has(taskId)); expect( await taskService.sendMessageToDescendantAgentTask(rootId, taskId, "again", "tool-end") ).toEqual({ @@ -3364,7 +3368,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { // Nothing published: the cascade's record still names the predecessor it captured. expect(entryOf(config, taskId)?.taskAttemptId).toBe(previous); expect(svc.ownedAttemptByTaskId.has(taskId)).toBe(false); - expect(svc.workspaceStopRecords.get(taskId)?.attemptId).toBe(previous); + expect(svc.taskStopLatch.workspaceStopRecords.get(taskId)?.attemptId).toBe(previous); cleanupGate.resolve(); await stop; // Once settled, a reactivation proceeds under a fresh id. @@ -3602,7 +3606,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { expect(failure).toBeNull(); expect(clearQueue).not.toHaveBeenCalled(); expect(stopStream).not.toHaveBeenCalled(); - expect(svc.workspaceStopRecords.has(taskId)).toBe(false); + expect(svc.taskStopLatch.workspaceStopRecords.has(taskId)).toBe(false); expect(taskService.isWorkspaceStopInProgress(taskId)).toBe(false); expect(svc.attemptSettlementByTaskId.get(taskId)?.attemptId).not.toBe(foreignAttemptId); return; @@ -3616,7 +3620,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { expect(entryOf(config, taskId)?.taskLaunchError).toContain("recovery attempts"); expect(failure?.errorType).toBe("task_recovery_limit"); await waitForCondition(() => !taskService.isWorkspaceStopInProgress(taskId)); - expect(svc.workspaceStopRecords.has(taskId)).toBe(false); + expect(svc.taskStopLatch.workspaceStopRecords.has(taskId)).toBe(false); } ); }); @@ -3816,7 +3820,7 @@ describe("TaskService attempt identity and send admission (G1)", () => { ).toBeNull(); expect(clearQueue).not.toHaveBeenCalled(); expect(stopStream).not.toHaveBeenCalled(); - expect(svc.workspaceStopRecords.has(taskId)).toBe(false); + expect(svc.taskStopLatch.workspaceStopRecords.has(taskId)).toBe(false); expect(svc.attemptSettlementByTaskId.get(taskId)?.attemptId).not.toBe(foreign); }); test("a resume-failure rollback restores only the reawakened attempt: a row another backend re-admitted meanwhile keeps its running status", async () => { diff --git a/src/node/services/taskService.reawakening.test.ts b/src/node/services/taskService.reawakening.test.ts index 99f54a876fa..01099eec154 100644 --- a/src/node/services/taskService.reawakening.test.ts +++ b/src/node/services/taskService.reawakening.test.ts @@ -468,9 +468,9 @@ describe("TaskService", () => { // free it — otherwise the child stays barred from peer messaging until restart even after // every admission-visible marker refuses on its own. const internals = taskService as unknown as { - workspaceStopsInProgress: Map; + taskStopLatch: { workspaceStopsInProgress: Map }; }; - expect(internals.workspaceStopsInProgress.has("leaf-a")).toBe(true); + expect(internals.taskStopLatch.workspaceStopsInProgress.has("leaf-a")).toBe(true); // A STALE handle settling for the same workspace is NOT settlement for the live execution: // the mirror still points at wst_leaf, so releasing here would let the still-running child @@ -487,14 +487,14 @@ describe("TaskService", () => { "wst_stale" ); expect(staleInterrupt.success).toBe(true); - expect(internals.workspaceStopsInProgress.has("leaf-a")).toBe(true); + expect(internals.taskStopLatch.workspaceStopsInProgress.has("leaf-a")).toBe(true); const interrupted = await workspaceTurnManagerFor(taskService).interruptWorkspaceTurn( "tree-root", "wst_leaf" ); expect(interrupted.success).toBe(true); - expect(internals.workspaceStopsInProgress.has("leaf-a")).toBe(false); + expect(internals.taskStopLatch.workspaceStopsInProgress.has("leaf-a")).toBe(false); }); test("successful no-op stream stop still retains the latch for an unsettled PREPARING execution the cascade could not settle", async () => { @@ -546,8 +546,10 @@ describe("TaskService", () => { interruptSpy.mockRestore(); expect(findWorkspaceInConfig(config, "leaf-a")?.taskStatus).toBe("reported"); - const internals = taskService as unknown as { workspaceStopsInProgress: Map }; - expect(internals.workspaceStopsInProgress.has("leaf-a")).toBe(true); + const internals = taskService as unknown as { + taskStopLatch: { workspaceStopsInProgress: Map }; + }; + expect(internals.taskStopLatch.workspaceStopsInProgress.has("leaf-a")).toBe(true); // User resume clears the level-triggered suppression; only the retained latch refuses the // prepared turn's child until its execution settles. @@ -568,7 +570,7 @@ describe("TaskService", () => { "wst_leaf" ); expect(interrupted.success).toBe(true); - expect(internals.workspaceStopsInProgress.has("leaf-a")).toBe(false); + expect(internals.taskStopLatch.workspaceStopsInProgress.has("leaf-a")).toBe(false); }); test("settlement with a swallowed mirror write still refuses peer sends via registration removal", async () => { @@ -615,9 +617,9 @@ describe("TaskService", () => { expect(interruptSpy).toHaveBeenCalledTimes(1); interruptSpy.mockRestore(); const internals = taskService as unknown as { - workspaceStopsInProgress: Map; + taskStopLatch: { workspaceStopsInProgress: Map }; }; - expect(internals.workspaceStopsInProgress.has("leaf-a")).toBe(true); + expect(internals.taskStopLatch.workspaceStopsInProgress.has("leaf-a")).toBe(true); // Settlement's terminal mirror write does not land (another writer replaced it), so the // on-disk mirror keeps claiming "running". Releasing the latch on the @@ -641,7 +643,7 @@ describe("TaskService", () => { expect( workspaceTurnManagerFor(taskService).getLiveWorkspaceTurnRegistration("leaf-a") ).toBeUndefined(); - expect(internals.workspaceStopsInProgress.has("leaf-a")).toBe(false); + expect(internals.taskStopLatch.workspaceStopsInProgress.has("leaf-a")).toBe(false); taskService.resetAutoResumeCount("branch-a"); expect( @@ -702,7 +704,7 @@ describe("TaskService", () => { status: "interrupted", }); const internals = taskService as unknown as { - workspaceStopsInProgress: Map; + taskStopLatch: { workspaceStopsInProgress: Map }; }; expect( workspaceTurnManagerFor(taskService).getLiveWorkspaceTurnRegistration("leaf-a") @@ -712,7 +714,7 @@ describe("TaskService", () => { expect.objectContaining({ abandonPartial: false }) ); // No owner left to settle: the latch releases with the cascade instead of at restart. - expect(internals.workspaceStopsInProgress.has("leaf-a")).toBe(false); + expect(internals.taskStopLatch.workspaceStopsInProgress.has("leaf-a")).toBe(false); // The owner's terminal wake for the handle its own Stop interrupted is suppressed (same as // task_stop), so a restart cannot resurrect it as a wake-up. const attention = await new TerminalAttentionStore(config).get( @@ -760,8 +762,10 @@ describe("TaskService", () => { taskService.markParentWorkspaceInterrupted("branch-a"); await taskService.terminateAllDescendantAgentTasks("branch-a"); - const internals = taskService as unknown as { workspaceStopsInProgress: Map }; - expect(internals.workspaceStopsInProgress.has("leaf-a")).toBe(false); + const internals = taskService as unknown as { + taskStopLatch: { workspaceStopsInProgress: Map }; + }; + expect(internals.taskStopLatch.workspaceStopsInProgress.has("leaf-a")).toBe(false); }); test("sendAgentTreeMessage refuses root sends while an interrupted workspace turn winds down", async () => { diff --git a/src/node/services/taskService.reportDecisionHold.test.ts b/src/node/services/taskService.reportDecisionHold.test.ts index 3ebc277eb1b..a6b77199995 100644 --- a/src/node/services/taskService.reportDecisionHold.test.ts +++ b/src/node/services/taskService.reportDecisionHold.test.ts @@ -53,7 +53,7 @@ interface Internals { attemptSettlementByTaskId: Map; admittedSendsByTaskId: Map>; streamEndDecisionsByTaskId: Map>; - workspaceStopRecords: Map; + taskStopLatch: { workspaceStopRecords: Map }; emitWorkspaceMetadata: (workspaceId: string) => Promise; workspaceEventLocks: { withLock(key: string, operation: () => Promise): Promise }; } @@ -1250,7 +1250,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => { expect(onCanceled).toHaveBeenCalledTimes(1); expect(onCanceled).toHaveBeenCalledWith("Queued message cleared before dispatch."); expect(outstanding(svc, childId)).toHaveLength(0); - expect(svc.workspaceStopRecords.has(childId)).toBe(false); + expect(svc.taskStopLatch.workspaceStopRecords.has(childId)).toBe(false); expect(taskService.isWorkspaceStopInProgress(childId)).toBe(false); expect(svc.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)).toEqual([ "pending", diff --git a/src/node/services/taskService.settlementReceipts.test.ts b/src/node/services/taskService.settlementReceipts.test.ts index 04fcfefac3a..9dd1eb5c1d9 100644 --- a/src/node/services/taskService.settlementReceipts.test.ts +++ b/src/node/services/taskService.settlementReceipts.test.ts @@ -60,7 +60,7 @@ const PREDECESSOR = "att_00000000000000f1"; interface Internals { ownedAttemptByTaskId: Map; attemptSettlementByTaskId: Map; - workspaceStopRecords: Map; + taskStopLatch: { workspaceStopRecords: Map }; attemptSettlementListenersByTaskId: Map void>>; waitForWorkspaceStopsToRelease: (ids: readonly string[], deadlineMs: number) => Promise; releaseSharedDesktopTaskOnUserStop: (taskId: string, abortOrigin: unknown) => Promise; @@ -617,7 +617,7 @@ describe("TaskService settlement receipt producers (G2)", () => { expect((await taskService.stopDescendantAgentTask(rootId, taskId)).success).toBe(true); // A failed write still releases the latch (no latch pinned until restart). await waitForCondition(() => !taskService.isWorkspaceStopInProgress(taskId)); - expect(svc.workspaceStopRecords.has(taskId)).toBe(false); + expect(svc.taskStopLatch.workspaceStopRecords.has(taskId)).toBe(false); } expect(entryOf(config, taskId)?.taskStatus).toBe("interrupted"); expect(svc.attemptSettlementByTaskId.get(taskId)).toMatchObject({ @@ -955,7 +955,7 @@ describe("TaskService settlement receipt producers (G2)", () => { }); expect((await taskService.stopDescendantAgentTask(rootId, taskId)).success).toBe(true); await waitForCondition(() => !taskService.isWorkspaceStopInProgress(taskId)); - expect(svc.workspaceStopRecords.has(taskId)).toBe(false); + expect(svc.taskStopLatch.workspaceStopRecords.has(taskId)).toBe(false); expect(entryOf(config, taskId)?.taskAttemptId).toBe(successor); // Neither id gets a receipt: the predecessor's settlement vouches only while the row names it. await expectNoReceipt(config, taskId, successor); @@ -1203,7 +1203,7 @@ describe("TaskService settlement receipt producers (G2)", () => { // Every other release condition holds once the cleanup paid back; the pending decision // alone keeps the latch, and no receipt write has even begun. const record = () => - svc.workspaceStopRecords.get(taskId) as + svc.taskStopLatch.workspaceStopRecords.get(taskId) as | { cleanupInFlight: number; receipt?: string } | undefined; await waitForCondition(() => record()?.cleanupInFlight === 0); diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index b28b3061e3a..0747170821b 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -13,7 +13,6 @@ import type { z } from "zod"; import { TASK_CREATE_WAIT_WARNING_MS, TASK_TERMINATION_STOP_STREAM_AGGREGATE_TIMEOUT_MS, - TASK_TERMINATION_STOP_STREAM_TIMEOUT_MS, TASK_TERMINATION_WORKSPACE_REMOVE_TIMEOUT_MS, } from "@/constants/terminationTimeouts"; import { SLOW_STARTUP_WARN_THRESHOLD_MS } from "@/constants/startup"; @@ -112,6 +111,7 @@ import { TASK_FAMILY_MESSAGE_MAX_CHARS, } from "@/constants/taskMessages"; import { log } from "@/node/services/log"; +import { TaskStopLatch } from "@/node/services/taskStopLatch"; import { PeerSendParking } from "@/node/services/peerSendParking"; import { TaskWaiters, @@ -646,7 +646,7 @@ export interface AgentTaskReport { * can be matched exactly against the attempt it belongs to. Persisted status, a stop cascade or * startup reconciliation never mint one: a prior backend's attempt has no owner in this process. */ -interface OwnedTaskAttempt { +export interface OwnedTaskAttempt { readonly generation: number; readonly source: string; /** @@ -690,7 +690,7 @@ interface AttemptSettlementEntry { * the seam). Identity is immutable: an obligation admitted under attempt X can never enter, or be * removed from, a successor attempt's stop record. */ -interface AdmittedSend { +export interface AdmittedSend { readonly taskId: string; readonly attemptId: string; /** Set for sends admitted while this process owns the attempt (receipt authority, later change). */ @@ -736,7 +736,7 @@ interface AdmittedSend { * was decided for is live: input sent after it resolved but before that turn settles targets the * same attempt and must still find it (preflight refusal, or the dequeue gate's refusal). */ -interface StreamEndDecision { +export interface StreamEndDecision { readonly attempt: OwnedTaskAttempt | undefined; readonly attemptId: string; readonly messageId: string; @@ -1393,8 +1393,8 @@ const CHILD_GOAL_ARBITRATED_STREAMS_MAX = 32; */ export const ATTEMPT_CLOSURE_SETTLE_WAIT_MS = 5_000; -/** See TaskService.workspaceStopRecords. */ -interface WorkspaceStopRecord { +/** See TaskStopLatch.workspaceStopRecords. */ +export interface WorkspaceStopRecord { /** Latch releases (one per overlapping cascade) run together once release conditions hold. */ releases: Array<() => void>; /** @@ -1440,19 +1440,12 @@ interface WorkspaceStopRecord { */ receipt?: "writing" | "written" | "failed"; } -/** - * The stop record's release conditions hold: no cleanup owed, a durable stop marker, no captured - * turn or pending admission left, and the captured execution settled. - */ -function isWorkspaceStopRecordSettled(record: WorkspaceStopRecord): boolean { - if (record.cleanupInFlight > 0) return false; - return ( - record.stopPersisted && - record.capturedTurns.size === 0 && - record.pendingAdmissions.size === 0 && - record.executionSettled - ); -} + +/** Result of TaskService.decideSettlementReceipt. */ +export type SettlementReceiptDecision = + | { kind: "memory-only" } + | { kind: "unreadable" } + | { kind: "write"; attemptId: string; parentWorkspaceId: string; owners: string[] }; /** Lineage proof (evaluateAttemptLineage) from the predecessor's receipt in the parent's dir. */ function lineageProofFromReceipt( @@ -2022,21 +2015,13 @@ export class TaskService implements AgentTaskIntegration { */ private interruptedParentWorkspaceIds = new Set(); /** - * Monotonic per-workspace stop generation, bumped synchronously at every explicit stop - * boundary (user Stop suppression, task_stop interruption, workspace-turn interruption). - * Peer-send admission captures its endpoints' generations and refuses when any changed: - * unlike the level-triggered suppression set and persisted statuses — which a quick user - * resume clears between probe evaluations — a bump latches, so an intervening stop keeps - * the in-flight send stale forever. - */ - private readonly workspaceStopEpochs = new Map(); - /** - * User Stops per workspace (markParentWorkspaceInterrupted). The stop epoch above also counts + * User Stops per workspace (markParentWorkspaceInterrupted). The stop epoch also counts * owner interrupts of a delegated turn, which a message waiting for that turn survives (see * flushParkedPeerSends), so parking needs this separate count to tell a user Stop apart. */ private readonly workspaceUserStopEpochs = new Map(); private readonly peerSendParking: PeerSendParking; + private readonly taskStopLatch: TaskStopLatch; /** * Child stream ends cut by a host-selected continuation (queued input or context-budget * hand-over) whose successor has not yet streamed or been withdrawn. Keyed by task; settled @@ -2059,23 +2044,6 @@ export class TaskService implements AgentTaskIntegration { attemptId: AttemptFence; } >(); - /** - * Level-triggered stop latches: workspace IDs whose stop cascade is currently between its - * synchronous epoch bump and terminal status persistence. The epoch map alone cannot refuse - * a send that ENTERS during that window — the post-bump generation becomes the send's clean - * baseline while the endpoints' persisted statuses still read running — so peer-send - * admission also rejects any endpoint chain holding an in-progress stop latch. Refcounted: - * overlapping stops (a subtree stop racing a workspace-turn interrupt on the same workspace) - * must not clear each other's latch. - */ - private readonly workspaceStopsInProgress = new Map(); - /** - * Latched workspace IDs whose held latch deferred a queued-task launch (own or parent latch; - * see startReservedAgentTask). The launch record was handed back to "queued", and the queue is - * rechecked exactly when that latch clears — not on every release (startup normalizes stopped - * trees before any drain) and not on an unrelated scheduler trigger. - */ - private readonly queueRechecksOwedByStopLatch = new Set(); /** * Attempt ownership ledger (see OwnedTaskAttempt); replaced before any new admission, dropped * once the attempt's report is durable (releaseReportedTaskAttempt). @@ -2120,15 +2088,6 @@ export class TaskService implements AgentTaskIntegration { private readonly reawakeningsInFlight = new Map(); /** Per task, the stream-end decisions of this process's owned attempts (see StreamEndDecision). */ private readonly streamEndDecisionsByTaskId = new Map(); - /** Stop latches retained past their cascade because the stop could not be confirmed; released on authoritative terminal settlement. */ - /** - * Ownership of every in-progress stop, keyed by workspace (see beginWorkspaceStop). The latch - * a record guards drops only when BOTH hold: the captured owner settled authoritatively - * (admitted turn generation ended; matching execution mirror settled) AND no cleanup issued - * by the cascade is still in flight. Persisted statuses, stop results, and missing - * registrations are not evidence — an accepted-but-preparing turn can still start. - */ - private readonly workspaceStopRecords = new Map(); /** Tracks consecutive auto-resumes per workspace. Reset when a user message is sent. */ private consecutiveAutoResumes = new Map(); @@ -2432,12 +2391,14 @@ export class TaskService implements AgentTaskIntegration { return null; } + /** WorkspaceTurnTaskHost (see TaskStopLatch.bumpWorkspaceStopEpoch). */ bumpWorkspaceStopEpoch(workspaceId: string): void { - this.workspaceStopEpochs.set(workspaceId, (this.workspaceStopEpochs.get(workspaceId) ?? 0) + 1); + this.taskStopLatch.bumpWorkspaceStopEpoch(workspaceId); } + /** AgentTaskIntegration (see TaskStopLatch.getWorkspaceStopEpoch). */ getWorkspaceStopEpoch(workspaceId: string): number { - return this.workspaceStopEpochs.get(workspaceId) ?? 0; + return this.taskStopLatch.getWorkspaceStopEpoch(workspaceId); } /** WorkspaceTurnTaskHost: a delegated turn's registration release (#4997). */ @@ -2455,355 +2416,27 @@ export class TaskService implements AgentTaskIntegration { return this.peerSendParking.flushParkedPeerSends(targetId); } - /** - * Hold a stop-in-progress latch for the given workspaces until the returned release runs. - * Callers latch synchronously with the epoch bump and release only after the cascade has - * persisted terminal state, so sends entering mid-stop are refused at admission. - */ + /** WorkspaceTurnTaskHost (see TaskStopLatch.latchWorkspaceStopsInProgress). */ latchWorkspaceStopsInProgress(workspaceIds: readonly string[]): () => void { - for (const id of workspaceIds) { - this.workspaceStopsInProgress.set(id, (this.workspaceStopsInProgress.get(id) ?? 0) + 1); - } - return () => { - for (const id of workspaceIds) { - const count = this.workspaceStopsInProgress.get(id) ?? 0; - if (count <= 1) { - this.workspaceStopsInProgress.delete(id); - if (this.queueRechecksOwedByStopLatch.delete(id)) { - this.scheduleMaybeStartQueuedTasks(); - } - this.notifyAttemptSettlementListeners(id); - // An interrupted delegated turn releases its registration while this latch is held. - this.peerSendParking.scheduleParkedPeerSendFlush(id); - } else { - this.workspaceStopsInProgress.set(id, count - 1); - } - } - }; + return this.taskStopLatch.latchWorkspaceStopsInProgress(workspaceIds); } + /** AgentTaskIntegration: a stop cascade holds the workspace's latch. */ isWorkspaceStopInProgress(workspaceId: string): boolean { - return this.workspaceStopsInProgress.has(workspaceId); - } - - /** - * Stop-cascade launch barrier: true (and the queue recheck owed to the held latch recorded) - * while the task's own or its parent's stop latch is held. The record stays/returns to - * "queued"; the latch release rechecks the queue (see queueRechecksOwedByStopLatch). - */ - private deferLaunchWhileStopInProgress( - taskId: string, - parentWorkspaceId: string | undefined - ): boolean { - let deferred = false; - for (const id of [taskId, parentWorkspaceId]) { - if (id != null && this.isWorkspaceStopInProgress(id)) { - this.queueRechecksOwedByStopLatch.add(id); - deferred = true; - } - } - return deferred; - } - - /** - * Phase A of a stop cascade (synchronous, under the global mutex): bump the epoch, latch the - * workspace and record what must settle before the latch may drop — the admitted turn - * generation (owner of any preparing/streaming work) and the live execution mirror. Nothing - * admitted after this point exists (admission barrier), so these captures are complete. - * Overlapping cascades add their release to the same record instead of resetting ownership. - */ - private beginWorkspaceStop(workspaceId: string): WorkspaceStopRecord { - this.bumpWorkspaceStopEpoch(workspaceId); - const release = this.latchWorkspaceStopsInProgress([workspaceId]); - const existing = this.workspaceStopRecords.get(workspaceId); - if (existing != null) { - // An overlapping cascade adds its release and plans its own cleanup; ownership of the - // earlier cascade's counts is untouched. - existing.releases.push(release); - existing.cleanupInFlight += 1; - return existing; - } - const capturedTurn = this.workspaceService.getActiveTurnGeneration(workspaceId); - const capturedExecutionId = - this.getWorkspaceTurnManager().getLiveWorkspaceTurnRegistration(workspaceId)?.handleId; - const ownedAttempt = this.ownedAttemptByTaskId.get(workspaceId); - const attemptId = ownedAttempt?.attemptId ?? this.currentTaskAttemptId(workspaceId); - const record: WorkspaceStopRecord = { - releases: [release], - cleanupInFlight: 1, - attemptId, - capturedTurns: new Set(capturedTurn != null ? [capturedTurn] : []), - pendingAdmissions: new Set(), - capturedExecutionId, - // A registration whose mirror already persisted terminally is settled (its settlement - // callback ran before this capture and will not run again). - executionSettled: - capturedExecutionId == null || - this.isExecutionMirrorSettled(workspaceId, capturedExecutionId), - stopPersisted: false, - ownedAttempt, - capturedStreamMessageId: - this.getWorkspaceTurnManager().captureQueueCutAttributionSnapshot(workspaceId).activeStream - ?.messageId, - }; - // Sends admitted against this attempt that have not claimed a turn yet are part of what - // must settle: captured by reference so a later obligation of a successor attempt (or one - // already superseded) can never enter this record. - for (const send of this.admittedSendsByTaskId.get(workspaceId) ?? []) { - if (send.attemptId !== attemptId) continue; - if (send.state !== "pending" && send.state !== "enqueued") continue; - record.pendingAdmissions.add(send); - send.capturedBy.add(record); - } - this.workspaceStopRecords.set(workspaceId, record); - return record; + return this.taskStopLatch.isWorkspaceStopInProgress(workspaceId); } - /** - * Phase B (outside the global mutex): every descendant's clearQueue + stopStream, issued - * exactly once while its latch is held and targeted at the execution captured in Phase A, raced - * concurrently under a per-child and one aggregate deadline. A timeout abandons the WAIT only: - * the original promise keeps its cleanup ownership and its later settlement rechecks release. - * Must run for EVERY id latched in Phase A (each planned one cleanup), so callers pass the - * full latched set even when Phase A failed midway. - */ - private async runWorkspaceStopCleanup( + /** Tests spy on this; TaskStopLatch owns Phase B. */ + private runWorkspaceStopCleanup( targets: readonly string[], - options: { - label: string; - abandonPartial: boolean; - clearQueue: boolean; - /** - * Live continuation handles captured in Phase A (see WorkspaceStopRecord.capturedExecutionId), - * interrupted here so their mirror settles: the stream stop below is a `system` abort, which - * never settles a workspace-turn handle (only a user abort does), so without this the record - * would wait for an execution settlement that never comes and hold the latch until restart. - */ - capturedExecutionsById?: ReadonlyMap; - onTimeout?: (workspaceId: string) => void; - } + options: Parameters[1] ): Promise { - const waits = targets.map(async (id) => { - // Consumes the unit planned by beginWorkspaceStop; the finally below pays it back. - const record = this.workspaceStopRecords.get(id); - const expectedMessageId = record?.capturedStreamMessageId; - const cleanup = (async () => { - if (options.clearQueue) { - // AgentSession stream-end cleanup auto-flushes queued messages, so a stopped - // descendant must not keep pending input; issued once, never re-issued later. The - // user's queued messages are handed back as held input, never silently dropped. - try { - const clearQueueResult = this.workspaceService.clearQueue(id, { - preserveUserInput: true, - }); - if (!clearQueueResult.success) { - log.debug(`${options.label}: clearQueue failed`, { - taskId: id, - error: clearQueueResult.error, - }); - } - } catch (error: unknown) { - log.debug(`${options.label}: clearQueue threw`, { taskId: id, error }); - } - } - const execution = options.capturedExecutionsById?.get(id); - if (execution != null) { - await this.interruptCapturedExecution(id, execution, options.label); - } - // Success is NOT stop confirmation (an accepted-but-PREPARING turn has no registered - // stream yet); only the owner's settlement recorded on the stop record confirms. - const stopResult = await this.aiService.stopStream(id, { - abandonPartial: options.abandonPartial, - ...(expectedMessageId != null ? { expectedMessageId } : {}), - }); - if (!stopResult.success) { - log.debug(`${options.label}: stopStream failed`, { taskId: id }); - } - })(); - // Ownership stays with the original promise: only ITS settlement (however late) counts - // the cleanup as finished, and even then it merely triggers a recheck. - void cleanup - .catch((error: unknown) => { - log.debug(`${options.label}: cleanup threw`, { taskId: id, error }); - }) - .finally(() => { - if (record != null) record.cleanupInFlight -= 1; - this.recheckWorkspaceStopRelease(id); - }); - const outcome = await raceWithAbortAndTimeout(cleanup, { - timeoutMs: TASK_TERMINATION_STOP_STREAM_TIMEOUT_MS, - }); - if (outcome.kind !== "ok") { - log.warn(`${options.label}: stopStream timed out; retaining stop latch`, { taskId: id }); - options.onTimeout?.(id); - } - }); - const aggregate = await raceWithAbortAndTimeout(Promise.allSettled(waits), { - timeoutMs: TASK_TERMINATION_STOP_STREAM_AGGREGATE_TIMEOUT_MS, - }); - if (aggregate.kind !== "ok") { - log.warn(`${options.label}: cleanup exceeded the aggregate deadline`, { - taskIds: targets, - }); - } + return this.taskStopLatch.runWorkspaceStopCleanup(targets, options); } - /** - * Settle the continuation handle a stop record captured (a reawakened child executes under a - * WorkspaceTurnManager handle): interruptWorkspaceTurn persists the handle and its execution - * mirror terminal within one settlement boundary and drops the live registration, which is the - * authoritative execution settlement the record waits on (releaseRetainedStopLatches). Same - * pairing as the task_stop subtree stop (finishSubtreeStopCleanup): the owner's terminal wake - * for a handle the stop itself interrupted is suppressed. A failure leaves the record waiting - * (fail closed), exactly as an unsettled execution does today. - */ - private async interruptCapturedExecution( - taskId: string, - execution: { ownerWorkspaceId: string; handleId: string }, - label: string - ): Promise { - const interrupted = await this.getWorkspaceTurnManager().interruptWorkspaceTurn( - execution.ownerWorkspaceId, - execution.handleId, - { scheduleQueueDrain: false } - ); - if (!interrupted.success) { - log.warn(`${label}: interruptWorkspaceTurn failed for the captured execution`, { - taskId, - handleId: execution.handleId, - error: interrupted.error, - }); - return; - } - await this.terminalAttention.suppressTerminalAttention({ - ownerWorkspaceId: execution.ownerWorkspaceId, - sourceKind: "workspace_turn", - sourceId: execution.handleId, - }); - } - - /** - * Phase C: drop the latch iff the captured owner settled, a durable stop marker exists and no - * cleanup is owed. Called from every settlement path; a late cleanup completion or a - * persisted-status caller only triggers this recheck and never supplies evidence by itself. - * A missing config entry is NOT evidence: a still-preparing session settles through its own - * disposal (the coordinator publishes the captured generation's idle transition). - */ - private recheckWorkspaceStopRelease(workspaceId: string): void { - const record = this.workspaceStopRecords.get(workspaceId); - if (record == null) return; - if (!isWorkspaceStopRecordSettled(record)) return; - // Execution-settlement producer: the receipt is durable before the latch releases, so nothing - // the latch holds back (a reawaken, a peer send, a queued launch) observes the stopped attempt - // as released while its receipt could still be missing. The record stays registered (latched) - // for the write; its completion rechecks, releasing the latch and recording the settlement in - // one step as before. A failed write (or an unreadable config) still releases, with the - // attempt left `closing` (fail closed) instead of pinning the latch until restart. - if (record.receipt === "writing") return; - if (this.isStopReceiptDecisionPending(workspaceId, record)) return; - if (record.receipt == null && this.startStopSettlementReceipt(workspaceId, record)) return; - this.releaseWorkspaceStopRecord(workspaceId, record); - } - - /** - * A receipt can be at stake: wait for the attempt's pending stream-end decision (its report - * may still publish); resolveStreamEndDecision rechecks. Without a receipt at stake the - * release keeps its previous timing. - */ - private isStopReceiptDecisionPending(workspaceId: string, record: WorkspaceStopRecord): boolean { - return ( - record.receipt == null && - record.attemptId != null && - this.ownsReceiptEligibleAttempt(workspaceId, record.ownedAttempt) && - this.findStreamEndDecision(workspaceId, record.attemptId)?.outcome === "pending" - ); - } - - /** - * Decide the stopped attempt's settlement receipt and start its write. Returns true while the - * write is in flight (the release waits for its completion recheck). An unreadable config marks - * the receipt failed and lets the release go ahead. - */ - private startStopSettlementReceipt(workspaceId: string, record: WorkspaceStopRecord): boolean { - const decision = this.decideSettlementReceipt(workspaceId, record.ownedAttempt, "stop-settled"); - if (decision.kind === "unreadable") record.receipt = "failed"; - if (decision.kind !== "write") return false; - record.receipt = "writing"; - void this.writeSettlementReceipt(workspaceId, decision, "execution-settled") - .then( - (written) => { - record.receipt = written ? "written" : "failed"; - }, - (error: unknown) => { - log.error("Stop release: settlement receipt write threw", { workspaceId, error }); - record.receipt = "failed"; - } - ) - .finally(() => this.recheckWorkspaceStopRelease(workspaceId)); - return true; - } - - /** Release step of recheckWorkspaceStopRelease: drop the latch and settle the stopped attempt. */ - private releaseWorkspaceStopRecord(workspaceId: string, record: WorkspaceStopRecord): void { - this.workspaceStopRecords.delete(workspaceId); - for (const release of record.releases) { - release(); - } - // Owner settled + cleanup finished + marker persisted: authoritative for the attempt captured - // in Phase A. An unknown (legacy) attempt has none to settle — stopping it proves nothing, - // but its id is closed to further sends until a new admission rotates it. - if (record.ownedAttempt == null && record.attemptId != null) { - this.closeAttemptAdmission(workspaceId, record.attemptId, undefined, "stop-settled"); - } - // A failed receipt write leaves the attempt `closing` (never settled without its receipt). - if (record.receipt !== "failed") { - this.settleOwnedTaskAttempt(workspaceId, record.ownedAttempt, "stop-settled"); - } - // A failed reawaken's settlement deferred to this Stop (#4310): drop or finish it now; no - // later disposal or turn settlement is guaranteed to recheck it. - this.recheckResumeFailureSettlement(workspaceId); - } - - /** - * Authoritative execution settlement (WorkspaceTurnManager settled the MATCHING execution - * mirror and removed its live registration). Records the evidence for the captured execution - * and rechecks; persisted interrupted statuses from other callers reach only the recheck. - */ + /** WorkspaceTurnTaskHost (see TaskStopLatch.releaseRetainedStopLatches). */ releaseRetainedStopLatches(workspaceId: string): void { - const record = this.workspaceStopRecords.get(workspaceId); - if (record == null) return; - const liveHandleId = - this.getWorkspaceTurnManager().getLiveWorkspaceTurnRegistration(workspaceId)?.handleId; - if ( - record.capturedExecutionId != null && - (liveHandleId !== record.capturedExecutionId || - this.isExecutionMirrorSettled(workspaceId, record.capturedExecutionId)) - ) { - record.executionSettled = true; - } - this.recheckWorkspaceStopRelease(workspaceId); - } - - /** The persisted execution mirror for this handle is terminal (settled by its owner). */ - private isExecutionMirrorSettled(workspaceId: string, handleId: string): boolean { - const workspace = findWorkspaceEntry(this.config.loadConfigOrDefault(), workspaceId)?.workspace; - return ( - workspace?.taskExecutionId === handleId && - !isActiveWorkspaceTurnTaskStatus(workspace.taskExecutionStatus) - ); - } - - /** Phase A recorded a durable stop marker for this workspace (see WorkspaceStopRecord). */ - private markWorkspaceStopPersisted(workspaceId: string): void { - const record = this.workspaceStopRecords.get(workspaceId); - if (record == null) return; - record.stopPersisted = true; - // A deferred child stream end cancels on this notification (the stop is in progress), so its - // pending decision cannot hold the latch. Every caller runs after its durable status write, or - // where no status is left to write (a removed row, or an already-terminal status that is the - // marker), on every branch, including those that skip recordTaskInterrupted: a terminate whose - // removal keeps the workspace (#6920), or a status already interrupted by an archive (#6967). - this.notifyAttemptSettlementListeners(workspaceId); + this.taskStopLatch.releaseRetainedStopLatches(workspaceId); } /** @@ -2827,10 +2460,7 @@ export class TaskService implements AgentTaskIntegration { } if (decisionTurnSettled) this.pruneStreamEndDecisions(workspaceId); this.recheckResumeFailureSettlement(workspaceId); - const record = this.workspaceStopRecords.get(workspaceId); - if (!record?.capturedTurns.has(turnGeneration)) return; - record.capturedTurns.delete(turnGeneration); - this.recheckWorkspaceStopRelease(workspaceId); + this.taskStopLatch.releaseCapturedTurn(workspaceId, turnGeneration); } /** @@ -2850,10 +2480,7 @@ export class TaskService implements AgentTaskIntegration { for (const decision of this.streamEndDecisionsByTaskId.get(workspaceId) ?? []) { if (decision.turn === previous) decision.turn = next; } - const record = this.workspaceStopRecords.get(workspaceId); - if (!record?.capturedTurns.has(previous)) return; - record.capturedTurns.delete(previous); - record.capturedTurns.add(next); + this.taskStopLatch.rebindCapturedTurn(workspaceId, previous, next); } /** @@ -3055,7 +2682,7 @@ export class TaskService implements AgentTaskIntegration { const attemptId = attempt.attemptId; const closed = this.isAttemptClosed(taskId, attemptId) || - this.workspaceStopRecords.get(taskId)?.attemptId === attemptId; + this.taskStopLatch.stopRecordAttemptId(taskId) === attemptId; if (!closed) return false; for (const send of this.admittedSendsByTaskId.get(taskId) ?? []) { if (send.attemptId === attemptId) return false; @@ -3104,10 +2731,7 @@ export class TaskService implements AgentTaskIntegration { taskId: string, attempt: OwnedTaskAttempt | undefined, settlementSource: string - ): - | { kind: "memory-only" } - | { kind: "unreadable" } - | { kind: "write"; attemptId: string; parentWorkspaceId: string; owners: string[] } { + ): SettlementReceiptDecision { if (!this.ownsReceiptEligibleAttempt(taskId, attempt)) return { kind: "memory-only" }; const attemptId = attempt.attemptId; if (!this.attemptCannotStillReport(taskId, attempt)) return { kind: "memory-only" }; @@ -3377,7 +3001,7 @@ export class TaskService implements AgentTaskIntegration { }); this.pruneStreamEndDecisions(taskId); // A stop record of this attempt waits for the decision (recheckWorkspaceStopRelease). - this.recheckWorkspaceStopRelease(taskId); + this.taskStopLatch.recheckWorkspaceStopRelease(taskId); this.workspaceService.drainQueuedMessagesIfIdle(taskId); return true; } @@ -3433,6 +3057,13 @@ export class TaskService implements AgentTaskIntegration { } } + /** TaskStopLatch's waits register here (TaskStopLatchDeps.addAttemptSettlementListener). */ + private addAttemptSettlementListener(taskId: string, listener: () => void): void { + const listeners = this.attemptSettlementListenersByTaskId.get(taskId) ?? new Set(); + listeners.add(listener); + this.attemptSettlementListenersByTaskId.set(taskId, listeners); + } + private removeAttemptSettlementListener(taskId: string, listener: () => void): void { const listeners = this.attemptSettlementListenersByTaskId.get(taskId); if (!listeners) return; @@ -3710,7 +3341,7 @@ export class TaskService implements AgentTaskIntegration { }) .finally(() => { record.cleanupInFlight -= 1; - this.recheckWorkspaceStopRelease(taskId); + this.taskStopLatch.recheckWorkspaceStopRelease(taskId); }); } send.capturedBy.clear(); @@ -3731,8 +3362,8 @@ export class TaskService implements AgentTaskIntegration { const records = [...send.capturedBy]; this.dischargeAdmittedSend(send); for (const record of records) { - if (this.workspaceStopRecords.get(taskId) === record) { - this.recheckWorkspaceStopRelease(taskId); + if (this.taskStopLatch.isCurrentStopRecord(taskId, record)) { + this.taskStopLatch.recheckWorkspaceStopRelease(taskId); } } this.recheckResumeFailureSettlement(taskId); @@ -3886,36 +3517,12 @@ export class TaskService implements AgentTaskIntegration { return settlement?.attempt === owned ? settlement.phase : undefined; } - /** - * Resolve once none of `workspaceIds` holds a Stop latch, or at `deadlineMs`. A Stop releases - * its latch only after its settlement receipt is durable or failed (recheckWorkspaceStopRelease), - * and the release notifies the attempt settlement listeners. - */ - private async waitForWorkspaceStopsToRelease( + /** Tests spy on this; TaskStopLatch owns the wait. */ + private waitForWorkspaceStopsToRelease( workspaceIds: readonly string[], deadlineMs: number ): Promise { - for (const id of workspaceIds) { - while (this.isWorkspaceStopInProgress(id)) { - const remainingMs = deadlineMs - Date.now(); - if (remainingMs <= 0) { - log.warn("Stop latch still held at the cleanup deadline; its receipt may be missing", { - taskId: id, - }); - return; - } - const released = Promise.withResolvers(); - const listener = () => released.resolve(); - const listeners = this.attemptSettlementListenersByTaskId.get(id) ?? new Set(); - listeners.add(listener); - this.attemptSettlementListenersByTaskId.set(id, listeners); - try { - await raceWithAbortAndTimeout(released.promise, { timeoutMs: remainingMs }); - } finally { - this.removeAttemptSettlementListener(id, listener); - } - } - } + return this.taskStopLatch.waitForWorkspaceStopsToRelease(workspaceIds, deadlineMs); } /** Resolve once the task's settlement entry leaves `closing` (or the bound elapses). */ @@ -4725,6 +4332,36 @@ export class TaskService implements AgentTaskIntegration { terminalAttentionStore?: TerminalAttentionStore, private readonly desktopInputCoordinator = new DesktopInputCoordinator(config) ) { + this.taskStopLatch = new TaskStopLatch({ + config: () => this.config, + aiService: () => this.aiService, + workspaceService: () => this.workspaceService, + turnManager: () => this.getWorkspaceTurnManager(), + terminalAttention: () => this.terminalAttention, + scheduleMaybeStartQueuedTasks: () => this.scheduleMaybeStartQueuedTasks(), + scheduleParkedPeerSendFlush: (workspaceId) => + this.peerSendParking.scheduleParkedPeerSendFlush(workspaceId), + notifyAttemptSettlementListeners: (taskId) => this.notifyAttemptSettlementListeners(taskId), + addAttemptSettlementListener: (taskId, listener) => + this.addAttemptSettlementListener(taskId, listener), + removeAttemptSettlementListener: (taskId, listener) => + this.removeAttemptSettlementListener(taskId, listener), + ownedAttempt: (taskId) => this.ownedAttemptByTaskId.get(taskId), + currentTaskAttemptId: (taskId) => this.currentTaskAttemptId(taskId), + admittedSends: (taskId) => this.admittedSendsByTaskId.get(taskId), + ownsReceiptEligibleAttempt: (taskId, attempt) => + this.ownsReceiptEligibleAttempt(taskId, attempt), + findStreamEndDecision: (taskId, attemptId) => this.findStreamEndDecision(taskId, attemptId), + decideSettlementReceipt: (taskId, attempt, settlementSource) => + this.decideSettlementReceipt(taskId, attempt, settlementSource), + writeSettlementReceipt: (taskId, target, source) => + this.writeSettlementReceipt(taskId, target, source), + closeAttemptAdmission: (taskId, attemptId, ownedAttempt, source) => + this.closeAttemptAdmission(taskId, attemptId, ownedAttempt, source), + settleOwnedTaskAttempt: (taskId, attempt, source) => + this.settleOwnedTaskAttempt(taskId, attempt, source), + recheckResumeFailureSettlement: (taskId) => this.recheckResumeFailureSettlement(taskId), + }); this.peerSendParking = new PeerSendParking({ flush: (targetId) => this.flushParkedPeerSends(targetId), getWorkspaceTurnManager: () => this.getWorkspaceTurnManager(), @@ -7673,7 +7310,7 @@ export class TaskService implements AgentTaskIntegration { // "queued", so a deferred launch hands its record back or it is stranded; the latch release // rechecks the queue. Only a still-reserved record is reverted — a status written meanwhile // (an explicit Stop's "interrupted") stays as is and is never relaunched. - if (this.deferLaunchWhileStopInProgress(plan.taskId, plan.parentWorkspaceId)) { + if (this.taskStopLatch.deferLaunchWhileStopInProgress(plan.taskId, plan.parentWorkspaceId)) { log.debug("startReservedAgentTask deferred: a stop is in progress", { taskId: plan.taskId }); try { await this.editActiveWorkspaceEntry( @@ -7692,7 +7329,7 @@ export class TaskService implements AgentTaskIntegration { await this.emitWorkspaceMetadata(plan.taskId); // The release-time drain may have seen "starting" while this rollback was awaiting disk. // Complete the handoff after publishing "queued", or leave the next release owning it. - if (!this.deferLaunchWhileStopInProgress(plan.taskId, plan.parentWorkspaceId)) { + if (!this.taskStopLatch.deferLaunchWhileStopInProgress(plan.taskId, plan.parentWorkspaceId)) { this.scheduleMaybeStartQueuedTasks(); } return; @@ -8730,15 +8367,15 @@ export class TaskService implements AgentTaskIntegration { this.hasPendingAdmissions(taskId) : liveUnderLaunchAttempt(); if (rowStillOurs && liveExecution) { - this.beginWorkspaceStop(taskId); - this.markWorkspaceStopPersisted(taskId); + this.taskStopLatch.beginWorkspaceStop(taskId); + this.taskStopLatch.markWorkspaceStopPersisted(taskId); await this.runWorkspaceStopCleanup([taskId], { label: "TaskService.create launch failure", abandonPartial: true, clearQueue: true, }); await this.waitForAttemptClosureToSettle(taskId, ATTEMPT_CLOSURE_SETTLE_WAIT_MS); - if (this.workspaceStopRecords.has(taskId)) { + if (this.taskStopLatch.hasStopRecord(taskId)) { log.warn("Task.create launch failure: the failed launch's attempt is still live", { taskId, message, @@ -11106,7 +10743,7 @@ export class TaskService implements AgentTaskIntegration { // clean while statuses still read running); each latch drops once its owner settled and // cleanup finished. for (const id of taskIds) { - this.beginWorkspaceStop(id); + this.taskStopLatch.beginWorkspaceStop(id); latched.push(id); } const activeWorkspaceTurns = await this.getWorkspaceTurnManager().listAllWorkspaceTurns({ @@ -11118,7 +10755,7 @@ export class TaskService implements AgentTaskIntegration { if (!current) { // Removed since the subtree snapshot: no status is left to persist, but the latch // still waits for the captured owner and the planned cleanup (no absent-config bypass). - this.markWorkspaceStopPersisted(id); + this.taskStopLatch.markWorkspaceStopPersisted(id); continue; } const status = current.workspace.taskStatus ?? "running"; @@ -11127,7 +10764,7 @@ export class TaskService implements AgentTaskIntegration { ACTIVE_AGENT_TASK_STATUSES.has(status) || this.aiService.isStreaming(id); if (!executionActive && activeHandles.length === 0) { // Already terminal with nothing live: the persisted status is the stop marker. - this.markWorkspaceStopPersisted(id); + this.taskStopLatch.markWorkspaceStopPersisted(id); continue; } activeHandlesById.set( @@ -11160,7 +10797,7 @@ export class TaskService implements AgentTaskIntegration { }, { allowMissing: true } ); - this.markWorkspaceStopPersisted(id); + this.taskStopLatch.markWorkspaceStopPersisted(id); if (parentWorkspaceId != null) { await this.terminalAttention.suppressTerminalAttention({ ownerWorkspaceId: parentWorkspaceId, @@ -11278,7 +10915,7 @@ export class TaskService implements AgentTaskIntegration { const terminationError = new Error("Task terminated"); for (const id of toTerminate) { - this.beginWorkspaceStop(id); + this.taskStopLatch.beginWorkspaceStop(id); latched.push(id); this.waiters.deleteCompletedReport(id); this.rejectWaiters(id, terminationError); @@ -11292,7 +10929,7 @@ export class TaskService implements AgentTaskIntegration { }, { allowMissing: true } ); - this.markWorkspaceStopPersisted(id); + this.taskStopLatch.markWorkspaceStopPersisted(id); } } finally { // Phase B: every stop at once, bounded per child and in aggregate; a timed-out stop keeps @@ -11725,7 +11362,7 @@ export class TaskService implements AgentTaskIntegration { // baseline and could wake a cousin or the root after Stop. Each latch holds until the // captured owner settles and the descendant's cleanup finishes (recheckWorkspaceStopRelease). for (const id of descendants) { - const record = this.beginWorkspaceStop(id); + const record = this.taskStopLatch.beginWorkspaceStop(id); latched.push(id); // Every latched descendant, a preserved completed report included (its continuation can // be executing): synchronously here, so the retirement withdraws an in-flight wake and @@ -11765,14 +11402,14 @@ export class TaskService implements AgentTaskIntegration { // Missing descendants should still reject prompt waiters promptly so task_await does // not hang until timeout after a parent hard interrupt races with external cleanup. // There is no status left to persist; the owner conditions still apply. - this.markWorkspaceStopPersisted(id); + this.taskStopLatch.markWorkspaceStopPersisted(id); this.rejectWaiters(id, interruptionError); log.debug("terminateAllDescendantAgentTasks: descendant workspace missing", { taskId: id, }); continue; } - this.markWorkspaceStopPersisted(id); + this.taskStopLatch.markWorkspaceStopPersisted(id); if (preservedCompletedDescendant) { // A reawakened completed child executes under a live workspace-turn handle while its @@ -16427,7 +16064,10 @@ export class TaskService implements AgentTaskIntegration { // Stop-cascade barrier (see startReservedAgentTask): a task whose own or parent's stop // latch is held is not even reserved; the latch release rechecks the queue. if ( - this.deferLaunchWhileStopInProgress(taskId, coerceNonEmptyString(task.parentWorkspaceId)) + this.taskStopLatch.deferLaunchWhileStopInProgress( + taskId, + coerceNonEmptyString(task.parentWorkspaceId) + ) ) { continue; } @@ -16752,7 +16392,7 @@ export class TaskService implements AgentTaskIntegration { // and refuses a missing one, and a removed ancestor's descendants were removed before it. // Best-effort: a stop handled after this call, or a removal by another backend, can still // leave an entry behind. - this.workspaceStopEpochs.delete(workspaceId); + this.taskStopLatch.forgetStopEpoch(workspaceId); this.workspaceUserStopEpochs.delete(workspaceId); // Per-task goal arbitration history (bounded per task, but not across removed tasks). this.childGoalArbitratedStreams.delete(workspaceId); @@ -17130,7 +16770,7 @@ export class TaskService implements AgentTaskIntegration { this.resumeFailureSettlementByTaskId.delete(taskId); return; } - if (this.workspaceStopRecords.has(taskId)) return; + if (this.taskStopLatch.hasStopRecord(taskId)) return; if (!this.attemptCannotStillReport(taskId, attempt)) return; this.resumeFailureSettlementByTaskId.delete(taskId); this.persistOwnedAttemptSettlement(taskId, attempt, "idle-settled", "resume-failed").catch( @@ -19039,7 +18679,7 @@ export class TaskService implements AgentTaskIntegration { let stopRecord: WorkspaceStopRecord | undefined; if (expectedAttemptId == null && sampleLiveExecution()) { await using _lock = await this.mutex.acquire(); - stopRecord = this.beginWorkspaceStop(workspaceId); + stopRecord = this.taskStopLatch.beginWorkspaceStop(workspaceId); } let superseded = false; let transitionedToInterrupted = false; @@ -19084,7 +18724,7 @@ export class TaskService implements AgentTaskIntegration { await using _lock = await this.mutex.acquire(); rowStillExpected = this.currentTaskAttemptId(workspaceId) === expectedAttemptId; if (rowStillExpected && sampleLiveExecution()) { - stopRecord = this.beginWorkspaceStop(workspaceId); + stopRecord = this.taskStopLatch.beginWorkspaceStop(workspaceId); } } if (!rowStillExpected) { @@ -19144,7 +18784,7 @@ export class TaskService implements AgentTaskIntegration { } if (stopRecord != null) { - this.markWorkspaceStopPersisted(workspaceId); + this.taskStopLatch.markWorkspaceStopPersisted(workspaceId); // Phase B for the single task: clearQueue + stopStream against the captured execution; // release (and the in-memory settlement) follows the captured turn's own settlement. await this.runWorkspaceStopCleanup([workspaceId], { diff --git a/src/node/services/taskStopLatch.test.ts b/src/node/services/taskStopLatch.test.ts new file mode 100644 index 00000000000..c1b3137d41f --- /dev/null +++ b/src/node/services/taskStopLatch.test.ts @@ -0,0 +1,499 @@ +import { describe, expect, mock, test } from "bun:test"; + +import type { ProjectsConfig } from "@/node/config"; +import type { + AdmittedSend, + OwnedTaskAttempt, + SettlementReceiptDecision, + StreamEndDecision, +} from "@/node/services/taskService"; +import { TaskStopLatch, type TaskStopLatchDeps } from "@/node/services/taskStopLatch"; + +const TASK = "task-a"; +const PARENT = "parent-a"; +const ATTEMPT = "att_00000000000000a1"; + +function ownedAttempt(receiptEligible = true): OwnedTaskAttempt { + return { generation: 1, source: "test", attemptId: ATTEMPT, receiptEligible }; +} + +function send(state: AdmittedSend["state"], attemptId = ATTEMPT): AdmittedSend { + return { taskId: TASK, attemptId, state, capturedBy: new Set() } as unknown as AdmittedSend; +} + +function streamEnd(outcome: StreamEndDecision["outcome"]): StreamEndDecision { + return { attempt: undefined, attemptId: ATTEMPT, messageId: "msg-1", outcome, turn: undefined }; +} + +/** A config whose only workspace is TASK, with this execution mirror. */ +function configWithMirror(taskExecutionId: string, taskExecutionStatus: string): ProjectsConfig { + const workspace = { id: TASK, taskExecutionId, taskExecutionStatus }; + return { + projects: new Map([["/project", { workspaces: [workspace] }]]), + } as unknown as ProjectsConfig; +} + +interface HarnessState { + activeTurn: symbol | undefined; + liveHandleId: string | undefined; + activeStreamMessageId: string | undefined; + owned: OwnedTaskAttempt | undefined; + currentAttemptId: string | undefined; + sends: Set; + eligible: boolean; + streamEnd: StreamEndDecision | undefined; + decision: SettlementReceiptDecision; + write: Promise; + config: ProjectsConfig; +} + +/** Fake TaskService side: records every dep call in `events`, in call order. */ +function createHarness() { + const events: string[] = []; + const listeners = new Map void>>(); + const state: HarnessState = { + activeTurn: undefined, + liveHandleId: undefined, + activeStreamMessageId: "msg-1", + owned: undefined, + currentAttemptId: ATTEMPT, + sends: new Set(), + eligible: false, + streamEnd: undefined, + decision: { kind: "memory-only" }, + write: Promise.resolve(true), + config: { projects: new Map() }, + }; + const clearQueue = mock((id: string, _options?: { preserveUserInput?: boolean }) => { + events.push(`clearQueue:${id}`); + return { success: true as const, data: undefined }; + }); + const interruptWorkspaceTurn = mock((owner: string, handleId: string, _options?: unknown) => { + events.push(`interrupt:${owner}:${handleId}`); + return Promise.resolve({ success: true, data: undefined }); + }); + const deps: TaskStopLatchDeps = { + config: () => ({ loadConfigOrDefault: () => state.config }), + aiService: () => ({ + stopStream: mock((id: string, options?: { expectedMessageId?: string }) => { + events.push(`stopStream:${id}:${options?.expectedMessageId ?? ""}`); + return Promise.resolve({ success: true as const, data: undefined }); + }), + }), + workspaceService: () => ({ + getActiveTurnGeneration: () => state.activeTurn, + clearQueue, + }), + turnManager: () => + ({ + getLiveWorkspaceTurnRegistration: () => + state.liveHandleId != null ? { handleId: state.liveHandleId } : undefined, + captureQueueCutAttributionSnapshot: () => ({ + activeStream: + state.activeStreamMessageId != null + ? { messageId: state.activeStreamMessageId } + : undefined, + }), + interruptWorkspaceTurn, + }) as unknown as ReturnType, + terminalAttention: () => ({ + suppressTerminalAttention: (params) => { + events.push(`suppress:${params.sourceId}`); + return Promise.resolve(); + }, + }), + scheduleMaybeStartQueuedTasks: () => events.push("scheduleQueue"), + scheduleParkedPeerSendFlush: (id) => events.push(`peerFlush:${id}`), + notifyAttemptSettlementListeners: (id) => { + events.push(`notify:${id}:record=${latch.hasStopRecord(id)}`); + for (const listener of [...(listeners.get(id) ?? [])]) listener(); + }, + addAttemptSettlementListener: (id, listener) => { + const set = listeners.get(id) ?? new Set(); + set.add(listener); + listeners.set(id, set); + }, + removeAttemptSettlementListener: (id, listener) => { + listeners.get(id)?.delete(listener); + if (listeners.get(id)?.size === 0) listeners.delete(id); + }, + ownedAttempt: () => state.owned, + currentTaskAttemptId: () => state.currentAttemptId, + admittedSends: () => state.sends, + ownsReceiptEligibleAttempt: () => state.eligible, + findStreamEndDecision: () => state.streamEnd, + decideSettlementReceipt: (id) => { + events.push(`decide:${id}`); + return state.decision; + }, + writeSettlementReceipt: (id) => { + events.push(`write:${id}`); + return state.write; + }, + closeAttemptAdmission: (id, attemptId) => events.push(`close:${id}:${attemptId ?? ""}`), + settleOwnedTaskAttempt: (id, attempt) => + events.push(`settle:${id}:${attempt?.attemptId ?? "none"}`), + recheckResumeFailureSettlement: (id) => events.push(`resumeRecheck:${id}`), + }; + const latch = new TaskStopLatch(deps); + return { latch, events, listeners, state, clearQueue, interruptWorkspaceTurn }; +} + +/** Phase A + the cascade's durable marker + Phase B's cleanup paid back (no Phase B run). */ +function stopUntilOnlyTurnsHold(h: ReturnType) { + const record = h.latch.beginWorkspaceStop(TASK); + h.latch.markWorkspaceStopPersisted(TASK); + record.cleanupInFlight -= 1; + return record; +} + +describe("TaskStopLatch latches", () => { + test("overlapping latches release only when the last holder releases", () => { + const h = createHarness(); + const first = h.latch.latchWorkspaceStopsInProgress([TASK]); + const second = h.latch.latchWorkspaceStopsInProgress([TASK]); + first(); + expect(h.latch.isWorkspaceStopInProgress(TASK)).toBe(true); + expect(h.events).toEqual([]); + second(); + expect(h.latch.isWorkspaceStopInProgress(TASK)).toBe(false); + // No launch was deferred, so the queue is not rechecked. + expect(h.events).toEqual([`notify:${TASK}:record=false`, `peerFlush:${TASK}`]); + }); + + test("a deferred launch makes the clearing latch recheck the queue once, before notifying", () => { + const h = createHarness(); + expect(h.latch.deferLaunchWhileStopInProgress(TASK, PARENT)).toBe(false); + const release = h.latch.latchWorkspaceStopsInProgress([PARENT]); + expect(h.latch.deferLaunchWhileStopInProgress(TASK, PARENT)).toBe(true); + expect(h.latch.deferLaunchWhileStopInProgress(TASK, undefined)).toBe(false); + release(); + expect(h.events).toEqual([ + "scheduleQueue", + `notify:${PARENT}:record=false`, + `peerFlush:${PARENT}`, + ]); + h.events.length = 0; + h.latch.latchWorkspaceStopsInProgress([PARENT])(); + expect(h.events).not.toContain("scheduleQueue"); + }); + + test("epochs count bumps and a forgotten epoch restarts at zero", () => { + const h = createHarness(); + h.latch.bumpWorkspaceStopEpoch(TASK); + h.latch.bumpWorkspaceStopEpoch(TASK); + expect(h.latch.getWorkspaceStopEpoch(TASK)).toBe(2); + expect(h.latch.getWorkspaceStopEpoch(PARENT)).toBe(0); + h.latch.forgetStopEpoch(TASK); + expect(h.latch.getWorkspaceStopEpoch(TASK)).toBe(0); + }); +}); + +describe("TaskStopLatch.beginWorkspaceStop", () => { + test("captures the owner and only this attempt's unadmitted sends, by reference", () => { + const h = createHarness(); + const turn = Symbol("turn"); + h.state.activeTurn = turn; + h.state.liveHandleId = "wst_live"; + h.state.owned = ownedAttempt(); + const pending = send("pending"); + const enqueued = send("enqueued"); + const admitted = send("admitted"); + const otherAttempt = send("pending", "att_00000000000000b2"); + h.state.sends = new Set([pending, enqueued, admitted, otherAttempt]); + + const record = h.latch.beginWorkspaceStop(TASK); + + expect(h.latch.getWorkspaceStopEpoch(TASK)).toBe(1); + expect(h.latch.isWorkspaceStopInProgress(TASK)).toBe(true); + expect(h.latch.isCurrentStopRecord(TASK, record)).toBe(true); + expect(h.latch.stopRecordAttemptId(TASK)).toBe(ATTEMPT); + expect(record.ownedAttempt).toBe(h.state.owned); + expect([...record.capturedTurns]).toEqual([turn]); + expect(record.capturedExecutionId).toBe("wst_live"); + // The live mirror is not terminal in the (empty) config. + expect(record.executionSettled).toBe(false); + expect(record.capturedStreamMessageId).toBe("msg-1"); + expect([...record.pendingAdmissions]).toEqual([pending, enqueued]); + expect(pending.capturedBy.has(record)).toBe(true); + expect(admitted.capturedBy.size).toBe(0); + expect(otherAttempt.capturedBy.size).toBe(0); + }); + + test("falls back to the persisted attempt id without an owned attempt", () => { + const h = createHarness(); + h.state.currentAttemptId = "att_00000000000000c3"; + h.state.activeStreamMessageId = undefined; + const record = h.latch.beginWorkspaceStop(TASK); + expect(record.capturedStreamMessageId).toBeUndefined(); + expect(record.ownedAttempt).toBeUndefined(); + expect(record.attemptId).toBe("att_00000000000000c3"); + expect(record.executionSettled).toBe(true); + }); + + test("a captured execution whose mirror is already terminal counts as settled", () => { + const h = createHarness(); + h.state.liveHandleId = "wst_live"; + h.state.config = configWithMirror("wst_live", "interrupted"); + expect(h.latch.beginWorkspaceStop(TASK).executionSettled).toBe(true); + const other = createHarness(); + other.state.liveHandleId = "wst_live"; + other.state.config = configWithMirror("wst_live", "running"); + expect(other.latch.beginWorkspaceStop(TASK).executionSettled).toBe(false); + }); + + test("an overlapping cascade joins the same record and plans one more cleanup", () => { + const h = createHarness(); + const first = h.latch.beginWorkspaceStop(TASK); + const second = h.latch.beginWorkspaceStop(TASK); + expect(second).toBe(first); + expect(first.releases).toHaveLength(2); + expect(first.cleanupInFlight).toBe(2); + expect(h.latch.getWorkspaceStopEpoch(TASK)).toBe(2); + }); +}); + +describe("TaskStopLatch release", () => { + test("a settled turn releases: record dropped before the latch, unowned attempt closed", () => { + const h = createHarness(); + const turn = Symbol("turn"); + h.state.activeTurn = turn; + stopUntilOnlyTurnsHold(h); + h.latch.releaseCapturedTurn(TASK, Symbol("other")); + expect(h.latch.hasStopRecord(TASK)).toBe(true); + h.events.length = 0; + + h.latch.releaseCapturedTurn(TASK, turn); + + expect(h.latch.hasStopRecord(TASK)).toBe(false); + expect(h.latch.isWorkspaceStopInProgress(TASK)).toBe(false); + expect(h.events).toEqual([ + `decide:${TASK}`, + `notify:${TASK}:record=false`, + `peerFlush:${TASK}`, + `close:${TASK}:${ATTEMPT}`, + `settle:${TASK}:none`, + `resumeRecheck:${TASK}`, + ]); + }); + + test("an owed cleanup keeps the record until it is paid back", () => { + const h = createHarness(); + const record = h.latch.beginWorkspaceStop(TASK); + h.latch.markWorkspaceStopPersisted(TASK); + h.latch.recheckWorkspaceStopRelease(TASK); + expect(h.latch.hasStopRecord(TASK)).toBe(true); + record.cleanupInFlight -= 1; + h.latch.recheckWorkspaceStopRelease(TASK); + expect(h.latch.hasStopRecord(TASK)).toBe(false); + }); + + test("the record waits for the durable stop marker; marking it only notifies", () => { + const h = createHarness(); + const record = h.latch.beginWorkspaceStop(TASK); + record.cleanupInFlight -= 1; + h.latch.recheckWorkspaceStopRelease(TASK); + expect(h.latch.hasStopRecord(TASK)).toBe(true); + h.events.length = 0; + h.latch.markWorkspaceStopPersisted(TASK); + expect(h.events).toEqual([`notify:${TASK}:record=true`]); + h.latch.recheckWorkspaceStopRelease(TASK); + expect(h.latch.hasStopRecord(TASK)).toBe(false); + }); + + test("turn and record accessors are no-ops without a stop record", () => { + const h = createHarness(); + const turn = Symbol("turn"); + expect(h.latch.stopRecordAttemptId(TASK)).toBeUndefined(); + h.latch.releaseCapturedTurn(TASK, turn); + h.latch.rebindCapturedTurn(TASK, turn, Symbol("next")); + h.latch.markWorkspaceStopPersisted(TASK); + h.latch.releaseRetainedStopLatches(TASK); + expect(h.events).toEqual([]); + }); + + test("a superseded turn rebinds the wait to its successor", () => { + const h = createHarness(); + const previous = Symbol("previous"); + const next = Symbol("next"); + h.state.activeTurn = previous; + stopUntilOnlyTurnsHold(h); + h.latch.rebindCapturedTurn(TASK, previous, next); + h.latch.releaseCapturedTurn(TASK, previous); + expect(h.latch.hasStopRecord(TASK)).toBe(true); + h.latch.releaseCapturedTurn(TASK, next); + expect(h.latch.hasStopRecord(TASK)).toBe(false); + }); + + test("an eligible attempt without a stream-end decision does not wait for one", () => { + const h = createHarness(); + h.state.owned = ownedAttempt(); + h.state.eligible = true; + stopUntilOnlyTurnsHold(h); + h.latch.recheckWorkspaceStopRelease(TASK); + expect(h.latch.hasStopRecord(TASK)).toBe(false); + }); + + test("a pending stream-end decision holds the release while a receipt is at stake", () => { + const h = createHarness(); + h.state.owned = ownedAttempt(); + h.state.eligible = true; + h.state.streamEnd = streamEnd("pending"); + stopUntilOnlyTurnsHold(h); + h.latch.recheckWorkspaceStopRelease(TASK); + expect(h.latch.hasStopRecord(TASK)).toBe(true); + expect(h.events).not.toContain(`decide:${TASK}`); + + h.state.streamEnd = streamEnd("nonreport"); + h.latch.recheckWorkspaceStopRelease(TASK); + expect(h.latch.hasStopRecord(TASK)).toBe(false); + }); + + test("the receipt is written before the latch releases; a written receipt settles", async () => { + const h = createHarness(); + h.state.owned = ownedAttempt(); + h.state.decision = { kind: "write", attemptId: ATTEMPT, parentWorkspaceId: PARENT, owners: [] }; + const write = Promise.withResolvers(); + h.state.write = write.promise; + const record = stopUntilOnlyTurnsHold(h); + h.events.length = 0; + + h.latch.recheckWorkspaceStopRelease(TASK); + expect(record.receipt).toBe("writing"); + // A recheck while the write is in flight starts no second write. + h.latch.recheckWorkspaceStopRelease(TASK); + expect(h.events).toEqual([`decide:${TASK}`, `write:${TASK}`]); + expect(h.latch.isWorkspaceStopInProgress(TASK)).toBe(true); + + write.resolve(true); + await write.promise; + await Promise.resolve(); + await Promise.resolve(); + + expect(record.receipt).toBe("written"); + expect(h.latch.isWorkspaceStopInProgress(TASK)).toBe(false); + expect(h.events).toContain(`settle:${TASK}:${ATTEMPT}`); + // An owned attempt is settled, not closed by the stop. + expect(h.events.some((e) => e.startsWith("close:"))).toBe(false); + }); + + test.each([ + ["a failed write", "write" as const], + ["an unreadable config", "unreadable" as const], + ])("%s still releases, without settling the attempt", async (_label, kind) => { + const h = createHarness(); + h.state.owned = ownedAttempt(); + h.state.decision = + kind === "write" + ? { kind, attemptId: ATTEMPT, parentWorkspaceId: PARENT, owners: [] } + : { kind }; + h.state.write = Promise.resolve(false); + const record = stopUntilOnlyTurnsHold(h); + h.latch.recheckWorkspaceStopRelease(TASK); + for (let i = 0; i < 5; i++) await Promise.resolve(); + expect(record.receipt).toBe("failed"); + expect(h.latch.isWorkspaceStopInProgress(TASK)).toBe(false); + expect(h.events.some((e) => e.startsWith("settle:"))).toBe(false); + expect(h.events).toContain(`resumeRecheck:${TASK}`); + }); + + test("the release waits for the durable stop marker and the captured execution", () => { + const h = createHarness(); + h.state.liveHandleId = "wst_live"; + const record = h.latch.beginWorkspaceStop(TASK); + record.cleanupInFlight -= 1; + h.latch.recheckWorkspaceStopRelease(TASK); + expect(h.latch.hasStopRecord(TASK)).toBe(true); + h.latch.markWorkspaceStopPersisted(TASK); + // The same handle is still live and its mirror is not terminal: not settled. + h.latch.releaseRetainedStopLatches(TASK); + expect(record.executionSettled).toBe(false); + expect(h.latch.hasStopRecord(TASK)).toBe(true); + h.state.liveHandleId = undefined; + h.latch.releaseRetainedStopLatches(TASK); + expect(record.executionSettled).toBe(true); + expect(h.latch.hasStopRecord(TASK)).toBe(false); + }); + + test("Phase B clears the queue, stops the captured stream and pays its cleanup back", async () => { + const h = createHarness(); + h.latch.beginWorkspaceStop(TASK); + h.latch.markWorkspaceStopPersisted(TASK); + h.events.length = 0; + + const onTimeout = mock((_id: string) => undefined); + await h.latch.runWorkspaceStopCleanup([TASK], { + label: "test", + abandonPartial: true, + clearQueue: true, + onTimeout, + }); + await Promise.resolve(); + + expect(h.events.slice(0, 2)).toEqual([`clearQueue:${TASK}`, `stopStream:${TASK}:msg-1`]); + // The user's queued input is handed back, never dropped. + expect(h.clearQueue).toHaveBeenCalledWith(TASK, { preserveUserInput: true }); + expect(onTimeout).not.toHaveBeenCalled(); + expect(h.latch.hasStopRecord(TASK)).toBe(false); + expect(h.latch.isWorkspaceStopInProgress(TASK)).toBe(false); + }); + + test("Phase B interrupts a captured execution before the stream stop", async () => { + const h = createHarness(); + h.latch.beginWorkspaceStop(TASK); + h.events.length = 0; + await h.latch.runWorkspaceStopCleanup([TASK], { + label: "test", + abandonPartial: false, + clearQueue: false, + capturedExecutionsById: new Map([[TASK, { ownerWorkspaceId: PARENT, handleId: "wst_live" }]]), + }); + expect(h.events.slice(0, 3)).toEqual([ + `interrupt:${PARENT}:wst_live`, + "suppress:wst_live", + `stopStream:${TASK}:msg-1`, + ]); + // The stop drains no queue: the cascade owns what runs next. + expect(h.interruptWorkspaceTurn).toHaveBeenCalledWith(PARENT, "wst_live", { + scheduleQueueDrain: false, + }); + }); + + test("Phase B for a workspace without a record stops any stream", async () => { + const h = createHarness(); + await h.latch.runWorkspaceStopCleanup([TASK], { + label: "test", + abandonPartial: false, + clearQueue: false, + }); + expect(h.events[0]).toBe(`stopStream:${TASK}:`); + }); +}); + +describe("TaskStopLatch.waitForWorkspaceStopsToRelease", () => { + test("resolves once the latch drops and removes its listener", async () => { + const h = createHarness(); + const release = h.latch.latchWorkspaceStopsInProgress([TASK]); + let resolved = false; + const wait = h.latch + .waitForWorkspaceStopsToRelease([PARENT, TASK], Date.now() + 60_000) + .then(() => { + resolved = true; + }); + await Promise.resolve(); + expect(resolved).toBe(false); + expect(h.listeners.get(TASK)?.size).toBe(1); + release(); + await wait; + expect(resolved).toBe(true); + expect(h.listeners.has(TASK)).toBe(false); + }); + + test("returns at the deadline while the latch is still held", async () => { + const h = createHarness(); + h.latch.latchWorkspaceStopsInProgress([TASK]); + await h.latch.waitForWorkspaceStopsToRelease([TASK], Date.now() - 1); + expect(h.latch.isWorkspaceStopInProgress(TASK)).toBe(true); + expect(h.listeners.has(TASK)).toBe(false); + }); +}); diff --git a/src/node/services/taskStopLatch.ts b/src/node/services/taskStopLatch.ts new file mode 100644 index 00000000000..f1ff3af3991 --- /dev/null +++ b/src/node/services/taskStopLatch.ts @@ -0,0 +1,567 @@ +import type { Config } from "@/node/config"; +import { + TASK_TERMINATION_STOP_STREAM_AGGREGATE_TIMEOUT_MS, + TASK_TERMINATION_STOP_STREAM_TIMEOUT_MS, +} from "@/constants/terminationTimeouts"; +import type { AIService } from "@/node/services/aiService"; +import { log } from "@/node/services/log"; +import { isActiveWorkspaceTurnTaskStatus } from "@/node/services/taskHandleStore"; +import type { SubagentAttemptSettlementSource } from "@/node/services/subagentAttemptSettlements"; +import type { + AdmittedSend, + OwnedTaskAttempt, + SettlementReceiptDecision, + StreamEndDecision, + WorkspaceStopRecord, +} from "@/node/services/taskService"; +import { findWorkspaceEntry } from "@/node/services/taskUtils"; +import type { WorkspaceHost } from "@/node/services/taskWorkspaceSeam"; +import type { TerminalAttentionNotifier } from "@/node/services/terminalAttentionNotifier"; +import type { WorkspaceTurnManager } from "@/node/services/workspaceTurnManager"; +import { raceWithAbortAndTimeout } from "@/node/utils/concurrency/withTimeout"; + +/** + * Everything the stop latch reads from TaskService. Each entry is a call-time closure over + * TaskService, never a captured value or bound method: tests swap TaskService fields and patch + * its methods after construction. Attempt, send and stream-end state stays on TaskService and is + * reached only through these closures. + */ +export interface TaskStopLatchDeps { + config: () => Pick; + aiService: () => Pick; + workspaceService: () => Pick; + turnManager: () => Pick< + WorkspaceTurnManager, + | "getLiveWorkspaceTurnRegistration" + | "captureQueueCutAttributionSnapshot" + | "interruptWorkspaceTurn" + >; + terminalAttention: () => Pick; + scheduleMaybeStartQueuedTasks: () => void; + /** PeerSendParking.scheduleParkedPeerSendFlush. */ + scheduleParkedPeerSendFlush: (workspaceId: string) => void; + notifyAttemptSettlementListeners: (taskId: string) => void; + addAttemptSettlementListener: (taskId: string, listener: () => void) => void; + removeAttemptSettlementListener: (taskId: string, listener: () => void) => void; + /** This process's owned attempt for the task (TaskService.ownedAttemptByTaskId). */ + ownedAttempt: (taskId: string) => OwnedTaskAttempt | undefined; + currentTaskAttemptId: (taskId: string) => string | undefined; + /** The task's live set of send obligations; its sends are captured by reference. */ + admittedSends: (taskId: string) => ReadonlySet | undefined; + ownsReceiptEligibleAttempt: (taskId: string, attempt: OwnedTaskAttempt | undefined) => boolean; + findStreamEndDecision: (taskId: string, attemptId: string) => StreamEndDecision | undefined; + decideSettlementReceipt: ( + taskId: string, + attempt: OwnedTaskAttempt | undefined, + settlementSource: string + ) => SettlementReceiptDecision; + writeSettlementReceipt: ( + taskId: string, + target: Extract, + source: SubagentAttemptSettlementSource + ) => Promise; + closeAttemptAdmission: ( + taskId: string, + attemptId: string | undefined, + ownedAttempt: OwnedTaskAttempt | undefined, + source: string + ) => void; + settleOwnedTaskAttempt: ( + taskId: string, + attempt: OwnedTaskAttempt | undefined, + source: string + ) => void; + recheckResumeFailureSettlement: (taskId: string) => void; +} + +/** + * The stop record's release conditions hold: no cleanup owed, a durable stop marker, no captured + * turn or pending admission left, and the captured execution settled. + */ +function isWorkspaceStopRecordSettled(record: WorkspaceStopRecord): boolean { + if (record.cleanupInFlight > 0) return false; + return ( + record.stopPersisted && + record.capturedTurns.size === 0 && + record.pendingAdmissions.size === 0 && + record.executionSettled + ); +} + +/** + * Stop latch state: the per-workspace stop epochs, the refcounted stop-in-progress latches, the + * queue rechecks a held latch owes and the stop records that decide when a latch may drop + * (Phase A capture, Phase B cleanup, Phase C release). The attempt settlement a release records + * stays on TaskService and runs through the deps. + */ +export class TaskStopLatch { + /** + * Monotonic per-workspace stop generation, bumped synchronously at every explicit stop + * boundary (user Stop suppression, task_stop interruption, workspace-turn interruption). + * Peer-send admission captures its endpoints' generations and refuses when any changed: + * unlike the level-triggered suppression set and persisted statuses — which a quick user + * resume clears between probe evaluations — a bump latches, so an intervening stop keeps + * the in-flight send stale forever. + */ + readonly workspaceStopEpochs = new Map(); + /** + * Level-triggered stop latches: workspace IDs whose stop cascade is currently between its + * synchronous epoch bump and terminal status persistence. The epoch map alone cannot refuse + * a send that ENTERS during that window — the post-bump generation becomes the send's clean + * baseline while the endpoints' persisted statuses still read running — so peer-send + * admission also rejects any endpoint chain holding an in-progress stop latch. Refcounted: + * overlapping stops (a subtree stop racing a workspace-turn interrupt on the same workspace) + * must not clear each other's latch. + */ + readonly workspaceStopsInProgress = new Map(); + /** + * Latched workspace IDs whose held latch deferred a queued-task launch (own or parent latch; + * see startReservedAgentTask). The launch record was handed back to "queued", and the queue is + * rechecked exactly when that latch clears — not on every release (startup normalizes stopped + * trees before any drain) and not on an unrelated scheduler trigger. + */ + private readonly queueRechecksOwedByStopLatch = new Set(); + /** Stop latches retained past their cascade because the stop could not be confirmed; released on authoritative terminal settlement. */ + /** + * Ownership of every in-progress stop, keyed by workspace (see beginWorkspaceStop). The latch + * a record guards drops only when BOTH hold: the captured owner settled authoritatively + * (admitted turn generation ended; matching execution mirror settled) AND no cleanup issued + * by the cascade is still in flight. Persisted statuses, stop results, and missing + * registrations are not evidence — an accepted-but-preparing turn can still start. + */ + readonly workspaceStopRecords = new Map(); + + constructor(private readonly deps: TaskStopLatchDeps) {} + + bumpWorkspaceStopEpoch(workspaceId: string): void { + this.workspaceStopEpochs.set(workspaceId, (this.workspaceStopEpochs.get(workspaceId) ?? 0) + 1); + } + + getWorkspaceStopEpoch(workspaceId: string): number { + return this.workspaceStopEpochs.get(workspaceId) ?? 0; + } + + /** + * Hold a stop-in-progress latch for the given workspaces until the returned release runs. + * Callers latch synchronously with the epoch bump and release only after the cascade has + * persisted terminal state, so sends entering mid-stop are refused at admission. + */ + latchWorkspaceStopsInProgress(workspaceIds: readonly string[]): () => void { + for (const id of workspaceIds) { + this.workspaceStopsInProgress.set(id, (this.workspaceStopsInProgress.get(id) ?? 0) + 1); + } + return () => { + for (const id of workspaceIds) { + const count = this.workspaceStopsInProgress.get(id) ?? 0; + if (count <= 1) { + this.workspaceStopsInProgress.delete(id); + if (this.queueRechecksOwedByStopLatch.delete(id)) { + this.deps.scheduleMaybeStartQueuedTasks(); + } + this.deps.notifyAttemptSettlementListeners(id); + // An interrupted delegated turn releases its registration while this latch is held. + this.deps.scheduleParkedPeerSendFlush(id); + } else { + this.workspaceStopsInProgress.set(id, count - 1); + } + } + }; + } + + isWorkspaceStopInProgress(workspaceId: string): boolean { + return this.workspaceStopsInProgress.has(workspaceId); + } + + /** + * Stop-cascade launch barrier: true (and the queue recheck owed to the held latch recorded) + * while the task's own or its parent's stop latch is held. The record stays/returns to + * "queued"; the latch release rechecks the queue (see queueRechecksOwedByStopLatch). + */ + deferLaunchWhileStopInProgress(taskId: string, parentWorkspaceId: string | undefined): boolean { + let deferred = false; + for (const id of [taskId, parentWorkspaceId]) { + if (id != null && this.isWorkspaceStopInProgress(id)) { + this.queueRechecksOwedByStopLatch.add(id); + deferred = true; + } + } + return deferred; + } + + /** + * Phase A of a stop cascade (synchronous, under the global mutex): bump the epoch, latch the + * workspace and record what must settle before the latch may drop — the admitted turn + * generation (owner of any preparing/streaming work) and the live execution mirror. Nothing + * admitted after this point exists (admission barrier), so these captures are complete. + * Overlapping cascades add their release to the same record instead of resetting ownership. + */ + beginWorkspaceStop(workspaceId: string): WorkspaceStopRecord { + this.bumpWorkspaceStopEpoch(workspaceId); + const release = this.latchWorkspaceStopsInProgress([workspaceId]); + const existing = this.workspaceStopRecords.get(workspaceId); + if (existing != null) { + // An overlapping cascade adds its release and plans its own cleanup; ownership of the + // earlier cascade's counts is untouched. + existing.releases.push(release); + existing.cleanupInFlight += 1; + return existing; + } + const capturedTurn = this.deps.workspaceService().getActiveTurnGeneration(workspaceId); + const capturedExecutionId = this.deps + .turnManager() + .getLiveWorkspaceTurnRegistration(workspaceId)?.handleId; + const ownedAttempt = this.deps.ownedAttempt(workspaceId); + const attemptId = ownedAttempt?.attemptId ?? this.deps.currentTaskAttemptId(workspaceId); + const record: WorkspaceStopRecord = { + releases: [release], + cleanupInFlight: 1, + attemptId, + capturedTurns: new Set(capturedTurn != null ? [capturedTurn] : []), + pendingAdmissions: new Set(), + capturedExecutionId, + // A registration whose mirror already persisted terminally is settled (its settlement + // callback ran before this capture and will not run again). + executionSettled: + capturedExecutionId == null || + this.isExecutionMirrorSettled(workspaceId, capturedExecutionId), + stopPersisted: false, + ownedAttempt, + capturedStreamMessageId: this.deps + .turnManager() + .captureQueueCutAttributionSnapshot(workspaceId).activeStream?.messageId, + }; + // Sends admitted against this attempt that have not claimed a turn yet are part of what + // must settle: captured by reference so a later obligation of a successor attempt (or one + // already superseded) can never enter this record. + for (const send of this.deps.admittedSends(workspaceId) ?? []) { + if (send.attemptId !== attemptId) continue; + if (send.state !== "pending" && send.state !== "enqueued") continue; + record.pendingAdmissions.add(send); + send.capturedBy.add(record); + } + this.workspaceStopRecords.set(workspaceId, record); + return record; + } + + /** + * Phase B (outside the global mutex): every descendant's clearQueue + stopStream, issued + * exactly once while its latch is held and targeted at the execution captured in Phase A, raced + * concurrently under a per-child and one aggregate deadline. A timeout abandons the WAIT only: + * the original promise keeps its cleanup ownership and its later settlement rechecks release. + * Must run for EVERY id latched in Phase A (each planned one cleanup), so callers pass the + * full latched set even when Phase A failed midway. + */ + async runWorkspaceStopCleanup( + targets: readonly string[], + options: { + label: string; + abandonPartial: boolean; + clearQueue: boolean; + /** + * Live continuation handles captured in Phase A (see WorkspaceStopRecord.capturedExecutionId), + * interrupted here so their mirror settles: the stream stop below is a `system` abort, which + * never settles a workspace-turn handle (only a user abort does), so without this the record + * would wait for an execution settlement that never comes and hold the latch until restart. + */ + capturedExecutionsById?: ReadonlyMap; + onTimeout?: (workspaceId: string) => void; + } + ): Promise { + const waits = targets.map(async (id) => { + // Consumes the unit planned by beginWorkspaceStop; the finally below pays it back. + const record = this.workspaceStopRecords.get(id); + const expectedMessageId = record?.capturedStreamMessageId; + const cleanup = (async () => { + if (options.clearQueue) { + // AgentSession stream-end cleanup auto-flushes queued messages, so a stopped + // descendant must not keep pending input; issued once, never re-issued later. The + // user's queued messages are handed back as held input, never silently dropped. + try { + const clearQueueResult = this.deps.workspaceService().clearQueue(id, { + preserveUserInput: true, + }); + if (!clearQueueResult.success) { + log.debug(`${options.label}: clearQueue failed`, { + taskId: id, + error: clearQueueResult.error, + }); + } + } catch (error: unknown) { + log.debug(`${options.label}: clearQueue threw`, { taskId: id, error }); + } + } + const execution = options.capturedExecutionsById?.get(id); + if (execution != null) { + await this.interruptCapturedExecution(id, execution, options.label); + } + // Success is NOT stop confirmation (an accepted-but-PREPARING turn has no registered + // stream yet); only the owner's settlement recorded on the stop record confirms. + const stopResult = await this.deps.aiService().stopStream(id, { + abandonPartial: options.abandonPartial, + ...(expectedMessageId != null ? { expectedMessageId } : {}), + }); + if (!stopResult.success) { + log.debug(`${options.label}: stopStream failed`, { taskId: id }); + } + })(); + // Ownership stays with the original promise: only ITS settlement (however late) counts + // the cleanup as finished, and even then it merely triggers a recheck. + void cleanup + .catch((error: unknown) => { + log.debug(`${options.label}: cleanup threw`, { taskId: id, error }); + }) + .finally(() => { + if (record != null) record.cleanupInFlight -= 1; + this.recheckWorkspaceStopRelease(id); + }); + const outcome = await raceWithAbortAndTimeout(cleanup, { + timeoutMs: TASK_TERMINATION_STOP_STREAM_TIMEOUT_MS, + }); + if (outcome.kind !== "ok") { + log.warn(`${options.label}: stopStream timed out; retaining stop latch`, { taskId: id }); + options.onTimeout?.(id); + } + }); + const aggregate = await raceWithAbortAndTimeout(Promise.allSettled(waits), { + timeoutMs: TASK_TERMINATION_STOP_STREAM_AGGREGATE_TIMEOUT_MS, + }); + if (aggregate.kind !== "ok") { + log.warn(`${options.label}: cleanup exceeded the aggregate deadline`, { + taskIds: targets, + }); + } + } + + /** + * Settle the continuation handle a stop record captured (a reawakened child executes under a + * WorkspaceTurnManager handle): interruptWorkspaceTurn persists the handle and its execution + * mirror terminal within one settlement boundary and drops the live registration, which is the + * authoritative execution settlement the record waits on (releaseRetainedStopLatches). Same + * pairing as the task_stop subtree stop (finishSubtreeStopCleanup): the owner's terminal wake + * for a handle the stop itself interrupted is suppressed. A failure leaves the record waiting + * (fail closed), exactly as an unsettled execution does today. + */ + private async interruptCapturedExecution( + taskId: string, + execution: { ownerWorkspaceId: string; handleId: string }, + label: string + ): Promise { + const interrupted = await this.deps + .turnManager() + .interruptWorkspaceTurn(execution.ownerWorkspaceId, execution.handleId, { + scheduleQueueDrain: false, + }); + if (!interrupted.success) { + log.warn(`${label}: interruptWorkspaceTurn failed for the captured execution`, { + taskId, + handleId: execution.handleId, + error: interrupted.error, + }); + return; + } + await this.deps.terminalAttention().suppressTerminalAttention({ + ownerWorkspaceId: execution.ownerWorkspaceId, + sourceKind: "workspace_turn", + sourceId: execution.handleId, + }); + } + + /** + * Phase C: drop the latch iff the captured owner settled, a durable stop marker exists and no + * cleanup is owed. Called from every settlement path; a late cleanup completion or a + * persisted-status caller only triggers this recheck and never supplies evidence by itself. + * A missing config entry is NOT evidence: a still-preparing session settles through its own + * disposal (the coordinator publishes the captured generation's idle transition). + */ + recheckWorkspaceStopRelease(workspaceId: string): void { + const record = this.workspaceStopRecords.get(workspaceId); + if (record == null) return; + if (!isWorkspaceStopRecordSettled(record)) return; + // Execution-settlement producer: the receipt is durable before the latch releases, so nothing + // the latch holds back (a reawaken, a peer send, a queued launch) observes the stopped attempt + // as released while its receipt could still be missing. The record stays registered (latched) + // for the write; its completion rechecks, releasing the latch and recording the settlement in + // one step as before. A failed write (or an unreadable config) still releases, with the + // attempt left `closing` (fail closed) instead of pinning the latch until restart. + if (record.receipt === "writing") return; + if (this.isStopReceiptDecisionPending(workspaceId, record)) return; + if (record.receipt == null && this.startStopSettlementReceipt(workspaceId, record)) return; + this.releaseWorkspaceStopRecord(workspaceId, record); + } + + /** + * A receipt can be at stake: wait for the attempt's pending stream-end decision (its report + * may still publish); resolveStreamEndDecision rechecks. Without a receipt at stake the + * release keeps its previous timing. + */ + private isStopReceiptDecisionPending(workspaceId: string, record: WorkspaceStopRecord): boolean { + return ( + record.receipt == null && + record.attemptId != null && + this.deps.ownsReceiptEligibleAttempt(workspaceId, record.ownedAttempt) && + this.deps.findStreamEndDecision(workspaceId, record.attemptId)?.outcome === "pending" + ); + } + + /** + * Decide the stopped attempt's settlement receipt and start its write. Returns true while the + * write is in flight (the release waits for its completion recheck). An unreadable config marks + * the receipt failed and lets the release go ahead. + */ + private startStopSettlementReceipt(workspaceId: string, record: WorkspaceStopRecord): boolean { + const decision = this.deps.decideSettlementReceipt( + workspaceId, + record.ownedAttempt, + "stop-settled" + ); + if (decision.kind === "unreadable") record.receipt = "failed"; + if (decision.kind !== "write") return false; + record.receipt = "writing"; + void this.deps + .writeSettlementReceipt(workspaceId, decision, "execution-settled") + .then( + (written) => { + record.receipt = written ? "written" : "failed"; + }, + (error: unknown) => { + log.error("Stop release: settlement receipt write threw", { workspaceId, error }); + record.receipt = "failed"; + } + ) + .finally(() => this.recheckWorkspaceStopRelease(workspaceId)); + return true; + } + + /** Release step of recheckWorkspaceStopRelease: drop the latch and settle the stopped attempt. */ + private releaseWorkspaceStopRecord(workspaceId: string, record: WorkspaceStopRecord): void { + this.workspaceStopRecords.delete(workspaceId); + for (const release of record.releases) { + release(); + } + // Owner settled + cleanup finished + marker persisted: authoritative for the attempt captured + // in Phase A. An unknown (legacy) attempt has none to settle — stopping it proves nothing, + // but its id is closed to further sends until a new admission rotates it. + if (record.ownedAttempt == null && record.attemptId != null) { + this.deps.closeAttemptAdmission(workspaceId, record.attemptId, undefined, "stop-settled"); + } + // A failed receipt write leaves the attempt `closing` (never settled without its receipt). + if (record.receipt !== "failed") { + this.deps.settleOwnedTaskAttempt(workspaceId, record.ownedAttempt, "stop-settled"); + } + // A failed reawaken's settlement deferred to this Stop (#4310): drop or finish it now; no + // later disposal or turn settlement is guaranteed to recheck it. + this.deps.recheckResumeFailureSettlement(workspaceId); + } + + /** + * Authoritative execution settlement (WorkspaceTurnManager settled the MATCHING execution + * mirror and removed its live registration). Records the evidence for the captured execution + * and rechecks; persisted interrupted statuses from other callers reach only the recheck. + */ + releaseRetainedStopLatches(workspaceId: string): void { + const record = this.workspaceStopRecords.get(workspaceId); + if (record == null) return; + const liveHandleId = this.deps + .turnManager() + .getLiveWorkspaceTurnRegistration(workspaceId)?.handleId; + if ( + record.capturedExecutionId != null && + (liveHandleId !== record.capturedExecutionId || + this.isExecutionMirrorSettled(workspaceId, record.capturedExecutionId)) + ) { + record.executionSettled = true; + } + this.recheckWorkspaceStopRelease(workspaceId); + } + + /** The persisted execution mirror for this handle is terminal (settled by its owner). */ + private isExecutionMirrorSettled(workspaceId: string, handleId: string): boolean { + const workspace = findWorkspaceEntry( + this.deps.config().loadConfigOrDefault(), + workspaceId + )?.workspace; + return ( + workspace?.taskExecutionId === handleId && + !isActiveWorkspaceTurnTaskStatus(workspace.taskExecutionStatus) + ); + } + + /** Phase A recorded a durable stop marker for this workspace (see WorkspaceStopRecord). */ + markWorkspaceStopPersisted(workspaceId: string): void { + const record = this.workspaceStopRecords.get(workspaceId); + if (record == null) return; + record.stopPersisted = true; + // A deferred child stream end cancels on this notification (the stop is in progress), so its + // pending decision cannot hold the latch. Every caller runs after its durable status write, or + // where no status is left to write (a removed row, or an already-terminal status that is the + // marker), on every branch, including those that skip recordTaskInterrupted: a terminate whose + // removal keeps the workspace (#6920), or a status already interrupted by an archive (#6967). + this.deps.notifyAttemptSettlementListeners(workspaceId); + } + + /** + * Resolve once none of `workspaceIds` holds a Stop latch, or at `deadlineMs`. A Stop releases + * its latch only after its settlement receipt is durable or failed (recheckWorkspaceStopRelease), + * and the release notifies the attempt settlement listeners. + */ + async waitForWorkspaceStopsToRelease( + workspaceIds: readonly string[], + deadlineMs: number + ): Promise { + for (const id of workspaceIds) { + while (this.isWorkspaceStopInProgress(id)) { + const remainingMs = deadlineMs - Date.now(); + if (remainingMs <= 0) { + log.warn("Stop latch still held at the cleanup deadline; its receipt may be missing", { + taskId: id, + }); + return; + } + const released = Promise.withResolvers(); + const listener = () => released.resolve(); + this.deps.addAttemptSettlementListener(id, listener); + try { + await raceWithAbortAndTimeout(released.promise, { timeoutMs: remainingMs }); + } finally { + this.deps.removeAttemptSettlementListener(id, listener); + } + } + } + } + + /** noteWorkspaceRemoved: drop the removed workspace's stop generation (#5334). */ + forgetStopEpoch(workspaceId: string): void { + this.workspaceStopEpochs.delete(workspaceId); + } + + /** A stop cascade still holds a record for this workspace. */ + hasStopRecord(workspaceId: string): boolean { + return this.workspaceStopRecords.has(workspaceId); + } + + /** `record` is still the workspace's registered stop record (not yet released). */ + isCurrentStopRecord(workspaceId: string, record: WorkspaceStopRecord): boolean { + return this.workspaceStopRecords.get(workspaceId) === record; + } + + /** The attempt the workspace's registered stop record stops; undefined without a record. */ + stopRecordAttemptId(workspaceId: string): string | undefined { + return this.workspaceStopRecords.get(workspaceId)?.attemptId; + } + + /** TaskService.recordWorkspaceTurnSettled: release the settled turn from the stop record. */ + releaseCapturedTurn(workspaceId: string, turnGeneration: symbol): void { + const record = this.workspaceStopRecords.get(workspaceId); + if (!record?.capturedTurns.has(turnGeneration)) return; + record.capturedTurns.delete(turnGeneration); + this.recheckWorkspaceStopRelease(workspaceId); + } + + /** TaskService.recordWorkspaceTurnSuperseded: the stop record waits on the successor turn. */ + rebindCapturedTurn(workspaceId: string, previous: symbol, next: symbol): void { + const record = this.workspaceStopRecords.get(workspaceId); + if (!record?.capturedTurns.has(previous)) return; + record.capturedTurns.delete(previous); + record.capturedTurns.add(next); + } +} diff --git a/src/node/services/workspaceService.removalCascade.test.ts b/src/node/services/workspaceService.removalCascade.test.ts index 179d40573e9..1290cf95669 100644 --- a/src/node/services/workspaceService.removalCascade.test.ts +++ b/src/node/services/workspaceService.removalCascade.test.ts @@ -191,8 +191,9 @@ describe("parent removal cascades over its sub-agent tree across two backends", const userStopEpochs = ( a.taskService as unknown as { workspaceUserStopEpochs: Map } ).workspaceUserStopEpochs; - const stopEpochs = (a.taskService as unknown as { workspaceStopEpochs: Map }) - .workspaceStopEpochs; + const stopEpochs = ( + a.taskService as unknown as { taskStopLatch: { workspaceStopEpochs: Map } } + ).taskStopLatch.workspaceStopEpochs; expect([...userStopEpochs.keys()].sort()).toEqual([...allIds].sort()); expect([...stopEpochs.keys()].sort()).toEqual([...allIds].sort());