feat(FN-4465): complete Step 3 — scheduler stale task reporter

Fusion-Task-Id: FN-4465
Fusion-Task-Lineage: 4365d4cb-ab93-4ed5-add9-3e14b26237ad

packages/engine/src/__tests__/scheduler.test.ts

@@ -7,6 +7,8 @@ import { existsSync } from "node:fs";
 import { readFile } from "node:fs/promises";
 import { schedulerLog } from "../logger.js";
 
+const staleReporterReportMock = vi.fn();
+
 // Mock fs modules
 vi.mock("node:fs", async (importOriginal) => {
   const actual = await importOriginal<typeof import("node:fs")>();
@@ -32,6 +34,12 @@ vi.mock("../logger.js", () => ({
   },
 }));
 
+vi.mock("../stale-task-reporter.js", () => ({
+  StaleTaskReporter: vi.fn().mockImplementation(() => ({
+    report: staleReporterReportMock,
+  })),
+}));
+
 // Helper to create mock tasks
 function createMockTask(overrides: Partial<Task> = {}): Task {
   return {
@@ -146,6 +154,9 @@ describe("filterPathsByIgnoreList", () => {
 });
 
 describe("Scheduler", () => {
+  beforeEach(() => {
+    staleReporterReportMock.mockReset().mockResolvedValue({ surfaced: 0 });
+  });
   // Helper to create mock MissionStore (shared across mission-related test suites)
   function createMockMissionStore(overrides = {}) {
     return {
@@ -170,6 +181,70 @@ describe("Scheduler", () => {
     };
   }
 
+  describe("stale task reporter integration", () => {
+    it("invokes reporter from schedule", async () => {
+      vi.mocked(existsSync).mockReturnValue(true);
+      vi.mocked(readFile).mockResolvedValue("# Task\nDo something");
+      const store = createMockStore({
+        listTasks: vi.fn().mockResolvedValue([createMockTask({ id: "FN-STALE", column: "todo", dependencies: [] })]),
+        getSettings: vi.fn().mockResolvedValue({
+          maxConcurrent: 2,
+          maxWorktrees: 4,
+          staleInProgressWarningMs: 1000,
+          staleInReviewWarningMs: 2000,
+        }),
+      });
+      const scheduler = new Scheduler(store);
+      (scheduler as unknown as { running: boolean }).running = true;
+      await scheduler.schedule();
+      expect(staleReporterReportMock).toHaveBeenCalledTimes(1);
+    });
+
+    it("does not throw when reporter errors", async () => {
+      staleReporterReportMock.mockRejectedValueOnce(new Error("boom"));
+      vi.mocked(existsSync).mockReturnValue(true);
+      vi.mocked(readFile).mockResolvedValue("# Task\nDo something");
+      const store = createMockStore({
+        listTasks: vi.fn().mockResolvedValue([createMockTask({ id: "FN-STALE", column: "todo", dependencies: [] })]),
+        getSettings: vi.fn().mockResolvedValue({
+          maxConcurrent: 2,
+          maxWorktrees: 4,
+          staleInProgressWarningMs: 1000,
+          staleInReviewWarningMs: 2000,
+        }),
+      });
+      const scheduler = new Scheduler(store);
+      (scheduler as unknown as { running: boolean }).running = true;
+      await expect(scheduler.schedule()).resolves.toBeUndefined();
+      expect(schedulerLog.warn).toHaveBeenCalledWith("Stale task reporter failed", expect.any(Error));
+    });
+
+    it("rate-limits back-to-back reporter runs", async () => {
+      vi.useFakeTimers();
+      try {
+        vi.setSystemTime(new Date("2026-05-14T12:00:00.000Z"));
+        vi.mocked(existsSync).mockReturnValue(true);
+        vi.mocked(readFile).mockResolvedValue("# Task\nDo something");
+        const store = createMockStore({
+          listTasks: vi.fn().mockResolvedValue([createMockTask({ id: "FN-STALE", column: "todo", dependencies: [] })]),
+          getSettings: vi.fn().mockResolvedValue({
+            maxConcurrent: 2,
+            maxWorktrees: 4,
+            staleInProgressWarningMs: 5000,
+            staleInReviewWarningMs: 10000,
+          }),
+        });
+        const scheduler = new Scheduler(store);
+        (scheduler as unknown as { running: boolean }).running = true;
+        await scheduler.schedule();
+        await scheduler.schedule();
+        expect(staleReporterReportMock).toHaveBeenCalledTimes(1);
+      } finally {
+        vi.useRealTimers();
+      }
+    });
+  });
+
   describe("constructor", () => {
     it("initializes with default options", () => {
       const store = createMockStore();

packages/engine/src/__tests__/stale-task-reporter.test.ts

@@ -0,0 +1,88 @@
+import { describe, expect, it, vi } from "vitest";
+import type { Task, TaskStore } from "@fusion/core";
+import { StaleTaskReporter } from "../stale-task-reporter.js";
+
+function createTask(overrides: Partial<Task> = {}): Task {
+  return {
+    id: "FN-1",
+    description: "test",
+    column: "in-progress",
+    dependencies: [],
+    steps: [],
+    currentStep: 0,
+    paused: false,
+    log: [],
+    updatedAt: "2026-05-14T00:00:00.000Z",
+    createdAt: "2026-05-14T00:00:00.000Z",
+    ...overrides,
+  } as Task;
+}
+
+function createStore(taskSets: { inProgress?: Task[]; inReview?: Task[] } = {}, settings: Record<string, unknown> = {}): TaskStore {
+  return {
+    getSettings: vi.fn().mockResolvedValue(settings),
+    listTasks: vi.fn().mockImplementation(async ({ column }) => {
+      if (column === "in-progress") return taskSets.inProgress ?? [];
+      if (column === "in-review") return taskSets.inReview ?? [];
+      return [];
+    }),
+    logEntry: vi.fn().mockResolvedValue(undefined),
+  } as unknown as TaskStore;
+}
+
+describe("StaleTaskReporter", () => {
+  it("no-ops under threshold", async () => {
+    const now = Date.parse("2026-05-14T08:00:00.000Z");
+    const store = createStore({
+      inProgress: [createTask({ columnMovedAt: new Date(now - 60_000).toISOString() })],
+    }, { staleInProgressWarningMs: 4 * 60 * 60_000, staleInProgressCriticalMs: 24 * 60 * 60_000 });
+    const reporter = new StaleTaskReporter({ store, now: () => now });
+    const result = await reporter.report();
+    expect(result.surfaced).toBe(0);
+    expect(store.logEntry).not.toHaveBeenCalled();
+  });
+
+  it("emits warning once and rate-limits repeat within window", async () => {
+    const now = Date.parse("2026-05-14T08:00:00.000Z");
+    const task = createTask({ columnMovedAt: new Date(now - 5 * 60 * 60_000).toISOString() });
+    const store = createStore({ inProgress: [task] }, { staleInProgressWarningMs: 4 * 60 * 60_000, staleInProgressCriticalMs: 24 * 60 * 60_000 });
+    const reporter = new StaleTaskReporter({ store, now: () => now });
+
+    expect((await reporter.report()).surfaced).toBe(1);
+    task.log.push({ timestamp: new Date(now).toISOString(), action: "Stale task age threshold crossed [warning]: column=in-progress paused=false ageMs=1 warningThresholdMs=1 criticalThresholdMs=1" });
+    expect((await reporter.report()).surfaced).toBe(0);
+  });
+
+  it("emits on warning->critical and critical->warning level changes", async () => {
+    const now = Date.parse("2026-05-14T12:00:00.000Z");
+    const task = createTask({
+      columnMovedAt: new Date(now - 30 * 60 * 60_000).toISOString(),
+      log: [{ timestamp: new Date(now - 60_000).toISOString(), action: "Stale task age threshold crossed [warning]: x" }],
+    });
+    const store = createStore({ inProgress: [task] }, { staleInProgressWarningMs: 4 * 60 * 60_000, staleInProgressCriticalMs: 24 * 60 * 60_000 });
+    const reporter = new StaleTaskReporter({ store, now: () => now });
+    expect((await reporter.report()).surfaced).toBe(1);
+
+    task.columnMovedAt = new Date(now - 6 * 60 * 60_000).toISOString();
+    task.log = [{ timestamp: new Date(now - 60_000).toISOString(), action: "Stale task age threshold crossed [critical]: x" }];
+    expect((await reporter.report()).surfaced).toBe(1);
+  });
+
+  it("skips merge-confirmed and recently-updated tasks", async () => {
+    const now = Date.parse("2026-05-14T12:00:00.000Z");
+    const store = createStore({
+      inProgress: [createTask({ columnMovedAt: new Date(now - 30 * 60 * 60_000).toISOString(), mergeDetails: { mergeConfirmed: true } })],
+      inReview: [createTask({ id: "FN-2", column: "in-review", columnMovedAt: new Date(now - 30 * 60 * 60_000).toISOString(), updatedAt: new Date(now).toISOString() })],
+    }, { staleInProgressWarningMs: 4 * 60 * 60_000, staleInProgressCriticalMs: 24 * 60 * 60_000, staleInReviewWarningMs: 24 * 60 * 60_000, staleInReviewCriticalMs: 3 * 24 * 60 * 60_000 });
+    const reporter = new StaleTaskReporter({ store, now: () => now });
+    expect((await reporter.report()).surfaced).toBe(0);
+  });
+
+  it("returns zero and skips scans when all thresholds disabled", async () => {
+    const store = createStore({}, { staleInProgressWarningMs: 0, staleInProgressCriticalMs: 0, staleInReviewWarningMs: 0, staleInReviewCriticalMs: 0 });
+    const reporter = new StaleTaskReporter({ store });
+    const result = await reporter.report();
+    expect(result.surfaced).toBe(0);
+    expect(store.listTasks).not.toHaveBeenCalled();
+  });
+});

packages/engine/src/scheduler.ts

@@ -24,6 +24,7 @@ import type { NodeDispatchValidationResult } from "./node-dispatch-validation.js
 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";
 
 /**
  * Check whether two sets of file scope paths overlap.
@@ -187,6 +188,8 @@ export class Scheduler {
   private wasPermanentAgentUnavailable = new Set<string>();
   /** Tracks dispatch-queued reason signatures to avoid per-tick log spam. */
   private wasDispatchQueuedReasonLogged = new Set<string>();
+  private readonly staleTaskReporter: StaleTaskReporter;
+  private lastStaleTaskReportAt = 0;
 
   /**
    * Async listener guard convention:
@@ -199,6 +202,7 @@ export class Scheduler {
     private store: TaskStore,
     private options: SchedulerOptions = {},
   ) {
+    this.staleTaskReporter = new StaleTaskReporter({ store: this.store });
     /**
      * Event-driven scheduling: when a task is created, trigger a scheduling
      * pass immediately instead of waiting for the next poll interval.
@@ -1040,6 +1044,18 @@ export class Scheduler {
           if (scope.length > 0) activeScopes.set(task.id, scope);
         }
       }
+
+      const staleWarningWindows = [settings.staleInProgressWarningMs, settings.staleInReviewWarningMs]
+        .filter((value): value is number => typeof value === "number" && Number.isFinite(value) && value > 0);
+      const minWarningMs = staleWarningWindows.length > 0 ? Math.min(...staleWarningWindows) : 0;
+      if (minWarningMs > 0 && Date.now() - this.lastStaleTaskReportAt >= minWarningMs) {
+        try {
+          await this.staleTaskReporter.report();
+          this.lastStaleTaskReportAt = Date.now();
+        } catch (error) {
+          schedulerLog.warn("Stale task reporter failed", error);
+        }
+      }
     } catch (err) {
       schedulerLog.error("Scheduling error:", err);
     } finally {

packages/engine/src/stale-task-reporter.ts

@@ -0,0 +1,120 @@
+import {
+  getTaskAgeStalenessSignal,
+  type Task,
+  type TaskStore,
+  type Settings,
+} from "@fusion/core";
+import { schedulerLog } from "./logger.js";
+
+const STALE_LOG_PREFIX = "Stale task age threshold crossed";
+const STALE_LOG_RE = /^Stale task age threshold crossed \[(warning|critical)\]/;
+
+interface StaleTaskReporterOptions {
+  store: TaskStore;
+  logger?: Pick<typeof schedulerLog, "log" | "warn" | "error">;
+  now?: () => number;
+}
+
+export class StaleTaskReporter {
+  private readonly store: TaskStore;
+  private readonly logger: Pick<typeof schedulerLog, "log" | "warn" | "error">;
+  private readonly now: () => number;
+
+  constructor(options: StaleTaskReporterOptions) {
+    this.store = options.store;
+    this.logger = options.logger ?? schedulerLog;
+    this.now = options.now ?? (() => Date.now());
+  }
+
+  async report(): Promise<{ surfaced: number }> {
+    const settings = await this.store.getSettings();
+    const thresholds = this.getThresholds(settings);
+    const hasAnyThreshold = Object.values(thresholds).some((value) => typeof value === "number" && value > 0);
+    if (!hasAnyThreshold) {
+      return { surfaced: 0 };
+    }
+
+    const cycleStartMs = this.now();
+    const [inProgress, inReview] = await Promise.all([
+      this.store.listTasks({ column: "in-progress", slim: false }),
+      this.store.listTasks({ column: "in-review", slim: false }),
+    ]);
+
+    let surfaced = 0;
+    for (const task of [...inProgress, ...inReview]) {
+      const updatedAtMs = Date.parse(task.updatedAt);
+      if (Number.isFinite(updatedAtMs) && updatedAtMs >= cycleStartMs) {
+        continue;
+      }
+
+      let signal;
+      try {
+        signal = getTaskAgeStalenessSignal(task, { now: cycleStartMs, thresholds });
+      } catch (error) {
+        if (error instanceof RangeError) {
+          this.logger.warn(`Stale task reporter disabled by invalid thresholds: ${error.message}`);
+          return { surfaced };
+        }
+        throw error;
+      }
+      if (!signal) {
+        continue;
+      }
+
+      if (!this.shouldEmit(task, signal.level, signal.warningThresholdMs, signal.criticalThresholdMs, cycleStartMs)) {
+        continue;
+      }
+
+      const message = `${STALE_LOG_PREFIX} [${signal.level}]: column=${signal.column} paused=${String(signal.paused)} ageMs=${signal.ageMs} warningThresholdMs=${signal.warningThresholdMs} criticalThresholdMs=${signal.criticalThresholdMs}`;
+      await this.store.logEntry(task.id, message);
+      this.logger.log(message);
+      surfaced++;
+    }
+
+    return { surfaced };
+  }
+
+  private getThresholds(settings: Settings) {
+    return {
+      inProgressWarningMs: settings.staleInProgressWarningMs,
+      inProgressCriticalMs: settings.staleInProgressCriticalMs,
+      inReviewWarningMs: settings.staleInReviewWarningMs,
+      inReviewCriticalMs: settings.staleInReviewCriticalMs,
+    };
+  }
+
+  private shouldEmit(
+    task: Task,
+    level: "warning" | "critical",
+    warningThresholdMs: number,
+    criticalThresholdMs: number,
+    nowMs: number,
+  ): boolean {
+    const last = [...(task.log ?? [])].reverse().find((entry) => STALE_LOG_RE.test(entry.action));
+    if (!last) {
+      return true;
+    }
+
+    const match = last.action.match(STALE_LOG_RE);
+    if (!match) {
+      return true;
+    }
+
+    const lastLevel = match[1] as "warning" | "critical";
+    if (lastLevel !== level) {
+      return true;
+    }
+
+    const lastTs = Date.parse(last.timestamp);
+    if (!Number.isFinite(lastTs)) {
+      return true;
+    }
+
+    const windowMs = level === "critical" ? criticalThresholdMs : warningThresholdMs;
+    if (windowMs <= 0) {
+      return true;
+    }
+
+    return nowMs - lastTs >= windowMs;
+  }
+}