feat(FN-5403): merge fusion/fn-5403

packages/engine/src/__tests__/executor-abort-all-in-flight.test.ts

@@ -0,0 +1,123 @@
+import "./executor-test-helpers.js";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import { TaskExecutor } from "../executor.js";
+import { executorLog } from "../logger.js";
+import { resetExecutorMocks } from "./executor-test-helpers.js";
+
+type Listener = (...args: any[]) => void;
+
+function createEventedStore() {
+  const listeners = new Map<string, Set<Listener>>();
+  return {
+    store: {
+      on: vi.fn((event: string, listener: Listener) => {
+        const set = listeners.get(event) ?? new Set<Listener>();
+        set.add(listener);
+        listeners.set(event, set);
+      }),
+      off: vi.fn((event: string, listener: Listener) => {
+        listeners.get(event)?.delete(listener);
+      }),
+      getSettings: vi.fn().mockResolvedValue({ globalPause: false, enginePaused: false }),
+      listTasks: vi.fn().mockResolvedValue([]),
+    } as any,
+  };
+}
+
+describe("TaskExecutor.abortAllInFlight", () => {
+  beforeEach(() => {
+    resetExecutorMocks();
+    vi.clearAllMocks();
+  });
+
+  it("aborts and disposes all active surfaces and logs a summary", async () => {
+    const { store } = createEventedStore();
+    const executor = new TaskExecutor(store, "/tmp/test");
+    const logSpy = vi.spyOn(executorLog, "log");
+
+    const taskAbort = vi.fn().mockResolvedValue(undefined);
+    const taskDispose = vi.fn();
+    (executor as any).activeSessions.set("FN-1", {
+      session: { abort: taskAbort, dispose: taskDispose },
+      seenSteeringIds: new Set<string>(),
+    });
+    (executor as any).activeSessions.set("FN-2", {
+      session: { abort: vi.fn().mockResolvedValue(undefined), dispose: vi.fn() },
+      seenSteeringIds: new Set<string>(),
+    });
+    (executor as any).activeStepExecutors.set("FN-1", {
+      abortAllSessionBash: vi.fn(),
+      terminateAllSessions: vi.fn().mockResolvedValue(undefined),
+    });
+    (executor as any).activeWorkflowStepSessions.set("FN-3", {
+      abort: vi.fn().mockResolvedValue(undefined),
+      dispose: vi.fn(),
+    });
+    (executor as any).activeSubagentSessions.set("FN-4", new Set([{ dispose: vi.fn() }]));
+
+    const childAbort = vi.fn().mockResolvedValue(undefined);
+    const childDispose = vi.fn();
+    (executor as any).childSessions.set("agent-1", { abort: childAbort, dispose: childDispose });
+
+    await executor.abortAllInFlight("engine stop");
+
+    expect(taskAbort).toHaveBeenCalledTimes(1);
+    expect(taskDispose).toHaveBeenCalledTimes(1);
+    expect(childAbort).toHaveBeenCalledTimes(1);
+    expect(childDispose).toHaveBeenCalledTimes(1);
+    expect((executor as any).activeSessions.size).toBe(0);
+    expect((executor as any).activeStepExecutors.size).toBe(0);
+    expect((executor as any).activeWorkflowStepSessions.size).toBe(0);
+    expect((executor as any).activeSubagentSessions.size).toBe(0);
+    expect((executor as any).childSessions.size).toBe(0);
+    expect(logSpy).toHaveBeenCalledWith("abortAllInFlight: aborted 4 task surface(s) — engine stop");
+  });
+
+  it("continues when one surface abort path rejects", async () => {
+    const { store } = createEventedStore();
+    const executor = new TaskExecutor(store, "/tmp/test");
+    const warnSpy = vi.spyOn(executorLog, "warn").mockImplementation(() => undefined as any);
+
+    (executor as any).activeSessions.set("FN-ERR", {
+      session: {
+        abort: vi.fn().mockRejectedValue(new Error("boom")),
+        dispose: vi.fn(),
+      },
+      seenSteeringIds: new Set<string>(),
+    });
+
+    const healthyAbort = vi.fn().mockResolvedValue(undefined);
+    (executor as any).activeSessions.set("FN-OK", {
+      session: { abort: healthyAbort, dispose: vi.fn() },
+      seenSteeringIds: new Set<string>(),
+    });
+
+    await expect(executor.abortAllInFlight("engine stop")).resolves.toBeUndefined();
+    expect(healthyAbort).toHaveBeenCalledTimes(1);
+    expect(warnSpy).toHaveBeenCalled();
+  });
+
+  it("is a no-op when there are no active surfaces", async () => {
+    const { store } = createEventedStore();
+    const executor = new TaskExecutor(store, "/tmp/test");
+    const logSpy = vi.spyOn(executorLog, "log");
+
+    await expect(executor.abortAllInFlight("engine stop")).resolves.toBeUndefined();
+
+    expect(logSpy).toHaveBeenCalledWith("abortAllInFlight: aborted 0 task surface(s) — engine stop");
+  });
+
+  it("propagates reason into per-task abort path", async () => {
+    const { store } = createEventedStore();
+    const executor = new TaskExecutor(store, "/tmp/test");
+    (executor as any).activeSessions.set("FN-1", {
+      session: { abort: vi.fn().mockResolvedValue(undefined), dispose: vi.fn() },
+      seenSteeringIds: new Set<string>(),
+    });
+
+    const abortSpy = vi.spyOn(executor as any, "awaitAbortInFlightTaskWork");
+    await executor.abortAllInFlight("engine stop");
+
+    expect(abortSpy).toHaveBeenCalledWith("FN-1", "engine stop");
+  });
+});

