diff --git a/packages/jimmy/src/cron/__tests__/catchup.test.ts b/packages/jimmy/src/cron/__tests__/catchup.test.ts index c1bb73ee..e01afcf8 100644 --- a/packages/jimmy/src/cron/__tests__/catchup.test.ts +++ b/packages/jimmy/src/cron/__tests__/catchup.test.ts @@ -7,6 +7,7 @@ import { readCheckpoint, writeCheckpoint, lastRunAtFromDisk, + mostRecentRun, } from "../catchup.js"; import type { CronJob } from "../../shared/types.js"; @@ -173,6 +174,48 @@ describe("checkpoint persistence", () => { }); }); +describe("mostRecentRun", () => { + it("returns the in-memory start when the disk log is empty", () => { + // An on-time fire still running: nothing on disk yet, but a start is tracked. + expect(mostRecentRun(null, ms("2026-06-03T01:00:00Z"))).toBe( + ms("2026-06-03T01:00:00Z"), + ); + }); + + it("returns the disk run when there is no in-memory start", () => { + // Fresh process (restart): in-memory map empty, fall back to the run-log. + expect(mostRecentRun(ms("2026-06-03T01:00:00Z"), null)).toBe( + ms("2026-06-03T01:00:00Z"), + ); + }); + + it("returns null when both are absent", () => { + expect(mostRecentRun(null, null)).toBeNull(); + }); + + it("prefers the more recent of the two", () => { + const older = ms("2026-06-02T01:00:00Z"); + const newer = ms("2026-06-03T01:00:00Z"); + expect(mostRecentRun(older, newer)).toBe(newer); + expect(mostRecentRun(newer, older)).toBe(newer); + }); + + it("dedups a long on-time fire whose disk log is still yesterday's run", () => { + // The actual production bug: yesterday's completion on disk, today's fire + // started but not yet logged. The merged value must clear the dedup gate so + // computeMissedFires does NOT replay today's slot. + const yesterdayOnDisk = ms("2026-06-02T01:00:03Z"); + const startedToday = ms("2026-06-03T01:00:01Z"); + const merged = mostRecentRun(yesterdayOnDisk, startedToday); + const { replay } = computeMissedFires([job({})], { + ...DEFAULTS, + lastCheck: ms("2026-06-03T00:30:00Z"), + lastRunAt: () => merged, + }); + expect(replay).toHaveLength(0); + }); +}); + describe("lastRunAtFromDisk", () => { it("returns the timestamp of the last run-log entry", () => { const dir = fs.mkdtempSync(path.join(os.tmpdir(), "runs-")); diff --git a/packages/jimmy/src/cron/__tests__/precheck.test.ts b/packages/jimmy/src/cron/__tests__/precheck.test.ts new file mode 100644 index 00000000..8af47892 --- /dev/null +++ b/packages/jimmy/src/cron/__tests__/precheck.test.ts @@ -0,0 +1,64 @@ +import { describe, it, expect } from "vitest"; +import { runPrecheck, DEFAULT_PRECHECK_TIMEOUT_MS } from "../precheck.js"; + +// Pin cwd to a real dir: vitest sets JINN_HOME to an uncreated tmp path, and the +// default cwd would otherwise spawn-fail. Production cwd is the real ~/.jinn. +const run = (pc: Parameters[0]) => runPrecheck(pc, { cwd: process.cwd() }); + +describe("runPrecheck — exit-code contract", () => { + it("decision=proceed when the command exits 0", async () => { + const r = await run({ command: "exit 0", skipExitCodes: [10] }); + expect(r.decision).toBe("proceed"); + expect(r.exitCode).toBe(0); + expect(r.timedOut).toBe(false); + }); + + it("decision=skip when the exit code is in skipExitCodes", async () => { + const r = await run({ command: "exit 10", skipExitCodes: [10, 20] }); + expect(r.decision).toBe("skip"); + expect(r.exitCode).toBe(10); + }); + + it("treats a second listed skip code (e.g. lock contention) as skip", async () => { + const r = await run({ command: "exit 20", skipExitCodes: [10, 20] }); + expect(r.decision).toBe("skip"); + expect(r.exitCode).toBe(20); + }); + + it("decision=error for a non-zero exit NOT in skipExitCodes (real failure not swallowed)", async () => { + // e.g. the jinn-watcher prefilter's 21 = wacli outage — must surface, not skip. + const r = await run({ command: "exit 21", skipExitCodes: [10, 20] }); + expect(r.decision).toBe("error"); + expect(r.exitCode).toBe(21); + }); + + it("with no skipExitCodes, ANY non-zero exit is an error (never skips blind)", async () => { + const r = await run({ command: "exit 1" }); + expect(r.decision).toBe("error"); + expect(r.exitCode).toBe(1); + }); + + it("decision=error and timedOut=true when the command exceeds timeoutMs", async () => { + const r = await run({ command: "sleep 5", skipExitCodes: [10], timeoutMs: 150 }); + expect(r.decision).toBe("error"); + expect(r.timedOut).toBe(true); + expect(r.exitCode).toBeNull(); + }); + + it("decision=error on invalid config (missing/empty command)", async () => { + const r = await run({ command: "" }); + expect(r.decision).toBe("error"); + expect(r.exitCode).toBeNull(); + expect(r.stderr).toMatch(/missing or empty command/); + }); + + it("captures stdout from the gate", async () => { + const r = await run({ command: "echo hello; exit 0" }); + expect(r.decision).toBe("proceed"); + expect(r.stdout).toMatch(/hello/); + }); + + it("default timeout constant is exported and sane", () => { + expect(DEFAULT_PRECHECK_TIMEOUT_MS).toBeGreaterThan(0); + }); +}); diff --git a/packages/jimmy/src/cron/__tests__/runner.test.ts b/packages/jimmy/src/cron/__tests__/runner.test.ts index f07554ca..69759968 100644 --- a/packages/jimmy/src/cron/__tests__/runner.test.ts +++ b/packages/jimmy/src/cron/__tests__/runner.test.ts @@ -23,6 +23,13 @@ vi.mock("../../shared/logger.js", () => ({ }, })); +// Stub the precheck gate (its own exit-code logic is unit-tested in precheck.test.ts); +// here we control its decision to assert how runCronJob branches. +vi.mock("../precheck.js", () => ({ runPrecheck: vi.fn() })); + +// Stub ops alert so a precheck_error doesn't try to reach Telegram. +vi.mock("../../shared/ops-alert.js", () => ({ opsAlert: vi.fn().mockResolvedValue(undefined) })); + function makeJob(overrides: Partial = {}): CronJob { return { id: "test-job", @@ -180,3 +187,91 @@ describe("runCronJob — latency alerting", () => { expect(alertMsg).toContain("failed"); }); }); + +describe("runCronJob — precheck gate", () => { + beforeEach(async () => { + // Clear accumulated call history on the module-mocked fns and re-arm async ones. + const { runPrecheck } = await import("../precheck.js"); + const { opsAlert } = await import("../../shared/ops-alert.js"); + const { appendRunLog } = await import("../jobs.js"); + (runPrecheck as any).mockReset(); + (opsAlert as any).mockReset().mockResolvedValue(undefined); + (appendRunLog as any).mockClear(); + }); + + const withPrecheck = (overrides = {}) => + makeJob({ precheck: { command: "x", skipExitCodes: [10] }, ...overrides }); + + it("spawns the session when precheck decides proceed", async () => { + const { runPrecheck } = await import("../precheck.js"); + (runPrecheck as any).mockResolvedValue({ decision: "proceed", exitCode: 0, signal: null, timedOut: false, stdout: "", stderr: "", durationMs: 5 }); + const sessionManager = makeMockSessionManager(0); + const connectors = new Map([["slack", makeMockConnector()]]); + + await runCronJob(withPrecheck(), sessionManager, makeConfig(), connectors); + + expect(sessionManager.route).toHaveBeenCalledTimes(1); + }); + + it("does NOT spawn a session on skip, and logs gated-skip (no alert)", async () => { + const { appendRunLog } = await import("../jobs.js"); + const { runPrecheck } = await import("../precheck.js"); + const { opsAlert } = await import("../../shared/ops-alert.js"); + (runPrecheck as any).mockResolvedValue({ decision: "skip", exitCode: 10, signal: null, timedOut: false, stdout: "", stderr: "", durationMs: 5 }); + const sessionManager = makeMockSessionManager(0); + const connector = makeMockConnector(); + const connectors = new Map([["slack", connector]]); + + await runCronJob(withPrecheck(), sessionManager, makeConfig(), connectors); + + expect(sessionManager.route).not.toHaveBeenCalled(); + expect(opsAlert).not.toHaveBeenCalled(); + expect(connector.sendMessage).not.toHaveBeenCalled(); + expect(appendRunLog).toHaveBeenCalledWith( + "test-job", + expect.objectContaining({ status: "gated-skip" }), + ); + }); + + it("does NOT spawn a session on precheck error, logs precheck_error AND ops-alerts", async () => { + const { appendRunLog } = await import("../jobs.js"); + const { runPrecheck } = await import("../precheck.js"); + const { opsAlert } = await import("../../shared/ops-alert.js"); + (runPrecheck as any).mockResolvedValue({ decision: "error", exitCode: 21, signal: null, timedOut: false, stdout: "", stderr: "wacli down", durationMs: 5 }); + const sessionManager = makeMockSessionManager(0); + const connectors = new Map([["slack", makeMockConnector()]]); + + await runCronJob(withPrecheck(), sessionManager, makeConfig(), connectors); + + expect(sessionManager.route).not.toHaveBeenCalled(); + expect(opsAlert).toHaveBeenCalledTimes(1); + expect(appendRunLog).toHaveBeenCalledWith( + "test-job", + expect.objectContaining({ status: "precheck_error" }), + ); + }); + + it("treats a precheck timeout as an error (no session)", async () => { + const { runPrecheck } = await import("../precheck.js"); + const { opsAlert } = await import("../../shared/ops-alert.js"); + (runPrecheck as any).mockResolvedValue({ decision: "error", exitCode: null, signal: "SIGTERM", timedOut: true, stdout: "", stderr: "", durationMs: 60000 }); + const sessionManager = makeMockSessionManager(0); + const connectors = new Map([["slack", makeMockConnector()]]); + + await runCronJob(withPrecheck(), sessionManager, makeConfig(), connectors); + + expect(sessionManager.route).not.toHaveBeenCalled(); + expect(opsAlert).toHaveBeenCalledTimes(1); + }); + + it("ignores precheck entirely for jobs without a precheck field (backward compat)", async () => { + const { runPrecheck } = await import("../precheck.js"); + const sessionManager = makeMockSessionManager(0); + const connectors = new Map([["slack", makeMockConnector()]]); + + await runCronJob(makeJob(), sessionManager, makeConfig(), connectors); + + expect(runPrecheck).not.toHaveBeenCalled(); + expect(sessionManager.route).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/jimmy/src/cron/catchup.ts b/packages/jimmy/src/cron/catchup.ts index d6e5f4c6..af7895c1 100644 --- a/packages/jimmy/src/cron/catchup.ts +++ b/packages/jimmy/src/cron/catchup.ts @@ -131,6 +131,22 @@ export function computeMissedFires( return { replay, tooOld }; } +/** + * Combine a job's on-disk last-run with its in-memory last-start (either may be + * null), returning the more recent. The in-memory start covers an on-time fire + * that is still running and has not yet written its completion entry to the + * run-log — without it, catch-up would replay a slot the live scheduler already + * fired. See runner.ts `lastStartedAt` for why the start is tracked in memory. + */ +export function mostRecentRun( + disk: number | null, + started: number | null, +): number | null { + if (disk === null) return started; + if (started === null) return disk; + return Math.max(disk, started); +} + /** Read the persisted last-sweep checkpoint (ms epoch), or null. */ export function readCheckpoint(file: string): number | null { try { diff --git a/packages/jimmy/src/cron/precheck.ts b/packages/jimmy/src/cron/precheck.ts new file mode 100644 index 00000000..8b3c12a3 --- /dev/null +++ b/packages/jimmy/src/cron/precheck.ts @@ -0,0 +1,90 @@ +import { execFile } from "node:child_process"; +import type { CronJob } from "../shared/types.js"; +import { JINN_HOME } from "../shared/paths.js"; + +export type PrecheckDecision = "proceed" | "skip" | "error"; + +export interface PrecheckResult { + decision: PrecheckDecision; + /** Numeric exit code, or null if the process never exited cleanly (timeout / spawn error). */ + exitCode: number | null; + /** Signal that terminated the process, if any (e.g. "SIGTERM" on timeout). */ + signal: string | null; + timedOut: boolean; + stdout: string; + stderr: string; + durationMs: number; +} + +export const DEFAULT_PRECHECK_TIMEOUT_MS = 60_000; +const MAX_OUTPUT_BYTES = 1024 * 1024; + +/** + * Run a cron job's `precheck` gate and classify the outcome. NEVER rejects — + * every failure mode resolves to a result with decision "error" so the caller + * can branch deterministically. + * + * Contract (see CronJob.precheck): exit 0 → proceed; exit ∈ skipExitCodes → + * skip; any other non-zero (or timeout / spawn error / missing command) → error. + */ +export function runPrecheck( + precheck: NonNullable, + opts?: { cwd?: string; env?: NodeJS.ProcessEnv }, +): Promise { + const start = Date.now(); + return new Promise((resolve) => { + const command = precheck?.command; + if (typeof command !== "string" || command.trim() === "") { + resolve({ + decision: "error", + exitCode: null, + signal: null, + timedOut: false, + stdout: "", + stderr: "precheck: missing or empty command", + durationMs: Date.now() - start, + }); + return; + } + const timeoutMs = precheck.timeoutMs ?? DEFAULT_PRECHECK_TIMEOUT_MS; + const skipExitCodes = precheck.skipExitCodes ?? []; + + execFile( + "/bin/bash", + ["-c", command], + { + cwd: opts?.cwd ?? JINN_HOME, + env: opts?.env ?? process.env, + timeout: timeoutMs, + maxBuffer: MAX_OUTPUT_BYTES, + }, + (err, stdout, stderr) => { + const durationMs = Date.now() - start; + const out = (stdout ?? "").toString(); + const errOut = (stderr ?? "").toString(); + + if (!err) { + resolve({ decision: "proceed", exitCode: 0, signal: null, timedOut: false, stdout: out, stderr: errOut, durationMs }); + return; + } + + const e = err as NodeJS.ErrnoException & { code?: number | string; killed?: boolean; signal?: string }; + const signal = e.signal ?? null; + // execFile sets killed=true and a termination signal when the timeout fires. + const timedOut = e.killed === true && (signal === "SIGTERM" || signal === "SIGKILL"); + const exitCode = typeof e.code === "number" ? e.code : null; + + if (timedOut) { + resolve({ decision: "error", exitCode: null, signal, timedOut: true, stdout: out, stderr: errOut, durationMs }); + return; + } + if (exitCode !== null && skipExitCodes.includes(exitCode)) { + resolve({ decision: "skip", exitCode, signal: null, timedOut: false, stdout: out, stderr: errOut, durationMs }); + return; + } + // Non-zero exit not on the skip list, or a spawn error (ENOENT etc.) → fail loud. + resolve({ decision: "error", exitCode, signal, timedOut: false, stdout: out, stderr: errOut, durationMs }); + }, + ); + }); +} diff --git a/packages/jimmy/src/cron/runner.ts b/packages/jimmy/src/cron/runner.ts index 1f86baac..bb77d0ad 100644 --- a/packages/jimmy/src/cron/runner.ts +++ b/packages/jimmy/src/cron/runner.ts @@ -8,6 +8,7 @@ import type { SessionManager } from "../sessions/manager.js"; import { resolveJobBudget, SESSION_BUDGET_STOP_PREFIX } from "../sessions/budget.js"; import { getSession } from "../sessions/registry.js"; import { opsAlert } from "../shared/ops-alert.js"; +import { runPrecheck } from "./precheck.js"; /** Provenance for a catch-up replay (a fire the host slept through). */ export interface CronRunMeta { @@ -18,6 +19,26 @@ export interface CronRunMeta { olderFiresSkipped?: number; } +/** + * ms epoch of each job's most recent *start*, recorded synchronously the moment + * runCronJob is invoked — i.e. before the (often multi-minute) LLM session it + * awaits. The run-log on disk is only written when the session COMPLETES, so a + * job that is still running has no fresh disk entry; without this, the ~5-min + * catch-up sweep sees a stale run-log and wrongly replays the slot the live + * scheduler already fired (every cron whose runtime outlasts the gap to the next + * sweep would double-fire). Catch-up dedup consults this alongside the disk log. + * + * Intentionally in-memory only: lost on process restart, which is correct — a + * fresh start must fall back to the on-disk run-log so genuinely-missed fires + * (slept/crashed through) still replay. + */ +const lastStartedAt = new Map(); + +/** ms epoch of a job's most recent in-process start, or null if none this run. */ +export function lastStartedAtMs(jobId: string): number | null { + return lastStartedAt.get(jobId) ?? null; +} + export async function runCronJob( job: CronJob, sessionManager: SessionManager, @@ -25,6 +46,7 @@ export async function runCronJob( connectors: Map, meta?: CronRunMeta, ): Promise { + lastStartedAt.set(job.id, Date.now()); const catchUpFields = meta?.catchUp ? { catchUp: true, @@ -35,6 +57,54 @@ export async function runCronJob( const startTime = Date.now(); logger.info(`Cron job "${job.name}" (${job.id}) starting`); + // Deterministic pre-gate: run a cheap shell check BEFORE creating the + // (expensive) LLM session, and skip the session entirely when there's no + // work. See CronJob.precheck for the exit-code contract. + if (job.precheck) { + const precheckStartedAt = new Date().toISOString(); + const pc = await runPrecheck(job.precheck); + if (pc.decision !== "proceed") { + const durationMs = Date.now() - startTime; + const isSkip = pc.decision === "skip"; + appendRunLog(job.id, { + timestamp: precheckStartedAt, + sessionKey: null, + sessionId: null, + status: isSkip ? "gated-skip" : "precheck_error", + durationMs, + precheck: { + exitCode: pc.exitCode, + timedOut: pc.timedOut, + durationMs: pc.durationMs, + stderr: pc.stderr.slice(0, 500), + }, + error: isSkip + ? null + : `precheck failed (exit ${pc.exitCode ?? "?"}${pc.timedOut ? ", timed out" : ""}): ${pc.stderr.slice(0, 200)}`, + ...catchUpFields, + resultPreview: null, + }); + if (isSkip) { + logger.info( + `Cron job "${job.name}" (${job.id}) gated-skip via precheck (exit ${pc.exitCode}) in ${durationMs}ms`, + ); + } else { + logger.error( + `Cron job "${job.name}" (${job.id}) precheck error (exit ${pc.exitCode ?? "none"}, timedOut=${pc.timedOut})`, + ); + // A precheck ERROR is an unexpected failure (not a normal no-work skip) — surface it. + await opsAlert( + `Cron "${job.name}" (${job.id}) precheck FAILED — exit ${pc.exitCode ?? "none"}${pc.timedOut ? " (timed out)" : ""}. ` + + `No session was spawned. stderr: ${pc.stderr.slice(0, 300) || "(none)"}`, + ).catch(() => {}); + } + return; + } + logger.debug( + `Cron job "${job.name}" (${job.id}) precheck passed (exit 0) in ${pc.durationMs}ms — proceeding`, + ); + } + const delivery = job.delivery || config.cron?.defaultDelivery; const cooSlug = config.portal?.portalName?.toLowerCase() || "jinn"; if (delivery && job.employee && job.employee !== cooSlug) { diff --git a/packages/jimmy/src/cron/scheduler.ts b/packages/jimmy/src/cron/scheduler.ts index b3303080..6835a47f 100644 --- a/packages/jimmy/src/cron/scheduler.ts +++ b/packages/jimmy/src/cron/scheduler.ts @@ -5,7 +5,7 @@ import type { JinnConfig, Connector, } from "../shared/types.js"; -import { runCronJob } from "./runner.js"; +import { runCronJob, lastStartedAtMs } from "./runner.js"; import { logger } from "../shared/logger.js"; import type { SessionManager } from "../sessions/manager.js"; import { loadJobs, saveJobs } from "./jobs.js"; @@ -14,6 +14,7 @@ import { readCheckpoint, writeCheckpoint, lastRunAtFromDisk, + mostRecentRun, } from "./catchup.js"; import { CRON_CATCHUP_STATE, CRON_RUNS } from "../shared/paths.js"; @@ -93,7 +94,10 @@ export async function catchUpMissed(now: number = Date.now()): Promise { maxLookbackMs: CATCHUP_MAX_LOOKBACK_MS, graceMs: CATCHUP_GRACE_MS, dedupSlopMs: CATCHUP_DEDUP_SLOP_MS, - lastRunAt: (id) => lastRunAtFromDisk(id, CRON_RUNS), + // Merge the on-disk run-log with the in-memory last-start so a still-running + // on-time fire (not yet logged to disk) is not mistaken for a missed slot. + lastRunAt: (id) => + mostRecentRun(lastRunAtFromDisk(id, CRON_RUNS), lastStartedAtMs(id)), }); const lookbackH = Math.round(CATCHUP_MAX_LOOKBACK_MS / 3_600_000); for (const t of tooOld) { diff --git a/packages/jimmy/src/shared/types.ts b/packages/jimmy/src/shared/types.ts index 853f7bd7..f5651303 100644 --- a/packages/jimmy/src/shared/types.ts +++ b/packages/jimmy/src/shared/types.ts @@ -216,6 +216,28 @@ export interface CronJob { * instead of being runlog-only. Default false = no alert. */ sideEffects?: boolean; }; + /** Optional deterministic gate run BEFORE any LLM session is created. Lets a + * cheap shell command decide whether the (expensive) session is worth + * spawning — e.g. a watcher whose prefilter usually finds no work. The + * command runs via `/bin/bash -c` in JINN_HOME with the gateway's env. + * + * Exit-code contract: + * - 0 → PROCEED (spawn the session as normal) + * - code ∈ skipExitCodes → gated-skip (no session, no alert, runlog only) + * - any other non-zero → precheck_error (no session, ops alert) — so a + * real failure the script exits non-zero on (e.g. + * a dependency outage) is NOT silently skipped + * - timeout → precheck_error + * `skipExitCodes` is required to get any skip behaviour; if omitted/empty, + * every non-zero exit is an error (fail loud, never skip blind). */ + precheck?: { + /** Shell command run before session creation (via `/bin/bash -c`). */ + command: string; + /** Max run time before the gate is killed and treated as an error. Default 60000. */ + timeoutMs?: number; + /** Exit codes that mean "no work — skip this fire" (gated-skip). */ + skipExitCodes?: number[]; + }; } export interface CronDelivery {