feat(db): add a job registry for scheduled background work

Background jobs each ship their own timer today, so there is no list of what
is scheduled, no history of what ran, and no way to pause one without an
environment variable and a restart. The registry gives them one home: a jobs
table holding the schedule, a job_runs table holding the outcomes, and a
loopback-only API to inspect and control both.

Cron jobs read their expression through an optional cronGetter rather than the
stored column, so an operator changing OMNIROUTE_WARMUP_CRON does not need the
row rewritten. register() is an idempotent upsert that refreshes the schedule
but never overwrites `enabled` or `created_at`, which is what lets a job be
re-registered on every boot without discarding the operator's toggle.

Run history is pruned per job rather than globally, and safeRun records a
failure for a handler that throws as well as one that returns success:false,
so a crashing job leaves a trail instead of a gap.

The API is under /api/jobs and gated to loopback in the route guard. It can
trigger a run and flip a job off, which is runtime administration and does not
belong on a remotely reachable surface.

Signed-off-by: Minxi Hou <houminxi@gmail.com>

tests/unit/api/jobs.test.ts

@@ -0,0 +1,210 @@
+/** /api/jobs route tests . */
+
+import test from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+
+const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-jobs-api-"));
+process.env.DATA_DIR = TEST_DATA_DIR;
+process.env.DISABLE_SQLITE_AUTO_BACKUP = "true";
+process.env.OMNIROUTE_WARMUP_ENABLED = "1";
+
+const core = await import("../../../src/lib/db/core.ts");
+const { getJobRegistry, __resetJobRegistry } =
+  await import("../../../src/lib/jobRegistry/index.ts");
+const route = await import("../../../src/app/api/jobs/route.ts");
+const runsRoute = await import("../../../src/app/api/jobs/[id]/runs/route.ts");
+const enableRoute = await import("../../../src/app/api/jobs/[id]/enable/route.ts");
+const disableRoute = await import("../../../src/app/api/jobs/[id]/disable/route.ts");
+const runNowRoute = await import("../../../src/app/api/jobs/[id]/run-now/route.ts");
+const { isLocalOnlyPath } = await import("../../../src/server/authz/routeGuard.ts");
+
+function resetAll() {
+  try {
+    getJobRegistry().stopAll();
+  } catch {
+    // no singleton yet
+  }
+  __resetJobRegistry();
+  core.resetDbInstance();
+  fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true });
+  fs.mkdirSync(TEST_DATA_DIR, { recursive: true });
+}
+
+test.beforeEach(() => {
+  resetAll();
+});
+
+test.after(() => {
+  core.resetDbInstance();
+  fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true });
+});
+
+function params(id: string) {
+  return Promise.resolve({ id });
+}
+
+async function json(res: Response) {
+  return res.json();
+}
+
+test("GET /api/jobs -> 200 + list with lastRun, seeds present", async () => {
+  const reg = getJobRegistry();
+  // Register a custom job so listJobs reflects runtime registrations.
+  reg.register({
+    id: "custom",
+    type: "interval",
+    cron: null,
+    intervalMs: 1000,
+    enabled: true,
+    envFlag: null,
+    config: {},
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+    handler: async () => ({ success: true }),
+  });
+  const res = await route.GET();
+  assert.equal(res.status, 200);
+  const body = (await json(res)) as { data: Array<{ id: string; lastRun: unknown }> };
+  assert.ok(Array.isArray(body.data));
+  const ids = body.data.map((j) => j.id);
+  assert.ok(ids.includes("budget_reset"), "seeded budget_reset present");
+  assert.ok(ids.includes("warmup"), "seeded warmup present");
+  assert.ok(ids.includes("custom"), "runtime-registered custom present");
+  // DTO whitelist: no handler/timer leaked.
+  for (const job of body.data) {
+    assert.ok(!("handler" in job), "DTO must not expose handler");
+    assert.ok(Object.prototype.hasOwnProperty.call(job, "lastRun"), "lastRun present");
+  }
+});
+
+test("GET /api/jobs/:id/runs -> 200 + history", async () => {
+  const reg = getJobRegistry();
+  reg.register({
+    id: "custom",
+    type: "interval",
+    cron: null,
+    intervalMs: 1000,
+    enabled: true,
+    envFlag: null,
+    config: {},
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+    handler: async () => ({ success: true, recordsAffected: 2 }),
+  });
+  await reg.runNow("custom");
+  await new Promise((r) => setTimeout(r, 30));
+  const res = await runsRoute.GET(new Request("http://localhost/api/jobs/custom/runs"), {
+    params: params("custom"),
+  });
+  assert.equal(res.status, 200);
+  const body = (await json(res)) as { data: Array<{ status: string }> };
+  assert.ok(body.data.length >= 1);
+  assert.equal(body.data[0].status, "success");
+});
+
+test("GET /api/jobs/unknown/runs -> 404", async () => {
+  const res = await runsRoute.GET(new Request("http://localhost/api/jobs/nope/runs"), {
+    params: params("nope"),
+  });
+  assert.equal(res.status, 404);
+});
+
+test("POST /api/jobs/:id/enable -> 200 + enabled:true", async () => {
+  const reg = getJobRegistry();
+  reg.register({
+    id: "custom",
+    type: "interval",
+    cron: null,
+    intervalMs: 1000,
+    enabled: false,
+    envFlag: null,
+    config: {},
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+    handler: async () => ({ success: true }),
+  });
+  const res = await enableRoute.POST(
+    new Request("http://localhost/api/jobs/custom/enable", { method: "POST" }),
+    { params: params("custom") }
+  );
+  assert.equal(res.status, 200);
+  const body = (await json(res)) as { data: { id: string; enabled: boolean } };
+  assert.deepEqual(body.data, { id: "custom", enabled: true });
+  assert.equal(reg.listJobs().find((j) => j.id === "custom")!.enabled, true);
+});
+
+test("POST /api/jobs/:id/disable -> 200 + enabled:false", async () => {
+  const reg = getJobRegistry();
+  reg.register({
+    id: "custom",
+    type: "interval",
+    cron: null,
+    intervalMs: 1000,
+    enabled: true,
+    envFlag: null,
+    config: {},
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+    handler: async () => ({ success: true }),
+  });
+  const res = await disableRoute.POST(
+    new Request("http://localhost/api/jobs/custom/disable", { method: "POST" }),
+    { params: params("custom") }
+  );
+  assert.equal(res.status, 200);
+  const body = (await json(res)) as { data: { id: string; enabled: boolean } };
+  assert.deepEqual(body.data, { id: "custom", enabled: false });
+  assert.equal(reg.listJobs().find((j) => j.id === "custom")!.enabled, false);
+});
+
+test("POST /api/jobs/:id/run-now -> 200 + started:true", async () => {
+  const reg = getJobRegistry();
+  reg.register({
+    id: "custom",
+    type: "interval",
+    cron: null,
+    intervalMs: 1000,
+    enabled: true,
+    envFlag: null,
+    config: {},
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+    handler: async () => ({ success: true }),
+  });
+  const res = await runNowRoute.POST(
+    new Request("http://localhost/api/jobs/custom/run-now", { method: "POST" }),
+    { params: params("custom") }
+  );
+  assert.equal(res.status, 200);
+  const body = (await json(res)) as { data: { started: boolean } };
+  assert.equal(body.data.started, true);
+});
+
+test("POST /api/jobs/unknown/run-now -> 404", async () => {
+  const res = await runNowRoute.POST(
+    new Request("http://localhost/api/jobs/nope/run-now", { method: "POST" }),
+    { params: params("nope") }
+  );
+  assert.equal(res.status, 404);
+});
+
+test("LOCAL_ONLY guard: /api/jobs and children are loopback-only", () => {
+  assert.equal(isLocalOnlyPath("/api/jobs"), true, "bare /api/jobs");
+  assert.equal(isLocalOnlyPath("/api/jobs/"), true, "/api/jobs/");
+  assert.equal(isLocalOnlyPath("/api/jobs/budget_reset/runs"), true, "runs sub-path");
+  assert.equal(isLocalOnlyPath("/api/jobs/warmup/enable"), true, "enable sub-path");
+  assert.equal(isLocalOnlyPath("/api/jobs/warmup/disable"), true, "disable sub-path");
+  assert.equal(isLocalOnlyPath("/api/jobs/warmup/run-now"), true, "run-now sub-path");
+});
+
+test("error responses do not leak stack traces", async () => {
+  // 404 path - assert the error message is sanitized (no "at /" frame).
+  const res = await runsRoute.GET(new Request("http://localhost/api/jobs/nope/runs"), {
+    params: params("nope"),
+  });
+  const body = (await json(res)) as { error: { message: string } };
+  assert.ok(!body.error.message.includes("at /"), "error message must not leak stack path");
+});