packages/engine/src/__tests__/reliability-interactions/engine-stop-aborts-execution.test.ts

@@ -0,0 +1,95 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { InProcessRuntime } from "../../runtimes/in-process-runtime.js";
+
+function makeExecutor(overrides: Record<string, unknown> = {}) {
+  return {
+    activeWorktrees: new Map(),
+    abortAllSessionBash: vi.fn(),
+    abortAllInFlight: vi.fn().mockResolvedValue(undefined),
+    disposeEphemeralTimers: vi.fn(),
+    ...overrides,
+  };
+}
+
+describe("FN-5403 reliability interactions: engine stop aborts execution", () => {
+  beforeEach(() => vi.useFakeTimers());
+  afterEach(() => vi.useRealTimers());
+
+  it("FN-5403: engine stop aborts executor AI sessions before drain completes", async () => {
+    const runtime = new InProcessRuntime({ projectId: "p", workingDirectory: "/tmp", isolationMode: "in-process" } as any, {} as any) as any;
+    let aborted = false;
+    let disposed = false;
+    runtime.status = "active";
+    runtime.taskStore = { getSettings: vi.fn().mockResolvedValue({ runtimeStopDrainMs: 1 }) };
+    runtime.pluginRunner = { shutdown: vi.fn().mockResolvedValue(undefined) };
+    runtime.worktreePool = { drain: vi.fn().mockReturnValue([]) };
+    runtime.executor = makeExecutor({
+      abortAllInFlight: vi.fn().mockImplementation(async () => {
+        aborted = true;
+        disposed = true;
+      }),
+    });
+
+    await runtime.stop();
+    expect(aborted).toBe(true);
+    expect(disposed).toBe(true);
+  });
+
+  it("FN-5403: engine stop does not wait the legacy 30 s for natural completion", async () => {
+    const runtime = new InProcessRuntime({ projectId: "p", workingDirectory: "/tmp", isolationMode: "in-process" } as any, {} as any) as any;
+    runtime.status = "active";
+    runtime.taskStore = { getSettings: vi.fn().mockResolvedValue({ runtimeStopDrainMs: 10 }) };
+    runtime.pluginRunner = { shutdown: vi.fn().mockResolvedValue(undefined) };
+    runtime.worktreePool = { drain: vi.fn().mockReturnValue([]) };
+    runtime.executor = makeExecutor({ activeWorktrees: new Map([["FN-1", { taskId: "FN-1" }]]) });
+
+    const stopPromise = runtime.stop();
+    await vi.advanceTimersByTimeAsync(50);
+    await expect(stopPromise).resolves.toBeUndefined();
+  });
+
+  it("FN-5403: engine stop interacts with TriageProcessor.stop", async () => {
+    const runtime = new InProcessRuntime({ projectId: "p", workingDirectory: "/tmp", isolationMode: "in-process" } as any, {} as any) as any;
+    runtime.status = "active";
+    const triageSessions = new Map([["FN-T", {}]]);
+    runtime.triageProcessor = { stop: vi.fn().mockImplementation(() => triageSessions.clear()) };
+    runtime.taskStore = { getSettings: vi.fn().mockResolvedValue({ runtimeStopDrainMs: 0 }) };
+    runtime.pluginRunner = { shutdown: vi.fn().mockResolvedValue(undefined) };
+    runtime.worktreePool = { drain: vi.fn().mockReturnValue([]) };
+    runtime.executor = makeExecutor();
+
+    await runtime.stop();
+    expect(runtime.triageProcessor.stop).toHaveBeenCalledTimes(1);
+    expect(runtime.executor.abortAllInFlight).toHaveBeenCalledWith("engine stop");
+    expect(triageSessions.size).toBe(0);
+  });
+
+  it("FN-5403: engine stop preserves task:moved cleanup contract", async () => {
+    const runtime = new InProcessRuntime({ projectId: "p", workingDirectory: "/tmp", isolationMode: "in-process" } as any, {} as any) as any;
+    runtime.status = "active";
+    const updateTask = vi.fn();
+    const moveTask = vi.fn();
+    runtime.taskStore = { getSettings: vi.fn().mockResolvedValue({ runtimeStopDrainMs: 0 }), updateTask, moveTask };
+    runtime.pluginRunner = { shutdown: vi.fn().mockResolvedValue(undefined) };
+    runtime.worktreePool = { drain: vi.fn().mockReturnValue([]) };
+    runtime.executor = makeExecutor();
+
+    await runtime.stop();
+    expect(updateTask).not.toHaveBeenCalled();
+    expect(moveTask).not.toHaveBeenCalled();
+  });
+
+  it("FN-5403: engine stop with runtimeStopDrainMs=0 still aborts before exiting", async () => {
+    const runtime = new InProcessRuntime({ projectId: "p", workingDirectory: "/tmp", isolationMode: "in-process" } as any, {} as any) as any;
+    runtime.status = "active";
+    const timeoutSpy = vi.spyOn(globalThis, "setTimeout");
+    runtime.taskStore = { getSettings: vi.fn().mockResolvedValue({ runtimeStopDrainMs: 0 }) };
+    runtime.pluginRunner = { shutdown: vi.fn().mockResolvedValue(undefined) };
+    runtime.worktreePool = { drain: vi.fn().mockReturnValue([]) };
+    runtime.executor = makeExecutor({ activeWorktrees: new Map([["FN-1", { taskId: "FN-1" }]]) });
+
+    await runtime.stop();
+    expect(runtime.executor.abortAllInFlight).toHaveBeenCalledWith("engine stop");
+    expect(timeoutSpy).not.toHaveBeenCalledWith(expect.any(Function), 500);
+  });
+});

