fix(cron): resilient file watcher + periodic reconciler (closes #15)

packages/jimmy/src/cron/__tests__/reconciler.test.ts

@@ -0,0 +1,65 @@
+import { describe, it, expect } from "vitest";
+import { signatureOfJobs } from "../scheduler.js";
+import type { CronJob } from "../../shared/types.js";
+
+function job(overrides: Partial<CronJob>): CronJob {
+  return {
+    id: "j1",
+    name: "test-job",
+    enabled: true,
+    schedule: "* * * * *",
+    timezone: "Asia/Kuala_Lumpur",
+    engine: "claude",
+    model: "sonnet",
+    employee: "jin",
+    prompt: "do the thing",
+    ...overrides,
+  } as CronJob;
+}
+
+describe("signatureOfJobs", () => {
+  it("is stable for the same input", () => {
+    const a = [job({ id: "a" }), job({ id: "b" })];
+    const b = [job({ id: "a" }), job({ id: "b" })];
+    expect(signatureOfJobs(a)).toBe(signatureOfJobs(b));
+  });
+
+  it("is order-independent", () => {
+    const a = [job({ id: "a" }), job({ id: "b" })];
+    const b = [job({ id: "b" }), job({ id: "a" })];
+    expect(signatureOfJobs(a)).toBe(signatureOfJobs(b));
+  });
+
+  it("excludes disabled jobs", () => {
+    const baseline = [job({ id: "a" })];
+    const withDisabled = [job({ id: "a" }), job({ id: "b", enabled: false })];
+    expect(signatureOfJobs(baseline)).toBe(signatureOfJobs(withDisabled));
+  });
+
+  it("changes when an enabled job's schedule changes", () => {
+    const before = [job({ id: "a", schedule: "0 * * * *" })];
+    const after = [job({ id: "a", schedule: "*/15 * * * *" })];
+    expect(signatureOfJobs(before)).not.toBe(signatureOfJobs(after));
+  });
+
+  it("changes when an enabled job's prompt changes (closure captures it)", () => {
+    // Critical: runCronJob captures the prompt in its closure at schedule
+    // time, so prompt edits must trigger a rescheduling pass. The
+    // reconciler relies on this signature to detect that case.
+    const before = [job({ prompt: "old" })];
+    const after = [job({ prompt: "new" })];
+    expect(signatureOfJobs(before)).not.toBe(signatureOfJobs(after));
+  });
+
+  it("changes when a disabled job is enabled", () => {
+    const before = [job({ id: "a", enabled: false })];
+    const after = [job({ id: "a", enabled: true })];
+    expect(signatureOfJobs(before)).not.toBe(signatureOfJobs(after));
+  });
+
+  it("does not change when a disabled job's fields are edited", () => {
+    const before = [job({ id: "x", enabled: true }), job({ id: "y", enabled: false, prompt: "p1" })];
+    const after = [job({ id: "x", enabled: true }), job({ id: "y", enabled: false, prompt: "p2" })];
+    expect(signatureOfJobs(before)).toBe(signatureOfJobs(after));
+  });
+});

packages/jimmy/src/cron/reconciler.ts

@@ -0,0 +1,69 @@
+import { logger } from "../shared/logger.js";
+import { loadJobs } from "./jobs.js";
+import {
+  reloadScheduler,
+  getScheduledSignature,
+  signatureOfJobs,
+} from "./scheduler.js";
+
+/**
+ * Periodic safety net for the cron scheduler.
+ *
+ * The file-watcher (`gateway/watcher.ts`) is the primary mechanism that
+ * picks up `cron/jobs.json` edits and calls `reloadScheduler`. But
+ * chokidar's single-file watch can silently die — atomic-rename writes
+ * (vim, git checkout, editor saves) unlink the watched inode and the
+ * underlying fs handle becomes invalid. When that happens, jobs.json edits
+ * stop being noticed and newly-enabled jobs never get scheduled.
+ *
+ * This reconciler ticks on a low-frequency timer and diffs the persisted
+ * jobs.json signature against the live scheduler's last-loaded signature.
+ * On divergence it forces a `reloadScheduler` — belt-and-suspenders.
+ *
+ * See nyem69/jinn#15 for the full diagnosis.
+ */
+
+const DEFAULT_INTERVAL_MS = 5 * 60 * 1000; // 5 minutes
+
+let timer: ReturnType<typeof setInterval> | null = null;
+
+export function startCronReconciler(intervalMs = DEFAULT_INTERVAL_MS): void {
+  if (timer) return; // already running
+  timer = setInterval(tickReconciler, intervalMs);
+  // Don't keep the process alive just for the reconciler.
+  timer.unref?.();
+  logger.info(
+    `Cron reconciler started (every ${Math.round(intervalMs / 1000)}s)`,
+  );
+}
+
+export function stopCronReconciler(): void {
+  if (timer) clearInterval(timer);
+  timer = null;
+}
+
+/**
+ * One reconciler tick — exported for tests. Safe to call manually.
+ * Returns true if a reload was forced, false otherwise.
+ */
+export function tickReconciler(): boolean {
+  try {
+    const jobs = loadJobs();
+    const persisted = signatureOfJobs(jobs);
+    const live = getScheduledSignature();
+    if (persisted !== live) {
+      logger.warn(
+        `Cron reconciler: persisted jobs.json diverges from live scheduler ` +
+          `(live=${live.slice(0, 8) || "<none>"} persisted=${persisted.slice(0, 8)}); forcing reloadScheduler`,
+      );
+      reloadScheduler(jobs);
+      return true;
+    }
+    return false;
+  } catch (err) {
+    logger.warn(
+      `Cron reconciler tick failed: ${(err as Error).message ?? err}`,
+    );
+    return false;
+  }
+}

packages/jimmy/src/cron/scheduler.ts

@@ -1,3 +1,4 @@
+import crypto from "node:crypto";
 import cron from "node-cron";
 import type {
   CronJob,
@@ -14,6 +15,38 @@ let currentSessionManager: SessionManager;
 let currentConfig: JinnConfig;
 let currentConnectors: Map<string, Connector>;
 
+// Signature of the last set of jobs successfully passed to scheduleJobs().
+// The reconciler (cron/reconciler.ts) diffs this against the persisted
+// jobs.json signature on a timer to catch silent file-watcher drops —
+// see nyem69/jinn#15.
+let lastScheduledSig = "";
+
+/**
+ * Canonical hash of the *enabled* jobs' scheduling-relevant fields.
+ * Includes prompt/engine/model/employee because runCronJob captures these
+ * at schedule time — editing any of them in jobs.json must trigger a
+ * rescheduling pass for the live closure to pick up the change.
+ */
+export function signatureOfJobs(jobs: CronJob[]): string {
+  const slim = jobs
+    .filter((j) => j.enabled)
+    .map((j) => ({
+      id: j.id,
+      schedule: j.schedule,
+      timezone: j.timezone ?? null,
+      engine: j.engine ?? null,
+      model: j.model ?? null,
+      employee: j.employee ?? null,
+      prompt: j.prompt ?? null,
+    }))
+    .sort((a, b) => a.id.localeCompare(b.id));
+  return crypto.createHash("sha1").update(JSON.stringify(slim)).digest("hex");
+}
+
+export function getScheduledSignature(): string {
+  return lastScheduledSig;
+}
+
 export function startScheduler(
   jobs: CronJob[],
   sessionManager: SessionManager,
@@ -57,6 +90,7 @@ function scheduleJobs(jobs: CronJob[]): void {
     tasks.push(task);
     logger.info(`Scheduled cron job "${job.name}" (${job.schedule})`);
   }
+  lastScheduledSig = signatureOfJobs(jobs);
 }
 
 export async function triggerCronJob(idOrName: string): Promise<CronJob | undefined> {

packages/jimmy/src/gateway/server.ts

@@ -25,6 +25,7 @@ import { WhatsAppConnector } from "../connectors/whatsapp/index.js";
 import { TelegramConnector } from "../connectors/telegram/index.js";
 import { loadJobs } from "../cron/jobs.js";
 import { startScheduler, reloadScheduler, stopScheduler } from "../cron/scheduler.js";
+import { startCronReconciler, stopCronReconciler } from "../cron/reconciler.js";
 import { scanOrg } from "./org.js";
 
 
@@ -584,6 +585,10 @@ export async function startGateway(
   // Start cron scheduler
   const cronJobs = loadJobs();
   startScheduler(cronJobs, sessionManager, config, connectorMap);
+  // Belt-and-suspenders for the file-watcher path: periodically diff the
+  // persisted jobs.json signature against the live scheduler and force a
+  // reload on divergence. See nyem69/jinn#15.
+  startCronReconciler();
   logger.info(`Loaded ${cronJobs.length} cron job(s)`);
 
   // Mutable config reference for hot-reload
@@ -822,7 +827,8 @@ export async function startGateway(
     claudeEngine.killAll();
     codexEngine.killAll();
 
-    // Stop cron scheduler
+    // Stop cron scheduler + reconciler
+    stopCronReconciler();
     stopScheduler();
 
     // Stop connectors

packages/jimmy/src/gateway/watcher.ts

@@ -101,31 +101,66 @@ export function syncSkillSymlinks(): void {
   }
 }
 
-export function startWatchers(callbacks: WatcherCallbacks): void {
-  const DEBOUNCE_MS = 500;
-
-  const configWatcher = watch(CONFIG_PATH, {
+/**
+ * Resilient single-file watcher.
+ *
+ * Watches the *parent directory* filtered to a single file (depth 0).
+ * Why: atomic-rename writes (vim, git checkout, many editors) unlink the
+ * original inode and create a new one. A direct file watch's underlying
+ * fs handle becomes invalid in that case and chokidar silently stops
+ * emitting events — which was the root cause of nyem69/jinn#15. A
+ * directory watch persists across inode replacements, and listening for
+ * both `change` and `add` catches in-place edits *and* atomic replaces.
+ */
+function watchSingleFile(
+  filePath: string,
+  label: string,
+  callback: () => void,
+  debounceMs: number,
+): FSWatcher {
+  const dir = path.dirname(filePath);
+  const file = path.basename(filePath);
+  const w = watch(dir, {
     ignoreInitial: true,
     awaitWriteFinish: { stabilityThreshold: 300 },
+    depth: 0,
+    ignored: (p) => {
+      // The watched directory itself must not be ignored.
+      if (path.resolve(p) === path.resolve(dir)) return false;
+      return path.basename(p) !== file;
+    },
   });
-  configWatcher.on(
-    "change",
-    debounce(() => {
-      logger.info("config.yaml changed, reloading...");
-      callbacks.onConfigReload();
-    }, DEBOUNCE_MS),
+  const fire = debounce(() => {
+    logger.info(`${label} changed, reloading...`);
+    try {
+      callback();
+    } catch (err) {
+      logger.warn(`${label} reload callback failed: ${(err as Error).message ?? err}`);
+    }
+  }, debounceMs);
+  w.on("change", fire);
+  w.on("add", fire); // atomic replace lands here, not "change"
+  w.on("error", (err) => {
+    logger.warn(`${label} watcher error: ${(err as Error).message ?? err}`);
+  });
+  return w;
+}
+
+export function startWatchers(callbacks: WatcherCallbacks): void {
+  const DEBOUNCE_MS = 500;
+
+  const configWatcher = watchSingleFile(
+    CONFIG_PATH,
+    "config.yaml",
+    callbacks.onConfigReload,
+    DEBOUNCE_MS,
   );
 
-  const cronWatcher = watch(CRON_JOBS, {
-    ignoreInitial: true,
-    awaitWriteFinish: { stabilityThreshold: 300 },
-  });
-  cronWatcher.on(
-    "change",
-    debounce(() => {
-      logger.info("cron/jobs.json changed, reloading...");
-      callbacks.onCronReload();
-    }, DEBOUNCE_MS),
+  const cronWatcher = watchSingleFile(
+    CRON_JOBS,
+    "cron/jobs.json",
+    callbacks.onCronReload,
+    DEBOUNCE_MS,
   );
 
   const orgWatcher = watch(ORG_DIR, {