From 68bfa42b9bfcdba928b7fb7ab6a711196b709d9a Mon Sep 17 00:00:00 2001 From: Cameron Beeley Date: Wed, 24 Jun 2026 14:39:31 +0100 Subject: [PATCH] =?UTF-8?q?feat(cron):=20on-exit=20cron=20schedule=20kind?= =?UTF-8?q?=20=E2=80=94=20fire=20a=20job=20when=20a=20watched=20command=20?= =?UTF-8?q?exits?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an `on-exit` cron schedule kind: a job fires once when a watched command/process exits, via gateway ProcessSupervisor exit watchers. Covers CLI (`--on-exit`/`--on-exit-cwd`), tool/protocol schema, RPC list-filter, Control UI + macOS read-only display, SQLite round-trip, and origin-aware wake routing. Restart-safe one-shot (persists completion before firing); platform-aware shell; bounded watched-command execution. Squashed from 22 iterative commits for a clean rebase onto current main. --- DESIGN-cron-on-exit.md | 29 ++ .../OpenClaw/CronJobEditor+Helpers.swift | 18 + apps/macos/Sources/OpenClaw/CronModels.swift | 11 +- .../OpenClaw/CronSettings+Helpers.swift | 3 + packages/gateway-protocol/src/schema/cron.ts | 12 + scripts/proof-cron-on-exit.mts | 170 ++++++++ src/agents/tools/cron-tool-canonicalize.ts | 15 +- .../tools/cron-tool.flat-params.test.ts | 58 +++ src/agents/tools/cron-tool.test.ts | 31 ++ src/agents/tools/cron-tool.ts | 12 +- src/cli/cron-cli/register.cron-add.ts | 10 +- .../cron-cli/schedule-options.onexit.test.ts | 44 ++ src/cli/cron-cli/schedule-options.ts | 36 +- src/cli/cron-cli/shared.test.ts | 19 + src/cli/cron-cli/shared.ts | 4 + src/cron/normalize.test.ts | 34 ++ src/cron/normalize.ts | 31 +- src/cron/persisted-shape.onexit.test.ts | 37 ++ src/cron/persisted-shape.ts | 13 +- src/cron/schedule.test.ts | 9 + src/cron/schedule.ts | 6 + src/cron/service-contract.ts | 7 +- src/cron/service.ts | 14 +- src/cron/service/list-page-types.ts | 2 +- src/cron/service/ops.ts | 17 +- src/cron/store/row-codec.schedule.test.ts | 54 +++ src/cron/store/row-codec.ts | 26 +- src/cron/types.ts | 14 + src/gateway/cron-exit-watchers.test.ts | 390 ++++++++++++++++++ src/gateway/cron-exit-watchers.ts | 241 +++++++++++ src/gateway/server-cron-lazy.test.ts | 28 ++ src/gateway/server-cron-lazy.ts | 12 +- src/gateway/server-cron.test.ts | 118 +++++- src/gateway/server-cron.ts | 109 ++++- src/gateway/server-reload-handlers.test.ts | 73 +++- src/gateway/server-reload-handlers.ts | 2 + src/gateway/server-runtime-services.test.ts | 19 + src/gateway/server-runtime-services.ts | 2 + src/plugins/hook-types.ts | 5 + ui/src/ui/controllers/cron.test.ts | 97 +++++ ui/src/ui/controllers/cron.ts | 30 +- ui/src/ui/presenter.test.ts | 40 ++ ui/src/ui/presenter.ts | 5 + ui/src/ui/types.ts | 3 +- ui/src/ui/ui-types.ts | 4 +- 45 files changed, 1869 insertions(+), 45 deletions(-) create mode 100644 DESIGN-cron-on-exit.md create mode 100644 scripts/proof-cron-on-exit.mts create mode 100644 src/cli/cron-cli/schedule-options.onexit.test.ts create mode 100644 src/cron/persisted-shape.onexit.test.ts create mode 100644 src/cron/store/row-codec.schedule.test.ts create mode 100644 src/gateway/cron-exit-watchers.test.ts create mode 100644 src/gateway/cron-exit-watchers.ts create mode 100644 ui/src/ui/presenter.test.ts diff --git a/DESIGN-cron-on-exit.md b/DESIGN-cron-on-exit.md new file mode 100644 index 00000000000..8229f357c6b --- /dev/null +++ b/DESIGN-cron-on-exit.md @@ -0,0 +1,29 @@ +# feat(cron): `on-exit` schedule — fire a job when a watched command/process exits + +## Problem + +Event-driven wakes that start a fresh agent turn already work (the `wake`/`system event` RPC). But an agent cannot reliably arm "wake me when this command/process exits" itself: CLI backends run each turn as a supervisor-spawned **detached process group** that is `signalProcessTree(SIGTERM→SIGKILL)`'d at turn end (`src/process/supervisor/adapters/child.ts`, intentional, #71662). Any process the agent backgrounds via `exec` is in that tree and dies with the turn. The only escape (`setsid` + raw `node dist/entry.js system event …`) is hand-rolled, fragile, and observed to take down the host. Applies to **all** spawn-and-kill CLI backends (claude-cli verified), not the TLS proxy. + +## Design + +A new cron **schedule kind** `on-exit`, executed by a **gateway-supervisor-owned watcher** — independent of #83738 (rides the existing main-session cron run pipeline, not the manual wake path). + +- `CronSchedule` gains `{ kind: "on-exit"; command: string; cwd?: string }` (PID-watch variant deferred). +- `computeNextRunAtMs()` returns `undefined` for `on-exit` → the time-based timer never fires it. +- `createCronExitWatchers()` (extracted into `src/gateway/cron-exit-watchers.ts`, wired from `buildGatewayCronService` via `reconcileExitWatchers()` / `stopExitWatchers()`) owns the watcher lifecycle, backed by `getProcessSupervisor()`. It exposes `reconcile(jobs) / cancel(jobId) / cancelAll() / activeJobIds()`. + - On reconcile, each enabled `on-exit` job reserves a watcher slot synchronously (with an `armToken`), then spawns the command via `supervisor.spawn({ mode:"child", scopeKey:"cron-exit:", replaceExistingScope:true, argv:[shell.command, ...shell.argsFor(command)], cwd?, captureOutput:true })`. The shell is platform-aware via `resolveExitWatchShell()`: `cmd.exe /d /s /c` on Windows, `bash -lc` on POSIX. + - The watcher lives under the **gateway** supervisor tree, so per-turn CLI teardown never touches it. An async spawn/wait that loses ownership (job cancelled or re-armed for a changed command/cwd) is a no-op via the `armToken`/slot-identity check. + - `await run.wait()` → on exit, the one-shot completion is **persisted (the job disabled in the store) BEFORE the job fires**, fail-closed: if the store write or `run.wait()` rejects, the job does NOT fire — so a gateway restart cannot re-arm and double-fire the same exit. Only then does it fire via the existing cron run pipeline; the woken turn sees the exit code + last output lines. + - Job `remove`/`disable`, or a changed command/cwd → `reconcile()` cancels or re-arms the watcher (`cancel(jobId)`; `cancelAll()` on shutdown). +- Delivery to the originating conversation is the **existing** `executeMainSessionCronJob` path (`resolveMainSessionCronDeliveryContext`) — already correct on main; no dependency on #83738. + +## Reuse / no new delivery code + +Everything after "process exited" is the current cron run→system-event→delivery pipeline. The only new surface: the schedule kind, its validation, the watcher lifecycle, and the tool/schema plumbing to create such a job. + +## Out of scope + +- PID-watch (`{ kind:"on-exit"; pid }`) — follow-up. +- Re-arm/repeat on each exit — v1 is one-shot (job disables after firing, like a one-shot `at`). +- This PR **stacks on #83738** and reuses its origin-aware wake as the firing + mechanism; it adds only the process-exit _trigger_ (the supervisor watcher). diff --git a/apps/macos/Sources/OpenClaw/CronJobEditor+Helpers.swift b/apps/macos/Sources/OpenClaw/CronJobEditor+Helpers.swift index 4c05bb29916..9e72c25e83a 100644 --- a/apps/macos/Sources/OpenClaw/CronJobEditor+Helpers.swift +++ b/apps/macos/Sources/OpenClaw/CronJobEditor+Helpers.swift @@ -39,6 +39,12 @@ extension CronJobEditor { self.scheduleKind = .cron self.cronExpr = expr self.cronTz = tz ?? "" + case .onExit: + // on-exit jobs are CLI-managed and have no editor form yet; fall back to + // the cron form so the editor still OPENS instead of failing to compile. + // Saving an on-exit job from the editor should be gated to avoid rewriting + // it as cron — tracked as a macOS read-only-editor follow-up. + self.scheduleKind = .cron } switch job.payload { @@ -74,6 +80,18 @@ extension CronJobEditor { } func buildPayload() throws -> [String: AnyCodable] { + // Gate on-exit saves: the editor has no on-exit schedule form (it falls back to + // the cron form), so saving would rewrite the on-exit schedule as cron and + // corrupt the job. Block it until a read-only/native on-exit editor exists. + if let job, case .onExit = job.schedule { + throw NSError( + domain: "Cron", + code: 0, + userInfo: [ + NSLocalizedDescriptionKey: + "on-exit cron jobs can't be edited in the macOS app yet; manage them with the CLI.", + ]) + } let name = try self.requireName() let description = self.trimmed(self.description) let agentId = self.trimmed(self.agentId) diff --git a/apps/macos/Sources/OpenClaw/CronModels.swift b/apps/macos/Sources/OpenClaw/CronModels.swift index 8c6191155e0..09e4ee76b44 100644 --- a/apps/macos/Sources/OpenClaw/CronModels.swift +++ b/apps/macos/Sources/OpenClaw/CronModels.swift @@ -66,14 +66,16 @@ enum CronSchedule: Codable, Equatable { case at(at: String) case every(everyMs: Int, anchorMs: Int?) case cron(expr: String, tz: String?) + case onExit(command: String, cwd: String?) - enum CodingKeys: String, CodingKey { case kind, at, atMs, everyMs, anchorMs, expr, tz } + enum CodingKeys: String, CodingKey { case kind, at, atMs, everyMs, anchorMs, expr, tz, command, cwd } var kind: String { switch self { case .at: "at" case .every: "every" case .cron: "cron" + case .onExit: "on-exit" } } @@ -105,6 +107,10 @@ enum CronSchedule: Codable, Equatable { self = try .cron( expr: container.decode(String.self, forKey: .expr), tz: container.decodeIfPresent(String.self, forKey: .tz)) + case "on-exit": + self = try .onExit( + command: container.decode(String.self, forKey: .command), + cwd: container.decodeIfPresent(String.self, forKey: .cwd)) default: throw DecodingError.dataCorruptedError( forKey: .kind, @@ -125,6 +131,9 @@ enum CronSchedule: Codable, Equatable { case let .cron(expr, tz): try container.encode(expr, forKey: .expr) try container.encodeIfPresent(tz, forKey: .tz) + case let .onExit(command, cwd): + try container.encode(command, forKey: .command) + try container.encodeIfPresent(cwd, forKey: .cwd) } } diff --git a/apps/macos/Sources/OpenClaw/CronSettings+Helpers.swift b/apps/macos/Sources/OpenClaw/CronSettings+Helpers.swift index 873b0741e34..81dc97eb9e7 100644 --- a/apps/macos/Sources/OpenClaw/CronSettings+Helpers.swift +++ b/apps/macos/Sources/OpenClaw/CronSettings+Helpers.swift @@ -27,6 +27,9 @@ extension CronSettings { case let .cron(expr, tz): if let tz, !tz.isEmpty { return "cron \(expr) (\(tz))" } return "cron \(expr)" + case let .onExit(command, cwd): + if let cwd, !cwd.isEmpty { return "on exit: \(command) (cwd: \(cwd))" } + return "on exit: \(command)" } } diff --git a/packages/gateway-protocol/src/schema/cron.ts b/packages/gateway-protocol/src/schema/cron.ts index 4f191e02707..e0eb21aaf69 100644 --- a/packages/gateway-protocol/src/schema/cron.ts +++ b/packages/gateway-protocol/src/schema/cron.ts @@ -83,6 +83,7 @@ const CronJobsScheduleKindFilterSchema = Type.Union([ Type.Literal("at"), Type.Literal("every"), Type.Literal("cron"), + Type.Literal("on-exit"), ]); const CronJobsLastRunStatusFilterSchema = Type.Union([ Type.Literal("all"), @@ -223,6 +224,17 @@ export const CronScheduleSchema = Type.Union([ }, { additionalProperties: false }, ), + Type.Object( + { + // Event-driven trigger: fires once when the gateway-owned watcher running + // `command` exits. Survives per-turn CLI teardown (runs under the gateway + // ProcessSupervisor, not the turn process tree). + kind: Type.Literal("on-exit"), + command: NonEmptyString, + cwd: Type.Optional(NonEmptyString), + }, + { additionalProperties: false }, + ), ]); /** Full cron payload for new jobs. */ diff --git a/scripts/proof-cron-on-exit.mts b/scripts/proof-cron-on-exit.mts new file mode 100644 index 00000000000..e219279fbe9 --- /dev/null +++ b/scripts/proof-cron-on-exit.mts @@ -0,0 +1,170 @@ +// Live-proof harness for PR #92037 (cron `on-exit` schedule kind). +// +// Drives the REAL gateway exit-watcher (`createCronExitWatchers`) against the +// REAL ProcessSupervisor (`getProcessSupervisor`) with a REAL short-lived child +// command, and captures the actual arm -> exit -> persist-before-fire -> fire +// lifecycle. The persistCompletion/fireOnExit sinks mirror the real wiring in +// server-cron.ts (disable-before-fire; force-run after exit) and only RECORD/LOG. +// +// Run: pnpm exec tsx scripts/proof-cron-on-exit.mts +// +// All identifiers are synthetic. No real Telegram chat ids / session keys. + +import type { CronJob } from "../src/cron/types.js"; +import { + createCronExitWatchers, + resolveExitWatchShell, +} from "../src/gateway/cron-exit-watchers.js"; +import { getProcessSupervisor } from "../src/process/supervisor/index.js"; + +const isWin = process.platform === "win32"; +// Commands phrased for the shell the watcher actually resolves on this host +// (cmd.exe /d /s /c on Windows, bash -lc on POSIX). +const delayThenExit = (code: number) => + isWin ? `ping -n 3 127.0.0.1 > nul & exit ${code}` : `sleep 2; exit ${code}`; +const longRunning = () => (isWin ? `ping -n 31 127.0.0.1 > nul` : `sleep 30`); + +type FireEvent = { jobId: string; exitCode: number | null }; + +const events: { armed: string[]; persisted: string[]; fired: FireEvent[] } = { + armed: [], + persisted: [], + fired: [], +}; +// Monotonic call-order log so we can assert persist-before-fire directly +// (the watcher's fail-closed guarantee), not merely that both happened. +const order: string[] = []; + +const logger = { + info: (obj: unknown, msg?: string) => { + const o = obj as { jobId?: string; exitCode?: number | null; reason?: string }; + if (msg?.includes("watcher armed") && o.jobId) { + events.armed.push(o.jobId); + } + console.log(`[cron-exit] ${msg ?? ""} ${JSON.stringify(obj)}`); + }, + warn: (obj: unknown, msg?: string) => + console.log(`[cron-exit][warn] ${msg ?? ""} ${JSON.stringify(obj)}`), +}; + +const watchers = createCronExitWatchers({ + getProcessSupervisor, + // Real wiring disables the one-shot job in the store before firing. + persistCompletion: async (jobId) => { + events.persisted.push(jobId); + order.push(`persist:${jobId}`); + console.log(`[cron-exit] persistCompletion (job disabled, enabled=false) jobId=${jobId}`); + }, + // Real wiring routes this into cron.run(job.id, "force"). + fireOnExit: (job, exit) => { + events.fired.push({ jobId: job.id, exitCode: exit.exitCode }); + order.push(`fire:${job.id}`); + console.log(`[gateway/cron] cron.run force jobId=${job.id}`); + }, + logger, +}); + +function onExitJob(id: string, command: string, extra?: Partial): CronJob { + return { + id, + enabled: true, + schedule: { kind: "on-exit", command }, + sessionKey: `agent:main:telegram:direct:SYN:thread:SYN`, + payload: { text: `on-exit[${id}]` }, + ...extra, + } as unknown as CronJob; +} + +const sleep = (ms: number) => + new Promise((r) => { + setTimeout(r, ms); + }); +async function waitFor(pred: () => boolean, timeoutMs: number, pollMs = 100): Promise { + const start = Date.now(); + while (Date.now() - start < timeoutMs) { + if (pred()) { + return true; + } + await sleep(pollMs); + } + return pred(); +} + +let failures = 0; +function assert(label: string, cond: boolean): void { + console.log(` ${cond ? "PASS" : "FAIL"}: ${label}`); + if (!cond) { + failures++; + } +} + +async function run(): Promise { + const shell = resolveExitWatchShell(); + console.log(`=== PR #92037 on-exit live proof (real watcher + real ProcessSupervisor) ===`); + console.log( + `platform=${process.platform} shell=${shell.command} argv=${JSON.stringify(shell.argsFor(""))}`, + ); + + // Scenario A: arm a real watcher; the watched command runs ~2s then exits 7. + // Expect: armed -> exited -> persistCompletion BEFORE fire -> fire with exitCode 7. + console.log(`\n=== A. arm -> watched command exits (code 7) -> force-run fires ===`); + const a = onExitJob("onexit-A", delayThenExit(7)); + watchers.reconcile([a]); + assert( + "watcher is active immediately after reconcile", + watchers.activeJobIds().includes("onexit-A"), + ); + const aFired = await waitFor(() => events.fired.some((e) => e.jobId === "onexit-A"), 15000); + assert("watcher armed (logged) for onexit-A", events.armed.includes("onexit-A")); + assert("job fired after the command exited", aFired); + const aEvt = events.fired.find((e) => e.jobId === "onexit-A"); + assert("exit code 7 captured from the real child", aEvt?.exitCode === 7); + assert( + "persistCompletion ran BEFORE fire (fail-closed ordering)", + order.includes("persist:onexit-A") && + order.indexOf("persist:onexit-A") < order.indexOf("fire:onexit-A"), + ); + assert("fire routed to the cron force-run sink", aEvt?.jobId === a.id); + + // Scenario B: arm a long-running watcher, cancel before exit -> NO fire. + console.log(`\n=== B. arm -> cancel before exit -> no fire (revocation) ===`); + const b = onExitJob("onexit-B", longRunning()); + watchers.reconcile([b]); + await waitFor(() => events.armed.includes("onexit-B"), 8000); + assert("watcher armed for onexit-B", events.armed.includes("onexit-B")); + watchers.cancel("onexit-B"); + assert( + "watcher removed from active set after cancel", + !watchers.activeJobIds().includes("onexit-B"), + ); + await sleep(2500); + assert("cancelled watcher never fired", !events.fired.some((e) => e.jobId === "onexit-B")); + + // Scenario C: reconcile without the job cancels its watcher. + console.log(`\n=== C. reconcile-removal cancels the watcher ===`); + const c = onExitJob("onexit-C", longRunning()); + watchers.reconcile([c]); + await waitFor(() => watchers.activeJobIds().includes("onexit-C"), 8000); + assert("watcher active for onexit-C", watchers.activeJobIds().includes("onexit-C")); + watchers.reconcile([]); // job gone + assert("reconcile([]) cancelled onexit-C", !watchers.activeJobIds().includes("onexit-C")); +} + +async function main(): Promise { + try { + await run(); + } finally { + // Always tear down watchers so a thrown assertion can't leak the + // long-running ping/sleep children (B, C) until their 24h timeout. + watchers.cancelAll(); + await sleep(300); + } + console.log(`\n=== RESULT: ${failures === 0 ? "ALL PASS" : `${failures} FAILURE(S)`} ===`); + process.exit(failures === 0 ? 0 : 1); +} + +main().catch((err: unknown) => { + console.error("proof harness crashed:", err); + watchers.cancelAll(); + process.exit(1); +}); diff --git a/src/agents/tools/cron-tool-canonicalize.ts b/src/agents/tools/cron-tool-canonicalize.ts index 377aa55e1dd..c4804fbe87b 100644 --- a/src/agents/tools/cron-tool-canonicalize.ts +++ b/src/agents/tools/cron-tool-canonicalize.ts @@ -6,7 +6,7 @@ import { timestampMsToIsoString } from "@openclaw/normalization-core/number-coercion"; import { isRecord } from "../../utils.js"; -const CRON_SCHEDULE_KINDS = ["at", "every", "cron"] as const; +const CRON_SCHEDULE_KINDS = ["at", "every", "cron", "on-exit"] as const; const CRON_PAYLOAD_KINDS = ["systemEvent", "agentTurn"] as const; const CRON_FLAT_PAYLOAD_KEYS = [ "message", @@ -32,6 +32,8 @@ const CRON_FLAT_SCHEDULE_KEYS = [ "stagger", "staggerMs", "exact", + "command", + "cwd", ] as const; const CRON_RECOVERABLE_OBJECT_KEYS: ReadonlySet = new Set([ "name", @@ -54,7 +56,7 @@ const CRON_RECOVERABLE_OBJECT_KEYS: ReadonlySet = new Set([ ]); function isCronScheduleKind(value: unknown): value is (typeof CRON_SCHEDULE_KINDS)[number] { - return value === "at" || value === "every" || value === "cron"; + return typeof value === "string" && (CRON_SCHEDULE_KINDS as readonly string[]).includes(value); } function isCronPayloadKind(value: unknown): value is (typeof CRON_PAYLOAD_KINDS)[number] { @@ -178,7 +180,12 @@ function canonicalizeCronToolSchedule(value: Record): void { schedule.kind = "cron"; } - for (const key of ["anchorMs", "tz", "staggerMs"] as const) { + const movedCommand = moveDefinedField({ source: value, target: schedule, from: "command" }); + if (movedCommand && !isCronScheduleKind(schedule.kind)) { + schedule.kind = "on-exit"; + } + + for (const key of ["anchorMs", "tz", "staggerMs", "cwd"] as const) { hasSchedule = moveDefinedField({ source: value, target: schedule, from: key }) || hasSchedule; } hasSchedule = @@ -198,6 +205,8 @@ function canonicalizeCronToolSchedule(value: Record): void { schedule.kind = "every"; } else if (schedule.expr !== undefined) { schedule.kind = "cron"; + } else if (schedule.command !== undefined) { + schedule.kind = "on-exit"; } } diff --git a/src/agents/tools/cron-tool.flat-params.test.ts b/src/agents/tools/cron-tool.flat-params.test.ts index f2d64a7ee58..c9f404286b9 100644 --- a/src/agents/tools/cron-tool.flat-params.test.ts +++ b/src/agents/tools/cron-tool.flat-params.test.ts @@ -119,6 +119,64 @@ describe("cron tool flat-params", () => { }); }); + it("rejects flat on-exit schedule shorthand for add", async () => { + const tool = createCronTool(undefined, { callGatewayTool: callGatewayToolMock }); + + await expect( + tool.execute("call-flat-onexit-add", { + action: "add", + name: "rebuild on exit", + kind: "on-exit", + command: "pnpm build", + cwd: "/repo", + message: "rebuilt", + }), + ).rejects.toThrow("cron on-exit schedules cannot be created or edited"); + expect(callGatewayToolMock).not.toHaveBeenCalled(); + }); + + it("rejects flat command schedule shorthand for add", async () => { + const tool = createCronTool(undefined, { callGatewayTool: callGatewayToolMock }); + + await expect( + tool.execute("call-flat-onexit-infer", { + action: "add", + name: "watch build", + command: "make", + message: "done", + }), + ).rejects.toThrow("cron on-exit schedules cannot be created or edited"); + expect(callGatewayToolMock).not.toHaveBeenCalled(); + }); + + it("rejects flat on-exit schedule shorthand for update", async () => { + const tool = createCronTool(undefined, { callGatewayTool: callGatewayToolMock }); + + await expect( + tool.execute("call-flat-onexit-update", { + action: "update", + jobId: "job-onexit", + kind: "on-exit", + command: "pnpm build", + cwd: "/repo", + }), + ).rejects.toThrow("cron on-exit schedules cannot be created or edited"); + expect(callGatewayToolMock).not.toHaveBeenCalled(); + }); + + it("rejects flat command schedule shorthand for update", async () => { + const tool = createCronTool(undefined, { callGatewayTool: callGatewayToolMock }); + + await expect( + tool.execute("call-flat-onexit-update-infer", { + action: "update", + jobId: "job-infer", + command: "make", + }), + ).rejects.toThrow("cron on-exit schedules cannot be created or edited"); + expect(callGatewayToolMock).not.toHaveBeenCalled(); + }); + it("passes local cron wall-clock expression and timezone through add", async () => { const tool = createCronTool(undefined, { callGatewayTool: callGatewayToolMock }); diff --git a/src/agents/tools/cron-tool.test.ts b/src/agents/tools/cron-tool.test.ts index c257e8bdda7..484673c39bf 100644 --- a/src/agents/tools/cron-tool.test.ts +++ b/src/agents/tools/cron-tool.test.ts @@ -1004,6 +1004,22 @@ describe("cron tool", () => { expect(callGatewayMock).not.toHaveBeenCalled(); }); + it("rejects on-exit schedules from the agent cron tool on add", async () => { + const tool = createTestCronTool(); + + await expect( + tool.execute("call-on-exit-add", { + action: "add", + job: { + name: "watch command", + schedule: { kind: "on-exit", command: "make" }, + payload: { kind: "agentTurn", message: "done" }, + }, + }), + ).rejects.toThrow("cron on-exit schedules cannot be created or edited"); + expect(callGatewayMock).not.toHaveBeenCalled(); + }); + it.each([ ["delivery.channel", { channel: " ", to: "chat-1" }], ["delivery.to", { mode: "announce", channel: "telegram", to: " \t" }], @@ -2013,6 +2029,21 @@ describe("cron tool", () => { expect(callGatewayMock).not.toHaveBeenCalled(); }); + it("rejects on-exit schedules from the agent cron tool on update", async () => { + const tool = createTestCronTool(); + + await expect( + tool.execute("call-on-exit-update", { + action: "update", + id: "job-4", + patch: { + schedule: { kind: "on-exit", command: "make" }, + }, + }), + ).rejects.toThrow("cron on-exit schedules cannot be created or edited"); + expect(callGatewayMock).not.toHaveBeenCalled(); + }); + it("recovers flattened payload patch params for update action", async () => { callGatewayMock.mockResolvedValueOnce({ ok: true }); diff --git a/src/agents/tools/cron-tool.ts b/src/agents/tools/cron-tool.ts index f28b3cef883..7544dffbde1 100644 --- a/src/agents/tools/cron-tool.ts +++ b/src/agents/tools/cron-tool.ts @@ -408,7 +408,7 @@ function stripExistingContext(text: string) { return text.slice(0, index).trim(); } -function assertNoCronCommandPayload(value: unknown): void { +function assertNoCronShellExecution(value: unknown): void { if (!isRecord(value)) { return; } @@ -418,6 +418,12 @@ function assertNoCronCommandPayload(value: unknown): void { "cron command payloads cannot be created or edited through the agent cron tool; use the CLI or Gateway API.", ); } + const schedule = isRecord(value.schedule) ? value.schedule : undefined; + if (schedule?.kind === "on-exit") { + throw new Error( + "cron on-exit schedules cannot be created or edited through the agent cron tool; use the CLI or Gateway API.", + ); + } } function normalizeCronToolsAllow(values: readonly string[]): string[] { @@ -1026,7 +1032,7 @@ Use jobId canonical; id accepted compat. contextMessages (0-10) adds previous me throw new Error("job required"); } const canonicalJob = canonicalizeCronToolObject(params.job as Record); - assertNoCronCommandPayload(canonicalJob); + assertNoCronShellExecution(canonicalJob); assertCronDeliveryInputNonBlankFields(canonicalJob.delivery); const job = normalizeCronJobCreate(canonicalJob, { @@ -1150,7 +1156,7 @@ Use jobId canonical; id accepted compat. contextMessages (0-10) adds previous me const canonicalPatch = canonicalizeCronToolObject( params.patch as Record, ); - assertNoCronCommandPayload(canonicalPatch); + assertNoCronShellExecution(canonicalPatch); assertCronDeliveryInputNonBlankFields(canonicalPatch.delivery); const patch = normalizeCronJobPatch(canonicalPatch) ?? canonicalPatch; if (recoveredFlatPatch && isEmptyRecoveredCronPatch(patch)) { diff --git a/src/cli/cron-cli/register.cron-add.ts b/src/cli/cron-cli/register.cron-add.ts index a332618f6c9..ef7a4303d51 100644 --- a/src/cli/cron-cli/register.cron-add.ts +++ b/src/cli/cron-cli/register.cron-add.ts @@ -99,6 +99,11 @@ export function registerCronAddCommand(cron: Command) { ) .option("--every ", "Run every duration (e.g. 10m, 1h)") .option("--cron ", "Cron expression (5-field or 6-field with seconds)") + .option( + "--on-exit ", + "Fire once when this watched command exits (event trigger; survives turn teardown)", + ) + .option("--on-exit-cwd ", "Working directory for the --on-exit watched command") .option( "--tz ", "Timezone for cron expressions (IANA; cron default: Gateway host local timezone)", @@ -152,12 +157,15 @@ export function registerCronAddCommand(cron: Command) { const hasScheduleFlag = typeof opts.at === "string" || typeof opts.cron === "string" || - typeof opts.every === "string"; + typeof opts.every === "string" || + typeof opts.onExit === "string"; const positionalSchedule = hasScheduleFlag ? undefined : nameArg; const schedule = resolveCronCreateScheduleFromArgs({ at: opts.at, cron: opts.cron, every: opts.every, + onExit: opts.onExit, + onExitCwd: opts.onExitCwd, exact: opts.exact, positionalSchedule, stagger: opts.stagger, diff --git a/src/cli/cron-cli/schedule-options.onexit.test.ts b/src/cli/cron-cli/schedule-options.onexit.test.ts new file mode 100644 index 00000000000..9d0db1a3f48 --- /dev/null +++ b/src/cli/cron-cli/schedule-options.onexit.test.ts @@ -0,0 +1,44 @@ +import { describe, expect, it } from "vitest"; +import { resolveCronCreateScheduleFromArgs } from "./schedule-options.js"; + +describe("resolveCronCreateScheduleFromArgs --on-exit", () => { + it("builds an on-exit schedule from --on-exit (+ optional cwd)", () => { + expect(resolveCronCreateScheduleFromArgs({ onExit: "make build" })).toEqual({ + kind: "on-exit", + command: "make build", + }); + expect(resolveCronCreateScheduleFromArgs({ onExit: "./watch.sh", onExitCwd: "/repo" })).toEqual( + { kind: "on-exit", command: "./watch.sh", cwd: "/repo" }, + ); + }); + + it("rejects --on-exit combined with another schedule", () => { + expect(() => resolveCronCreateScheduleFromArgs({ onExit: "make", every: "10m" })).toThrow( + /exactly one schedule/, + ); + }); + + it("rejects --on-exit combined with a positional schedule", () => { + expect(() => + resolveCronCreateScheduleFromArgs({ onExit: "make", positionalSchedule: "10m" }), + ).toThrow(/positional schedule or one of/); + }); + + it("rejects --tz/--stagger with --on-exit", () => { + expect(() => + resolveCronCreateScheduleFromArgs({ onExit: "make", tz: "Asia/Shanghai" }), + ).toThrow(/not valid with --on-exit/); + }); + + it("rejects orphan --on-exit-cwd (flag-only) instead of a confusing generic error", () => { + expect(() => resolveCronCreateScheduleFromArgs({ onExitCwd: "/repo" })).toThrow( + /--on-exit-cwd requires --on-exit/, + ); + }); + + it("rejects --on-exit-cwd alongside a positional schedule (cwd would be silently dropped)", () => { + expect(() => + resolveCronCreateScheduleFromArgs({ onExitCwd: "/repo", positionalSchedule: "10m" }), + ).toThrow(/--on-exit-cwd requires --on-exit/); + }); +}); diff --git a/src/cli/cron-cli/schedule-options.ts b/src/cli/cron-cli/schedule-options.ts index 3df3366883c..1e8c87aa896 100644 --- a/src/cli/cron-cli/schedule-options.ts +++ b/src/cli/cron-cli/schedule-options.ts @@ -7,6 +7,8 @@ type ScheduleOptionInput = { at?: unknown; cron?: unknown; every?: unknown; + onExit?: unknown; + onExitCwd?: unknown; exact?: unknown; stagger?: unknown; tz?: unknown; @@ -20,6 +22,8 @@ type NormalizedScheduleOptions = { at: string; cronExpr: string; every: string; + onExitCommand: string; + onExitCwd: string | undefined; requestedStaggerMs: number | undefined; tz: string | undefined; }; @@ -33,13 +37,16 @@ export type CronEditScheduleRequest = /** Resolve explicit `--at`, `--every`, or `--cron` options for cron creation. */ export function resolveCronCreateSchedule(options: ScheduleOptionInput): CronSchedule { const normalized = normalizeScheduleOptions(options); + if (normalized.onExitCwd && !normalized.onExitCommand) { + throw new Error("--on-exit-cwd requires --on-exit."); + } const chosen = countChosenSchedules(normalized); if (chosen !== 1) { - throw new Error("Choose exactly one schedule: --at, --every, or --cron"); + throw new Error("Choose exactly one schedule: --at, --every, --cron, or --on-exit"); } const schedule = resolveDirectSchedule(normalized); if (!schedule) { - throw new Error("Choose exactly one schedule: --at, --every, or --cron"); + throw new Error("Choose exactly one schedule: --at, --every, --cron, or --on-exit"); } return schedule; } @@ -54,7 +61,7 @@ export function resolveCronCreateScheduleFromArgs( } const normalized = normalizeScheduleOptions(options); if (countChosenSchedules(normalized) > 0) { - throw new Error("Choose a positional schedule or one of --at, --every, or --cron."); + throw new Error("Choose a positional schedule or one of --at, --every, --cron, or --on-exit."); } const every = parseEverySchedule(positionalSchedule); return resolveCronCreateSchedule({ @@ -118,14 +125,20 @@ function normalizeScheduleOptions(options: ScheduleOptionInput): NormalizedSched at: normalizeOptionalString(options.at) ?? "", every: normalizeOptionalString(options.every) ?? "", cronExpr: normalizeOptionalString(options.cron) ?? "", + onExitCommand: normalizeOptionalString(options.onExit) ?? "", + onExitCwd: normalizeOptionalString(options.onExitCwd), tz: normalizeOptionalString(options.tz), requestedStaggerMs: parseCronStaggerMs({ staggerRaw, useExact }), }; } function countChosenSchedules(options: NormalizedScheduleOptions): number { - return [Boolean(options.at), Boolean(options.every), Boolean(options.cronExpr)].filter(Boolean) - .length; + return [ + Boolean(options.at), + Boolean(options.every), + Boolean(options.cronExpr), + Boolean(options.onExitCommand), + ].filter(Boolean).length; } function parseEverySchedule(value: string): string | undefined { @@ -139,6 +152,9 @@ function looksLikeCronExpression(value: string): boolean { } function resolveDirectSchedule(options: NormalizedScheduleOptions): CronSchedule | undefined { + if (options.onExitCwd && !options.onExitCommand) { + throw new Error("--on-exit-cwd requires --on-exit."); + } if (options.tz && options.every) { throw new Error("--tz is only valid with --cron or offset-less --at"); } @@ -167,5 +183,15 @@ function resolveDirectSchedule(options: NormalizedScheduleOptions): CronSchedule staggerMs: options.requestedStaggerMs, }; } + if (options.onExitCommand) { + if (options.tz || options.requestedStaggerMs !== undefined) { + throw new Error("--tz/--stagger/--exact are not valid with --on-exit"); + } + return { + kind: "on-exit", + command: options.onExitCommand, + ...(options.onExitCwd ? { cwd: options.onExitCwd } : {}), + }; + } return undefined; } diff --git a/src/cli/cron-cli/shared.test.ts b/src/cli/cron-cli/shared.test.ts index 95a9b5ce652..2ea2b978273 100644 --- a/src/cli/cron-cli/shared.test.ts +++ b/src/cli/cron-cli/shared.test.ts @@ -121,6 +121,25 @@ describe("printCronList", () => { expectLogsToInclude(logs, "(stagger 5m)"); }); + it("shows on-exit schedules in list and show output", () => { + const job = createBaseJob({ + id: "on-exit-job", + name: "Watch build", + schedule: { kind: "on-exit", command: "pnpm build", cwd: "/repo" }, + sessionTarget: "main", + state: {}, + payload: { kind: "systemEvent", text: "done" }, + }); + + const list = createRuntimeLogCapture(); + printCronList([job], list.runtime); + expectLogsToInclude(list.logs, "on-exit pnpm build @ /repo"); + + const show = createRuntimeLogCapture(); + printCronShow(job, show.runtime); + expectLogsToInclude(show.logs, "schedule: on-exit pnpm build @ /repo"); + }); + it("shows dash for unset agentId instead of default", () => { const { logs, runtime } = createRuntimeLogCapture(); const job = createBaseJob({ diff --git a/src/cli/cron-cli/shared.ts b/src/cli/cron-cli/shared.ts index 8f784d3f7d9..3eec1f933d3 100644 --- a/src/cli/cron-cli/shared.ts +++ b/src/cli/cron-cli/shared.ts @@ -384,6 +384,10 @@ const formatSchedule = (schedule: CronSchedule | undefined) => { if (schedule?.kind === "every") { return `every ${formatDurationHuman(schedule.everyMs)}`; } + if (schedule?.kind === "on-exit") { + const cwd = schedule.cwd ? ` @ ${schedule.cwd}` : ""; + return `on-exit ${schedule.command}${cwd}`; + } if (schedule?.kind !== "cron") { return "-"; } diff --git a/src/cron/normalize.test.ts b/src/cron/normalize.test.ts index 1e419b817f7..199c491a6c8 100644 --- a/src/cron/normalize.test.ts +++ b/src/cron/normalize.test.ts @@ -1039,3 +1039,37 @@ describe("normalizeCronJobPatch", () => { expect(validateCronUpdateParams({ id: "job-1", patch: normalized })).toBe(true); }); }); + +describe("on-exit schedule normalization", () => { + it("keeps command/cwd and strips time fields for on-exit jobs", () => { + const normalized = normalizeCronJobCreate({ + name: "watch build", + schedule: { + kind: "on-exit", + command: "make build", + cwd: "/repo", + // stale fields from a prior kind that must be dropped + everyMs: 1000, + expr: "* * * * *", + at: "2026-01-01T00:00:00Z", + }, + payload: { kind: "systemEvent", text: "build done" }, + sessionTarget: "main", + }); + expect(normalized).not.toBeNull(); + expect(normalized?.schedule).toEqual({ kind: "on-exit", command: "make build", cwd: "/repo" }); + expect(validateCronAddParams(normalized)).toBe(true); + }); + + it("drops command/cwd when normalizing a non-on-exit schedule", () => { + const normalized = normalizeCronJobCreate({ + name: "interval", + schedule: { kind: "every", everyMs: 5000, command: "leftover", cwd: "/x" }, + payload: { kind: "systemEvent", text: "tick" }, + sessionTarget: "main", + }); + expect(normalized).not.toBeNull(); + expect(normalized?.schedule).not.toHaveProperty("command"); + expect(normalized?.schedule).not.toHaveProperty("cwd"); + }); +}); diff --git a/src/cron/normalize.ts b/src/cron/normalize.ts index 63210dc7b5d..9c6404151e0 100644 --- a/src/cron/normalize.ts +++ b/src/cron/normalize.ts @@ -94,8 +94,13 @@ function hasAgentTurnOnlyPayloadHint(payload: UnknownRecord): boolean { function coerceSchedule(schedule: UnknownRecord) { const next: UnknownRecord = { ...schedule }; const rawKind = normalizeLowercaseStringOrEmpty(schedule.kind); - const kind = rawKind === "at" || rawKind === "every" || rawKind === "cron" ? rawKind : undefined; + const kind = + rawKind === "at" || rawKind === "every" || rawKind === "cron" || rawKind === "on-exit" + ? rawKind + : undefined; const exprRaw = normalizeOptionalString(schedule.expr) ?? ""; + const commandRaw = normalizeOptionalString(schedule.command) ?? ""; + const cwdRaw = normalizeOptionalString(schedule.cwd) ?? ""; const everyMs = coerceFiniteScheduleNumber(schedule.everyMs); const anchorMs = coerceFiniteScheduleNumber(schedule.anchorMs); const atString = normalizeOptionalString(schedule.at) ?? ""; @@ -124,6 +129,16 @@ function coerceSchedule(schedule: UnknownRecord) { if (anchorMs !== undefined && anchorMs >= 0) { next.anchorMs = Math.floor(anchorMs); } + if (commandRaw) { + next.command = commandRaw; + } else if ("command" in next) { + delete next.command; + } + if (cwdRaw) { + next.cwd = cwdRaw; + } else if ("cwd" in next) { + delete next.cwd; + } const staggerMs = normalizeCronStaggerMs(schedule.staggerMs); if (staggerMs !== undefined) { next.staggerMs = staggerMs; @@ -148,6 +163,20 @@ function coerceSchedule(schedule: UnknownRecord) { delete next.at; delete next.everyMs; delete next.anchorMs; + delete next.command; + delete next.cwd; + } else if (next.kind === "on-exit") { + delete next.at; + delete next.everyMs; + delete next.anchorMs; + delete next.expr; + delete next.tz; + delete next.staggerMs; + } + + if (next.kind !== "on-exit") { + delete next.command; + delete next.cwd; } return next; diff --git a/src/cron/persisted-shape.onexit.test.ts b/src/cron/persisted-shape.onexit.test.ts new file mode 100644 index 00000000000..963f4a17767 --- /dev/null +++ b/src/cron/persisted-shape.onexit.test.ts @@ -0,0 +1,37 @@ +// Regression: on-exit jobs must pass the persisted-shape validator (else they +// cannot be saved to the cron store or survive a gateway restart). +import { describe, expect, it } from "vitest"; +import { getInvalidPersistedCronJobReason } from "./persisted-shape.js"; + +function onExitCandidate(overrides: Record = {}) { + return { + id: "job-1", + schedule: { kind: "on-exit", command: "make build" }, + payload: { kind: "systemEvent", text: "done" }, + sessionTarget: "main", + ...overrides, + }; +} + +describe("getInvalidPersistedCronJobReason on-exit", () => { + it("accepts a well-formed on-exit job", () => { + expect(getInvalidPersistedCronJobReason(onExitCandidate())).toBeNull(); + }); + + it("rejects an on-exit job with an empty/missing command", () => { + expect( + getInvalidPersistedCronJobReason( + onExitCandidate({ schedule: { kind: "on-exit", command: "" } }), + ), + ).toBe("invalid-schedule"); + expect( + getInvalidPersistedCronJobReason(onExitCandidate({ schedule: { kind: "on-exit" } })), + ).toBe("invalid-schedule"); + }); + + it("still rejects genuinely unknown schedule kinds", () => { + expect( + getInvalidPersistedCronJobReason(onExitCandidate({ schedule: { kind: "whenever" } })), + ).toBe("invalid-schedule"); + }); +}); diff --git a/src/cron/persisted-shape.ts b/src/cron/persisted-shape.ts index 6293ced3f92..37ed6dc4bc7 100644 --- a/src/cron/persisted-shape.ts +++ b/src/cron/persisted-shape.ts @@ -31,7 +31,12 @@ export function getInvalidPersistedCronJobReason( } const scheduleRecord = schedule as Record; const scheduleKind = scheduleRecord.kind; - if (scheduleKind !== "at" && scheduleKind !== "every" && scheduleKind !== "cron") { + if ( + scheduleKind !== "at" && + scheduleKind !== "every" && + scheduleKind !== "cron" && + scheduleKind !== "on-exit" + ) { return "invalid-schedule"; } if (scheduleKind === "at") { @@ -52,6 +57,12 @@ export function getInvalidPersistedCronJobReason( return "invalid-schedule"; } } + if (scheduleKind === "on-exit") { + const command = scheduleRecord.command; + if (typeof command !== "string" || command.trim().length === 0) { + return "invalid-schedule"; + } + } const payload = candidate.payload; if (!payload || typeof payload !== "object" || Array.isArray(payload)) { return "missing-payload"; diff --git a/src/cron/schedule.test.ts b/src/cron/schedule.test.ts index 024a554ac28..e3c9477ab93 100644 --- a/src/cron/schedule.test.ts +++ b/src/cron/schedule.test.ts @@ -253,3 +253,12 @@ describe("coerceFiniteScheduleNumber", () => { expect(coerceFiniteScheduleNumber(undefined)).toBeUndefined(); }); }); + +describe("computeNextRunAtMs on-exit", () => { + it("never reports a time-due run for on-exit schedules (event-driven)", () => { + expect(computeNextRunAtMs({ kind: "on-exit", command: "sleep 1" }, Date.now())).toBeUndefined(); + expect( + computeNextRunAtMs({ kind: "on-exit", command: "make build", cwd: "/repo" }, 0), + ).toBeUndefined(); + }); +}); diff --git a/src/cron/schedule.ts b/src/cron/schedule.ts index 7a5e4592b99..9a26043fcc8 100644 --- a/src/cron/schedule.ts +++ b/src/cron/schedule.ts @@ -77,6 +77,12 @@ export function computeNextRunAtMs(schedule: CronSchedule, nowMs: number): numbe return anchor + steps * everyMs; } + if (schedule.kind === "on-exit") { + // Event-driven trigger: never time-due. The gateway watcher calls + // enqueueRun when the watched command exits. + return undefined; + } + const cron = resolveCronFromSchedule(schedule); if (!cron) { return undefined; diff --git a/src/cron/service-contract.ts b/src/cron/service-contract.ts index 167ced95c83..395e46a5586 100644 --- a/src/cron/service-contract.ts +++ b/src/cron/service-contract.ts @@ -12,12 +12,15 @@ import type { CronUpdateResult, CronWakeMode, } from "./service/state.js"; -import type { CronJob } from "./types.js"; +import type { CronJob, CronPayload } from "./types.js"; type CronWakeResult = { ok: true } | { ok: false; reason?: "unwakeable-session-key" }; /** Result shape for direct/queued cron runs. */ export type CronServiceRunResult = CronRunResult; +export type CronServiceRunOptions = { + payload?: CronPayload; +}; /** Public cron service facade used by gateway, plugin SDK, and tests. */ export interface CronServiceContract { @@ -29,7 +32,7 @@ export interface CronServiceContract { add(input: CronAddInput): Promise; update(id: string, patch: CronUpdateInput): Promise; remove(id: string): Promise; - run(id: string, mode?: CronRunMode): Promise; + run(id: string, mode?: CronRunMode, opts?: CronServiceRunOptions): Promise; enqueueRun(id: string, mode?: CronRunMode): Promise; getJob(id: string): CronJob | undefined; readJob(id: string): Promise; diff --git a/src/cron/service.ts b/src/cron/service.ts index 843fb11be31..ea67119ab38 100644 --- a/src/cron/service.ts +++ b/src/cron/service.ts @@ -1,5 +1,9 @@ /** Stateful CronService facade around the locked service operation helpers. */ -import type { CronServiceContract, CronServiceRunResult } from "./service-contract.js"; +import type { + CronServiceContract, + CronServiceRunOptions, + CronServiceRunResult, +} from "./service-contract.js"; import type { CronListPageOptions } from "./service/list-page-types.js"; import * as ops from "./service/ops.js"; import { @@ -51,8 +55,12 @@ export class CronService implements CronServiceContract { return await ops.remove(this.state, id); } - async run(id: string, mode?: "due" | "force"): Promise { - return await ops.run(this.state, id, mode); + async run( + id: string, + mode?: "due" | "force", + opts?: CronServiceRunOptions, + ): Promise { + return await ops.run(this.state, id, mode, opts); } async enqueueRun(id: string, mode?: "due" | "force"): Promise { diff --git a/src/cron/service/list-page-types.ts b/src/cron/service/list-page-types.ts index d6a975639c5..8633c3177a5 100644 --- a/src/cron/service/list-page-types.ts +++ b/src/cron/service/list-page-types.ts @@ -5,7 +5,7 @@ import type { CronJob, CronRunStatus } from "../types.js"; export type CronJobsEnabledFilter = "all" | "enabled" | "disabled"; /** Schedule-kind filter accepted by paginated cron listing. */ -export type CronJobsScheduleKindFilter = "all" | "at" | "every" | "cron"; +export type CronJobsScheduleKindFilter = "all" | "at" | "every" | "cron" | "on-exit"; /** Last-run status filter, including jobs that have not produced a status yet. */ export type CronJobsLastRunStatusFilter = "all" | CronRunStatus | "unknown"; diff --git a/src/cron/service/ops.ts b/src/cron/service/ops.ts index e4a8254b425..52ffaca8302 100644 --- a/src/cron/service/ops.ts +++ b/src/cron/service/ops.ts @@ -19,7 +19,7 @@ import { resolveCronDeliveryPlan, resolveFailureDestination } from "../delivery- import { createCronRunDiagnosticsFromError } from "../run-diagnostics.js"; import { createCronExecutionId } from "../run-id.js"; import { cronSchedulingInputsEqual } from "../schedule-identity.js"; -import type { CronJob, CronJobCreate, CronJobPatch } from "../types.js"; +import type { CronJob, CronJobCreate, CronJobPatch, CronPayload } from "../types.js"; import { normalizeCronRunErrorText } from "./execution-errors.js"; import { failureNotificationDeliveryFromJobState } from "./failure-alerts.js"; import { @@ -350,7 +350,8 @@ function resolveScheduleKindFilter(opts?: CronListPageOptions): CronJobsSchedule opts?.scheduleKind === "all" || opts?.scheduleKind === "at" || opts?.scheduleKind === "every" || - opts?.scheduleKind === "cron" + opts?.scheduleKind === "cron" || + opts?.scheduleKind === "on-exit" ) { return opts.scheduleKind; } @@ -639,6 +640,11 @@ type PreparedManualRun = } | { ok: false }; +type ManualRunOptions = { + runId?: string; + payload?: CronPayload; +}; + type ManualRunDisposition = | Extract | { ok: true; runnable: true }; @@ -847,7 +853,7 @@ async function prepareManualRun( state: CronServiceState, id: string, mode?: "due" | "force", - opts?: { runId?: string }, + opts?: ManualRunOptions, ): Promise { const preflight = await inspectManualRunPreflight(state, id, mode); if (!preflight.ok) { @@ -894,6 +900,9 @@ async function prepareManualRun( // Execute against a snapshot so later reload/merge can preserve delivery // target writeback from disk without mutating the running object. const executionJob = structuredClone(job); + if (opts?.payload) { + executionJob.payload = structuredClone(opts.payload); + } return { ok: true, ran: true, @@ -1043,7 +1052,7 @@ export async function run( state: CronServiceState, id: string, mode?: "due" | "force", - opts?: { runId?: string }, + opts?: ManualRunOptions, ) { const prepared = await prepareManualRun(state, id, mode, opts); if (!prepared.ok || !prepared.ran) { diff --git a/src/cron/store/row-codec.schedule.test.ts b/src/cron/store/row-codec.schedule.test.ts new file mode 100644 index 00000000000..32f241597c8 --- /dev/null +++ b/src/cron/store/row-codec.schedule.test.ts @@ -0,0 +1,54 @@ +// Round-trips each CronSchedule kind through the SQLite column codec so the +// on-exit command/cwd persistence (v1 reuses schedule_expr/schedule_tz) is +// covered alongside the existing kinds. +import { describe, expect, it } from "vitest"; +import type { CronSchedule } from "../types.js"; +import { bindScheduleColumns, scheduleFromRow } from "./row-codec.js"; +import type { CronJobRow } from "./schema.js"; + +function roundTrip(schedule: CronSchedule): CronSchedule | null { + const cols = bindScheduleColumns(schedule); + // scheduleFromRow only reads the schedule_* / at / every_ms / anchor_ms / + // stagger_ms columns; the rest of the row is irrelevant here. + return scheduleFromRow(cols as unknown as CronJobRow); +} + +describe("schedule column codec round-trip", () => { + it("round-trips an on-exit schedule with command + cwd", () => { + expect(roundTrip({ kind: "on-exit", command: "make build", cwd: "/repo" })).toEqual({ + kind: "on-exit", + command: "make build", + cwd: "/repo", + }); + }); + + it("round-trips an on-exit schedule without cwd", () => { + expect(roundTrip({ kind: "on-exit", command: "./watch.sh" })).toEqual({ + kind: "on-exit", + command: "./watch.sh", + }); + }); + + it("keeps existing kinds intact (no cross-talk from on-exit column reuse)", () => { + expect(roundTrip({ kind: "every", everyMs: 60_000 })).toEqual({ + kind: "every", + everyMs: 60_000, + }); + expect(roundTrip({ kind: "cron", expr: "0 9 * * *", tz: "Asia/Shanghai" })).toEqual({ + kind: "cron", + expr: "0 9 * * *", + tz: "Asia/Shanghai", + }); + expect(roundTrip({ kind: "at", at: "2026-01-01T00:00:00.000Z" })).toEqual({ + kind: "at", + at: "2026-01-01T00:00:00.000Z", + }); + }); + + it("an on-exit row is decoded as on-exit, not cron (schedule_kind disambiguates)", () => { + const cols = bindScheduleColumns({ kind: "on-exit", command: "sleep 5" }); + expect(cols.schedule_kind).toBe("on-exit"); + const decoded = scheduleFromRow(cols as unknown as CronJobRow); + expect(decoded?.kind).toBe("on-exit"); + }); +}); diff --git a/src/cron/store/row-codec.ts b/src/cron/store/row-codec.ts index 7a03b0e0be4..b320a8beace 100644 --- a/src/cron/store/row-codec.ts +++ b/src/cron/store/row-codec.ts @@ -21,7 +21,7 @@ import { getCronStoreKysely } from "./schema.js"; import { bindStateColumns, stateFromRow } from "./state-codec.js"; import type { LoadedCronStore } from "./types.js"; -function bindScheduleColumns( +export function bindScheduleColumns( schedule: CronSchedule, ): Pick< CronJobInsert, @@ -49,6 +49,21 @@ function bindScheduleColumns( stagger_ms: null, }; } + if (schedule.kind === "on-exit") { + // v1: reuse existing nullable TEXT columns to round-trip the watcher's + // command (schedule_expr) and cwd (schedule_tz) without a schema migration. + // schedule_kind disambiguates from cron. (Dedicated columns are a possible + // follow-up if reviewers prefer.) + return { + schedule_kind: "on-exit", + at: null, + every_ms: null, + anchor_ms: null, + schedule_expr: schedule.command, + schedule_tz: schedule.cwd ?? null, + stagger_ms: null, + }; + } return { schedule_kind: "cron", at: null, @@ -189,7 +204,7 @@ export function assertCronStoreCanPersist(store: CronStoreFile): void { } } -function scheduleFromRow(row: CronJobRow): CronSchedule | null { +export function scheduleFromRow(row: CronJobRow): CronSchedule | null { if (row.schedule_kind === "at" && row.at) { return { kind: "at", at: row.at }; } @@ -208,6 +223,13 @@ function scheduleFromRow(row: CronJobRow): CronSchedule | null { ...(row.stagger_ms != null ? { staggerMs: normalizeNumber(row.stagger_ms) } : {}), }; } + if (row.schedule_kind === "on-exit" && row.schedule_expr) { + return { + kind: "on-exit", + command: row.schedule_expr, + ...(row.schedule_tz ? { cwd: row.schedule_tz } : {}), + }; + } return null; } diff --git a/src/cron/types.ts b/src/cron/types.ts index b61a167b7b7..7ef4479942a 100644 --- a/src/cron/types.ts +++ b/src/cron/types.ts @@ -15,6 +15,20 @@ export type CronSchedule = tz?: string; /** Optional deterministic stagger window in milliseconds (0 keeps exact schedule). */ staggerMs?: number; + } + | { + /** + * Event-driven (non-time) trigger: the job fires once when a gateway-owned + * watcher process running `command` exits. The watcher lives under the + * gateway ProcessSupervisor, NOT inside any agent turn's process tree, so + * it survives the per-turn spawn-and-kill teardown that CLI backends apply + * (#71662). On exit the job runs through the normal cron run pipeline, so + * delivery to the bound session works exactly like a scheduled main job. + * `computeNextRunAtMs` returns undefined for this kind (never time-due). + */ + kind: "on-exit"; + command: string; + cwd?: string; }; /** Runtime target that decides whether a job joins main, isolated, or a named session. */ diff --git a/src/gateway/cron-exit-watchers.test.ts b/src/gateway/cron-exit-watchers.test.ts new file mode 100644 index 00000000000..5a4b1b629e2 --- /dev/null +++ b/src/gateway/cron-exit-watchers.test.ts @@ -0,0 +1,390 @@ +import { describe, expect, it, vi } from "vitest"; +import type { CronJob } from "../cron/types.js"; +import { + createCronExitWatchers, + type CronExitResult, + resolveExitWatchShell, +} from "./cron-exit-watchers.js"; + +type Deferred = { + resolve: (exit: { exitCode: number | null; reason: string }) => void; + reject: (err: unknown) => void; +}; + +type FireOnExit = (job: CronJob, exit: CronExitResult) => Promise; + +/** + * Minimal fake ProcessSupervisor: each spawn returns a run whose wait() is + * controlled by the test, so we can deterministically drive "command exited". + */ +function makeFakeSupervisor(opts: { deferSpawn?: boolean } = {}) { + const runs: { scopeKey?: string; runId: string; deferred: Deferred; cancelled: boolean }[] = []; + const cancelledScopes: string[] = []; + const runCancels: string[] = []; + let counter = 0; + let releaseSpawn: (() => void) | undefined; + const spawnGate = opts.deferSpawn + ? new Promise((res) => { + releaseSpawn = res; + }) + : Promise.resolve(); + const supervisor = { + spawn: vi.fn(async (input: { scopeKey?: string }) => { + await spawnGate; + counter += 1; + const runId = `run-${counter}`; + let resolveWait!: (exit: { exitCode: number | null; reason: string }) => void; + let rejectWait!: (err: unknown) => void; + const waitPromise = new Promise<{ exitCode: number | null; reason: string }>((res, rej) => { + resolveWait = res; + rejectWait = rej; + }); + // Pre-attach a no-op catch so a test-driven rejection never escapes as an + // unhandled rejection if the run loses ownership before it awaits wait(). + waitPromise.catch(() => {}); + const entry = { + scopeKey: input.scopeKey, + runId, + deferred: { resolve: resolveWait, reject: rejectWait }, + cancelled: false, + }; + runs.push(entry); + return { + runId, + startedAtMs: 0, + wait: () => + waitPromise.then((e) => ({ + ...e, + exitSignal: null, + durationMs: 1, + stdout: "", + stderr: "", + timedOut: false, + noOutputTimedOut: false, + })), + cancel: () => { + entry.cancelled = true; + runCancels.push(runId); + }, + }; + }), + cancelScope: vi.fn((scopeKey: string) => { + cancelledScopes.push(scopeKey); + }), + }; + return { + supervisor, + runs, + cancelled: cancelledScopes, + cancelledScopes, + runCancels, + releaseSpawn: () => releaseSpawn?.(), + }; +} + +function onExitJob(id: string, command = "true", enabled = true): CronJob { + return { + id, + name: id, + enabled, + createdAtMs: 1, + updatedAtMs: 1, + schedule: { kind: "on-exit", command }, + sessionTarget: "main", + wakeMode: "now", + payload: { kind: "systemEvent", text: "done" }, + delivery: { mode: "none" }, + state: {}, + } as unknown as CronJob; +} + +const noopLogger = { info: () => {}, warn: () => {} }; + +const flush = async () => { + await Promise.resolve(); + await Promise.resolve(); +}; + +describe("createCronExitWatchers", () => { + it("arms a watcher for an enabled on-exit job and fires the job on exit", async () => { + const { supervisor, runs } = makeFakeSupervisor(); + const order: string[] = []; + const persistCompletion = vi.fn(async () => { + order.push("persist"); + }); + const fireOnExit = vi.fn(async (_job: CronJob, _exit: CronExitResult) => { + order.push("fire"); + }); + const w = createCronExitWatchers({ + getProcessSupervisor: () => supervisor as never, + persistCompletion, + fireOnExit, + logger: noopLogger, + }); + + w.reconcile([onExitJob("job-a")]); + await flush(); + expect(supervisor.spawn).toHaveBeenCalledTimes(1); + expect(w.activeJobIds()).toEqual(["job-a"]); + expect(fireOnExit).not.toHaveBeenCalled(); + + // Watched command exits → job fires through the run pipeline. + runs[0].deferred.resolve({ exitCode: 0, reason: "exit" }); + await flush(); + expect(fireOnExit).toHaveBeenCalledTimes(1); + expect(fireOnExit.mock.calls[0]?.[0].id).toBe("job-a"); + expect(fireOnExit.mock.calls[0]?.[1]).toMatchObject({ + exitCode: 0, + reason: "exit", + stdout: "", + stderr: "", + }); + // One-shot terminal state is persisted BEFORE firing (restart-safe). + expect(persistCompletion).toHaveBeenCalledWith("job-a"); + expect(order).toEqual(["persist", "fire"]); + }); + + it("a fired job stays unarmed across a simulated restart (disabled in store → not re-run)", async () => { + // persistCompletion disables the job; after a restart the reconcile sees a + // disabled job and must NOT re-arm (which would re-run the command). + const { supervisor, runs } = makeFakeSupervisor(); + const w = createCronExitWatchers({ + getProcessSupervisor: () => supervisor as never, + persistCompletion: vi.fn(async () => {}), + fireOnExit: vi.fn(async () => {}), + logger: noopLogger, + }); + w.reconcile([onExitJob("job-a")]); + await flush(); + runs[0].deferred.resolve({ exitCode: 0, reason: "exit" }); + await flush(); + expect(supervisor.spawn).toHaveBeenCalledTimes(1); + // Simulate restart: a fresh manager reconciling the now-disabled persisted job. + const restarted = createCronExitWatchers({ + getProcessSupervisor: () => supervisor as never, + persistCompletion: vi.fn(async () => {}), + fireOnExit: vi.fn(async () => {}), + logger: noopLogger, + }); + restarted.reconcile([onExitJob("job-a", "sleep 1", false)]); // enabled=false after completion + await flush(); + expect(supervisor.spawn).toHaveBeenCalledTimes(1); // no re-spawn → command not re-run + expect(restarted.activeJobIds()).toEqual([]); + }); + + it("does NOT fire when persistCompletion fails (fail closed to avoid replay)", async () => { + const { supervisor, runs } = makeFakeSupervisor(); + const fireOnExit = vi.fn(async () => {}); + const w = createCronExitWatchers({ + getProcessSupervisor: () => supervisor as never, + persistCompletion: vi.fn(async () => { + throw new Error("store write failed"); + }), + fireOnExit, + logger: noopLogger, + }); + w.reconcile([onExitJob("job-a")]); + await flush(); + runs[0].deferred.resolve({ exitCode: 0, reason: "exit" }); + await flush(); + expect(fireOnExit).not.toHaveBeenCalled(); + expect(w.activeJobIds()).toEqual([]); + w.reconcile([onExitJob("job-a")]); + await flush(); + expect(supervisor.spawn).toHaveBeenCalledTimes(2); + expect(w.activeJobIds()).toEqual(["job-a"]); + }); + + it("releases the slot without firing when run.wait() rejects (fail closed on unknown outcome)", async () => { + const { supervisor, runs } = makeFakeSupervisor(); + const persistCompletion = vi.fn(async () => {}); + const fireOnExit = vi.fn(async () => {}); + const w = createCronExitWatchers({ + getProcessSupervisor: () => supervisor as never, + persistCompletion, + fireOnExit, + logger: noopLogger, + }); + + w.reconcile([onExitJob("job-a")]); + await flush(); + expect(supervisor.spawn).toHaveBeenCalledTimes(1); + expect(w.activeJobIds()).toEqual(["job-a"]); + + // wait() rejects (e.g. supervisor error) instead of resolving with an exit. + runs[0].deferred.reject(new Error("supervisor wait blew up")); + await flush(); + + // Fail closed: no fire, no persisted terminal state on an unknown outcome. + expect(fireOnExit).not.toHaveBeenCalled(); + expect(persistCompletion).not.toHaveBeenCalled(); + // Slot released so a subsequent reconcile can re-arm the job. + expect(w.activeJobIds()).toEqual([]); + w.reconcile([onExitJob("job-a")]); + await flush(); + expect(supervisor.spawn).toHaveBeenCalledTimes(2); + expect(w.activeJobIds()).toEqual(["job-a"]); + }); + + it("replaces the watcher when the watched command changes", async () => { + const { supervisor, cancelledScopes } = makeFakeSupervisor(); + const w = createCronExitWatchers({ + getProcessSupervisor: () => supervisor as never, + persistCompletion: vi.fn(async () => {}), + fireOnExit: vi.fn(async () => {}), + logger: noopLogger, + }); + w.reconcile([onExitJob("job-a", "sleep 1")]); + await flush(); + expect(supervisor.spawn).toHaveBeenCalledTimes(1); + // Same job id, different command → cancel the stale watcher and re-arm. + w.reconcile([onExitJob("job-a", "sleep 999")]); + await flush(); + expect(cancelledScopes).toContain("cron-exit:job-a"); + expect(supervisor.spawn).toHaveBeenCalledTimes(2); + }); + + it("fires with the latest job snapshot when non-schedule fields change", async () => { + const { supervisor, runs } = makeFakeSupervisor(); + const fireOnExit = vi.fn(async () => {}); + const w = createCronExitWatchers({ + getProcessSupervisor: () => supervisor as never, + persistCompletion: vi.fn(async () => {}), + fireOnExit, + logger: noopLogger, + }); + w.reconcile([onExitJob("job-a")]); + await flush(); + + w.reconcile([ + { + ...onExitJob("job-a"), + payload: { kind: "systemEvent", text: "updated" }, + } as CronJob, + ]); + runs[0].deferred.resolve({ exitCode: 0, reason: "exit" }); + await flush(); + + expect(supervisor.spawn).toHaveBeenCalledTimes(1); + expect(fireOnExit.mock.calls[0]?.[0]).toMatchObject({ + payload: { kind: "systemEvent", text: "updated" }, + }); + }); + + it("cancels and kills an in-flight spawn when the job is removed mid-spawn", async () => { + const fake = makeFakeSupervisor({ deferSpawn: true }); + const fireOnExit = vi.fn(async () => {}); + const w = createCronExitWatchers({ + getProcessSupervisor: () => fake.supervisor as never, + persistCompletion: vi.fn(async () => {}), + fireOnExit, + logger: noopLogger, + }); + w.reconcile([onExitJob("job-a")]); + await flush(); // spawn is awaiting the gate (in flight, untracked child) + w.reconcile([]); // remove the job while the spawn is in flight + fake.releaseSpawn(); // spawn now resolves + await flush(); + await flush(); + // The orphaned child is killed and the job never fires. + expect(fake.runCancels.length).toBe(1); + expect(fireOnExit).not.toHaveBeenCalled(); + expect(w.activeJobIds()).toEqual([]); + }); + + it("does not arm a watcher for time-based or disabled jobs", async () => { + const { supervisor } = makeFakeSupervisor(); + const w = createCronExitWatchers({ + getProcessSupervisor: () => supervisor as never, + persistCompletion: vi.fn(async () => {}), + fireOnExit: vi.fn(async () => {}), + logger: noopLogger, + }); + const everyJob = { + ...onExitJob("timer"), + schedule: { kind: "every", everyMs: 1000 }, + } as unknown as CronJob; + w.reconcile([everyJob, onExitJob("disabled", "true", false)]); + await flush(); + expect(supervisor.spawn).not.toHaveBeenCalled(); + expect(w.activeJobIds()).toEqual([]); + }); + + it("is idempotent: re-reconciling the same job does not double-arm", async () => { + const { supervisor } = makeFakeSupervisor(); + const w = createCronExitWatchers({ + getProcessSupervisor: () => supervisor as never, + persistCompletion: vi.fn(async () => {}), + fireOnExit: vi.fn(async () => {}), + logger: noopLogger, + }); + w.reconcile([onExitJob("job-a")]); + await flush(); + w.reconcile([onExitJob("job-a")]); + await flush(); + expect(supervisor.spawn).toHaveBeenCalledTimes(1); + }); + + it("cancels the watcher when the job is removed from the set", async () => { + const { supervisor, cancelled } = makeFakeSupervisor(); + const w = createCronExitWatchers({ + getProcessSupervisor: () => supervisor as never, + persistCompletion: vi.fn(async () => {}), + fireOnExit: vi.fn(async () => {}), + logger: noopLogger, + }); + w.reconcile([onExitJob("job-a")]); + await flush(); + w.reconcile([]); + expect(cancelled).toContain("cron-exit:job-a"); + expect(w.activeJobIds()).toEqual([]); + }); + + it("does not fire a job whose watcher was cancelled before exit", async () => { + const { supervisor, runs } = makeFakeSupervisor(); + const fireOnExit = vi.fn(async () => {}); + const w = createCronExitWatchers({ + getProcessSupervisor: () => supervisor as never, + persistCompletion: vi.fn(async () => {}), + fireOnExit, + logger: noopLogger, + }); + w.reconcile([onExitJob("job-a")]); + await flush(); + w.reconcile([]); // cancel before the command exits + runs[0].deferred.resolve({ exitCode: 0, reason: "manual-cancel" }); + await flush(); + expect(fireOnExit).not.toHaveBeenCalled(); + }); + + it("is one-shot: a fired job is not re-armed on a later reconcile", async () => { + const { supervisor, runs } = makeFakeSupervisor(); + const w = createCronExitWatchers({ + getProcessSupervisor: () => supervisor as never, + persistCompletion: vi.fn(async () => {}), + fireOnExit: vi.fn(async () => {}), + logger: noopLogger, + }); + w.reconcile([onExitJob("job-a")]); + await flush(); + runs[0].deferred.resolve({ exitCode: 0, reason: "exit" }); + await flush(); + w.reconcile([onExitJob("job-a")]); + await flush(); + expect(supervisor.spawn).toHaveBeenCalledTimes(1); + }); +}); + +describe("resolveExitWatchShell", () => { + it("uses cmd.exe on Windows so native gateways without bash can run on-exit", () => { + const shell = resolveExitWatchShell("win32"); + expect(shell.command).toMatch(/cmd\.exe$/i); + expect(shell.argsFor("echo hi")).toEqual(["/d", "/s", "/c", "echo hi"]); + }); + + it("uses bash -lc on POSIX (unchanged from prior behavior)", () => { + expect(resolveExitWatchShell("linux").command).toBe("bash"); + expect(resolveExitWatchShell("linux").argsFor("echo hi")).toEqual(["-lc", "echo hi"]); + expect(resolveExitWatchShell("darwin").command).toBe("bash"); + }); +}); diff --git a/src/gateway/cron-exit-watchers.ts b/src/gateway/cron-exit-watchers.ts new file mode 100644 index 00000000000..1bd594a91f4 --- /dev/null +++ b/src/gateway/cron-exit-watchers.ts @@ -0,0 +1,241 @@ +import type { CronJob } from "../cron/types.js"; +import { markOpenClawExecEnv } from "../infra/openclaw-exec-env.js"; +import type { ManagedRun, ProcessSupervisor } from "../process/supervisor/index.js"; + +/** + * Safety bound for a watched command, so a hung/never-exiting command cannot + * keep a gateway-owned process alive forever. Generous (24h) because on-exit + * legitimately watches long-running commands (builds, deploys); on timeout the + * watch ends and the job fires like any other exit. + */ +const ON_EXIT_WATCH_TIMEOUT_MS = 24 * 60 * 60 * 1000; + +type Logger = { + info: (obj: unknown, msg?: string) => void; + warn: (obj: unknown, msg?: string) => void; +}; + +type OnExitCronJob = CronJob & { schedule: Extract }; + +export type CronExitResult = { + exitCode: number | null; + reason: string; + stdout: string; + stderr: string; + timedOut: boolean; + noOutputTimedOut: boolean; +}; + +export type CronExitWatchers = { + reconcile: (jobs: CronJob[]) => void; + cancel: (jobId: string) => void; + cancelAll: () => void; + activeJobIds: () => string[]; +}; + +const SCOPE_PREFIX = "cron-exit"; + +function scopeKey(jobId: string): string { + return `${SCOPE_PREFIX}:${jobId}`; +} + +function isWatchableExitJob(job: CronJob): job is OnExitCronJob { + return job.enabled && job.schedule.kind === "on-exit"; +} + +/** + * Resolve the shell used to run watched commands. Native Windows gateways use + * cmd.exe; POSIX gateways keep bash -lc. + */ +export function resolveExitWatchShell(platform: NodeJS.Platform = process.platform): { + command: string; + argsFor: (command: string) => string[]; +} { + if (platform === "win32") { + return { + command: process.env.ComSpec ?? "cmd.exe", + // /d skip AutoRun, /s strip outer quotes, /c run then exit. + argsFor: (command: string) => ["/d", "/s", "/c", command], + }; + } + return { command: "bash", argsFor: (command: string) => ["-lc", command] }; +} + +export function createCronExitWatchers(params: { + getProcessSupervisor: () => ProcessSupervisor; + persistCompletion: (jobId: string) => Promise; + fireOnExit: (job: CronJob, exit: CronExitResult) => void | Promise; + logger: Logger; + shell?: { command: string; argsFor: (command: string) => string[] }; +}): CronExitWatchers { + const shell = params.shell ?? resolveExitWatchShell(); + // jobId -> watcher state. `armToken` identifies the current arm so an async + // spawn/wait that loses ownership (the job was cancelled or re-armed for a + // changed command) becomes a no-op. The slot is reserved synchronously in + // arm() BEFORE the spawn awaits, so a concurrent cancel can act on an + // in-flight spawn. `fired` marks one-shot completion. + type WatcherSlot = { + armToken: object; + job: OnExitCronJob; + run: ManagedRun | undefined; + fired: boolean; + command: string; + cwd: string | undefined; + }; + const active = new Map(); + + const cancel = (jobId: string) => { + const slot = active.get(jobId); + if (!slot) { + return; + } + active.delete(jobId); + // Cancel an already-spawned child; an in-flight spawn (run undefined) is + // killed by the arm() ownership check once it resolves. + slot.run?.cancel("manual-cancel"); + try { + params.getProcessSupervisor().cancelScope(scopeKey(jobId), "manual-cancel"); + } catch (err) { + params.logger.warn({ err: String(err), jobId }, "cron-exit: cancel watcher failed"); + } + }; + + const arm = (job: OnExitCronJob) => { + const command = job.schedule.command; + const cwd = job.schedule.cwd; + const armToken: object = {}; + // Reserve the slot synchronously so a concurrent cancel/replace can observe + // and act on this arm before the child is spawned. + const slot: WatcherSlot = { armToken, job, run: undefined, fired: false, command, cwd }; + active.set(job.id, slot); + const owns = () => active.get(job.id) === slot && slot.armToken === armToken; + void (async () => { + let run: ManagedRun; + try { + run = await params.getProcessSupervisor().spawn({ + sessionId: `cron-exit:${job.id}`, + backendId: "cron-exit-watch", + scopeKey: scopeKey(job.id), + replaceExistingScope: true, + mode: "child", + argv: [shell.command, ...shell.argsFor(command)], + ...(cwd ? { cwd } : {}), + // Mark the child as an OpenClaw-launched subprocess (loop protection / + // detection) and bound its lifetime — consistent with how cron + // command-payload jobs run via runCommandWithTimeout. + env: markOpenClawExecEnv({ ...process.env }), + timeoutMs: ON_EXIT_WATCH_TIMEOUT_MS, + captureOutput: true, + }); + } catch (err) { + if (owns()) { + active.delete(job.id); + } + params.logger.warn({ err: String(err), jobId: job.id }, "cron-exit: watcher spawn failed"); + return; + } + if (!owns()) { + // Cancelled or re-armed (changed command/cwd) while the spawn was in + // flight — kill this now-orphaned child instead of leaking it. + run.cancel("manual-cancel"); + return; + } + slot.run = run; + params.logger.info({ jobId: job.id, runId: run.runId, command }, "cron-exit: watcher armed"); + let exit: Awaited>; + try { + exit = await run.wait(); + } catch (err) { + // run.wait() rejected (e.g. supervisor error) rather than resolving with + // an exit. Release the slot so a future reconcile can re-arm, and avoid + // an unhandled rejection. FAIL CLOSED: do not fire on an unknown outcome. + if (owns()) { + active.delete(job.id); + } + params.logger.warn( + { err: String(err), jobId: job.id }, + "cron-exit: run.wait() rejected; released watcher slot without firing", + ); + return; + } + if (!owns()) { + return; + } + params.logger.info( + { jobId: job.id, exitCode: exit.exitCode, reason: exit.reason }, + "cron-exit: watched command exited; firing job", + ); + // Persist the terminal one-shot state BEFORE firing. FAIL CLOSED: if the + // store write fails we do NOT wake — waking without a persisted terminal + // state would let a gateway restart re-arm and re-run the command. + try { + await params.persistCompletion(job.id); + } catch (err) { + if (owns()) { + active.delete(job.id); + } + params.logger.warn( + { err: String(err), jobId: job.id }, + "cron-exit: persistCompletion failed; NOT firing (fail closed to avoid replay)", + ); + return; + } + slot.fired = true; + try { + await params.fireOnExit(slot.job, { + exitCode: exit.exitCode, + reason: exit.reason, + stdout: exit.stdout, + stderr: exit.stderr, + timedOut: exit.timedOut, + noOutputTimedOut: exit.noOutputTimedOut, + }); + } catch (err) { + params.logger.warn( + { err: String(err), jobId: job.id }, + "cron-exit: fireOnExit after exit failed", + ); + } + })(); + }; + + const reconcile = (jobs: CronJob[]) => { + const want = new Map(jobs.filter(isWatchableExitJob).map((j) => [j.id, j] as const)); + // Cancel watchers whose job is gone or no longer watchable. + for (const jobId of Array.from(active.keys())) { + if (!want.has(jobId)) { + cancel(jobId); + } + } + for (const [jobId, job] of want) { + const slot = active.get(jobId); + if (slot) { + // Already tracked. A fired one-shot stays put (re-watch = re-add). If + // the watched command/cwd changed, cancel the stale watcher and re-arm. + if (slot.fired) { + continue; + } + const { command, cwd } = job.schedule; + if (slot.command === command && slot.cwd === cwd) { + slot.job = job; + continue; + } + cancel(jobId); + } + arm(job); + } + }; + + const cancelAll = () => { + for (const jobId of Array.from(active.keys())) { + cancel(jobId); + } + }; + + return { + reconcile, + cancel, + cancelAll, + activeJobIds: () => Array.from(active.keys()), + }; +} diff --git a/src/gateway/server-cron-lazy.test.ts b/src/gateway/server-cron-lazy.test.ts index 046fe6a84e2..2a0ceb1827a 100644 --- a/src/gateway/server-cron-lazy.test.ts +++ b/src/gateway/server-cron-lazy.test.ts @@ -57,6 +57,18 @@ describe("createLazyGatewayCronState", () => { expect(cron["readJob"]).toHaveBeenCalledWith("demo"); }); + it("forwards run payload overrides to the loaded cron service", async () => { + const cron = createCronService(); + hoisted.setState(createCronState(cron)); + + const lazy = createLazyGatewayCronState(createParams()); + const payload = { kind: "systemEvent" as const, text: "done" }; + await lazy.cron.run("demo", "force", { payload }); + + expect(hoisted.buildGatewayCronService).toHaveBeenCalledTimes(1); + expect(cron["run"]).toHaveBeenCalledWith("demo", "force", { payload }); + }); + it("starts the loaded cron service once", async () => { const cron = createCronService(); hoisted.setState(createCronState(cron)); @@ -121,6 +133,22 @@ describe("createLazyGatewayCronState", () => { expect(lazy.cronEnabled).toBe(false); expect(hoisted.buildGatewayCronService).not.toHaveBeenCalled(); }); + + it("does not reconcile exit watchers when cron is disabled", async () => { + const cron = createCronService(); + const reconcileExitWatchers = vi.fn(async () => {}); + hoisted.setState({ + ...createCronState(cron), + cronEnabled: false, + reconcileExitWatchers, + }); + + const lazy = createLazyGatewayCronState(createParams({ cron: { enabled: false } })); + await lazy.cron.start(); + + expect(cron["start"]).toHaveBeenCalledTimes(1); + expect(reconcileExitWatchers).not.toHaveBeenCalled(); + }); }); function createParams(overrides: Partial = {}) { diff --git a/src/gateway/server-cron-lazy.ts b/src/gateway/server-cron-lazy.ts index 5264327ee41..5390b6c0618 100644 --- a/src/gateway/server-cron-lazy.ts +++ b/src/gateway/server-cron-lazy.ts @@ -53,11 +53,17 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew } resolved.started = true; await resolved.state.cron.start(); + // Arm on-exit watchers for jobs loaded from the store at startup (no + // change event fires for already-persisted jobs). + if (resolved.state.cronEnabled) { + await resolved.state.reconcileExitWatchers?.(); + } // If stop raced the lazy import/start path, immediately stop the loaded // scheduler so shutdown does not leave a background loop alive. if (stopped && resolved.started) { resolved.started = false; resolved.state.cron.stop(); + resolved.state.stopExitWatchers?.(); } }, stop() { @@ -65,6 +71,7 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew if (loaded) { loaded.started = false; loaded.state.cron.stop(); + loaded.state.stopExitWatchers?.(); return; } if (loading) { @@ -77,6 +84,7 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew } resolved.started = false; resolved.state.cron.stop(); + resolved.state.stopExitWatchers?.(); }) .catch(() => {}); } @@ -99,8 +107,8 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew async remove(id) { return await (await load()).state.cron.remove(id); }, - async run(id, mode) { - return await (await load()).state.cron.run(id, mode); + async run(id, mode, opts) { + return await (await load()).state.cron.run(id, mode, opts); }, async enqueueRun(id, mode) { return await (await load()).state.cron.enqueueRun(id, mode); diff --git a/src/gateway/server-cron.test.ts b/src/gateway/server-cron.test.ts index 3ef102dd99f..bd112c62686 100644 --- a/src/gateway/server-cron.test.ts +++ b/src/gateway/server-cron.test.ts @@ -26,6 +26,7 @@ const { abortAndDrainEmbeddedAgentRunMock, retireSessionMcpRuntimeMock, requestSafeGatewayRestartMock, + getProcessSupervisorMock, } = vi.hoisted(() => ({ enqueueSystemEventMock: vi.fn(), consumeSelectedSystemEventEntriesMock: vi.fn((_sessionKey, entries) => entries ?? []), @@ -78,6 +79,10 @@ const { cooldownMsApplied: 0, }, })), + getProcessSupervisorMock: vi.fn(() => ({ + spawn: vi.fn(), + cancelScope: vi.fn(), + })), })); function enqueueSystemEvent(text: string, opts?: unknown) { @@ -185,7 +190,12 @@ vi.mock("../agents/agent-bundle-mcp-tools.js", () => ({ retireSessionMcpRuntime: retireSessionMcpRuntimeMock, })); -import { buildGatewayCronService } from "./server-cron.js"; +vi.mock("../process/supervisor/index.js", () => ({ + getProcessSupervisor: getProcessSupervisorMock, +})); + +import type { CronJob } from "../cron/types.js"; +import { buildGatewayCronService, fireOnExitJob } from "./server-cron.js"; function createCronConfig(name: string): OpenClawConfig { const tmpDir = path.join(os.tmpdir(), `${name}-${Date.now()}`); @@ -290,12 +300,57 @@ describe("buildGatewayCronService", () => { abortAndDrainEmbeddedAgentRunMock.mockClear(); retireSessionMcpRuntimeMock.mockClear(); requestSafeGatewayRestartMock.mockClear(); + getProcessSupervisorMock.mockReset(); + getProcessSupervisorMock.mockReturnValue({ + spawn: vi.fn(), + cancelScope: vi.fn(), + }); getGlobalHookRunnerMock.mockReturnValue({ hasHooks: (hookName: string) => hookName === "cron_changed", runCronChanged: runCronChangedMock, }); }); + it("stops on-exit watcher children when the direct cron service stops", async () => { + vi.stubEnv("OPENCLAW_SKIP_CRON", "0"); + const cancelRun = vi.fn(); + const cancelScope = vi.fn(); + const spawn = vi.fn(async () => ({ + runId: "run-on-exit", + startedAtMs: 0, + wait: () => new Promise(() => {}), + cancel: cancelRun, + })); + getProcessSupervisorMock.mockReturnValue({ spawn, cancelScope }); + const cfg = createCronConfig("server-cron-stop-exit-watchers"); + loadConfigMock.mockReturnValue(cfg); + const state = buildGatewayCronService({ + cfg, + deps: {} as CliDeps, + broadcast: () => {}, + }); + + const job = await state.cron.add({ + name: "watch build", + enabled: true, + schedule: { kind: "on-exit", command: "sleep 60" }, + payload: { kind: "systemEvent", text: "done" }, + sessionTarget: "main", + wakeMode: "next-heartbeat", + }); + await state.reconcileExitWatchers?.(); + + try { + await vi.waitFor(() => expect(spawn).toHaveBeenCalledTimes(1)); + state.cron.stop(); + expect(cancelRun).toHaveBeenCalledWith("manual-cancel"); + expect(cancelScope).toHaveBeenCalledWith(`cron-exit:${job.id}`, "manual-cancel"); + } finally { + state.cron.stop(); + vi.unstubAllEnvs(); + } + }); + it("backs off isolated cron setup timeout without gateway restart", async () => { vi.useFakeTimers(); const cfg = createCronConfig("server-cron-isolated-setup-timeout"); @@ -1620,3 +1675,64 @@ describe("buildGatewayCronService", () => { } }); }); + +describe("fireOnExitJob (on-exit fire routing)", () => { + type ForceRunMock = (jobId: string, payload?: CronJob["payload"]) => Promise; + + const job = (payload: unknown, extra: Partial = {}): CronJob => + ({ id: "job-x", payload, ...extra }) as unknown as CronJob; + const exit = { + exitCode: 3, + reason: "exit", + stdout: "built ok\n", + stderr: "warned\n", + timedOut: false, + noOutputTimedOut: false, + }; + + it("executes an agentTurn payload via the force-run path, not a text wake", async () => { + const run = vi.fn(async () => {}); + const wake = vi.fn(); + await fireOnExitJob(job({ kind: "agentTurn", message: "go" }), exit, { + run, + }); + expect(run.mock.calls[0]?.[1]).toMatchObject({ + kind: "agentTurn", + message: expect.stringContaining("Exit code: 3"), + }); + expect(run.mock.calls[0]?.[1]).toMatchObject({ + message: expect.stringContaining("stdout:\nbuilt ok"), + }); + expect(run.mock.calls[0]?.[0]).toBe("job-x"); + expect(wake).not.toHaveBeenCalled(); + }); + + it("executes a command payload via the force-run path", async () => { + const run = vi.fn(async () => {}); + const wake = vi.fn(); + await fireOnExitJob(job({ kind: "command", argv: ["echo", "hi"] }), exit, { + run, + }); + expect(run).toHaveBeenCalledWith("job-x", undefined); + expect(wake).not.toHaveBeenCalled(); + }); + + it("executes a systemEvent payload via the force-run path", async () => { + const run = vi.fn(async () => {}); + const wake = vi.fn(); + await fireOnExitJob( + job({ kind: "systemEvent", text: "done" }, { sessionKey: "sk-1", agentId: "agent-1" }), + exit, + { run }, + ); + expect(run.mock.calls[0]?.[1]).toMatchObject({ + kind: "systemEvent", + text: expect.stringContaining("Exit code: 3"), + }); + expect(run.mock.calls[0]?.[1]).toMatchObject({ + text: expect.stringContaining("stderr:\nwarned"), + }); + expect(run.mock.calls[0]?.[0]).toBe("job-x"); + expect(wake).not.toHaveBeenCalled(); + }); +}); diff --git a/src/gateway/server-cron.ts b/src/gateway/server-cron.ts index 46692140a93..824ea808ec4 100644 --- a/src/gateway/server-cron.ts +++ b/src/gateway/server-cron.ts @@ -15,7 +15,10 @@ import { import { resolveStorePath } from "../config/sessions/paths.js"; import type { AgentDefaultsConfig } from "../config/types.agent-defaults.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; -import { redactCronCommandSummaryForExternalDelivery } from "../cron/command-output-summary.js"; +import { + buildCronCommandSummary, + redactCronCommandSummaryForExternalDelivery, +} from "../cron/command-output-summary.js"; import { runCronCommandJob } from "../cron/command-runner.js"; import { resolveCronStoredDeliveryContext } from "../cron/delivery-context.js"; import { resolveCronDeliveryPlan, sendCronAnnouncePayloadStrict } from "../cron/delivery.js"; @@ -28,7 +31,7 @@ import { resolveCronSessionTargetSessionKey, } from "../cron/session-target.js"; import { resolveCronJobsStorePath } from "../cron/store.js"; -import type { CronJob } from "../cron/types.js"; +import type { CronJob, CronPayload } from "../cron/types.js"; import { formatErrorMessage } from "../infra/errors.js"; import { resolveMainScopedEventSessionKey } from "../infra/event-session-routing.js"; import { runHeartbeatOnce } from "../infra/heartbeat-runner.js"; @@ -45,6 +48,7 @@ import type { PluginHookGatewayCronService, PluginHookGatewayContext, } from "../plugins/hook-types.js"; +import { getProcessSupervisor } from "../process/supervisor/index.js"; import { normalizeAgentId, resolveEventSessionKey, @@ -52,6 +56,7 @@ import { } from "../routing/session-key.js"; import { defaultRuntime } from "../runtime.js"; import { parseAgentSessionKey } from "../sessions/session-key-utils.js"; +import { createCronExitWatchers, type CronExitResult } from "./cron-exit-watchers.js"; import { dispatchGatewayCronFinishedNotifications, sendGatewayCronFailureAlert, @@ -61,8 +66,58 @@ export type GatewayCronState = { cron: CronServiceContract; storePath: string; cronEnabled: boolean; + reconcileExitWatchers?: () => Promise; + stopExitWatchers?: () => void; }; +function formatOnExitRunSummary(exit: CronExitResult): string { + const lines = [ + "Watched command finished.", + `Exit code: ${exit.exitCode ?? "none"}`, + `Reason: ${exit.reason}`, + ]; + const output = buildCronCommandSummary({ stdout: exit.stdout, stderr: exit.stderr }); + return output ? `${lines.join("\n")}\n\nOutput:\n${output}` : lines.join("\n"); +} + +function addOnExitRunSummary(payload: CronPayload, exit: CronExitResult): CronPayload { + const summary = formatOnExitRunSummary(exit); + if (payload.kind === "systemEvent") { + return { ...payload, text: `${payload.text}\n\n${summary}` }; + } + if (payload.kind === "agentTurn") { + return { ...payload, message: `${payload.message}\n\n${summary}` }; + } + return payload; +} + +/** + * On-exit jobs use the normal force-run path so every payload kind records + * run state, history, notifications, and delivery outcomes consistently. + */ +export async function fireOnExitJob( + job: CronJob, + exit: CronExitResult, + deps: { + run: (jobId: string, payload?: CronPayload) => Promise; + }, +): Promise { + const payload = addOnExitRunSummary(job.payload, exit); + await deps.run(job.id, payload === job.payload ? undefined : payload); +} + +function reconcileCronExitWatchers(params: { + cronEnabled: boolean; + exitWatchers: ReturnType; + jobs: CronJob[]; +}) { + if (!params.cronEnabled) { + params.exitWatchers.cancelAll(); + return; + } + params.exitWatchers.reconcile(params.jobs); +} + /** Pick only the keys whose values are not `undefined` from an object. */ function pickDefined>( obj: T, @@ -319,6 +374,27 @@ export function buildGatewayCronService(params: { }); }; + // Built after cron so watcher exit callbacks can call back into the service. + const exitWatchersRef: { current: ReturnType | undefined } = { + current: undefined, + }; + const reconcileExitWatchers = async () => { + if (!exitWatchersRef.current) { + return; + } + try { + const result = await cron.list({ includeDisabled: true }); + const jobs: CronJob[] = Array.isArray(result) ? result : (result as { jobs: CronJob[] }).jobs; + reconcileCronExitWatchers({ + cronEnabled, + exitWatchers: exitWatchersRef.current, + jobs, + }); + } catch (err) { + cronLogger.warn({ err: String(err) }, "cron-exit: reconcile failed"); + } + }; + const cron = new CronService({ storePath, cronEnabled, @@ -618,6 +694,10 @@ export function buildGatewayCronService(params: { ...(hookSummary !== undefined ? { summary: hookSummary } : {}), }; runCronChangedHook(hookEvt); + // Re-arm / cancel on-exit watchers when the job set changes. + if (evt.action === "added" || evt.action === "updated" || evt.action === "removed") { + void reconcileExitWatchers(); + } if (evt.action === "finished") { const job = evt.job ?? cron.getJob(evt.jobId); dispatchGatewayCronFinishedNotifications({ @@ -666,5 +746,28 @@ export function buildGatewayCronService(params: { }, }); - return { cron, storePath, cronEnabled }; + exitWatchersRef.current = createCronExitWatchers({ + getProcessSupervisor, + persistCompletion: async (jobId) => { + await cron.update(jobId, { enabled: false }); + }, + fireOnExit: (job, exit) => + fireOnExitJob(job, exit, { + run: (jobId, payload) => cron.run(jobId, "force", payload ? { payload } : undefined), + }), + logger: cronLogger, + }); + const stopCron = cron.stop.bind(cron); + cron.stop = () => { + stopCron(); + exitWatchersRef.current?.cancelAll(); + }; + + return { + cron, + storePath, + cronEnabled, + reconcileExitWatchers, + stopExitWatchers: () => exitWatchersRef.current?.cancelAll(), + }; } diff --git a/src/gateway/server-reload-handlers.test.ts b/src/gateway/server-reload-handlers.test.ts index bb3bd5760ae..ea324f1c1d4 100644 --- a/src/gateway/server-reload-handlers.test.ts +++ b/src/gateway/server-reload-handlers.test.ts @@ -64,6 +64,13 @@ const hoisted = vi.hoisted(() => ({ clearCurrentProviderAuthState: vi.fn(() => {}), warmCurrentProviderAuthStateOffMainThread: vi.fn(async (_cfg: OpenClawConfig) => {}), disposeAllSessionMcpRuntimes: vi.fn(async () => {}), + buildGatewayCronService: vi.fn(() => ({ + cron: { start: vi.fn(async () => {}), stop: vi.fn() }, + storePath: "/tmp/rebuilt-cron.json", + cronEnabled: true, + reconcileExitWatchers: vi.fn(async () => {}), + stopExitWatchers: vi.fn(), + })), })); vi.mock("../hooks/gmail-watcher.js", () => ({ @@ -152,6 +159,14 @@ vi.mock("../agents/agent-bundle-mcp-tools.js", () => ({ disposeAllSessionMcpRuntimes: hoisted.disposeAllSessionMcpRuntimes, })); +vi.mock("./server-cron.js", async () => { + const actual = await vi.importActual("./server-cron.js"); + return { + ...actual, + buildGatewayCronService: hoisted.buildGatewayCronService, + }; +}); + function createReloadHandlersForTest( logReload = { info: vi.fn(), warn: vi.fn() }, channels?: { @@ -160,21 +175,28 @@ function createReloadHandlersForTest( }, ) { const cron = { start: vi.fn(async () => {}), stop: vi.fn() }; + const stopExitWatchers = vi.fn(); const heartbeatRunner = { stop: vi.fn(), updateConfig: vi.fn(), }; - return createGatewayReloadHandlers({ + const setState = vi.fn(); + const handlers = createGatewayReloadHandlers({ deps: {} as never, broadcast: vi.fn(), getState: () => ({ hooksConfig: {} as never, hookClientIpConfig: {} as never, heartbeatRunner: heartbeatRunner as never, - cronState: { cron, storePath: "/tmp/cron.json", cronEnabled: false } as never, + cronState: { + cron, + storePath: "/tmp/cron.json", + cronEnabled: false, + stopExitWatchers, + } as never, channelHealthMonitor: null, }), - setState: vi.fn(), + setState, startChannel: channels?.start ?? vi.fn(async () => {}), stopChannel: channels?.stop ?? vi.fn(async () => {}), stopPostReadySidecars: vi.fn(), @@ -190,6 +212,7 @@ function createReloadHandlersForTest( logReload, createHealthMonitor: () => null, }); + return { ...handlers, cron, heartbeatRunner, setState, stopExitWatchers }; } afterEach(() => { @@ -211,10 +234,54 @@ afterEach(() => { hoisted.warmCurrentProviderAuthStateOffMainThread.mockClear(); hoisted.disposeAllSessionMcpRuntimes.mockClear(); hoisted.disposeAllSessionMcpRuntimes.mockResolvedValue(undefined); + hoisted.buildGatewayCronService.mockClear(); clearSecretsRuntimeSnapshot(); }); describe("gateway hot reload model state", () => { + it("stops old cron exit watchers and reconciles rebuilt ones after cron restart", async () => { + const newCron = { start: vi.fn(async () => {}), stop: vi.fn() }; + const newReconcileExitWatchers = vi.fn(async () => {}); + const rebuiltCronState = { + cron: newCron, + storePath: "/tmp/rebuilt-cron.json", + cronEnabled: true, + reconcileExitWatchers: newReconcileExitWatchers, + stopExitWatchers: vi.fn(), + }; + hoisted.buildGatewayCronService.mockReturnValueOnce(rebuiltCronState); + const { applyHotReload, cron, setState, stopExitWatchers } = createReloadHandlersForTest(); + + await applyHotReload( + { + changedPaths: ["cron"], + restartGateway: false, + restartReasons: [], + hotReasons: ["cron"], + reloadHooks: false, + restartGmailWatcher: false, + restartCron: true, + restartHeartbeat: false, + restartHealthMonitor: false, + reloadPlugins: false, + restartChannels: new Set(), + disposeMcpRuntimes: false, + noopPaths: [], + }, + {} as OpenClawConfig, + ); + + expect(cron.stop).toHaveBeenCalledTimes(1); + expect(stopExitWatchers).toHaveBeenCalledTimes(1); + expect(newCron.start).toHaveBeenCalledTimes(1); + await vi.waitFor(() => expect(newReconcileExitWatchers).toHaveBeenCalledTimes(1)); + expect(setState).toHaveBeenCalledWith( + expect.objectContaining({ + cronState: rebuiltCronState, + }), + ); + }); + it("resets prepared model runtime state for every hot reload and rewarms after plugin reload", async () => { const reloadPlugins = vi.fn(async (): Promise => { hoisted.reloadEvents.push("reload-plugins"); diff --git a/src/gateway/server-reload-handlers.ts b/src/gateway/server-reload-handlers.ts index 3b46a8549c6..47b55fcfaee 100644 --- a/src/gateway/server-reload-handlers.ts +++ b/src/gateway/server-reload-handlers.ts @@ -465,6 +465,7 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) if (plan.restartCron) { params.onCronRestart?.(); state.cronState.cron.stop(); + state.cronState.stopExitWatchers?.(); nextState.cronState = buildGatewayCronService({ cfg: nextConfig, deps: params.deps, @@ -472,6 +473,7 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) }); startGatewayCronWithLogging({ cron: nextState.cronState.cron, + afterStart: nextState.cronState.reconcileExitWatchers, logCron: params.logCron, }); } diff --git a/src/gateway/server-runtime-services.test.ts b/src/gateway/server-runtime-services.test.ts index 980be6971fe..4d94a7691fc 100644 --- a/src/gateway/server-runtime-services.test.ts +++ b/src/gateway/server-runtime-services.test.ts @@ -60,6 +60,7 @@ const { activateGatewayScheduledServices, runGatewayPostReadyMaintenance, scheduleGatewayPostReadyMaintenance, + startGatewayCronWithLogging, startGatewayRuntimeServices, } = await import("./server-runtime-services.js"); @@ -142,6 +143,24 @@ describe("server-runtime-services", () => { expect(hoisted.stopModelPricingRefresh).toHaveBeenCalledTimes(1); }); + it("runs cron afterStart after startup succeeds", async () => { + const order: string[] = []; + const cron = { + start: vi.fn(async () => { + order.push("start"); + }), + }; + const afterStart = vi.fn(async () => { + order.push("after-start"); + }); + const logCron = { error: vi.fn() }; + + startGatewayCronWithLogging({ cron, afterStart, logCron }); + + await vi.waitFor(() => expect(order).toEqual(["start", "after-start"])); + expect(logCron.error).not.toHaveBeenCalled(); + }); + it("does not start model pricing refresh after scheduled services stop before import settles", async () => { const { services } = activateScheduledServicesForTest(); diff --git a/src/gateway/server-runtime-services.ts b/src/gateway/server-runtime-services.ts index e0b973475ec..294c116e76e 100644 --- a/src/gateway/server-runtime-services.ts +++ b/src/gateway/server-runtime-services.ts @@ -26,10 +26,12 @@ export type GatewayMaintenanceHandles = NonNullable< /** Starts cron without making gateway startup wait for cron initialization. */ export function startGatewayCronWithLogging(params: { cron: { start: () => Promise }; + afterStart?: () => Promise; logCron: { error: (message: string) => void }; }): void { void params.cron .start() + .then(() => params.afterStart?.()) .catch((err: unknown) => params.logCron.error(`failed to start: ${String(err)}`)); } diff --git a/src/plugins/hook-types.ts b/src/plugins/hook-types.ts index 49e41ae90f5..4d387519225 100644 --- a/src/plugins/hook-types.ts +++ b/src/plugins/hook-types.ts @@ -890,6 +890,11 @@ export type PluginHookGatewayCronJob = { kind: "every"; everyMs?: number; anchorMs?: number; + } + | { + kind: "on-exit"; + command?: string; + cwd?: string; }; sessionTarget?: string; wakeMode?: string; diff --git a/ui/src/ui/controllers/cron.test.ts b/ui/src/ui/controllers/cron.test.ts index 83d4d5f8a28..fd96cf5c44c 100644 --- a/ui/src/ui/controllers/cron.test.ts +++ b/ui/src/ui/controllers/cron.test.ts @@ -743,6 +743,91 @@ describe("cron controller", () => { expect(patch).not.toHaveProperty("payload"); }); + it("preserves on-exit schedules when editing Control UI metadata", async () => { + const request = vi.fn(async (method: string, _payload?: unknown) => { + if (method === "cron.update") { + return { id: "job-on-exit" }; + } + if (method === "cron.list") { + return { jobs: [{ id: "job-on-exit" }] }; + } + if (method === "cron.status") { + return { enabled: true, jobs: 1, nextWakeAtMs: null }; + } + return {}; + }); + const job = { + id: "job-on-exit", + name: "On exit", + enabled: true, + createdAtMs: 0, + updatedAtMs: 0, + schedule: { kind: "on-exit" as const, command: "make build", cwd: "/repo" }, + sessionTarget: "isolated" as const, + wakeMode: "next-heartbeat" as const, + payload: { kind: "agentTurn" as const, message: "report" }, + delivery: { mode: "none" as const }, + state: {}, + }; + const state = createState({ + client: { request } as unknown as CronState["client"], + cronJobs: [job], + }); + + startCronEdit(state, job); + state.cronForm.name = "On exit renamed"; + state.cronForm.cronExpr = ""; + await addCronJob(state); + + const updateCall = findRequestCall(request.mock.calls, "cron.update"); + const patch = requestPatch(updateCall); + expect(patch.name).toBe("On exit renamed"); + expect(patch).not.toHaveProperty("schedule"); + expect(state.cronFieldErrors).toEqual({}); + }); + + it("applies schedule edits when changing an on-exit job to a regular schedule", async () => { + const request = vi.fn(async (method: string, _payload?: unknown) => { + if (method === "cron.update") { + return { id: "job-on-exit" }; + } + if (method === "cron.list") { + return { jobs: [{ id: "job-on-exit" }] }; + } + if (method === "cron.status") { + return { enabled: true, jobs: 1, nextWakeAtMs: null }; + } + return {}; + }); + const job = { + id: "job-on-exit", + name: "On exit", + enabled: true, + createdAtMs: 0, + updatedAtMs: 0, + schedule: { kind: "on-exit" as const, command: "make build", cwd: "/repo" }, + sessionTarget: "isolated" as const, + wakeMode: "next-heartbeat" as const, + payload: { kind: "agentTurn" as const, message: "report" }, + delivery: { mode: "none" as const }, + state: {}, + }; + const state = createState({ + client: { request } as unknown as CronState["client"], + cronJobs: [job], + }); + + startCronEdit(state, job); + state.cronForm.scheduleKind = "every"; + state.cronForm.everyAmount = "5"; + state.cronForm.everyUnit = "minutes"; + await addCronJob(state); + + const updateCall = findRequestCall(request.mock.calls, "cron.update"); + const patch = requestPatch(updateCall); + expect(patch.schedule).toEqual({ kind: "every", everyMs: 300_000 }); + }); + it('keeps implicit announce delivery implicit when editing a job that shows "last" in the form', async () => { const request = vi.fn(async (method: string, _payload?: unknown) => { if (method === "cron.update") { @@ -1283,6 +1368,18 @@ describe("cron controller", () => { expect(errors.deliveryTo).toBe("cron.errors.webhookUrlInvalid"); }); + it("does not require cron expression fields for on-exit schedules", () => { + const errors = validateCronForm({ + ...DEFAULT_CRON_FORM, + name: "on exit", + scheduleKind: "on-exit", + cronExpr: "", + payloadKind: "agentTurn", + payloadText: "report", + }); + expect(errors.cronExpr).toBeUndefined(); + }); + it("blocks add/update submit when validation errors exist", async () => { const request = vi.fn(async () => ({})); const state = createState({ diff --git a/ui/src/ui/controllers/cron.ts b/ui/src/ui/controllers/cron.ts index a7935ca96c1..3f6c7bf93cf 100644 --- a/ui/src/ui/controllers/cron.ts +++ b/ui/src/ui/controllers/cron.ts @@ -44,7 +44,7 @@ export type CronFieldKey = export type CronFieldErrors = Partial>; -export type CronJobsScheduleKindFilter = "all" | "at" | "every" | "cron"; +export type CronJobsScheduleKindFilter = "all" | "at" | "every" | "cron" | "on-exit"; export type CronJobsLastStatusFilter = "all" | CronRunStatus | "unknown"; export type CronRunsLoadStatus = "ok" | "error" | "skipped"; @@ -131,7 +131,7 @@ export function validateCronForm(form: CronFormState): CronFieldErrors { if (amount <= 0) { errors.everyAmount = "cron.errors.everyAmountInvalid"; } - } else { + } else if (form.scheduleKind === "cron") { if (!form.cronExpr.trim()) { errors.cronExpr = "cron.errors.cronExprRequired"; } @@ -412,9 +412,14 @@ export function getVisibleCronJobs( }); } -function resolveCronJobScheduleKind(job: CronJob): CronJob["schedule"]["kind"] | null { +function resolveCronJobScheduleKind(job: CronJob): string | null { const scheduleKind = (job.schedule as { kind?: unknown } | null | undefined)?.kind; - if (scheduleKind === "at" || scheduleKind === "every" || scheduleKind === "cron") { + if ( + scheduleKind === "at" || + scheduleKind === "every" || + scheduleKind === "cron" || + scheduleKind === "on-exit" + ) { return scheduleKind; } return null; @@ -568,7 +573,7 @@ function jobToForm(job: CronJob, prev: CronFormState): CronFormState { const parsed = parseEverySchedule(job.schedule.everyMs); next.everyAmount = parsed.everyAmount; next.everyUnit = parsed.everyUnit; - } else { + } else if (job.schedule.kind === "cron") { next.cronExpr = job.schedule.expr; next.cronTz = job.schedule.tz ?? ""; const staggerFields = parseStaggerSchedule(job.schedule.staggerMs); @@ -576,6 +581,8 @@ function jobToForm(job: CronJob, prev: CronFormState): CronFormState { next.staggerAmount = staggerFields.staggerAmount; next.staggerUnit = staggerFields.staggerUnit; } + // Other schedule kinds (e.g. on-exit) are shown read-only in the list and have no + // editable schedule form fields; leave the cron/at/every fields at their defaults. return normalizeCronFormState(next); } @@ -712,11 +719,18 @@ export async function addCronJob(state: CronState): Promise { return; } - const schedule = buildCronSchedule(form); const editingJob = state.cronEditingJobId ? state.cronJobs.find((job) => job.id === state.cronEditingJobId) : undefined; const editingPayload = editingJob ? getCronJobPayload(editingJob) : null; + // Preserve on-exit only while the edit form still points at on-exit; if the + // user selects an editable schedule kind, the update must apply it. + const preserveSchedule = Boolean( + state.cronEditingJobId && + (editingJob?.schedule as { kind?: string } | undefined)?.kind === "on-exit" && + form.scheduleKind === "on-exit", + ); + const schedule = preserveSchedule ? undefined : buildCronSchedule(form); const preserveLockedPayload = Boolean( state.cronEditingJobId && form.payloadLocked && editingPayload?.kind === "command", ); @@ -775,12 +789,14 @@ export async function addCronJob(state: CronState): Promise { sessionKey, enabled: form.enabled, deleteAfterRun: form.deleteAfterRun, - schedule, sessionTarget: form.sessionTarget, wakeMode: form.wakeMode, delivery, failureAlert, }; + if (schedule) { + job.schedule = schedule; + } if (payload) { job.payload = payload; } diff --git a/ui/src/ui/presenter.test.ts b/ui/src/ui/presenter.test.ts new file mode 100644 index 00000000000..39d5ef81b06 --- /dev/null +++ b/ui/src/ui/presenter.test.ts @@ -0,0 +1,40 @@ +// Control UI tests cover cron schedule presentation. +import { describe, expect, it } from "vitest"; +import { formatCronSchedule } from "./presenter.ts"; +import type { CronJob } from "./types.ts"; + +function job(schedule: CronJob["schedule"]): CronJob { + return { + id: "job", + name: "Job", + enabled: true, + createdAtMs: 0, + updatedAtMs: 0, + schedule, + sessionTarget: "main", + wakeMode: "next-heartbeat", + payload: { kind: "systemEvent", text: "test" }, + }; +} + +describe("formatCronSchedule", () => { + it("formats every schedules", () => { + expect(formatCronSchedule(job({ kind: "every", everyMs: 60_000 }))).toBe("Every 1m"); + }); + + it("formats cron schedules", () => { + expect(formatCronSchedule(job({ kind: "cron", expr: "0 * * * *" }))).toBe("Cron 0 * * * *"); + }); + + it("formats on-exit schedules with the watched command instead of falling through to cron", () => { + expect(formatCronSchedule(job({ kind: "on-exit", command: "make build" }))).toBe( + "On exit: make build", + ); + }); + + it("includes the working directory for on-exit schedules when set", () => { + expect(formatCronSchedule(job({ kind: "on-exit", command: "./watch.sh", cwd: "/repo" }))).toBe( + "On exit: ./watch.sh (cwd: /repo)", + ); + }); +}); diff --git a/ui/src/ui/presenter.ts b/ui/src/ui/presenter.ts index b0be10634d4..793c17008bb 100644 --- a/ui/src/ui/presenter.ts +++ b/ui/src/ui/presenter.ts @@ -63,6 +63,11 @@ export function formatCronSchedule(job: CronJob) { if (s.kind === "every") { return `Every ${formatDurationHuman(s.everyMs)}`; } + if (s.kind === "on-exit") { + // on-exit jobs carry a watched command (+ optional cwd), not a cron expr; + // without this branch they fall through and render "Cron undefined". + return `On exit: ${s.command}${s.cwd ? ` (cwd: ${s.cwd})` : ""}`; + } return `Cron ${s.expr}${s.tz ? ` (${s.tz})` : ""}`; } diff --git a/ui/src/ui/types.ts b/ui/src/ui/types.ts index bb839b711ad..16c47aae23b 100644 --- a/ui/src/ui/types.ts +++ b/ui/src/ui/types.ts @@ -590,7 +590,8 @@ export type CronSortDir = "asc" | "desc"; export type CronSchedule = | { kind: "at"; at: string } | { kind: "every"; everyMs: number; anchorMs?: number } - | { kind: "cron"; expr: string; tz?: string; staggerMs?: number }; + | { kind: "cron"; expr: string; tz?: string; staggerMs?: number } + | { kind: "on-exit"; command: string; cwd?: string }; export type CronSessionTarget = "main" | "isolated" | "current" | `session:${string}`; export type CronWakeMode = "next-heartbeat" | "now"; diff --git a/ui/src/ui/ui-types.ts b/ui/src/ui/ui-types.ts index 180bbb57b82..57e9cc69cc4 100644 --- a/ui/src/ui/ui-types.ts +++ b/ui/src/ui/ui-types.ts @@ -49,7 +49,9 @@ export type CronFormState = { clearAgent: boolean; enabled: boolean; deleteAfterRun: boolean; - scheduleKind: "at" | "every" | "cron"; + // on-exit jobs are shown read-only in the form (the form can't edit a watched + // command); the schedule is preserved verbatim on save, never rebuilt. + scheduleKind: "at" | "every" | "cron" | "on-exit"; scheduleAt: string; everyAmount: string; everyUnit: "minutes" | "hours" | "days";