fix(fusion): handle a stopped or swapped worker like any other cut

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();

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");
+  });
 });

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?.({

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