feat(jobs): move the budget reset and token health check onto the registry

Both jobs owned their own timer and started themselves as an import side effect,
so nothing could report whether they were running, when they last ran, or why a
run failed. They now register with the job registry and are started from it, which
also means their schedule and run history are visible through /api/jobs.

startAll() runs each interval job's first tick synchronously, so both entry points
start the registry only after initializeCloudSync() has been awaited. The old
wiring reached that ordering two different ways: the budget reset was started
after the init call, and the health check's first sweep sat behind a 10s timer.
Replacing both with one startAll() would otherwise have moved the two handlers
in front of the initialisation they run against.

Both entry points also register the same pair of jobs. Registering one and not
the other is how a background job goes missing without anything failing.

sweep() now returns how many connections it swept, so the health check can record
a real records_affected the way the budget reset does. The migration documents
that column as a per-job count, and hardcoding zero would have left one of the two
jobs reporting a number the schema promises but the code never produces. A skipped
or empty sweep reports zero. Every existing caller ignores the return value.

The token health check keeps its own disable semantics: the handler still calls
isHealthCheckDisabled() before sweeping, so OMNIROUTE_DISABLE_TOKEN_HEALTHCHECK,
the production-build phase and the automated-test guard behave as before. Its
registry adapter lives in src/lib/jobs/ next to the budget reset rather than in
tokenHealthCheck.ts, which is already above its frozen size ceiling on the base
branch and should not grow further. The adapter lets a failing sweep throw rather
than reporting it itself, matching the budget reset: safeRun records a thrown
error as a failure run with its message.

The warmup job is seeded disabled. Its handler arrives with the warmup scheduler,
and startAll() filters on enabled before it looks for a handler, so seeding it
enabled here would warn about the missing handler on every boot.

src/lib/db/migrations/139_job_registry.sql

