feat(FN-4825): complete Step 3 — add scheduler unreachable-owner audits

Fusion-Task-Id: FN-4825
Fusion-Task-Lineage: dc4e633e-fe46-47a8-9ecc-f032073caee9

packages/engine/src/__tests__/scheduler-node-unreachable-audit.test.ts

@@ -0,0 +1,123 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import type { RunAuditEventInput, Task, TaskStore } from "@fusion/core";
+import { existsSync } from "node:fs";
+import { readFile } from "node:fs/promises";
+import { Scheduler } from "../scheduler.js";
+
+vi.mock("node:fs", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("node:fs")>();
+  return { ...actual, existsSync: vi.fn() };
+});
+
+vi.mock("node:fs/promises", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("node:fs/promises")>();
+  return { ...actual, readFile: vi.fn() };
+});
+
+function createTask(overrides: Partial<Task> = {}): Task {
+  return {
+    id: "FN-1",
+    description: "x",
+    column: "todo",
+    dependencies: [],
+    steps: [],
+    currentStep: 0,
+    log: [],
+    createdAt: "2026-01-01T00:00:00Z",
+    updatedAt: "2026-01-01T00:00:00Z",
+    nodeId: "node-task",
+    checkedOutBy: "agent-owner",
+    checkoutNodeId: "node-owner",
+    ...overrides,
+  } as Task;
+}
+
+function createStore(task: Task) {
+  const recordRunAuditEvent = vi.fn().mockResolvedValue(undefined);
+  const store = {
+    listTasks: vi.fn().mockResolvedValue([task]),
+    getSettings: vi.fn().mockResolvedValue({
+      maxConcurrent: 1,
+      maxWorktrees: 1,
+      owningNodeHandoffPolicy: "block",
+      unavailableNodePolicy: "block",
+    }),
+    getTask: vi.fn().mockResolvedValue(task),
+    updateTask: vi.fn().mockResolvedValue(undefined),
+    moveTask: vi.fn().mockResolvedValue(undefined),
+    parseFileScopeFromPrompt: vi.fn().mockResolvedValue([]),
+    logEntry: vi.fn().mockResolvedValue(undefined),
+    getRootDir: vi.fn().mockReturnValue("/tmp/test"),
+    getTasksDir: vi.fn().mockReturnValue("/tmp/test/.fusion/tasks"),
+    recordRunAuditEvent,
+    on: vi.fn(),
+    off: vi.fn(),
+  } as unknown as TaskStore;
+  return { store, recordRunAuditEvent };
+}
+
+describe("Scheduler node-unreachable audit", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    vi.mocked(existsSync).mockReturnValue(true);
+    vi.mocked(readFile).mockResolvedValue("# Task\nBody");
+  });
+
+  it("emits park-action once per blocked task", async () => {
+    const { store, recordRunAuditEvent } = createStore(createTask());
+    const scheduler = new Scheduler(store, {
+      nodeHealthMonitor: { getNodeHealth: vi.fn(() => "offline") } as any,
+    });
+    (scheduler as unknown as { running: boolean }).running = true;
+
+    await scheduler.schedule();
+    await scheduler.schedule();
+
+    expect(recordRunAuditEvent).toHaveBeenCalledTimes(1);
+    const event = recordRunAuditEvent.mock.calls[0][0] as RunAuditEventInput;
+    expect(event.mutationType).toBe("task:auto-recover-node-unreachable");
+    expect(event.metadata).toMatchObject({
+      handoffAction: "park",
+      decisionPath: "scheduler-handoff-park",
+      ownerNodeId: "node-owner",
+      ownerNodeHealth: "offline",
+    });
+  });
+
+  it("emits reassign-local audit metadata", async () => {
+    const task = createTask({ id: "FN-2" });
+    const { store, recordRunAuditEvent } = createStore(task);
+    vi.mocked(store.getSettings).mockResolvedValue({
+      maxConcurrent: 1,
+      maxWorktrees: 1,
+      owningNodeHandoffPolicy: "reassign-local",
+      unavailableNodePolicy: "block",
+    } as any);
+    const scheduler = new Scheduler(store, {
+      nodeHealthMonitor: { getNodeHealth: vi.fn((id: string) => (id === "node-owner" ? "offline" : "online")) } as any,
+    });
+    (scheduler as unknown as { running: boolean }).running = true;
+
+    await scheduler.schedule();
+
+    const event = recordRunAuditEvent.mock.calls[0][0] as RunAuditEventInput;
+    expect(event.metadata).toMatchObject({
+      handoffAction: "reassign-local",
+      decisionPath: "scheduler-handoff-reassign-local",
+      dispatchNodeBefore: "node-task",
+      dispatchNodeAfter: undefined,
+    });
+  });
+
+  it("does not emit node-unreachable audit on healthy owner dispatch", async () => {
+    const { store, recordRunAuditEvent } = createStore(createTask({ id: "FN-3" }));
+    const scheduler = new Scheduler(store, {
+      nodeHealthMonitor: { getNodeHealth: vi.fn(() => "online") } as any,
+    });
+    (scheduler as unknown as { running: boolean }).running = true;
+
+    await scheduler.schedule();
+
+    expect(recordRunAuditEvent).not.toHaveBeenCalled();
+  });
+});

