feat(FN-4813): complete Step 5 — add isolation transition runtime restart path

Fusion-Task-Id: FN-4813 Fusion-Task-Lineage: 846893a5-2afa-4817-8f64-8c444d2fd713
2026-05-16 17:03:12 -07:00
parent 2ca18e983a
commit 63adddadf9
10 changed files with 513 additions and 96 deletions
--- a/packages/engine/src/tests/hybrid-executor.test.ts
+++ b/packages/engine/src/tests/hybrid-executor.test.ts
@@ -16,6 +16,7 @@ const mockProjectManagerInstances: Array<{
  getGlobalMetrics: ReturnType<typeof vi.fn>;
  acquireGlobalSlot: ReturnType<typeof vi.fn>;
  releaseGlobalSlot: ReturnType<typeof vi.fn>;
+  restartProjectRuntime: ReturnType<typeof vi.fn>;
  stopAll: ReturnType<typeof vi.fn>;
  on: ReturnType<typeof vi.fn>;
 }> = [];
@@ -71,6 +72,7 @@ vi.mock("../project-manager.js", () => ({
      }),
      acquireGlobalSlot: vi.fn().mockResolvedValue(true),
      releaseGlobalSlot: vi.fn().mockResolvedValue(undefined),
+      restartProjectRuntime: vi.fn().mockResolvedValue(undefined),
      stopAll: vi.fn().mockImplementation(() => {
        mockRuntimes.clear();
        mockProjectIds.length = 0;
@@ -133,6 +135,7 @@ describe("HybridExecutor", () => {
      registerProject: vi.fn().mockResolvedValue(mockProject),
      unregisterProject: vi.fn().mockResolvedValue(undefined),
      updateProject: vi.fn().mockResolvedValue(mockProject),
+      transitionProjectIsolation: vi.fn().mockResolvedValue({ ok: true }),
      getGlobalConcurrencyState: vi.fn().mockResolvedValue({
        globalMaxConcurrent: 4,
        currentlyActive: 0,
@@ -475,4 +478,26 @@ describe("HybridExecutor", () => {
      );
    });
  });
+
+  describe("transitionProjectIsolation", () => {
+    beforeEach(async () => {
+      await executor.initialize();
+      await executor.addProject({
+        projectId: "proj_test123",
+        workingDirectory: "/tmp/test-project",
+        isolationMode: "in-process",
+        maxConcurrent: 2,
+        maxWorktrees: 4,
+      });
+    });
+
+    it("rolls back persisted isolation mode when restart is blocked by active tasks", async () => {
+      const manager = mockProjectManagerInstances[0];
+      manager?.restartProjectRuntime.mockRejectedValueOnce({ kind: "active_tasks", count: 3 });
+
+      const result = await executor.transitionProjectIsolation("proj_test123", "child-process");
+      expect(result).toEqual({ ok: false, reason: "active_tasks", activeTaskCount: 3 });
+      expect(mockCentralCore.updateProject).toHaveBeenCalledWith("proj_test123", { isolationMode: "in-process" });
+    });
+  });
 });
--- a/packages/engine/src/tests/owning-node-handoff.integration.test.ts
+++ b/packages/engine/src/tests/owning-node-handoff.integration.test.ts
@@ -1,71 +1,82 @@
-import { describe, expect, it, vi } from "vitest";
-import type { Task, TaskStore } from "@fusion/core";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { mkdtempSync } from "node:fs";
+import { rm } from "node:fs/promises";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
+import { TaskStore, type OwningNodeHandoffPolicy } from "@fusion/core";
 import { MeshLeaseManager } from "../mesh-lease-manager.js";
+import type { NodeHealthMonitor } from "../node-health-monitor.js";

-function baseTask(overrides: Partial<Task> = {}): Task {
-  return {
-    id: "FN-1",
-    description: "handoff",
-    column: "in-progress",
-    dependencies: [],
-    steps: [],
-    currentStep: 0,
-    log: [],
-    createdAt: "2026-01-01T00:00:00.000Z",
-    updatedAt: "2026-01-01T00:00:00.000Z",
-    checkedOutBy: "agent-1",
-    checkedOutAt: "2026-01-01T00:00:00.000Z",
-    checkoutLeaseRenewedAt: "2026-01-01T00:00:00.000Z",
-    checkoutLeaseEpoch: 1,
-    checkoutNodeId: "node-owner",
-    ...overrides,
-  };
-}
-
-function createTaskStore(task: Task) {
-  let current = { ...task };
-  const updateTask = vi.fn(async (_id: string, patch: Partial<Task>) => {
-    current = { ...current, ...patch };
-    return current;
-  });
-  const taskStore = {
-    getTask: vi.fn(async () => current),
-    updateTask,
-    moveTask: vi.fn(async () => current),
-    logEntry: vi.fn(async () => undefined),
-  } as unknown as TaskStore;
-  return { taskStore, updateTask, getCurrent: () => current };
+function makeTmpDir(): string {
+  return mkdtempSync(join(tmpdir(), "fn-owning-handoff-test-"));
 }

 describe("MeshLeaseManager owning-node handoff integration", () => {
-  it.each([
-    { policy: "block", selfOwned: false, expectRecovered: false },
-    { policy: "reassign-to-local", selfOwned: false, expectRecovered: true },
-    { policy: "reassign-any-healthy", selfOwned: false, expectRecovered: true },
-    { policy: "block", selfOwned: true, expectRecovered: true },
-    { policy: "reassign-to-local", selfOwned: true, expectRecovered: true },
-    { policy: "reassign-any-healthy", selfOwned: true, expectRecovered: true },
-  ] as const)("policy=$policy selfOwned=$selfOwned", async ({ policy, selfOwned, expectRecovered }) => {
-    const ownerNodeId = selfOwned ? "node-local" : "node-owner";
-    const { taskStore, updateTask, getCurrent } = createTaskStore(baseTask({ checkoutNodeId: ownerNodeId }));
+  let rootDir: string;
+  let globalDir: string;
+  let taskStore: TaskStore;
+  let taskId: string;

+  beforeEach(async () => {
+    rootDir = makeTmpDir();
+    globalDir = join(rootDir, ".fusion-global");
+    taskStore = new TaskStore(rootDir, globalDir);
+    await taskStore.init();
+    taskId = (await taskStore.createTask({ description: "handoff" })).id;
+  });
+
+  afterEach(async () => {
+    taskStore?.close();
+    await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 });
+  });
+
+  async function seedLease(ownerNodeId: string): Promise<void> {
+    await taskStore.updateTask(taskId, {
+      column: "in-progress",
+      checkedOutBy: "agent-1",
+      checkedOutAt: "2026-05-01T00:00:00.000Z",
+      checkoutLeaseRenewedAt: "2026-05-01T00:00:00.000Z",
+      checkoutLeaseEpoch: 1,
+      checkoutNodeId: ownerNodeId,
+    });
+  }
+
+  async function runCase(policy: OwningNodeHandoffPolicy, ownerNodeId: string): Promise<boolean> {
    const manager = new MeshLeaseManager({
      taskStore,
-      nodeHealthMonitor: { getNodeHealth: vi.fn(() => "offline") } as any,
      localNodeId: "node-local",
      getHandoffPolicy: async () => policy,
+      nodeHealthMonitor: {
+        getNodeHealth: () => "offline",
+      } as unknown as NodeHealthMonitor,
    });
+    return manager.recoverAbandonedLease(taskId, "test-owner-unavailable", { preserveProgress: true });
+  }

-    const recovered = await manager.recoverAbandonedLease("FN-1", "test");
-    expect(recovered).toBe(expectRecovered);
+  it("applies handoff policy matrix for peer-owned leases", async () => {
+    await seedLease("node-peer");
+    expect(await runCase("block", "node-peer")).toBe(false);
+    let task = await taskStore.getTask(taskId);
+    expect(task?.checkedOutBy).toBe("agent-1");

-    if (expectRecovered) {
-      expect(updateTask).toHaveBeenCalled();
-      expect(getCurrent().checkedOutBy).toBeNull();
-      expect(getCurrent().checkoutNodeId).toBeNull();
-    } else {
-      expect(updateTask).not.toHaveBeenCalled();
-      expect(getCurrent().checkedOutBy).toBe("agent-1");
+    await seedLease("node-peer");
+    expect(await runCase("reassign-to-local", "node-peer")).toBe(true);
+    task = await taskStore.getTask(taskId);
+    expect(task?.checkedOutBy ?? null).toBeNull();
+
+    await seedLease("node-peer");
+    expect(await runCase("reassign-any-healthy", "node-peer")).toBe(true);
+    task = await taskStore.getTask(taskId);
+    expect(task?.checkedOutBy ?? null).toBeNull();
+  });
+
+  it("recovers self-owned leases regardless of policy", async () => {
+    for (const policy of ["block", "reassign-to-local", "reassign-any-healthy"] as const) {
+      await seedLease("node-local");
+      const recovered = await runCase(policy, "node-local");
+      expect(recovered).toBe(true);
+      const task = await taskStore.getTask(taskId);
+      expect(task?.checkedOutBy ?? null).toBeNull();
    }
  });
 });
--- a/packages/engine/src/tests/scheduler-node-routing.test.ts
+++ b/packages/engine/src/tests/scheduler-node-routing.test.ts
@@ -401,4 +401,67 @@ describe("Scheduler node routing", () => {

    expect((healthMonitor.getNodeHealth as ReturnType<typeof vi.fn>)).not.toHaveBeenCalled();
  });
+
+  it("parks dispatch when owning-node handoff policy blocks peer-owner takeover", async () => {
+    const task = createMockTask({ id: "FN-118", nodeId: "node-online", checkoutNodeId: "node-owner", checkedOutBy: "agent-owner" });
+    const store = createMockStore(task, {
+      maxConcurrent: 1,
+      maxWorktrees: 1,
+      unavailableNodePolicy: "block",
+      owningNodeHandoffPolicy: "block",
+    });
+    const healthMonitor = createMockHealthMonitor({ "node-online": "online", "node-owner": "offline" });
+    const scheduler = new Scheduler(store, { nodeHealthMonitor: healthMonitor });
+    (scheduler as unknown as { running: boolean }).running = true;
+
+    await scheduler.schedule();
+
+    expect(store.updateTask).not.toHaveBeenCalled();
+    expect(store.logEntry).toHaveBeenCalledWith(
+      task.id,
+      "Owning-node handoff parked dispatch: handoff_blocked_by_policy",
+    );
+  });
+
+  it("forces local dispatch when owning-node handoff returns reassign-local", async () => {
+    const task = createMockTask({ id: "FN-119", nodeId: "node-online", checkoutNodeId: "node-owner", checkedOutBy: "agent-owner" });
+    const store = createMockStore(task, {
+      maxConcurrent: 1,
+      maxWorktrees: 1,
+      unavailableNodePolicy: "block",
+      owningNodeHandoffPolicy: "reassign-to-local",
+    });
+    const healthMonitor = createMockHealthMonitor({ "node-online": "online", "node-owner": "offline" });
+    const scheduler = new Scheduler(store, { nodeHealthMonitor: healthMonitor });
+    (scheduler as unknown as { running: boolean }).running = true;
+
+    await scheduler.schedule();
+
+    expect(store.updateTask).toHaveBeenCalledWith(task.id, expect.objectContaining({
+      effectiveNodeId: null,
+      effectiveNodeSource: "local",
+    }));
+    expect(store.logEntry).toHaveBeenCalledWith(task.id, "Owning-node handoff applied: owner_offline_local_takes_over");
+  });
+
+  it("keeps non-local routing when owning-node handoff returns reassign-any", async () => {
+    const task = createMockTask({ id: "FN-120", nodeId: "node-online", checkoutNodeId: "node-owner", checkedOutBy: "agent-owner" });
+    const store = createMockStore(task, {
+      maxConcurrent: 1,
+      maxWorktrees: 1,
+      unavailableNodePolicy: "block",
+      owningNodeHandoffPolicy: "reassign-any-healthy",
+    });
+    const healthMonitor = createMockHealthMonitor({ "node-online": "online", "node-owner": "offline" });
+    const scheduler = new Scheduler(store, { nodeHealthMonitor: healthMonitor });
+    (scheduler as unknown as { running: boolean }).running = true;
+
+    await scheduler.schedule();
+
+    expect(store.updateTask).toHaveBeenCalledWith(task.id, expect.objectContaining({
+      effectiveNodeId: "node-online",
+      effectiveNodeSource: "task-override",
+    }));
+    expect(store.logEntry).toHaveBeenCalledWith(task.id, "Owning-node handoff applied: owner_offline_any_healthy_eligible");
+  });
 });
--- a/packages/engine/src/hybrid-executor.ts
+++ b/packages/engine/src/hybrid-executor.ts
@@ -1,5 +1,5 @@
 import { EventEmitter } from "node:events";
-import type { Task, CentralCore, RegisteredProject } from "@fusion/core";
+import type { Task, CentralCore, RegisteredProject, IsolationMode } from "@fusion/core";
 import { ProjectManager } from "./project-manager.js";
 import { NodeHealthMonitor } from "./node-health-monitor.js";
 import type {
@@ -48,6 +48,8 @@ export interface HybridExecutorEvents {
  "project:added": [data: { projectId: string; projectName: string }];
  /** Emitted when a project runtime is removed */
  "project:removed": [data: { projectId: string; projectName: string }];
+  /** Emitted when a project runtime is restarted */
+  "project:runtime-restarted": [data: { projectId: string; projectName: string; isolationMode: IsolationMode; reason?: string }];
 }

 /**
@@ -309,6 +311,47 @@ export class HybridExecutor extends EventEmitter<HybridExecutorEvents> {
    return existingRuntime;
  }

+
+  /**
+   * Transition project isolation mode and restart runtime to apply changes.
+   *
+   * Non-force transitions that fail runtime restart due to active tasks roll back
+   * the persisted isolationMode change to its previous value before returning.
+   */
+  async transitionProjectIsolation(
+    projectId: string,
+    nextMode: IsolationMode,
+    opts?: { force?: boolean },
+  ): Promise<{ ok: true } | { ok: false; reason: string; activeTaskCount?: number }> {
+    const current = await this.centralCore.getProject(projectId);
+    if (!current) {
+      return { ok: false, reason: "project_not_found" };
+    }
+
+    const transition = await this.centralCore.transitionProjectIsolation(projectId, nextMode, opts);
+    if (!transition.ok) {
+      return transition;
+    }
+
+    try {
+      await this.projectManager.restartProjectRuntime(projectId, {
+        reason: `isolation-transition:${current.isolationMode}->${nextMode}`,
+        force: opts?.force,
+      });
+      return { ok: true };
+    } catch (error) {
+      if (!opts?.force && error && typeof error === "object" && (error as { kind?: unknown }).kind === "active_tasks") {
+        await this.centralCore.updateProject(projectId, { isolationMode: current.isolationMode });
+        return {
+          ok: false,
+          reason: "active_tasks",
+          activeTaskCount: Number((error as { count?: unknown }).count) || 0,
+        };
+      }
+      throw error;
+    }
+  }
+
  /**
   * Get a runtime by project ID.
   */
@@ -438,6 +481,10 @@ export class HybridExecutor extends EventEmitter<HybridExecutorEvents> {
    this.projectManager.on("runtime:removed", (data) => {
      this.emit("project:removed", data);
    });
+
+    this.projectManager.on("project:runtime-restarted", (data) => {
+      this.emit("project:runtime-restarted", data);
+    });
  }

  /**
--- a/packages/engine/src/project-manager.ts
+++ b/packages/engine/src/project-manager.ts
@@ -45,6 +45,21 @@ export interface ProjectManagerEvents {
  "runtime:added": [data: { projectId: string; projectName: string }];
  /** Emitted when a runtime is removed */
  "runtime:removed": [data: { projectId: string; projectName: string }];
+  /** Emitted when a runtime is restarted */
+  "project:runtime-restarted": [data: { projectId: string; projectName: string; isolationMode: import("@fusion/core").IsolationMode; reason?: string }];
+}
+
+
+
+export interface ProjectRuntimeRestartBlockedError {
+  kind: "active_tasks";
+  count: number;
+}
+
+export function isProjectRuntimeRestartBlockedError(error: unknown): error is ProjectRuntimeRestartBlockedError {
+  if (!error || typeof error !== "object") return false;
+  const candidate = error as { kind?: unknown; count?: unknown };
+  return candidate.kind === "active_tasks" && typeof candidate.count === "number";
 }

 /**
@@ -463,6 +478,43 @@ export class ProjectManager extends EventEmitter<ProjectManagerEvents> {
    }
  }

+
+  async restartProjectRuntime(projectId: string, options?: { reason?: string; force?: boolean }): Promise<void> {
+    const runtime = this.runtimes.get(projectId);
+    if (!runtime) {
+      throw new Error(`Runtime not found for project ${projectId}`);
+    }
+
+    const metrics = runtime.getMetrics();
+    if (!options?.force && metrics.inFlightTasks > 0) {
+      throw { kind: "active_tasks", count: metrics.inFlightTasks } satisfies ProjectRuntimeRestartBlockedError;
+    }
+
+    const project = await this.centralCore.getProject(projectId);
+    if (!project) {
+      throw new Error(`Project not found: ${projectId}`);
+    }
+
+    const workingDirectory = await this.centralCore.resolveLocalProjectWorkingDirectory(projectId);
+    await this.removeProject(projectId);
+
+    await this.addProject({
+      projectId,
+      workingDirectory,
+      isolationMode: project.isolationMode,
+      maxConcurrent: project.settings?.maxConcurrent ?? 2,
+      maxWorktrees: project.settings?.maxWorktrees ?? 4,
+      settings: project.settings,
+    });
+
+    this.emit("project:runtime-restarted", {
+      projectId,
+      projectName: project.name,
+      isolationMode: project.isolationMode,
+      reason: options?.reason,
+    });
+  }
+
  /**
   * Stop all runtimes and clean up.
   */
--- a/packages/engine/src/scheduler.ts
+++ b/packages/engine/src/scheduler.ts
@@ -999,64 +999,58 @@ export class Scheduler {
          this.wasNodeDispatchValidationBlocked.delete(task.id);
        }

-        // Enforce unavailable-node policy
+        // Enforce unavailable-node policy + owning-node handoff policy
        if (effectiveNode.nodeId !== undefined && this.options.nodeHealthMonitor) {
-          let skipUnavailableNodePolicy = false;
-
-          if (freshTask.checkoutNodeId) {
+          if (freshTask.checkoutNodeId && freshTask.checkedOutBy) {
            const ownerNodeHealth = this.options.nodeHealthMonitor.getNodeHealth(freshTask.checkoutNodeId);
            if (ownerNodeHealth === "offline" || ownerNodeHealth === "error") {
              const handoffDecision = decideOwningNodeHandoff({
                task: freshTask,
                ownerNodeId: freshTask.checkoutNodeId,
                ownerNodeHealth,
-                localNodeId: settings.defaultNodeId ?? "local",
+                localNodeId: "local",
                handoffPolicy: settings.owningNodeHandoffPolicy,
              });

              if (handoffDecision.action === "park") {
                if (!this.wasNodeBlocked.has(task.id)) {
                  this.wasNodeBlocked.add(task.id);
-                  schedulerLog.log(`Task ${task.id} dispatch blocked — ${handoffDecision.reason}`);
-                  await this.store.logEntry(task.id, handoffDecision.reason);
+                  const reason = `Owning-node handoff parked dispatch: ${handoffDecision.reason}`;
+                  schedulerLog.log(`Task ${task.id} dispatch blocked — ${reason}`);
+                  await this.store.logEntry(task.id, reason);
                }
                continue;
              }

+              await this.store.logEntry(task.id, `Owning-node handoff applied: ${handoffDecision.reason}`);
              if (handoffDecision.action === "reassign-local") {
                effectiveNode = { nodeId: undefined, source: "local" };
-                await this.store.logEntry(task.id, `Owner handoff: ${handoffDecision.reason}`);
-              } else if (handoffDecision.action === "reassign-any") {
-                skipUnavailableNodePolicy = true;
-                await this.store.logEntry(task.id, `Owner handoff: ${handoffDecision.reason}`);
              }
            }
          }

-          if (!skipUnavailableNodePolicy) {
-            const nodeHealth = this.options.nodeHealthMonitor.getNodeHealth(effectiveNode.nodeId);
-            const decision = applyUnavailableNodePolicy({
-              effectiveNode,
-              nodeHealth,
-              policy: settings.unavailableNodePolicy,
-            });
+          const nodeHealth = this.options.nodeHealthMonitor.getNodeHealth(effectiveNode.nodeId);
+          const decision = applyUnavailableNodePolicy({
+            effectiveNode,
+            nodeHealth,
+            policy: settings.unavailableNodePolicy,
+          });

-            if (!decision.allowed) {
-              if (!this.wasNodeBlocked.has(task.id)) {
-                this.wasNodeBlocked.add(task.id);
-                schedulerLog.log(`Task ${task.id} dispatch blocked — ${decision.reason}`);
-                await this.store.logEntry(task.id, decision.reason);
-              }
-              continue;
-            }
-
-            this.wasNodeBlocked.delete(task.id);
-
-            if (decision.fallbackToLocal) {
-              schedulerLog.log(`Task ${task.id} falling back to local — ${decision.reason}`);
+          if (!decision.allowed) {
+            if (!this.wasNodeBlocked.has(task.id)) {
+              this.wasNodeBlocked.add(task.id);
+              schedulerLog.log(`Task ${task.id} dispatch blocked — ${decision.reason}`);
              await this.store.logEntry(task.id, decision.reason);
-              effectiveNode = { nodeId: undefined, source: "local" };
            }
+            continue;
+          }
+
+          this.wasNodeBlocked.delete(task.id);
+
+          if (decision.fallbackToLocal) {
+            schedulerLog.log(`Task ${task.id} falling back to local — ${decision.reason}`);
+            await this.store.logEntry(task.id, decision.reason);
+            effectiveNode = { nodeId: undefined, source: "local" };
          }
        }