feat(FN-4813): complete Step 4 — wire handoff into scheduler and lease recovery

Fusion-Task-Id: FN-4813
Fusion-Task-Lineage: 846893a5-2afa-4817-8f64-8c444d2fd713

packages/engine/src/__tests__/owning-node-handoff.integration.test.ts

@@ -0,0 +1,71 @@
+import { describe, expect, it, vi } from "vitest";
+import type { Task, TaskStore } from "@fusion/core";
+import { MeshLeaseManager } from "../mesh-lease-manager.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 };
+}
+
+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 }));
+
+    const manager = new MeshLeaseManager({
+      taskStore,
+      nodeHealthMonitor: { getNodeHealth: vi.fn(() => "offline") } as any,
+      localNodeId: "node-local",
+      getHandoffPolicy: async () => policy,
+    });
+
+    const recovered = await manager.recoverAbandonedLease("FN-1", "test");
+    expect(recovered).toBe(expectRecovered);
+
+    if (expectRecovered) {
+      expect(updateTask).toHaveBeenCalled();
+      expect(getCurrent().checkedOutBy).toBeNull();
+      expect(getCurrent().checkoutNodeId).toBeNull();
+    } else {
+      expect(updateTask).not.toHaveBeenCalled();
+      expect(getCurrent().checkedOutBy).toBe("agent-1");
+    }
+  });
+});

packages/engine/src/mesh-lease-manager.ts

@@ -1,11 +1,23 @@
-import type { AgentStore, RunMutationContext, Task, TaskStore } from "@fusion/core";
+import type {
+  AgentStore,
+  OwningNodeHandoffPolicy,
+  RunMutationContext,
+  Task,
+  TaskStore,
+} from "@fusion/core";
 import type { NodeHealthMonitor } from "./node-health-monitor.js";
+import { decideOwningNodeHandoff } from "./node-routing-policy.js";
+import { createLogger } from "./logger.js";
+
+const meshLeaseManagerLog = createLogger("mesh-lease-manager");
 
 export interface MeshLeaseManagerOptions {
   taskStore: TaskStore;
   agentStore?: AgentStore;
   nodeHealthMonitor?: NodeHealthMonitor;
   getExecutingTaskIds?: () => Set<string>;
+  localNodeId?: string;
+  getHandoffPolicy?: () => Promise<OwningNodeHandoffPolicy | undefined>;
 }
 
 export interface LeaseRecoveryContext {
@@ -78,6 +90,27 @@ export class MeshLeaseManager {
       return false;
     }
 
+    if ((stale.reason === "owner_node_offline" || stale.reason === "owner_node_error")
+      && task.checkoutNodeId
+      && this.options.nodeHealthMonitor) {
+      const ownerNodeHealth = this.options.nodeHealthMonitor.getNodeHealth(task.checkoutNodeId);
+      const handoffPolicy = await this.options.getHandoffPolicy?.();
+      const handoffDecision = decideOwningNodeHandoff({
+        task,
+        ownerNodeId: task.checkoutNodeId,
+        ownerNodeHealth,
+        localNodeId: this.options.localNodeId ?? "local",
+        handoffPolicy,
+      });
+
+      if (handoffDecision.action === "park") {
+        meshLeaseManagerLog.log(
+          `mesh-lease: handoff parked taskId=${task.id} reason=${handoffDecision.reason}`,
+        );
+        return false;
+      }
+    }
+
     const nextEpoch = (task.checkoutLeaseEpoch ?? 0) + 1;
     await this.options.taskStore.updateTask(
       taskId,

packages/engine/src/scheduler.ts

@@ -22,7 +22,7 @@ import { type PrMonitor, type PrComment } from "./pr-monitor.js";
 import { reconcileMissionFeatureState } from "./mission-feature-sync.js";
 import { evaluateSpecStaleness, getPromptPath } from "./spec-staleness.js";
 import { resolveEffectiveNode } from "./effective-node.js";
-import { applyUnavailableNodePolicy } from "./node-routing-policy.js";
+import { applyUnavailableNodePolicy, decideOwningNodeHandoff } from "./node-routing-policy.js";
 import type { NodeDispatchValidationResult } from "./node-dispatch-validation.js";
 import type { MeshLeaseManager } from "./mesh-lease-manager.js";
 import { selectPermanentAgentForTask } from "./agent-assignment.js";
@@ -1001,28 +1001,62 @@ export class Scheduler {
 
         // Enforce unavailable-node policy
         if (effectiveNode.nodeId !== undefined && this.options.nodeHealthMonitor) {
-          const nodeHealth = this.options.nodeHealthMonitor.getNodeHealth(effectiveNode.nodeId);
-          const decision = applyUnavailableNodePolicy({
-            effectiveNode,
-            nodeHealth,
-            policy: settings.unavailableNodePolicy,
-          });
+          let skipUnavailableNodePolicy = false;
 
-          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);
+          if (freshTask.checkoutNodeId) {
+            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",
+                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);
+                }
+                continue;
+              }
+
+              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}`);
+              }
             }
-            continue;
           }
 
-          this.wasNodeBlocked.delete(task.id);
+          if (!skipUnavailableNodePolicy) {
+            const nodeHealth = this.options.nodeHealthMonitor.getNodeHealth(effectiveNode.nodeId);
+            const decision = applyUnavailableNodePolicy({
+              effectiveNode,
+              nodeHealth,
+              policy: settings.unavailableNodePolicy,
+            });
 
-          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" };
+            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}`);
+              await this.store.logEntry(task.id, decision.reason);
+              effectiveNode = { nodeId: undefined, source: "local" };
+            }
           }
         }