diff --git a/src/node/services/attemptLedger.ts b/src/node/services/attemptLedger.ts index 8b0e54418ab..c01343fa054 100644 --- a/src/node/services/attemptLedger.ts +++ b/src/node/services/attemptLedger.ts @@ -194,9 +194,9 @@ export interface AttemptLedgerDeps { ) => Promise; settleChildGoalPause: (workspaceId: string) => Promise; findStreamEndDecision: (taskId: string, attemptId: string) => StreamEndDecision | undefined; - /** Drop the stream-end decisions no send obligation can still read (TaskService). */ + /** Drop the stream-end decisions no send obligation can still read (StreamEndLedger). */ pruneStreamEndDecisions: (taskId: string) => void; - /** Drop every stream-end decision of the task (TaskService.streamEndDecisionsByTaskId). */ + /** Drop every stream-end decision of the task (StreamEndLedger). */ discardStreamEndDecisions: (taskId: string) => void; inspectAttemptOutcome: ( taskId: string, diff --git a/src/node/services/streamEndLedger.test.ts b/src/node/services/streamEndLedger.test.ts new file mode 100644 index 00000000000..f35dd4310ef --- /dev/null +++ b/src/node/services/streamEndLedger.test.ts @@ -0,0 +1,451 @@ +import { describe, expect, test } from "bun:test"; + +import type { StreamEndEvent } from "@/common/types/stream"; +import type { ProjectsConfig } from "@/node/config"; +import { StreamEndLedger, type StreamEndLedgerDeps } from "@/node/services/streamEndLedger"; +import type { + AdmittedSend, + OwnedTaskAttempt, + StreamEndTaskOrigin, +} from "@/node/services/taskService"; +import type { + QueueCutAttributionSnapshot, + QueueCutReceipt, +} from "@/node/services/taskWorkspaceSeam"; +import type { StreamAccountingReceiptOutcome } from "@/node/services/workspaceGoalService"; +import { MutexMap } from "@/node/utils/concurrency/mutexMap"; + +const TASK = "task-a"; +const ATTEMPT = "att_00000000000000a1"; +const OTHER = "att_00000000000000b2"; + +function attempt(attemptId: string = ATTEMPT): OwnedTaskAttempt { + return { attemptId, receiptEligible: true } as unknown as OwnedTaskAttempt; +} + +function send(attemptId: string, state: string): AdmittedSend { + return { taskId: TASK, attemptId, state } as unknown as AdmittedSend; +} + +function configWithRow(row: Record | undefined): ProjectsConfig { + const workspaces = row == null ? [] : [{ id: TASK, ...row }]; + return { projects: new Map([["/project", { workspaces }]]) } as unknown as ProjectsConfig; +} + +const streamEnd = { workspaceId: TASK, messageId: "msg-1" } as unknown as StreamEndEvent; +const snapshot = {} as unknown as QueueCutAttributionSnapshot; + +function deferral(overrides: Record = {}) { + return { + sourceMessageId: "msg-1", + sourceTurnGeneration: Symbol("source"), + continuationEntryId: "entry-1", + executionId: "exec-1", + stopEpoch: 0, + attemptId: ATTEMPT, + ...overrides, + }; +} + +/** A ledger over fake deps; `calls` records the order of the deps that act. */ +function harness() { + const state = { + config: configWithRow({ taskStatus: "running", taskAttemptId: ATTEMPT }), + activeTurn: undefined as symbol | undefined, + sends: new Set(), + stopEpoch: 0, + executionId: "exec-1" as string | null, + admissionOpen: true, + receipt: undefined as QueueCutReceipt | undefined, + disposeResult: true, + accounting: Promise.withResolvers(), + handlerResults: [] as Array<"deferred" | undefined | Error>, + }; + const calls: string[] = []; + const listeners = new Set<() => void>(); + const locks = new MutexMap(); + const deps: StreamEndLedgerDeps = { + config: () => ({ loadConfigOrDefault: () => state.config }), + workspaceService: () => ({ + getActiveTurnGeneration: () => state.activeTurn, + drainQueuedMessagesIfIdle: () => calls.push("drain"), + disposeQueueCut: () => { + calls.push("dispose"); + return state.disposeResult; + }, + getQueueCutReceipt: () => state.receipt, + }), + workspaceGoalService: () => ({ + streamAccountingReceiptOutcome: () => { + calls.push("accounting"); + return state.accounting.promise; + }, + }), + workspaceEventLocks: () => locks, + handleStreamEnd: () => { + calls.push("handle"); + const result = state.handlerResults.shift(); + return result instanceof Error ? Promise.reject(result) : Promise.resolve(result); + }, + recoverTaskFromIncompleteStreamEnd: (_workspaceId, status, streamAttemptId) => { + calls.push(`recover:${String(status)}:${String(streamAttemptId)}`); + return Promise.resolve(); + }, + getWorkspaceStopEpoch: () => state.stopEpoch, + getAgentTaskExecutionId: () => state.executionId, + recheckWorkspaceStopRelease: () => { + // The decision is pruned before the stop latch rechecks it. + calls.push(`recheck:${String(ledger.streamEndDecisionsByTaskId.has(TASK))}`); + }, + attemptAdmissionOpen: () => state.admissionOpen, + addAttemptSettlementListener: (_taskId, listener) => { + calls.push("add-listener"); + listeners.add(listener); + }, + removeAttemptSettlementListener: (_taskId, listener) => { + calls.push("remove-listener"); + listeners.delete(listener); + }, + sendsOf: () => state.sends, + }; + const ledger = new StreamEndLedger(deps); + return { ledger, state, calls, listeners, locks }; +} + +function origin(owned: OwnedTaskAttempt | undefined): StreamEndTaskOrigin { + return { executionId: "exec-1", stopEpoch: 0, ownedAttempt: owned }; +} + +/** Waits until the event lock of TASK is free (the handler run finished). */ +async function settle(locks: MutexMap): Promise { + for (let i = 0; i < 5; i++) await locks.withLock(TASK, () => Promise.resolve()); +} + +describe("StreamEndLedger decisions", () => { + test("register binds the owned attempt object and the turn live in the event tick", () => { + const t = harness(); + const turn = Symbol("turn"); + t.state.activeTurn = turn; + const owned = attempt(); + const decision = t.ledger.registerStreamEndDecision(TASK, "msg-1", owned, undefined); + t.state.activeTurn = undefined; + expect(decision?.attempt).toBe(owned); + expect(decision?.turn).toBe(turn); + expect(t.ledger.findPendingStreamEndDecision(TASK, owned)).toBe(decision); + expect(t.ledger.registerStreamEndDecision(TASK, "msg-2", undefined, undefined)).toBeUndefined(); + }); + + test("an unowned decision is keyed by the persisted attempt id and never matches an owner", () => { + const t = harness(); + const decision = t.ledger.registerStreamEndDecision(TASK, "msg-1", undefined, ATTEMPT); + expect(t.ledger.findPendingStreamEndDecision(TASK, undefined, ATTEMPT)).toBe(decision); + expect(t.ledger.findPendingStreamEndDecision(TASK, attempt())).toBeUndefined(); + expect(t.ledger.findPendingStreamEndDecision(TASK, undefined)).toBeUndefined(); + expect(t.ledger.findPendingStreamEndDecision(TASK, attempt(OTHER))).toBeUndefined(); + }); + + test("find prefers the pending decision of the attempt, else its latest", () => { + const t = harness(); + t.state.sends.add(send(ATTEMPT, "pending")); + const first = t.ledger.registerStreamEndDecision(TASK, "msg-1", undefined, ATTEMPT); + const pending = t.ledger.registerStreamEndDecision(TASK, "msg-2", undefined, ATTEMPT); + const third = t.ledger.registerStreamEndDecision(TASK, "msg-3", undefined, ATTEMPT); + t.ledger.resolveStreamEndDecision(TASK, first, "nonreport"); + t.ledger.resolveStreamEndDecision(TASK, third, "nonreport"); + expect(t.ledger.findStreamEndDecision(TASK, ATTEMPT)).toBe(pending); + t.ledger.resolveStreamEndDecision(TASK, pending, "nonreport"); + expect(t.ledger.findStreamEndDecision(TASK, ATTEMPT)).toBe(third); + expect(t.ledger.findStreamEndDecision(TASK, OTHER)).toBeUndefined(); + }); + + test("resolve runs once, downgrades the receipt, then prunes, rechecks the stop and drains", () => { + const t = harness(); + const owned = attempt(); + const decision = t.ledger.registerStreamEndDecision(TASK, "msg-1", owned, undefined); + expect(t.ledger.resolveStreamEndDecision(TASK, decision, "published")).toBe(true); + expect(owned.receiptEligible).toBe(false); + expect(t.calls).toEqual(["recheck:false", "drain"]); + expect(t.ledger.resolveStreamEndDecision(TASK, decision, "nonreport")).toBe(false); + expect(decision?.outcome).toBe("published"); + expect(t.calls).toEqual(["recheck:false", "drain"]); + expect(t.ledger.resolveStreamEndDecision(TASK, undefined, "nonreport")).toBe(false); + }); + + test("a nonreport decision keeps the attempt receipt-eligible", () => { + const t = harness(); + const owned = attempt(); + const decision = t.ledger.registerStreamEndDecision(TASK, "msg-1", owned, undefined); + t.ledger.resolveStreamEndDecision(TASK, decision, "nonreport"); + expect(owned.receiptEligible).toBe(true); + }); + + test("prune keeps pending, read and live-turn terminal decisions only", () => { + const t = harness(); + const turn = Symbol("turn"); + t.state.sends.add(send(OTHER, "enqueued")); + t.state.sends.add(send("att_unread", "admitted")); + const pending = t.ledger.registerStreamEndDecision(TASK, "m1", undefined, ATTEMPT); + const read = t.ledger.registerStreamEndDecision(TASK, "m2", undefined, OTHER); + t.state.activeTurn = turn; + const terminal = t.ledger.registerStreamEndDecision(TASK, "m3", undefined, "att_terminal"); + const liveNonreport = t.ledger.registerStreamEndDecision(TASK, "m4", undefined, "att_unread"); + t.ledger.resolveStreamEndDecision(TASK, read, "nonreport"); + t.ledger.resolveStreamEndDecision(TASK, terminal, "indeterminate"); + t.ledger.resolveStreamEndDecision(TASK, liveNonreport, "nonreport"); + expect(t.ledger.streamEndDecisionsByTaskId.get(TASK)).toEqual([pending!, read!, terminal!]); + + // The reporting turn settles: only actual readers keep the terminal decision now. + t.ledger.settleTurn(TASK, Symbol("unrelated")); + expect(terminal?.turn).toBe(turn); + t.ledger.settleTurn(TASK, turn); + expect(t.ledger.streamEndDecisionsByTaskId.get(TASK)).toEqual([pending!, read!]); + + t.state.sends.clear(); + t.ledger.resolveStreamEndDecision(TASK, pending, "nonreport"); + expect(t.ledger.streamEndDecisionsByTaskId.has(TASK)).toBe(false); + }); + + test("prune keeps the same array when nothing is dropped", () => { + const t = harness(); + t.ledger.registerStreamEndDecision(TASK, "m1", undefined, ATTEMPT); + const before = t.ledger.streamEndDecisionsByTaskId.get(TASK); + expect(before).toHaveLength(1); + t.ledger.pruneStreamEndDecisions(TASK); + expect(t.ledger.streamEndDecisionsByTaskId.get(TASK)).toBe(before); + }); + + test("rebindTurn moves decisions of the superseded turn to its successor", () => { + const t = harness(); + const previous = Symbol("previous"); + const next = Symbol("next"); + t.state.activeTurn = previous; + const decision = t.ledger.registerStreamEndDecision(TASK, "m1", undefined, ATTEMPT); + t.ledger.rebindTurn(TASK, Symbol("unrelated"), next); + expect(decision?.turn).toBe(previous); + t.ledger.rebindTurn(TASK, previous, next); + expect(decision?.turn).toBe(next); + }); + + test("settling a turn no decision was decided for prunes nothing", () => { + const t = harness(); + t.state.sends.add(send(ATTEMPT, "pending")); + const decision = t.ledger.registerStreamEndDecision(TASK, "m1", undefined, ATTEMPT); + t.ledger.resolveStreamEndDecision(TASK, decision, "nonreport"); + // The reader went away without a prune: only a settled decision turn prunes now. + t.state.sends.clear(); + t.ledger.settleTurn(TASK, Symbol("unrelated")); + expect(t.ledger.findStreamEndDecision(TASK, ATTEMPT)).toBe(decision); + }); + + test("discard drops every decision of the task", () => { + const t = harness(); + t.ledger.registerStreamEndDecision(TASK, "m1", undefined, ATTEMPT); + t.ledger.discardStreamEndDecisions(TASK); + expect(t.ledger.findStreamEndDecision(TASK, ATTEMPT)).toBeUndefined(); + }); + + test("a reported row resolves the attempt's decision indeterminate", () => { + const t = harness(); + const decision = t.ledger.registerStreamEndDecision(TASK, "m1", undefined, ATTEMPT); + expect(t.ledger.resolveDecisionOfReportedAttempt(TASK, ATTEMPT)).toBe(false); + expect(t.ledger.resolveDecisionOfReportedAttempt(TASK, undefined)).toBe(false); + t.state.config = configWithRow(undefined); + expect(t.ledger.resolveDecisionOfReportedAttempt(TASK, ATTEMPT)).toBe(false); + t.state.config = configWithRow({ taskStatus: "reported", taskAttemptId: ATTEMPT }); + expect(t.ledger.resolveDecisionOfReportedAttempt(TASK, OTHER)).toBe(false); + expect(decision?.outcome).toBe("pending"); + expect(t.ledger.resolveDecisionOfReportedAttempt(TASK, ATTEMPT)).toBe(true); + expect(decision?.outcome).toBe("indeterminate"); + }); +}); + +describe("StreamEndLedger stream-end handler runs", () => { + test("a handler return resolves nonreport, a throw indeterminate", async () => { + const t = harness(); + const ok = t.ledger.registerStreamEndDecision(TASK, "m1", undefined, ATTEMPT); + t.state.handlerResults.push(undefined); + t.ledger.runStreamEndHandler(streamEnd, snapshot, origin(undefined), ok); + await settle(t.locks); + expect(ok?.outcome).toBe("nonreport"); + + const thrown = t.ledger.registerStreamEndDecision(TASK, "m2", undefined, OTHER); + t.state.handlerResults.push(new Error("handler failed")); + t.ledger.runStreamEndHandler(streamEnd, snapshot, origin(undefined), thrown); + await settle(t.locks); + expect(thrown?.outcome).toBe("indeterminate"); + }); + + test("a deferred stream waits for its accounting, then re-runs with the same decision", async () => { + const t = harness(); + const owned = attempt(); + const decision = t.ledger.registerStreamEndDecision(TASK, "m1", owned, undefined); + t.state.handlerResults.push("deferred", undefined); + t.ledger.runStreamEndHandler(streamEnd, snapshot, origin(owned), decision); + await settle(t.locks); + expect(decision?.outcome).toBe("pending"); + expect(t.calls).toEqual(["handle", "add-listener", "accounting"]); + expect(t.listeners.size).toBe(1); + + t.state.accounting.resolve("settled"); + await settle(t.locks); + expect(t.calls.slice(0, 5)).toEqual([ + "handle", + "add-listener", + "accounting", + "remove-listener", + "handle", + ]); + expect(t.listeners.size).toBe(0); + expect(decision?.outcome).toBe("nonreport"); + }); + + test("the deferral listener runs synchronously once and cancels a closed attempt", async () => { + const t = harness(); + t.state.admissionOpen = false; + const decision = t.ledger.registerStreamEndDecision(TASK, "m1", undefined, ATTEMPT); + t.state.handlerResults.push("deferred", undefined); + t.ledger.runStreamEndHandler( + streamEnd, + snapshot, + { ...origin(undefined), unownedAttemptId: ATTEMPT }, + decision + ); + await settle(t.locks); + // Cancelled without the accounting settling: the stream end re-ran. + expect(t.calls.filter((call) => call === "handle")).toHaveLength(2); + expect(t.listeners.size).toBe(0); + expect(decision?.outcome).toBe("nonreport"); + }); + + test("a deferral whose attempt reported meanwhile resolves indeterminate before the re-run", async () => { + const t = harness(); + const owned = attempt(); + const decision = t.ledger.registerStreamEndDecision(TASK, "m1", owned, undefined); + t.state.handlerResults.push("deferred", undefined); + t.ledger.runStreamEndHandler(streamEnd, snapshot, origin(owned), decision); + await settle(t.locks); + t.state.config = configWithRow({ taskStatus: "reported", taskAttemptId: ATTEMPT }); + // The listener sees the row no longer reads running and cancels the wait. + for (const listener of t.listeners) listener(); + await settle(t.locks); + expect(decision?.outcome).toBe("indeterminate"); + expect(owned.receiptEligible).toBe(false); + }); + + test("a deferral whose row is gone cancels at once", async () => { + const t = harness(); + t.state.config = configWithRow(undefined); + const decision = t.ledger.registerStreamEndDecision(TASK, "m1", undefined, ATTEMPT); + t.state.handlerResults.push("deferred", undefined); + t.ledger.runStreamEndHandler( + streamEnd, + snapshot, + { ...origin(undefined), unownedAttemptId: ATTEMPT }, + decision + ); + await settle(t.locks); + expect(t.calls.filter((call) => call === "handle")).toHaveLength(2); + expect(decision?.outcome).toBe("nonreport"); + }); + + test("a stop epoch or execution that moved on cancels the deferral", async () => { + for (const move of ["stop", "execution"] as const) { + const t = harness(); + const owned = attempt(); + const decision = t.ledger.registerStreamEndDecision(TASK, "m1", owned, undefined); + t.state.handlerResults.push("deferred", undefined); + t.ledger.runStreamEndHandler(streamEnd, snapshot, origin(owned), decision); + await settle(t.locks); + expect(t.listeners.size).toBe(1); + if (move === "stop") t.state.stopEpoch = 1; + else t.state.executionId = "exec-2"; + for (const listener of t.listeners) listener(); + await settle(t.locks); + expect(t.listeners.size).toBe(0); + expect(decision?.outcome).toBe("nonreport"); + } + }); +}); + +describe("StreamEndLedger deferred cut stream ends", () => { + test("no deferral reconciles to false", async () => { + const t = harness(); + expect(await t.ledger.reconcileDeferredTaskStreamEnd(TASK)).toBe(false); + expect(t.calls).toEqual([]); + }); + + test("an obsolete deferral is dropped and its cut disposed without recovery", async () => { + // Each case breaks exactly one condition of a live deferral. + const live = { + parentWorkspaceId: "p", + taskStatus: "running", + taskAttemptId: ATTEMPT, + taskExecutionId: "exec-1", + }; + const cases: Array<{ row?: Record; stopEpoch?: number }> = [ + { row: undefined }, + { row: { ...live, parentWorkspaceId: undefined } }, + { row: { ...live, taskStatus: "awaiting_report" } }, + { row: { ...live, taskAttemptId: OTHER } }, + { row: { ...live, taskExecutionId: "x" } }, + { row: live, stopEpoch: 1 }, + ]; + for (const c of cases) { + const t = harness(); + t.state.config = configWithRow(c.row); + t.state.stopEpoch = c.stopEpoch ?? 0; + t.ledger.recordDeferral(TASK, deferral()); + expect(await t.ledger.reconcileDeferredTaskStreamEnd(TASK)).toBe(false); + expect(t.ledger.hasDeferral(TASK)).toBe(false); + expect(t.calls).toEqual(["dispose"]); + } + }); + + test("the successor outcome settles or keeps the deferral", async () => { + const live = { + parentWorkspaceId: "p", + taskStatus: "running", + taskAttemptId: ATTEMPT, + taskExecutionId: "exec-1", + }; + const receipt = (successor: string) => + ({ + successor, + sourceTurnGeneration: Symbol("s"), + sourceHandled: true, + disposed: false, + }) as unknown as QueueCutReceipt; + const run = async (r: QueueCutReceipt | undefined, disposeResult = true) => { + const t = harness(); + t.state.config = configWithRow(live); + t.state.receipt = r; + t.state.disposeResult = disposeResult; + const d = deferral(); + t.ledger.recordDeferral(TASK, d); + expect(t.ledger.deferralOf(TASK)).toBe(d); + const result = await t.ledger.reconcileDeferredTaskStreamEnd(TASK); + return { result, kept: t.ledger.hasDeferral(TASK), calls: t.calls }; + }; + expect(await run(undefined)).toEqual({ result: false, kept: false, calls: [] }); + expect(await run(receipt("streaming"))).toEqual({ result: false, kept: false, calls: [] }); + expect(await run(receipt("pending"))).toEqual({ result: false, kept: true, calls: [] }); + expect(await run(receipt("admitted"))).toEqual({ result: false, kept: true, calls: [] }); + expect(await run(receipt("canceled"))).toEqual({ + result: true, + kept: false, + calls: ["dispose", `recover:running:${ATTEMPT}`], + }); + expect(await run(receipt("prestream-failed"), false)).toEqual({ + result: true, + kept: false, + calls: ["dispose"], + }); + }); + + test("forgetDeferral drops the deferral", () => { + const t = harness(); + t.ledger.recordDeferral(TASK, deferral()); + t.ledger.forgetDeferral(TASK); + expect(t.ledger.deferralOf(TASK)).toBeUndefined(); + }); +}); diff --git a/src/node/services/streamEndLedger.ts b/src/node/services/streamEndLedger.ts new file mode 100644 index 00000000000..373c814dd0e --- /dev/null +++ b/src/node/services/streamEndLedger.ts @@ -0,0 +1,419 @@ +import assert from "node:assert/strict"; +import type { StreamEndEvent } from "@/common/types/stream"; +import { hasCompletedAgentReport } from "@/common/utils/agentTaskCompletion"; +import type { Config, Workspace as WorkspaceConfigEntry } from "@/node/config"; +import { log } from "@/node/services/log"; +import type { + AdmittedSend, + AttemptFence, + OwnedTaskAttempt, + StreamEndDecision, + StreamEndTaskOrigin, +} from "@/node/services/taskService"; +import { findWorkspaceEntry, rowSupersedes } from "@/node/services/taskUtils"; +import type { QueueCutAttributionSnapshot, WorkspaceHost } from "@/node/services/taskWorkspaceSeam"; +import type { WorkspaceGoalService } from "@/node/services/workspaceGoalService"; +import type { MutexMap } from "@/node/utils/concurrency/mutexMap"; + +/** + * A child stream end cut by a host-selected continuation (see StreamEndLedger's + * deferredTaskStreamEnds). + */ +export interface TaskStreamEndDeferral { + sourceMessageId: string; + sourceTurnGeneration: symbol; + continuationEntryId: string; + executionId: string | undefined; + stopEpoch: number; + /** + * The cut stream's attempt: a row another writer re-admitted since drops the deferral, and + * its recovery is CAS'd on this attempt (#4414). + */ + attemptId: AttemptFence; +} + +/** + * Everything the stream-end ledger 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. Stop and attempt state stays with its owners and is reached + * only through these closures. + */ +export interface StreamEndLedgerDeps { + config: () => Pick; + workspaceService: () => Pick< + WorkspaceHost, + | "getActiveTurnGeneration" + | "drainQueuedMessagesIfIdle" + | "disposeQueueCut" + | "getQueueCutReceipt" + >; + workspaceGoalService: () => + | Pick + | undefined; + workspaceEventLocks: () => Pick, "withLock">; + /** TaskService.handleStreamEnd (stays on TaskService). */ + handleStreamEnd: ( + event: StreamEndEvent, + queueCutSnapshot: QueueCutAttributionSnapshot, + taskOrigin: StreamEndTaskOrigin + ) => Promise<"deferred" | undefined>; + /** TaskService.recoverTaskFromIncompleteStreamEnd (stays on TaskService). */ + recoverTaskFromIncompleteStreamEnd: ( + workspaceId: string, + status: WorkspaceConfigEntry["taskStatus"], + streamAttemptId: AttemptFence + ) => Promise; + /** TaskService.getWorkspaceStopEpoch. */ + getWorkspaceStopEpoch: (workspaceId: string) => number; + /** TaskService.getAgentTaskExecutionId. */ + getAgentTaskExecutionId: (taskId: string) => string | null; + /** TaskStopLatch.recheckWorkspaceStopRelease. */ + recheckWorkspaceStopRelease: (workspaceId: string) => void; + /** AttemptLedger.attemptAdmissionOpen. */ + attemptAdmissionOpen: (taskId: string, attemptId: string) => boolean; + /** AttemptLedger.addAttemptSettlementListener. */ + addAttemptSettlementListener: (taskId: string, listener: () => void) => void; + /** AttemptLedger.removeAttemptSettlementListener. */ + removeAttemptSettlementListener: (taskId: string, listener: () => void) => void; + /** AttemptLedger.sendsOf: the task's admitted send obligations. */ + sendsOf: (taskId: string) => ReadonlySet; +} + +/** + * Stream-end decisions (see StreamEndDecision) of this process's owned attempts and the child + * stream ends deferred by a host-selected continuation, with the stream-end handler runs that + * resolve them. handleStreamEnd and the stream-end listener stay on TaskService. + */ +export class StreamEndLedger { + /** + * 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 + * from the recorded successor outcome (QueueCutReceipt), never from an idle probe, and cleared + * on terminal transitions. In-memory only: after a restart a running task with no stream is + * handled by startup recovery. + */ + private readonly deferredTaskStreamEnds = new Map(); + /** Per task, the stream-end decisions of this process's owned attempts (see StreamEndDecision). */ + readonly streamEndDecisionsByTaskId = new Map(); + + constructor(private readonly deps: StreamEndLedgerDeps) {} + + /** + * Run handleStreamEnd under the workspace event lock, owning the stream's decision: resolved + * `nonreport` on every return that neither published nor failed, `indeterminate` on a throw — + * unless the handler deferred the stream (child goal arbitration waiting on its accounting), + * which transfers the still-pending decision to deferChildStreamEnd. + */ + runStreamEndHandler( + event: StreamEndEvent, + queueCutSnapshot: QueueCutAttributionSnapshot, + taskOrigin: StreamEndTaskOrigin, + decision: StreamEndDecision | undefined + ): void { + void this.deps + .workspaceEventLocks() + .withLock(event.workspaceId, async () => { + let deferred = false; + try { + deferred = + (await this.deps.handleStreamEnd(event, queueCutSnapshot, taskOrigin)) === "deferred"; + } catch (error: unknown) { + // Unknown outcome: never let held follow-ups run as if the attempt continued. + this.resolveStreamEndDecision(event.workspaceId, decision, "indeterminate"); + throw error; + } finally { + // Every return path that neither published nor failed left the attempt continuing. + if (!deferred) this.resolveStreamEndDecision(event.workspaceId, decision, "nonreport"); + } + if (deferred) this.deferChildStreamEnd(event, queueCutSnapshot, taskOrigin, decision); + }) + .catch((error: unknown) => { + log.error("TaskService.handleStreamEnd failed", { error }); + }); + } + + /** + * A child stream whose goal accounting had not settled when arbitration needed it (see + * arbitrateChildGoalAtStreamEnd): accounting completion is authoritative, so nothing is decided + * from unsettled state and the task is not finalized. The stream's pending decision (held + * follow-ups keep waiting) moves here; outside the event lock this waits — unbounded — for the + * retained receipt to settle, or for the attempt to close / be superseded / be stopped (or its + * row to stop reading running), then re-runs the original stream end (same event, attempt, + * turn and stop fences) under the lock. + */ + private deferChildStreamEnd( + event: StreamEndEvent, + queueCutSnapshot: QueueCutAttributionSnapshot, + taskOrigin: StreamEndTaskOrigin, + decision: StreamEndDecision | undefined + ): void { + const goalService = this.deps.workspaceGoalService(); + assert(goalService != null, "deferChildStreamEnd requires the goal service"); + const taskId = event.workspaceId; + const attemptId = taskOrigin.ownedAttempt?.attemptId ?? taskOrigin.unownedAttemptId; + const cancelled = Promise.withResolvers(); + const listener = () => { + if ( + attemptId == null || + !this.deps.attemptAdmissionOpen(taskId, attemptId) || + taskOrigin.stopEpoch !== this.deps.getWorkspaceStopEpoch(taskId) || + taskOrigin.executionId !== this.deps.getAgentTaskExecutionId(taskId) || + // arbitrateChildGoalAtStreamEnd's status gate (its owed-pause gate needs no mirror: the + // pause marker is written only with a terminal status): a row that no longer reads + // running (a report adopted from disk, an archive) can never take a goal turn, so the + // accounting decides nothing and the re-run takes the same branch now as after it (#6994). + findWorkspaceEntry(this.deps.config().loadConfigOrDefault(), taskId)?.workspace + .taskStatus !== "running" + ) { + cancelled.resolve(); + } + }; + this.deps.addAttemptSettlementListener(taskId, listener); + listener(); + log.info("[task-goal] stream accounting unsettled; deferring the child stream end", { + taskId, + messageId: event.messageId, + }); + void Promise.race([ + goalService.streamAccountingReceiptOutcome(taskId, event.messageId), + cancelled.promise, + ]) + .finally(() => this.deps.removeAttemptSettlementListener(taskId, listener)) + .then(() => { + // #7003: this stream deferred while its row read running, and nothing moves a running row + // to a completed report under the same attempt but a publication (ours resolved the + // decision already) or another backend's. So that attempt reported: the re-run's + // `reported` branch must not leave the decision `nonreport`, or a held follow-up would + // start a new turn of a reported attempt. + this.resolveDecisionOfReportedAttempt(taskId, attemptId, decision); + this.runStreamEndHandler(event, queueCutSnapshot, taskOrigin, decision); + }) + .catch((error: unknown) => { + log.error("TaskService: deferred child stream end failed", { taskId, error }); + this.resolveStreamEndDecision(taskId, decision, "indeterminate"); + }); + } + + /** + * Register the pending decision for the ended stream of `attempt` (owned) or, when this process + * owns none, of the persisted `unownedAttemptId`. Synchronous, event tick only. + */ + registerStreamEndDecision( + taskId: string, + messageId: string, + attempt: OwnedTaskAttempt | undefined, + unownedAttemptId: string | undefined + ): StreamEndDecision | undefined { + const attemptId = attempt != null ? attempt.attemptId : unownedAttemptId; + if (attemptId == null) return undefined; + const decision: StreamEndDecision = { + attempt, + attemptId, + messageId, + outcome: "pending", + turn: this.deps.workspaceService().getActiveTurnGeneration(taskId), + }; + const decisions = this.streamEndDecisionsByTaskId.get(taskId) ?? []; + decisions.push(decision); + this.streamEndDecisionsByTaskId.set(taskId, decisions); + return decision; + } + + /** The decision a send bound to `attemptId` must honor: the pending one, else the latest. */ + findStreamEndDecision(taskId: string, attemptId: string): StreamEndDecision | undefined { + let latest: StreamEndDecision | undefined; + for (const decision of this.streamEndDecisionsByTaskId.get(taskId) ?? []) { + if (decision.attemptId !== attemptId) continue; + if (decision.outcome === "pending") return decision; + latest = decision; + } + return latest; + } + + /** + * The handler's own pending decision for exactly `attempt` — or, for an unowned stream + * (`attempt` undefined), the unowned decision for `unownedAttemptId` — none once resolved or + * replaced. + */ + findPendingStreamEndDecision( + taskId: string, + attempt: OwnedTaskAttempt | undefined, + unownedAttemptId?: string + ): StreamEndDecision | undefined { + const attemptId = attempt != null ? attempt.attemptId : unownedAttemptId; + if (attemptId == null) return undefined; + const decision = this.findStreamEndDecision(taskId, attemptId); + return decision?.outcome === "pending" && decision.attempt === attempt ? decision : undefined; + } + + /** + * Resolve exactly once (later calls, including the handler's finally, are no-ops), drop the + * decision when nothing is left to read it, and wake the workspace's idle drain so a held entry + * either dispatches (nonreport) or is refused and handed back (published/indeterminate) now + * rather than at the next unrelated drain. Returns whether this call resolved it. + */ + resolveStreamEndDecision( + taskId: string, + decision: StreamEndDecision | undefined, + outcome: Exclude + ): boolean { + if (decision?.outcome !== "pending") return false; + decision.outcome = outcome; + // A stream that published (or may have published) the report ends the attempt WITH a report: + // it can never produce a no-report receipt, even after this decision is pruned below. + if (outcome !== "nonreport" && decision.attempt != null) { + decision.attempt.receiptEligible = false; + } + log.debug("[task-attempt] stream-end decision resolved", { + taskId, + attemptId: decision.attemptId, + messageId: decision.messageId, + outcome, + }); + this.pruneStreamEndDecisions(taskId); + // A stop record of this attempt waits for the decision (recheckWorkspaceStopRelease). + this.deps.recheckWorkspaceStopRelease(taskId); + this.deps.workspaceService().drainQueuedMessagesIfIdle(taskId); + return true; + } + + /** + * Drop resolved decisions no pending/enqueued obligation of their attempt can still read — a + * terminal one only once its reporting turn settled too (a later reader can still arrive while + * that turn is live; see StreamEndDecision). + */ + pruneStreamEndDecisions(taskId: string): void { + const decisions = this.streamEndDecisionsByTaskId.get(taskId); + if (decisions == null) return; + const readers = new Set(); + for (const send of this.deps.sendsOf(taskId)) { + if (send.state === "pending" || send.state === "enqueued") readers.add(send.attemptId); + } + const kept = decisions.filter( + (decision) => + decision.outcome === "pending" || + readers.has(decision.attemptId) || + (decision.outcome !== "nonreport" && decision.turn != null) + ); + if (kept.length === 0) this.streamEndDecisionsByTaskId.delete(taskId); + else if (kept.length !== decisions.length) this.streamEndDecisionsByTaskId.set(taskId, kept); + } + + /** + * Settle a deferred cut from the recorded successor outcome. Runs under the task's + * workspaceEventLock (from handleStreamEnd, the queue-change listener, stream-abort and error + * handlers). Only `canceled`/`prestream-failed` (recover once) and `streaming` (transfer) settle + * it; `pending`/`admitted` keep waiting. A task whose execution or stop epoch moved on, or that + * left `running`, drops the deferral without recovery. + */ + async reconcileDeferredTaskStreamEnd(workspaceId: string): Promise { + const deferral = this.deferredTaskStreamEnds.get(workspaceId); + if (deferral == null) return false; + const entry = findWorkspaceEntry(this.deps.config().loadConfigOrDefault(), workspaceId); + if ( + entry?.workspace.parentWorkspaceId == null || + entry.workspace.taskStatus !== "running" || + rowSupersedes(entry.workspace, deferral.attemptId) || + entry.workspace.taskExecutionId !== deferral.executionId || + this.deps.getWorkspaceStopEpoch(workspaceId) !== deferral.stopEpoch + ) { + this.deferredTaskStreamEnds.delete(workspaceId); + this.deps.workspaceService().disposeQueueCut(workspaceId, deferral.continuationEntryId); + return false; + } + const receipt = this.deps + .workspaceService() + .getQueueCutReceipt(workspaceId, deferral.continuationEntryId); + if (receipt == null || receipt.successor === "streaming") { + this.deferredTaskStreamEnds.delete(workspaceId); + return false; + } + if (receipt.successor !== "canceled" && receipt.successor !== "prestream-failed") { + return false; + } + this.deferredTaskStreamEnds.delete(workspaceId); + if (!this.deps.workspaceService().disposeQueueCut(workspaceId, deferral.continuationEntryId)) { + return true; + } + log.debug("Deferred task stream end recovering: continuation withdrawn", { + workspaceId, + sourceMessageId: deferral.sourceMessageId, + continuationEntryId: deferral.continuationEntryId, + successor: receipt.successor, + }); + // Recovery acts for the cut stream's attempt only (its writes and prompt are CAS'd on it). + await this.deps.recoverTaskFromIncompleteStreamEnd( + workspaceId, + entry.workspace.taskStatus, + deferral.attemptId + ); + return true; + } + + /** + * The row reads a completed report for `attemptId` (#7003): that attempt reported, possibly on + * another backend. Its pending stream-end decision (`decision`, else the attempt's) resolves + * `indeterminate`, as publication's not-durable branch does: held follow-ups are refused and + * handed back, never run as a new turn of a reported attempt. Returns whether the row reads so. + */ + resolveDecisionOfReportedAttempt( + taskId: string, + attemptId: string | undefined, + decision?: StreamEndDecision + ): boolean { + if (attemptId == null) return false; + const workspace = findWorkspaceEntry( + this.deps.config().loadConfigOrDefault(), + taskId + )?.workspace; + if (workspace?.taskAttemptId !== attemptId || !hasCompletedAgentReport(workspace)) return false; + this.resolveStreamEndDecision( + taskId, + decision ?? this.findStreamEndDecision(taskId, attemptId), + "indeterminate" + ); + return true; + } + + /** + * Owner settlement of `turnGeneration` (TaskService.recordWorkspaceTurnSettled): a terminal + * decision held for this turn's late readers is kept now only by actual readers. + */ + settleTurn(workspaceId: string, turnGeneration: symbol): void { + let decisionTurnSettled = false; + for (const decision of this.streamEndDecisionsByTaskId.get(workspaceId) ?? []) { + if (decision.turn !== turnGeneration) continue; + decision.turn = undefined; + decisionTurnSettled = true; + } + if (decisionTurnSettled) this.pruneStreamEndDecisions(workspaceId); + } + + /** Supersession (TaskService.recordWorkspaceTurnSuperseded): rebind decisions to `next`. */ + rebindTurn(workspaceId: string, previous: symbol, next: symbol): void { + for (const decision of this.streamEndDecisionsByTaskId.get(workspaceId) ?? []) { + if (decision.turn === previous) decision.turn = next; + } + } + + /** A new attempt begins (AttemptLedger.beginOwnedTaskAttempt): drop the task's decisions. */ + discardStreamEndDecisions(taskId: string): void { + this.streamEndDecisionsByTaskId.delete(taskId); + } + + hasDeferral(taskId: string): boolean { + return this.deferredTaskStreamEnds.has(taskId); + } + + deferralOf(taskId: string): TaskStreamEndDeferral | undefined { + return this.deferredTaskStreamEnds.get(taskId); + } + + recordDeferral(taskId: string, deferral: TaskStreamEndDeferral): void { + this.deferredTaskStreamEnds.set(taskId, deferral); + } + + forgetDeferral(taskId: string): void { + this.deferredTaskStreamEnds.delete(taskId); + } +} diff --git a/src/node/services/taskService.childGoals.test.ts b/src/node/services/taskService.childGoals.test.ts index 5a1e8401abd..a5f56b5fb65 100644 --- a/src/node/services/taskService.childGoals.test.ts +++ b/src/node/services/taskService.childGoals.test.ts @@ -1209,7 +1209,7 @@ describe("TaskService child goals", () => { expect(await t.taskService.markInterruptedTaskRunning(childId)).toBe(true); const internal = t.taskService as unknown as { attemptLedger: { ownedAttemptByTaskId: Map }; - streamEndDecisionsByTaskId: Map>; + streamEndLedger: { streamEndDecisionsByTaskId: Map> }; }; expect(internal.attemptLedger.ownedAttemptByTaskId.get(childId)?.receiptEligible).toBe(true); await t.setChildGoal(); @@ -1220,9 +1220,9 @@ describe("TaskService child goals", () => { t.goals.beginStreamAccountingReceipt(childId, "assistant-1"); await streamEnd(t.taskService, t.proseEnd("assistant-1")); expect(t.child()?.taskStatus).toBe("running"); - expect(internal.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)).toEqual([ - "pending", - ]); + expect( + internal.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome) + ).toEqual(["pending"]); return { t, internal }; } @@ -1239,7 +1239,7 @@ describe("TaskService child goals", () => { data: { stoppedTaskIds: [childId] }, }); expect(t.taskService.isWorkspaceStopInProgress(childId)).toBe(false); - expect(internal.streamEndDecisionsByTaskId.has(childId)).toBe(false); + expect(internal.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false); expect(t.child()?.taskStatus).toBe("interrupted"); }); @@ -1259,7 +1259,7 @@ describe("TaskService child goals", () => { await waitForCondition(() => !t.taskService.isWorkspaceStopInProgress(childId), { timeoutMs: 1_000, }); - expect(internal.streamEndDecisionsByTaskId.has(childId)).toBe(false); + expect(internal.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false); }); // #6920 control: a terminate whose removal succeeds. The removal deregisters the workspace and @@ -1309,7 +1309,7 @@ describe("TaskService child goals", () => { expect(await waiter).toBe("rejected: Task terminated"); expect(rejectWaiters.mock.calls.filter((call) => call[0] === childId)).toHaveLength(1); await waitForCondition(() => !t.taskService.isWorkspaceStopInProgress(childId)); - expect(internal.streamEndDecisionsByTaskId.has(childId)).toBe(false); + expect(internal.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false); expect(removeQueuedMessagesByDedupeKeyPrefix).not.toHaveBeenCalled(); expect(t.sends()).toHaveLength(0); expect(await t.parentReports()).toHaveLength(0); @@ -1342,7 +1342,7 @@ describe("TaskService child goals", () => { await waitForCondition(() => !t.taskService.isWorkspaceStopInProgress(childId), { timeoutMs: 1_000, }); - expect(internal.streamEndDecisionsByTaskId.has(childId)).toBe(false); + expect(internal.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false); } ); @@ -1361,9 +1361,12 @@ describe("TaskService child goals", () => { expect(report.reportMarkdown).toBe("Report from another backend."); expect(t.child()?.taskStatus).toBe("reported"); - await waitForCondition(() => !internal.streamEndDecisionsByTaskId.has(childId), { - timeoutMs: 1_000, - }); + await waitForCondition( + () => !internal.streamEndLedger.streamEndDecisionsByTaskId.has(childId), + { + timeoutMs: 1_000, + } + ); // Settled, not continued: no goal turn, recovery prompt or report send for the child. expect(t.sends()).toHaveLength(0); }); @@ -1476,9 +1479,9 @@ describe("TaskService child goals", () => { expect(waited).toBe("still waiting"); expect(t.child()?.taskStatus).toBe("running"); - expect(internal.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)).toEqual([ - "pending", - ]); + expect( + internal.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome) + ).toEqual(["pending"]); }); // #7003 legacy artifacts over an interrupted row keep today's self-heal: adopted as the report. @@ -1514,9 +1517,12 @@ describe("TaskService child goals", () => { expect(report.reportMarkdown).toBe("Legacy report."); expect(t.child()?.taskStatus).toBe("reported"); - await waitForCondition(() => !internal.streamEndDecisionsByTaskId.has(childId), { - timeoutMs: 1_000, - }); + await waitForCondition( + () => !internal.streamEndLedger.streamEndDecisionsByTaskId.has(childId), + { + timeoutMs: 1_000, + } + ); }); // #7003: another backend's `reported` for the deferred stream's own attempt lands, and then any @@ -1524,7 +1530,7 @@ describe("TaskService child goals", () => { // must settle the decision as indeterminate (held follow-ups refused), not as nonreport. test("a deferral re-run after another backend reported its attempt settles as indeterminate", async () => { const { t, internal } = await deferStreamEndOfOwnedAttempt(); - const decision = internal.streamEndDecisionsByTaskId.get(childId)?.[0]; + const decision = internal.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.[0]; expect(decision?.outcome).toBe("pending"); await t.editChild((workspace) => { workspace.taskStatus = "reported"; @@ -1535,9 +1541,12 @@ describe("TaskService child goals", () => { t.taskService as unknown as { notifyAttemptSettlementListeners: (taskId: string) => void } ).notifyAttemptSettlementListeners(childId); - await waitForCondition(() => !internal.streamEndDecisionsByTaskId.has(childId), { - timeoutMs: 1_000, - }); + await waitForCondition( + () => !internal.streamEndLedger.streamEndDecisionsByTaskId.has(childId), + { + timeoutMs: 1_000, + } + ); expect(decision?.outcome).toBe("indeterminate"); expect(t.sends()).toHaveLength(0); }); @@ -1579,9 +1588,9 @@ describe("TaskService child goals", () => { expect(listeners.get(childId)?.size).toBe(1); expect(t.child()?.taskStatus).toBe("running"); - expect(internal.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)).toEqual([ - "pending", - ]); + expect( + internal.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome) + ).toEqual(["pending"]); }); // #7003: a row moved to awaiting_report while the stream end is deferred (no notify on that @@ -1597,9 +1606,12 @@ describe("TaskService child goals", () => { t.taskService as unknown as { notifyAttemptSettlementListeners: (taskId: string) => void } ).notifyAttemptSettlementListeners(childId); - await waitForCondition(() => !internal.streamEndDecisionsByTaskId.has(childId), { - timeoutMs: 1_000, - }); + await waitForCondition( + () => !internal.streamEndLedger.streamEndDecisionsByTaskId.has(childId), + { + timeoutMs: 1_000, + } + ); await t.untilStatus("reported"); expect(t.child()?.taskStatus).toBe("reported"); // The parent's delivery lands after the status write. diff --git a/src/node/services/taskService.reportDecisionHold.test.ts b/src/node/services/taskService.reportDecisionHold.test.ts index 1a9d617d2ad..b4d16fa4db5 100644 --- a/src/node/services/taskService.reportDecisionHold.test.ts +++ b/src/node/services/taskService.reportDecisionHold.test.ts @@ -54,7 +54,9 @@ interface Internals { attemptSettlementByTaskId: Map; admittedSendsByTaskId: Map>; }; - streamEndDecisionsByTaskId: Map>; + streamEndLedger: { + streamEndDecisionsByTaskId: Map>; + }; taskStopLatch: { workspaceStopRecords: Map }; emitWorkspaceMetadata: (workspaceId: string) => Promise; workspaceEventLocks: { withLock(key: string, operation: () => Promise): Promise }; @@ -346,7 +348,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => { // The session's terminal drain runs while TaskService still decides: the entry is HELD // — still queued, no turn claimed, nothing dispatched. await yieldMacrotasks(3); - expect(svc.streamEndDecisionsByTaskId.get(childId)?.[0]).toMatchObject({ + expect(svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.[0]).toMatchObject({ attemptId: attemptA, }); } @@ -368,7 +370,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => { expect(svc.attemptLedger.ownedAttemptByTaskId.has(childId)).toBe(false); // The follow-up's obligation was discharged exactly once and the decision pruned. expect(outstanding(svc, childId)).toHaveLength(0); - expect(svc.streamEndDecisionsByTaskId.has(childId)).toBe(false); + expect(svc.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false); // The follow-up is held by the session (nothing is pushed into the composer), refused // because the task reported — and published with that reason. expect(stack.heldTexts()).toEqual(["follow-up text"]); @@ -507,7 +509,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => { // Its obligation was discharged, and the decision is gone once the reporting turn settled // (bounded: nothing retained past the turn it was decided for). expect(outstanding(svc, childId)).toHaveLength(0); - expect(svc.streamEndDecisionsByTaskId.has(childId)).toBe(false); + expect(svc.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false); // After the turn settled, a manual send is a new admission that mints a fresh attempt. expect(await taskService.markInterruptedTaskRunning(childId)).toBe(false); expect(entryOf(config, childId)?.taskAttemptId).not.toBe(attemptA); @@ -855,7 +857,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => { ); // The decision reads `nonreport` for as long as a queued reader (the nudge) is bound to A. expect( - (svc.streamEndDecisionsByTaskId.get(childId) ?? []).every( + (svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId) ?? []).every( (decision) => decision.outcome === "nonreport" ) ).toBe(true); @@ -896,7 +898,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => { // Refused as indeterminate — not as a completed report and not as a plain stale attempt. expect(onCanceled).toHaveBeenCalledWith(TASK_REPORT_OUTCOME_INDETERMINATE_UNSENT_MESSAGE); expect(outstanding(svc, childId)).toHaveLength(0); - expect(svc.streamEndDecisionsByTaskId.has(childId)).toBe(false); + expect(svc.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false); } finally { await stack.cleanup(); } @@ -951,13 +953,14 @@ describe("report-decision hold for queued follow-ups (real host)", () => { ).toEqual(Ok(undefined)); stack.endStream(0, {}, true); await until( - () => svc.streamEndDecisionsByTaskId.get(childId)?.[0]?.outcome === "pending", + () => + svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.[0]?.outcome === "pending", "deferred decision" ); await yieldMacrotasks(30); - expect(svc.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)).toEqual([ - "pending", - ]); + expect( + svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome) + ).toEqual(["pending"]); expect(completions).toHaveLength(1); if (rowStatus === "reported") { @@ -989,7 +992,8 @@ describe("report-decision hold for queued follow-ups (real host)", () => { }); expect(report.reportMarkdown).toBe("Report from another backend."); await until( - () => svc.streamEndDecisionsByTaskId.get(childId)?.[0]?.outcome !== "pending", + () => + svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.[0]?.outcome !== "pending", "decision settled" ); await until(() => !workspaceService.hasQueuedMessages(childId), "queue drained"); @@ -1055,7 +1059,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => { expect(completions).toHaveLength(1); expect(stack.heldTexts()).toEqual(["follow-up text"]); expect(outstanding(svc, childId)).toHaveLength(0); - expect(svc.streamEndDecisionsByTaskId.has(childId)).toBe(false); + expect(svc.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false); } finally { gate.resolve(); await stack.cleanup(); @@ -1175,7 +1179,9 @@ describe("report-decision hold for queued follow-ups (real host)", () => { // Held: still queued, obligation still enqueued, decision pending. expect(workspaceService.hasQueuedMessages(childId)).toBe(true); expect(outstanding(svc, childId).map((send) => send.state)).toEqual(["enqueued"]); - expect(svc.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)).toEqual(["pending"]); + expect( + svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome) + ).toEqual(["pending"]); return { gate, attemptA, onCanceled }; } @@ -1205,13 +1211,16 @@ describe("report-decision hold for queued follow-ups (real host)", () => { expect(onCanceled).toHaveBeenCalledWith("Queued message cleared before dispatch."); // The obligation is discharged by the clear; the decision still belongs to the handler. expect(outstanding(svc, childId)).toHaveLength(0); - expect(svc.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome)).toEqual([ - "pending", - ]); + expect( + svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome) + ).toEqual(["pending"]); expect(completions).toHaveLength(1); // Handler resumes: publishes and releases A; no reader is left, so the decision is dropped. gate.resolve(); - await until(() => !svc.streamEndDecisionsByTaskId.has(childId), "decision pruned"); + await until( + () => !svc.streamEndLedger.streamEndDecisionsByTaskId.has(childId), + "decision pruned" + ); await until(() => !svc.attemptLedger.ownedAttemptByTaskId.has(childId), "attempt released"); await yieldMacrotasks(5); expect(entryOf(config, childId)).toMatchObject({ @@ -1256,12 +1265,15 @@ describe("report-decision hold for queued follow-ups (real host)", () => { expect(outstanding(svc, childId)).toHaveLength(0); 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", - ]); + expect( + svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId)?.map((d) => d.outcome) + ).toEqual(["pending"]); expect(completions).toHaveLength(1); gate.resolve(); - await until(() => !svc.streamEndDecisionsByTaskId.has(childId), "decision pruned"); + await until( + () => !svc.streamEndLedger.streamEndDecisionsByTaskId.has(childId), + "decision pruned" + ); await until(() => !svc.attemptLedger.ownedAttemptByTaskId.has(childId), "attempt released"); await yieldMacrotasks(5); expect(entryOf(config, childId)).toMatchObject({ @@ -1351,7 +1363,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => { expect(completions).toHaveLength(1); expect(stack.heldTexts()).toEqual(["follow-up text"]); expect(outstanding(svc, childId)).toHaveLength(0); - expect(svc.streamEndDecisionsByTaskId.has(childId)).toBe(false); + expect(svc.streamEndLedger.streamEndDecisionsByTaskId.has(childId)).toBe(false); // The hold came from the decision alone: no ownership (settlement/receipt authority) granted. expect(svc.attemptLedger.ownedAttemptByTaskId.has(childId)).toBe(false); expect(entryOf(config, childId)?.taskAttemptId).toBe(redriven); @@ -1394,7 +1406,9 @@ describe("report-decision hold for queued follow-ups (real host)", () => { stack.endStream(0, { report: "done by the re-drive" }, true); // No decision of the ended stream is ever keyed by B. expect( - (svc.streamEndDecisionsByTaskId.get(childId) ?? []).map((decision) => decision.attemptId) + (svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId) ?? []).map( + (decision) => decision.attemptId + ) ).not.toContain(foreign); await until(() => completions.length === 2, "B's follow-up dispatched"); await yieldMacrotasks(20); @@ -1457,7 +1471,9 @@ describe("report-decision hold for queued follow-ups (real host)", () => { await until(() => rotated, "the publication write"); await until( () => - !(svc.streamEndDecisionsByTaskId.get(childId) ?? []).some((d) => d.outcome === "pending"), + !(svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId) ?? []).some( + (d) => d.outcome === "pending" + ), "the decision resolved" ); await yieldMacrotasks(20); @@ -1472,7 +1488,9 @@ describe("report-decision hold for queued follow-ups (real host)", () => { ).toBeNull(); // The re-drive's decision stayed keyed by its own attempt, and no ownership was granted. expect( - (svc.streamEndDecisionsByTaskId.get(childId) ?? []).map((decision) => decision.attemptId) + (svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId) ?? []).map( + (decision) => decision.attemptId + ) ).not.toContain(foreign); expect(redriven).not.toBe(foreign); expect(svc.attemptLedger.ownedAttemptByTaskId.has(childId)).toBe(false); @@ -1533,7 +1551,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => { await until(() => rotated, "the rotation"); await until( () => - !(svc.streamEndDecisionsByTaskId.get(childId) ?? []).some( + !(svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId) ?? []).some( (decision) => decision.outcome === "pending" ), "the decision resolved" @@ -1658,7 +1676,9 @@ describe("report-decision hold for queued follow-ups (real host)", () => { await rotation; await until( () => - !(svc.streamEndDecisionsByTaskId.get(childId) ?? []).some((d) => d.outcome === "pending"), + !(svc.streamEndLedger.streamEndDecisionsByTaskId.get(childId) ?? []).some( + (d) => d.outcome === "pending" + ), "the decision resolved" ); await yieldMacrotasks(20); diff --git a/src/node/services/taskService.settlementReceipts.test.ts b/src/node/services/taskService.settlementReceipts.test.ts index 90c11ff19a6..41646bf0b00 100644 --- a/src/node/services/taskService.settlementReceipts.test.ts +++ b/src/node/services/taskService.settlementReceipts.test.ts @@ -83,7 +83,9 @@ interface Internals { emitWorkspaceMetadata: (workspaceId: string) => Promise; writeSettlementReceipt: (...args: unknown[]) => Promise; workspaceEventLocks: { withLock: (key: string, fn: () => Promise) => Promise }; - streamEndDecisionsByTaskId: Map>; + streamEndLedger: { + streamEndDecisionsByTaskId: Map>; + }; } const internals = (service: TaskService) => service as unknown as Internals; @@ -1202,9 +1204,9 @@ describe("TaskService settlement receipt producers (G2)", () => { const gate = Promise.withResolvers(); const held = svc.workspaceEventLocks.withLock(taskId, () => gate.promise); emitter.emit("stream-end", streamEnd(taskId, "assistant-1")); - expect(svc.streamEndDecisionsByTaskId.get(taskId)?.map((d) => d.outcome)).toEqual([ - "pending", - ]); + expect( + svc.streamEndLedger.streamEndDecisionsByTaskId.get(taskId)?.map((d) => d.outcome) + ).toEqual(["pending"]); const { stopped } = await startStopAwaitingRelease(taskService, rootId, taskId); // Every other release condition holds once the cleanup paid back; the pending decision // alone keeps the latch, and no receipt write has even begun. @@ -1249,16 +1251,16 @@ describe("TaskService settlement receipt producers (G2)", () => { }); emitter.emit("stream-end", streamEnd(taskId, "assistant-1", "done")); await waitForCondition(() => blocked); - expect(svc.streamEndDecisionsByTaskId.get(taskId)?.map((d) => d.outcome)).toEqual([ - "pending", - ]); + expect( + svc.streamEndLedger.streamEndDecisionsByTaskId.get(taskId)?.map((d) => d.outcome) + ).toEqual(["pending"]); const { stopped } = await startStopAwaitingRelease(taskService, rootId, taskId); await new Promise((resolve) => setImmediate(resolve)); await expectNoReceipt(config, taskId, attemptId); gate.resolve(); expect((await stopped).success).toBe(true); expect(taskService.isWorkspaceStopInProgress(taskId)).toBe(false); - await waitForCondition(() => !svc.streamEndDecisionsByTaskId.has(taskId)); + await waitForCondition(() => !svc.streamEndLedger.streamEndDecisionsByTaskId.has(taskId)); expect(entryOf(config, taskId)).toMatchObject({ taskStatus: "reported", taskAttemptId: attemptId, diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 7815f493c96..7851a07f86f 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -105,6 +105,7 @@ import { log } from "@/node/services/log"; import { ATTEMPT_CLOSURE_SETTLE_WAIT_MS, AttemptLedger } from "@/node/services/attemptLedger"; export { ATTEMPT_CLOSURE_SETTLE_WAIT_MS } from "@/node/services/attemptLedger"; import { TaskStopLatch } from "@/node/services/taskStopLatch"; +import { StreamEndLedger } from "@/node/services/streamEndLedger"; import { PeerSendParking } from "@/node/services/peerSendParking"; import { TaskWaiters, @@ -1368,7 +1369,7 @@ const QUEUED_LAUNCH_LEASE_RETRY_MS = 1_000; type ChildGoalTurnOutcome = "none" | "handled" | "failed" | "deferred"; /** The ended stream's task origin, captured in the stream-end event's own tick. */ -interface StreamEndTaskOrigin { +export interface StreamEndTaskOrigin { executionId: string | null; stopEpoch: number; ownedAttempt: OwnedTaskAttempt | undefined; @@ -1479,7 +1480,7 @@ function markedTerminalFailure( * a CAS on it), or `null` when the caller has no proven attempt (a legacy/pre-identity row, a * parent- or timer-driven decision), which keeps today's unconditional writes. */ -type AttemptFence = string | null; +export type AttemptFence = string | null; /** * Who a stream event belongs to, read in the event's own tick (resolveStreamAttemptAtEvent): @@ -1899,30 +1900,7 @@ export class TaskService implements AgentTaskIntegration { private readonly peerSendParking: PeerSendParking; private readonly taskStopLatch: TaskStopLatch; private readonly attemptLedger: AttemptLedger; - /** - * 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 - * from the recorded successor outcome (QueueCutReceipt), never from an idle probe, and cleared - * on terminal transitions. In-memory only: after a restart a running task with no stream is - * handled by startup recovery. - */ - private readonly deferredTaskStreamEnds = new Map< - string, - { - sourceMessageId: string; - sourceTurnGeneration: symbol; - continuationEntryId: string; - executionId: string | undefined; - stopEpoch: number; - /** - * The cut stream's attempt: a row another writer re-admitted since drops the deferral, and - * its recovery is CAS'd on this attempt (#4414). - */ - attemptId: AttemptFence; - } - >(); - /** Per task, the stream-end decisions of this process's owned attempts (see StreamEndDecision). */ - private readonly streamEndDecisionsByTaskId = new Map(); + private readonly streamEndLedger: StreamEndLedger; /** Tracks consecutive auto-resumes per workspace. Reset when a user message is sent. */ private consecutiveAutoResumes = new Map(); @@ -2287,13 +2265,7 @@ export class TaskService implements AgentTaskIntegration { } } // A terminal decision held for this turn's late readers: only actual readers keep it now. - let decisionTurnSettled = false; - for (const decision of this.streamEndDecisionsByTaskId.get(workspaceId) ?? []) { - if (decision.turn !== turnGeneration) continue; - decision.turn = undefined; - decisionTurnSettled = true; - } - if (decisionTurnSettled) this.pruneStreamEndDecisions(workspaceId); + this.streamEndLedger.settleTurn(workspaceId, turnGeneration); this.attemptLedger.recheckResumeFailureSettlement(workspaceId); this.taskStopLatch.releaseCapturedTurn(workspaceId, turnGeneration); } @@ -2312,9 +2284,7 @@ export class TaskService implements AgentTaskIntegration { send.inherited = true; } } - for (const decision of this.streamEndDecisionsByTaskId.get(workspaceId) ?? []) { - if (decision.turn === previous) decision.turn = next; - } + this.streamEndLedger.rebindTurn(workspaceId, previous, next); this.taskStopLatch.rebindCapturedTurn(workspaceId, previous, next); } @@ -2364,210 +2334,6 @@ export class TaskService implements AgentTaskIntegration { return this.attemptLedger.writeSettlementReceipt(taskId, target, source); } - // --------------------------------------------------------------------------------------------- - // Stream-end decisions (see StreamEndDecision) - // --------------------------------------------------------------------------------------------- - - /** - * Run handleStreamEnd under the workspace event lock, owning the stream's decision: resolved - * `nonreport` on every return that neither published nor failed, `indeterminate` on a throw — - * unless the handler deferred the stream (child goal arbitration waiting on its accounting), - * which transfers the still-pending decision to deferChildStreamEnd. - */ - private runStreamEndHandler( - event: StreamEndEvent, - queueCutSnapshot: QueueCutAttributionSnapshot, - taskOrigin: StreamEndTaskOrigin, - decision: StreamEndDecision | undefined - ): void { - void this.workspaceEventLocks - .withLock(event.workspaceId, async () => { - let deferred = false; - try { - deferred = - (await this.handleStreamEnd(event, queueCutSnapshot, taskOrigin)) === "deferred"; - } catch (error: unknown) { - // Unknown outcome: never let held follow-ups run as if the attempt continued. - this.resolveStreamEndDecision(event.workspaceId, decision, "indeterminate"); - throw error; - } finally { - // Every return path that neither published nor failed left the attempt continuing. - if (!deferred) this.resolveStreamEndDecision(event.workspaceId, decision, "nonreport"); - } - if (deferred) this.deferChildStreamEnd(event, queueCutSnapshot, taskOrigin, decision); - }) - .catch((error: unknown) => { - log.error("TaskService.handleStreamEnd failed", { error }); - }); - } - - /** - * A child stream whose goal accounting had not settled when arbitration needed it (see - * arbitrateChildGoalAtStreamEnd): accounting completion is authoritative, so nothing is decided - * from unsettled state and the task is not finalized. The stream's pending decision (held - * follow-ups keep waiting) moves here; outside the event lock this waits — unbounded — for the - * retained receipt to settle, or for the attempt to close / be superseded / be stopped (or its - * row to stop reading running), then re-runs the original stream end (same event, attempt, - * turn and stop fences) under the lock. - */ - private deferChildStreamEnd( - event: StreamEndEvent, - queueCutSnapshot: QueueCutAttributionSnapshot, - taskOrigin: StreamEndTaskOrigin, - decision: StreamEndDecision | undefined - ): void { - const goalService = this.workspaceGoalService; - assert(goalService != null, "deferChildStreamEnd requires the goal service"); - const taskId = event.workspaceId; - const attemptId = taskOrigin.ownedAttempt?.attemptId ?? taskOrigin.unownedAttemptId; - const cancelled = Promise.withResolvers(); - const listener = () => { - if ( - attemptId == null || - !this.attemptLedger.attemptAdmissionOpen(taskId, attemptId) || - taskOrigin.stopEpoch !== this.getWorkspaceStopEpoch(taskId) || - taskOrigin.executionId !== this.getAgentTaskExecutionId(taskId) || - // arbitrateChildGoalAtStreamEnd's status gate (its owed-pause gate needs no mirror: the - // pause marker is written only with a terminal status): a row that no longer reads - // running (a report adopted from disk, an archive) can never take a goal turn, so the - // accounting decides nothing and the re-run takes the same branch now as after it (#6994). - findWorkspaceEntry(this.config.loadConfigOrDefault(), taskId)?.workspace.taskStatus !== - "running" - ) { - cancelled.resolve(); - } - }; - this.attemptLedger.addAttemptSettlementListener(taskId, listener); - listener(); - log.info("[task-goal] stream accounting unsettled; deferring the child stream end", { - taskId, - messageId: event.messageId, - }); - void Promise.race([ - goalService.streamAccountingReceiptOutcome(taskId, event.messageId), - cancelled.promise, - ]) - .finally(() => this.attemptLedger.removeAttemptSettlementListener(taskId, listener)) - .then(() => { - // #7003: this stream deferred while its row read running, and nothing moves a running row - // to a completed report under the same attempt but a publication (ours resolved the - // decision already) or another backend's. So that attempt reported: the re-run's - // `reported` branch must not leave the decision `nonreport`, or a held follow-up would - // start a new turn of a reported attempt. - this.resolveDecisionOfReportedAttempt(taskId, attemptId, decision); - this.runStreamEndHandler(event, queueCutSnapshot, taskOrigin, decision); - }) - .catch((error: unknown) => { - log.error("TaskService: deferred child stream end failed", { taskId, error }); - this.resolveStreamEndDecision(taskId, decision, "indeterminate"); - }); - } - - /** - * Register the pending decision for the ended stream of `attempt` (owned) or, when this process - * owns none, of the persisted `unownedAttemptId`. Synchronous, event tick only. - */ - private registerStreamEndDecision( - taskId: string, - messageId: string, - attempt: OwnedTaskAttempt | undefined, - unownedAttemptId: string | undefined - ): StreamEndDecision | undefined { - const attemptId = attempt != null ? attempt.attemptId : unownedAttemptId; - if (attemptId == null) return undefined; - const decision: StreamEndDecision = { - attempt, - attemptId, - messageId, - outcome: "pending", - turn: this.workspaceService.getActiveTurnGeneration(taskId), - }; - const decisions = this.streamEndDecisionsByTaskId.get(taskId) ?? []; - decisions.push(decision); - this.streamEndDecisionsByTaskId.set(taskId, decisions); - return decision; - } - - /** The decision a send bound to `attemptId` must honor: the pending one, else the latest. */ - private findStreamEndDecision(taskId: string, attemptId: string): StreamEndDecision | undefined { - let latest: StreamEndDecision | undefined; - for (const decision of this.streamEndDecisionsByTaskId.get(taskId) ?? []) { - if (decision.attemptId !== attemptId) continue; - if (decision.outcome === "pending") return decision; - latest = decision; - } - return latest; - } - - /** - * The handler's own pending decision for exactly `attempt` — or, for an unowned stream - * (`attempt` undefined), the unowned decision for `unownedAttemptId` — none once resolved or - * replaced. - */ - private findPendingStreamEndDecision( - taskId: string, - attempt: OwnedTaskAttempt | undefined, - unownedAttemptId?: string - ): StreamEndDecision | undefined { - const attemptId = attempt != null ? attempt.attemptId : unownedAttemptId; - if (attemptId == null) return undefined; - const decision = this.findStreamEndDecision(taskId, attemptId); - return decision?.outcome === "pending" && decision.attempt === attempt ? decision : undefined; - } - - /** - * Resolve exactly once (later calls, including the handler's finally, are no-ops), drop the - * decision when nothing is left to read it, and wake the workspace's idle drain so a held entry - * either dispatches (nonreport) or is refused and handed back (published/indeterminate) now - * rather than at the next unrelated drain. Returns whether this call resolved it. - */ - private resolveStreamEndDecision( - taskId: string, - decision: StreamEndDecision | undefined, - outcome: Exclude - ): boolean { - if (decision?.outcome !== "pending") return false; - decision.outcome = outcome; - // A stream that published (or may have published) the report ends the attempt WITH a report: - // it can never produce a no-report receipt, even after this decision is pruned below. - if (outcome !== "nonreport" && decision.attempt != null) { - decision.attempt.receiptEligible = false; - } - log.debug("[task-attempt] stream-end decision resolved", { - taskId, - attemptId: decision.attemptId, - messageId: decision.messageId, - outcome, - }); - this.pruneStreamEndDecisions(taskId); - // A stop record of this attempt waits for the decision (recheckWorkspaceStopRelease). - this.taskStopLatch.recheckWorkspaceStopRelease(taskId); - this.workspaceService.drainQueuedMessagesIfIdle(taskId); - return true; - } - - /** - * Drop resolved decisions no pending/enqueued obligation of their attempt can still read — a - * terminal one only once its reporting turn settled too (a later reader can still arrive while - * that turn is live; see StreamEndDecision). - */ - private pruneStreamEndDecisions(taskId: string): void { - const decisions = this.streamEndDecisionsByTaskId.get(taskId); - if (decisions == null) return; - const readers = new Set(); - for (const send of this.attemptLedger.sendsOf(taskId)) { - if (send.state === "pending" || send.state === "enqueued") readers.add(send.attemptId); - } - const kept = decisions.filter( - (decision) => - decision.outcome === "pending" || - readers.has(decision.attemptId) || - (decision.outcome !== "nonreport" && decision.turn != null) - ); - if (kept.length === 0) this.streamEndDecisionsByTaskId.delete(taskId); - else if (kept.length !== decisions.length) this.streamEndDecisionsByTaskId.set(taskId, kept); - } - /** Tests call this; AttemptLedger owns the listeners. */ private notifyAttemptSettlementListeners(taskId: string): void { this.attemptLedger.notifyAttemptSettlementListeners(taskId); @@ -3039,7 +2805,7 @@ export class TaskService implements AgentTaskIntegration { * Delayed source events are separately fenced by their execution identity and Stop epoch. */ private clearTaskRecovery(taskId: string): void { - this.deferredTaskStreamEnds.delete(taskId); + this.streamEndLedger.forgetDeferral(taskId); this.workspaceService.clearQueueCutReceipts(taskId); const removal = this.workspaceService.removeQueuedMessagesByDedupeKeyPrefix( taskId, @@ -3178,10 +2944,11 @@ export class TaskService implements AgentTaskIntegration { editWorkspaceEntry: (workspaceId, updater, options) => this.editWorkspaceEntry(workspaceId, updater, options), settleChildGoalPause: (workspaceId) => this.settleChildGoalPause(workspaceId), - findStreamEndDecision: (taskId, attemptId) => this.findStreamEndDecision(taskId, attemptId), - pruneStreamEndDecisions: (taskId) => this.pruneStreamEndDecisions(taskId), + findStreamEndDecision: (taskId, attemptId) => + this.streamEndLedger.findStreamEndDecision(taskId, attemptId), + pruneStreamEndDecisions: (taskId) => this.streamEndLedger.pruneStreamEndDecisions(taskId), discardStreamEndDecisions: (taskId) => { - this.streamEndDecisionsByTaskId.delete(taskId); + this.streamEndLedger.discardStreamEndDecisions(taskId); }, // Through TaskService: tests spy on inspectAttemptOutcome on the instance. inspectAttemptOutcome: (taskId, options) => this.inspectAttemptOutcome(taskId, options), @@ -3214,7 +2981,8 @@ export class TaskService implements AgentTaskIntegration { admittedSends: (taskId) => this.attemptLedger.sendsOf(taskId), ownsReceiptEligibleAttempt: (taskId, attempt) => this.attemptLedger.ownsReceiptEligibleAttempt(taskId, attempt), - findStreamEndDecision: (taskId, attemptId) => this.findStreamEndDecision(taskId, attemptId), + findStreamEndDecision: (taskId, attemptId) => + this.streamEndLedger.findStreamEndDecision(taskId, attemptId), decideSettlementReceipt: (taskId, attempt, settlementSource) => this.decideSettlementReceipt(taskId, attempt, settlementSource), writeSettlementReceipt: (taskId, target, source) => @@ -3226,6 +2994,27 @@ export class TaskService implements AgentTaskIntegration { recheckResumeFailureSettlement: (taskId) => this.attemptLedger.recheckResumeFailureSettlement(taskId), }); + this.streamEndLedger = new StreamEndLedger({ + config: () => this.config, + workspaceService: () => this.workspaceService, + workspaceGoalService: () => this.workspaceGoalService, + workspaceEventLocks: () => this.workspaceEventLocks, + handleStreamEnd: (event, queueCutSnapshot, taskOrigin) => + this.handleStreamEnd(event, queueCutSnapshot, taskOrigin), + recoverTaskFromIncompleteStreamEnd: (workspaceId, status, streamAttemptId) => + this.recoverTaskFromIncompleteStreamEnd(workspaceId, status, streamAttemptId), + getWorkspaceStopEpoch: (workspaceId) => this.getWorkspaceStopEpoch(workspaceId), + getAgentTaskExecutionId: (taskId) => this.getAgentTaskExecutionId(taskId), + recheckWorkspaceStopRelease: (workspaceId) => + this.taskStopLatch.recheckWorkspaceStopRelease(workspaceId), + attemptAdmissionOpen: (taskId, attemptId) => + this.attemptLedger.attemptAdmissionOpen(taskId, attemptId), + addAttemptSettlementListener: (taskId, listener) => + this.attemptLedger.addAttemptSettlementListener(taskId, listener), + removeAttemptSettlementListener: (taskId, listener) => + this.attemptLedger.removeAttemptSettlementListener(taskId, listener), + sendsOf: (taskId) => this.attemptLedger.sendsOf(taskId), + }); this.peerSendParking = new PeerSendParking({ flush: (targetId) => this.flushParkedPeerSends(targetId), getWorkspaceTurnManager: () => this.getWorkspaceTurnManager(), @@ -3282,13 +3071,13 @@ export class TaskService implements AgentTaskIntegration { }; // The decision this handler owes for the attempt's ended stream, registered in the event's // own tick so the session's turn-completion drain (a later microtask) finds it. - const decision = this.registerStreamEndDecision( + const decision = this.streamEndLedger.registerStreamEndDecision( payload.workspaceId, payload.messageId, taskOrigin.ownedAttempt, taskOrigin.unownedAttemptId ); - this.runStreamEndHandler(payload, queueCutSnapshot, taskOrigin, decision); + this.streamEndLedger.runStreamEndHandler(payload, queueCutSnapshot, taskOrigin, decision); }); this.aiService.on("stream-abort", (payload: unknown) => { @@ -3331,10 +3120,10 @@ export class TaskService implements AgentTaskIntegration { // Successor outcomes (withdrawn, admitted, streaming) are published as queue changes. this.workspaceService.onQueuedMessageChanged((workspaceId) => { - if (!this.deferredTaskStreamEnds.has(workspaceId)) return; + if (!this.streamEndLedger.hasDeferral(workspaceId)) return; this.workspaceEventLocks .withLock(workspaceId, async () => { - await this.reconcileDeferredTaskStreamEnd(workspaceId); + await this.streamEndLedger.reconcileDeferredTaskStreamEnd(workspaceId); }) .catch((error: unknown) => { log.error("TaskService.reconcileDeferredTaskStreamEnd failed", { error }); @@ -16316,9 +16105,9 @@ export class TaskService implements AgentTaskIntegration { // BEFORE the recovery/continuation sends below, so a user follow-up queued during the stream // still dispatches ahead of them, exactly as it did before the hold existed. if (reportArgs == null && !(isPlanLike && proposePlanResult) && status !== "interrupted") { - this.resolveStreamEndDecision( + this.streamEndLedger.resolveStreamEndDecision( workspaceId, - this.findPendingStreamEndDecision( + this.streamEndLedger.findPendingStreamEndDecision( workspaceId, taskOrigin.ownedAttempt, taskOrigin.unownedAttemptId @@ -16437,9 +16226,9 @@ export class TaskService implements AgentTaskIntegration { status, streamAttemptId, () => - this.resolveStreamEndDecision( + this.streamEndLedger.resolveStreamEndDecision( workspaceId, - this.findPendingStreamEndDecision( + this.streamEndLedger.findPendingStreamEndDecision( workspaceId, taskOrigin.ownedAttempt, taskOrigin.unownedAttemptId @@ -16485,9 +16274,9 @@ export class TaskService implements AgentTaskIntegration { // exec. Decide (and wake the held queue) before the handoff, as for any other non-report // end above: the kickoff's admission reads a still-pending decision for its attempt as // stale and would be refused, leaving the task running with no exec turn. - this.resolveStreamEndDecision( + this.streamEndLedger.resolveStreamEndDecision( workspaceId, - this.findPendingStreamEndDecision( + this.streamEndLedger.findPendingStreamEndDecision( workspaceId, taskOrigin.ownedAttempt, taskOrigin.unownedAttemptId @@ -16507,8 +16296,8 @@ export class TaskService implements AgentTaskIntegration { discardCutReceipt(); return; } - if (await this.reconcileDeferredTaskStreamEnd(workspaceId)) return; - const pendingCut = this.deferredTaskStreamEnds.get(workspaceId); + if (await this.streamEndLedger.reconcileDeferredTaskStreamEnd(workspaceId)) return; + const pendingCut = this.streamEndLedger.deferralOf(workspaceId); // A different turn can be admitted ahead of the designated continuation. Its incomplete // end does not withdraw that continuation or transfer the old cut's recovery obligation. if (pendingCut != null && pendingCut.sourceTurnGeneration !== queueCutSnapshot.turnGeneration) { @@ -16536,7 +16325,7 @@ export class TaskService implements AgentTaskIntegration { expectedAttemptId: streamAttemptId, }); } - this.deferredTaskStreamEnds.set(workspaceId, { + this.streamEndLedger.recordDeferral(workspaceId, { sourceMessageId: event.messageId, sourceTurnGeneration: receipt.sourceTurnGeneration, continuationEntryId, @@ -16545,7 +16334,7 @@ export class TaskService implements AgentTaskIntegration { attemptId: streamAttemptId, }); // The successor may have been withdrawn between the cut and this classification. - await this.reconcileDeferredTaskStreamEnd(workspaceId); + await this.streamEndLedger.reconcileDeferredTaskStreamEnd(workspaceId); return; } } @@ -17135,58 +16924,6 @@ export class TaskService implements AgentTaskIntegration { }); } - /** - * Settle a deferred cut from the recorded successor outcome. Runs under the task's - * workspaceEventLock (from handleStreamEnd, the queue-change listener, stream-abort and error - * handlers). Only `canceled`/`prestream-failed` (recover once) and `streaming` (transfer) settle - * it; `pending`/`admitted` keep waiting. A task whose execution or stop epoch moved on, or that - * left `running`, drops the deferral without recovery. - */ - private async reconcileDeferredTaskStreamEnd(workspaceId: string): Promise { - const deferral = this.deferredTaskStreamEnds.get(workspaceId); - if (deferral == null) return false; - const entry = findWorkspaceEntry(this.config.loadConfigOrDefault(), workspaceId); - if ( - entry?.workspace.parentWorkspaceId == null || - entry.workspace.taskStatus !== "running" || - rowSupersedes(entry.workspace, deferral.attemptId) || - entry.workspace.taskExecutionId !== deferral.executionId || - this.getWorkspaceStopEpoch(workspaceId) !== deferral.stopEpoch - ) { - this.deferredTaskStreamEnds.delete(workspaceId); - this.workspaceService.disposeQueueCut(workspaceId, deferral.continuationEntryId); - return false; - } - const receipt = this.workspaceService.getQueueCutReceipt( - workspaceId, - deferral.continuationEntryId - ); - if (receipt == null || receipt.successor === "streaming") { - this.deferredTaskStreamEnds.delete(workspaceId); - return false; - } - if (receipt.successor !== "canceled" && receipt.successor !== "prestream-failed") { - return false; - } - this.deferredTaskStreamEnds.delete(workspaceId); - if (!this.workspaceService.disposeQueueCut(workspaceId, deferral.continuationEntryId)) { - return true; - } - log.debug("Deferred task stream end recovering: continuation withdrawn", { - workspaceId, - sourceMessageId: deferral.sourceMessageId, - continuationEntryId: deferral.continuationEntryId, - successor: receipt.successor, - }); - // Recovery acts for the cut stream's attempt only (its writes and prompt are CAS'd on it). - await this.recoverTaskFromIncompleteStreamEnd( - workspaceId, - entry.workspace.taskStatus, - deferral.attemptId - ); - return true; - } - private async handleStreamAbort( event: StreamAbortEvent, /** Captured by the stream-abort listener in the event's own tick, before the lock wait. */ @@ -17198,7 +16935,7 @@ export class TaskService implements AgentTaskIntegration { this.bumpWorkspaceStopEpoch(event.workspaceId); this.clearTaskRecovery(event.workspaceId); } else { - await this.reconcileDeferredTaskStreamEnd(event.workspaceId); + await this.streamEndLedger.reconcileDeferredTaskStreamEnd(event.workspaceId); } // Settles a continuation handle (execution mirror) first. A reawakened child is ALSO // `running` in its stable status (markInterruptedTaskRunning), and the desktop ledger treats @@ -17429,7 +17166,7 @@ export class TaskService implements AgentTaskIntegration { // A startup error can wait behind the source handler until its continuation failure is // recorded. That disposition already requested recovery; do not spend the budget twice. - if (await this.reconcileDeferredTaskStreamEnd(workspaceId)) return; + if (await this.streamEndLedger.reconcileDeferredTaskStreamEnd(workspaceId)) return; if (status !== "awaiting_report") { // Retryable errors during `running` are handled by the agent session's @@ -17918,9 +17655,9 @@ export class TaskService implements AgentTaskIntegration { // continues. Decide (and wake the held queue) before the completion prompt, as for any // other non-report end: the prompt's admission reads a still-pending decision for its // attempt as stale and would be refused, leaving the task awaiting a report forever. - this.resolveStreamEndDecision( + this.streamEndLedger.resolveStreamEndDecision( args.workspaceId, - this.findPendingStreamEndDecision( + this.streamEndLedger.findPendingStreamEndDecision( args.workspaceId, args.reportedAttempt, args.reportedAttempt == null ? (args.streamAttemptId ?? undefined) : undefined @@ -18705,9 +18442,9 @@ export class TaskService implements AgentTaskIntegration { // stream's pending decision before the completion prompt, as handleStreamEnd does for any // other non-report end: the prompt's admission reads a still-pending decision for its // attempt as stale and is refused, leaving the workflow step awaiting a report forever. - this.resolveStreamEndDecision( + this.streamEndLedger.resolveStreamEndDecision( childWorkspaceId, - this.findPendingStreamEndDecision( + this.streamEndLedger.findPendingStreamEndDecision( childWorkspaceId, reportedAttempt, reportedAttempt == null ? (reportedAttemptId ?? undefined) : undefined @@ -18974,9 +18711,9 @@ export class TaskService implements AgentTaskIntegration { ); // Never a continuation of the reporting attempt; its held entries are refused (they also // read stale at their own gates, since the row names another attempt). - this.resolveStreamEndDecision( + this.streamEndLedger.resolveStreamEndDecision( childWorkspaceId, - this.findPendingStreamEndDecision( + this.streamEndLedger.findPendingStreamEndDecision( childWorkspaceId, reportedAttempt, unownedReportedAttemptId @@ -19032,9 +18769,9 @@ export class TaskService implements AgentTaskIntegration { this.resolveWaiters(childWorkspaceId, reportArgs); this.scheduleMaybeStartQueuedTasks(); // Reported row, no release, nobody to report to: not a continuation of the attempt. - this.resolveStreamEndDecision( + this.streamEndLedger.resolveStreamEndDecision( childWorkspaceId, - this.findPendingStreamEndDecision( + this.streamEndLedger.findPendingStreamEndDecision( childWorkspaceId, reportedAttempt, unownedReportedAttemptId @@ -19113,9 +18850,9 @@ export class TaskService implements AgentTaskIntegration { // so held follow-ups are refused rather than run under an attempt of unknown standing. if (persistedInEveryAncestor) { this.attemptLedger.releaseReportedTaskAttempt(childWorkspaceId, reportedAttempt); - this.resolveStreamEndDecision( + this.streamEndLedger.resolveStreamEndDecision( childWorkspaceId, - this.findPendingStreamEndDecision( + this.streamEndLedger.findPendingStreamEndDecision( childWorkspaceId, reportedAttempt, unownedReportedAttemptId @@ -19123,9 +18860,9 @@ export class TaskService implements AgentTaskIntegration { "published" ); } else { - this.resolveStreamEndDecision( + this.streamEndLedger.resolveStreamEndDecision( childWorkspaceId, - this.findPendingStreamEndDecision( + this.streamEndLedger.findPendingStreamEndDecision( childWorkspaceId, reportedAttempt, unownedReportedAttemptId @@ -19185,7 +18922,7 @@ export class TaskService implements AgentTaskIntegration { // After the committed `reported` (ours above, or the other backend's): a report bound to the // row's attempt settles that attempt's stream-end decision (#7003); the notification then // re-runs its deferral (the row no longer reads running). - if (this.resolveDecisionOfReportedAttempt(taskId, artifact.taskAttemptId)) { + if (this.streamEndLedger.resolveDecisionOfReportedAttempt(taskId, artifact.taskAttemptId)) { this.notifyAttemptSettlementListeners(taskId); } if (adopt) { @@ -19211,27 +18948,6 @@ export class TaskService implements AgentTaskIntegration { thinkingLevel: artifact.thinkingLevel, }; } - /** - * The row reads a completed report for `attemptId` (#7003): that attempt reported, possibly on - * another backend. Its pending stream-end decision (`decision`, else the attempt's) resolves - * `indeterminate`, as publication's not-durable branch does: held follow-ups are refused and - * handed back, never run as a new turn of a reported attempt. Returns whether the row reads so. - */ - private resolveDecisionOfReportedAttempt( - taskId: string, - attemptId: string | undefined, - decision?: StreamEndDecision - ): boolean { - if (attemptId == null) return false; - const workspace = findWorkspaceEntry(this.config.loadConfigOrDefault(), taskId)?.workspace; - if (workspace?.taskAttemptId !== attemptId || !hasCompletedAgentReport(workspace)) return false; - this.resolveStreamEndDecision( - taskId, - decision ?? this.findStreamEndDecision(taskId, attemptId), - "indeterminate" - ); - return true; - } /** * waitForAgentReport's adoption of a persisted report for a row that has not reported: refused diff --git a/src/node/services/taskStopLatch.ts b/src/node/services/taskStopLatch.ts index 62968cf1596..99caaab62fe 100644 --- a/src/node/services/taskStopLatch.ts +++ b/src/node/services/taskStopLatch.ts @@ -23,8 +23,8 @@ 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. + * its methods after construction. Attempt, send and stream-end state stays with its owners + * (AttemptLedger, StreamEndLedger) and is reached only through these closures. */ export interface TaskStopLatchDeps { config: () => Pick;