tests/unit/db/jobRegistryDb.test.ts

@@ -0,0 +1,288 @@
+/**
+ * Tests for jobRegistry persistence (migration 136 + jobRegistryDb.ts).
+ *
+ * Verifies:
+ *  - 3 built-in jobs seeded by the migration
+ *  - upsertJob insert + column-level update (preserves enabled + created_at)
+ *  - recordRun writes ISO-8601 timestamps
+ *  - pruneRuns count dimension (150 rows -> keep 100)
+ *  - pruneRuns time dimension (rows older than 30 days deleted)
+ *  - pruneRuns dual dimension (150 rows, 50 older than 30 days -> keep 100)
+ *  - cleanupOrphanedRuns fixes stale 'running' records past the timeout
+ *
+ * Runs against an isolated temp DATA_DIR so the real ~/.omniroute DB is never touched.
+ */
+
+import test from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+
+const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-jr-"));
+process.env.DATA_DIR = TEST_DATA_DIR;
+process.env.DISABLE_SQLITE_AUTO_BACKUP = "true";
+
+const core = await import("../../../src/lib/db/core.ts");
+const db = await import("../../../src/lib/db/jobRegistryDb.ts");
+
+function resetDb() {
+  core.resetDbInstance();
+  fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true });
+  fs.mkdirSync(TEST_DATA_DIR, { recursive: true });
+}
+
+test.beforeEach(() => {
+  resetDb();
+});
+
+test.after(() => {
+  core.resetDbInstance();
+  fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true });
+});
+
+test("seed: migration registers 3 built-in jobs", () => {
+  const jobs = db.getAllJobs();
+  assert.equal(jobs.length, 3);
+  const ids = jobs.map((j) => j.id).sort();
+  assert.deepEqual(ids, ["budget_reset", "token_health_check", "warmup"]);
+});
+
+test("seed: warmup is cron type with env gate + envDefault=false", () => {
+  const warmup = db.getAllJobs().find((j) => j.id === "warmup");
+  assert.ok(warmup);
+  assert.equal(warmup.type, "cron");
+  assert.equal(warmup.cron, "0 7 * * *");
+  assert.equal(warmup.envFlag, "OMNIROUTE_WARMUP_ENABLED");
+  assert.equal(warmup.enabled, true);
+  assert.equal(warmup.config.envDefault, false);
+  assert.equal(warmup.config.timezone, "America/Los_Angeles");
+});
+
+test("seed: budget_reset is interval type with no env gate", () => {
+  const budget = db.getAllJobs().find((j) => j.id === "budget_reset");
+  assert.ok(budget);
+  assert.equal(budget.type, "interval");
+  assert.equal(budget.intervalMs, 600000);
+  assert.equal(budget.envFlag, null);
+});
+
+test("upsertJob: insert a new job then update scheduling fields", () => {
+  const created = new Date().toISOString();
+  db.upsertJob({
+    id: "custom",
+    type: "interval",
+    cron: null,
+    intervalMs: 1000,
+    enabled: true,
+    envFlag: null,
+    config: { foo: "bar" },
+    createdAt: created,
+    updatedAt: created,
+  });
+  let job = db.getJob("custom");
+  assert.ok(job);
+  assert.equal(job.type, "interval");
+  assert.equal(job.intervalMs, 1000);
+  assert.deepEqual(job.config, { foo: "bar" });
+
+  // Update interval + config; created_at must not change.
+  db.upsertJob({
+    ...job!,
+    intervalMs: 2000,
+    config: { foo: "baz" },
+  });
+  job = db.getJob("custom");
+  assert.equal(job!.intervalMs, 2000);
+  assert.deepEqual(job!.config, { foo: "baz" });
+  assert.equal(job!.createdAt, created);
+});
+
+test("upsertJob: does NOT overwrite enabled (user toggle preserved)", () => {
+  const ts = new Date().toISOString();
+  db.upsertJob({
+    id: "toggle",
+    type: "interval",
+    cron: null,
+    intervalMs: 1000,
+    enabled: true,
+    envFlag: null,
+    config: {},
+    createdAt: ts,
+    updatedAt: ts,
+  });
+  db.updateJobEnabled("toggle", false);
+  // Re-register with enabled=true - must not flip the user's disabled state.
+  db.upsertJob({
+    id: "toggle",
+    type: "interval",
+    cron: null,
+    intervalMs: 1000,
+    enabled: true,
+    envFlag: null,
+    config: {},
+    createdAt: ts,
+    updatedAt: ts,
+  });
+  assert.equal(db.getJob("toggle")!.enabled, false);
+});
+
+test("recordRun: writes a completed run with ISO timestamps", () => {
+  db.upsertJob({
+    id: "j",
+    type: "interval",
+    cron: null,
+    intervalMs: 1000,
+    enabled: true,
+    envFlag: null,
+    config: {},
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+  });
+  db.recordRun("j", "success", {
+    startedAt: "2026-01-01T00:00:00.000Z",
+    durationMs: 123,
+    recordsAffected: 7,
+  });
+  const runs = db.getRuns("j");
+  assert.equal(runs.length, 1);
+  assert.equal(runs[0].status, "success");
+  assert.equal(runs[0].recordsAffected, 7);
+  assert.equal(runs[0].durationMs, 123);
+  assert.equal(runs[0].startedAt, "2026-01-01T00:00:00.000Z");
+  assert.ok(runs[0].finishedAt, "finishedAt must be set for a completed run");
+});
+
+test("recordRun: running status leaves finishedAt NULL", () => {
+  db.upsertJob({
+    id: "j",
+    type: "interval",
+    cron: null,
+    intervalMs: 1000,
+    enabled: true,
+    envFlag: null,
+    config: {},
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+  });
+  db.recordRun("j", "running", { startedAt: "2026-01-01T00:00:00.000Z" });
+  const runs = db.getRuns("j");
+  assert.equal(runs[0].status, "running");
+  assert.equal(runs[0].finishedAt, null);
+});
+
+test("pruneRuns: count dimension keeps the most recent 100 of 150", () => {
+  db.upsertJob({
+    id: "j",
+    type: "interval",
+    cron: null,
+    intervalMs: 1000,
+    enabled: true,
+    envFlag: null,
+    config: {},
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+  });
+  // Insert 150 rows, oldest first, 1s apart.
+  for (let i = 0; i < 150; i++) {
+    const t = new Date(Date.UTC(2026, 0, 1, 0, 0, i)).toISOString();
+    db.recordRun("j", "success", { startedAt: t, durationMs: 1 });
+  }
+  db.pruneRuns("j", 100, 30);
+  const runs = db.getRuns("j", 200);
+  assert.equal(runs.length, 100);
+  // The oldest surviving row is the 50th (i=50); rows i=0..49 are pruned.
+  const oldestSec = new Date(runs[runs.length - 1].startedAt).getUTCSeconds();
+  assert.equal(oldestSec, 50);
+});
+
+test("pruneRuns: time dimension deletes rows older than 30 days (once outside recent-100)", () => {
+  // Dual-dimension semantics: a row is deleted only when it is BOTH outside the
+  // recent-100 window AND older than maxDays. So we insert 100 recent rows to push
+  // the 40-day-old row out of the recent-100, where age then prunes it.
+  db.upsertJob({
+    id: "j",
+    type: "interval",
+    cron: null,
+    intervalMs: 1000,
+    enabled: true,
+    envFlag: null,
+    config: {},
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+  });
+  const now = Date.now();
+  db.recordRun("j", "success", {
+    startedAt: new Date(now - 40 * 86_400_000).toISOString(),
+    durationMs: 1,
+  });
+  // 100 genuinely-recent rows (hourly, within the last ~4 days) fill the recent-100 window.
+  for (let i = 0; i < 100; i++) {
+    db.recordRun("j", "success", {
+      startedAt: new Date(now - i * 3_600_000).toISOString(),
+      durationMs: 1,
+    });
+  }
+  db.pruneRuns("j", 100, 30);
+  const runs = db.getRuns("j", 200);
+  // The 40-day-old row is outside recent-100 and >30 days -> deleted.
+  // Of the 100 recent rows, those aged 31..99 days are also pruned (outside recent-100? no -
+  // they're within recent-100 by rank). Only the single 40-day outlier is deleted.
+  const hasOld = runs.some((r) => new Date(r.startedAt).getTime() < now - 30 * 86_400_000);
+  assert.equal(hasOld, false, "no run older than 30 days should survive");
+  assert.equal(runs.length, 100);
+});
+
+test("pruneRuns: dual dimension - old rows inside the recent-100 are kept", () => {
+  db.upsertJob({
+    id: "j",
+    type: "interval",
+    cron: null,
+    intervalMs: 1000,
+    enabled: true,
+    envFlag: null,
+    config: {},
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+  });
+  const now = Date.now();
+  // 150 rows total: the 50 oldest are >30 days ago, the newest 100 are recent.
+  for (let i = 0; i < 150; i++) {
+    // Row i=0 oldest. Newest 100 (i=50..149) are recent; oldest 50 (i=0..49) are 40 days old.
+    const ageDays = i < 50 ? 40 : 1;
+    const t = new Date(now - ageDays * 86_400_000 - (149 - i) * 1000).toISOString();
+    db.recordRun("j", "success", { startedAt: t, durationMs: 1 });
+  }
+  db.pruneRuns("j", 100, 30);
+  const runs = db.getRuns("j", 200);
+  // The 50 rows older than 30 days are all outside the recent-100 -> deleted.
+  // The 100 recent rows survive.
+  assert.equal(runs.length, 100);
+});
+
+test("cleanupOrphanedRuns: fixes only running rows past the timeout", () => {
+  db.upsertJob({
+    id: "j",
+    type: "interval",
+    cron: null,
+    intervalMs: 1000,
+    enabled: true,
+    envFlag: null,
+    config: {},
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+  });
+  const now = Date.now();
+  // Orphaned: running, started 10 minutes ago.
+  db.recordRun("j", "running", { startedAt: new Date(now - 10 * 60_000).toISOString() });
+  // Fresh: running, started now - must NOT be touched.
+  db.recordRun("j", "running", { startedAt: new Date(now).toISOString() });
+
+  db.cleanupOrphanedRuns(5);
+  const runs = db.getRuns("j", 200);
+  const orphaned = runs.find((r) => new Date(r.startedAt).getTime() <= now - 10 * 60_000);
+  const fresh = runs.find((r) => new Date(r.startedAt).getTime() > now - 60_000);
+  assert.equal(orphaned.status, "failure");
+  assert.equal(orphaned.errorMessage, "orphaned: exceeded timeout");
+  assert.equal(fresh.status, "running");
+});