packages/engine/src/scheduler.ts

@@ -28,6 +28,7 @@ import type { MeshLeaseManager } from "./mesh-lease-manager.js";
 import { selectPermanentAgentForTask } from "./agent-assignment.js";
 import type { AutoClaimSnapshotManager } from "./auto-claim-snapshot.js";
 import { StaleTaskReporter } from "./stale-task-reporter.js";
+import { createRunAuditor, generateSyntheticRunId } from "./run-audit.js";
 
 /**
  * Check whether two sets of file scope paths overlap.
@@ -508,6 +509,43 @@ export class Scheduler {
     await this.store.logEntry(taskId, reason);
   }
 
+  private async emitNodeUnreachableRecoveryAudit(
+    task: Task,
+    metadata: {
+      ownerNodeId: string;
+      ownerNodeHealth: "offline" | "error";
+      handoffAction: "park" | "reassign-local" | "reassign-any";
+      handoffReason: string;
+      decisionPath: "scheduler-handoff-park" | "scheduler-handoff-reassign-local" | "scheduler-handoff-reassign-any";
+      newColumn: string;
+      dispatchNodeBefore?: string;
+      dispatchNodeAfter?: string;
+    },
+  ): Promise<void> {
+    const auditor = createRunAuditor(this.store, {
+      runId: generateSyntheticRunId("scheduler", task.id),
+      agentId: "scheduler",
+      taskId: task.id,
+      taskLineageId: task.lineageId,
+      phase: "dispatch-owning-node-handoff",
+    });
+
+    try {
+      await auditor.database({
+        type: "task:auto-recover-node-unreachable",
+        target: task.id,
+        metadata: {
+          previousColumn: task.column,
+          ...metadata,
+        },
+      });
+    } catch (error) {
+      schedulerLog.warn(
+        `Task ${task.id} failed to emit node-unreachable auto-recovery audit: ${error instanceof Error ? error.message : String(error)}`,
+      );
+    }
+  }
+
   private async emitHighOverlapFanoutWarnings(tasks: Task[]): Promise<void> {
     const fanoutMap = computeBlockerFanoutMap(tasks, 3);
     const seenBlockers = new Set<string>();
@@ -1015,6 +1053,16 @@ export class Scheduler {
               if (handoffDecision.action === "park") {
                 if (!this.wasNodeBlocked.has(task.id)) {
                   this.wasNodeBlocked.add(task.id);
+                  await this.emitNodeUnreachableRecoveryAudit(freshTask, {
+                    ownerNodeId: freshTask.checkoutNodeId,
+                    ownerNodeHealth,
+                    handoffAction: handoffDecision.action,
+                    handoffReason: handoffDecision.reason,
+                    decisionPath: "scheduler-handoff-park",
+                    newColumn: freshTask.column,
+                    dispatchNodeBefore: effectiveNode.nodeId,
+                    dispatchNodeAfter: effectiveNode.nodeId,
+                  });
                   const reason = `Owning-node handoff parked dispatch: ${handoffDecision.reason}`;
                   schedulerLog.log(`Task ${task.id} dispatch blocked — ${reason}`);
                   await this.store.logEntry(task.id, reason);
@@ -1023,9 +1071,23 @@ export class Scheduler {
               }
 
               await this.store.logEntry(task.id, `Owning-node handoff applied: ${handoffDecision.reason}`);
+              const dispatchNodeBefore = effectiveNode.nodeId;
               if (handoffDecision.action === "reassign-local") {
                 effectiveNode = { nodeId: undefined, source: "local" };
               }
+              await this.emitNodeUnreachableRecoveryAudit(freshTask, {
+                ownerNodeId: freshTask.checkoutNodeId,
+                ownerNodeHealth,
+                handoffAction: handoffDecision.action,
+                handoffReason: handoffDecision.reason,
+                decisionPath:
+                  handoffDecision.action === "reassign-local"
+                    ? "scheduler-handoff-reassign-local"
+                    : "scheduler-handoff-reassign-any",
+                newColumn: freshTask.column,
+                dispatchNodeBefore,
+                dispatchNodeAfter: effectiveNode.nodeId,
+              });
             }
           }