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..d10721953 100644 --- a/src/tools/fusion/worker-runner.ts +++ b/src/tools/fusion/worker-runner.ts @@ -663,7 +663,13 @@ 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. 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]!) @@ -1469,6 +1475,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 +1556,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 +1649,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