tests/unit/lib/jobRegistry/registry.test.ts

@@ -0,0 +1,594 @@
+/** JobRegistry runtime tests . Uses real timers + isolated temp DB. */
+
+// Access private internals (avoids `as any`).
+type TestRegistry = JobRegistry & {
+  timers: Map<string, unknown>;
+  cronFailCount: Map<string, number>;
+};
+function regInternals(reg: JobRegistry): TestRegistry {
+  return reg as TestRegistry;
+}
+
+import test from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+
+import type { JobDefinition } from "@/lib/jobRegistry/core.ts";
+
+const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-jr-rt-"));
+process.env.DATA_DIR = TEST_DATA_DIR;
+process.env.DISABLE_SQLITE_AUTO_BACKUP = "true";
+
+const core = await import("@/lib/db/core.ts");
+const index = await import("@/lib/jobRegistry/index.ts");
+const { getJobRegistry, __resetJobRegistry } = index;
+
+let nowIso: string;
+function def(
+  id: string,
+  handler: () => Promise<{ success: boolean; recordsAffected?: number; error?: string }>,
+  over: Record<string, unknown> = {}
+): JobDefinition {
+  return {
+    id,
+    type: "interval",
+    cron: null,
+    intervalMs: 100000,
+    enabled: true,
+    envFlag: null,
+    config: {},
+    createdAt: nowIso,
+    updatedAt: nowIso,
+    handler,
+    ...over,
+  } as JobDefinition;
+}
+
+function resetAll() {
+  // Stop any live timers on the current singleton before wiping the DB, otherwise
+  // an in-flight safeRun from the previous test writes to a closed/wiped DB (FK error).
+  try {
+    getJobRegistry().stopAll();
+  } catch {
+    // no singleton yet on first run
+  }
+  __resetJobRegistry();
+  core.resetDbInstance();
+  fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true });
+  fs.mkdirSync(TEST_DATA_DIR, { recursive: true });
+}
+
+test.beforeEach(() => {
+  nowIso = new Date().toISOString();
+  resetAll();
+});
+
+test.after(() => {
+  core.resetDbInstance();
+  fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true });
+});
+
+test("register + start (interval) fires handler immediately", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  reg.register(
+    def("j", async () => {
+      calls++;
+      return { success: true, recordsAffected: 0 };
+    })
+  );
+  reg.start("j");
+  await new Promise((r) => setTimeout(r, 30));
+  assert.equal(calls, 1, "interval fires once on start");
+  reg.stop("j");
+});
+
+test("re-entrancy: second tick skipped while handler is running", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  let release: (v: void) => void = () => {};
+  const gate = new Promise<void>((res) => (release = res));
+  reg.register(
+    def(
+      "j",
+      async () => {
+        calls++;
+        await gate;
+        return { success: true };
+      },
+      { intervalMs: 30 }
+    )
+  );
+  reg.start("j"); // fires immediately, blocks on gate
+  await new Promise((r) => setTimeout(r, 10));
+  assert.equal(calls, 1);
+  // ~100ms passes; setInterval would tick again but must be skipped (running guard).
+  await new Promise((r) => setTimeout(r, 120));
+  assert.equal(calls, 1, "must not re-enter while handler runs");
+  release();
+  await new Promise((r) => setTimeout(r, 30));
+  reg.stop("j");
+});
+
+test("cron nextTick: handler fires via recursive setTimeout", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  reg.register(
+    def(
+      "j",
+      async () => {
+        calls++;
+        return { success: true };
+      },
+      {
+        type: "cron",
+        cron: "* * * * * *", // every second (6-field) for fast deterministic test
+        intervalMs: null,
+        config: { timezone: "UTC" },
+      }
+    )
+  );
+  reg.start("j");
+  await new Promise((r) => setTimeout(r, 1400));
+  reg.stop("j");
+  assert.ok(calls >= 1, `cron handler should fire at least once, got ${calls}`);
+});
+
+test("runNow: manual trigger starts a disabled job's run", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  reg.register(
+    def("j", async () => {
+      calls++;
+      return { success: true };
+    })
+  );
+  const res = await reg.runNow("j");
+  assert.deepEqual(res, { started: true });
+  await new Promise((r) => setTimeout(r, 30));
+  assert.equal(calls, 1);
+});
+
+test("runNow: disabled job returns reason=disabled", async () => {
+  const reg = getJobRegistry();
+  reg.register(def("j", async () => ({ success: true }), { enabled: false }));
+  const res = await reg.runNow("j");
+  assert.deepEqual(res, { started: false, reason: "disabled" });
+});
+
+test("runNow: unknown job returns reason=not_found", async () => {
+  const reg = getJobRegistry();
+  const res = await reg.runNow("nope");
+  assert.deepEqual(res, { started: false, reason: "not_found" });
+});
+
+test("runNow: queue depth=1, coalesces concurrent triggers then re-runs", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  let release: (v: void) => void = () => {};
+  const gate = new Promise<void>((res) => (release = res));
+  reg.register(
+    def("j", async () => {
+      calls++;
+      await gate;
+      return { success: true };
+    })
+  );
+  const first = await reg.runNow("j"); // starts, blocks on gate
+  assert.deepEqual(first, { started: true });
+  // While running, a second runNow should queue (depth=1) and resolve after.
+  const queued = reg.runNow("j");
+  // The queued promise must not pile up: only one extra fire after release.
+  release();
+  const queuedRes = await queued;
+  await new Promise((r) => setTimeout(r, 30));
+  assert.equal(calls, 2, "initial + one queued re-run = 2 fires, no pile-up");
+  assert.deepEqual(queuedRes, { started: true });
+});
+
+test("runNow: env gate blocks when env explicitly disabled", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  reg.register(
+    def("j", async () => ({ success: true }), { envFlag: "OMNIROUTE_TEST_JOB_ENABLED" })
+  );
+  process.env.OMNIROUTE_TEST_JOB_ENABLED = "0";
+  const res = await reg.runNow("j");
+  delete process.env.OMNIROUTE_TEST_JOB_ENABLED;
+  assert.deepEqual(res, { started: false, reason: "env_disabled" });
+  assert.equal(calls, 0);
+});
+
+test("setEnabled(false) stops timer; handler no longer fires", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  reg.register(
+    def(
+      "j",
+      async () => {
+        calls++;
+        return { success: true };
+      },
+      { intervalMs: 40 }
+    )
+  );
+  reg.start("j");
+  await new Promise((r) => setTimeout(r, 25));
+  reg.setEnabled("j", false);
+  await new Promise((r) => setTimeout(r, 120));
+  reg.stop("j");
+  assert.equal(calls, 1, "only the immediate fire before disable");
+});
+
+test("setEnabled(true) restarts a stopped job", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  reg.register(
+    def(
+      "j",
+      async () => {
+        calls++;
+        return { success: true };
+      },
+      { intervalMs: 40 }
+    )
+  );
+  reg.setEnabled("j", true);
+  await new Promise((r) => setTimeout(r, 130));
+  reg.stop("j");
+  assert.ok(calls >= 2, `expected repeated fires, got ${calls}`);
+});
+
+test("envFlag gate generic: unset env fires (defaultWhenUnset=true)", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  reg.register(
+    def(
+      "j",
+      async () => {
+        calls++;
+        return { success: true };
+      },
+      {
+        type: "cron",
+        cron: "* * * * * *",
+        intervalMs: null,
+        envFlag: "OMNIROUTE_GENERIC_JOB_ENABLED", // unset -> default true -> fires
+        config: { timezone: "UTC" },
+      }
+    )
+  );
+  reg.start("j");
+  await new Promise((r) => setTimeout(r, 1400));
+  reg.stop("j");
+  assert.ok(calls >= 1, "unset env with default=true should fire");
+});
+
+test("envFlag gate warmup: unset env does NOT fire (envDefault=false)", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  reg.register(
+    def("j", async () => ({ success: true }), {
+      type: "cron",
+      cron: "* * * * * *",
+      intervalMs: null,
+      envFlag: "OMNIROUTE_WARMUP_ENABLED",
+      config: { timezone: "UTC", envDefault: false },
+    })
+  );
+  reg.start("j");
+  await new Promise((r) => setTimeout(r, 1400));
+  reg.stop("j");
+  assert.equal(calls, 0, "unset env with envDefault=false must not fire");
+});
+
+test("startAll: registers + starts all enabled jobs; throws if no handlers", async () => {
+  const reg = getJobRegistry();
+  await assert.rejects(() => reg.startAll(), /No handlers registered/);
+
+  let a = 0;
+  let b = 0;
+  reg.register(
+    def("a", async () => {
+      a++;
+      return { success: true, recordsAffected: 0 };
+    })
+  );
+  reg.register(
+    def("b", async () => {
+      b++;
+      return { success: true, recordsAffected: 0 };
+    })
+  );
+  await reg.startAll();
+  await new Promise((r) => setTimeout(r, 30));
+  assert.ok(a >= 1, "job a started");
+  assert.ok(b >= 1, "job b started");
+  reg.stopAll();
+});
+
+test("startAll isolation: a missing handler skips that job, starts the rest", async () => {
+  const reg = getJobRegistry();
+  let a = 0;
+  reg.register(
+    def("a", async () => {
+      a++;
+      return { success: true, recordsAffected: 0 };
+    })
+  );
+  // Seed a job in DB with no handler registered in the registry.
+  core.getDbInstance();
+  const { upsertJob } = await import("@/lib/db/jobRegistryDb.ts");
+  upsertJob({
+    id: "orphan",
+    type: "interval",
+    cron: null,
+    intervalMs: 1000,
+    enabled: true,
+    envFlag: null,
+    config: {},
+    createdAt: nowIso,
+    updatedAt: nowIso,
+  });
+  await reg.startAll();
+  await new Promise((r) => setTimeout(r, 30));
+  assert.ok(a >= 1, "registered job a still starts despite orphan in DB");
+  reg.stopAll();
+});
+
+test("startAll isolation: one job whose start() throws does not abort the rest", async () => {
+  const reg = getJobRegistry();
+  let a = 0;
+  let c = 0;
+  for (const [id, counter] of [
+    ["a", () => a++],
+    ["boom", () => 0],
+    ["c", () => c++],
+  ] as Array<[string, () => number]>) {
+    reg.register(
+      def(id, async () => {
+        counter();
+        return { success: true, recordsAffected: 0 };
+      })
+    );
+  }
+
+  // start() reads the DB (getJob), so it can throw for one job while the others
+  // are fine. Nothing in the registry catches per-job, so this asserts the loop
+  // in startAll() isolates it.
+  const realStart = reg.start.bind(reg);
+  reg.start = (jobId: string): void => {
+    if (jobId === "boom") throw new Error("start blew up for boom");
+    realStart(jobId);
+  };
+
+  try {
+    await reg.startAll();
+  } finally {
+    reg.start = realStart;
+  }
+
+  await new Promise((r) => setTimeout(r, 30));
+  assert.ok(a >= 1, "job registered before the throwing one still started");
+  assert.ok(c >= 1, "job registered after the throwing one still started");
+  reg.stopAll();
+});
+
+test("error isolation: handler throw records failure and survives", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  reg.register(
+    def("j", async () => {
+      calls++;
+      if (calls === 1) throw new Error("boom at /some/file.ts");
+      return { success: true };
+    })
+  );
+  await reg.runNow("j");
+  await new Promise((r) => setTimeout(r, 30));
+  const runs = reg.getRuns("j");
+  assert.equal(runs.length, 1);
+  assert.equal(runs[0].status, "failure");
+  // Error message must be sanitized - no stack path leaked.
+  assert.ok(!runs[0].errorMessage?.includes("at /"), "error must be sanitized");
+});
+
+test("getRuns returns recent records newest-first", async () => {
+  const reg = getJobRegistry();
+  reg.register(def("j", async () => ({ success: true, recordsAffected: 0 })));
+  await reg.runNow("j");
+  await reg.runNow("j");
+  await new Promise((r) => setTimeout(r, 30));
+  const runs = reg.getRuns("j");
+  assert.equal(runs.length, 2);
+  assert.ok(
+    new Date(runs[0].startedAt).getTime() >= new Date(runs[1].startedAt).getTime(),
+    "newest first"
+  );
+});
+
+test("listJobs returns registered jobs with handlers", () => {
+  const reg = getJobRegistry();
+  reg.register(def("j", async () => ({ success: true })));
+  const jobs = reg.listJobs();
+  const j = jobs.find((x) => x.id === "j");
+  assert.ok(j, "registered job j should appear in listJobs");
+  assert.equal(typeof j!.handler, "function");
+});
+
+test("cron: invalid expression stops re-scheduling after MAX_CRON_PARSE_FAILURES", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  reg.register(
+    def(
+      "badcron",
+      async () => {
+        calls++;
+        return { success: true };
+      },
+      {
+        type: "cron",
+        cron: "not-a-valid-cron-expression",
+        intervalMs: null,
+        config: { timezone: "UTC" },
+      }
+    )
+  );
+  reg.start("badcron");
+  const failCount = regInternals(reg).cronFailCount.get("badcron") ?? 0;
+  assert.ok(failCount >= 1, `parse should fail, got failCount=${failCount}`);
+  assert.equal(calls, 0, "invalid cron handler should never fire");
+  reg.stop("badcron");
+});
+
+test("runNow: queued trigger waits for current run then re-fires", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  let release: (v: void) => void = () => {};
+  const gate = new Promise<void>((res) => (release = res));
+  reg.register(
+    def("slow", async () => {
+      calls++;
+      await gate;
+      return { success: true };
+    })
+  );
+  const first = await reg.runNow("slow");
+  assert.deepEqual(first, { started: true });
+  const queued = reg.runNow("slow");
+  release();
+  const queuedRes = await queued;
+  await new Promise((r) => setTimeout(r, 30));
+  assert.equal(calls, 2, "initial + one queued re-run = 2 fires");
+  assert.deepEqual(queuedRes, { started: true });
+});
+
+test("dispose clears timers and cron failure counts", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  reg.register(
+    def(
+      "j",
+      async () => {
+        calls++;
+        return { success: true };
+      },
+      { intervalMs: 30 }
+    )
+  );
+  reg.start("j");
+  await new Promise((r) => setTimeout(r, 10));
+  assert.ok(regInternals(reg).timers.has("j"));
+  reg.dispose();
+  assert.equal(regInternals(reg).timers.size, 0);
+  assert.equal(regInternals(reg).cronFailCount.size, 0);
+  const callsBefore = calls;
+  await new Promise((r) => setTimeout(r, 100));
+  assert.equal(calls, callsBefore, "disposed timers must not fire");
+});
+
+test("cronGetter: re-reads cron on each fire", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  let cronExpr = "*/2 * * * * *";
+  reg.register({
+    id: "dynamic",
+    type: "cron",
+    cron: cronExpr,
+    intervalMs: null,
+    enabled: true,
+    envFlag: null,
+    config: { timezone: "UTC" },
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+    handler: async () => {
+      calls++;
+      return { success: true };
+    },
+    cronGetter: () => cronExpr,
+  });
+  reg.start("dynamic");
+  await new Promise((r) => setTimeout(r, 2500));
+  const callsAfterInitial = calls;
+  assert.ok(callsAfterInitial >= 1, `should fire with initial cron, got ${callsAfterInitial}`);
+  cronExpr = "0 0 1 1 * 2099";
+  await new Promise((r) => setTimeout(r, 3000));
+  reg.stop("dynamic");
+  const extraFires = calls - callsAfterInitial;
+  assert.ok(extraFires <= 1, `cronGetter change should stop fires, got ${extraFires} extra`);
+});
+
+test("cronFailCount: resets after successful parse", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  reg.register({
+    id: "recover",
+    type: "cron",
+    cron: "not-a-cron",
+    intervalMs: null,
+    enabled: true,
+    envFlag: null,
+    config: { timezone: "UTC" },
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+    handler: async () => {
+      calls++;
+      return { success: true };
+    },
+  });
+  reg.start("recover");
+  const failCountAfterFail = regInternals(reg).cronFailCount.get("recover") ?? 0;
+  assert.ok(failCountAfterFail >= 1, `parse should fail, got failCount=${failCountAfterFail}`);
+  reg.register({
+    id: "recover",
+    type: "cron",
+    cron: "* * * * * *",
+    intervalMs: null,
+    enabled: true,
+    envFlag: null,
+    config: { timezone: "UTC" },
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+    handler: async () => {
+      calls++;
+      return { success: true };
+    },
+    cronGetter: () => "* * * * * *",
+  });
+  reg.stop("recover");
+  reg.start("recover");
+  const failCountAfterRecovery = regInternals(reg).cronFailCount.get("recover") ?? 0;
+  assert.equal(failCountAfterRecovery, 0, "successful parse should reset failure count");
+  reg.stop("recover");
+});
+
+test("cronGetter: throws falls back to static cron", async () => {
+  const reg = getJobRegistry();
+  let calls = 0;
+  let shouldThrow = true;
+  reg.register({
+    id: "throwing",
+    type: "cron",
+    cron: "* * * * * *",
+    intervalMs: null,
+    enabled: true,
+    envFlag: null,
+    config: { timezone: "UTC" },
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+    handler: async () => {
+      calls++;
+      return { success: true };
+    },
+    cronGetter: () => {
+      if (shouldThrow) throw new Error("transient failure");
+      return "* * * * * *";
+    },
+  });
+  reg.start("throwing");
+  await new Promise((r) => setTimeout(r, 1100));
+  assert.ok(calls >= 1, "job should fire with static cron fallback when cronGetter throws");
+  reg.stop("throwing");
+});

