From c64d9f8440669ef74b61c475be20c4625a0d1d60 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Tue, 27 Oct 2026 07:20:59 +0000 Subject: [PATCH 1/2] =?UTF-8?q?=F0=9F=A4=96=20refactor:=20move=20workflow?= =?UTF-8?q?=20run=20attention=20queue=20into=20WorkflowRunAttention?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move-only. The in-memory workflow terminal wake queue (pendingWorkflowRunAttention), the per-run settlement chain (workflowRunSettlementByRun) and their methods (queueWorkflowRunAttention, markWorkflowRunTerminalAttentionSettled, settleWorkflowRunTerminalAttention, clearWorkflowRunDowngradeSettlement) move into a plain WorkflowRunAttention class. TaskService builds it with call-time closures for sessionsDir and terminalAttentionStore, keeps the two public methods as one-line facades, and the drain reaches the queue through queuedRunIds/dropOwner/dropRun. Tests that seed the queue through a cast now point at workflowRunAttention.pendingWorkflowRunAttention. The new class gets its own unit tests with a fake store. --- .../taskService.shared.testHarness.ts | 6 +- src/node/services/taskService.test.ts | 48 ++- src/node/services/taskService.ts | 192 +---------- .../services/workflowRunAttention.test.ts | 325 ++++++++++++++++++ src/node/services/workflowRunAttention.ts | 225 ++++++++++++ 5 files changed, 605 insertions(+), 191 deletions(-) create mode 100644 src/node/services/workflowRunAttention.test.ts create mode 100644 src/node/services/workflowRunAttention.ts diff --git a/src/node/services/taskService.shared.testHarness.ts b/src/node/services/taskService.shared.testHarness.ts index 16fa5379ddb..81a0fc6ea6e 100644 --- a/src/node/services/taskService.shared.testHarness.ts +++ b/src/node/services/taskService.shared.testHarness.ts @@ -460,8 +460,10 @@ export function queuedWorkflowRunAttention( ownerWorkspaceId: string ): ReadonlySet { const queue = ( - taskService as unknown as { pendingWorkflowRunAttention: Map> } - ).pendingWorkflowRunAttention; + taskService as unknown as { + workflowRunAttention: { pendingWorkflowRunAttention: Map> }; + } + ).workflowRunAttention.pendingWorkflowRunAttention; return new Set(queue.get(ownerWorkspaceId)); } diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index d46897d120c..b78eb2a57ea 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -1479,8 +1479,10 @@ describe("TaskService", () => { const sweep = () => sweepOwnerWorkflowRunAttention(taskService, parentId); ( - taskService as unknown as { pendingWorkflowRunAttention: Map> } - ).pendingWorkflowRunAttention.set(parentId, new Set([runId])); + taskService as unknown as { + workflowRunAttention: { pendingWorkflowRunAttention: Map> }; + } + ).workflowRunAttention.pendingWorkflowRunAttention.set(parentId, new Set([runId])); await drain(parentId); ownerBusy = false; @@ -1810,8 +1812,10 @@ describe("TaskService", () => { // Residual private seed: noteWorkflowRunTerminalAttention would also start a drain that // delivers and settles this run itself, racing the two settlements under test. ( - taskService as unknown as { pendingWorkflowRunAttention: Map> } - ).pendingWorkflowRunAttention.set(parentId, new Set([runId])); + taskService as unknown as { + workflowRunAttention: { pendingWorkflowRunAttention: Map> }; + } + ).workflowRunAttention.pendingWorkflowRunAttention.set(parentId, new Set([runId])); // Park the older generation's settlement inside its first marker write: the newer // generation's settlement (started while the older one is parked) can then only @@ -2069,8 +2073,13 @@ describe("TaskService", () => { // Residual private seed: noteWorkflowRunTerminalAttention would also start a drain that // evaluates this entry concurrently with the settlement calls under test. ( - taskService as unknown as { pendingWorkflowRunAttention: Map> } - ).pendingWorkflowRunAttention.set(parentId, new Set(["wfr_marker_soft_fail"])); + taskService as unknown as { + workflowRunAttention: { pendingWorkflowRunAttention: Map> }; + } + ).workflowRunAttention.pendingWorkflowRunAttention.set( + parentId, + new Set(["wfr_marker_soft_fail"]) + ); const settleSpy = spyOn(terminalAttentionStore, "recordSettled") // Lazy rejection: an eager mockRejectedValueOnce promise trips bun's unhandled-rejection // detector on this host before the call consumes it. @@ -2295,8 +2304,10 @@ describe("TaskService", () => { // Residual private seed so ONE drain observes both runs: a second note would schedule a // second drain, giving the older group another chance outside the cycle under test. ( - taskService as unknown as { pendingWorkflowRunAttention: Map> } - ).pendingWorkflowRunAttention.set(parentId, new Set([oldRunId, newRunId])); + taskService as unknown as { + workflowRunAttention: { pendingWorkflowRunAttention: Map> }; + } + ).workflowRunAttention.pendingWorkflowRunAttention.set(parentId, new Set([oldRunId, newRunId])); taskService.scheduleTerminalAttentionDrain(parentId); await flushTerminalAttentionDrains(taskService); @@ -2421,8 +2432,10 @@ describe("TaskService", () => { // Residual private seed so ONE drain observes both runs: a second note would schedule a // second drain that could deliver the older group even if the same-cycle re-poke broke. ( - taskService as unknown as { pendingWorkflowRunAttention: Map> } - ).pendingWorkflowRunAttention.set(parentId, new Set([oldRunId, newRunId])); + taskService as unknown as { + workflowRunAttention: { pendingWorkflowRunAttention: Map> }; + } + ).workflowRunAttention.pendingWorkflowRunAttention.set(parentId, new Set([oldRunId, newRunId])); taskService.scheduleTerminalAttentionDrain(parentId); await flushTerminalAttentionDrains(taskService); @@ -3031,8 +3044,13 @@ describe("TaskService", () => { // and the second drain would deliver the other group before the assertions below. const terminalAttentionStore = new TerminalAttentionStore(config); ( - taskService as unknown as { pendingWorkflowRunAttention: Map> } - ).pendingWorkflowRunAttention.set(parentId, new Set(["wfr_split_plan", "wfr_split_exec"])); + taskService as unknown as { + workflowRunAttention: { pendingWorkflowRunAttention: Map> }; + } + ).workflowRunAttention.pendingWorkflowRunAttention.set( + parentId, + new Set(["wfr_split_plan", "wfr_split_exec"]) + ); taskService.scheduleTerminalAttentionDrain(parentId); await flushTerminalAttentionDrains(taskService); @@ -3116,8 +3134,10 @@ describe("TaskService", () => { const terminalAttentionStore = new TerminalAttentionStore(config); // Residual private seed so ONE drain observes both runs (see the test above). ( - taskService as unknown as { pendingWorkflowRunAttention: Map> } - ).pendingWorkflowRunAttention.set( + taskService as unknown as { + workflowRunAttention: { pendingWorkflowRunAttention: Map> }; + } + ).workflowRunAttention.pendingWorkflowRunAttention.set( parentId, new Set(["wfr_pin_split_pinned", "wfr_pin_split_unpinned"]) ); diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index e5a01e9b78d..cf7d5de1f50 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -344,6 +344,7 @@ import { type TerminalAttentionNotification, type TerminalAttentionOutcome, } from "@/node/services/terminalAttentionStore"; +import { WorkflowRunAttention } from "@/node/services/workflowRunAttention"; import { readAgentWorkflowRunReferences } from "@/node/services/agentWorkflowRunReferences"; import type { AgentWorkflowRunStrictPin } from "@/node/services/agentWorkflowRunReferences"; import { normalizeWorkflowAgentReportPayloadForHostSchema } from "@/common/utils/tools/workflowReportPayload"; @@ -1970,16 +1971,8 @@ export class TaskService implements AgentTaskIntegration { // the whole map, so each report can cut at most one busy turn, and only the turn that was live // at its delivery (see cutBusyOwnerTurnForSubagentAttention). private readonly freshSubagentAttentionTurnByOwner = new Map>(); - // Terminal settlements of the same run must not overlap: settlement is multi-step (stable - // refresh, generation marker, post-write mismatch delete), so an older generation reaching - // its mismatch delete after a newer settlement's stable refresh would remove the newer - // generation's valid marker and let a downgraded build re-deliver a consumed result. - private readonly workflowRunSettlementByRun = new Map>(); - // Owed workflow terminal wakes (owner -> runIds believed terminal and not yet settled). An - // in-memory work queue over durable state, not a delivery record: entries are (re)derived - // from run records + settled markers at startup and on the periodic sweep and added by live - // terminal callbacks, so losing the map merely delays a wake until the next sweep. - private readonly pendingWorkflowRunAttention = new Map>(); + // Owed workflow terminal wakes and the per-run settlement chain (see WorkflowRunAttention). + private readonly workflowRunAttention: WorkflowRunAttention; // Owner -> wake-group key -> epoch ms before which the drain skips selecting the group. A // group whose send was rejected for group-specific reasons (for example an unresolvable // strictly pinned agent) would otherwise be re-selected newest-first by every drain and @@ -4836,6 +4829,10 @@ export class TaskService implements AgentTaskIntegration { (this.parkedPeerSendsByTarget.get(targetId)?.length ?? 0), }); this.terminalAttentionStore = terminalAttentionStore ?? new TerminalAttentionStore(config); + this.workflowRunAttention = new WorkflowRunAttention({ + sessionsDir: () => this.config.sessionsDir, + terminalAttentionStore: () => this.terminalAttentionStore, + }); this.gitPatchArtifactService = new GitPatchArtifactService(config, historyService.locks); this.aiService.on("stream-end", (payload: unknown) => { @@ -12264,7 +12261,7 @@ export class TaskService implements AgentTaskIntegration { continue; } } - if (this.queueWorkflowRunAttention(workspace.id, run.id)) { + if (this.workflowRunAttention.queueWorkflowRunAttention(workspace.id, run.id)) { queuedCount += 1; } queuedForWorkspace = true; @@ -12303,7 +12300,7 @@ export class TaskService implements AgentTaskIntegration { if (!isTerminalWorkflowRunStatus(params.status)) { return; } - this.queueWorkflowRunAttention(params.ownerWorkspaceId, params.runId); + this.workflowRunAttention.queueWorkflowRunAttention(params.ownerWorkspaceId, params.runId); this.scheduleTerminalAttentionDrain(params.ownerWorkspaceId); } @@ -12323,20 +12320,6 @@ export class TaskService implements AgentTaskIntegration { return ownerWorkspaceIds.length; } - /** Returns true when the run was newly queued for this owner. */ - private queueWorkflowRunAttention(ownerWorkspaceId: string, runId: string): boolean { - let runIds = this.pendingWorkflowRunAttention.get(ownerWorkspaceId); - if (runIds == null) { - runIds = new Set(); - this.pendingWorkflowRunAttention.set(ownerWorkspaceId, runIds); - } - if (runIds.has(runId)) { - return false; - } - runIds.add(runId); - return true; - } - /** * Tool-path access to the invocation-boundary snapshot recorded into the * agent-workflow-runs sidecar (see recordBackgroundWorkflowRunReference). @@ -12348,13 +12331,7 @@ export class TaskService implements AgentTaskIntegration { return this.workspaceService.getWorkflowInvocationBoundaryMessageId(workspaceId, runId); } - /** - * Durable "this terminal generation needs no further wake" marker, keyed by the run's - * terminal updatedAt: a later resume produces a new generation and thereby re-arms - * attention without any reset bookkeeping. Write-once and best-effort by design: if the - * write fails or never happens, the next drain evaluation re-derives the same answer from - * run + history evidence and merely re-attempts the marker. - */ + /** See WorkflowRunAttention.markWorkflowRunTerminalAttentionSettled. */ async markWorkflowRunTerminalAttentionSettled(params: { ownerWorkspaceId: string; runId: string; @@ -12362,148 +12339,15 @@ export class TaskService implements AgentTaskIntegration { runUpdatedAt: string; settledAs: "delivered" | "superseded"; }): Promise { - assert( - params.ownerWorkspaceId.length > 0, - "markWorkflowRunTerminalAttentionSettled requires ownerWorkspaceId" - ); - assert(params.runId.length > 0, "markWorkflowRunTerminalAttentionSettled requires runId"); - if (!isTerminalWorkflowRunStatus(params.status)) { - return; - } - const key = `${params.ownerWorkspaceId}\u0000${params.runId}`; - const previous = this.workflowRunSettlementByRun.get(key) ?? Promise.resolve(); - const run = previous - .catch(() => undefined) - .then(() => this.settleWorkflowRunTerminalAttention(params)); - const tracked = run - .then( - () => undefined, - () => undefined - ) - .finally(() => { - if (this.workflowRunSettlementByRun.get(key) === tracked) { - this.workflowRunSettlementByRun.delete(key); - } - }); - this.workflowRunSettlementByRun.set(key, tracked); - return await run; - } - - private async settleWorkflowRunTerminalAttention(params: { - ownerWorkspaceId: string; - runId: string; - status: WorkflowRunStatus; - runUpdatedAt: string; - settledAs: "delivered" | "superseded"; - }): Promise { - try { - // Downgrade compatibility: the previous build dedupes its startup re-derivation on the - // stable un-suffixed workflow_run id, so settling only the generation marker would let - // a downgraded build re-create a pending wake for a result the user already consumed. - // Stable-first ordering keeps the generation marker (this build's authority) retryable: - // if either write fails, the queue entry survives and the next drain re-settles both. - // The refresh (not write-once) matters: a stale previous-generation stable marker can - // survive its best-effort restart-time clear, and a downgraded build would read it as - // consumption of THIS generation's result. The recorded generation also gives the - // sweep's upgrade fallback exact evidence immune to wall-clock corrections. - await this.terminalAttentionStore.recordSettled( - { - ownerWorkspaceId: params.ownerWorkspaceId, - sourceKind: "workflow_run", - sourceId: params.runId, - generationId: params.runUpdatedAt, - terminalOutcome: terminalAttentionOutcome(params.status), - status: params.settledAs, - }, - { wholeSourceRefresh: true } - ); - await this.terminalAttentionStore.recordSettled({ - ownerWorkspaceId: params.ownerWorkspaceId, - sourceKind: "workflow_run", - sourceId: params.runId, - generationId: params.runUpdatedAt, - terminalOutcome: terminalAttentionOutcome(params.status), - status: params.settledAs, - }); - } catch (error: unknown) { - // Contain marker I/O here so no caller fails on bookkeeping: workflow_resume and - // task_await must still return the run's durable result, and the drain must move on to - // its other candidates. Keep the queue entry so the next drain re-evaluates from run + - // history evidence and re-attempts the marker; at worst a truthful wake re-delivers. - log.warn("Failed to record workflow terminal settlement marker", { - ownerWorkspaceId: params.ownerWorkspaceId, - runId: params.runId, - settledAs: params.settledAs, - error, - }); - return; - } - // Post-write revalidation: a wake-turn workflow_resume can restart the run while this - // settlement's snapshot was in flight, so the restart-time stable clear can land BEFORE - // the stable write above re-creates the marker. That marker postdates the newer - // generation's updatedAt, so the sweep's upgrade fallback (and a downgraded build's - // whole-run dedupe) would permanently suppress the newer result, and the by-run-id queue - // delete below would drop its owed wake. Reading the run AFTER the writes closes the - // write-side race: any restart after this read re-clears the stable marker itself and - // its terminal callback re-queues behind this deletion. - let currentRunUpdatedAt: string | null; - try { - const runStore = new WorkflowRunStore({ - sessionDir: path.join(this.config.sessionsDir, params.ownerWorkspaceId), - }); - currentRunUpdatedAt = (await runStore.getRun(params.runId)).updatedAt; - } catch { - currentRunUpdatedAt = null; - } - if (currentRunUpdatedAt !== params.runUpdatedAt) { - // The settled snapshot is no longer the run's newest generation (or the run is - // unreadable): the stable whole-run marker must not outlive the snapshot. The - // generation marker stays; it truthfully settles only this snapshot. - await this.clearWorkflowRunDowngradeSettlement({ - ownerWorkspaceId: params.ownerWorkspaceId, - runId: params.runId, - }); - if (currentRunUpdatedAt != null) { - // The queue entry now represents the newer generation's owed wake: leave it for the - // drain the terminal callback scheduled (the sweep backstops a lost poke). - return; - } - // Unreadable run: fall through to the delete. With no stable marker surviving, the - // sweep re-derives any owed newer generation from durable state. - } - this.pendingWorkflowRunAttention.get(params.ownerWorkspaceId)?.delete(params.runId); + return await this.workflowRunAttention.markWorkflowRunTerminalAttentionSettled(params); } - /** - * Downgrade-compat bookkeeping only: settlement dual-writes a stable un-suffixed marker for - * the previous build's whole-run dedupe (see markWorkflowRunTerminalAttentionSettled), and a - * restarted run invalidates it. Without this delete, downgrading after a resume would leave - * the old build refusing to enqueue the run's newer result behind the stale stable marker. - * This build reads the stable marker only as recency-gated upgrade evidence behind - * generation markers (see sweepWorkflowRunTerminalAttention), so the delete stays - * best-effort and must never fail the status transition. - */ + /** See WorkflowRunAttention.clearWorkflowRunDowngradeSettlement. */ async clearWorkflowRunDowngradeSettlement(params: { ownerWorkspaceId: string; runId: string; }): Promise { - assert( - params.ownerWorkspaceId.length > 0, - "clearWorkflowRunDowngradeSettlement requires ownerWorkspaceId" - ); - assert(params.runId.length > 0, "clearWorkflowRunDowngradeSettlement requires runId"); - try { - await this.terminalAttentionStore.delete( - params.ownerWorkspaceId, - TerminalAttentionStore.notificationId("workflow_run", params.runId) - ); - } catch (error: unknown) { - log.warn("Failed to clear stale workflow downgrade settlement marker", { - ownerWorkspaceId: params.ownerWorkspaceId, - runId: params.runId, - error, - }); - } + return await this.workflowRunAttention.clearWorkflowRunDowngradeSettlement(params); } private async suppressTerminalAttention(params: { @@ -13352,9 +13196,7 @@ export class TaskService implements AgentTaskIntegration { } } let pending = allPending.filter((notification) => notification.sourceKind !== "workflow_run"); - const queuedWorkflowRunIds = Array.from( - this.pendingWorkflowRunAttention.get(ownerWorkspaceId) ?? [] - ); + const queuedWorkflowRunIds = this.workflowRunAttention.queuedRunIds(ownerWorkspaceId); if (pending.length === 0 && queuedWorkflowRunIds.length === 0) { return; } @@ -13364,7 +13206,7 @@ export class TaskService implements AgentTaskIntegration { if (entry == null) { // Owner workspace no longer exists: the terminal artifacts remain retrievable elsewhere. // Queue only, no markers: a settled-marker write would recreate the deleted session dir. - this.pendingWorkflowRunAttention.delete(ownerWorkspaceId); + this.workflowRunAttention.dropOwner(ownerWorkspaceId); for (const notification of pending) { await this.terminalAttentionStore.markSuperseded(ownerWorkspaceId, notification.id); } @@ -13374,7 +13216,7 @@ export class TaskService implements AgentTaskIntegration { if (isWorkspaceArchived(entry.workspace.archivedAt, entry.workspace.unarchivedAt)) { // Workflow wakes stay unsettled while archived: the sweep skips archived workspaces, so // dropping the queue parks them until an unarchive-time sweep re-derives the entries. - this.pendingWorkflowRunAttention.delete(ownerWorkspaceId); + this.workflowRunAttention.dropOwner(ownerWorkspaceId); for (const notification of pending) { await this.terminalAttentionStore.markSuperseded(ownerWorkspaceId, notification.id); } @@ -13483,7 +13325,7 @@ export class TaskService implements AgentTaskIntegration { continue; } if (workflowPrompt.outcome === "drop") { - this.pendingWorkflowRunAttention.get(ownerWorkspaceId)?.delete(runId); + this.workflowRunAttention.dropRun(ownerWorkspaceId, runId); continue; } if (workflowPrompt.outcome === "settle") { diff --git a/src/node/services/workflowRunAttention.test.ts b/src/node/services/workflowRunAttention.test.ts new file mode 100644 index 00000000000..3b9f81c67ce --- /dev/null +++ b/src/node/services/workflowRunAttention.test.ts @@ -0,0 +1,325 @@ +import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test"; +import * as fsPromises from "fs/promises"; +import * as os from "node:os"; +import * as path from "node:path"; +import { TerminalAttentionStore } from "@/node/services/terminalAttentionStore"; +import { WorkflowRunAttention } from "@/node/services/workflowRunAttention"; +import { WorkflowRunStore } from "@/node/services/workflows/WorkflowRunStore"; + +type RecordSettled = TerminalAttentionStore["recordSettled"]; +type DeleteMarker = TerminalAttentionStore["delete"]; + +interface FakeStore { + recordSettled: ReturnType>; + delete: ReturnType>; +} + +function createFakeStore(): FakeStore { + return { + recordSettled: mock(() => Promise.resolve()), + delete: mock(() => Promise.resolve()), + }; +} + +const OWNER = "owner-workspace"; + +describe("WorkflowRunAttention", () => { + let sessionsDir: string; + let store: FakeStore; + let attention: WorkflowRunAttention; + + beforeEach(async () => { + sessionsDir = await fsPromises.mkdtemp(path.join(os.tmpdir(), "workflow-run-attention-")); + store = createFakeStore(); + attention = new WorkflowRunAttention({ + sessionsDir: () => sessionsDir, + terminalAttentionStore: () => store as unknown as TerminalAttentionStore, + }); + }); + + afterEach(async () => { + await fsPromises.rm(sessionsDir, { recursive: true, force: true }); + }); + + /** Creates a completed run on disk and returns its terminal generation (updatedAt). */ + async function createCompletedRun(runId: string, ownerWorkspaceId = OWNER): Promise { + const runStore = new WorkflowRunStore({ + sessionDir: path.join(sessionsDir, ownerWorkspaceId), + }); + await runStore.createRun({ + id: runId, + workspaceId: ownerWorkspaceId, + workflow: { + name: "research", + description: "Research workflow", + scope: "built-in", + executable: true, + }, + source: "export default function workflow() { return { reportMarkdown: 'done' }; }\n", + args: {}, + attentionPolicy: "notify_on_terminal", + now: "2026-06-19T00:00:00.000Z", + }); + await runStore.appendStatus(runId, "running", "2026-06-19T00:00:01.000Z"); + return (await runStore.appendStatus(runId, "completed", "2026-06-19T00:00:02.000Z")).updatedAt; + } + + function settle(runId: string, runUpdatedAt: string, ownerWorkspaceId = OWNER) { + return attention.markWorkflowRunTerminalAttentionSettled({ + ownerWorkspaceId, + runId, + status: "completed", + runUpdatedAt, + settledAs: "delivered", + }); + } + + describe("queue", () => { + test("reports only the first queueing of a run and keeps insertion order", () => { + expect(attention.queueWorkflowRunAttention(OWNER, "wfr_b")).toBe(true); + expect(attention.queueWorkflowRunAttention(OWNER, "wfr_a")).toBe(true); + expect(attention.queueWorkflowRunAttention(OWNER, "wfr_b")).toBe(false); + expect(attention.queuedRunIds(OWNER)).toEqual(["wfr_b", "wfr_a"]); + expect(attention.queuedRunIds("other-owner")).toEqual([]); + }); + + test("queuedRunIds returns a snapshot the caller can hold across queue changes", () => { + attention.queueWorkflowRunAttention(OWNER, "wfr_a"); + const snapshot = attention.queuedRunIds(OWNER); + attention.queueWorkflowRunAttention(OWNER, "wfr_b"); + expect(snapshot).toEqual(["wfr_a"]); + }); + + test("dropRun removes one run and dropOwner removes the whole owner", () => { + attention.queueWorkflowRunAttention(OWNER, "wfr_a"); + attention.queueWorkflowRunAttention(OWNER, "wfr_b"); + attention.queueWorkflowRunAttention("other-owner", "wfr_c"); + + attention.dropRun(OWNER, "wfr_a"); + attention.dropRun("missing-owner", "wfr_a"); + expect(attention.queuedRunIds(OWNER)).toEqual(["wfr_b"]); + + attention.dropOwner(OWNER); + expect(attention.queuedRunIds(OWNER)).toEqual([]); + expect(attention.queuedRunIds("other-owner")).toEqual(["wfr_c"]); + // A dropped owner can be re-queued from scratch. + expect(attention.queueWorkflowRunAttention(OWNER, "wfr_b")).toBe(true); + }); + }); + + describe("markWorkflowRunTerminalAttentionSettled", () => { + test("ignores a non-terminal status", async () => { + attention.queueWorkflowRunAttention(OWNER, "wfr_running"); + await attention.markWorkflowRunTerminalAttentionSettled({ + ownerWorkspaceId: OWNER, + runId: "wfr_running", + status: "running", + runUpdatedAt: "2026-06-19T00:00:01.000Z", + settledAs: "delivered", + }); + expect(store.recordSettled).not.toHaveBeenCalled(); + expect(attention.queuedRunIds(OWNER)).toEqual(["wfr_running"]); + }); + + test("writes the stable marker before the generation marker and dequeues the run", async () => { + const runId = "wfr_current"; + const generation = await createCompletedRun(runId); + attention.queueWorkflowRunAttention(OWNER, runId); + attention.queueWorkflowRunAttention(OWNER, "wfr_other"); + + await settle(runId, generation); + + const calls = store.recordSettled.mock.calls; + expect(calls).toHaveLength(2); + expect(calls[0][0]).toMatchObject({ sourceId: runId, generationId: generation }); + expect(calls[0][1]).toEqual({ wholeSourceRefresh: true }); + expect(calls[1][0]).toMatchObject({ + ownerWorkspaceId: OWNER, + sourceKind: "workflow_run", + sourceId: runId, + generationId: generation, + terminalOutcome: "completed", + status: "delivered", + }); + expect(calls[1][1]).toBeUndefined(); + expect(store.delete).not.toHaveBeenCalled(); + expect(attention.queuedRunIds(OWNER)).toEqual(["wfr_other"]); + }); + + test("keeps the queue entry and resolves when a marker write fails", async () => { + const runId = "wfr_write_fails"; + const generation = await createCompletedRun(runId); + attention.queueWorkflowRunAttention(OWNER, runId); + store.recordSettled.mockImplementationOnce(() => Promise.reject(new Error("disk full"))); + + await settle(runId, generation); + + expect(store.recordSettled).toHaveBeenCalledTimes(1); + expect(store.delete).not.toHaveBeenCalled(); + expect(attention.queuedRunIds(OWNER)).toEqual([runId]); + }); + + test("clears the stable marker but keeps the queue entry for a newer generation", async () => { + const runId = "wfr_restarted"; + const generation = await createCompletedRun(runId); + attention.queueWorkflowRunAttention(OWNER, runId); + + await settle(runId, `${generation}-older`); + + expect(store.delete.mock.calls).toEqual([ + [OWNER, TerminalAttentionStore.notificationId("workflow_run", runId)], + ]); + expect(attention.queuedRunIds(OWNER)).toEqual([runId]); + }); + + test("clears the stable marker and dequeues the run when it is unreadable", async () => { + const runId = "wfr_missing"; + attention.queueWorkflowRunAttention(OWNER, runId); + + await settle(runId, "2026-06-19T00:00:02.000Z"); + + expect(store.delete.mock.calls).toEqual([ + [OWNER, TerminalAttentionStore.notificationId("workflow_run", runId)], + ]); + expect(attention.queuedRunIds(OWNER)).toEqual([]); + }); + + test("serializes settlements of one run and lets other runs proceed", async () => { + const runId = "wfr_serialized"; + const otherRunId = "wfr_parallel"; + const generation = await createCompletedRun(runId); + const otherGeneration = await createCompletedRun(otherRunId); + + // Park every settlement of runId inside its first marker write until released, in order. + const order: string[] = []; + const releases: Array<() => void> = []; + let reached: () => void = () => undefined; + store.recordSettled.mockImplementation(async (record, options) => { + if (record.sourceId === runId && options?.wholeSourceRefresh === true) { + order.push(`start:${record.status}`); + await new Promise((resolve) => { + releases.push(resolve); + reached(); + }); + } + }); + const nextReached = () => + new Promise((resolve) => { + reached = resolve; + }); + + let parked = nextReached(); + const first = attention.markWorkflowRunTerminalAttentionSettled({ + ownerWorkspaceId: OWNER, + runId, + status: "completed", + runUpdatedAt: generation, + settledAs: "superseded", + }); + await parked; + const second = settle(runId, generation); + + // Another run of the same owner is not blocked by the parked settlement. + await settle(otherRunId, otherGeneration); + expect(order).toEqual(["start:superseded"]); + + parked = nextReached(); + releases[0](); + await first; + await parked; + expect(order).toEqual(["start:superseded", "start:delivered"]); + + // A third settlement queues behind the still-parked second one: finishing the first must + // not have dropped the second one's chain entry. + const third = settle(runId, generation); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(order).toEqual(["start:superseded", "start:delivered"]); + + parked = nextReached(); + releases[1](); + await second; + await parked; + expect(order).toEqual(["start:superseded", "start:delivered", "start:delivered"]); + releases[2](); + await third; + }); + + test("rejects an empty owner or run id and accepts one-character ids", async () => { + const base = { + status: "completed", + runUpdatedAt: "2026-06-19T00:00:02.000Z", + settledAs: "delivered", + } as const; + await expect( + attention.markWorkflowRunTerminalAttentionSettled({ + ...base, + ownerWorkspaceId: "", + runId: "r", + }) + ).rejects.toThrow(); + await expect( + attention.markWorkflowRunTerminalAttentionSettled({ + ...base, + ownerWorkspaceId: "o", + runId: "", + }) + ).rejects.toThrow(); + expect(store.recordSettled).not.toHaveBeenCalled(); + + await attention.markWorkflowRunTerminalAttentionSettled({ + ...base, + ownerWorkspaceId: "o", + runId: "r", + }); + expect(store.recordSettled).toHaveBeenCalledTimes(2); + }); + + test("reads the store and sessions dir from deps at call time", async () => { + const runId = "wfr_swapped"; + const generation = await createCompletedRun(runId); + const swappedStore = createFakeStore(); + let currentStore = store; + attention = new WorkflowRunAttention({ + sessionsDir: () => sessionsDir, + terminalAttentionStore: () => currentStore as unknown as TerminalAttentionStore, + }); + currentStore = swappedStore; + attention.queueWorkflowRunAttention(OWNER, runId); + + await settle(runId, generation); + + expect(store.recordSettled).not.toHaveBeenCalled(); + expect(swappedStore.recordSettled).toHaveBeenCalledTimes(2); + expect(attention.queuedRunIds(OWNER)).toEqual([]); + }); + }); + + describe("clearWorkflowRunDowngradeSettlement", () => { + test("deletes the stable marker and swallows a delete failure", async () => { + store.delete.mockImplementationOnce(() => Promise.reject(new Error("locked"))); + + await attention.clearWorkflowRunDowngradeSettlement({ ownerWorkspaceId: OWNER, runId: "r" }); + + expect(store.delete.mock.calls).toEqual([ + [OWNER, TerminalAttentionStore.notificationId("workflow_run", "r")], + ]); + }); + + test("accepts one-character ids", async () => { + await attention.clearWorkflowRunDowngradeSettlement({ ownerWorkspaceId: "o", runId: "r" }); + expect(store.delete.mock.calls).toEqual([ + ["o", TerminalAttentionStore.notificationId("workflow_run", "r")], + ]); + }); + + test("rejects an empty owner or run id", async () => { + await expect( + attention.clearWorkflowRunDowngradeSettlement({ ownerWorkspaceId: "", runId: "r" }) + ).rejects.toThrow(); + await expect( + attention.clearWorkflowRunDowngradeSettlement({ ownerWorkspaceId: OWNER, runId: "" }) + ).rejects.toThrow(); + expect(store.delete).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/src/node/services/workflowRunAttention.ts b/src/node/services/workflowRunAttention.ts new file mode 100644 index 00000000000..616c31db877 --- /dev/null +++ b/src/node/services/workflowRunAttention.ts @@ -0,0 +1,225 @@ +import assert from "node:assert/strict"; +import * as path from "node:path"; +import { isTerminalWorkflowRunStatus, type WorkflowRunStatus } from "@/common/types/workflow"; +import { log } from "@/node/services/log"; +import { terminalAttentionOutcome } from "@/node/services/taskWorkspaceSeam"; +import { TerminalAttentionStore } from "@/node/services/terminalAttentionStore"; +import { WorkflowRunStore } from "@/node/services/workflows/WorkflowRunStore"; + +/** + * Call-time closures into TaskService: tests swap TaskService fields at runtime, so the owner + * must read them when it runs instead of capturing them at construction. + */ +export interface WorkflowRunAttentionDeps { + sessionsDir: () => string; + terminalAttentionStore: () => TerminalAttentionStore; +} + +/** + * Owns TaskService's in-memory workflow terminal wake queue and the per-run settlement chain + * that writes the durable settled markers. The drain and the sweep stay on TaskService and + * reach the queue through the accessors below. + */ +export class WorkflowRunAttention { + // Terminal settlements of the same run must not overlap: settlement is multi-step (stable + // refresh, generation marker, post-write mismatch delete), so an older generation reaching + // its mismatch delete after a newer settlement's stable refresh would remove the newer + // generation's valid marker and let a downgraded build re-deliver a consumed result. + private readonly workflowRunSettlementByRun = new Map>(); + // Owed workflow terminal wakes (owner -> runIds believed terminal and not yet settled). An + // in-memory work queue over durable state, not a delivery record: entries are (re)derived + // from run records + settled markers at startup and on the periodic sweep and added by live + // terminal callbacks, so losing the map merely delays a wake until the next sweep. + readonly pendingWorkflowRunAttention = new Map>(); + + constructor(private readonly deps: WorkflowRunAttentionDeps) {} + + /** Snapshot of the owner's queued run IDs, in insertion order. */ + queuedRunIds(ownerWorkspaceId: string): string[] { + return Array.from(this.pendingWorkflowRunAttention.get(ownerWorkspaceId) ?? []); + } + + /** Forget every queued run of the owner (the sweep re-derives them from durable state). */ + dropOwner(ownerWorkspaceId: string): void { + this.pendingWorkflowRunAttention.delete(ownerWorkspaceId); + } + + /** Forget one queued run of the owner without writing any marker. */ + dropRun(ownerWorkspaceId: string, runId: string): void { + this.pendingWorkflowRunAttention.get(ownerWorkspaceId)?.delete(runId); + } + + /** Returns true when the run was newly queued for this owner. */ + queueWorkflowRunAttention(ownerWorkspaceId: string, runId: string): boolean { + let runIds = this.pendingWorkflowRunAttention.get(ownerWorkspaceId); + if (runIds == null) { + runIds = new Set(); + this.pendingWorkflowRunAttention.set(ownerWorkspaceId, runIds); + } + if (runIds.has(runId)) { + return false; + } + runIds.add(runId); + return true; + } + + /** + * Durable "this terminal generation needs no further wake" marker, keyed by the run's + * terminal updatedAt: a later resume produces a new generation and thereby re-arms + * attention without any reset bookkeeping. Write-once and best-effort by design: if the + * write fails or never happens, the next drain evaluation re-derives the same answer from + * run + history evidence and merely re-attempts the marker. + */ + async markWorkflowRunTerminalAttentionSettled(params: { + ownerWorkspaceId: string; + runId: string; + status: WorkflowRunStatus; + runUpdatedAt: string; + settledAs: "delivered" | "superseded"; + }): Promise { + assert( + params.ownerWorkspaceId.length > 0, + "markWorkflowRunTerminalAttentionSettled requires ownerWorkspaceId" + ); + assert(params.runId.length > 0, "markWorkflowRunTerminalAttentionSettled requires runId"); + if (!isTerminalWorkflowRunStatus(params.status)) { + return; + } + const key = `${params.ownerWorkspaceId}\u0000${params.runId}`; + const previous = this.workflowRunSettlementByRun.get(key) ?? Promise.resolve(); + const run = previous + .catch(() => undefined) + .then(() => this.settleWorkflowRunTerminalAttention(params)); + const tracked = run + .then( + () => undefined, + () => undefined + ) + .finally(() => { + if (this.workflowRunSettlementByRun.get(key) === tracked) { + this.workflowRunSettlementByRun.delete(key); + } + }); + this.workflowRunSettlementByRun.set(key, tracked); + return await run; + } + + private async settleWorkflowRunTerminalAttention(params: { + ownerWorkspaceId: string; + runId: string; + status: WorkflowRunStatus; + runUpdatedAt: string; + settledAs: "delivered" | "superseded"; + }): Promise { + try { + // Downgrade compatibility: the previous build dedupes its startup re-derivation on the + // stable un-suffixed workflow_run id, so settling only the generation marker would let + // a downgraded build re-create a pending wake for a result the user already consumed. + // Stable-first ordering keeps the generation marker (this build's authority) retryable: + // if either write fails, the queue entry survives and the next drain re-settles both. + // The refresh (not write-once) matters: a stale previous-generation stable marker can + // survive its best-effort restart-time clear, and a downgraded build would read it as + // consumption of THIS generation's result. The recorded generation also gives the + // sweep's upgrade fallback exact evidence immune to wall-clock corrections. + await this.deps.terminalAttentionStore().recordSettled( + { + ownerWorkspaceId: params.ownerWorkspaceId, + sourceKind: "workflow_run", + sourceId: params.runId, + generationId: params.runUpdatedAt, + terminalOutcome: terminalAttentionOutcome(params.status), + status: params.settledAs, + }, + { wholeSourceRefresh: true } + ); + await this.deps.terminalAttentionStore().recordSettled({ + ownerWorkspaceId: params.ownerWorkspaceId, + sourceKind: "workflow_run", + sourceId: params.runId, + generationId: params.runUpdatedAt, + terminalOutcome: terminalAttentionOutcome(params.status), + status: params.settledAs, + }); + } catch (error: unknown) { + // Contain marker I/O here so no caller fails on bookkeeping: workflow_resume and + // task_await must still return the run's durable result, and the drain must move on to + // its other candidates. Keep the queue entry so the next drain re-evaluates from run + + // history evidence and re-attempts the marker; at worst a truthful wake re-delivers. + log.warn("Failed to record workflow terminal settlement marker", { + ownerWorkspaceId: params.ownerWorkspaceId, + runId: params.runId, + settledAs: params.settledAs, + error, + }); + return; + } + // Post-write revalidation: a wake-turn workflow_resume can restart the run while this + // settlement's snapshot was in flight, so the restart-time stable clear can land BEFORE + // the stable write above re-creates the marker. That marker postdates the newer + // generation's updatedAt, so the sweep's upgrade fallback (and a downgraded build's + // whole-run dedupe) would permanently suppress the newer result, and the by-run-id queue + // delete below would drop its owed wake. Reading the run AFTER the writes closes the + // write-side race: any restart after this read re-clears the stable marker itself and + // its terminal callback re-queues behind this deletion. + let currentRunUpdatedAt: string | null; + try { + const runStore = new WorkflowRunStore({ + sessionDir: path.join(this.deps.sessionsDir(), params.ownerWorkspaceId), + }); + currentRunUpdatedAt = (await runStore.getRun(params.runId)).updatedAt; + } catch { + currentRunUpdatedAt = null; + } + if (currentRunUpdatedAt !== params.runUpdatedAt) { + // The settled snapshot is no longer the run's newest generation (or the run is + // unreadable): the stable whole-run marker must not outlive the snapshot. The + // generation marker stays; it truthfully settles only this snapshot. + await this.clearWorkflowRunDowngradeSettlement({ + ownerWorkspaceId: params.ownerWorkspaceId, + runId: params.runId, + }); + if (currentRunUpdatedAt != null) { + // The queue entry now represents the newer generation's owed wake: leave it for the + // drain the terminal callback scheduled (the sweep backstops a lost poke). + return; + } + // Unreadable run: fall through to the delete. With no stable marker surviving, the + // sweep re-derives any owed newer generation from durable state. + } + this.pendingWorkflowRunAttention.get(params.ownerWorkspaceId)?.delete(params.runId); + } + + /** + * Downgrade-compat bookkeeping only: settlement dual-writes a stable un-suffixed marker for + * the previous build's whole-run dedupe (see markWorkflowRunTerminalAttentionSettled), and a + * restarted run invalidates it. Without this delete, downgrading after a resume would leave + * the old build refusing to enqueue the run's newer result behind the stale stable marker. + * This build reads the stable marker only as recency-gated upgrade evidence behind + * generation markers (see sweepWorkflowRunTerminalAttention), so the delete stays + * best-effort and must never fail the status transition. + */ + async clearWorkflowRunDowngradeSettlement(params: { + ownerWorkspaceId: string; + runId: string; + }): Promise { + assert( + params.ownerWorkspaceId.length > 0, + "clearWorkflowRunDowngradeSettlement requires ownerWorkspaceId" + ); + assert(params.runId.length > 0, "clearWorkflowRunDowngradeSettlement requires runId"); + try { + await this.deps + .terminalAttentionStore() + .delete( + params.ownerWorkspaceId, + TerminalAttentionStore.notificationId("workflow_run", params.runId) + ); + } catch (error: unknown) { + log.warn("Failed to clear stale workflow downgrade settlement marker", { + ownerWorkspaceId: params.ownerWorkspaceId, + runId: params.runId, + error, + }); + } + } +} From a67670bdb1718e931285d539203688130a21b476 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Tue, 27 Oct 2026 07:58:09 +0000 Subject: [PATCH 2/2] =?UTF-8?q?=F0=9F=A4=96=20tests:=20let=20the=20await-t?= =?UTF-8?q?henable=20lint=20rule=20see=20bun:test=20async=20matchers=20in?= =?UTF-8?q?=20WorkflowRunAttention=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/node/services/workflowRunAttention.test.ts | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/node/services/workflowRunAttention.test.ts b/src/node/services/workflowRunAttention.test.ts index 3b9f81c67ce..067628d9e12 100644 --- a/src/node/services/workflowRunAttention.test.ts +++ b/src/node/services/workflowRunAttention.test.ts @@ -1,3 +1,4 @@ +/* eslint-disable @typescript-eslint/await-thenable -- bun:test async matchers return thenables the rule cannot see */ import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test"; import * as fsPromises from "fs/promises"; import * as os from "node:os"; @@ -256,14 +257,14 @@ describe("WorkflowRunAttention", () => { ownerWorkspaceId: "", runId: "r", }) - ).rejects.toThrow(); + ).rejects.toThrow(/requires/); await expect( attention.markWorkflowRunTerminalAttentionSettled({ ...base, ownerWorkspaceId: "o", runId: "", }) - ).rejects.toThrow(); + ).rejects.toThrow(/requires/); expect(store.recordSettled).not.toHaveBeenCalled(); await attention.markWorkflowRunTerminalAttentionSettled({ @@ -315,10 +316,10 @@ describe("WorkflowRunAttention", () => { test("rejects an empty owner or run id", async () => { await expect( attention.clearWorkflowRunDowngradeSettlement({ ownerWorkspaceId: "", runId: "r" }) - ).rejects.toThrow(); + ).rejects.toThrow(/requires/); await expect( attention.clearWorkflowRunDowngradeSettlement({ ownerWorkspaceId: OWNER, runId: "" }) - ).rejects.toThrow(); + ).rejects.toThrow(/requires/); expect(store.delete).not.toHaveBeenCalled(); }); });