Merge pull request #606 from fabro-sh/fix/cancellation-interrupt-lifecycle

Harden cancellation and interrupt lifecycles

apps/fabro-web/app/components/runs-list/row-actions-menu.tsx

@@ -12,9 +12,13 @@ import {
   canCancel,
   canDelete,
   canUnarchive,
+  cancellationActionLabel,
+  cancellationSuccessMessage,
   cancelRun,
+  deleteErrorMessage,
   deleteRun,
   denyRun,
+  isCancellationPendingState,
   mapError,
   retryRun,
   unarchiveRun,
@@ -32,7 +36,8 @@ const MENU_ITEM_DANGER_CLASS =
 export function RowActionsMenu({ run }: { run: RunWithStatus }) {
   const { mutate } = useSWRConfig();
   const { push } = useToast();
-  const [pending, setPending] = useState(false);
+  const [pendingAction, setPendingAction] = useState<LifecycleAction | "delete" | null>(null);
+  const [optimisticallyCancelled, setOptimisticallyCancelled] = useState(false);
   const [deleteDialogOpen, setDeleteDialogOpen] = useState(false);
   const [idCopied, setIdCopied] = useState(false);
 
@@ -44,6 +49,12 @@ export function RowActionsMenu({ run }: { run: RunWithStatus }) {
   const showUnarchive = canUnarchive(status);
   const showCancel = canCancel(status);
   const showDelete = canDelete(status);
+  const cancellationPending = isCancellationPendingState(
+    status,
+    run.pendingControl,
+    pendingAction === "cancel" || optimisticallyCancelled,
+  );
+  const pending = pendingAction !== null || cancellationPending;
 
   const hasLifecycle = showRetry || showArchive || showUnarchive;
   const hasDestructive = showDeny || showCancel || showDelete;
@@ -51,17 +62,25 @@ export function RowActionsMenu({ run }: { run: RunWithStatus }) {
   async function runAction<T>(
     label: LifecycleAction,
     action: () => Promise<T>,
-    successMessage: string,
+    successMessage: string | ((result: T) => string),
   ) {
     if (pending) return;
-    setPending(true);
+    setPendingAction(label);
     try {
-      await action();
-      push({ message: successMessage });
+      const result = await action();
+      if (label === "cancel") {
+        setOptimisticallyCancelled(true);
+      }
+      push({
+        message:
+          typeof successMessage === "function"
+            ? successMessage(result)
+            : successMessage,
+      });
     } catch (error) {
       push({ message: mapError(error, label), tone: "error" });
     } finally {
-      setPending(false);
+      setPendingAction(null);
       mutateRunListCaches(mutate);
     }
   }
@@ -81,15 +100,14 @@ export function RowActionsMenu({ run }: { run: RunWithStatus }) {
 
   async function handleDeleteConfirm() {
     if (pending) return;
-    setPending(true);
+    setPendingAction("delete");
     try {
       await deleteRun(run.id);
       push({ message: "Deleted run." });
     } catch (error) {
-      // deleteRun throws LifecycleActionError shapes via lifecycleActionErrorFromError
-      push({ message: mapError(error, "archive"), tone: "error" });
+      push({ message: deleteErrorMessage(error), tone: "error" });
     } finally {
-      setPending(false);
+      setPendingAction(null);
       setDeleteDialogOpen(false);
       mutateRunListCaches(mutate);
     }
@@ -208,12 +226,12 @@ export function RowActionsMenu({ run }: { run: RunWithStatus }) {
               <button
                 type="button"
                 onClick={() =>
-                  void runAction("cancel", () => cancelRun(run.id), "Cancelled run.")
+                  void runAction("cancel", () => cancelRun(run.id), cancellationSuccessMessage)
                 }
                 disabled={pending}
                 className={MENU_ITEM_DANGER_CLASS}
               >
-                Cancel
+                {cancellationActionLabel(cancellationPending)}
               </button>
             </MenuItem>
           )}

apps/fabro-web/app/components/steer-bar.test.tsx

@@ -0,0 +1,22 @@
+import { describe, expect, test } from "bun:test";
+import { createElement } from "react";
+import { renderToStaticMarkup } from "react-dom/server";
+
+import {
+  isInterruptDisabled,
+  SteerWaitingStatus,
+} from "./steer-bar";
+
+describe("SteerBar", () => {
+  test("shows durable waiting state and prevents a second interrupt", () => {
+    expect(isInterruptDisabled(true, false)).toBe(true);
+    expect(isInterruptDisabled(false, true)).toBe(true);
+    expect(isInterruptDisabled(false, false)).toBe(false);
+
+    const html = renderToStaticMarkup(
+      createElement(SteerWaitingStatus, { waitingForSteer: true }),
+    );
+    expect(html).toContain('role="status"');
+    expect(html).toContain("Interrupted — waiting for steering");
+  });
+});

apps/fabro-web/app/components/steer-bar.tsx

@@ -13,6 +13,7 @@ import { ErrorMessage } from "./ui";
 
 export interface SteerBarProps {
   runId: string;
+  waitingForSteer?: boolean;
   ref?: Ref<SteerBarHandle>;
 }
 
@@ -20,13 +21,38 @@ export interface SteerBarHandle {
   focus(): void;
 }
 
-export function SteerBar({ runId, ref }: SteerBarProps) {
+export function isInterruptDisabled(
+  waitingForSteer: boolean,
+  mutationPending: boolean,
+): boolean {
+  return waitingForSteer || mutationPending;
+}
+
+export function SteerWaitingStatus({
+  waitingForSteer,
+}: {
+  waitingForSteer: boolean;
+}) {
+  if (!waitingForSteer) return null;
+  return (
+    <p role="status" className="mt-2 text-xs text-amber">
+      Interrupted — waiting for steering
+    </p>
+  );
+}
+
+export function SteerBar({
+  runId,
+  waitingForSteer = false,
+  ref,
+}: SteerBarProps) {
   const [text, setText] = useState("");
   const [errorMessage, setErrorMessage] = useState<string | null>(null);
   const textareaRef = useRef<HTMLTextAreaElement | null>(null);
   const steer = useSteerRun(runId);
   const interrupt = useInterruptRun(runId);
   const pending = steer.isMutating || interrupt.isMutating;
+  const interruptDisabled = isInterruptDisabled(waitingForSteer, pending);
 
   useImperativeHandle(ref, () => ({
     focus() {
@@ -49,7 +75,7 @@ export function SteerBar({ runId, ref }: SteerBarProps) {
   }
 
   async function fireInterrupt() {
-    if (pending) return;
+    if (interruptDisabled) return;
     setErrorMessage(null);
     try {
       await interrupt.trigger();
@@ -91,7 +117,7 @@ export function SteerBar({ runId, ref }: SteerBarProps) {
         <button
           type="button"
           onClick={() => void fireInterrupt()}
-          disabled={pending}
+          disabled={interruptDisabled}
           className="inline-flex shrink-0 items-center gap-2 rounded-md bg-overlay px-3 py-2 text-sm font-medium text-amber outline-1 -outline-offset-1 outline-amber/40 transition-colors hover:bg-amber/15 focus-visible:outline-2 focus-visible:outline-offset-2 focus-visible:outline-amber disabled:cursor-not-allowed disabled:opacity-60"
         >
           {interrupt.isMutating ? "Interrupting…" : "Interrupt"}
@@ -109,6 +135,7 @@ export function SteerBar({ runId, ref }: SteerBarProps) {
           <ErrorMessage message={errorMessage} />
         </div>
       )}
+      <SteerWaitingStatus waitingForSteer={waitingForSteer} />
     </form>
   );
 }

apps/fabro-web/app/data/runs.test.ts

@@ -83,6 +83,7 @@ describe("mapRunListItem", () => {
         html_url: "https://github.com/fabro-sh/fabro/pull/123",
       },
     });
+    summary.lifecycle.pending_control = "cancel";
     const item = mapRunListItem(summary);
     expect(item.id).toBe("01ABC");
     expect(item.title).toBe("Server supplied title");
@@ -94,6 +95,7 @@ describe("mapRunListItem", () => {
     expect(item.lifecycleStatus).toBe("paused");
     expect(item.number).toBe(123);
     expect(item.pullRequestUrl).toBe("https://github.com/fabro-sh/fabro/pull/123");
+    expect(item.pendingControl).toBe("cancel");
   });
 
   test("uses a fallback title when the server title is blank", () => {

apps/fabro-web/app/data/runs.ts

@@ -3,6 +3,7 @@ import {
   BoardColumn,
   type Principal,
   type Run,
+  type RunControlAction,
   type RunSize,
   type RunStatus as ApiRunStatus,
 } from "@qltysh/fabro-api-client";
@@ -26,6 +27,7 @@ export interface RunItem {
   column?: BoardColumn;
   lifecycleStatus?: RunStatus | null;
   lifecycleStatusLabel?: string;
+  pendingControl?: RunControlAction | null;
   number?: number;
   pullRequestUrl?: string;
   additions?: number;
@@ -99,6 +101,7 @@ export function mapRunListItem(item: Run): RunItem {
     column: columnForRun(item) ?? undefined,
     lifecycleStatus,
     lifecycleStatusLabel: lifecycleStatusLabel(item.lifecycle.status, item.lifecycle.archived),
+    pendingControl: item.lifecycle.pending_control,
     number: item.pull_request?.number,
     pullRequestUrl: item.pull_request?.html_url,
     elapsed: item.timing != null ? formatDurationMs(item.timing.wall_time_ms) : undefined,

apps/fabro-web/app/lib/board-events.test.tsx

@@ -38,6 +38,7 @@ describe("shouldRefreshBoardForEvent", () => {
   test("refreshes board for run and interview status changes only", () => {
     expect(shouldRefreshBoardForEvent("run.running")).toBe(true);
     expect(shouldRefreshBoardForEvent("run.blocked")).toBe(true);
+    expect(shouldRefreshBoardForEvent("run.cancel.requested")).toBe(true);
     expect(shouldRefreshBoardForEvent("interview.completed")).toBe(true);
     expect(shouldRefreshBoardForEvent("checkpoint.completed")).toBe(false);
   });

apps/fabro-web/app/lib/board-events.ts

@@ -35,6 +35,9 @@ const BOARD_STATUS_EVENTS = new Set([
   "run.unpaused",
   "run.blocked",
   "run.unblocked",
+  "run.cancel.requested",
+  "run.pause.requested",
+  "run.unpause.requested",
   "run.completed",
   "run.failed",
   "run.archived",

apps/fabro-web/app/lib/mutations.ts

@@ -119,8 +119,9 @@ function useLifecycleMutation(
       onSuccess: (result) => {
         if (!id || !result.ok) return;
         if (intent !== "retry") {
-          // Retry doesn't mutate the source run, so skip invalidating its detail/billing keys.
-          void mutate(queryKeys.runs.detail(id));
+          // Keep the returned lifecycle state visible while revalidation
+          // observes the durable follow-up event (notably a 202 cancel).
+          void mutate(queryKeys.runs.detail(id), result.run, { revalidate: true });
           void mutate(queryKeys.runs.billing(id));
         }
         mutateRunListCaches(mutate);

apps/fabro-web/app/lib/query-keys.test.ts

@@ -75,6 +75,7 @@ describe("queryKeys", () => {
       queryKeys.runs.graph("run-1", "LR"),
       queryKeys.runs.graph("run-1", "TB"),
       queryKeys.runs.detail("run-1"),
+      queryKeys.runs.state("run-1"),
       queryKeys.runs.stageEvents("run-1", "stage-1"),
       queryKeys.runs.stageContextWindow("run-1", "stage-1"),
     ]);

apps/fabro-web/app/lib/run-actions.test.ts

@@ -15,9 +15,13 @@ import {
   canCancel,
   canRetry,
   canUnarchive,
+  cancellationActionLabel,
+  cancellationSuccessMessage,
   cancelRun,
   deleteRuns,
   isTerminalCancelledRun,
+  isCancellationPending,
+  isCancellationPendingState,
   mapError,
   retryRun,
   unarchiveRun,
@@ -173,6 +177,20 @@ describe("run lifecycle actions", () => {
     }
   });
 
+  test("cancelRun parses a 202 response as a pending cancellation", async () => {
+    const run = makeRun({ kind: "running" });
+    run.lifecycle.pending_control = "cancel";
+    stubGeneratedAxiosOnce({
+      status: 202,
+      body: run,
+    });
+
+    const result = await cancelRun("run-1");
+    expect(result.lifecycle.status.kind).toBe("running");
+    expect(result.lifecycle.pending_control).toBe("cancel");
+    expect(cancellationSuccessMessage(result)).toBe("Cancellation requested.");
+  });
+
   test("archiveRun parses a 200 response", async () => {
     stubGeneratedAxiosOnce({
       status: 200,
@@ -448,4 +466,22 @@ describe("run lifecycle actions", () => {
       ),
     ).toBe(false);
   });
+
+  test("cancellation pending combines durable state with the active mutation", () => {
+    const running = makeRun({ kind: "running" });
+    expect(isCancellationPending(running)).toBe(false);
+    expect(isCancellationPending(running, true)).toBe(true);
+
+    running.lifecycle.pending_control = "cancel";
+    expect(isCancellationPending(running)).toBe(true);
+
+    const cancelled = makeRun({ kind: "failed", reason: "cancelled" });
+    cancelled.lifecycle.pending_control = "cancel";
+    expect(isCancellationPending(cancelled, true)).toBe(false);
+    expect(cancellationSuccessMessage(cancelled)).toBe("Run cancelled.");
+    expect(isCancellationPendingState("running", "cancel")).toBe(true);
+    expect(isCancellationPendingState("failed", "cancel", true)).toBe(false);
+    expect(cancellationActionLabel(true)).toBe("Cancelling…");
+    expect(cancellationActionLabel(false)).toBe("Cancel");
+  });
 });

apps/fabro-web/app/lib/run-actions.ts

@@ -6,6 +6,7 @@ import type {
   BatchRunLifecycleSummary,
   ErrorResponseEntry,
   Run,
+  RunControlAction,
 } from "@qltysh/fabro-api-client";
 
 import {
@@ -159,11 +160,40 @@ export function canDelete(status: string | null | undefined): boolean {
   return status === "archived";
 }
 
-export function isTerminalCancelledRun(run: Run): boolean {
+export function isTerminalCancelledRun(run: Pick<Run, "lifecycle">): boolean {
   const status = run.lifecycle.status;
   return status.kind === "failed" && status.reason === "cancelled";
 }
 
+export function isCancellationPending(
+  run: Pick<Run, "lifecycle"> | null | undefined,
+  mutationPending = false,
+): boolean {
+  return isCancellationPendingState(
+    run?.lifecycle.status.kind,
+    run?.lifecycle.pending_control,
+    mutationPending,
+  );
+}
+
+export function isCancellationPendingState(
+  status: string | null | undefined,
+  pendingControl: RunControlAction | null | undefined,
+  mutationPending = false,
+): boolean {
+  return canCancel(status) && (mutationPending || pendingControl === "cancel");
+}
+
+export function cancellationSuccessMessage(run: Pick<Run, "lifecycle">): string {
+  return isTerminalCancelledRun(run)
+    ? "Run cancelled."
+    : "Cancellation requested.";
+}
+
+export function cancellationActionLabel(pending: boolean): string {
+  return pending ? "Cancelling…" : "Cancel";
+}
+
 export function deleteErrorMessage(error: unknown): string {
   if (isLifecycleActionError(error)) {
     if (error.status === 409) {

apps/fabro-web/app/lib/run-events.test.tsx

@@ -43,6 +43,7 @@ describe("queryKeysForRunEvent", () => {
   test("terminal events invalidate run-scoped resources", () => {
     expect(queryKeysForRunEvent("run-1", "run.completed")).toEqual([
       queryKeys.runs.detail("run-1"),
+      queryKeys.runs.state("run-1"),
       ...queryKeys.runs.filesAllScopes("run-1"),
       queryKeys.runs.commits("run-1"),
       queryKeys.runs.billing("run-1"),
@@ -60,6 +61,7 @@ describe("queryKeysForRunEvent", () => {
       queryKeys.runs.graph("run-1", "LR"),
       queryKeys.runs.graph("run-1", "TB"),
       queryKeys.runs.detail("run-1"),
+      queryKeys.runs.state("run-1"),
       queryKeys.runs.stageEvents("run-1", "verify@2"),
       queryKeys.runs.stageContextWindow("run-1", "verify@2"),
     ]);
@@ -81,6 +83,21 @@ describe("queryKeysForRunEvent", () => {
     ]);
   });
 
+  test("interrupt settlement invalidates projected control state and stage activity", () => {
+    expect(queryKeysForRunEvent("run-1", "agent.round.interrupted", "nap@1")).toEqual([
+      queryKeys.runs.state("run-1"),
+      queryKeys.runs.events("run-1", 1000),
+      queryKeys.runs.stageEvents("run-1", "nap@1"),
+      queryKeys.runs.stageContextWindow("run-1", "nap@1"),
+    ]);
+  });
+
+  test("cancel requests invalidate the durable run summary", () => {
+    expect(queryKeysForRunEvent("run-1", "run.cancel.requested")).toEqual([
+      queryKeys.runs.detail("run-1"),
+    ]);
+  });
+
   test("pair messages invalidate stage-scoped resources", () => {
     expect(queryKeysForRunEvent("run-1", "agent.pair.user_message", "nap@1")).toEqual([
       queryKeys.runs.stageEvents("run-1", "nap@1"),

apps/fabro-web/app/lib/run-events.ts

@@ -47,6 +47,9 @@ const RUN_SUMMARY_EVENTS = new Set([
   "run.unpaused",
   "run.blocked",
   "run.unblocked",
+  "run.cancel.requested",
+  "run.pause.requested",
+  "run.unpause.requested",
   "run.archived",
   "run.unarchived",
   "run.title.updated",
@@ -77,6 +80,7 @@ export const STAGE_ACTIVITY_EVENT_TYPES = [
   "agent.tool.completed",
   "agent.steering.injected",
   "agent.interrupt.injected",
+  "agent.round.interrupted",
   "agent.pair.user_message",
   "agent.pair.system_message",
   "command.started",
@@ -95,11 +99,17 @@ const STEERING_EVENTS = new Set([
   "run.steer",
   "agent.steering.injected",
   "agent.interrupt.injected",
+  "agent.round.interrupted",
   "agent.session.activated",
   "agent.session.deactivated",
   "agent.steer.buffered",
   "agent.steer.dropped",
 ]);
+const AGENT_CONTROL_STATE_EVENTS = new Set([
+  "agent.round.interrupted",
+  "agent.steering.injected",
+  "agent.session.deactivated",
+]);
 // Todo / task mutation events refresh `getRunState` consumers (so per-stage
 // todo projections update live) and the run events list.
 const TODO_EVENTS = new Set([
@@ -123,6 +133,7 @@ export function queryKeysForRunEvent(
   if (TERMINAL_EVENTS.has(event)) {
     return [
       queryKeys.runs.detail(runId),
+      queryKeys.runs.state(runId),
       ...queryKeys.runs.filesAllScopes(runId),
       queryKeys.runs.commits(runId),
       queryKeys.runs.billing(runId),
@@ -151,6 +162,7 @@ export function queryKeysForRunEvent(
       queryKeys.runs.graph(runId, "LR"),
       queryKeys.runs.graph(runId, "TB"),
       queryKeys.runs.detail(runId),
+      queryKeys.runs.state(runId),
     ];
     if (stageId) {
       keys.push(queryKeys.runs.stageEvents(runId, stageId));
@@ -161,6 +173,9 @@ export function queryKeysForRunEvent(
 
   if (STEERING_EVENTS.has(event)) {
     const keys: Key[] = [queryKeys.runs.events(runId, 1000)];
+    if (AGENT_CONTROL_STATE_EVENTS.has(event)) {
+      keys.unshift(queryKeys.runs.state(runId));
+    }
     if (stageId) {
       keys.push(queryKeys.runs.stageEvents(runId, stageId));
       keys.push(queryKeys.runs.stageContextWindow(runId, stageId));
@@ -238,6 +253,7 @@ function runInvalidation(runId: string, payload: RunEventPayload) {
 function resyncKeysForRun(runId: string) {
   return [
     queryKeys.runs.detail(runId),
+    queryKeys.runs.state(runId),
     ...queryKeys.runs.filesAllScopes(runId),
     queryKeys.runs.commits(runId),
     queryKeys.runs.billing(runId),

apps/fabro-web/app/routes/run-detail.test.ts

@@ -197,6 +197,7 @@ function makeRunSummary({
   title = "Run 1",
   askFabro = null as any,
   automation = null as any,
+  pendingControl = null as any,
 }: {
   status?: string;
   diffSummary?: any;
@@ -204,6 +205,7 @@ function makeRunSummary({
   title?: string;
   askFabro?: any;
   automation?: any;
+  pendingControl?: any;
 } = {}) {
   const apiStatus =
     status === "succeeded"
@@ -229,7 +231,7 @@ function makeRunSummary({
     lifecycle:        {
       status:          archived ? { kind: "succeeded", reason: "completed" } : apiStatus,
       approval:        null,
-      pending_control: null,
+      pending_control: pendingControl,
       queue_position:  null,
       error:           null,
       archived,
@@ -284,6 +286,7 @@ async function renderRunDetailHarness({
   title,
   askFabro = null,
   automation = null,
+  pendingControl = null,
 }: {
   initialEntry: string;
   status?: string;
@@ -293,8 +296,17 @@ async function renderRunDetailHarness({
   title?: string;
   askFabro?: any;
   automation?: any;
+  pendingControl?: any;
 }) {
-  currentRunSummary = makeRunSummary({ status, diffSummary, pullRequest, title, askFabro, automation });
+  currentRunSummary = makeRunSummary({
+    status,
+    diffSummary,
+    pullRequest,
+    title,
+    askFabro,
+    automation,
+    pendingControl,
+  });
   currentQuestions = questions;
   (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true;
 
@@ -803,6 +815,39 @@ describe("RunDetail full-height child routes", () => {
     expect(sandboxLinks).toHaveLength(0);
   });
 
+  test("keeps cancellation visibly pending from durable server state", async () => {
+    const renderer = await renderRunDetail({
+      initialEntry:   "/runs/run_1",
+      status:         "running",
+      pendingControl: "cancel",
+    });
+
+    const cancelButton = findButtonByText(renderer, "Cancelling…");
+    expect(cancelButton).toBeDefined();
+    expect(cancelButton!.props.disabled).toBe(true);
+  });
+
+  test("shows projected interrupt settlement until steering resumes", async () => {
+    currentRunState = {
+      stages: {
+        "code@1": {
+          agent_control: "waiting_for_steer",
+        },
+      },
+    };
+    const renderer = await renderRunDetail({
+      initialEntry: "/runs/run_1",
+      status:       "running",
+    });
+
+    const statuses = renderer.root.findAll(
+      (node) => node.type === "p" && node.props.role === "status",
+    );
+    expect(statuses.map(textFromTestNode)).toContain(
+      "Interrupted — waiting for steering",
+    );
+  });
+
   test("defers steer bar focus until after the Actions menu item click settles", async () => {
     const focusCalls: string[] = [];
 

apps/fabro-web/app/routes/run-detail.tsx

@@ -9,6 +9,7 @@ import {
   useMatches,
   useNavigate,
 } from "react-router";
+import { AgentControlState } from "@qltysh/fabro-api-client";
 
 import { type SteerBarHandle } from "../components/steer-bar";
 import { ErrorState } from "../components/state";
@@ -38,6 +39,7 @@ import {
   canRetry,
   deleteErrorMessage,
   deleteRun,
+  isCancellationPending,
   type LifecycleAction,
 } from "../lib/run-actions";
 import {
@@ -105,6 +107,9 @@ export default function RunDetail({ params }: { params: { id: string } }) {
   const filesCount = runQuery.data?.diff?.files_changed ?? null;
   const childrenCount = runQuery.data?.children_count ?? null;
   const hasSandbox = runHasSandbox(runStateQuery.data);
+  const waitingForSteer = Object.values(runStateQuery.data?.stages ?? {}).some(
+    (stage) => stage.agent_control === AgentControlState.WAITING_FOR_STEER,
+  );
   const tabs = buildRunDetailTabs({
     hasSandbox,
     filesCount,
@@ -157,7 +162,7 @@ export default function RunDetail({ params }: { params: { id: string } }) {
 
   const visibility = lifecycleActionVisibility(run.lifecycleStatus);
   const previewPending = previewMutation.isMutating;
-  const cancelPending = cancelMutation.isMutating;
+  const cancelPending = isCancellationPending(summary, cancelMutation.isMutating);
   const approvalActionVisible = canApprove(summary);
   const approvePending = approveMutation.isMutating;
   const denyPending = denyMutation.isMutating;
@@ -228,9 +233,9 @@ export default function RunDetail({ params }: { params: { id: string } }) {
         key:          "interrupt",
         label:        "Send interrupt",
         pendingLabel: "Interrupting…",
-          pending:      interruptMutation.isMutating,
-          disabled:     statusKind !== "running",
-          onSelect:     () => void interruptMutation.trigger(),
+        pending:      interruptMutation.isMutating,
+        disabled:     statusKind !== "running" || waitingForSteer,
+        onSelect:     () => void interruptMutation.trigger(),
       },
       {
         key:      "steer",
@@ -366,6 +371,7 @@ export default function RunDetail({ params }: { params: { id: string } }) {
             sidebarWidth={sidebarWidth}
             isResizing={isResizing}
             steerBarRef={steerBarRef}
+            waitingForSteer={waitingForSteer}
           />
         </div>
       )}

apps/fabro-web/app/routes/run-detail/docked-controls.tsx

@@ -133,6 +133,7 @@ export function RunDetailDockedControls({
   sidebarWidth,
   isResizing,
   steerBarRef,
+  waitingForSteer,
 }: {
   runId: string;
   hideSteerBar: boolean;
@@ -141,6 +142,7 @@ export function RunDetailDockedControls({
   sidebarWidth: number;
   isResizing: boolean;
   steerBarRef: RefObject<SteerBarHandle | null>;
+  waitingForSteer: boolean;
 }) {
   if (hideSteerBar && !hasPendingQuestions) return null;
 
@@ -156,7 +158,11 @@ export function RunDetailDockedControls({
       {hasPendingQuestions ? (
         <InterviewDock runId={runId} questions={pendingQuestions} />
       ) : (
-        <SteerBar ref={steerBarRef} runId={runId} />
+        <SteerBar
+          ref={steerBarRef}
+          runId={runId}
+          waitingForSteer={waitingForSteer}
+        />
       )}
     </div>
   );

apps/fabro-web/app/routes/run-detail/lifecycle-toasts.ts

@@ -8,7 +8,7 @@ import {
   canCancel,
   canDelete,
   canUnarchive,
-  isTerminalCancelledRun,
+  cancellationSuccessMessage,
   mapError,
   type LifecycleAction,
 } from "../../lib/run-actions";
@@ -92,7 +92,7 @@ export function handleLifecycleToastResult(
 
   if (intent === "cancel") {
     toastApi.push({
-      message: isTerminalCancelledRun(result.run) ? "Run cancelled." : "Cancellation requested.",
+      message: cancellationSuccessMessage(result.run),
     });
     return nextState;
   }

apps/fabro-web/app/routes/run-stages.test.ts

@@ -228,6 +228,31 @@ describe("eventsToActivity", () => {
     ]);
   });
 
+  test("renders settled interrupt as waiting for steering", () => {
+    const events: EventEnvelope[] = [
+      envelope(1, {
+        event: "agent.round.interrupted",
+        stage_id: "nap@1",
+        node_id: "nap",
+        properties: { generation: 1, visit: 1 },
+      }),
+      envelope(2, {
+        event: "agent.round.interrupted",
+        stage_id: "other@1",
+        node_id: "other",
+        properties: { generation: 1, visit: 1 },
+      }),
+    ];
+
+    expect(eventsToActivity(events, "nap@1")).toEqual([
+      {
+        kind: "interrupt",
+        ts: "2026-04-09T12:00:00Z",
+        content: "Interrupted — waiting for steering",
+      },
+    ]);
+  });
+
   test("renders pair messages as transcript turns for the matching stage", () => {
     const events: EventEnvelope[] = [
       envelope(1, {

apps/fabro-web/app/routes/run-stages.tsx

@@ -293,6 +293,13 @@ export function eventsToActivity(events: EventEnvelope[], stageId: string): Turn
       case "agent.interrupt.injected":
         turns.push({ kind: "interrupt", ts: e.ts, content: "Agent interrupted" });
         break;
+      case "agent.round.interrupted":
+        turns.push({
+          kind: "interrupt",
+          ts: e.ts,
+          content: "Interrupted — waiting for steering",
+        });
+        break;
       case "agent.pair.user_message": {
         const text = getString(props, "text") ?? e.text ?? "";
         if (text) {

apps/fabro-web/app/routes/runs.test.tsx

@@ -153,6 +153,7 @@ describe("runs route board mapping", () => {
     expect(shouldRefreshBoardForEvent("run.denied")).toBe(true);
     expect(shouldRefreshBoardForEvent("run.blocked")).toBe(true);
     expect(shouldRefreshBoardForEvent("run.unblocked")).toBe(true);
+    expect(shouldRefreshBoardForEvent("run.cancel.requested")).toBe(true);
     expect(shouldRefreshBoardForEvent("run.archived")).toBe(true);
     expect(shouldRefreshBoardForEvent("run.unarchived")).toBe(true);
     expect(shouldRefreshBoardForEvent("run.title.updated")).toBe(true);

docs/public/api-reference/fabro-api.yaml

@@ -1567,12 +1567,23 @@ paths:
       operationId: cancelRun
       tags: [Runs]
       summary: Cancel Run
-      description: Cancels a pending, runnable, or running run. Returns 409 if the run has already completed or been cancelled.
+      description: |
+        Cancels a pending, runnable, or running run. Pre-execution runs are
+        cancelled synchronously. Live runs return after the cancellation
+        request is durably recorded and continue converging to a terminal
+        cancelled state. Returns 409 if the run has already completed or been
+        cancelled.
       parameters:
         - $ref: "#/components/parameters/RunId"
       responses:
         "200":
-          description: Run cancelled
+          description: Run was cancelled synchronously before execution
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/Run"
+        "202":
+          description: Cancellation was durably requested for a live run
           content:
             application/json:
               schema:
@@ -10472,6 +10483,7 @@ components:
       required:
         - first_event_seq
         - usage
+        - agent_control
         - state
       properties:
         first_event_seq:
@@ -10571,6 +10583,9 @@ components:
             - $ref: "#/components/schemas/StageContextWindowProjection"
             - type: "null"
           description: Latest content-free context-window snapshot for this agent stage.
+        agent_control:
+          $ref: "#/components/schemas/AgentControlState"
+          description: Whether the agent is executing normally or waiting for steering after an interrupt.
         state:
           $ref: "#/components/schemas/StageState"
           description: Lifecycle state of the stage projection.
@@ -12347,6 +12362,13 @@ components:
 
     # ── Stage / Turn Schemas ─────────────────────────────────────────────
 
+    AgentControlState:
+      description: Control state of a live agent stage.
+      type: string
+      enum:
+        - running
+        - waiting_for_steer
+
     StageState:
       description: Lifecycle projection state of a workflow stage.
       type: string

lib/apps/fabro-cli/tests/it/cmd/runner.rs

@@ -840,7 +840,7 @@ fn detached_run_cancel_reaches_worker_over_control_websocket() {
                 .expect("cancel request should succeed");
             assert_reqwest_status(
                 response,
-                fabro_http::StatusCode::OK,
+                fabro_http::StatusCode::ACCEPTED,
                 format!("POST /api/v1/runs/{run_id}/cancel"),
             )
             .await;

lib/apps/fabro-server/src/server.rs

@@ -281,10 +281,35 @@ struct ManagedRun {
     cancel_tx: Option<oneshot::Sender<()>>,
     cancel_token: Option<CancellationToken>,
     worker_ref: Option<WorkerRef>,
+    /// Exact worker currently covered by a cancellation escalation task.
+    /// Prevents repeated cancel requests from arming duplicate watchdogs.
+    cancel_escalation_worker: Option<WorkerRef>,
     run_dir: Option<std::path::PathBuf>,
     execution_mode: RunExecutionMode,
 }
 
+impl ManagedRun {
+    /// True if cancellation should still escalate to `worker_ref`; clears a
+    /// stale escalation marker as a side effect.
+    fn escalation_still_current(&mut self, worker_ref: &WorkerRef) -> bool {
+        let matches_watchdog = self.cancel_escalation_worker.as_ref() == Some(worker_ref);
+        let still_current = matches_watchdog
+            && !self.status.is_terminal()
+            && self.worker_ref.as_ref() == Some(worker_ref);
+        if matches_watchdog && !still_current {
+            self.cancel_escalation_worker = None;
+        }
+        still_current
+    }
+
+    /// Clears the escalation marker if it is still owned by `worker_ref`.
+    fn clear_escalation_for(&mut self, worker_ref: &WorkerRef) {
+        if self.cancel_escalation_worker.as_ref() == Some(worker_ref) {
+            self.cancel_escalation_worker = None;
+        }
+    }
+}
+
 #[derive(Clone, Copy)]
 enum RunExecutionMode {
     Start,
@@ -2961,6 +2986,7 @@ fn clear_live_run_state(run: &mut ManagedRun) {
     run.cancel_tx = None;
     run.cancel_token = None;
     run.worker_ref = None;
+    run.cancel_escalation_worker = None;
 }
 
 fn cleanup_worker_control_bus_for_run(state: &AppState, run_id: RunId) {
@@ -3328,6 +3354,7 @@ fn managed_run(
         cancel_tx: None,
         cancel_token: None,
         worker_ref: None,
+        cancel_escalation_worker: None,
         run_dir: Some(run_dir),
         execution_mode,
     }
@@ -4479,7 +4506,15 @@ async fn append_control_request(
         RunControlAction::Pause => workflow_event::Event::RunPauseRequested { actor },
         RunControlAction::Unpause => workflow_event::Event::RunUnpauseRequested { actor },
     };
-    workflow_event::append_event(&run_store, &run_id, &event).await
+    if action == RunControlAction::Cancel {
+        workflow_event::append_event_if(&run_store, &run_id, &event, |projection| {
+            projection.pending_control != Some(RunControlAction::Cancel)
+        })
+        .await
+        .map(|_| ())
+    } else {
+        workflow_event::append_event(&run_store, &run_id, &event).await
+    }
 }
 
 /// Returns a 409 response with an actionable "unarchive first" message if the

lib/apps/fabro-server/src/server/handler/lifecycle.rs

@@ -2,6 +2,7 @@ use std::collections::HashSet;
 use std::sync::Arc;
 
 use chrono::Utc;
+use tokio::time::{Instant, sleep_until};
 
 use super::super::{
     ApiError, AppState, AskFabroReadiness, BatchDeleteRunsRequest, BatchDeleteRunsResponse,
@@ -15,7 +16,7 @@ use super::super::{
     TimelineEntryResponse, WORKER_CANCEL_GRACE, WorkflowError, append_control_request,
     clear_live_run_state, delete_run_internal, durable_run_status, get, load_pending_control,
     managed_run, operations, parse_run_id_path, persist_cancelled_run_status, post,
-    reject_if_archived, sleep, update_live_run_from_event, workflow_event,
+    reject_if_archived, update_live_run_from_event, workflow_event,
 };
 use super::runs::run_provenance;
 use crate::worker_runtime::WorkerRef;
@@ -349,17 +350,90 @@ async fn deny_run(
     run_response(state.as_ref(), id, StatusCode::OK).await
 }
 
-fn schedule_worker_force_stop(state: Arc<AppState>, run_id: RunId, worker_ref: WorkerRef) {
-    tokio::spawn(async move {
-        sleep(WORKER_CANCEL_GRACE).await;
-        let current_ref = {
-            let runs = state.runs.lock().expect("runs lock poisoned");
-            runs.get(&run_id).and_then(|run| run.worker_ref.clone())
+fn schedule_worker_cancel_escalation(state: Arc<AppState>, run_id: RunId, worker_ref: WorkerRef) {
+    let requested_at = Instant::now();
+    let armed = {
+        let mut runs = state.runs.lock().expect("runs lock poisoned");
+        let Some(run) = runs.get_mut(&run_id) else {
+            return;
         };
-        if current_ref.as_ref() == Some(&worker_ref)
-            && state.worker_runtime.is_alive(&worker_ref).await
-        {
-            state.worker_runtime.force_stop(&worker_ref).await;
+        if run.cancel_escalation_worker.as_ref() == Some(&worker_ref) {
+            false
+        } else {
+            run.cancel_escalation_worker = Some(worker_ref.clone());
+            true
+        }
+    };
+    if !armed {
+        tracing::debug!(
+            run_id = %run_id,
+            worker_kind = worker_ref.kind(),
+            worker_ref = ?worker_ref,
+            "Worker cancellation escalation is already armed"
+        );
+        return;
+    }
+
+    tokio::spawn(async move {
+        sleep_until(requested_at + WORKER_CANCEL_GRACE).await;
+        let should_escalate = {
+            let mut runs = state.runs.lock().expect("runs lock poisoned");
+            let Some(run) = runs.get_mut(&run_id) else {
+                return;
+            };
+            run.escalation_still_current(&worker_ref)
+        };
+        if !should_escalate {
+            tracing::debug!(
+                run_id = %run_id,
+                worker_kind = worker_ref.kind(),
+                worker_ref = ?worker_ref,
+                "Skipping stale worker cancellation escalation"
+            );
+            return;
+        }
+        if !state.worker_runtime.is_alive(&worker_ref).await {
+            let mut runs = state.runs.lock().expect("runs lock poisoned");
+            if let Some(run) = runs.get_mut(&run_id) {
+                run.clear_escalation_for(&worker_ref);
+            }
+            tracing::debug!(
+                run_id = %run_id,
+                worker_kind = worker_ref.kind(),
+                worker_ref = ?worker_ref,
+                "Skipping worker cancellation escalation because worker exited"
+            );
+            return;
+        }
+        let still_current = {
+            let mut runs = state.runs.lock().expect("runs lock poisoned");
+            let Some(run) = runs.get_mut(&run_id) else {
+                return;
+            };
+            run.escalation_still_current(&worker_ref)
+        };
+        if !still_current {
+            tracing::debug!(
+                run_id = %run_id,
+                worker_kind = worker_ref.kind(),
+                worker_ref = ?worker_ref,
+                "Skipping worker cancellation escalation after liveness check"
+            );
+            return;
+        }
+
+        let elapsed_ms = u64::try_from(requested_at.elapsed().as_millis()).unwrap_or(u64::MAX);
+        tracing::warn!(
+            run_id = %run_id,
+            worker_kind = worker_ref.kind(),
+            worker_ref = ?worker_ref,
+            elapsed_ms,
+            "Force-stopping worker after cancellation grace period"
+        );
+        state.worker_runtime.force_stop(&worker_ref).await;
+        let mut runs = state.runs.lock().expect("runs lock poisoned");
+        if let Some(run) = runs.get_mut(&run_id) {
+            run.clear_escalation_for(&worker_ref);
         }
     });
 }
@@ -470,21 +544,29 @@ async fn cancel_run(
     } else {
         false
     };
-    if !delivered_control {
-        if let Some(worker_ref) = worker_ref {
+    tracing::debug!(
+        run_id = %id,
+        delivered_control,
+        "Processed cooperative run cancellation signal"
+    );
+    if let Some(worker_ref) = worker_ref {
+        if !delivered_control {
             state.worker_runtime.request_stop(&worker_ref).await;
-            schedule_worker_force_stop(Arc::clone(&state), id, worker_ref);
         }
+        schedule_worker_cancel_escalation(Arc::clone(&state), id, worker_ref);
     }
 
-    if persist_cancelled_status {
+    let response_status = if persist_cancelled_status {
         if let Err(err) = persist_cancelled_run_status(state.as_ref(), id).await {
             return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
                 .into_response();
         }
-    }
+        StatusCode::OK
+    } else {
+        StatusCode::ACCEPTED
+    };
 
-    run_response(state.as_ref(), id, StatusCode::OK).await
+    run_response(state.as_ref(), id, response_status).await
 }
 
 async fn unmanaged_cancel_response(

lib/apps/fabro-server/src/server/tests.rs

@@ -37,6 +37,7 @@ use fabro_workflow::records::CheckpointExt;
 use httpmock::Method::{GET, POST};
 use httpmock::MockServer;
 use serde_json::json;
+use tokio::sync::Notify;
 use tokio_stream::StreamExt as _;
 use tokio_tungstenite::tungstenite::client::IntoClientRequest;
 use tokio_tungstenite::tungstenite::protocol::Message as WebSocketMessage;
@@ -2943,9 +2944,10 @@ fn worker_token_claims(cmd: &Command, state: &AppState) -> crate::worker_token::
 
 #[derive(Default)]
 struct RecordingWorkerRuntime {
-    requested: StdMutex<Vec<WorkerRef>>,
-    forced:    StdMutex<Vec<WorkerRef>>,
-    alive:     AtomicBool,
+    requested:     StdMutex<Vec<WorkerRef>>,
+    forced:        StdMutex<Vec<WorkerRef>>,
+    alive:         AtomicBool,
+    forced_notify: Notify,
 }
 
 impl RecordingWorkerRuntime {
@@ -2963,6 +2965,20 @@ impl RecordingWorkerRuntime {
     fn set_alive(&self, alive: bool) {
         self.alive.store(alive, Ordering::Relaxed);
     }
+
+    async fn wait_for_forced_ref(&self, worker_ref: &WorkerRef) {
+        tokio::time::timeout(std::time::Duration::from_secs(1), async {
+            loop {
+                let notified = self.forced_notify.notified();
+                if self.forced_refs().contains(worker_ref) {
+                    return;
+                }
+                notified.await;
+            }
+        })
+        .await
+        .expect("worker should be force-stopped after the cancellation grace period");
+    }
 }
 
 #[async_trait::async_trait]
@@ -2984,6 +3000,7 @@ impl WorkerRuntime for RecordingWorkerRuntime {
             .expect("forced lock poisoned")
             .push(worker_ref.clone());
         self.alive.store(false, Ordering::Relaxed);
+        self.forced_notify.notify_one();
     }
 
     async fn is_alive(&self, _worker_ref: &WorkerRef) -> bool {
@@ -14148,7 +14165,7 @@ async fn cancel_run_overwrites_pending_pause_request() {
         .body(Body::empty())
         .unwrap();
     let response = app.clone().oneshot(req).await.unwrap();
-    let body = response_json!(response, StatusCode::OK).await;
+    let body = response_json!(response, StatusCode::ACCEPTED).await;
     assert_eq!(run_json_pending_control(&body).as_str(), Some("cancel"));
 
     let summary = state
@@ -14165,19 +14182,29 @@ async fn cancel_run_overwrites_pending_pause_request() {
     );
 }
 
+async fn advance_past_worker_cancel_grace() {
+    tokio::task::yield_now().await;
+    tokio::time::advance(WORKER_CANCEL_GRACE).await;
+    for _ in 0..10 {
+        tokio::task::yield_now().await;
+    }
+}
+
 #[tokio::test]
 async fn cancel_run_requests_worker_runtime_stop_when_control_unavailable() {
     let runtime = StdArc::new(RecordingWorkerRuntime::default());
+    runtime.set_alive(true);
     let state = TestAppStateBuilder::new()
         .vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")])
         .worker_runtime(runtime.clone())
         .build();
     let app = crate::test_support::build_test_router(Arc::clone(&state));
-    let run_id = create_and_start_run(&app, MINIMAL_DOT)
+    let run_id = create_run(&app, MINIMAL_DOT)
         .await
         .parse::<RunId>()
         .unwrap();
     let worker_ref = test_worker_ref(u32::MAX);
+    tokio::time::pause();
 
     {
         let mut runs = state.runs.lock().expect("runs lock poisoned");
@@ -14193,9 +14220,203 @@ async fn cancel_run_requests_worker_runtime_stop_when_control_unavailable() {
         .body(Body::empty())
         .unwrap();
     let response = app.oneshot(req).await.unwrap();
-    assert_status!(response, StatusCode::OK).await;
+    assert_status!(response, StatusCode::ACCEPTED).await;
 
-    assert_eq!(runtime.requested_refs(), vec![worker_ref]);
+    assert_eq!(runtime.requested_refs(), vec![worker_ref.clone()]);
+
+    advance_past_worker_cancel_grace().await;
+    runtime.wait_for_forced_ref(&worker_ref).await;
+
+    assert_eq!(runtime.forced_refs(), vec![worker_ref]);
+}
+
+#[tokio::test]
+async fn cancel_run_force_stops_worker_when_delivered_control_does_not_converge() {
+    let runtime = StdArc::new(RecordingWorkerRuntime::default());
+    runtime.set_alive(true);
+    let state = TestAppStateBuilder::new()
+        .vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")])
+        .worker_runtime(runtime.clone())
+        .build();
+    let app = crate::test_support::build_test_router(Arc::clone(&state));
+    let run_id = create_run(&app, MINIMAL_DOT)
+        .await
+        .parse::<RunId>()
+        .unwrap();
+    let worker_ref = test_worker_ref(u32::MAX);
+    let (answer_transport, _receiver) = worker_transport_with_receiver(run_id).await;
+    tokio::time::pause();
+
+    {
+        let mut runs = state.runs.lock().expect("runs lock poisoned");
+        let managed_run = runs.get_mut(&run_id).expect("run should exist");
+        managed_run.status = RunStatus::Running;
+        managed_run.answer_transport = Some(answer_transport);
+        managed_run.worker_ref = Some(worker_ref.clone());
+    }
+
+    let req = Request::builder()
+        .method("POST")
+        .uri(api(&format!("/runs/{run_id}/cancel")))
+        .body(Body::empty())
+        .unwrap();
+    let response = app.oneshot(req).await.unwrap();
+    assert_status!(response, StatusCode::ACCEPTED).await;
+
+    assert!(runtime.requested_refs().is_empty());
+    assert!(runtime.forced_refs().is_empty());
+
+    advance_past_worker_cancel_grace().await;
+    runtime.wait_for_forced_ref(&worker_ref).await;
+
+    assert_eq!(runtime.forced_refs(), vec![worker_ref]);
+}
+
+#[tokio::test]
+async fn cancel_run_watchdog_does_not_stop_replacement_worker() {
+    let runtime = StdArc::new(RecordingWorkerRuntime::default());
+    runtime.set_alive(true);
+    let state = TestAppStateBuilder::new()
+        .vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")])
+        .worker_runtime(runtime.clone())
+        .build();
+    let app = crate::test_support::build_test_router(Arc::clone(&state));
+    let run_id = create_run(&app, MINIMAL_DOT)
+        .await
+        .parse::<RunId>()
+        .unwrap();
+    let cancelled_worker_ref = test_worker_ref(u32::MAX - 1);
+    let replacement_worker_ref = test_worker_ref(u32::MAX);
+    let (answer_transport, _receiver) = worker_transport_with_receiver(run_id).await;
+    tokio::time::pause();
+
+    {
+        let mut runs = state.runs.lock().expect("runs lock poisoned");
+        let managed_run = runs.get_mut(&run_id).expect("run should exist");
+        managed_run.status = RunStatus::Running;
+        managed_run.answer_transport = Some(answer_transport);
+        managed_run.worker_ref = Some(cancelled_worker_ref);
+    }
+
+    let req = Request::builder()
+        .method("POST")
+        .uri(api(&format!("/runs/{run_id}/cancel")))
+        .body(Body::empty())
+        .unwrap();
+    let response = app.oneshot(req).await.unwrap();
+    assert_status!(response, StatusCode::ACCEPTED).await;
+
+    tokio::task::yield_now().await;
+    {
+        let mut runs = state.runs.lock().expect("runs lock poisoned");
+        let managed_run = runs.get_mut(&run_id).expect("run should exist");
+        managed_run.worker_ref = Some(replacement_worker_ref);
+    }
+    advance_past_worker_cancel_grace().await;
+
+    assert!(runtime.forced_refs().is_empty());
+}
+
+#[tokio::test]
+async fn cancel_run_watchdog_does_not_stop_worker_after_live_ref_clears() {
+    let runtime = StdArc::new(RecordingWorkerRuntime::default());
+    runtime.set_alive(true);
+    let state = TestAppStateBuilder::new()
+        .vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")])
+        .worker_runtime(runtime.clone())
+        .build();
+    let app = crate::test_support::build_test_router(Arc::clone(&state));
+    let run_id = create_run(&app, MINIMAL_DOT)
+        .await
+        .parse::<RunId>()
+        .unwrap();
+    let worker_ref = test_worker_ref(u32::MAX);
+    let (answer_transport, _receiver) = worker_transport_with_receiver(run_id).await;
+    tokio::time::pause();
+
+    {
+        let mut runs = state.runs.lock().expect("runs lock poisoned");
+        let managed_run = runs.get_mut(&run_id).expect("run should exist");
+        managed_run.status = RunStatus::Running;
+        managed_run.answer_transport = Some(answer_transport);
+        managed_run.worker_ref = Some(worker_ref);
+    }
+
+    let req = Request::builder()
+        .method("POST")
+        .uri(api(&format!("/runs/{run_id}/cancel")))
+        .body(Body::empty())
+        .unwrap();
+    let response = app.oneshot(req).await.unwrap();
+    assert_status!(response, StatusCode::ACCEPTED).await;
+
+    tokio::task::yield_now().await;
+    {
+        let mut runs = state.runs.lock().expect("runs lock poisoned");
+        let managed_run = runs.get_mut(&run_id).expect("run should exist");
+        managed_run.worker_ref = None;
+    }
+    advance_past_worker_cancel_grace().await;
+
+    assert!(runtime.forced_refs().is_empty());
+}
+
+#[tokio::test]
+async fn repeated_cancel_request_arms_one_watchdog_and_persists_one_intent() {
+    let runtime = StdArc::new(RecordingWorkerRuntime::default());
+    runtime.set_alive(true);
+    let state = TestAppStateBuilder::new()
+        .vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")])
+        .worker_runtime(runtime.clone())
+        .build();
+    let app = crate::test_support::build_test_router(Arc::clone(&state));
+    let run_id = create_run(&app, MINIMAL_DOT)
+        .await
+        .parse::<RunId>()
+        .unwrap();
+    let worker_ref = test_worker_ref(u32::MAX);
+    let (answer_transport, _receiver) = worker_transport_with_receiver(run_id).await;
+    tokio::time::pause();
+
+    {
+        let mut runs = state.runs.lock().expect("runs lock poisoned");
+        let managed_run = runs.get_mut(&run_id).expect("run should exist");
+        managed_run.status = RunStatus::Running;
+        managed_run.answer_transport = Some(answer_transport);
+        managed_run.worker_ref = Some(worker_ref.clone());
+    }
+
+    let first_request = Request::builder()
+        .method("POST")
+        .uri(api(&format!("/runs/{run_id}/cancel")))
+        .body(Body::empty())
+        .unwrap();
+    let second_request = Request::builder()
+        .method("POST")
+        .uri(api(&format!("/runs/{run_id}/cancel")))
+        .body(Body::empty())
+        .unwrap();
+    let (first_response, second_response) = tokio::join!(
+        app.clone().oneshot(first_request),
+        app.clone().oneshot(second_request)
+    );
+    assert_status!(first_response.unwrap(), StatusCode::ACCEPTED).await;
+    assert_status!(second_response.unwrap(), StatusCode::ACCEPTED).await;
+
+    let run_store = state.stores.runs.open_run_reader(&run_id).await.unwrap();
+    let request_count = run_store
+        .list_events()
+        .await
+        .unwrap()
+        .iter()
+        .filter(|event| event.event.event_name() == "run.cancel.requested")
+        .count();
+    assert_eq!(request_count, 1);
+
+    advance_past_worker_cancel_grace().await;
+    runtime.wait_for_forced_ref(&worker_ref).await;
+
+    assert_eq!(runtime.forced_refs(), vec![worker_ref]);
 }
 
 #[tokio::test]
@@ -14248,7 +14469,7 @@ async fn cancel_durably_blocked_in_process_run_cancels_pending_interview_without
         .body(Body::empty())
         .unwrap();
     let response = app.oneshot(req).await.unwrap();
-    assert_status!(response, StatusCode::OK).await;
+    assert_status!(response, StatusCode::ACCEPTED).await;
 
     let submission = tokio::time::timeout(std::time::Duration::from_millis(100), ask)
         .await
@@ -14790,8 +15011,7 @@ id = "local"
         if matches!(
             live_status_before_cancel,
             Some(
-                RunStatus::Runnable
-                    | RunStatus::Starting
+                RunStatus::Starting
                     | RunStatus::Running
                     | RunStatus::Blocked { .. }
                     | RunStatus::Paused { .. }
@@ -14805,8 +15025,7 @@ id = "local"
         matches!(
             live_status_before_cancel,
             Some(
-                RunStatus::Runnable
-                    | RunStatus::Starting
+                RunStatus::Starting
                     | RunStatus::Running
                     | RunStatus::Blocked { .. }
                     | RunStatus::Paused { .. }
@@ -14825,7 +15044,7 @@ id = "local"
     let response_body = body_json(response.into_body()).await;
     assert_eq!(
         response_status,
-        StatusCode::OK,
+        StatusCode::ACCEPTED,
         "unexpected cancel response body: {response_body}; live status before cancel: {live_status_before_cancel:?}"
     );
 
@@ -14887,7 +15106,7 @@ async fn cancel_before_run_transitions_to_running_returns_empty_attach_stream()
         .body(Body::empty())
         .unwrap();
     let response = app.clone().oneshot(req).await.unwrap();
-    assert_status!(response, StatusCode::OK).await;
+    assert_status!(response, StatusCode::ACCEPTED).await;
 
     runner.await.unwrap();
 

lib/apps/fabro-server/src/worker_runtime.rs

@@ -21,7 +21,8 @@ pub(crate) trait WorkerRuntime: Send + Sync {
     async fn is_alive(&self, worker_ref: &WorkerRef) -> bool;
 }
 
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq, strum::IntoStaticStr)]
+#[strum(serialize_all = "snake_case")]
 pub(crate) enum WorkerRef {
     /// A worker running as a local subprocess. `pre_exec_setpgid` ensures the
     /// child is the leader of its own process group with `pgid == pid`, so a
@@ -29,6 +30,12 @@ pub(crate) enum WorkerRef {
     Local { pid: u32 },
 }
 
+impl WorkerRef {
+    pub(crate) fn kind(&self) -> &'static str {
+        self.into()
+    }
+}
+
 pub(crate) struct WorkerLaunchSpec {
     pub(crate) executable:             PathBuf,
     pub(crate) server_target:          String,

lib/apps/fabro-server/tests/it/scenario/lifecycle.rs

@@ -256,7 +256,7 @@ async fn full_http_lifecycle_cancel() {
     let response = app.clone().oneshot(req).await.unwrap();
     let body = response_json(
         response,
-        StatusCode::OK,
+        StatusCode::ACCEPTED,
         format!("POST /api/v1/runs/{run_id}/cancel"),
     )
     .await;
@@ -323,7 +323,7 @@ async fn cancel_at_human_gate_persists_cancelled_terminal_event() {
     let response = app.clone().oneshot(req).await.unwrap();
     response_status(
         response,
-        StatusCode::OK,
+        StatusCode::ACCEPTED,
         format!("POST /api/v1/runs/{run_id}/cancel"),
     )
     .await;

lib/components/fabro-agent/src/agent_profile.rs

@@ -1,14 +1,11 @@
-use std::sync::Arc;
-
 use fabro_llm::types::ToolDefinition;
 use fabro_model::{AgentProfileKind, Catalog, Model, ProviderId};
-use tokio::sync::Mutex;
 
 use crate::profiles::EnvContext;
 use crate::sandbox::Sandbox;
 use crate::skills::Skill;
 use crate::subagent::{
-    SessionFactory, SubAgentManager, make_close_agent_tool, make_send_input_tool,
+    SessionFactory, SubAgentSupervisor, make_close_agent_tool, make_send_input_tool,
     make_spawn_agent_tool, make_wait_tool,
 };
 use crate::tool_registry::ToolRegistry;
@@ -57,21 +54,21 @@ pub trait AgentProfile: Send + Sync {
 
     fn register_subagent_tools(
         &mut self,
-        manager: Arc<Mutex<SubAgentManager>>,
+        supervisor: SubAgentSupervisor,
         session_factory: SessionFactory,
         current_depth: usize,
     ) {
         self.tool_registry_mut().register(make_spawn_agent_tool(
-            manager.clone(),
+            supervisor.clone(),
             session_factory,
             current_depth,
         ));
         self.tool_registry_mut()
-            .register(make_send_input_tool(manager.clone()));
+            .register(make_send_input_tool(supervisor.clone()));
         self.tool_registry_mut()
-            .register(make_wait_tool(manager.clone()));
+            .register(make_wait_tool(supervisor.clone()));
         self.tool_registry_mut()
-            .register(make_close_agent_tool(manager));
+            .register(make_close_agent_tool(supervisor));
     }
 }
 

lib/components/fabro-agent/src/cli.rs

@@ -26,16 +26,15 @@ use fabro_util::terminal::Styles;
 use fabro_vault::SecretStore;
 use tokio::io::{AsyncWriteExt, stdout};
 use tokio::signal;
-use tokio::sync::Mutex as AsyncMutex;
 
 use crate::config::{ToolApprovalAdapter, ToolApprovalFn, ToolHookCallback, ToolSecrets};
 use crate::error::InterruptReason;
-use crate::subagent::{SessionFactory, SubAgentManager};
+use crate::subagent::{SessionFactory, SubAgentSupervisor};
 use crate::tool_permissions::{is_auto_approved, tool_category};
 use crate::tools::WebFetchSummarizer;
 use crate::{
     AgentEvent, AgentProfile, AnthropicProfile, GeminiProfile, LocalSandbox, Message,
-    OpenAiProfile, Sandbox, Session, SessionOptions,
+    OpenAiProfile, Sandbox, Session, SessionOptions, SessionShutdownReason,
 };
 
 #[expect(
@@ -600,10 +599,8 @@ pub async fn run_with_args_and_client_and_catalog(
     };
 
     // Register subagent tools
-    let manager = Arc::new(AsyncMutex::new(SubAgentManager::new(
-        config.max_subagent_depth,
-    )));
-    let manager_for_callback = manager.clone();
+    let supervisor = SubAgentSupervisor::new(config.max_subagent_depth);
+    let supervisor_for_session = supervisor.clone();
     let factory_client = client.clone();
     let factory_model = model.clone();
     let factory_catalog = Arc::clone(&catalog);
@@ -640,22 +637,13 @@ pub async fn run_with_args_and_client_and_catalog(
             None,
         )
     });
-    profile.register_subagent_tools(manager, factory, 0);
+    profile.register_subagent_tools(supervisor.clone(), factory, 0);
     let profile: Arc<dyn AgentProfile> = Arc::from(profile);
 
-    let mut session = Session::new(
-        client,
-        profile,
-        env,
-        config,
-        Some(manager_for_callback.clone()),
-    );
+    let mut session = Session::new(client, profile, env, config, Some(supervisor_for_session));
 
     // Wire subagent event callback to parent session's emitter
-    manager_for_callback
-        .lock()
-        .await
-        .set_event_callback(session.sub_agent_event_callback());
+    supervisor.set_event_callback(session.sub_agent_event_callback());
 
     // SIGINT handler
     let cancel_token = session.cancel_token();
@@ -797,8 +785,18 @@ pub async fn run_with_args_and_client_and_catalog(
     });
 
     // Initialize and run
-    session.initialize().await?;
-    let result = session.process_input(&args.prompt).await;
+    let result = match session.initialize().await {
+        Ok(()) => session.process_input(&args.prompt).await,
+        Err(error) => Err(error),
+    };
+    let shutdown_reason = if result.is_ok() {
+        SessionShutdownReason::Completed
+    } else if session.cancel_token().is_cancelled() {
+        SessionShutdownReason::Cancelled
+    } else {
+        SessionShutdownReason::Error
+    };
+    session.shutdown(shutdown_reason).await;
 
     if matches!(output_format, OutputFormat::Text) {
         // Print assistant text to stdout
@@ -1235,11 +1233,11 @@ mod tests {
             None,
             test_catalog(),
         );
-        let manager = Arc::new(AsyncMutex::new(SubAgentManager::new(1)));
+        let supervisor = SubAgentSupervisor::new(1);
         let factory: SessionFactory = Arc::new(|| {
             panic!("factory should not be called in this test");
         });
-        profile.register_subagent_tools(manager, factory, 0);
+        profile.register_subagent_tools(supervisor, factory, 0);
 
         let names = profile.tool_registry().names();
         assert!(names.contains(&"spawn_agent".to_string()));

lib/components/fabro-agent/src/lib.rs

@@ -61,13 +61,11 @@ pub use sandbox::{
     shell_quote,
 };
 pub use session::{
-    CompletionCoordinator, Session, SessionControlHandle, SessionInputTiming, StaticEnvProvider,
-    SteeringItem, ToolEnvProvider,
+    CompletionCoordinator, Session, SessionControlHandle, SessionInputTiming,
+    SessionShutdownReason, StaticEnvProvider, SteeringItem, ToolEnvProvider,
 };
 pub use skills::Skill;
-pub use subagent::{
-    SubAgent, SubAgentEventCallback, SubAgentManager, SubAgentResult, SubAgentStatus,
-};
+pub use subagent::{SubAgentEventCallback, SubAgentResult, SubAgentStatus, SubAgentSupervisor};
 pub use todo_runtime::TodoRuntime;
 pub use todo_tools::{
     make_task_create_tool, make_task_get_tool, make_task_list_tool, make_task_update_tool,

lib/components/fabro-agent/src/profiles/anthropic.rs

@@ -272,10 +272,8 @@ impl AgentProfile for AnthropicProfile {
 mod tests {
     use std::sync::Arc;
 
-    use tokio::sync::Mutex as AsyncMutex;
-
     use super::*;
-    use crate::subagent::{SessionFactory, SubAgentManager};
+    use crate::subagent::{SessionFactory, SubAgentSupervisor};
     use crate::test_support::MockSandbox;
 
     fn test_catalog() -> Arc<Catalog> {
@@ -386,11 +384,11 @@ mod tests {
         assert!(!prompt.contains("Subagents are valuable for independent work"));
 
         let mut profile = AnthropicProfile::new("claude-sonnet-4-20250514");
-        let manager = Arc::new(AsyncMutex::new(SubAgentManager::new(3)));
+        let supervisor = SubAgentSupervisor::new(3);
         let factory: SessionFactory = Arc::new(|| {
             panic!("should not be called in test");
         });
-        profile.register_subagent_tools(manager, factory, 0);
+        profile.register_subagent_tools(supervisor, factory, 0);
         let prompt = profile.build_system_prompt(&env, &EnvContext::default(), &[], None, &[]);
 
         assert!(prompt.contains("Subagents are valuable for independent work"));
@@ -471,12 +469,12 @@ mod tests {
         let mut profile = AnthropicProfile::new("claude-sonnet-4-20250514");
         assert_eq!(profile.tool_registry().names().len(), 12);
 
-        let manager = Arc::new(AsyncMutex::new(SubAgentManager::new(3)));
+        let supervisor = SubAgentSupervisor::new(3);
         let factory: SessionFactory = Arc::new(|| {
             panic!("should not be called in test");
         });
 
-        profile.register_subagent_tools(manager, factory, 0);
+        profile.register_subagent_tools(supervisor, factory, 0);
 
         let names = profile.tool_registry().names();
         assert_eq!(names.len(), 16, "should have 12 base + 4 subagent tools");

lib/components/fabro-agent/src/profiles/gemini.rs

@@ -227,10 +227,8 @@ in the project.";
 mod tests {
     use std::sync::Arc;
 
-    use tokio::sync::Mutex as AsyncMutex;
-
     use super::*;
-    use crate::subagent::{SessionFactory, SubAgentManager};
+    use crate::subagent::{SessionFactory, SubAgentSupervisor};
     use crate::test_support::MockSandbox;
 
     fn test_catalog() -> Arc<Catalog> {
@@ -329,11 +327,11 @@ mod tests {
     #[test]
     fn gemini_subagent_tools_registered() {
         let mut profile = GeminiProfile::new("gemini-2.0-flash");
-        let manager = Arc::new(AsyncMutex::new(SubAgentManager::new(3)));
+        let supervisor = SubAgentSupervisor::new(3);
         let factory: SessionFactory = Arc::new(|| {
             panic!("should not be called");
         });
-        profile.register_subagent_tools(manager, factory, 0);
+        profile.register_subagent_tools(supervisor, factory, 0);
         let names = profile.tool_registry().names();
         assert_eq!(names.len(), 14);
         assert!(names.contains(&"spawn_agent".to_string()));

lib/components/fabro-agent/src/profiles/openai.rs

@@ -290,10 +290,8 @@ in the project.");
 mod tests {
     use std::sync::Arc;
 
-    use tokio::sync::Mutex as AsyncMutex;
-
     use super::*;
-    use crate::subagent::{SessionFactory, SubAgentManager};
+    use crate::subagent::{SessionFactory, SubAgentSupervisor};
     use crate::test_support::MockSandbox;
 
     fn test_catalog() -> Arc<Catalog> {
@@ -384,9 +382,9 @@ mod tests {
         let mut profile = OpenAiProfile::new("o3-mini");
         assert_eq!(profile.tool_registry().names().len(), 9);
 
-        let manager = Arc::new(AsyncMutex::new(SubAgentManager::new(3)));
+        let supervisor = SubAgentSupervisor::new(3);
         let factory: SessionFactory = Arc::new(|| panic!("should not be called in test"));
-        profile.register_subagent_tools(manager, factory, 0);
+        profile.register_subagent_tools(supervisor, factory, 0);
         assert_eq!(profile.tool_registry().names().len(), 13);
     }
 

lib/components/fabro-agent/src/session.rs

@@ -21,7 +21,7 @@ use fabro_types::{
     StageContextWindowProjection, SteeringMessage,
 };
 use futures::StreamExt;
-use tokio::sync::{Mutex as AsyncMutex, Notify, broadcast};
+use tokio::sync::{Notify, broadcast};
 use tokio::time;
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, info, warn};
@@ -44,7 +44,7 @@ use crate::sandbox::Sandbox;
 use crate::skills::{
     ExpandedInput, Skill, default_skill_dirs, discover_skills, expand_skill, make_use_skill_tool,
 };
-use crate::subagent::{SubAgentCallbackEvent, SubAgentEventCallback, SubAgentManager};
+use crate::subagent::{SubAgentCallbackEvent, SubAgentEventCallback, SubAgentSupervisor};
 use crate::tool_execution::execute_tool_calls;
 use crate::tool_registry::ToolDefinitionWithSource;
 use crate::types::{
@@ -76,6 +76,13 @@ pub struct SessionInputTiming {
     pub tool:      Duration,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum SessionShutdownReason {
+    Completed,
+    Cancelled,
+    Error,
+}
+
 /// Take the value out of `start`, add its elapsed time to `total`. Used by
 /// `run_single_input` to accumulate inference and tool spans at well-defined
 /// boundaries (stream open, retry, error, cancel, end-of-loop).
@@ -106,8 +113,10 @@ impl From<SteeringMessage> for SteeringItem {
 
 #[derive(Default)]
 struct ControlState {
-    queue:             VecDeque<SteeringItem>,
+    queue: VecDeque<SteeringItem>,
     waiting_for_steer: bool,
+    interrupt_generation: u64,
+    settled_interrupt_generation: u64,
 }
 
 /// Trait that lets the workflow layer keep an agent in `process_input` when a
@@ -164,6 +173,7 @@ impl SessionControlHandle {
     pub fn interrupt(&self, _actor: Option<Principal>) {
         {
             let mut control = self.control.lock().expect("control state lock poisoned");
+            control.interrupt_generation = control.interrupt_generation.saturating_add(1);
             if control.queue.is_empty() {
                 control.waiting_for_steer = true;
             }
@@ -227,8 +237,20 @@ impl SessionControlHandle {
         item: SteeringItem,
         cap: usize,
     ) -> Option<SteeringItem> {
-        let evicted = self.push_bounded(item, cap);
+        let evicted = {
+            let mut control = self.control.lock().expect("control state lock poisoned");
+            let evicted = if control.queue.len() >= cap {
+                control.queue.pop_front()
+            } else {
+                None
+            };
+            control.interrupt_generation = control.interrupt_generation.saturating_add(1);
+            control.queue.push_back(item);
+            control.waiting_for_steer = false;
+            evicted
+        };
         self.cancel_round();
+        self.notify.notify_waiters();
         evicted
     }
 
@@ -251,7 +273,7 @@ impl SessionControlHandle {
     fn interrupt_then_enqueue(&self, item: SteeringItem) {
         {
             let mut control = self.control.lock().expect("control state lock poisoned");
-            control.waiting_for_steer = true;
+            control.interrupt_generation = control.interrupt_generation.saturating_add(1);
             control.queue.push_back(item);
             control.waiting_for_steer = false;
         }
@@ -334,6 +356,7 @@ pub struct Session {
     history: History,
     event_emitter: Emitter,
     state: SessionState,
+    ended: bool,
     llm_client: Client,
     provider_profile: Arc<dyn AgentProfile>,
     sandbox: Arc<dyn Sandbox>,
@@ -350,7 +373,7 @@ pub struct Session {
     activated_skill_context_observed: bool,
     file_tracker: FileTracker,
     tool_env_provider: Option<Arc<dyn ToolEnvProvider>>,
-    subagent_manager: Option<Arc<AsyncMutex<SubAgentManager>>>,
+    subagent_supervisor: Option<SubAgentSupervisor>,
     completion_coordinator: Option<Arc<dyn CompletionCoordinator>>,
     last_input_timing: SessionInputTiming,
     last_input_usage: TokenCounts,
@@ -364,7 +387,7 @@ impl Session {
         provider_profile: Arc<dyn AgentProfile>,
         sandbox: Arc<dyn Sandbox>,
         config: SessionOptions,
-        subagent_manager: Option<Arc<AsyncMutex<SubAgentManager>>>,
+        subagent_supervisor: Option<SubAgentSupervisor>,
     ) -> Self {
         let id = uuid::Uuid::new_v4().to_string();
         Self {
@@ -374,6 +397,7 @@ impl Session {
             history: History::default(),
             event_emitter: Emitter::new(),
             state: SessionState::Idle,
+            ended: false,
             llm_client,
             provider_profile,
             sandbox,
@@ -390,7 +414,7 @@ impl Session {
             activated_skill_context_observed: false,
             file_tracker: FileTracker::default(),
             tool_env_provider: None,
-            subagent_manager,
+            subagent_supervisor,
             completion_coordinator: None,
             last_input_timing: SessionInputTiming::default(),
             last_input_usage: TokenCounts::default(),
@@ -414,7 +438,7 @@ impl Session {
         provider_profile: Arc<dyn AgentProfile>,
         sandbox: Arc<dyn Sandbox>,
         config: SessionOptions,
-        subagent_manager: Option<Arc<AsyncMutex<SubAgentManager>>>,
+        subagent_supervisor: Option<SubAgentSupervisor>,
     ) -> Result<Self, LlmError> {
         let client = Client::from_source(source, catalog).await?;
         Ok(Self::new(
@@ -422,7 +446,7 @@ impl Session {
             provider_profile,
             sandbox,
             config,
-            subagent_manager,
+            subagent_supervisor,
         ))
     }
 
@@ -433,14 +457,14 @@ impl Session {
         provider_profile: Arc<dyn AgentProfile>,
         sandbox: Arc<dyn Sandbox>,
         config: SessionOptions,
-        subagent_manager: Option<Arc<AsyncMutex<SubAgentManager>>>,
+        subagent_supervisor: Option<SubAgentSupervisor>,
     ) -> Result<Self, Error> {
         let mut session = Self::new(
             llm_client,
             provider_profile,
             sandbox,
             config,
-            subagent_manager,
+            subagent_supervisor,
         );
         session.id = record.id.to_string();
         // from_record represents a fresh root session by default; callers
@@ -1128,18 +1152,20 @@ impl Session {
         })
     }
 
-    /// Transition the session state machine, emitting events and running
-    /// cleanup as appropriate for each transition.
+    /// Transition the in-memory session state machine.
     ///
     /// Valid transitions (matches the Attractor spec):
     /// - Idle → Thinking
     /// - Thinking → Executing
     /// - Thinking → Idle  (emits ProcessingEnd)
     /// - Executing → Thinking
-    /// - Thinking → Closed (emits SessionEnded)
-    /// - Executing → Closed (emits SessionEnded)
-    /// - Idle → Closed (emits SessionEnded)
-    /// - any → Closed (interrupt/error — emits SessionEnded)
+    /// - Thinking → Closed
+    /// - Executing → Closed
+    /// - Idle → Closed
+    /// - any → Closed (interrupt/error)
+    ///
+    /// Async resource cleanup and `SessionEnded` emission belong to
+    /// [`Self::shutdown`], never to this synchronous transition helper.
     fn transition(&mut self, to: SessionState) {
         let from = self.state;
         if from == to {
@@ -1160,17 +1186,6 @@ impl Session {
             "Invalid session state transition: {from:?} -> {to:?}"
         );
 
-        if to == SessionState::Closed && from != SessionState::Closed {
-            // Clean up subagents before emitting SessionEnded
-            if let Some(ref manager) = self.subagent_manager {
-                if let Ok(mut mgr) = manager.try_lock() {
-                    mgr.close_all();
-                }
-            }
-            self.event_emitter
-                .emit(self.id.clone(), AgentEvent::SessionEnded);
-        }
-
         if matches!(from, SessionState::Thinking | SessionState::Executing)
             && to == SessionState::Idle
         {
@@ -1181,10 +1196,24 @@ impl Session {
         self.state = to;
     }
 
-    pub fn close(&mut self) -> bool {
-        let was_open = self.state != SessionState::Closed;
+    /// Close the session and resolve all owned child tasks before emitting
+    /// `SessionEnded`. Returns `true` only for the call that performs shutdown.
+    pub async fn shutdown(&mut self, reason: SessionShutdownReason) -> bool {
+        if self.ended {
+            return false;
+        }
+        if reason == SessionShutdownReason::Cancelled {
+            self.set_interrupt_reason(InterruptReason::Cancelled);
+            self.cancel_token.cancel();
+        }
         self.transition(SessionState::Closed);
-        was_open
+        if let Some(supervisor) = &self.subagent_supervisor {
+            supervisor.shutdown_all().await;
+        }
+        self.ended = true;
+        self.event_emitter
+            .emit(self.id.clone(), AgentEvent::SessionEnded);
+        true
     }
 
     pub fn set_reasoning_effort(&mut self, effort: Option<ReasoningEffort>) {
@@ -1317,8 +1346,14 @@ impl Session {
             handle.abort();
         }
 
-        // Only transition to Idle if the session wasn't closed by an error
-        if self.state != SessionState::Closed {
+        if self.state == SessionState::Closed {
+            let reason = if self.cancel_token.is_cancelled() {
+                SessionShutdownReason::Cancelled
+            } else {
+                SessionShutdownReason::Error
+            };
+            self.shutdown(reason).await;
+        } else {
             self.transition(SessionState::Idle);
         }
 
@@ -1378,7 +1413,7 @@ impl Session {
             // swap in a fresh one before draining and rebuilding state.
             // (Terminal cancel via `cancel_token` is handled by the explicit
             // check below and by `interrupted_error()`.)
-            {
+            let round_was_interrupted = {
                 let needs_refresh = self
                     .round_token
                     .read()
@@ -1388,15 +1423,37 @@ impl Session {
                     *self.round_token.write().expect("round token lock poisoned") =
                         CancellationToken::new();
                 }
-            }
+                needs_refresh
+            };
 
             // Terminal cancellation wins even when a control interrupt has
             // parked the session waiting for steering.
             if self.cancel_token.is_cancelled() {
-                self.close();
+                self.shutdown(SessionShutdownReason::Cancelled).await;
                 return Err(self.interrupted_error());
             }
 
+            if round_was_interrupted {
+                let generations = {
+                    let mut control = self
+                        .control_state
+                        .lock()
+                        .expect("control state lock poisoned");
+                    let first = control.settled_interrupt_generation.saturating_add(1);
+                    let last = control.interrupt_generation;
+                    control.settled_interrupt_generation = last;
+                    if first <= last {
+                        (first..=last).collect::<Vec<_>>()
+                    } else {
+                        Vec::new()
+                    }
+                };
+                for generation in generations {
+                    self.event_emitter
+                        .emit(self.id.clone(), AgentEvent::RoundInterrupted { generation });
+                }
+            }
+
             // Drain pending steering messages at the top of every iteration
             // so steering pushed mid-round is delivered as the first turn of
             // the next round. A pure interrupt with no queued steer parks the
@@ -1464,7 +1521,7 @@ impl Session {
             } else {
                 record_elapsed(&mut inference_start, &mut timing.inference);
                 if self.cancel_token.is_cancelled() {
-                    self.close();
+                    self.shutdown(SessionShutdownReason::Cancelled).await;
                     return Err(self.interrupted_error());
                 }
                 // Round-only cancel before stream opened — re-iterate to
@@ -1539,7 +1596,7 @@ impl Session {
                 if self.cancel_token.is_cancelled() {
                     drop(event_stream);
                     record_elapsed(&mut inference_start, &mut timing.inference);
-                    self.close();
+                    self.shutdown(SessionShutdownReason::Cancelled).await;
                     return Err(self.interrupted_error());
                 }
 
@@ -1757,6 +1814,9 @@ impl Session {
             // completion coordinator: it can return `true` to force one more
             // iteration when a steer arrived during the final response.
             if tool_calls.is_empty() {
+                if round_token.is_cancelled() {
+                    continue;
+                }
                 let should_continue = self
                     .completion_coordinator
                     .as_ref()
@@ -1823,7 +1883,7 @@ impl Session {
 
             // Terminal cancel takes precedence: close and return.
             if self.cancel_token.is_cancelled() {
-                self.close();
+                self.shutdown(SessionShutdownReason::Cancelled).await;
                 return Err(self.interrupted_error());
             }
 
@@ -1936,7 +1996,7 @@ impl Session {
             tokio::select! {
                 biased;
                 () = self.cancel_token.cancelled() => {
-                    self.close();
+                    self.shutdown(SessionShutdownReason::Cancelled).await;
                     return Err(self.interrupted_error());
                 }
                 () = notified => {}
@@ -2060,7 +2120,7 @@ mod tests {
     use super::*;
     use crate::config::{ToolAccess, ToolAccessPolicy, ToolApprovalAdapter, ToolExposureMode};
     use crate::skills::{Skill, make_use_skill_tool};
-    use crate::subagent::SubAgentStatus;
+    use crate::subagent::{SubAgentStatus, make_wait_tool};
     use crate::test_support::*;
     use crate::tool_registry::{RegisteredTool, ToolContext, ToolRegistry, ToolSource};
 
@@ -2215,13 +2275,51 @@ mod tests {
         }
     }
 
+    struct BlockingFirstStreamProvider {
+        first_started: Arc<Notify>,
+        response:      Response,
+        call_index:    AtomicUsize,
+    }
+
+    impl BlockingFirstStreamProvider {
+        fn new(response: Response) -> Self {
+            Self {
+                first_started: Arc::new(Notify::new()),
+                response,
+                call_index: AtomicUsize::new(0),
+            }
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl ProviderAdapter for BlockingFirstStreamProvider {
+        fn name(&self) -> &'static str {
+            "mock"
+        }
+
+        async fn complete(&self, _request: &Request) -> Result<Response, LlmError> {
+            Err(LlmError::Configuration {
+                message: "BlockingFirstStreamProvider does not implement complete()".into(),
+                source:  None,
+            })
+        }
+
+        async fn stream(&self, _request: &Request) -> Result<StreamEventStream, LlmError> {
+            if self.call_index.fetch_add(1, Ordering::SeqCst) == 0 {
+                self.first_started.notify_one();
+                return std::future::pending().await;
+            }
+            Ok(response_to_stream(self.response.clone()))
+        }
+    }
+
     async fn make_session_with_provider(provider: Arc<dyn ProviderAdapter>) -> Session {
         make_session_with_provider_and_manager(provider, None).await
     }
 
     async fn make_session_with_provider_and_manager(
         provider: Arc<dyn ProviderAdapter>,
-        subagent_manager: Option<Arc<AsyncMutex<SubAgentManager>>>,
+        subagent_supervisor: Option<SubAgentSupervisor>,
     ) -> Session {
         let client = make_client(provider).await;
         let profile = Arc::new(TestProfile::new());
@@ -2231,7 +2329,7 @@ mod tests {
             profile,
             env,
             SessionOptions::default(),
-            subagent_manager,
+            subagent_supervisor,
         )
     }
 
@@ -2489,6 +2587,7 @@ mod tests {
     #[tokio::test]
     async fn pure_interrupt_waits_until_later_steer() {
         let mut session = make_session(vec![text_response("OK")]).await;
+        let mut events = session.subscribe();
         let handle = session.control_handle();
         handle.interrupt(None);
 
@@ -2506,6 +2605,13 @@ mod tests {
         let turns = session.history().turns();
         assert!(matches!(&turns[1], Message::Steering { content, .. } if content == "resume now"));
         assert!(!handle.is_waiting_for_steer());
+        let generations = std::iter::from_fn(|| events.try_recv().ok())
+            .filter_map(|event| match event.event {
+                AgentEvent::RoundInterrupted { generation } => Some(generation),
+                _ => None,
+            })
+            .collect::<Vec<_>>();
+        assert_eq!(generations, vec![1]);
     }
 
     #[tokio::test]
@@ -2517,14 +2623,152 @@ mod tests {
         handle.interrupt_then_steer("stop now".to_string(), None);
         session.process_input("start").await.unwrap();
 
-        let mut found_text = None;
-        while let Ok(ev) = rx.try_recv() {
-            if let AgentEvent::SteeringInjected { text, .. } = ev.event {
-                found_text = Some(text);
-                break;
-            }
-        }
-        assert_eq!(found_text.as_deref(), Some("stop now"));
+        let events = std::iter::from_fn(|| rx.try_recv().ok())
+            .map(|event| event.event)
+            .collect::<Vec<_>>();
+        let settled = events
+            .iter()
+            .position(|event| matches!(event, AgentEvent::RoundInterrupted { generation: 1 }))
+            .unwrap();
+        let steered = events
+            .iter()
+            .position(|event| {
+                matches!(
+                    event,
+                    AgentEvent::SteeringInjected { text, .. } if text == "stop now"
+                )
+            })
+            .unwrap();
+        assert!(settled < steered);
+        assert!(!handle.is_waiting_for_steer());
+    }
+
+    #[tokio::test]
+    async fn interrupt_during_inference_settles_once_before_steering_resumes() {
+        let provider = Arc::new(BlockingFirstStreamProvider::new(text_response("resumed")));
+        let first_started = Arc::clone(&provider.first_started);
+        let mut session = make_session_with_provider(provider.clone()).await;
+        let control = session.control_handle();
+        let mut controller_events = session.subscribe();
+        let mut recorded_events = session.subscribe();
+        let control_for_controller = control.clone();
+        let controller = tokio::spawn(async move {
+            first_started.notified().await;
+            control_for_controller.interrupt(None);
+            wait_for_agent_event(&mut controller_events, |event| {
+                matches!(event, AgentEvent::RoundInterrupted { generation: 1 })
+            })
+            .await;
+            assert!(control_for_controller.is_waiting_for_steer());
+            control_for_controller.steer("resume inference".into(), None);
+        });
+
+        timeout(Duration::from_secs(1), session.process_input("start"))
+            .await
+            .expect("inference interrupt should settle and resume")
+            .unwrap();
+        controller.await.unwrap();
+
+        let events = std::iter::from_fn(|| recorded_events.try_recv().ok())
+            .map(|event| event.event)
+            .collect::<Vec<_>>();
+        assert_eq!(
+            events
+                .iter()
+                .filter(|event| matches!(event, AgentEvent::RoundInterrupted { .. }))
+                .count(),
+            1
+        );
+        let settled = events
+            .iter()
+            .position(|event| matches!(event, AgentEvent::RoundInterrupted { .. }))
+            .unwrap();
+        let steered = events
+            .iter()
+            .position(|event| matches!(event, AgentEvent::SteeringInjected { .. }))
+            .unwrap();
+        assert!(settled < steered);
+        assert_eq!(provider.call_index.load(Ordering::SeqCst), 2);
+        assert!(!control.is_waiting_for_steer());
+    }
+
+    #[tokio::test]
+    async fn interrupt_during_tool_settles_once_after_balancing_tool_result() {
+        let blocking_tool = RegisteredTool {
+            definition: ToolDefinition {
+                name:        "block".into(),
+                description: "Blocks until interrupted".into(),
+                parameters:  serde_json::json!({"type": "object"}),
+            },
+            executor:   Arc::new(|_args, ctx| {
+                Box::pin(async move {
+                    ctx.cancel.cancelled().await;
+                    Err("Cancelled".to_string())
+                })
+            }),
+            source:     ToolSource::Native,
+        };
+        let mut registry = ToolRegistry::new();
+        registry.register(blocking_tool);
+        let responses = vec![
+            tool_call_response("block", "call_block", serde_json::json!({})),
+            text_response("resumed"),
+        ];
+        let mut session = make_session_with_tools(responses, registry).await;
+        let control = session.control_handle();
+        let mut controller_events = session.subscribe();
+        let mut recorded_events = session.subscribe();
+        let control_for_controller = control.clone();
+        let controller = tokio::spawn(async move {
+            wait_for_agent_event(&mut controller_events, |event| {
+                matches!(
+                    event,
+                    AgentEvent::ToolCallStarted { tool_name, .. } if tool_name == "block"
+                )
+            })
+            .await;
+            control_for_controller.interrupt(None);
+            wait_for_agent_event(&mut controller_events, |event| {
+                matches!(event, AgentEvent::RoundInterrupted { generation: 1 })
+            })
+            .await;
+            assert!(control_for_controller.is_waiting_for_steer());
+            control_for_controller.steer("resume after tool".into(), None);
+        });
+
+        timeout(
+            Duration::from_secs(1),
+            session.process_input("use the tool"),
+        )
+        .await
+        .expect("tool interrupt should settle and resume")
+        .unwrap();
+        controller.await.unwrap();
+
+        let events = std::iter::from_fn(|| recorded_events.try_recv().ok())
+            .map(|event| event.event)
+            .collect::<Vec<_>>();
+        assert_eq!(
+            events
+                .iter()
+                .filter(|event| matches!(event, AgentEvent::RoundInterrupted { .. }))
+                .count(),
+            1
+        );
+        let tool_completed = events
+            .iter()
+            .position(|event| matches!(event, AgentEvent::ToolCallCompleted { .. }))
+            .unwrap();
+        let settled = events
+            .iter()
+            .position(|event| matches!(event, AgentEvent::RoundInterrupted { .. }))
+            .unwrap();
+        assert!(tool_completed < settled);
+        assert!(matches!(
+            session.history().turns().get(2),
+            Some(Message::ToolResults { .. })
+        ));
+        assert!(!control.is_waiting_for_steer());
     }
 
     #[tokio::test]
@@ -2614,7 +2858,7 @@ mod tests {
 
         session.initialize().await.unwrap();
         session.process_input("Hi").await.unwrap();
-        session.close();
+        session.shutdown(SessionShutdownReason::Completed).await;
 
         // Collect events
         let mut events = Vec::new();
@@ -2929,7 +3173,7 @@ mod tests {
     #[tokio::test]
     async fn closed_session_rejects_input() {
         let mut session = make_session(vec![]).await;
-        session.close();
+        session.shutdown(SessionShutdownReason::Completed).await;
         assert_eq!(session.state(), SessionState::Closed);
 
         let result = session.process_input("Hello").await;
@@ -2942,8 +3186,8 @@ mod tests {
         let mut session = make_session(vec![]).await;
         let mut rx = session.subscribe();
 
-        assert!(session.close());
-        assert!(!session.close());
+        assert!(session.shutdown(SessionShutdownReason::Completed).await);
+        assert!(!session.shutdown(SessionShutdownReason::Completed).await);
 
         let events: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
         assert_eq!(
@@ -2958,7 +3202,7 @@ mod tests {
     #[tokio::test]
     async fn closed_session_does_not_emit_session_start() {
         let mut session = make_session(vec![]).await;
-        session.close();
+        session.shutdown(SessionShutdownReason::Completed).await;
 
         let mut rx = session.subscribe();
         let result = session.process_input("Hello").await;
@@ -3198,7 +3442,7 @@ mod tests {
         session.initialize().await.unwrap();
         session.process_input("one").await.unwrap();
         session.process_input("two").await.unwrap();
-        session.close();
+        session.shutdown(SessionShutdownReason::Completed).await;
 
         let mut session_start_count = 0;
         let mut session_end_count = 0;
@@ -4648,35 +4892,209 @@ mod tests {
         );
     }
 
-    #[tokio::test]
-    async fn close_cleans_up_subagents_before_emitting_session_ended() {
-        use crate::subagent::SubAgentManager;
+    async fn make_parent_waiting_on_blocked_subagent()
+    -> (Session, SubAgentSupervisor, String, CancellationToken) {
+        let block_until_cancelled = RegisteredTool {
+            definition: ToolDefinition {
+                name:        "block_until_cancelled".into(),
+                description: "Waits until cancelled".into(),
+                parameters:  serde_json::json!({"type": "object"}),
+            },
+            executor:   Arc::new(|_args, ctx| {
+                Box::pin(async move {
+                    ctx.cancel.cancelled().await;
+                    Ok("cancelled".to_string())
+                })
+            }),
+            source:     ToolSource::Native,
+        };
+        let mut child_registry = ToolRegistry::new();
+        child_registry.register(block_until_cancelled);
+        let child = make_session_with_tools(
+            vec![tool_call_response(
+                "block_until_cancelled",
+                "child_call",
+                serde_json::json!({}),
+            )],
+            child_registry,
+        )
+        .await;
+        let child_cancel = child.cancel_token();
 
-        let manager = Arc::new(AsyncMutex::new(SubAgentManager::new(3)));
+        let supervisor = SubAgentSupervisor::new(3);
+        let agent_id = supervisor
+            .spawn(child, "block until cancelled".into(), 0)
+            .unwrap();
+
+        let mut parent_registry = ToolRegistry::new();
+        parent_registry.register(make_wait_tool(supervisor.clone()));
+        let parent_provider = Arc::new(ScriptedStreamProvider::new(vec![
+            ScriptedStreamCall::Response(Box::new(tool_call_response(
+                "wait",
+                "parent_wait_call",
+                serde_json::json!({ "agent_id": agent_id }),
+            ))),
+            ScriptedStreamCall::Response(Box::new(text_response("resumed"))),
+        ]));
+        let client = make_client(parent_provider).await;
+        let profile = Arc::new(TestProfile::with_tools(parent_registry));
+        let env = Arc::new(MockSandbox::default());
+        let session = Session::new(
+            client,
+            profile,
+            env,
+            SessionOptions::default(),
+            Some(supervisor.clone()),
+        );
+        supervisor.set_event_callback(session.sub_agent_event_callback());
+
+        (session, supervisor, agent_id, child_cancel)
+    }
+
+    async fn wait_for_agent_event(
+        rx: &mut broadcast::Receiver<SessionEvent>,
+        predicate: impl Fn(&AgentEvent) -> bool,
+    ) {
+        loop {
+            let event = rx
+                .recv()
+                .await
+                .expect("session event stream should remain open");
+            if predicate(&event.event) {
+                return;
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn control_interrupt_during_subagent_wait_closes_child_and_resumes_after_steer() {
+        let (mut session, manager, agent_id, child_cancel) =
+            make_parent_waiting_on_blocked_subagent().await;
+        let control = session.control_handle();
+        let mut events = session.subscribe();
+        let mut recorded_events = session.subscribe();
+        let control_for_controller = control.clone();
+        let controller = tokio::spawn(async move {
+            wait_for_agent_event(&mut events, |event| {
+                matches!(
+                    event,
+                    AgentEvent::ToolCallStarted { tool_name, .. } if tool_name == "wait"
+                )
+            })
+            .await;
+            control_for_controller.interrupt(None);
+            wait_for_agent_event(&mut events, |event| {
+                matches!(event, AgentEvent::SubAgentClosed { .. })
+            })
+            .await;
+            wait_for_agent_event(&mut events, |event| {
+                matches!(event, AgentEvent::RoundInterrupted { generation: 1 })
+            })
+            .await;
+            assert!(control_for_controller.is_waiting_for_steer());
+            control_for_controller.steer("resume after interrupt".into(), None);
+        });
+
+        timeout(
+            Duration::from_secs(1),
+            session.process_input("wait for the child"),
+        )
+        .await
+        .expect("interrupt should unblock the subagent wait")
+        .unwrap();
+        controller.await.unwrap();
+
+        assert_eq!(session.state(), SessionState::Idle);
+        assert!(child_cancel.is_cancelled());
+        assert!(matches!(
+            manager.status(&agent_id),
+            Some(SubAgentStatus::Closed)
+        ));
+        let events = std::iter::from_fn(|| recorded_events.try_recv().ok())
+            .map(|event| event.event)
+            .collect::<Vec<_>>();
+        assert_eq!(
+            events
+                .iter()
+                .filter(|event| matches!(event, AgentEvent::RoundInterrupted { .. }))
+                .count(),
+            1
+        );
+        let child_closed = events
+            .iter()
+            .position(|event| matches!(event, AgentEvent::SubAgentClosed { .. }))
+            .unwrap();
+        let settled = events
+            .iter()
+            .position(|event| matches!(event, AgentEvent::RoundInterrupted { .. }))
+            .unwrap();
+        assert!(child_closed < settled);
+        assert!(!control.is_waiting_for_steer());
+    }
+
+    #[tokio::test]
+    async fn terminal_cancel_during_subagent_wait_closes_child_and_session() {
+        let (mut session, manager, agent_id, child_cancel) =
+            make_parent_waiting_on_blocked_subagent().await;
+        let cancel = session.cancel_token();
+        let mut events = session.subscribe();
+        let controller = tokio::spawn(async move {
+            wait_for_agent_event(&mut events, |event| {
+                matches!(
+                    event,
+                    AgentEvent::ToolCallStarted { tool_name, .. } if tool_name == "wait"
+                )
+            })
+            .await;
+            cancel.cancel();
+        });
+
+        let result = timeout(
+            Duration::from_secs(1),
+            session.process_input("wait for the child"),
+        )
+        .await
+        .expect("terminal cancellation should unblock the subagent wait");
+        controller.await.unwrap();
+
+        assert!(matches!(
+            result,
+            Err(Error::Interrupted(InterruptReason::Cancelled))
+        ));
+        assert_eq!(session.state(), SessionState::Closed);
+        assert!(child_cancel.is_cancelled());
+        assert!(matches!(
+            manager.status(&agent_id),
+            Some(SubAgentStatus::Closed)
+        ));
+    }
+
+    #[tokio::test]
+    async fn shutdown_cleans_up_subagents_before_emitting_session_ended() {
+        let supervisor = SubAgentSupervisor::new(3);
 
         let provider = Arc::new(ScriptedStreamProvider::new(vec![
             ScriptedStreamCall::Response(Box::new(text_response("done"))),
         ]));
         let mut session =
-            make_session_with_provider_and_manager(provider, Some(manager.clone())).await;
+            make_session_with_provider_and_manager(provider, Some(supervisor.clone())).await;
 
-        // Wire the manager's event callback to the session's emitter
-        manager
-            .lock()
-            .await
-            .set_event_callback(session.sub_agent_event_callback());
+        supervisor.set_event_callback(session.sub_agent_event_callback());
 
-        // Spawn a subagent
-        let child = make_session(vec![text_response("child done")]).await;
-        let agent_id = manager.lock().await.spawn(child, "task".into(), 0).unwrap();
+        let child_provider = Arc::new(DelayedStreamProvider::new(
+            vec![text_response("child done")],
+            Duration::from_mins(1),
+        ));
+        let child = make_session_with_provider(child_provider).await;
+        let agent_id = supervisor.spawn(child, "task".into(), 0).unwrap();
 
         // Collect events
         let mut rx = session.subscribe();
-        session.close();
+        session.shutdown(SessionShutdownReason::Completed).await;
 
         // The subagent should have been closed
         assert!(matches!(
-            manager.lock().await.status(&agent_id),
+            supervisor.status(&agent_id),
             Some(SubAgentStatus::Closed)
         ));
 

lib/components/fabro-agent/src/types.rs

@@ -292,6 +292,11 @@ pub enum AgentEvent {
         #[serde(default, skip_serializing_if = "Option::is_none")]
         actor: Option<fabro_types::Principal>,
     },
+    /// The cancelled round has fully unwound and the session is ready to
+    /// consume queued steering or wait for a later steering message.
+    RoundInterrupted {
+        generation: u64,
+    },
     CompactionStarted {
         estimated_tokens:    usize,
         context_window_size: usize,
@@ -464,6 +469,9 @@ impl AgentEvent {
             Self::SteeringInjected { text, .. } => {
                 debug!(session_id, text_len = text.len(), "Steering injected");
             }
+            Self::RoundInterrupted { generation } => {
+                debug!(session_id, generation, "Agent round interrupted");
+            }
             Self::CompactionStarted {
                 estimated_tokens,
                 context_window_size,

lib/components/fabro-agent/tests/it/parity_matrix.rs

@@ -11,7 +11,7 @@ use std::sync::Arc;
 use fabro_agent::subagent::SessionFactory;
 use fabro_agent::{
     AgentEvent, AgentProfile, AnthropicProfile, GeminiProfile, LocalSandbox, OpenAiProfile,
-    Session, SessionOptions, SubAgentManager, WebFetchSummarizer,
+    Session, SessionOptions, SubAgentSupervisor, WebFetchSummarizer,
 };
 use fabro_auth::EnvCredentialSource;
 use fabro_llm::client::Client;
@@ -20,7 +20,6 @@ use fabro_llm::providers::{OpenAiAdapter, OpenAiCompatibleAdapter};
 use fabro_model::catalog::{LlmCatalogSettings, ProviderCatalogSettings};
 use fabro_model::{Catalog, ModelHandle, ProviderId};
 use fabro_test::{TwinScenario, TwinScenarios, TwinToolCall, twin_openai};
-use tokio::sync::Mutex as AsyncMutex;
 
 type Provider = ProviderId;
 
@@ -85,7 +84,7 @@ async fn make_session(
 
     // Register subagent tools so spawn_agent / wait / send_input / close_agent are
     // available
-    let manager = Arc::new(AsyncMutex::new(SubAgentManager::new(3)));
+    let supervisor = SubAgentSupervisor::new(3);
     let factory_client = client.clone();
     let factory_model: String = model.to_string();
     let factory_cwd = cwd.to_path_buf();
@@ -123,10 +122,16 @@ async fn make_session(
             None,
         )
     });
-    profile.register_subagent_tools(manager, factory, 0);
+    profile.register_subagent_tools(supervisor.clone(), factory, 0);
 
     let profile: Arc<dyn AgentProfile> = Arc::from(profile);
-    Session::new(client, profile, env, SessionOptions::default(), None)
+    Session::new(
+        client,
+        profile,
+        env,
+        SessionOptions::default(),
+        Some(supervisor),
+    )
 }
 
 async fn make_session_with_config(

lib/components/fabro-store/src/run_state.rs

@@ -8,8 +8,8 @@ use fabro_types::run_event::{
 };
 use fabro_types::settings::run::{EnvironmentProvider, RunEnvironmentSettings};
 use fabro_types::{
-    ActivatedSkill, AskFabro, BilledModelUsage, BilledTokenCounts, Checkpoint, CheckpointRecord,
-    CommandTermination, Conclusion, EventBody, FailureCategory, FailureSignature,
+    ActivatedSkill, AgentControlState, AskFabro, BilledModelUsage, BilledTokenCounts, Checkpoint,
+    CheckpointRecord, CommandTermination, Conclusion, EventBody, FailureCategory, FailureSignature,
     InterviewQuestionRecord, McpServerProjection, McpServerStatus, Outcome, PendingInterviewRecord,
     PendingReason, PullRequestLink, RepositoryRef, Run, RunApproval, RunApprovalState,
     RunBillingSummary, RunControlAction, RunDiff, RunEvent, RunId, RunLifecycle, RunLinks,
@@ -352,6 +352,7 @@ impl RunProjectionReducer for RunProjection {
                     return Ok(());
                 };
                 stage.state = StageState::Retrying;
+                stage.agent_control = AgentControlState::Running;
             }
             EventBody::StagePrompt(props) => {
                 let Some(stage) = stage_at_stored_or_visit(self, stored, props.visit, event.seq)
@@ -388,6 +389,7 @@ impl RunProjectionReducer for RunProjection {
                     stage.model = Some(billing.model().clone());
                 }
                 stage.state = StageState::from(outcome.status);
+                stage.agent_control = AgentControlState::Running;
             }
             EventBody::StageFailed(props) => {
                 let failure_reason = props.failure.as_ref().map(|detail| detail.message.clone());
@@ -411,6 +413,7 @@ impl RunProjectionReducer for RunProjection {
                 }
                 stage.state =
                     stage_state_from_failure(props.will_retry, failure_category, stage.termination);
+                stage.agent_control = AgentControlState::Running;
             }
             EventBody::AgentMessage(props) => {
                 let Some(stage) = stage_at_stored_or_visit(self, stored, props.visit, event.seq)
@@ -433,6 +436,27 @@ impl RunProjectionReducer for RunProjection {
                 stage.provider_used = Some(StageModelUsage::from_agent_session_activated(props));
                 stage.permission_level = props.permission_level;
             }
+            EventBody::AgentRoundInterrupted(props) => {
+                let Some(stage) = stage_at_stored_or_visit(self, stored, props.visit, event.seq)
+                else {
+                    return Ok(());
+                };
+                stage.agent_control = AgentControlState::WaitingForSteer;
+            }
+            EventBody::AgentSteeringInjected(props) => {
+                let Some(stage) = stage_at_stored_or_visit(self, stored, props.visit, event.seq)
+                else {
+                    return Ok(());
+                };
+                stage.agent_control = AgentControlState::Running;
+            }
+            EventBody::AgentSessionDeactivated(props) => {
+                let Some(stage) = stage_at_stored_or_visit(self, stored, props.visit, event.seq)
+                else {
+                    return Ok(());
+                };
+                stage.agent_control = AgentControlState::Running;
+            }
             EventBody::AgentToolsAvailable(props) => {
                 let Some(stage) = stage_at_stored_or_visit(self, stored, props.visit, event.seq)
                 else {
@@ -544,6 +568,7 @@ impl RunProjectionReducer for RunProjection {
                 });
                 stage.timing = Some(fabro_types::StageTiming::wall_only(props.duration_ms));
                 stage.state = StageState::from(outcome);
+                stage.agent_control = AgentControlState::Running;
             }
             EventBody::TodoCreated(props) => {
                 if !should_project_root_agent_todo_event(stored, props.list_kind) {
@@ -1274,6 +1299,7 @@ fn apply_agent_terminal(
     stage.output = Some(output);
     stage.termination = Some(termination);
     stage.script_timing = Some(script_timing);
+    stage.agent_control = AgentControlState::Running;
     Ok(())
 }
 
@@ -1296,11 +1322,12 @@ mod tests {
     use fabro_types::run_event::{
         AgentAcpCancelledProps, AgentAcpCompletedProps, AgentAcpStartedProps,
         AgentAcpTimedOutProps, AgentMcpFailedProps, AgentMcpReadyProps, AgentMcpToolSummary,
-        AgentMessageProps, AgentSessionActivatedProps, AgentSessionEndedProps,
-        AgentSessionStartedProps, AgentSkillActivatedProps, AgentSkillActivationSource,
-        AgentSkillSummary, AgentSkillsDiscoveredProps, AgentSubClosedProps, AgentSubCompletedProps,
-        AgentSubFailedProps, AgentSubSpawnedProps, AgentToolCategory, AgentToolSource,
-        AgentToolStartedProps, AgentToolSummary, AgentToolsAvailableProps,
+        AgentMessageProps, AgentRoundInterruptedProps, AgentSessionActivatedProps,
+        AgentSessionDeactivatedProps, AgentSessionEndedProps, AgentSessionStartedProps,
+        AgentSkillActivatedProps, AgentSkillActivationSource, AgentSkillSummary,
+        AgentSkillsDiscoveredProps, AgentSteeringInjectedProps, AgentSubClosedProps,
+        AgentSubCompletedProps, AgentSubFailedProps, AgentSubSpawnedProps, AgentToolCategory,
+        AgentToolSource, AgentToolStartedProps, AgentToolSummary, AgentToolsAvailableProps,
         CheckpointCompletedProps, InterviewCompletedProps, InterviewOption, InterviewStartedProps,
         ParallelBranchCompletedProps, ParallelBranchStartedProps, RunCompletedProps,
         RunControlEffectProps, StageCompletedProps, StageFailedProps, StagePromptProps,
@@ -1308,15 +1335,15 @@ mod tests {
     };
     use fabro_types::settings::run::{DockerfileSource, EnvironmentProvider};
     use fabro_types::{
-        AgentBackend, AutomationRef, BilledModelUsage, BilledTokenCounts, BlockedReason,
-        Checkpoint, CheckpointRecord, CommandTermination, EventBody, FailureCategory,
-        FailureDetail, FailureReason, Graph, McpServerStatus, Outcome, PendingReason,
-        PermissionLevel, PullRequestLink, QuestionType, ReasoningEffort, RunApprovalState,
-        RunBlobId, RunControlAction, RunDiff, RunEvent, RunSize, RunSpec, RunStatus, Speed,
-        StageContextWindowBreakdownItem, StageContextWindowCategory, StageContextWindowCountMethod,
-        StageContextWindowProjection, StageContextWindowStaleness, StageContextWindowWarning,
-        StageModelUsage, StageOutcome, StageState, SubAgentStatus, SuccessReason, WorkflowSettings,
-        first_event_seq, fixtures, test_support,
+        AgentBackend, AgentControlState, AutomationRef, BilledModelUsage, BilledTokenCounts,
+        BlockedReason, Checkpoint, CheckpointRecord, CommandTermination, EventBody,
+        FailureCategory, FailureDetail, FailureReason, Graph, McpServerStatus, Outcome,
+        PendingReason, PermissionLevel, PullRequestLink, QuestionType, ReasoningEffort,
+        RunApprovalState, RunBlobId, RunControlAction, RunDiff, RunEvent, RunSize, RunSpec,
+        RunStatus, Speed, StageContextWindowBreakdownItem, StageContextWindowCategory,
+        StageContextWindowCountMethod, StageContextWindowProjection, StageContextWindowStaleness,
+        StageContextWindowWarning, StageModelUsage, StageOutcome, StageState, SubAgentStatus,
+        SuccessReason, WorkflowSettings, first_event_seq, fixtures, test_support,
     };
     use serde_json::json;
 
@@ -4914,6 +4941,86 @@ mod tests {
             StageId::new("code", 1)
         }
 
+        #[test]
+        fn interrupt_settlement_and_steering_update_agent_control_projection() {
+            let mut state = initialized_projection();
+            let stage_id = stage_id();
+
+            state
+                .apply_event(&test_stage_event(
+                    1,
+                    EventBody::AgentRoundInterrupted(AgentRoundInterruptedProps {
+                        generation: 1,
+                        visit:      1,
+                    }),
+                    stage_id.clone(),
+                ))
+                .unwrap();
+            assert_eq!(
+                state.stage(&stage_id).unwrap().agent_control,
+                AgentControlState::WaitingForSteer
+            );
+
+            state
+                .apply_event(&test_stage_event(
+                    2,
+                    EventBody::AgentSteeringInjected(AgentSteeringInjectedProps {
+                        text:  "continue".to_string(),
+                        visit: 1,
+                    }),
+                    stage_id.clone(),
+                ))
+                .unwrap();
+            assert_eq!(
+                state.stage(&stage_id).unwrap().agent_control,
+                AgentControlState::Running
+            );
+
+            state
+                .apply_event(&test_stage_event(
+                    3,
+                    EventBody::AgentRoundInterrupted(AgentRoundInterruptedProps {
+                        generation: 2,
+                        visit:      1,
+                    }),
+                    stage_id.clone(),
+                ))
+                .unwrap();
+            state
+                .apply_event(&test_stage_event(
+                    4,
+                    EventBody::AgentSessionDeactivated(AgentSessionDeactivatedProps { visit: 1 }),
+                    stage_id.clone(),
+                ))
+                .unwrap();
+            assert_eq!(
+                state.stage(&stage_id).unwrap().agent_control,
+                AgentControlState::Running
+            );
+
+            state
+                .apply_event(&test_stage_event(
+                    5,
+                    EventBody::AgentRoundInterrupted(AgentRoundInterruptedProps {
+                        generation: 3,
+                        visit:      1,
+                    }),
+                    stage_id.clone(),
+                ))
+                .unwrap();
+            state
+                .apply_event(&test_stage_event(
+                    6,
+                    EventBody::StageFailed(failed_props(10, false)),
+                    stage_id.clone(),
+                ))
+                .unwrap();
+            assert_eq!(
+                state.stage(&stage_id).unwrap().agent_control,
+                AgentControlState::Running
+            );
+        }
+
         #[test]
         fn subagent_events_update_stage_projection() {
             let mut state = initialized_projection();

lib/components/fabro-workflow/src/event/convert.rs

@@ -680,6 +680,12 @@ fn event_body_from_event(event: &Event) -> EventBody {
                     visit: *visit,
                 })
             }
+            AgentEvent::RoundInterrupted { generation } => {
+                EventBody::AgentRoundInterrupted(fabro_types::AgentRoundInterruptedProps {
+                    generation: *generation,
+                    visit:      *visit,
+                })
+            }
             AgentEvent::CompactionStarted {
                 estimated_tokens,
                 context_window_size,
@@ -1801,6 +1807,30 @@ mod tests {
         }
     }
 
+    #[test]
+    fn agent_round_interrupted_populates_stage_session_and_generation() {
+        let stored = to_run_event(&fixtures::RUN_1, &Event::Agent {
+            stage:             "code".to_string(),
+            visit:             3,
+            event:             AgentEvent::RoundInterrupted { generation: 2 },
+            session_id:        Some("ses_1".to_string()),
+            parent_session_id: None,
+            tool_call_id:      None,
+        });
+
+        assert_eq!(stored.event_name(), "agent.round.interrupted");
+        assert_eq!(stored.node_id.as_deref(), Some("code"));
+        assert_eq!(stored.stage_id, Some(StageId::new("code", 3)));
+        assert_eq!(stored.session_id.as_deref(), Some("ses_1"));
+        match stored.body {
+            EventBody::AgentRoundInterrupted(props) => {
+                assert_eq!(props.generation, 2);
+                assert_eq!(props.visit, 3);
+            }
+            other => panic!("unexpected body: {other:?}"),
+        }
+    }
+
     #[test]
     fn stage_scope_populates_stage_id_on_non_stage_events() {
         // Events tied to a concrete stage execution but lacking scope in their

lib/components/fabro-workflow/src/event/names.rs

@@ -82,6 +82,7 @@ pub fn event_name(event: &Event) -> &'static str {
             AgentEvent::Warning { .. } => "agent.warning",
             AgentEvent::LoopDetected => "agent.loop.detected",
             AgentEvent::SteeringInjected { .. } => "agent.steering.injected",
+            AgentEvent::RoundInterrupted { .. } => "agent.round.interrupted",
             AgentEvent::CompactionStarted { .. } => "agent.compaction.started",
             AgentEvent::CompactionCompleted { .. } => "agent.compaction.completed",
             AgentEvent::LlmRetry { .. } => "agent.llm.retry",
@@ -191,6 +192,17 @@ mod tests {
             }),
             "agent.sub.spawned"
         );
+        assert_eq!(
+            event_name(&Event::Agent {
+                stage:             "code".to_string(),
+                visit:             1,
+                event:             AgentEvent::RoundInterrupted { generation: 1 },
+                session_id:        Some("session-1".to_string()),
+                parent_session_id: None,
+                tool_call_id:      None,
+            }),
+            "agent.round.interrupted"
+        );
         assert_eq!(
             event_name(&Event::AgentToolsAvailable {
                 node_id:    "code".to_string(),

lib/components/fabro-workflow/src/handler/llm/api.rs

@@ -3,12 +3,13 @@ use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
 use async_trait::async_trait;
-use fabro_agent::subagent::{SessionFactory, SubAgentManager};
+use fabro_agent::subagent::{SessionFactory, SubAgentSupervisor};
 use fabro_agent::tool_registry::{RegisteredTool, ToolContext, ToolRegistry, ToolSource};
 use fabro_agent::{
     AgentEvent, AgentProfile, AnthropicProfile, CompletionCoordinator, GeminiProfile,
-    Message as AgentMessage, OpenAiProfile, Sandbox, Session, SessionOptions, StaticEnvProvider,
-    ToolEnvProvider, ToolSecrets, register_question_tools,
+    Message as AgentMessage, OpenAiProfile, Sandbox, Session, SessionOptions,
+    SessionShutdownReason, StaticEnvProvider, ToolEnvProvider, ToolSecrets,
+    register_question_tools,
 };
 use fabro_auth::{CredentialSource, EnvCredentialSource};
 use fabro_graphviz::graph::{AttrValue, Node};
@@ -24,7 +25,7 @@ use fabro_model::{AgentProfileKind, Catalog, FallbackTarget, ModelRef, ProviderI
 use fabro_types::settings::run::RunModelControls;
 use fabro_types::{PermissionLevel, RunId, SessionCapability, StageId, StageTiming};
 use serde::de::DeserializeOwned;
-use tokio::sync::{Mutex as TokioMutex, mpsc};
+use tokio::sync::mpsc;
 use tokio::task::JoinHandle;
 use tokio_util::sync::CancellationToken;
 
@@ -155,21 +156,30 @@ fn begin_session_lifecycle(
     });
 }
 
-fn discard_session(
+async fn discard_session(
     session: &mut Session,
     lease: &mut Option<Arc<ActivationLease>>,
+    event_forwarder: &mut EventForwarder,
     emitter: &Arc<Emitter>,
 ) {
     if let Some(lease) = lease.take() {
         lease.release();
     }
     let session_id = session.id().to_string();
-    if session.close() {
-        emitter.emit(&Event::AgentSessionEnded {
-            session_id,
-            parent_session_id: None,
-        });
-    }
+    let reason = if session.cancel_token().is_cancelled() {
+        SessionShutdownReason::Cancelled
+    } else {
+        SessionShutdownReason::Error
+    };
+    session.shutdown(reason).await;
+    event_forwarder.wait_for_session_end().await;
+    // The agent-layer SessionEnded event is deliberately filtered by the
+    // bridge. This workflow-level event owns the durable session lifecycle,
+    // even when process_input already performed internal shutdown.
+    emitter.emit(&Event::AgentSessionEnded {
+        session_id,
+        parent_session_id: None,
+    });
 }
 
 fn build_profile(
@@ -539,6 +549,7 @@ fn emit_agent_tools_available(
 /// overtaking queued agent events.
 struct EventForwarder {
     processing_end_rx: mpsc::UnboundedReceiver<()>,
+    session_end_rx:    mpsc::UnboundedReceiver<()>,
     task:              JoinHandle<()>,
 }
 
@@ -549,6 +560,12 @@ impl EventForwarder {
         }
     }
 
+    async fn wait_for_session_end(&mut self) {
+        if self.session_end_rx.recv().await.is_none() {
+            tracing::warn!("Agent event forwarder stopped before session shutdown events");
+        }
+    }
+
     fn abort(&self) {
         self.task.abort();
     }
@@ -570,11 +587,15 @@ fn spawn_event_forwarder(
     let mut rx = session.subscribe();
     let root_session_id = session.id().to_string();
     let (processing_end_tx, processing_end_rx) = mpsc::unbounded_channel();
+    let (session_end_tx, session_end_rx) = mpsc::unbounded_channel();
     let task = tokio::spawn(async move {
         while let Ok(event) = rx.recv().await {
             let is_root_processing_end = event.session_id == root_session_id
                 && event.parent_session_id.is_none()
                 && matches!(&event.event, AgentEvent::ProcessingEnd);
+            let is_root_session_end = event.session_id == root_session_id
+                && event.parent_session_id.is_none()
+                && matches!(&event.event, AgentEvent::SessionEnded);
 
             // Reset watchdog on every event, including streaming deltas
             emitter.touch();
@@ -611,11 +632,15 @@ fn spawn_event_forwarder(
             if is_root_processing_end {
                 let _ = processing_end_tx.send(());
             }
+            if is_root_session_end {
+                let _ = session_end_tx.send(());
+            }
         }
     });
 
     EventForwarder {
         processing_end_rx,
+        session_end_rx,
         task,
     }
 }
@@ -833,10 +858,8 @@ impl AgentApiBackend {
             ..SessionOptions::default()
         };
 
-        let manager = Arc::new(TokioMutex::new(SubAgentManager::new(
-            config.max_subagent_depth,
-        )));
-        let manager_for_callback = manager.clone();
+        let supervisor = SubAgentSupervisor::new(config.max_subagent_depth);
+        let supervisor_for_session = supervisor.clone();
 
         // Build factory that creates child sessions WITHOUT subagent tools
         let factory_client = client.clone();
@@ -878,7 +901,7 @@ impl AgentApiBackend {
             session
         });
 
-        profile.register_subagent_tools(manager, factory, 0);
+        profile.register_subagent_tools(supervisor.clone(), factory, 0);
         register_question_tools(provider.profile_kind, profile.tool_registry_mut());
         if let Some(services) = fabro_run_tools {
             register_fabro_run_tools(profile.tool_registry_mut(), &services);
@@ -890,17 +913,14 @@ impl AgentApiBackend {
             profile,
             Arc::clone(sandbox),
             config,
-            Some(manager_for_callback.clone()),
+            Some(supervisor_for_session),
         );
         if let Some(provider) = tool_env {
             session.set_tool_env_provider(Arc::clone(provider));
         }
 
         // Wire subagent event callback to parent session's emitter
-        manager_for_callback
-            .lock()
-            .await
-            .set_event_callback(session.sub_agent_event_callback());
+        supervisor.set_event_callback(session.sub_agent_event_callback());
 
         Ok(session)
     }
@@ -938,7 +958,7 @@ impl AgentApiBackend {
         Ok(lease)
     }
 
-    fn shutdown_cached_sessions(&self, emitter: &Arc<Emitter>) {
+    async fn shutdown_cached_sessions(&self, emitter: &Arc<Emitter>) {
         let sessions: Vec<Session> = self
             .sessions
             .lock()
@@ -948,7 +968,7 @@ impl AgentApiBackend {
             .collect();
         for mut session in sessions {
             let session_id = session.id().to_string();
-            if session.close() {
+            if session.shutdown(SessionShutdownReason::Completed).await {
                 emitter.emit(&Event::AgentSessionEnded {
                     session_id,
                     parent_session_id: None,
@@ -1054,7 +1074,7 @@ impl AgentApiBackend {
 #[async_trait]
 impl CodergenBackend for AgentApiBackend {
     async fn shutdown(&self, emitter: &Arc<Emitter>) {
-        self.shutdown_cached_sessions(emitter);
+        self.shutdown_cached_sessions(emitter).await;
     }
 
     fn effective_request_controls(&self, node: &Node) -> Result<EffectiveRequestControls, Error> {
@@ -1286,12 +1306,14 @@ impl CodergenBackend for AgentApiBackend {
                 Err(err) => match classify_agent_error(err, allow_failover_primary) {
                     AgentApiErrorDisposition::Cancelled => {
                         bridge.abort();
-                        discard_session(&mut session, &mut lease, emitter);
+                        discard_session(&mut session, &mut lease, &mut event_forwarder, emitter)
+                            .await;
                         return Err(Error::Cancelled);
                     }
                     AgentApiErrorDisposition::Terminal(err) => {
                         bridge.abort();
-                        discard_session(&mut session, &mut lease, emitter);
+                        discard_session(&mut session, &mut lease, &mut event_forwarder, emitter)
+                            .await;
                         return Err(err);
                     }
                     AgentApiErrorDisposition::FailoverEligible(sdk_err) => {
@@ -1309,7 +1331,8 @@ impl CodergenBackend for AgentApiBackend {
                     Ok(active_lease) => lease = Some(active_lease),
                     Err(err) => {
                         bridge.abort();
-                        discard_session(&mut session, &mut lease, emitter);
+                        discard_session(&mut session, &mut lease, &mut event_forwarder, emitter)
+                            .await;
                         return Err(err);
                     }
                 }
@@ -1342,12 +1365,12 @@ impl CodergenBackend for AgentApiBackend {
             Err(err) => match classify_agent_error(err, allow_failover_primary) {
                 AgentApiErrorDisposition::Cancelled => {
                     bridge.abort();
-                    discard_session(&mut session, &mut lease, emitter);
+                    discard_session(&mut session, &mut lease, &mut event_forwarder, emitter).await;
                     return Err(Error::Cancelled);
                 }
                 AgentApiErrorDisposition::Terminal(err) => {
                     bridge.abort();
-                    discard_session(&mut session, &mut lease, emitter);
+                    discard_session(&mut session, &mut lease, &mut event_forwarder, emitter).await;
                     return Err(err);
                 }
                 AgentApiErrorDisposition::FailoverEligible(sdk_err) => {
@@ -1359,8 +1382,8 @@ impl CodergenBackend for AgentApiBackend {
                     let mut succeeded = false;
 
                     bridge.abort();
+                    discard_session(&mut session, &mut lease, &mut event_forwarder, emitter).await;
                     event_forwarder.abort();
-                    discard_session(&mut session, &mut lease, emitter);
 
                     for (index, target) in self.fallback_chain.iter().enumerate() {
                         emitter.emit_scoped(
@@ -1427,18 +1450,36 @@ impl CodergenBackend for AgentApiBackend {
                             match classify_agent_error(err, allow_failover_next) {
                                 AgentApiErrorDisposition::Cancelled => {
                                     bridge.abort();
-                                    discard_session(&mut session, &mut lease, emitter);
+                                    discard_session(
+                                        &mut session,
+                                        &mut lease,
+                                        &mut event_forwarder,
+                                        emitter,
+                                    )
+                                    .await;
                                     return Err(Error::Cancelled);
                                 }
                                 AgentApiErrorDisposition::Terminal(err) => {
                                     bridge.abort();
-                                    discard_session(&mut session, &mut lease, emitter);
+                                    discard_session(
+                                        &mut session,
+                                        &mut lease,
+                                        &mut event_forwarder,
+                                        emitter,
+                                    )
+                                    .await;
                                     return Err(err);
                                 }
                                 AgentApiErrorDisposition::FailoverEligible(sdk_err) => {
                                     last_err = Error::Llm(sdk_err);
                                     bridge.abort();
-                                    discard_session(&mut session, &mut lease, emitter);
+                                    discard_session(
+                                        &mut session,
+                                        &mut lease,
+                                        &mut event_forwarder,
+                                        emitter,
+                                    )
+                                    .await;
                                     continue;
                                 }
                             }
@@ -1452,7 +1493,13 @@ impl CodergenBackend for AgentApiBackend {
                             Ok(active_lease) => lease = Some(active_lease),
                             Err(err) => {
                                 bridge.abort();
-                                discard_session(&mut session, &mut lease, emitter);
+                                discard_session(
+                                    &mut session,
+                                    &mut lease,
+                                    &mut event_forwarder,
+                                    emitter,
+                                )
+                                .await;
                                 return Err(err);
                             }
                         }
@@ -1474,18 +1521,36 @@ impl CodergenBackend for AgentApiBackend {
                             Err(err) => match classify_agent_error(err, allow_failover_next) {
                                 AgentApiErrorDisposition::Cancelled => {
                                     bridge.abort();
-                                    discard_session(&mut session, &mut lease, emitter);
+                                    discard_session(
+                                        &mut session,
+                                        &mut lease,
+                                        &mut event_forwarder,
+                                        emitter,
+                                    )
+                                    .await;
                                     return Err(Error::Cancelled);
                                 }
                                 AgentApiErrorDisposition::Terminal(err) => {
                                     bridge.abort();
-                                    discard_session(&mut session, &mut lease, emitter);
+                                    discard_session(
+                                        &mut session,
+                                        &mut lease,
+                                        &mut event_forwarder,
+                                        emitter,
+                                    )
+                                    .await;
                                     return Err(err);
                                 }
                                 AgentApiErrorDisposition::FailoverEligible(sdk_err) => {
                                     last_err = Error::Llm(sdk_err);
                                     bridge.abort();
-                                    discard_session(&mut session, &mut lease, emitter);
+                                    discard_session(
+                                        &mut session,
+                                        &mut lease,
+                                        &mut event_forwarder,
+                                        emitter,
+                                    )
+                                    .await;
                                 }
                             },
                         }
@@ -1500,7 +1565,7 @@ impl CodergenBackend for AgentApiBackend {
         // bridge's `Drop` will abort the spawned task on early return.
         if let Err(err) = result {
             bridge.abort();
-            discard_session(&mut session, &mut lease, emitter);
+            discard_session(&mut session, &mut lease, &mut event_forwarder, emitter).await;
             return Err(err);
         }
 
@@ -1521,7 +1586,13 @@ impl CodergenBackend for AgentApiBackend {
                     Err(error) => {
                         if repair_attempts >= node.output_retries() {
                             bridge.abort();
-                            discard_session(&mut session, &mut lease, emitter);
+                            discard_session(
+                                &mut session,
+                                &mut lease,
+                                &mut event_forwarder,
+                                emitter,
+                            )
+                            .await;
                             return Err(Error::OutputSchemaValidation(
                                 structured_output::exhausted_failure_reason(node.output_retries()),
                             ));
@@ -1547,17 +1618,35 @@ impl CodergenBackend for AgentApiBackend {
                             Err(err) => match classify_agent_error(err, false) {
                                 AgentApiErrorDisposition::Cancelled => {
                                     bridge.abort();
-                                    discard_session(&mut session, &mut lease, emitter);
+                                    discard_session(
+                                        &mut session,
+                                        &mut lease,
+                                        &mut event_forwarder,
+                                        emitter,
+                                    )
+                                    .await;
                                     return Err(Error::Cancelled);
                                 }
                                 AgentApiErrorDisposition::Terminal(err) => {
                                     bridge.abort();
-                                    discard_session(&mut session, &mut lease, emitter);
+                                    discard_session(
+                                        &mut session,
+                                        &mut lease,
+                                        &mut event_forwarder,
+                                        emitter,
+                                    )
+                                    .await;
                                     return Err(err);
                                 }
                                 AgentApiErrorDisposition::FailoverEligible(sdk_err) => {
                                     bridge.abort();
-                                    discard_session(&mut session, &mut lease, emitter);
+                                    discard_session(
+                                        &mut session,
+                                        &mut lease,
+                                        &mut event_forwarder,
+                                        emitter,
+                                    )
+                                    .await;
                                     return Err(Error::Llm(sdk_err));
                                 }
                             },
@@ -1579,10 +1668,6 @@ impl CodergenBackend for AgentApiBackend {
         )?
         .with_reported_cost(total_cost);
 
-        // Collect files_touched from the shared tracking state.
-        let (files_touched, last_file_touched) = file_tracking_snapshot(&file_tracking);
-        drop(event_forwarder);
-
         if let Some(lease) = lease.take() {
             lease.release();
         }
@@ -1591,20 +1676,26 @@ impl CodergenBackend for AgentApiBackend {
         // the cached session is not left wired to this run's cancel token.
         if let Some(key) = reuse_key {
             bridge.abort();
+            drop(event_forwarder);
             self.sessions
                 .lock()
                 .expect("sessions mutex is never poisoned: no code panics while holding this lock")
                 .insert(key, session);
         } else {
+            bridge.abort();
             let session_id = session.id().to_string();
-            if session.close() {
-                emitter.emit(&Event::AgentSessionEnded {
-                    session_id,
-                    parent_session_id: None,
-                });
-            }
+            session.shutdown(SessionShutdownReason::Completed).await;
+            event_forwarder.wait_for_session_end().await;
+            emitter.emit(&Event::AgentSessionEnded {
+                session_id,
+                parent_session_id: None,
+            });
+            drop(event_forwarder);
         }
 
+        // Snapshot after non-cached shutdown so final child events are included.
+        let (files_touched, last_file_touched) = file_tracking_snapshot(&file_tracking);
+
         Ok(CodergenResult::Text {
             text: response,
             usage: Some(stage_usage),
@@ -2566,11 +2657,11 @@ reasoning = false
             AgentProfileKind::Anthropic,
             Arc::new(Catalog::from_builtin().unwrap()),
         );
-        let manager = Arc::new(TokioMutex::new(SubAgentManager::new(1)));
+        let supervisor = SubAgentSupervisor::new(1);
         let factory: SessionFactory = Arc::new(|| {
             panic!("factory should not be called in this test");
         });
-        profile.register_subagent_tools(manager, factory, 0);
+        profile.register_subagent_tools(supervisor, factory, 0);
 
         let names = profile.tool_registry().names();
         assert!(names.contains(&"spawn_agent".to_string()));
@@ -3077,6 +3168,67 @@ enabled = true
         assert!(backend.sessions.lock().unwrap().is_empty());
     }
 
+    #[tokio::test]
+    async fn session_end_barrier_preserves_child_close_ordering() {
+        let mut providers = HashMap::new();
+        providers.insert(
+            "openai".to_string(),
+            Arc::new(ShutdownTestProvider) as Arc<dyn ProviderAdapter>,
+        );
+        let client = Client::new(providers, Some("openai".to_string()), Vec::new());
+        let mut session = Session::new(
+            client,
+            Arc::new(ShutdownTestProfile::new()),
+            Arc::new(LocalSandbox::new(
+                tempfile::tempdir().unwrap().path().to_path_buf(),
+            )),
+            SessionOptions::default(),
+            None,
+        );
+        let emitter = Arc::new(Emitter::new(RunId::new()));
+        let event_names = Arc::new(Mutex::new(Vec::new()));
+        let event_names_for_listener = Arc::clone(&event_names);
+        emitter.on_event(move |event| {
+            event_names_for_listener
+                .lock()
+                .unwrap()
+                .push(event.event_name().to_string());
+        });
+        let context = Context::new();
+        let scope = StageScope::for_handler(&context, "code");
+        let file_tracking = Arc::new(Mutex::new(FileTracking {
+            pending: HashMap::new(),
+            touched: HashSet::new(),
+            last:    None,
+        }));
+        let mut forwarder = spawn_event_forwarder(
+            &session,
+            "code".to_string(),
+            scope,
+            Arc::clone(&emitter),
+            file_tracking,
+        );
+
+        session.sub_agent_event_callback()(
+            fabro_agent::subagent::SubAgentCallbackEvent::Lifecycle(AgentEvent::SubAgentClosed {
+                agent_id: "child-1".to_string(),
+                depth:    1,
+            }),
+        );
+        let session_id = session.id().to_string();
+        session.shutdown(SessionShutdownReason::Completed).await;
+        forwarder.wait_for_session_end().await;
+        emitter.emit(&Event::AgentSessionEnded {
+            session_id,
+            parent_session_id: None,
+        });
+
+        assert_eq!(event_names.lock().unwrap().as_slice(), [
+            "agent.sub.closed",
+            "agent.session.ended"
+        ]);
+    }
+
     // --- Bridge guard tests ---
 
     fn failover_eligible_llm_error() -> LlmError {

lib/foundation/fabro-api/build.rs

@@ -346,6 +346,7 @@ fn main() {
         ("StageOutcome", "fabro_types::StageOutcome", &[]),
         ("StageHandler", "fabro_types::StageHandler", &[]),
         ("StageState", "fabro_types::StageState", &[]),
+        ("AgentControlState", "fabro_types::AgentControlState", &[]),
         ("CommandTermination", "fabro_types::CommandTermination", &[]),
         ("StageModelUsage", "fabro_types::StageModelUsage", &[]),
         ("StageProjection", "fabro_types::StageProjection", &[]),

lib/foundation/fabro-api/src/lib.rs

@@ -39,13 +39,14 @@ pub mod types {
         BlockedReason, FailureReason, PendingReason, RunControlAction, RunStatus, SuccessReason,
     };
     pub use fabro_types::{
-        ActivatedSkill, AgentMcpToolSummary, AgentSkillActivationSource, AgentSkillSummary,
-        AgentToolCategory, AgentToolSource, AgentToolSummary, AgentToolsAvailableProps, AskFabro,
-        AuthMethod, AutomationRef, BilledTokenCounts, CommandTermination, Conclusion, ContentPart,
-        CreateVariableRequest, DiffStats, DiffSummary, DirtyStatus, EventEnvelope, ExecOutputTail,
-        FailureCategory, FailureDetail, FailureSignature, GitContext, IdpIdentity,
-        IntegrationConnectionKind, IntegrationConnectionState, IntegrationConnectionStatus,
-        IntegrationProvider, IntegrationStatus, InterviewOption, InterviewQuestionRecord,
+        ActivatedSkill, AgentControlState, AgentMcpToolSummary, AgentSkillActivationSource,
+        AgentSkillSummary, AgentToolCategory, AgentToolSource, AgentToolSummary,
+        AgentToolsAvailableProps, AskFabro, AuthMethod, AutomationRef, BilledTokenCounts,
+        CommandTermination, Conclusion, ContentPart, CreateVariableRequest, DiffStats, DiffSummary,
+        DirtyStatus, EventEnvelope, ExecOutputTail, FailureCategory, FailureDetail,
+        FailureSignature, GitContext, IdpIdentity, IntegrationConnectionKind,
+        IntegrationConnectionState, IntegrationConnectionStatus, IntegrationProvider,
+        IntegrationStatus, InterviewOption, InterviewQuestionRecord,
         McpServerDraft as CreateMcpServerRequest, McpServerProjection,
         McpServerReplace as ReplaceMcpServerRequest, McpServerStatus, McpServerView as McpServer,
         McpTransportView, Message, PairId, PairMessageId, PairMessageRecord, PairMessageRequest,

lib/foundation/fabro-api/tests/run_projection_round_trip.rs

@@ -94,6 +94,7 @@ fn run_projection_round_trips_populated_projection() {
                     "cache_read_tokens": 0,
                     "cache_write_tokens": 0
                 },
+                "agent_control": "running",
                 "state": "running"
             }
         }

lib/foundation/fabro-api/tests/stage_projection_round_trip.rs

@@ -1,7 +1,8 @@
 use std::any::{TypeId, type_name};
 
 use fabro_api::types::{
-    ActivatedSkill as ApiActivatedSkill, AgentMcpToolSummary as ApiAgentMcpToolSummary,
+    ActivatedSkill as ApiActivatedSkill, AgentControlState as ApiAgentControlState,
+    AgentMcpToolSummary as ApiAgentMcpToolSummary,
     AgentSkillActivationSource as ApiAgentSkillActivationSource,
     AgentSkillSummary as ApiAgentSkillSummary, AgentToolCategory as ApiAgentToolCategory,
     AgentToolSource as ApiAgentToolSource, AgentToolSummary as ApiAgentToolSummary,
@@ -20,13 +21,13 @@ use fabro_api::types::{
     SubAgentStatus as ApiSubAgentStatus, TodoListProjection as ApiTodoListProjection,
 };
 use fabro_types::{
-    ActivatedSkill, AgentMcpToolSummary, AgentSkillActivationSource, AgentSkillSummary,
-    AgentToolCategory, AgentToolSource, AgentToolSummary, AgentToolsAvailableProps,
-    McpServerProjection, McpServerStatus, PermissionLevel, SkillsProjection, StageContextWindow,
-    StageContextWindowBreakdownItem, StageContextWindowCategory, StageContextWindowCountMethod,
-    StageContextWindowProjection, StageContextWindowStaleness, StageContextWindowUnavailableReason,
-    StageContextWindowWarning, StageProjection, SubAgentProjection, SubAgentStatus, TodoListKind,
-    TodoListProjection,
+    ActivatedSkill, AgentControlState, AgentMcpToolSummary, AgentSkillActivationSource,
+    AgentSkillSummary, AgentToolCategory, AgentToolSource, AgentToolSummary,
+    AgentToolsAvailableProps, McpServerProjection, McpServerStatus, PermissionLevel,
+    SkillsProjection, StageContextWindow, StageContextWindowBreakdownItem,
+    StageContextWindowCategory, StageContextWindowCountMethod, StageContextWindowProjection,
+    StageContextWindowStaleness, StageContextWindowUnavailableReason, StageContextWindowWarning,
+    StageProjection, SubAgentProjection, SubAgentStatus, TodoListKind, TodoListProjection,
 };
 use serde_json::json;
 
@@ -37,6 +38,7 @@ fn stage_projection_reuses_canonical_type() {
 
 #[test]
 fn stage_projection_reuses_nested_agent_state_types() {
+    assert_same_type::<ApiAgentControlState, AgentControlState>();
     assert_same_type::<ApiTodoListProjection, TodoListProjection>();
     assert_same_type::<ApiSubAgentProjection, SubAgentProjection>();
     assert_same_type::<ApiSubAgentStatus, SubAgentStatus>();
@@ -198,6 +200,7 @@ fn stage_projection_round_trips_representative_json() {
             ],
             "warnings": []
         },
+        "agent_control": "running",
         "state": "succeeded"
     });
 

lib/foundation/fabro-types/src/lib.rs

@@ -115,11 +115,12 @@ pub use run_event::{
 pub use run_failure::RunFailure;
 pub use run_id::{RunId, fixtures};
 pub use run_projection::{
-    ActivatedSkill, CheckpointRecord, McpServerProjection, McpServerStatus, PendingInterviewRecord,
-    RunProjection, SkillsProjection, StageContextWindow, StageContextWindowBreakdownItem,
-    StageContextWindowCategory, StageContextWindowCountMethod, StageContextWindowProjection,
-    StageContextWindowStaleness, StageContextWindowUnavailableReason, StageContextWindowWarning,
-    StageModelUsage, StageProjection, SubAgentProjection, SubAgentStatus, first_event_seq,
+    ActivatedSkill, AgentControlState, CheckpointRecord, McpServerProjection, McpServerStatus,
+    PendingInterviewRecord, RunProjection, SkillsProjection, StageContextWindow,
+    StageContextWindowBreakdownItem, StageContextWindowCategory, StageContextWindowCountMethod,
+    StageContextWindowProjection, StageContextWindowStaleness, StageContextWindowUnavailableReason,
+    StageContextWindowWarning, StageModelUsage, StageProjection, SubAgentProjection,
+    SubAgentStatus, first_event_seq,
 };
 pub use run_sandbox::{
     RunSandbox, RunSandboxFailure, RunSandboxInstance, RunSandboxKind, RunSandboxPlan,

lib/foundation/fabro-types/src/run_event/agent.rs

@@ -201,6 +201,12 @@ pub struct AgentSteeringInjectedProps {
     pub visit: u32,
 }
 
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct AgentRoundInterruptedProps {
+    pub generation: u64,
+    pub visit:      u32,
+}
+
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct AgentPairUserMessageProps {
     pub pair_id:           PairId,

lib/foundation/fabro-types/src/run_event/mod.rs

@@ -222,6 +222,8 @@ pub enum EventBody {
     AgentLoopDetected(AgentLoopDetectedProps),
     #[serde(rename = "agent.steering.injected")]
     AgentSteeringInjected(AgentSteeringInjectedProps),
+    #[serde(rename = "agent.round.interrupted")]
+    AgentRoundInterrupted(AgentRoundInterruptedProps),
     #[serde(rename = "agent.pair.user_message")]
     AgentPairUserMessage(AgentPairUserMessageProps),
     #[serde(rename = "agent.pair.system_message")]
@@ -498,6 +500,7 @@ impl EventBody {
             Self::AgentWarning(_) => "agent.warning",
             Self::AgentLoopDetected(_) => "agent.loop.detected",
             Self::AgentSteeringInjected(_) => "agent.steering.injected",
+            Self::AgentRoundInterrupted(_) => "agent.round.interrupted",
             Self::AgentPairUserMessage(_) => "agent.pair.user_message",
             Self::AgentPairSystemMessage(_) => "agent.pair.system_message",
             Self::AgentInterruptInjected(_) => "agent.interrupt.injected",
@@ -669,6 +672,7 @@ fn is_known_event_name(event: &str) -> bool {
             | "agent.warning"
             | "agent.loop.detected"
             | "agent.steering.injected"
+            | "agent.round.interrupted"
             | "agent.pair.user_message"
             | "agent.pair.system_message"
             | "agent.interrupt.injected"
@@ -1183,6 +1187,29 @@ mod tests {
         assert_eq!(parsed.to_value().unwrap(), line);
     }
 
+    #[test]
+    fn agent_round_interrupted_round_trips_with_generation_and_stage() {
+        let line = json!({
+            "id": "evt_round_interrupted",
+            "ts": "2026-04-04T12:00:00Z",
+            "run_id": fixtures::RUN_1,
+            "event": "agent.round.interrupted",
+            "node_id": "code",
+            "node_label": "code",
+            "stage_id": "code@2",
+            "session_id": "ses_1",
+            "properties": { "generation": 3, "visit": 2 }
+        });
+
+        let parsed = RunEvent::from_value(line.clone()).unwrap();
+        assert!(matches!(
+            &parsed.body,
+            EventBody::AgentRoundInterrupted(props)
+                if props.generation == 3 && props.visit == 2
+        ));
+        assert_eq!(parsed.to_value().unwrap(), line);
+    }
+
     #[test]
     fn run_interrupt_then_steer_is_not_a_known_persisted_event() {
         let line = json!({

lib/foundation/fabro-types/src/run_projection.rs

@@ -370,9 +370,32 @@ pub struct StageProjection {
     pub mcp_servers:       Vec<McpServerProjection>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub context_window:    Option<StageContextWindowProjection>,
+    #[serde(default)]
+    pub agent_control:     AgentControlState,
     pub state:             StageState,
 }
 
+#[derive(
+    Debug,
+    Clone,
+    Copy,
+    Default,
+    PartialEq,
+    Eq,
+    serde::Serialize,
+    serde::Deserialize,
+    Display,
+    EnumString,
+    IntoStaticStr,
+)]
+#[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
+pub enum AgentControlState {
+    #[default]
+    Running,
+    WaitingForSteer,
+}
+
 #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
 pub struct SubAgentProjection {
     pub agent_id: String,
@@ -451,6 +474,7 @@ impl StageProjection {
             agent_tools: Vec::new(),
             mcp_servers: Vec::new(),
             context_window: None,
+            agent_control: AgentControlState::default(),
             provider_used: None,
             diff: None,
             script_invocation: None,
@@ -752,7 +776,9 @@ mod iter_stages_tests {
     use serde_json::json;
 
     use super::RunProjection;
-    use crate::{Graph, RunId, RunSpec, StageProjection, WorkflowSettings, test_support};
+    use crate::{
+        AgentControlState, Graph, RunId, RunSpec, StageProjection, WorkflowSettings, test_support,
+    };
 
     fn seq(n: u32) -> NonZeroU32 {
         NonZeroU32::new(n).unwrap()
@@ -825,7 +851,7 @@ mod iter_stages_tests {
     }
 
     #[test]
-    fn stage_projection_defaults_missing_agent_tools_to_empty_and_omits_empty_list() {
+    fn stage_projection_defaults_missing_agent_fields() {
         let value = json!({
             "first_event_seq": 1,
             "state": "running"
@@ -833,6 +859,7 @@ mod iter_stages_tests {
 
         let stage: StageProjection = serde_json::from_value(value).unwrap();
         assert!(stage.agent_tools.is_empty());
+        assert_eq!(stage.agent_control, AgentControlState::Running);
 
         let serialized = serde_json::to_value(stage).unwrap();
         assert!(

lib/packages/fabro-api-client/src/.openapi-generator/FILES

@@ -28,6 +28,7 @@ common.ts
 configuration.ts
 index.ts
 models/activated-skill.ts
+models/agent-control-state.ts
 models/agent-mcp-tool-summary.ts
 models/agent-message-props.ts
 models/agent-permissions.ts

lib/packages/fabro-api-client/src/api/runs-api.ts

@@ -288,7 +288,7 @@ export const RunsApiAxiosParamCreator = function (configuration?: Configuration)
             };
         },
         /**
-         * Cancels a pending, runnable, or running run. Returns 409 if the run has already completed or been cancelled.
+         * Cancels a pending, runnable, or running run. Pre-execution runs are cancelled synchronously. Live runs return after the cancellation request is durably recorded and continue converging to a terminal cancelled state. Returns 409 if the run has already completed or been cancelled.
          * @summary Cancel Run
          * @param {string} id Unique run identifier (ULID).
          * @param {*} [options] Override http request option.
@@ -1607,7 +1607,7 @@ export const RunsApiFp = function(configuration?: Configuration) {
             return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
         },
         /**
-         * Cancels a pending, runnable, or running run. Returns 409 if the run has already completed or been cancelled.
+         * Cancels a pending, runnable, or running run. Pre-execution runs are cancelled synchronously. Live runs return after the cancellation request is durably recorded and continue converging to a terminal cancelled state. Returns 409 if the run has already completed or been cancelled.
          * @summary Cancel Run
          * @param {string} id Unique run identifier (ULID).
          * @param {*} [options] Override http request option.
@@ -2060,7 +2060,7 @@ export const RunsApiFactory = function (configuration?: Configuration, basePath?
             return localVarFp.batchUnarchiveRuns(batchRunLifecycleRequest, options).then((request) => request(axios, basePath));
         },
         /**
-         * Cancels a pending, runnable, or running run. Returns 409 if the run has already completed or been cancelled.
+         * Cancels a pending, runnable, or running run. Pre-execution runs are cancelled synchronously. Live runs return after the cancellation request is durably recorded and continue converging to a terminal cancelled state. Returns 409 if the run has already completed or been cancelled.
          * @summary Cancel Run
          * @param {string} id Unique run identifier (ULID).
          * @param {*} [options] Override http request option.
@@ -2429,7 +2429,7 @@ export class RunsApi extends BaseAPI {
     }
 
     /**
-     * Cancels a pending, runnable, or running run. Returns 409 if the run has already completed or been cancelled.
+     * Cancels a pending, runnable, or running run. Pre-execution runs are cancelled synchronously. Live runs return after the cancellation request is durably recorded and continue converging to a terminal cancelled state. Returns 409 if the run has already completed or been cancelled.
      * @summary Cancel Run
      * @param {string} id Unique run identifier (ULID).
      * @param {*} [options] Override http request option.

lib/packages/fabro-api-client/src/models/agent-control-state.ts

@@ -0,0 +1,26 @@
+/* tslint:disable */
+/* eslint-disable */
+/**
+ * Fabro Run API
+ * HTTP API for managing Fabro workflow run executions.
+ *
+ * The version of the OpenAPI document: 0.1.0
+ *
+ *
+ * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
+ * https://openapi-generator.tech
+ * Do not edit the class manually.
+ */
+
+
+
+/**
+ * Control state of a live agent stage.
+ */
+
+export const AgentControlState = {
+    RUNNING: 'running',
+    WAITING_FOR_STEER: 'waiting_for_steer'
+} as const;
+
+export type AgentControlState = typeof AgentControlState[keyof typeof AgentControlState];

lib/packages/fabro-api-client/src/models/index.ts

@@ -1,4 +1,5 @@
 export * from './activated-skill';
+export * from './agent-control-state';
 export * from './agent-mcp-tool-summary';
 export * from './agent-message-props';
 export * from './agent-permissions';

lib/packages/fabro-api-client/src/models/stage-projection.ts

@@ -13,6 +13,9 @@
  */
 
 
+// May contain unused imports in some cases
+// @ts-ignore
+import type { AgentControlState } from './agent-control-state';
 // May contain unused imports in some cases
 // @ts-ignore
 import type { AgentToolSummary } from './agent-tool-summary';
@@ -108,6 +111,10 @@ export interface StageProjection {
      */
     'mcp_servers'?: Array<McpServerProjection>;
     'context_window'?: StageContextWindowProjection | null;
+    /**
+     * Whether the agent is executing normally or waiting for steering after an interrupt.
+     */
+    'agent_control': AgentControlState;
     /**
      * Lifecycle state of the stage projection.
      */