tests/unit/lib/jobRegistry/timeUtils.test.ts

@@ -0,0 +1,48 @@
+/**
+ * Tests for jobRegistry timeUtils (convertToTimeZone).
+ *
+ * Verifies wall-clock conversion across IANA timezones, date-boundary crossing,
+ * and DST handling (the conversion itself is DST-naive; cron-parser owns DST on the
+ * scheduling side - we only require the wall clock to read correctly).
+ */
+
+import test from "node:test";
+import assert from "node:assert/strict";
+import { convertToTimeZone } from "@/lib/jobRegistry/timeUtils.ts";
+
+test("convertToTimeZone: UTC noon -> Pacific standard time (UTC-8)", () => {
+  // 2026-01-15T12:00:00Z -> PST is UTC-8 -> 04:00 same day
+  const utc = new Date("2026-01-15T12:00:00Z");
+  const pt = convertToTimeZone(utc, "America/Los_Angeles");
+  assert.equal(pt.getHours(), 4);
+  assert.equal(pt.getDate(), 15);
+});
+
+test("convertToTimeZone: UTC noon -> Pacific daylight time (UTC-7)", () => {
+  // 2026-07-15T12:00:00Z -> PDT is UTC-7 -> 05:00 same day
+  const utc = new Date("2026-07-15T12:00:00Z");
+  const pt = convertToTimeZone(utc, "America/Los_Angeles");
+  assert.equal(pt.getHours(), 5);
+});
+
+test("convertToTimeZone: crosses date boundary backward (UTC -> Asia/Tokyo, UTC+9)", () => {
+  // 2026-01-15T20:00:00Z -> JST +9 -> 05:00 NEXT day
+  const utc = new Date("2026-01-15T20:00:00Z");
+  const jst = convertToTimeZone(utc, "Asia/Tokyo");
+  assert.equal(jst.getHours(), 5);
+  assert.equal(jst.getDate(), 16);
+});
+
+test("convertToTimeZone: UTC identity zone returns same wall clock", () => {
+  const utc = new Date("2026-03-01T13:45:00Z");
+  const same = convertToTimeZone(utc, "UTC");
+  assert.equal(same.getHours(), 13);
+  assert.equal(same.getMinutes(), 45);
+});
+
+test("convertToTimeZone: Europe/London winter (UTC+0) vs summer (UTC+1 BST)", () => {
+  const winter = convertToTimeZone(new Date("2026-01-15T12:00:00Z"), "Europe/London");
+  assert.equal(winter.getHours(), 12);
+  const summer = convertToTimeZone(new Date("2026-07-15T12:00:00Z"), "Europe/London");
+  assert.equal(summer.getHours(), 13);
+});