From c25518f12afeaa253b9126450c22f06fd562a6dd Mon Sep 17 00:00:00 2001 From: Nadia Rodionova <273119990+sosidudku1@users.noreply.github.com> Date: Wed, 11 Nov 2026 02:15:06 +0300 Subject: [PATCH 1/2] fix(fusion): handle a stopped or swapped worker like any other cut - A local worker the person stopped, or one swapped for the escalation model, now marks the local leg as possibly wedged, so the next local worker asks the server first instead of waiting out a queue budget. The cut is read before the re-run resets its flags. (ATO-1317) - The re-run on the escalation model reports its tool calls as they start, so a crash mid-re-run leaves its side effects in the checkpoint and the task is not offered as safe to repeat. (ATO-1316) - The fan-out pool starts nothing more once the turn is stopped, also when no budget cap is set. (ATO-1334) --- src/tools/fusion/escalation.test.ts | 31 +++++++++++++ src/tools/fusion/local-leg-health.test.ts | 17 +++++++ src/tools/fusion/worker-runner.test.ts | 18 ++++++++ src/tools/fusion/worker-runner.ts | 56 ++++++++++++++++++----- 4 files changed, 111 insertions(+), 11 deletions(-) diff --git a/src/tools/fusion/escalation.test.ts b/src/tools/fusion/escalation.test.ts index ff3572ce8..469730dbb 100644 --- a/src/tools/fusion/escalation.test.ts +++ b/src/tools/fusion/escalation.test.ts @@ -213,6 +213,37 @@ describe("advisor escalation of a stuck worker (ATO-942)", () => { }); }); +describe("the re-run's side effects reach the checkpoint at once (ATO-1316)", () => { + it("reports the re-run's tool calls as they start", async () => { + const h = harness(async (_s, options) => { + if (options.providerId === "cloud") { + options.eventHook?.({ + type: "llm_event", + event: { type: "tool_call_parsed", call: { tool: "os.shell.run", args: {} } }, + } as never); + return done(); + } + for (let i = 0; i < 3; i += 1) sameStep(options.eventHook, i); + await flush(); + for (let i = 3; i < 6; i += 1) sameStep(options.eventHook, i); + return { + session: createEmptySessionState({ id: "s-x", workingDir: "/repo" }), + reason: options.signal?.aborted ? "cancelled" : "reply", + stepCount: 6, + } as RunTurnResult; + }); + const tools: Array<[string, string]> = []; + await runWorkerTasks(h.deps, { + ...BASE, + tasks: TASK, + signal: new AbortController().signal, + escalation: { target: { ...TARGET, replace: true }, pricing: PRICING }, + onWorkerTool: (taskId, tool) => tools.push([taskId, tool]), + }); + expect(tools).toContainEqual(["t0", "os.shell.run"]); + }); +}); + describe("the person's stop and escalate on one worker (ATO-66)", () => { it("escalates a running worker on request and stops it alone", async () => { const live = new FusionLiveWorkers(); diff --git a/src/tools/fusion/local-leg-health.test.ts b/src/tools/fusion/local-leg-health.test.ts index 95afd881a..a9a1a79d3 100644 --- a/src/tools/fusion/local-leg-health.test.ts +++ b/src/tools/fusion/local-leg-health.test.ts @@ -5,6 +5,7 @@ import type { FusionWorkerMeta } from "../../session/fusion-worker-session.js"; import { createEmptySessionState } from "../../session/session-state.js"; import { FUSION_WORKER_ID_PREFIX } from "../../session/fusion-worker-session.js"; import { runWorkerTasks, type WorkerRunnerDeps } from "./worker-runner.js"; +import { FusionLiveWorkers } from "./escalation.js"; import { LocalLegHealth, LOCAL_LEG_UNRESPONSIVE_ERROR, @@ -167,4 +168,20 @@ describe("dispatch after a cut-off local worker (ATO-917)", () => { expect(probes).toBe(0); expect(deps.turns).toBe(1); }); + + it("marks the leg when the person stops a local worker (ATO-1317)", async () => { + const health = new LocalLegHealth(); + const live = new FusionLiveWorkers(); + const deps = harness(servedThenHung, { localLegHealth: health, liveWorkers: live }); + const run = runWorkerTasks(deps, { + ...BASE, + tasks: TASK, + signal: new AbortController().signal, + }); + await new Promise((r) => setTimeout(r, 0)); + expect(live.stop("s-parent", "t0")).toBe(true); + await run; + expect(health.isSuspect()).toBe(true); + expect(health.cause()).toBe("task t0 was stopped by the person"); + }); }); diff --git a/src/tools/fusion/worker-runner.test.ts b/src/tools/fusion/worker-runner.test.ts index eb89113e6..f29fe4c85 100644 --- a/src/tools/fusion/worker-runner.test.ts +++ b/src/tools/fusion/worker-runner.test.ts @@ -93,6 +93,24 @@ const BASE = { }; describe("runWorkerTasks", () => { + it("starts no further task after the turn is stopped, with no budget cap set (ATO-1334)", async () => { + const stop = new AbortController(); + const { deps, calls } = harness(async () => { + stop.abort(new Error("stopped")); + return turnResult({ reason: "cancelled" }); + }); + const results = await runWorkerTasks(deps, { + ...BASE, + tasks: tasks(3), + maxWorkers: 1, + signal: stop.signal, + }); + expect(calls).toHaveLength(1); + expect(results.map((r) => r.status)).toEqual(["cancelled", "cancelled", "cancelled"]); + expect(results[1]!.error).toBe("not started: the turn was stopped"); + expect(results[2]!.error).toBe("not started: the turn was stopped"); + }); + it("pins each worker turn to the local leg with the fusion origin and the worker filter", async () => { const { deps, calls } = harness(async ({ options }) => { options.eventHook?.({ diff --git a/src/tools/fusion/worker-runner.ts b/src/tools/fusion/worker-runner.ts index c0d72619f..31d77365f 100644 --- a/src/tools/fusion/worker-runner.ts +++ b/src/tools/fusion/worker-runner.ts @@ -663,7 +663,11 @@ export async function runWorkerTasks( const width = Math.max(1, Math.min(options.maxWorkers, options.tasks.length)); const runners = Array.from({ length: width }, async () => { for (let i = claim(); i < options.tasks.length; i = claim()) { - const stop = options.shouldStart?.(); + // A stopped turn starts nothing more, capped or not (ATO-1334): + // `shouldStart` is only wired when a budget cap is set. + const stop = options.signal.aborted + ? "the turn was stopped" + : options.shouldStart?.(); results[i] = stop === undefined ? await runOneTask(deps, options, options.tasks[i]!) @@ -1469,6 +1473,30 @@ async function runOneTask( disarmWorkerSession(session.id); } + // How the LOCAL turn ended, read before the re-run below resets these + // flags (ATO-1317). A turn cut by the person's stop or by the swap to + // the escalation model was cut as surely as one cut by its clock, and + // may have left the server just as wedged; without these two the next + // local worker waited out a whole queue budget before anyone asked. + const localCut = + replaceRequested || + stopByPerson.signal.aborted || + timedOutOutcome || + queuedOutcome || + stoppedByHandBack || + options.signal.aborted; + const localCutCause = timedOutOutcome + ? "timed out" + : queuedOutcome + ? "never served" + : stoppedByHandBack + ? "handed back" + : replaceRequested + ? "replaced by the escalation model" + : stopByPerson.signal.aborted + ? "stopped by the person" + : "cancelled"; + // The advisor's hint did not help and the operator allowed the // replacement path (`escalationReplace`): the task runs again on the // escalation model, with what the local worker found, under the @@ -1526,6 +1554,20 @@ async function runOneTask( eventHook: (event) => { markNoteRead(replaceSession.id, event); replaceCollector.observe(event); + // The re-run's side effects reach the checkpoint as they + // start, like the local worker's do (ATO-1316): a crash in + // the middle of the re-run must not offer the task as safe + // to repeat. + if ( + event.type === "llm_event" && + event.event.type === "tool_call_parsed" + ) { + try { + options.onWorkerTool?.(task.id, event.event.call.tool); + } catch { + // An observer never takes the worker down. + } + } if ( event.type === "llm_event" && event.event.type === "llm_completed" && @@ -1605,16 +1647,8 @@ async function runOneTask( // A local worker that ended by an abort — its clock, its queue wait, // its hand-back or the operator — may have left the server wedged // (ATO-917); the next local dispatch asks the server first. - if ( - options.localLeg === true && - (timedOutOutcome || - queuedOutcome || - stoppedByHandBack || - options.signal.aborted) - ) { - deps.localLegHealth?.noteCut( - `task ${task.id} was ${timedOutOutcome ? "timed out" : queuedOutcome ? "never served" : stoppedByHandBack ? "handed back" : "cancelled"}`, - ); + if (options.localLeg === true && localCut) { + deps.localLegHealth?.noteCut(`task ${task.id} was ${localCutCause}`); } // How long the server took to say anything. The one number that tells From 87c5f01d060a31b4d83a76f3834268a60c60e1c5 Mon Sep 17 00:00:00 2001 From: Nadia Rodionova <273119990+sosidudku1@users.noreply.github.com> Date: Wed, 11 Nov 2026 02:48:23 +0300 Subject: [PATCH 2/2] fix(fusion): name the spend cap, not a stop, when the cap ends the pool The cap aborts the same signal the pool now checks, so the next task read "not started: the turn was stopped". Ask the cap first. --- src/tools/fusion/worker-runner.ts | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/tools/fusion/worker-runner.ts b/src/tools/fusion/worker-runner.ts index 31d77365f..d10721953 100644 --- a/src/tools/fusion/worker-runner.ts +++ b/src/tools/fusion/worker-runner.ts @@ -664,10 +664,12 @@ export async function runWorkerTasks( const runners = Array.from({ length: width }, async () => { for (let i = claim(); i < options.tasks.length; i = claim()) { // A stopped turn starts nothing more, capped or not (ATO-1334): - // `shouldStart` is only wired when a budget cap is set. - const stop = options.signal.aborted - ? "the turn was stopped" - : options.shouldStart?.(); + // `shouldStart` is only wired when a budget cap is set. The cap is + // asked first: a cap reached aborts the same signal, and its row + // must say the cap, not the person, stopped it. + const stop = + options.shouldStart?.() ?? + (options.signal.aborted ? "the turn was stopped" : undefined); results[i] = stop === undefined ? await runOneTask(deps, options, options.tasks[i]!)