@@ -1,4 +1,4 @@
--- Migration 136: generic job registry (jobs + job_runs tables)
+-- Migration 139: generic job registry (jobs + job_runs tables)
 -- Job registry (#8848): centralized periodic-job scheduling + run history.
 --
 -- jobs:      one row per registered job (interval or cron), with env-flag gating
@@ -33,10 +33,14 @@ CREATE INDEX IF NOT EXISTS idx_jr_started_at ON job_runs(started_at);
 -- Built-in job registration (idempotent - INSERT OR IGNORE).
 -- env_flag = NULL means "no registry-level boolean gate"; per-job disable semantics
 -- live inside the handler itself (see token_health_check wrapper).
-INSERT OR IGNORE INTO jobs (id, type, cron, interval_ms, env_flag, config) VALUES
-  ('budget_reset', 'interval', NULL, 600000, NULL, '{}'),
-  ('warmup', 'cron', '0 7 * * *', NULL, 'OMNIROUTE_WARMUP_ENABLED', '{"timezone":"America/Los_Angeles","envDefault":false}'),
-  ('token_health_check', 'interval', NULL, 60000, NULL, '{}');
+-- 'warmup' is seeded disabled because its handler is not part of this change.
+-- startAll() filters on enabled before it looks for a handler, so a disabled row
+-- stays quiet instead of warning on every boot; the change that brings the warmup
+-- handler flips it on.
+INSERT OR IGNORE INTO jobs (id, type, cron, interval_ms, enabled, env_flag, config) VALUES
+  ('budget_reset', 'interval', NULL, 600000, 1, NULL, '{}'),
+  ('warmup', 'cron', '0 7 * * *', NULL, 0, 'OMNIROUTE_WARMUP_ENABLED', '{"timezone":"America/Los_Angeles","envDefault":false}'),
+  ('token_health_check', 'interval', NULL, 60000, 1, NULL, '{}');
 
 -- records_affected semantics:
 --   budget_reset       = number of budget records reset (UPDATE ... SET budget_used=0 row count)

src/lib/initCloudSync.ts

@@ -1,8 +1,9 @@
 import initializeCloudSync from "@/shared/services/initializeCloudSync";
-import { startBudgetResetJob } from "@/lib/jobs/budgetResetJob";
 import { startModelSyncScheduler } from "@/shared/services/modelSyncScheduler";
-import { startWarmupScheduler } from "@/lib/warmupScheduler";
 import { isAutomatedTestProcess } from "@/shared/utils/testProcess";
+import { getJobRegistry } from "@/lib/jobRegistry";
+import { registerBudgetResetJob } from "@/lib/jobs/budgetResetJob";
+import { registerTokenHealthCheck } from "@/lib/jobs/tokenHealthCheckJob";
 
 // Initialize runtime background sync services once per server process.
 let initialized = false;
@@ -29,12 +30,19 @@ export async function ensureCloudSyncInitialized() {
   }
   if (!initialized) {
     try {
-      const { initTokenHealthCheck } = await import("@/lib/tokenHealthCheck");
-      initTokenHealthCheck();
       await initializeCloudSync();
       startModelSyncScheduler();
-      startBudgetResetJob();
-      startWarmupScheduler();
+
+      // startAll() runs each interval job's first tick synchronously, so it has to
+      // come after initializeCloudSync(). The old wiring got that ordering two
+      // different ways: the budget reset was started right here, and the health
+      // check's first sweep sat behind a 10s timer. Awaiting the init is a firmer
+      // guarantee than the timer was.
+      const registry = getJobRegistry();
+      registerBudgetResetJob(registry);
+      registerTokenHealthCheck(registry);
+      await registry.startAll();
+
       initialized = true;
     } catch (error) {
       console.error("[ServerInit] Error initializing background sync services:", error);

src/lib/jobs/budgetResetJob.ts

@@ -1,40 +1,42 @@
 import { syncAllBudgetSchedules } from "@/domain/costRules";
+import type { JobRegistry } from "../jobRegistry/registry";
 
 const DEFAULT_INTERVAL_MS = 10 * 60 * 1000;
 
-let timer: NodeJS.Timeout | null = null;
-
 function getIntervalMs() {
   const raw = process.env.OMNIROUTE_BUDGET_RESET_JOB_INTERVAL_MS;
   const parsed = raw ? Number(raw) : Number.NaN;
   return Number.isFinite(parsed) && parsed >= 10_000 ? parsed : DEFAULT_INTERVAL_MS;
 }
 
-export function startBudgetResetJob() {
-  if (timer) {
-    return timer;
+/**
+ * Budget-reset handler - resets all budget schedules once per interval.
+ *
+ * Migrated to the JobRegistry: the old start/stop timer logic is
+ * gone; the registry owns scheduling + run recording. This module exports just the
+ * `run` handler and a `registerBudgetResetJob` wiring helper. Errors are no longer
+ * caught here - safeRun records them as a failure run.
+ */
+export async function run(): Promise<{ success: boolean; recordsAffected: number }> {
+  const result = syncAllBudgetSchedules(Date.now());
+  if (result.resetCount > 0) {
+    console.log(`[BudgetReset] processed=${result.processed} reset=${result.resetCount}`);
   }
-
-  const run = () => {
-    try {
-      const result = syncAllBudgetSchedules(Date.now());
-      if (result.resetCount > 0) {
-        console.log(`[BudgetReset] processed=${result.processed} reset=${result.resetCount}`);
-      }
-    } catch (error) {
-      console.error("[BudgetReset] Job failed:", error);
-    }
-  };
-
-  run();
-  timer = setInterval(run, getIntervalMs());
-  timer.unref?.();
-  return timer;
+  return { success: true, recordsAffected: result.resetCount };
 }
 
-export function stopBudgetResetJob() {
-  if (timer) {
-    clearInterval(timer);
-    timer = null;
-  }
+/** Wire the budget_reset job into a JobRegistry (idempotent - call at boot). */
+export function registerBudgetResetJob(registry: JobRegistry): void {
+  registry.register({
+    id: "budget_reset",
+    type: "interval",
+    cron: null,
+    intervalMs: getIntervalMs(),
+    enabled: true,
+    envFlag: null,
+    config: {},
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+    handler: run,
+  });
 }

src/lib/jobs/tokenHealthCheckJob.ts

@@ -0,0 +1,40 @@
+/**
+ * Registry adapter for the token health check.
+ *
+ * The sweep used to start itself on import. The registry owns the schedule now,
+ * so the adapter lives here next to the other job registrations rather than in
+ * tokenHealthCheck.ts, which is already at its size ceiling.
+ *
+ * Disable semantics are unchanged: isHealthCheckDisabled() still honours
+ * OMNIROUTE_DISABLE_TOKEN_HEALTHCHECK, NEXT_PHASE=phase-production-build and
+ * isAutomatedTestProcess(). The registry only fires the interval; the handler
+ * decides whether there is anything to do.
+ */
+import { isHealthCheckDisabled, sweep } from "@/lib/tokenHealthCheck";
+import type { JobRegistry } from "@/lib/jobRegistry/registry";
+
+const TOKEN_HEALTH_CHECK_INTERVAL_MS = 60_000;
+
+export function registerTokenHealthCheck(registry: JobRegistry): void {
+  const now = new Date().toISOString();
+  registry.register({
+    id: "token_health_check",
+    type: "interval",
+    cron: null,
+    intervalMs: TOKEN_HEALTH_CHECK_INTERVAL_MS,
+    enabled: true,
+    envFlag: null,
+    config: {},
+    createdAt: now,
+    updatedAt: now,
+    handler: async () => {
+      if (isHealthCheckDisabled()) {
+        return { success: true, recordsAffected: 0 };
+      }
+      // Errors are not caught here: safeRun records a thrown error as a failure run
+      // with its message, the same as the budget reset job.
+      const swept = await sweep();
+      return { success: true, recordsAffected: swept };
+    },
+  });
+}

src/lib/tokenHealthCheck.ts

@@ -267,7 +267,7 @@ function isEnvFlagEnabled(name: string): boolean {
   return TRUE_ENV_VALUES.has(value.trim().toLowerCase());
 }
 
-function isHealthCheckDisabled(): boolean {
+export function isHealthCheckDisabled(): boolean {
   return (
     isEnvFlagEnabled("OMNIROUTE_DISABLE_TOKEN_HEALTHCHECK") ||
     isBuildProcess() ||
@@ -421,16 +421,18 @@ export function stopTokenHealthCheck() {
 }
 
 // ── Core sweep (batch concurrent) ──────────────────────────────────────────
-export async function sweep() {
+/** Returns the number of connections swept, which the job registry records. */
+export async function sweep(): Promise<number> {
   const state = getHCState();
   if (state.sweeping) {
-    return log(`${LOG_PREFIX} Sweep skipped — previous sweep still in progress`);
+    log(`${LOG_PREFIX} Sweep skipped — previous sweep still in progress`);
+    return 0;
   }
   state.sweeping = true;
   try {
     const connections = await getProviderConnections({ authType: "oauth" });
 
-    if (!connections || connections.length === 0) return;
+    if (!connections || connections.length === 0) return 0;
 
     const staggerMs = parseInt(process.env.HEALTHCHECK_STAGGER_MS || "3000", 10);
     const total = connections.length;
@@ -471,8 +473,10 @@ export async function sweep() {
         await new Promise((resolve) => setTimeout(resolve, 0));
       }
     }
+    return total;
   } catch (err) {
     logError(`${LOG_PREFIX} Sweep error:`, err.message);
+    return 0;
   } finally {
     state.sweeping = false;
   }
@@ -1045,8 +1049,3 @@ export async function checkConnection(conn) {
     );
   }
 }
-
-// Auto-start when imported
-initTokenHealthCheck();
-
-export default initTokenHealthCheck;

src/server-init.ts

@@ -4,7 +4,8 @@ import { enforceWebRuntimeEnv } from "./lib/env/runtimeEnv";
 import { enforceSecrets } from "./shared/utils/secretsValidator";
 import { initAuditLog, cleanupExpiredLogs, logAuditEvent } from "./lib/compliance/index";
 import { initConsoleInterceptor } from "./lib/consoleInterceptor";
-import { startBudgetResetJob } from "./lib/jobs/budgetResetJob";
+import { registerBudgetResetJob } from "./lib/jobs/budgetResetJob";
+import { registerTokenHealthCheck } from "./lib/jobs/tokenHealthCheckJob";
 import { startReasoningCacheCleanupJob } from "./lib/jobs/reasoningCacheCleanupJob";
 import { startCleanupScheduler } from "./lib/db/cleanup";
 import { getSettings } from "./lib/db/settings";
@@ -113,7 +114,16 @@ async function startServer() {
     }
 
     await initializeCloudSync();
-    startBudgetResetJob();
+    // register() only persists the definition; startAll() is what arms the timers.
+    // This path does not call ensureCloudSyncInitialized(), so nothing else here
+    // would start the jobs on our behalf. It registers the same set as that path:
+    // starting one job and not the other is how a background job goes missing
+    // without anything failing.
+    const { getJobRegistry } = await import("./lib/jobRegistry");
+    const jobRegistry = getJobRegistry();
+    registerBudgetResetJob(jobRegistry);
+    registerTokenHealthCheck(jobRegistry);
+    await jobRegistry.startAll();
     startReasoningCacheCleanupJob();
     startCleanupScheduler();
     startRuntimeConfigHotReload();

tests/unit/db/jobRegistryDb.test.ts

@@ -1,5 +1,5 @@
 /**
- * Tests for jobRegistry persistence (migration 136 + jobRegistryDb.ts).
+ * Tests for jobRegistry persistence (migration 139 + jobRegistryDb.ts).
  *
  * Verifies:
  *  - 3 built-in jobs seeded by the migration
@@ -54,7 +54,9 @@ test("seed: warmup is cron type with env gate + envDefault=false", () => {
   assert.equal(warmup.type, "cron");
   assert.equal(warmup.cron, "0 7 * * *");
   assert.equal(warmup.envFlag, "OMNIROUTE_WARMUP_ENABLED");
-  assert.equal(warmup.enabled, true);
+  // Seeded disabled: the warmup handler is not registered by this change, and
+  // startAll() would otherwise warn about the missing handler on every boot.
+  assert.equal(warmup.enabled, false);
   assert.equal(warmup.config.envDefault, false);
   assert.equal(warmup.config.timezone, "America/Los_Angeles");
 });

tests/unit/lib/jobRegistry/boot-wiring.test.ts

@@ -0,0 +1,61 @@
+/**
+ * Boot wiring for the job registry.
+ *
+ * startAll() runs each interval job's first tick synchronously (see startInterval
+ * in registry.ts), so starting a job before initializeCloudSync() has been awaited
+ * runs its handler against a half-initialised app. The previous wiring avoided that
+ * two different ways: the budget reset was started after the init call, and the
+ * token health check's first sweep sat behind a 10s timer.
+ *
+ * Neither entry point can be driven from a unit test - both are process-boot
+ * functions that reach the real cloud sync - so these read the wiring, the same
+ * way tests/unit/model-sync-scheduler.test.ts already asserts on this file.
+ */
+
+import test from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import path from "node:path";
+
+function readSource(relativePath: string): string {
+  return fs.readFileSync(path.join(process.cwd(), relativePath), "utf8");
+}
+
+test("initCloudSync: cloud sync is initialised before the registry starts any job", () => {
+  const source = readSource("src/lib/initCloudSync.ts");
+  const initCall = source.indexOf("await initializeCloudSync(");
+  const startAllCall = source.indexOf("registry.startAll(");
+
+  assert.notEqual(initCall, -1, "initializeCloudSync() call not found");
+  assert.notEqual(startAllCall, -1, "registry.startAll() call not found");
+  assert.ok(
+    initCall < startAllCall,
+    "startAll() fires each interval job's first tick immediately, so it must come after initializeCloudSync()"
+  );
+});
+
+test("server-init: cloud sync is initialised before the registry starts any job", () => {
+  const source = readSource("src/server-init.ts");
+  const initCall = source.indexOf("await initializeCloudSync(");
+  const startAllCall = source.indexOf("jobRegistry.startAll(");
+
+  assert.notEqual(initCall, -1, "initializeCloudSync() call not found");
+  assert.notEqual(startAllCall, -1, "jobRegistry.startAll() call not found");
+  assert.ok(initCall < startAllCall, "startAll() must come after initializeCloudSync()");
+});
+
+test("both boot paths register the same jobs", () => {
+  for (const relativePath of ["src/lib/initCloudSync.ts", "src/server-init.ts"]) {
+    const source = readSource(relativePath);
+    assert.match(
+      source,
+      /registerBudgetResetJob\s*\(/,
+      `${relativePath} must register the budget reset job`
+    );
+    assert.match(
+      source,
+      /registerTokenHealthCheck\s*\(/,
+      `${relativePath} must register the token health check job`
+    );
+  }
+});

tests/unit/token-health-check-sweep.test.ts

@@ -104,9 +104,13 @@ test("sweep() skips re-entrant calls while a previous sweep is still in flight",
   assert.equal(isSweeping(), true, "first sweep should mark sweeping=true immediately");
 
   const start = Date.now();
-  await sweep(); // second, concurrent call — must be skipped by the guard
+  const skippedCount = await sweep(); // second, concurrent call — must be skipped by the guard
   const elapsedMs = Date.now() - start;
 
+  // The job registry records this as records_affected, so a skipped sweep has
+  // to report nothing rather than inheriting the in-flight sweep's count.
+  assert.equal(skippedCount, 0, "a skipped sweep reports 0 connections swept");
+
   // With 21 connections and a 300ms inter-batch stagger, a REAL second
   // sweep would take >= 300ms to clear the single batch gap. The guard must
   // make this call return near-
@@ -130,8 +134,9 @@ test("sweep() resets the sweeping flag after a normal completion, allowing the n
   process.env.HEALTHCHECK_STAGGER_MS = "0";
   await createNoOpOauthConnections(2, "sequential");
 
-  await sweep();
+  const sweptCount = await sweep();
   assert.equal(isSweeping(), false, "flag resets after first sweep completes");
+  assert.equal(sweptCount, 2, "sweep reports how many connections it swept");
 
   // A second, fully sequential call must not be skipped by leftover state —
   // it should run and complete normally rather than hang or short-circuit.
@@ -144,7 +149,8 @@ test("sweep() resets the sweeping flag even when there are no connections to pro
   process.env.HEALTHCHECK_STAGGER_MS = "0";
   // No connections created — sweep() takes the early `connections.length === 0` return.
 
-  await sweep();
+  const emptyCount = await sweep();
 
   assert.equal(isSweeping(), false, "sweeping flag must be false after an empty sweep");
+  assert.equal(emptyCount, 0, "an empty sweep reports 0 connections swept");
 });