packages/engine/src/executor.ts

@@ -1662,6 +1662,43 @@ export class TaskExecutor {
     }
   }
 
+  async abortAllInFlight(reason: string): Promise<void> {
+    const taskIds = new Set<string>([
+      ...this.activeSessions.keys(),
+      ...this.activeStepExecutors.keys(),
+      ...this.activeWorkflowStepSessions.keys(),
+      ...this.activeSubagentSessions.keys(),
+    ]);
+
+    for (const taskId of taskIds) {
+      try {
+        await this.awaitAbortInFlightTaskWork(taskId, reason);
+      } catch (err) {
+        executorLog.warn(`abortAllInFlight: failed to abort task ${taskId} — ${reason}: ${err}`);
+      }
+    }
+
+    for (const [agentId, session] of this.childSessions) {
+      try {
+        const sessionWithAbort = session as AgentSession & { abort?: () => Promise<void> };
+        if (typeof sessionWithAbort.abort === "function") {
+          await sessionWithAbort.abort();
+        }
+      } catch (err) {
+        executorLog.warn(`abortAllInFlight: failed to abort child session ${agentId} — ${reason}: ${err}`);
+      }
+
+      try {
+        session.dispose();
+      } catch (err) {
+        executorLog.warn(`abortAllInFlight: failed to dispose child session ${agentId} — ${reason}: ${err}`);
+      }
+    }
+    this.childSessions.clear();
+
+    executorLog.log(`abortAllInFlight: aborted ${taskIds.size} task surface(s) — ${reason}`);
+  }
+
   abortAllSessionBash(): void {
     for (const [taskId, { session }] of this.activeSessions) {
       try {

packages/engine/src/runtimes/__tests__/in-process-runtime.test.ts

@@ -194,6 +194,7 @@ vi.mock("../../executor.js", async () => {
       self.handleLoopDetected = vi.fn().mockResolvedValue(false);
       self.markStuckAborted = vi.fn();
       self.abortAllSessionBash = vi.fn().mockResolvedValue(undefined);
+      self.abortAllInFlight = vi.fn().mockResolvedValue(undefined);
       self.isEphemeralDeletionPending = vi.fn().mockReturnValue(false);
       self.disposeEphemeralTimers = vi.fn();
       self.activeWorktrees = new Map();
@@ -452,6 +453,93 @@ describe("InProcessRuntime", () => {
       expect(statusChanges).toContain("stopping");
       expect(statusChanges).toContain("stopped");
     }, 30000);
+
+    it("calls abortAllInFlight after bash abort and before drain checks", async () => {
+      await runtime.start();
+      const executor = (runtime as any).executor;
+      const callOrder: string[] = [];
+      executor.abortAllSessionBash.mockImplementation(() => {
+        callOrder.push("bash");
+      });
+      executor.abortAllInFlight.mockImplementation(async () => {
+        callOrder.push("inFlight");
+      });
+      const metricsSpy = vi.spyOn(runtime, "getMetrics").mockImplementation(() => {
+        callOrder.push("metrics");
+        return { inFlightTasks: 0, activeAgents: 0, lastActivityAt: new Date().toISOString() };
+      });
+
+      await runtime.stop();
+
+      expect(executor.abortAllInFlight).toHaveBeenCalledTimes(1);
+      expect(executor.abortAllInFlight).toHaveBeenCalledWith("engine stop");
+      expect(callOrder.indexOf("bash")).toBeLessThan(callOrder.indexOf("inFlight"));
+      expect(callOrder.indexOf("inFlight")).toBeLessThan(callOrder.indexOf("metrics"));
+      metricsSpy.mockRestore();
+    }, 30000);
+
+    it("honors runtimeStopDrainMs=0 and default 2000ms poll interval", async () => {
+      await runtime.start();
+      const timeoutSpy = vi.spyOn(globalThis, "setTimeout");
+      const executor = (runtime as any).executor;
+
+      mockTaskStoreSettings.runtimeStopDrainMs = 0;
+      executor.activeWorktrees.set("FN-1", { taskId: "FN-1" });
+      await runtime.stop();
+      expect(timeoutSpy).not.toHaveBeenCalledWith(expect.any(Function), 500);
+
+      delete mockTaskStoreSettings.runtimeStopDrainMs;
+      runtime = new InProcessRuntime(buildTestConfig(testDir), mockCentralCore);
+      await runtime.start();
+      const executor2 = (runtime as any).executor;
+      let metricCalls = 0;
+      executor2.activeWorktrees.set("FN-2", { taskId: "FN-2" });
+      const metricsSpy = vi.spyOn(runtime, "getMetrics").mockImplementation(() => {
+        metricCalls += 1;
+        if (metricCalls >= 2) {
+          executor2.activeWorktrees.clear();
+        }
+        return {
+          inFlightTasks: metricCalls === 1 ? 1 : 0,
+          activeAgents: 0,
+          lastActivityAt: new Date().toISOString(),
+        };
+      });
+
+      await runtime.stop();
+      expect(timeoutSpy).toHaveBeenCalledWith(expect.any(Function), 500);
+      metricsSpy.mockRestore();
+      timeoutSpy.mockRestore();
+    }, 30000);
+
+    it("logs post-abort drain timeout when in-flight tasks remain", async () => {
+      mockTaskStoreSettings.runtimeStopDrainMs = 50;
+      await runtime.start();
+      const warnSpy = vi.spyOn(runtimeLog, "warn").mockImplementation(() => undefined as any);
+      const executor = (runtime as any).executor;
+      executor.activeWorktrees.set("FN-stuck", { taskId: "FN-stuck" });
+      vi.spyOn(runtime, "getMetrics").mockImplementation(() => ({
+        inFlightTasks: 1,
+        activeAgents: 0,
+        lastActivityAt: new Date().toISOString(),
+      }));
+
+      await runtime.stop();
+
+      expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining("post-abort drain timeout"));
+    }, 30000);
+
+    it("continues stopping when abortAllInFlight throws", async () => {
+      await runtime.start();
+      const executor = (runtime as any).executor;
+      executor.abortAllInFlight.mockRejectedValueOnce(new Error("boom"));
+      const warnSpy = vi.spyOn(runtimeLog, "warn").mockImplementation(() => undefined as any);
+
+      await expect(runtime.stop()).resolves.toBeUndefined();
+
+      expect(runtime.getStatus()).toBe("stopped");
+      expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining("Failed to abort in-flight executor AI sessions"));
+    }, 30000);
   });
 
   describe("event forwarding", () => {

packages/engine/src/runtimes/in-process-runtime.ts

@@ -898,8 +898,6 @@ export class InProcessRuntime
       // process group), so killing the worker alone leaks vitest / npm / build
       // grandchildren as orphans. This call routes through pi-coding-agent's
       // AbortController -> killProcessTree, taking down the whole subtree.
-      // Sessions are intentionally NOT disposed here so near-complete steps
-      // can still wrap up during the drain window below.
       if (this.executor) {
         try {
           this.executor.abortAllSessionBash();
@@ -909,26 +907,41 @@ export class InProcessRuntime
         }
       }
 
-      // 8. Wait for active tasks to complete (30 second timeout)
-      const shutdownTimeout = 30000;
+      // 7c. Abort and dispose all in-flight AI sessions so shutdown does not
+      // continue streaming LLM output or tool calls during the drain phase.
+      if (this.executor) {
+        try {
+          await this.executor.abortAllInFlight("engine stop");
+          runtimeLog.log("Aborted in-flight executor AI sessions");
+        } catch (err) {
+          runtimeLog.warn(`Failed to abort in-flight executor AI sessions: ${err}`);
+        }
+      }
+
+      // 8. Wait for active tasks to drain after aborting live sessions.
+      const settings = this.taskStore ? await this.taskStore.getSettings() : undefined;
+      const shutdownTimeout = settings?.runtimeStopDrainMs ?? 2000;
       const startTime = Date.now();
 
-      while (Date.now() - startTime < shutdownTimeout) {
-        const metrics = this.getMetrics();
-        if (metrics.inFlightTasks === 0) {
-          break;
+      if (shutdownTimeout > 0) {
+        const pollIntervalMs = Math.min(500, shutdownTimeout);
+        while (Date.now() - startTime < shutdownTimeout) {
+          const metrics = this.getMetrics();
+          if (metrics.inFlightTasks === 0) {
+            break;
+          }
+          runtimeLog.log(
+            `Waiting for ${metrics.inFlightTasks} in-flight tasks to complete...`
+          );
+          await new Promise((resolve) => setTimeout(resolve, pollIntervalMs));
         }
-        runtimeLog.log(
-          `Waiting for ${metrics.inFlightTasks} in-flight tasks to complete...`
-        );
-        await new Promise((resolve) => setTimeout(resolve, 1000));
       }
 
       // Check if we timed out
       const finalMetrics = this.getMetrics();
       if (finalMetrics.inFlightTasks > 0) {
         runtimeLog.warn(
-          `Shutdown timeout reached with ${finalMetrics.inFlightTasks} tasks still in-flight`
+          `post-abort drain timeout: shutdown reached with ${finalMetrics.inFlightTasks} tasks still in-flight`
         );
       }