diff --git a/docs/automation/cron-jobs.md b/docs/automation/cron-jobs.md index 1d76cbfbcee8..85dd85b0f40b 100644 --- a/docs/automation/cron-jobs.md +++ b/docs/automation/cron-jobs.md @@ -122,6 +122,33 @@ This fires ~5–6 times per month instead of 0–1 times per month. OpenClaw use +### Command payloads + +Use command payloads for deterministic scripts that should run inside the Gateway scheduler without starting a model-backed isolated agent turn. Command jobs execute on the Gateway host, capture stdout/stderr, record the run in cron history, and reuse the same `announce`, `webhook`, and `none` delivery modes as isolated jobs. + + +Command cron is an operator-admin Gateway automation surface, not an agent +`tools.exec` call. Creating, updating, removing, or manually running cron jobs +requires `operator.admin`; scheduled command runs later execute inside the +Gateway process as that admin-authored automation. Agent exec policy such as +`tools.exec.mode`, approval prompts, and per-agent tool allowlists governs +model-visible exec tools, not command cron payloads. + + +```bash +openclaw cron create "*/15 * * * *" \ + --name "Queue depth probe" \ + --command "scripts/check-queue.sh" \ + --command-cwd "/srv/app" \ + --announce \ + --channel telegram \ + --to "-1001234567890" +``` + +`--command ` stores `argv: ["sh", "-lc", ]`. Use `--command-argv '["node","scripts/report.mjs"]'` when you want exact argv execution without shell parsing. Optional `--command-env KEY=VALUE`, `--command-input`, `--timeout-seconds`, `--no-output-timeout-seconds`, and `--output-max-bytes` fields control the process environment, stdin, and output bounds. + +If stdout is non-empty, that text is the delivered result. If stdout is empty and stderr is non-empty, stderr is delivered. If both streams are present, cron delivers a small `stdout:` / `stderr:` block. A zero exit code records the run as `ok`; non-zero exit, signal, timeout, or no-output timeout records `error` and can trigger failure alerts. A command that prints only `NO_REPLY` uses the normal cron silent-token suppression and posts nothing back to chat. + ### Payload options for isolated jobs @@ -246,6 +273,17 @@ Failure notifications follow a separate destination path: --webhook "https://example.invalid/openclaw/cron" ``` + + ```bash + openclaw cron create "*/15 * * * *" \ + --name "Queue depth probe" \ + --command "scripts/check-queue.sh" \ + --command-cwd "/srv/app" \ + --announce \ + --channel telegram \ + --to "-1001234567890" + ``` + ## Webhooks diff --git a/docs/cli/cron.md b/docs/cli/cron.md index 9a6d149e7a2d..d7a64f14989c 100644 --- a/docs/cli/cron.md +++ b/docs/cli/cron.md @@ -34,6 +34,27 @@ openclaw cron create "0 18 * * 1-5" \ --webhook "https://example.invalid/openclaw/cron" ``` +Use `--command` for deterministic shell-style jobs that should run inside OpenClaw cron without starting an isolated agent/model run: + + +Command cron jobs are admin-authored Gateway automation. Creating, editing, +removing, or manually running them requires `operator.admin`; the scheduled run +later executes in the Gateway process, not as an agent `tools.exec` tool call. +`tools.exec.*` and exec approvals still govern model-visible exec tools. + + +```bash +openclaw cron create "*/15 * * * *" \ + --name "Queue depth probe" \ + --command "scripts/check-queue.sh" \ + --command-cwd "/srv/app" \ + --announce \ + --channel telegram \ + --to "-1001234567890" +``` + +`--command ` stores `argv: ["sh", "-lc", ]`. Use `--command-argv '["node","scripts/report.mjs"]'` for exact argv execution. Command jobs capture stdout/stderr, record normal cron history, and route output through the same `announce`, `webhook`, or `none` delivery modes as isolated jobs. A command that prints only `NO_REPLY` is suppressed. + ## Sessions `--session` accepts `main`, `isolated`, `current`, or `session:`. @@ -92,6 +113,10 @@ Note: isolated cron runs treat run-level agent failures as job errors even when no reply payload is produced, so model/provider failures still increment error counters and trigger failure notifications. +Command cron jobs do not start an isolated agent turn. A zero exit code records +`ok`; non-zero exit, signal, timeout, or no-output timeout records `error` and +can trigger the same failure notification path. + If an isolated run times out before the first model request, `openclaw cron show` and `openclaw cron runs` include a phase-specific error such as `setup timed out before runner start` or @@ -252,6 +277,21 @@ openclaw cron create "0 7 * * *" \ `--light-context` applies to isolated agent-turn jobs only. For cron runs, lightweight mode keeps bootstrap context empty instead of injecting the full workspace bootstrap set. +Create a command job with exact argv, cwd, env, stdin, and output limits: + +```bash +openclaw cron create "*/30 * * * *" \ + --name "Position export" \ + --command-argv '["node","scripts/export-position.mjs"]' \ + --command-cwd "/srv/app" \ + --command-env "NODE_ENV=production" \ + --command-input '{"mode":"summary"}' \ + --timeout-seconds 120 \ + --no-output-timeout-seconds 30 \ + --output-max-bytes 65536 \ + --webhook "https://example.invalid/openclaw/cron" +``` + ## Common admin commands Manual run and inspection: diff --git a/packages/gateway-protocol/src/cron-validators.test.ts b/packages/gateway-protocol/src/cron-validators.test.ts index 1c4d0a0e89d1..2fbf27382cab 100644 --- a/packages/gateway-protocol/src/cron-validators.test.ts +++ b/packages/gateway-protocol/src/cron-validators.test.ts @@ -54,6 +54,36 @@ describe("cron protocol validators", () => { ).toBe(true); }); + it("accepts command cron payloads", () => { + expect( + validateCronAddParams({ + ...minimalAddParams, + sessionTarget: "isolated", + payload: { + kind: "command", + argv: ["sh", "-lc", "echo ok"], + cwd: "/srv/example", + env: { FOO: "bar" }, + input: "stdin", + timeoutSeconds: 30, + noOutputTimeoutSeconds: 5, + outputMaxBytes: 4096, + }, + }), + ).toBe(true); + expect( + validateCronUpdateParams({ + id: "job-1", + patch: { + payload: { + kind: "command", + argv: ["sh", "-lc", "echo updated"], + }, + }, + }), + ).toBe(true); + }); + it("rejects add params when required scheduling fields are missing", () => { const { wakeMode: _wakeMode, ...withoutWakeMode } = minimalAddParams; expect(validateCronAddParams(withoutWakeMode)).toBe(false); diff --git a/packages/gateway-protocol/src/schema/cron.ts b/packages/gateway-protocol/src/schema/cron.ts index d50db3e51f97..86b2741ec87d 100644 --- a/packages/gateway-protocol/src/schema/cron.ts +++ b/packages/gateway-protocol/src/schema/cron.ts @@ -27,6 +27,23 @@ function cronAgentTurnPayloadSchema(params: { message: TSchema; toolsAllow: TSch ); } +/** Builds command payload variants while preserving create/patch argv optionality. */ +function cronCommandPayloadSchema(params: { argv: TSchema }) { + return Type.Object( + { + kind: Type.Literal("command"), + argv: params.argv, + cwd: Type.Optional(Type.String({ minLength: 1 })), + env: Type.Optional(Type.Record(Type.String({ minLength: 1 }), Type.String())), + input: Type.Optional(Type.String()), + timeoutSeconds: Type.Optional(Type.Number({ minimum: 0 })), + noOutputTimeoutSeconds: Type.Optional(Type.Number({ minimum: 0 })), + outputMaxBytes: Type.Optional(Type.Integer({ minimum: 1 })), + }, + { additionalProperties: false }, + ); +} + /** Session target accepted by cron jobs. */ const CronSessionTargetSchema = Type.Union([ Type.Literal("main"), @@ -212,6 +229,9 @@ export const CronPayloadSchema = Type.Union([ message: NonEmptyString, toolsAllow: Type.Array(Type.String()), }), + cronCommandPayloadSchema({ + argv: Type.Array(NonEmptyString, { minItems: 1 }), + }), ]); /** Partial cron payload for job updates. */ @@ -227,6 +247,9 @@ export const CronPayloadPatchSchema = Type.Union([ message: Type.Optional(NonEmptyString), toolsAllow: Type.Union([Type.Array(Type.String()), Type.Null()]), }), + cronCommandPayloadSchema({ + argv: Type.Optional(Type.Array(NonEmptyString, { minItems: 1 })), + }), ]); /** Failure alert policy for repeated cron run failures. */ diff --git a/src/agents/tools/cron-tool.test.ts b/src/agents/tools/cron-tool.test.ts index 0c09bc0e3445..883daf9d4b37 100644 --- a/src/agents/tools/cron-tool.test.ts +++ b/src/agents/tools/cron-tool.test.ts @@ -637,6 +637,23 @@ describe("cron tool", () => { expect(params?.failureAlert).toBe(false); }); + it("rejects command payloads from the agent cron tool on add", async () => { + const tool = createTestCronTool(); + + await expect( + tool.execute("call-command-add", { + action: "add", + job: { + name: "command", + schedule: { at: new Date(123).toISOString() }, + sessionTarget: "isolated", + payload: { kind: "command", argv: ["sh", "-lc", "echo ok"] }, + }, + }), + ).rejects.toThrow("cron command payloads 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" }], @@ -1458,6 +1475,21 @@ describe("cron tool", () => { expect(params?.patch?.failureAlert).toBe(false); }); + it("rejects command payloads from the agent cron tool on update", async () => { + const tool = createTestCronTool(); + + await expect( + tool.execute("call-command-update", { + action: "update", + id: "job-4", + patch: { + payload: { kind: "command", argv: ["sh", "-lc", "echo ok"] }, + }, + }), + ).rejects.toThrow("cron command payloads 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 9ac0c7f396e3..0243cf68b512 100644 --- a/src/agents/tools/cron-tool.ts +++ b/src/agents/tools/cron-tool.ts @@ -339,6 +339,18 @@ function stripExistingContext(text: string) { return text.slice(0, index).trim(); } +function assertNoCronCommandPayload(value: unknown): void { + if (!isRecord(value)) { + return; + } + const payload = isRecord(value.payload) ? value.payload : undefined; + if (payload?.kind === "command") { + throw new Error( + "cron command payloads cannot be created or edited through the agent cron tool; use the CLI or Gateway API.", + ); + } +} + function truncateText(input: string, maxLen: number) { if (input.length <= maxLen) { return input; @@ -657,6 +669,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); assertCronDeliveryInputNonBlankFields(canonicalJob.delivery); const job = normalizeCronJobCreate(canonicalJob, { @@ -774,6 +787,7 @@ Use jobId canonical; id accepted compat. contextMessages (0-10) adds previous me const canonicalPatch = canonicalizeCronToolObject( params.patch as Record, ); + assertNoCronCommandPayload(canonicalPatch); assertCronDeliveryInputNonBlankFields(canonicalPatch.delivery); const patch = normalizeCronJobPatch(canonicalPatch) ?? canonicalPatch; if (recoveredFlatPatch && isEmptyRecoveredCronPatch(patch)) { diff --git a/src/cli/cron-cli.test.ts b/src/cli/cron-cli.test.ts index 81b3085017be..998774fa93ea 100644 --- a/src/cli/cron-cli.test.ts +++ b/src/cli/cron-cli.test.ts @@ -62,10 +62,17 @@ type CronUpdatePatch = { schedule?: { kind?: string; expr?: string; tz?: string; staggerMs?: number }; payload?: { kind?: string; + argv?: string[]; + cwd?: string; + env?: Record; + input?: string; message?: string; model?: string; thinking?: string; lightContext?: boolean; + timeoutSeconds?: number; + noOutputTimeoutSeconds?: number; + outputMaxBytes?: number; toolsAllow?: string[]; }; delivery?: { @@ -84,10 +91,17 @@ type CronAddParams = { schedule?: { kind?: string; at?: string; expr?: string; everyMs?: number; staggerMs?: number }; payload?: { kind?: string; + argv?: string[]; + cwd?: string; + env?: Record; + input?: string; message?: string; model?: string; thinking?: string; lightContext?: boolean; + timeoutSeconds?: number; + noOutputTimeoutSeconds?: number; + outputMaxBytes?: number; toolsAllow?: string[]; }; delivery?: { @@ -583,6 +597,57 @@ describe("cron cli", () => { expect(params?.payload?.kind).toBe("agentTurn"); }); + it("creates command cron payloads without an agent-turn message", async () => { + const params = await runCronAddAndGetParams([ + "--name", + "Shell probe", + "--every", + "10m", + "--command", + "echo ok", + "--command-cwd", + "/srv/app", + "--command-env", + "FOO=bar", + "--timeout-seconds", + "30", + "--no-output-timeout-seconds", + "5", + "--output-max-bytes", + "4096", + "--no-deliver", + ]); + + expect(params?.sessionTarget).toBe("isolated"); + expect(params?.payload).toMatchObject({ + kind: "command", + argv: ["sh", "-lc", "echo ok"], + cwd: "/srv/app", + env: { FOO: "bar" }, + timeoutSeconds: 30, + noOutputTimeoutSeconds: 5, + outputMaxBytes: 4096, + }); + expect(params?.delivery?.mode).toBe("none"); + }); + + it("rejects cron add with both message and command payloads", async () => { + await expectCronCommandExit([ + "cron", + "add", + "--name", + "Ambiguous", + "--cron", + "* * * * *", + "--message", + "hello", + "--command", + "echo ok", + ]); + + expectRuntimeErrorContaining("Choose exactly one payload"); + }); + it("supports --keep-after-run on cron add", async () => { await runCronCommand([ "cron", @@ -1007,6 +1072,56 @@ describe("cron cli", () => { expect(patch?.patch?.payload?.thinking).toBe("low"); }); + it("converts cron edit payloads to command argv", async () => { + const patch = await runCronEditAndGetPatch([ + "--command-argv", + '["node","scripts/report.mjs"," "]', + "--command-cwd", + "/srv/app", + ]); + + expect(patch?.patch?.payload).toEqual({ + kind: "command", + argv: ["node", "scripts/report.mjs", " "], + cwd: "/srv/app", + }); + }); + + it("updates command cron timeout without requiring argv to be repeated", async () => { + resetGatewayMock(); + callGatewayFromCli.mockImplementation( + async (method: string, _opts: unknown, params?: unknown) => { + if (method === "cron.status") { + return { enabled: true }; + } + if (method === "cron.list") { + return { + jobs: [ + { + ...createCronJob("job-1", "Command"), + payload: { kind: "command", argv: ["sh", "-lc", "echo ok"] }, + }, + ], + }; + } + return { ok: true, params }; + }, + ); + + const program = buildProgram(); + await program.parseAsync(["cron", "edit", "job-1", "--timeout-seconds", "120"], { + from: "user", + }); + + const patch = getGatewayCallParams<{ + patch?: { payload?: { kind?: string; timeoutSeconds?: number } }; + }>("cron.update"); + expect(patch?.patch?.payload).toEqual({ + kind: "command", + timeoutSeconds: 120, + }); + }); + it("sets and clears lightContext on cron edit", async () => { const setPatch = await runCronEditAndGetPatch(["--light-context", "--message", "hello"]); expect(setPatch?.patch?.payload?.lightContext).toBe(true); @@ -1034,7 +1149,7 @@ describe("cron cli", () => { }; }>("cron.update"); - expect(patch?.patch?.payload?.kind).toBe("agentTurn"); + expect(patch?.patch?.payload).toBeUndefined(); expect(patch?.patch?.delivery?.mode).toBe("announce"); expect(patch?.patch?.delivery?.channel).toBe("telegram"); expect(patch?.patch?.delivery?.to).toBe("19098680"); @@ -1052,7 +1167,7 @@ describe("cron cli", () => { "42", ]); - expect(patch?.patch?.payload?.kind).toBe("agentTurn"); + expect(patch?.patch?.payload).toBeUndefined(); expect(patch?.patch?.delivery?.mode).toBe("announce"); expect(patch?.patch?.delivery?.channel).toBe("telegram"); expect(patch?.patch?.delivery?.to).toBe("-100123"); @@ -1062,7 +1177,7 @@ describe("cron cli", () => { it("preserves existing delivery mode on thread-only cron edit patches", async () => { const patch = await runCronEditAndGetPatch(["--thread-id", "42"]); - expect(patch?.patch?.payload?.kind).toBe("agentTurn"); + expect(patch?.patch?.payload).toBeUndefined(); expect(patch?.patch?.delivery?.mode).toBeUndefined(); expect(patch?.patch?.delivery?.threadId).toBe(42); }); @@ -1133,7 +1248,7 @@ describe("cron cli", () => { it("updates delivery account without requiring --message on cron edit", async () => { const patch = await runCronEditAndGetPatch(["--account", " coordinator "]); - expect(patch?.patch?.payload?.kind).toBe("agentTurn"); + expect(patch?.patch?.payload).toBeUndefined(); expect(patch?.patch?.delivery?.accountId).toBe("coordinator"); expect(patch?.patch?.delivery?.mode).toBeUndefined(); }); diff --git a/src/cli/cron-cli/register.cron-add.ts b/src/cli/cron-cli/register.cron-add.ts index 974d8089a708..1c37ab1b5a80 100644 --- a/src/cli/cron-cli/register.cron-add.ts +++ b/src/cli/cron-cli/register.cron-add.ts @@ -17,6 +17,8 @@ import { coerceCronDeliveryPreviews, enrichCronJsonWithStatus, handleCronCliError, + parseCronCommandArgv, + parseCronCommandEnv, parseCronToolsAllow, printCronJson, printCronList, @@ -105,12 +107,23 @@ export function registerCronAddCommand(cron: Command) { .option("--exact", "Disable cron staggering (set stagger to 0)", false) .option("--system-event ", "System event payload (main session)") .option("--message ", "Agent message payload") + .option("--command ", "Command payload run as sh -lc on the Gateway") + .option("--command-argv ", "Command payload argv as JSON array of strings") + .option("--command-cwd ", "Working directory for command payloads") + .option( + "--command-env ", + "Environment override for command payloads (repeatable)", + (value: string, previous: string[] | undefined) => [...(previous ?? []), value], + ) + .option("--command-input ", "stdin for command payloads") .option( "--thinking ", "Thinking level for agent jobs (off|minimal|low|medium|high|xhigh)", ) .option("--model ", "Model override for agent jobs (provider/model or alias)") - .option("--timeout-seconds ", "Timeout seconds for agent jobs") + .option("--timeout-seconds ", "Timeout seconds for agent or command jobs") + .option("--no-output-timeout-seconds ", "No-output timeout seconds for command jobs") + .option("--output-max-bytes ", "Maximum captured stdout/stderr bytes for command jobs") .option("--light-context", "Use lightweight bootstrap context for agent jobs", false) .option("--tools ", "Tool allow-list (e.g. exec,read,write or exec read write)") .option("--announce", "Fallback-deliver final text to a chat", false) @@ -178,20 +191,59 @@ export function registerCronAddCommand(cron: Command) { const systemEvent = normalizeOptionalString(opts.systemEvent) ?? ""; const optionMessage = normalizeOptionalString(opts.message); const positionalMessage = normalizeOptionalString(messageArg); + const commandShell = normalizeOptionalString(opts.command); + const commandArgv = parseCronCommandArgv(opts.commandArgv); if (optionMessage && positionalMessage && optionMessage !== positionalMessage) { throw new Error( "Pass the cron job message either positionally or with --message, not both.", ); } const message = optionMessage ?? positionalMessage ?? ""; - const chosen = [Boolean(systemEvent), Boolean(message)].filter(Boolean).length; + if (commandShell && commandArgv) { + throw new Error( + "Pass command payload either with --command or --command-argv, not both.", + ); + } + const chosen = [ + Boolean(systemEvent), + Boolean(message), + Boolean(commandShell) || Boolean(commandArgv), + ].filter(Boolean).length; if (chosen !== 1) { - throw new Error("Choose exactly one payload: --system-event or --message"); + throw new Error( + "Choose exactly one payload: --system-event, --message, or --command", + ); } if (systemEvent) { return { kind: "systemEvent" as const, text: systemEvent }; } const timeoutSeconds = parsePositiveIntOrUndefined(opts.timeoutSeconds); + if (commandShell || commandArgv) { + const rawNoOutputTimeoutSeconds = + opts.noOutputTimeoutSeconds ?? + (typeof opts.outputTimeoutSeconds === "string" || + typeof opts.outputTimeoutSeconds === "number" + ? opts.outputTimeoutSeconds + : undefined); + const noOutputTimeoutSeconds = + parsePositiveIntOrUndefined(rawNoOutputTimeoutSeconds); + const outputMaxBytes = parsePositiveIntOrUndefined(opts.outputMaxBytes); + return { + kind: "command" as const, + argv: commandArgv ?? ["sh", "-lc", commandShell ?? ""], + cwd: normalizeOptionalString(opts.commandCwd), + env: parseCronCommandEnv(opts.commandEnv), + input: typeof opts.commandInput === "string" ? opts.commandInput : undefined, + timeoutSeconds: + timeoutSeconds && Number.isFinite(timeoutSeconds) ? timeoutSeconds : undefined, + noOutputTimeoutSeconds: + noOutputTimeoutSeconds && Number.isFinite(noOutputTimeoutSeconds) + ? noOutputTimeoutSeconds + : undefined, + outputMaxBytes: + outputMaxBytes && Number.isFinite(outputMaxBytes) ? outputMaxBytes : undefined, + }; + } return { kind: "agentTurn" as const, message, @@ -206,7 +258,8 @@ export function registerCronAddCommand(cron: Command) { const sessionSource = optionSource("session"); const sessionTargetRaw = normalizeOptionalString(opts.session) ?? ""; - const inferredSessionTarget = payload.kind === "agentTurn" ? "isolated" : "main"; + const inferredSessionTarget = + payload.kind === "agentTurn" || payload.kind === "command" ? "isolated" : "main"; const sessionTarget = sessionSource === "cli" ? normalizeCronSessionTargetOption(sessionTargetRaw) || "" @@ -227,17 +280,22 @@ export function registerCronAddCommand(cron: Command) { if (sessionTarget === "main" && payload.kind !== "systemEvent") { throw new Error("Main jobs require --system-event (systemEvent)."); } - if (isIsolatedLikeSessionTarget && payload.kind !== "agentTurn") { + if ( + isIsolatedLikeSessionTarget && + payload.kind !== "agentTurn" && + payload.kind !== "command" + ) { throw new Error( - "Isolated/current/custom-session jobs require --message (agentTurn).", + "Isolated/current/custom-session jobs require --message (agentTurn) or --command.", ); } if ( (opts.announce || typeof opts.deliver === "boolean") && - (!isIsolatedLikeSessionTarget || payload.kind !== "agentTurn") + (!isIsolatedLikeSessionTarget || + (payload.kind !== "agentTurn" && payload.kind !== "command")) ) { throw new Error( - "--announce/--no-deliver require a non-main agentTurn session target.", + "--announce/--no-deliver require a non-main agentTurn or command session target.", ); } @@ -252,10 +310,11 @@ export function registerCronAddCommand(cron: Command) { if ( (accountId || hasThreadId) && - (!isIsolatedLikeSessionTarget || payload.kind !== "agentTurn") + (!isIsolatedLikeSessionTarget || + (payload.kind !== "agentTurn" && payload.kind !== "command")) ) { throw new Error( - "--account and --thread-id require a non-main agentTurn job with delivery.", + "--account and --thread-id require a non-main agentTurn or command job with delivery.", ); } if (hasWebhook && hasChatDeliveryTarget) { @@ -264,7 +323,8 @@ export function registerCronAddCommand(cron: Command) { const deliveryMode = hasWebhook ? "webhook" - : isIsolatedLikeSessionTarget && payload.kind === "agentTurn" + : isIsolatedLikeSessionTarget && + (payload.kind === "agentTurn" || payload.kind === "command") ? hasAnnounce ? "announce" : hasNoDeliver @@ -288,7 +348,7 @@ export function registerCronAddCommand(cron: Command) { const sessionKey = normalizeOptionalString(opts.sessionKey); - if (payload.kind === "agentTurn" && !agentId) { + if ((payload.kind === "agentTurn" || payload.kind === "command") && !agentId) { defaultRuntime.error( theme.warn( "No --agent specified; the job will run with the configured default agent. " + diff --git a/src/cli/cron-cli/register.cron-edit.test.ts b/src/cli/cron-cli/register.cron-edit.test.ts index 5190bb191bcd..d95308284080 100644 --- a/src/cli/cron-cli/register.cron-edit.test.ts +++ b/src/cli/cron-cli/register.cron-edit.test.ts @@ -36,7 +36,7 @@ describe("cron edit command", () => { expect(help).toMatch(/also\s+implies --announce when used alone/); }); - it("keeps the documented --best-effort-deliver-only patch behavior (#83908)", async () => { + it("keeps --best-effort-deliver-only edits delivery-only (#83908)", async () => { const program = createCronProgram(); await program.parseAsync(["edit", "job-1", "--best-effort-deliver"], { from: "user" }); @@ -47,7 +47,6 @@ describe("cron edit command", () => { { id: "job-1", patch: { - payload: { kind: "agentTurn" }, delivery: { mode: "announce", bestEffort: true, @@ -57,7 +56,7 @@ describe("cron edit command", () => { ); }); - it("does not imply announce mode for --no-best-effort-deliver alone", async () => { + it("keeps --no-best-effort-deliver-only edits delivery-only", async () => { const program = createCronProgram(); await program.parseAsync(["edit", "job-1", "--no-best-effort-deliver"], { from: "user" }); @@ -68,7 +67,6 @@ describe("cron edit command", () => { { id: "job-1", patch: { - payload: { kind: "agentTurn" }, delivery: { bestEffort: false, }, @@ -76,4 +74,37 @@ describe("cron edit command", () => { }, ); }); + + it("preserves command payload kind for timeout-only edits", async () => { + callGatewayFromCli.mockImplementation(async (method: string) => { + if (method === "cron.list") { + return { + jobs: [ + { + id: "job-1", + payload: { kind: "command", argv: ["sh", "-lc", "echo ok"] }, + }, + ], + }; + } + return { ok: true }; + }); + const program = createCronProgram(); + + await program.parseAsync(["edit", "job-1", "--timeout-seconds", "12"], { from: "user" }); + + expect(callGatewayFromCli).toHaveBeenCalledWith( + "cron.update", + expect.objectContaining({ timeoutSeconds: "12" }), + { + id: "job-1", + patch: { + payload: { + kind: "command", + timeoutSeconds: 12, + }, + }, + }, + ); + }); }); diff --git a/src/cli/cron-cli/register.cron-edit.ts b/src/cli/cron-cli/register.cron-edit.ts index ddc7de439f8e..cb52571457e2 100644 --- a/src/cli/cron-cli/register.cron-edit.ts +++ b/src/cli/cron-cli/register.cron-edit.ts @@ -16,6 +16,8 @@ import { } from "./schedule-options.js"; import { getCronChannelOptions, + parseCronCommandArgv, + parseCronCommandEnv, parseCronToolsAllow, parseDurationMs, warnIfCronSchedulerDisabled, @@ -92,12 +94,23 @@ export function registerCronEditCommand(cron: Command) { .option("--exact", "Disable cron staggering (set stagger to 0)") .option("--system-event ", "Set systemEvent payload") .option("--message ", "Set agentTurn payload message") + .option("--command ", "Set command payload run as sh -lc on the Gateway") + .option("--command-argv ", "Set command payload argv as JSON array of strings") + .option("--command-cwd ", "Set command payload working directory") + .option( + "--command-env ", + "Set command payload environment overrides (repeatable)", + (value: string, previous: string[] | undefined) => [...(previous ?? []), value], + ) + .option("--command-input ", "Set command payload stdin") .option( "--thinking ", "Thinking level for agent jobs (off|minimal|low|medium|high|xhigh)", ) .option("--model ", "Model override for agent jobs") - .option("--timeout-seconds ", "Timeout seconds for agent jobs") + .option("--timeout-seconds ", "Timeout seconds for agent or command jobs") + .option("--no-output-timeout-seconds ", "No-output timeout seconds for command jobs") + .option("--output-max-bytes ", "Maximum captured stdout/stderr bytes for command jobs") .option("--light-context", "Enable lightweight bootstrap context for agent jobs") .option("--no-light-context", "Disable lightweight bootstrap context for agent jobs") .option("--tools ", "Tool allow-list (e.g. exec,read,write or exec read write)") @@ -143,9 +156,9 @@ export function registerCronEditCommand(cron: Command) { if (typeof opts.session === "string" && !sessionTarget) { throw new Error("--session must be main, isolated, current, or session:"); } - if (sessionTarget === "main" && opts.message) { + if (sessionTarget === "main" && (opts.message || opts.command || opts.commandArgv)) { throw new Error( - "Main jobs cannot use --message; use --system-event or --session isolated.", + "Main jobs cannot use --message or --command; use --system-event or --session isolated.", ); } if ( @@ -155,7 +168,7 @@ export function registerCronEditCommand(cron: Command) { opts.systemEvent ) { throw new Error( - "Isolated jobs cannot use --system-event; use --message or --session main.", + "Isolated jobs cannot use --system-event; use --message, --command, or --session main.", ); } const hasWebhookDelivery = typeof opts.webhook === "string"; @@ -240,6 +253,13 @@ export function registerCronEditCommand(cron: Command) { } const hasSystemEventPatch = typeof opts.systemEvent === "string"; + const commandShell = normalizeOptionalString(opts.command); + const commandArgv = parseCronCommandArgv(opts.commandArgv); + if (commandShell && commandArgv) { + throw new Error( + "Pass command payload either with --command or --command-argv, not both.", + ); + } const model = normalizeOptionalString(opts.model); const thinking = normalizeOptionalString(opts.thinking); const toolsAllow = parseCronToolsAllow(opts.tools); @@ -257,6 +277,20 @@ export function registerCronEditCommand(cron: Command) { if (rawTimeoutSeconds !== undefined && !hasTimeoutSeconds) { throw new Error("Invalid --timeout-seconds (must be a positive integer)."); } + const rawNoOutputTimeoutSeconds = + opts.noOutputTimeoutSeconds ?? + (typeof opts.outputTimeoutSeconds === "string" || + typeof opts.outputTimeoutSeconds === "number" + ? opts.outputTimeoutSeconds + : undefined); + const noOutputTimeoutSeconds = parseStrictPositiveInteger(rawNoOutputTimeoutSeconds); + if (rawNoOutputTimeoutSeconds !== undefined && noOutputTimeoutSeconds === undefined) { + throw new Error("Invalid --no-output-timeout-seconds (must be a positive integer)."); + } + const outputMaxBytes = parseStrictPositiveInteger(opts.outputMaxBytes); + if (opts.outputMaxBytes !== undefined && outputMaxBytes === undefined) { + throw new Error("Invalid --output-max-bytes (must be a positive integer)."); + } const hasDeliveryModeFlag = opts.announce || typeof opts.deliver === "boolean" || hasWebhookDelivery; const threadId = parseCronThreadIdOption(opts.threadId); @@ -268,23 +302,49 @@ export function registerCronEditCommand(cron: Command) { if (hasWebhookDelivery && (hasDeliveryTarget || hasDeliveryAccount)) { throw new Error("--webhook cannot be combined with chat delivery options."); } + const hasCommandSpecificPayloadField = + Boolean(commandShell) || + Boolean(commandArgv) || + typeof opts.commandCwd === "string" || + typeof opts.commandInput === "string" || + opts.commandEnv !== undefined || + noOutputTimeoutSeconds !== undefined || + outputMaxBytes !== undefined; + let timeoutOnlyPayloadKind: "agentTurn" | "command" | undefined; + if ( + hasTimeoutSeconds && + !hasCommandSpecificPayloadField && + typeof opts.message !== "string" && + !model && + !thinking && + typeof opts.lightContext !== "boolean" && + typeof opts.tools !== "string" && + !Array.isArray(opts.tools) && + !opts.clearTools + ) { + const existing = await loadCronJobForEditSchedulePatch(opts, String(id)); + timeoutOnlyPayloadKind = existing?.payload.kind === "command" ? "command" : "agentTurn"; + } const hasAgentTurnPayloadField = typeof opts.message === "string" || Boolean(model) || Boolean(thinking) || - hasTimeoutSeconds || + (hasTimeoutSeconds && + !hasCommandSpecificPayloadField && + timeoutOnlyPayloadKind !== "command") || typeof opts.lightContext === "boolean" || typeof opts.tools === "string" || Array.isArray(opts.tools) || opts.clearTools; - const hasAgentTurnPatch = - hasAgentTurnPayloadField || - Boolean(opts.announce) || - opts.deliver === true || - hasDeliveryTarget || - hasDeliveryAccount || - (hasBestEffort && !hasWebhookDelivery); - if (hasSystemEventPatch && hasAgentTurnPatch) { + const hasCommandPayloadField = + hasCommandSpecificPayloadField || + (hasTimeoutSeconds && + (hasCommandSpecificPayloadField || timeoutOnlyPayloadKind === "command")); + const hasAgentTurnPatch = hasAgentTurnPayloadField; + const hasCommandPatch = hasCommandPayloadField; + if ( + [hasSystemEventPatch, hasAgentTurnPatch, hasCommandPatch].filter(Boolean).length > 1 + ) { throw new Error("Choose at most one payload change"); } if (hasSystemEventPatch) { @@ -310,6 +370,32 @@ export function registerCronEditCommand(cron: Command) { payload.toolsAllow = toolsAllow; } patch.payload = payload; + } else if (hasCommandPatch) { + const payload: Record = { kind: "command" }; + assignIf(payload, "argv", commandArgv, Boolean(commandArgv)); + assignIf(payload, "argv", ["sh", "-lc", commandShell], Boolean(commandShell)); + assignIf( + payload, + "cwd", + normalizeOptionalString(opts.commandCwd), + typeof opts.commandCwd === "string", + ); + assignIf( + payload, + "env", + parseCronCommandEnv(opts.commandEnv), + opts.commandEnv !== undefined, + ); + assignIf(payload, "input", opts.commandInput, typeof opts.commandInput === "string"); + assignIf(payload, "timeoutSeconds", timeoutSeconds, hasTimeoutSeconds); + assignIf( + payload, + "noOutputTimeoutSeconds", + noOutputTimeoutSeconds, + noOutputTimeoutSeconds !== undefined, + ); + assignIf(payload, "outputMaxBytes", outputMaxBytes, outputMaxBytes !== undefined); + patch.payload = payload; } if (hasDeliveryModeFlag || hasDeliveryTarget || hasDeliveryAccount || hasBestEffort) { @@ -322,7 +408,7 @@ export function registerCronEditCommand(cron: Command) { : "none"; } else if ( opts.bestEffortDeliver === true || - (hasAgentTurnPayloadField && hasBestEffort) + ((hasAgentTurnPayloadField || hasCommandPayloadField) && hasBestEffort) ) { // Back-compat: best-effort true and payload edits historically implied announce mode. delivery.mode = "announce"; diff --git a/src/cli/cron-cli/shared.ts b/src/cli/cron-cli/shared.ts index b83f0537fbc3..4671ee8f8006 100644 --- a/src/cli/cron-cli/shared.ts +++ b/src/cli/cron-cli/shared.ts @@ -22,6 +22,46 @@ import { defaultRuntime, type RuntimeEnv } from "../../runtime.js"; import type { GatewayRpcOpts } from "../gateway-rpc.js"; import { callGatewayFromCli } from "../gateway-rpc.js"; +export function parseCronCommandArgv(value: unknown): string[] | undefined { + if (typeof value !== "string") { + return undefined; + } + let parsed: unknown; + try { + parsed = JSON.parse(value); + } catch { + throw new Error("--command-argv must be a JSON array of strings"); + } + if ( + !Array.isArray(parsed) || + parsed.length === 0 || + parsed.some((entry) => typeof entry !== "string" || entry.length === 0) + ) { + throw new Error("--command-argv must be a non-empty JSON array of non-empty strings"); + } + return parsed; +} + +export function parseCronCommandEnv(values: unknown): Record | undefined { + const rawValues = Array.isArray(values) ? values : typeof values === "string" ? [values] : []; + if (rawValues.length === 0) { + return undefined; + } + const env: Record = {}; + for (const raw of rawValues) { + if (typeof raw !== "string") { + throw new Error("--command-env must be KEY=VALUE"); + } + const idx = raw.indexOf("="); + const key = idx > 0 ? raw.slice(0, idx).trim() : ""; + if (!key) { + throw new Error("--command-env must be KEY=VALUE"); + } + env[key] = raw.slice(idx + 1); + } + return env; +} + export const getCronChannelOptions = () => { // Keep help truthful even before the plugin registry is bootstrapped. const pluginIds = listChannelPlugins() diff --git a/src/commands/doctor/cron/index.test.ts b/src/commands/doctor/cron/index.test.ts index cbf740a954b8..838ba6fae119 100644 --- a/src/commands/doctor/cron/index.test.ts +++ b/src/commands/doctor/cron/index.test.ts @@ -426,6 +426,23 @@ describe("maybeRepairLegacyCronStore", () => { expectNoteContaining("Cron store migrated to SQLite", "Doctor changes"); }); + it("archives legacy cron stores when an older migrated archive already exists", async () => { + const storePath = await makeTempStorePath(); + await writeCronStore(storePath, [createLegacyCronJob()]); + await fs.writeFile(`${storePath}.migrated`, "old archive", "utf-8"); + + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + + await expect(fs.stat(storePath)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(fs.readFile(`${storePath}.migrated`, "utf-8")).resolves.toBe("old archive"); + await expect(fs.stat(`${storePath}.migrated.2`)).resolves.toBeTruthy(); + expectNoteContaining("Cron store migrated to SQLite", "Doctor changes"); + }); + it("imports legacy-only jobs when SQLite already has cron rows", async () => { const storePath = await makeTempStorePath(); await writeCurrentCronStore(storePath, [ diff --git a/src/commands/doctor/cron/index.ts b/src/commands/doctor/cron/index.ts index c4a1825837e7..e6e747e4d082 100644 --- a/src/commands/doctor/cron/index.ts +++ b/src/commands/doctor/cron/index.ts @@ -336,7 +336,9 @@ export async function maybeRepairLegacyCronStore(params: { const normalized = normalizeStoredCronJobs(rawJobs); const notifyCount = rawJobs.filter((job) => job.notify === true).length; const dreamingStaleCount = countStaleDreamingJobs(rawJobs); - const previewLines = formatLegacyIssuePreview(normalized.issues); + const previewLines = formatLegacyIssuePreview(normalized.issues, { + unresolvedAgentTurnShellToolPrompt: normalized.unresolvedAgentTurnShellToolPromptJobs, + }); if (legacyStoreDetected) { previewLines.unshift( legacyImportCount > 0 diff --git a/src/commands/doctor/cron/legacy-store-migration.ts b/src/commands/doctor/cron/legacy-store-migration.ts index 58ad18411df3..19b3c1754447 100644 --- a/src/commands/doctor/cron/legacy-store-migration.ts +++ b/src/commands/doctor/cron/legacy-store-migration.ts @@ -33,9 +33,9 @@ async function archiveLegacyCronFile(filePath: string): Promise { if (!(await legacyCronFileExists(filePath))) { return; } - const archivePath = `${filePath}${LEGACY_CRON_ARCHIVE_SUFFIX}`; - if (await legacyCronFileExists(archivePath)) { - return; + let archivePath = `${filePath}${LEGACY_CRON_ARCHIVE_SUFFIX}`; + for (let index = 2; await legacyCronFileExists(archivePath); index += 1) { + archivePath = `${filePath}${LEGACY_CRON_ARCHIVE_SUFFIX}.${index}`; } await fs.rename(filePath, archivePath).catch(() => undefined); } diff --git a/src/commands/doctor/cron/payload-migration.ts b/src/commands/doctor/cron/payload-migration.ts index 76b615f759b5..a1ae5d798480 100644 --- a/src/commands/doctor/cron/payload-migration.ts +++ b/src/commands/doctor/cron/payload-migration.ts @@ -6,6 +6,39 @@ import { type UnknownRecord = Record; +type LegacyAgentTurnCommandPayload = { + command: string; + cwd?: string; + timeoutSeconds?: number; +}; + +const LEGACY_AGENT_TURN_COMMAND_MARKER_RE = /\bCommand to run\s*:/iu; +const LEGACY_AGENT_TURN_COMMAND_FIELD_RE = /^\s*-\s*(command|workdir|timeout)\s*:\s*(.*?)\s*$/iu; +const SHELL_TOOL_NAMES = new Set(["bash", "command", "exec", "process", "shell", "sh"]); +const SHELL_COMMAND_MESSAGE_RE = + /\b(?:bash|command|execute|exec|process|run|shell)\b[\s\S]{0,240}\b(?:python3?|node|bun|pnpm|npm|npx|yarn|sh|bash|sudo|cd|\.\/|\/[A-Za-z0-9._/-]+)\b/iu; +const LEGACY_DELIVERY_HINT_FIELDS = [ + "deliver", + "bestEffortDeliver", + "channel", + "provider", + "to", + "threadId", +] as const; + +function hasShellToolAccess(toolsAllow: unknown): boolean { + if (toolsAllow === undefined) { + return true; + } + if (!Array.isArray(toolsAllow)) { + return false; + } + return toolsAllow.some((tool) => { + const normalized = normalizeOptionalLowercaseString(tool); + return normalized === "*" || (normalized ? SHELL_TOOL_NAMES.has(normalized) : false); + }); +} + function toCanonicalOpenAIModelRef(value: unknown): string | undefined { const raw = readString(value); if (typeof raw !== "string") { @@ -28,6 +61,57 @@ function normalizeChannel(value: string): string { return normalizeOptionalLowercaseString(value) ?? ""; } +function parsePositiveInteger(value: string): number | undefined { + const trimmed = value.trim(); + if (!/^\d+$/u.test(trimmed)) { + return undefined; + } + const parsed = Number.parseInt(trimmed, 10); + return Number.isFinite(parsed) && parsed > 0 ? parsed : undefined; +} + +function readPositiveInteger(value: unknown): number | undefined { + return typeof value === "number" && Number.isFinite(value) && value > 0 + ? Math.floor(value) + : undefined; +} + +function parseLegacyAgentTurnCommandMessage(message: string): LegacyAgentTurnCommandPayload | null { + if (!LEGACY_AGENT_TURN_COMMAND_MARKER_RE.test(message)) { + return null; + } + + let command = ""; + let cwd: string | undefined; + let timeoutSeconds: number | undefined; + + for (const line of message.split(/\r?\n/u)) { + const match = LEGACY_AGENT_TURN_COMMAND_FIELD_RE.exec(line); + if (!match) { + continue; + } + const key = match[1]?.toLowerCase(); + const value = match[2]?.trim() ?? ""; + if (key === "command" && value && !command) { + command = value; + } else if (key === "workdir" && value && !cwd) { + cwd = value; + } else if (key === "timeout" && value && timeoutSeconds === undefined) { + timeoutSeconds = parsePositiveInteger(value); + } + } + + if (!command) { + return null; + } + + return { + command, + ...(cwd ? { cwd } : {}), + ...(timeoutSeconds ? { timeoutSeconds } : {}), + }; +} + /** Return true when a cron payload contains legacy `openai-codex/*` model refs. */ export function hasLegacyOpenAICodexCronModelRef(payload: UnknownRecord): boolean { if (toCanonicalOpenAIModelRef(payload.model)) { @@ -92,3 +176,58 @@ export function migrateLegacyCronPayload(payload: UnknownRecord): boolean { return mutated; } + +export function migrateLegacyAgentTurnCommandPayload(payload: UnknownRecord): boolean { + if (payload.kind !== "agentTurn") { + return false; + } + const message = readString(payload.message); + if (typeof message !== "string") { + return false; + } + const parsed = parseLegacyAgentTurnCommandMessage(message); + if (!parsed) { + return false; + } + if (!hasShellToolAccess(payload.toolsAllow)) { + return false; + } + + const timeoutSeconds = readPositiveInteger(payload.timeoutSeconds) ?? parsed.timeoutSeconds; + const deliveryHints: UnknownRecord = {}; + for (const key of LEGACY_DELIVERY_HINT_FIELDS) { + if (key in payload) { + deliveryHints[key] = payload[key]; + } + } + + for (const key of Object.keys(payload)) { + delete payload[key]; + } + + payload.kind = "command"; + payload.argv = ["sh", "-lc", parsed.command]; + if (parsed.cwd) { + payload.cwd = parsed.cwd; + } + if (timeoutSeconds !== undefined) { + payload.timeoutSeconds = timeoutSeconds; + } + Object.assign(payload, deliveryHints); + return true; +} + +export function hasUnresolvedAgentTurnShellToolPrompt(payload: UnknownRecord): boolean { + if (payload.kind !== "agentTurn") { + return false; + } + const message = readString(payload.message); + if (typeof message !== "string") { + return false; + } + const parsed = parseLegacyAgentTurnCommandMessage(message); + return ( + Boolean(parsed) || + (hasShellToolAccess(payload.toolsAllow) && SHELL_COMMAND_MESSAGE_RE.test(message)) + ); +} diff --git a/src/commands/doctor/cron/repair-plan.ts b/src/commands/doctor/cron/repair-plan.ts index 693339754115..579e0d510037 100644 --- a/src/commands/doctor/cron/repair-plan.ts +++ b/src/commands/doctor/cron/repair-plan.ts @@ -5,13 +5,28 @@ import { normalizeCronJobInput } from "../../../cron/normalize.js"; import type { CronJob } from "../../../cron/types.js"; export type CronLegacyIssueCounts = Partial>; +export type CronLegacyIssueDetails = { + unresolvedAgentTurnShellToolPrompt?: string[]; +}; function pluralize(count: number, noun: string) { return `${count} ${noun}${count === 1 ? "" : "s"}`; } +function formatJobNameList(names: string[] | undefined): string { + if (!names || names.length === 0) { + return ""; + } + const preview = names.slice(0, 5).map((name) => `\`${name}\``); + const remaining = names.length - preview.length; + return remaining > 0 ? `: ${preview.join(", ")} (+${remaining} more)` : `: ${preview.join(", ")}`; +} + /** Convert legacy cron issue counts into doctor preview lines. */ -export function formatLegacyIssuePreview(issues: CronLegacyIssueCounts): string[] { +export function formatLegacyIssuePreview( + issues: CronLegacyIssueCounts, + details: CronLegacyIssueDetails = {}, +): string[] { const lines: string[] = []; if (issues.jobId) { lines.push(`- ${pluralize(issues.jobId, "job")} still uses legacy \`jobId\``); @@ -38,6 +53,16 @@ export function formatLegacyIssuePreview(issues: CronLegacyIssueCounts): string[ `- ${pluralize(issues.legacyPayloadCodexModel, "job")} still uses legacy \`openai-codex/*\` cron model refs`, ); } + if (issues.legacyAgentTurnCommandPayload) { + lines.push( + `- ${pluralize(issues.legacyAgentTurnCommandPayload, "job")} uses an agent prompt to run a shell command`, + ); + } + if (issues.unresolvedAgentTurnShellToolPrompt) { + lines.push( + `- ${pluralize(issues.unresolvedAgentTurnShellToolPrompt, "job")} asks an isolated agent for shell/process tools and needs manual command conversion${formatJobNameList(details.unresolvedAgentTurnShellToolPrompt)}`, + ); + } if (issues.legacyPayloadProvider) { lines.push( `- ${pluralize(issues.legacyPayloadProvider, "job")} still uses payload \`provider\` as a delivery alias`, diff --git a/src/commands/doctor/cron/store-migration.test.ts b/src/commands/doctor/cron/store-migration.test.ts index 9af09819b3f2..37a50d3d110b 100644 --- a/src/commands/doctor/cron/store-migration.test.ts +++ b/src/commands/doctor/cron/store-migration.test.ts @@ -122,6 +122,165 @@ describe("normalizeStoredCronJobs", () => { expect(payload.fallbacks).toEqual(["anthropic/claude-opus-4.6", "openai/gpt-5.4-mini"]); }); + it("converts legacy agent command prompts into command cron payloads", () => { + const command = + "cd /home/openclaw/.razor/quant && ./scripts/system/run_position_control.sh --write-card --silent-token NO_REPLY"; + const { job, result } = normalizeOneJob( + makeLegacyJob({ + id: "quant-position-card", + schedule: { kind: "cron", expr: "*/30 * * * *", tz: "Europe/Madrid" }, + sessionTarget: "isolated", + payload: { + kind: "agentTurn", + message: [ + "Run this deterministic shell job once and report only the final JSON/status.", + "", + "Command to run:", + `- command: ${command}`, + "- workdir: /home/openclaw/.razor/quant", + "- background: false", + "- timeout: 840", + "", + "Final response contract:", + "- If the command prints exactly NO_REPLY, respond exactly NO_REPLY.", + "- Otherwise return the concise command output.", + ].join("\n"), + toolsAllow: ["bash", "process"], + lightContext: true, + timeoutSeconds: 900, + model: "openai/gpt-5.5", + deliver: true, + channel: "telegram", + to: "123", + }, + }), + ); + + expect(result.mutated).toBe(true); + expect(result.issues.legacyAgentTurnCommandPayload).toBe(1); + expect(job.delivery).toEqual({ mode: "announce", channel: "telegram", to: "123" }); + const payload = job.payload as Record; + expect(payload).toEqual({ + kind: "command", + argv: ["sh", "-lc", command], + cwd: "/home/openclaw/.razor/quant", + timeoutSeconds: 900, + }); + }); + + it("does not convert command-shaped prompts without shell tool access", () => { + const command = "python3 scripts/check_mail.py"; + const { job, result } = normalizeOneJob( + makeLegacyJob({ + id: "restricted-command-prompt", + schedule: { kind: "cron", expr: "*/30 * * * *", tz: "Europe/Madrid" }, + sessionTarget: "isolated", + payload: { + kind: "agentTurn", + message: [ + "Command to run:", + `- command: ${command}`, + "- workdir: /home/openclaw/.razor/clawd", + ].join("\n"), + toolsAllow: ["read", "message"], + }, + }), + ); + + expect(result.issues.legacyAgentTurnCommandPayload).toBeUndefined(); + expect(result.issues.unresolvedAgentTurnShellToolPrompt).toBe(1); + const payload = job.payload as Record; + expect(payload.kind).toBe("agentTurn"); + expect(payload.message).toContain(command); + expect(payload.toolsAllow).toEqual(["read", "message"]); + }); + + it("warns without converting mixed agent prompts that request shell tools", () => { + const { job, result } = normalizeOneJob( + makeLegacyJob({ + id: "mixed-agent-job", + schedule: { kind: "cron", expr: "0 9 * * *", tz: "Europe/Madrid" }, + sessionTarget: "isolated", + payload: { + kind: "agentTurn", + message: + "Run deterministic health first: python3 scripts/check_mail.py and then decide whether to send a summary.", + toolsAllow: ["bash", "read", "message"], + lightContext: true, + }, + }), + ); + + expect(result.issues.legacyAgentTurnCommandPayload).toBeUndefined(); + expect(result.issues.unresolvedAgentTurnShellToolPrompt).toBe(1); + expect(result.unresolvedAgentTurnShellToolPromptJobs).toEqual(["Legacy job"]); + const payload = job.payload as Record; + expect(payload.kind).toBe("agentTurn"); + expect(payload.message).toContain("Run deterministic health first"); + expect(payload.toolsAllow).toEqual(["bash", "read", "message"]); + }); + + it("warns on shell-style prompts with unrestricted tool access", () => { + const { result } = normalizeOneJob( + makeLegacyJob({ + id: "implicit-tools-shell-job", + schedule: { kind: "cron", expr: "0 9 * * *", tz: "Europe/Madrid" }, + sessionTarget: "isolated", + payload: { + kind: "agentTurn", + message: + "Run python3 scripts/check_mail.py and send a compact summary if anything changed.", + lightContext: true, + }, + }), + ); + + expect(result.issues.unresolvedAgentTurnShellToolPrompt).toBe(1); + expect(result.unresolvedAgentTurnShellToolPromptJobs).toEqual(["Legacy job"]); + }); + + it("warns on shell-style prompts with wildcard tool access", () => { + const { result } = normalizeOneJob( + makeLegacyJob({ + id: "wildcard-tools-shell-job", + schedule: { kind: "cron", expr: "0 9 * * *", tz: "Europe/Madrid" }, + sessionTarget: "isolated", + payload: { + kind: "agentTurn", + message: + "Execute ./scripts/check_mail.sh and send a compact summary if anything changed.", + toolsAllow: ["*"], + lightContext: true, + }, + }), + ); + + expect(result.issues.unresolvedAgentTurnShellToolPrompt).toBe(1); + expect(result.unresolvedAgentTurnShellToolPromptJobs).toEqual(["Legacy job"]); + }); + + it("does not warn on ordinary agent prompts that mention commands without shell tools", () => { + const { job, result } = normalizeOneJob( + makeLegacyJob({ + id: "ordinary-agent-job", + schedule: { kind: "cron", expr: "0 9 * * *", tz: "Europe/Madrid" }, + sessionTarget: "isolated", + payload: { + kind: "agentTurn", + message: "Explain whether the user should run python3 scripts/check_mail.py.", + toolsAllow: ["read", "message"], + lightContext: true, + }, + delivery: { mode: "announce" }, + }), + ); + + expect(result.issues.unresolvedAgentTurnShellToolPrompt).toBeUndefined(); + const payload = job.payload as Record; + expect(payload.kind).toBe("agentTurn"); + expect(payload.message).toContain("python3 scripts/check_mail.py"); + }); + it("does not report legacyPayloadKind for already-normalized payload kinds", () => { const jobs = [ { diff --git a/src/commands/doctor/cron/store-migration.ts b/src/commands/doctor/cron/store-migration.ts index d76742a467ee..f1878d031041 100644 --- a/src/commands/doctor/cron/store-migration.ts +++ b/src/commands/doctor/cron/store-migration.ts @@ -13,7 +13,12 @@ import { coerceFiniteScheduleNumber } from "../../../cron/schedule.js"; import { inferCronJobName } from "../../../cron/service/normalize.js"; import { normalizeCronStaggerMs, resolveDefaultCronStaggerMs } from "../../../cron/stagger.js"; import { normalizeLegacyDeliveryInput } from "./legacy-delivery.js"; -import { hasLegacyOpenAICodexCronModelRef, migrateLegacyCronPayload } from "./payload-migration.js"; +import { + hasUnresolvedAgentTurnShellToolPrompt, + hasLegacyOpenAICodexCronModelRef, + migrateLegacyAgentTurnCommandPayload, + migrateLegacyCronPayload, +} from "./payload-migration.js"; type CronStoreIssueKey = | "jobId" @@ -23,6 +28,8 @@ type CronStoreIssueKey = | "legacyScheduleCron" | "legacyPayloadKind" | "legacyPayloadCodexModel" + | "legacyAgentTurnCommandPayload" + | "unresolvedAgentTurnShellToolPrompt" | "legacyPayloadProvider" | "legacyTopLevelPayloadFields" | "legacyTopLevelDeliveryFields" @@ -34,6 +41,7 @@ type CronStoreIssues = Partial>; type NormalizeCronStoreJobsResult = { issues: CronStoreIssues; + unresolvedAgentTurnShellToolPromptJobs: string[]; jobs: Array>; mutated: boolean; removedJobs: Array<{ job: Record; reason: string; sourceIndex: number }>; @@ -238,6 +246,7 @@ export function normalizeStoredCronJobs( jobs: Array>, ): NormalizeCronStoreJobsResult { const issues: CronStoreIssues = {}; + const unresolvedAgentTurnShellToolPromptJobs: string[] = []; let mutated = false; const keptJobs: Array> = []; const removedJobs: NormalizeCronStoreJobsResult["removedJobs"] = []; @@ -409,6 +418,16 @@ export function normalizeStoredCronJobs( trackIssue("legacyPayloadProvider"); } } + if (migrateLegacyAgentTurnCommandPayload(payloadRecord)) { + mutated = true; + trackIssue("legacyAgentTurnCommandPayload"); + } else if (hasUnresolvedAgentTurnShellToolPrompt(payloadRecord)) { + trackIssue("unresolvedAgentTurnShellToolPrompt"); + const name = normalizeOptionalString(raw.name) ?? normalizeOptionalString(raw.id); + if (name) { + unresolvedAgentTurnShellToolPromptJobs.push(name); + } + } } const schedule = raw.schedule; @@ -545,7 +564,8 @@ export function normalizeStoredCronJobs( mutated = true; } } else { - const inferredSessionTarget = payloadKind === "agentTurn" ? "isolated" : "main"; + const inferredSessionTarget = + payloadKind === "agentTurn" || payloadKind === "command" ? "isolated" : "main"; if (raw.sessionTarget !== inferredSessionTarget) { raw.sessionTarget = inferredSessionTarget; mutated = true; @@ -553,18 +573,18 @@ export function normalizeStoredCronJobs( } const sessionTarget = normalizeOptionalLowercaseString(raw.sessionTarget) ?? ""; - const isIsolatedAgentTurn = + const isIsolatedRunnablePayload = sessionTarget === "isolated" || sessionTarget === "current" || sessionTarget.startsWith("session:") || - (sessionTarget === "" && payloadKind === "agentTurn"); + (sessionTarget === "" && (payloadKind === "agentTurn" || payloadKind === "command")); const hasDelivery = delivery && typeof delivery === "object" && !Array.isArray(delivery); const normalizedLegacy = normalizeLegacyDeliveryInput({ delivery: hasDelivery ? (delivery as Record) : null, payload: payloadRecord, }); - if (isIsolatedAgentTurn && payloadKind === "agentTurn") { + if (isIsolatedRunnablePayload && (payloadKind === "agentTurn" || payloadKind === "command")) { if (!hasDelivery && normalizedLegacy.delivery) { raw.delivery = normalizedLegacy.delivery; mutated = true; @@ -606,5 +626,5 @@ export function normalizeStoredCronJobs( jobs.splice(0, jobs.length, ...keptJobs); } - return { issues, jobs, mutated, removedJobs }; + return { issues, unresolvedAgentTurnShellToolPromptJobs, jobs, mutated, removedJobs }; } diff --git a/src/cron/command-runner.test.ts b/src/cron/command-runner.test.ts new file mode 100644 index 000000000000..207754d433e4 --- /dev/null +++ b/src/cron/command-runner.test.ts @@ -0,0 +1,157 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; +import { describe, expect, it } from "vitest"; +import { runCronCommandJob } from "./command-runner.js"; +import type { CronJob } from "./types.js"; + +function makeCommandJob(payload: Extract): CronJob { + const now = Date.now(); + return { + id: "command-job", + name: "Command job", + enabled: true, + createdAtMs: now, + updatedAtMs: now, + schedule: { kind: "every", everyMs: 60_000 }, + sessionTarget: "isolated", + wakeMode: "now", + payload, + state: {}, + }; +} + +describe("runCronCommandJob", () => { + it("runs command argv and returns stdout as the deliverable summary", async () => { + const result = await runCronCommandJob({ + job: makeCommandJob({ + kind: "command", + argv: [process.execPath, "-e", "process.stdout.write('hello from cron')"], + timeoutSeconds: 5, + }), + nowMs: () => 123, + }); + + expect(result.status).toBe("ok"); + expect(result.summary).toBe("hello from cron"); + expect(result.diagnostics?.entries[0]).toMatchObject({ + ts: 123, + source: "exec", + severity: "info", + exitCode: 0, + }); + }); + + it("preserves exact NO_REPLY stdout for outbound suppression", async () => { + const result = await runCronCommandJob({ + job: makeCommandJob({ + kind: "command", + argv: [process.execPath, "-e", "process.stdout.write('NO_REPLY\\n')"], + timeoutSeconds: 5, + }), + }); + + expect(result.status).toBe("ok"); + expect(result.summary).toBe("NO_REPLY"); + }); + + it("marks non-zero exit codes as cron errors and keeps stderr as summary", async () => { + const result = await runCronCommandJob({ + job: makeCommandJob({ + kind: "command", + argv: [process.execPath, "-e", "process.stderr.write('bad thing'); process.exit(7)"], + timeoutSeconds: 5, + }), + }); + + expect(result.status).toBe("error"); + expect(result.error).toBe("command exited with code 7"); + expect(result.summary).toBe("bad thing"); + expect(result.diagnostics?.entries[0]).toMatchObject({ + source: "exec", + severity: "error", + exitCode: 7, + }); + }); + + it("marks command timeouts as cron errors", async () => { + const result = await runCronCommandJob({ + job: makeCommandJob({ + kind: "command", + argv: [process.execPath, "-e", "setInterval(() => {}, 1000)"], + timeoutSeconds: 0.05, + }), + nowMs: () => 456, + }); + + expect(result.status).toBe("error"); + expect(result.error).toBe("command timed out"); + expect(result.diagnostics?.entries[0]).toMatchObject({ + ts: 456, + source: "exec", + severity: "error", + }); + }); + + it.skipIf(process.platform === "win32")("kills shell process groups on timeout", async () => { + const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-cron-command-")); + const markerPath = path.join(tempDir, "survived"); + const childScript = [ + `setTimeout(() => require("node:fs").writeFileSync(${JSON.stringify(markerPath)}, "alive"), 350)`, + "setInterval(() => {}, 1000)", + ].join(";"); + const shellCommand = `${JSON.stringify(process.execPath)} -e ${JSON.stringify(childScript)}`; + + const result = await runCronCommandJob({ + job: makeCommandJob({ + kind: "command", + argv: ["sh", "-lc", shellCommand], + timeoutSeconds: 0.05, + }), + }); + + expect(result.status).toBe("error"); + expect(result.error).toBe("command timed out"); + + await delay(700); + await expect(fs.access(markerPath)).rejects.toThrow(); + await fs.rm(tempDir, { recursive: true, force: true }); + }); + + it("marks no-output timeouts as cron errors", async () => { + const result = await runCronCommandJob({ + job: makeCommandJob({ + kind: "command", + argv: [process.execPath, "-e", "setInterval(() => {}, 1000)"], + timeoutSeconds: 5, + noOutputTimeoutSeconds: 0.05, + }), + }); + + expect(result.status).toBe("error"); + expect(result.error).toBe("command produced no output before noOutputTimeoutSeconds"); + expect(result.diagnostics?.entries[0]).toMatchObject({ + source: "exec", + severity: "error", + }); + }); + + it("marks aborted command runs as cron errors", async () => { + const controller = new AbortController(); + controller.abort(); + + const result = await runCronCommandJob({ + job: makeCommandJob({ + kind: "command", + argv: [process.execPath, "-e", "process.stdout.write('should not run')"], + timeoutSeconds: 5, + }), + abortSignal: controller.signal, + }); + + expect(result.status).toBe("error"); + expect(result.error).toBe("command stopped"); + expect(result.summary).toBeUndefined(); + }); +}); diff --git a/src/cron/command-runner.ts b/src/cron/command-runner.ts new file mode 100644 index 000000000000..fcce31b3996d --- /dev/null +++ b/src/cron/command-runner.ts @@ -0,0 +1,170 @@ +import { finiteSecondsToTimerSafeMilliseconds } from "@openclaw/normalization-core/number-coercion"; +import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; +import { runCommandWithTimeout } from "../process/exec.js"; +import type { CronRunDiagnostics, CronRunOutcome, CronRunStatus, CronJob } from "./types.js"; + +const DEFAULT_COMMAND_TIMEOUT_MS = 10 * 60_000; +const EFFECTIVELY_UNBOUNDED_TIMEOUT_MS = 2_147_483_647; + +function secondsToMs(value: number | undefined): number | undefined { + if (typeof value !== "number") { + return undefined; + } + if (value <= 0) { + return EFFECTIVELY_UNBOUNDED_TIMEOUT_MS; + } + return finiteSecondsToTimerSafeMilliseconds(value) ?? undefined; +} + +function formatCommand(argv: string[]): string { + return argv.map((arg) => JSON.stringify(arg)).join(" "); +} + +function trimOutput(value: string): string | undefined { + return normalizeOptionalString(value); +} + +function buildCommandSummary(params: { stdout: string; stderr: string }): string | undefined { + const stdout = trimOutput(params.stdout); + const stderr = trimOutput(params.stderr); + if (stdout && stderr) { + return `stdout:\n${stdout}\n\nstderr:\n${stderr}`; + } + return stdout ?? stderr; +} + +function commandErrorMessage(params: { + code: number | null; + signal: NodeJS.Signals | null; + termination: string; +}): string { + if (params.termination === "timeout") { + return "command timed out"; + } + if (params.termination === "no-output-timeout") { + return "command produced no output before noOutputTimeoutSeconds"; + } + if (params.termination === "signal") { + return params.signal ? `command stopped by signal ${params.signal}` : "command stopped"; + } + if (typeof params.code === "number") { + return `command exited with code ${params.code}`; + } + return "command failed"; +} + +function buildDiagnostics(params: { + command: string; + status: CronRunStatus; + summary?: string; + code: number | null; + signal: NodeJS.Signals | null; + stdoutTruncatedBytes?: number; + stderrTruncatedBytes?: number; + nowMs: () => number; +}): CronRunDiagnostics { + const truncated = + Boolean(params.stdoutTruncatedBytes && params.stdoutTruncatedBytes > 0) || + Boolean(params.stderrTruncatedBytes && params.stderrTruncatedBytes > 0); + return { + ...(params.summary ? { summary: params.summary } : {}), + entries: [ + { + ts: params.nowMs(), + source: "exec", + severity: params.status === "ok" ? "info" : "error", + message: params.summary + ? `command ${params.status}: ${params.command}` + : `command ${params.status} with no output: ${params.command}`, + exitCode: params.code, + truncated, + ...(params.signal ? { toolName: `signal:${params.signal}` } : {}), + }, + ], + }; +} + +/** Executes a cron command payload without starting an agent/model run. */ +export async function runCronCommandJob(params: { + job: CronJob; + abortSignal?: AbortSignal; + nowMs?: () => number; +}): Promise { + const nowMs = params.nowMs ?? Date.now; + const { payload } = params.job; + if (payload.kind !== "command") { + return { + status: "skipped", + error: 'command runner requires payload.kind="command"', + }; + } + if (!Array.isArray(payload.argv) || payload.argv.length === 0) { + return { + status: "skipped", + error: 'command payload requires non-empty "argv"', + }; + } + + const command = formatCommand(payload.argv); + const noOutputTimeoutMs = secondsToMs(payload.noOutputTimeoutSeconds); + try { + const result = await runCommandWithTimeout(payload.argv, { + timeoutMs: secondsToMs(payload.timeoutSeconds) ?? DEFAULT_COMMAND_TIMEOUT_MS, + ...(payload.cwd ? { cwd: payload.cwd } : {}), + ...(payload.input !== undefined ? { input: payload.input } : {}), + ...(payload.env ? { env: payload.env } : {}), + ...(noOutputTimeoutMs !== undefined ? { noOutputTimeoutMs } : {}), + ...(payload.outputMaxBytes !== undefined ? { maxOutputBytes: payload.outputMaxBytes } : {}), + ...(params.abortSignal ? { signal: params.abortSignal } : {}), + killProcessTree: true, + }); + const ok = + result.code === 0 && + !result.killed && + result.termination !== "timeout" && + result.termination !== "no-output-timeout" && + result.termination !== "signal"; + const status: CronRunStatus = ok ? "ok" : "error"; + const summary = buildCommandSummary({ stdout: result.stdout, stderr: result.stderr }); + const error = ok + ? undefined + : commandErrorMessage({ + code: result.code, + signal: result.signal, + termination: result.termination, + }); + return { + status, + ...(error ? { error } : {}), + ...(summary ? { summary } : {}), + diagnostics: buildDiagnostics({ + command, + status, + summary, + code: result.code, + signal: result.signal, + stdoutTruncatedBytes: result.stdoutTruncatedBytes, + stderrTruncatedBytes: result.stderrTruncatedBytes, + nowMs, + }), + }; + } catch (err) { + const error = err instanceof Error ? err.message : String(err); + return { + status: "error", + error, + diagnostics: { + summary: error, + entries: [ + { + ts: nowMs(), + source: "exec", + severity: "error", + message: `command failed to start: ${command}: ${error}`, + exitCode: null, + }, + ], + }, + }; + } +} diff --git a/src/cron/delivery-plan.ts b/src/cron/delivery-plan.ts index 0fbb76509183..00b31d06b652 100644 --- a/src/cron/delivery-plan.ts +++ b/src/cron/delivery-plan.ts @@ -98,15 +98,15 @@ export function resolveCronDeliveryPlan(job: CronJob): CronDeliveryPlan { }; } - const isIsolatedAgentTurn = - job.payload.kind === "agentTurn" && + const isDetachedOutputJob = + (job.payload.kind === "agentTurn" || job.payload.kind === "command") && typeof job.sessionTarget === "string" && (job.sessionTarget === "isolated" || job.sessionTarget === "current" || job.sessionTarget.startsWith("session:")); - // Isolated/current/session cron jobs default to announce delivery so their - // output reaches the initiating session unless the job opts out. - const resolvedMode = isIsolatedAgentTurn ? "announce" : "none"; + // Isolated/current/session output jobs default to announce delivery so their + // result reaches the initiating session unless the job opts out. + const resolvedMode = isDetachedOutputJob ? "announce" : "none"; return { mode: resolvedMode, diff --git a/src/cron/normalize.test.ts b/src/cron/normalize.test.ts index 9ac2a7bd5139..b2a427d8fefe 100644 --- a/src/cron/normalize.test.ts +++ b/src/cron/normalize.test.ts @@ -361,6 +361,52 @@ describe("normalizeCronJobCreate", () => { expect(delivery.mode).toBe("announce"); }); + it("defaults command payloads to isolated announce jobs", () => { + const normalized = normalizeCronJobCreate({ + name: "command default", + schedule: { kind: "every", everyMs: 60_000 }, + payload: { + kind: "command", + argv: ["sh", "-lc", "echo ok"], + cwd: " /srv/example ", + env: { FOO: "bar" }, + timeoutSeconds: 30, + noOutputTimeoutSeconds: 5, + outputMaxBytes: 4096, + }, + }) as unknown as Record; + + expect(normalized.sessionTarget).toBe("isolated"); + expect((normalized.delivery as Record).mode).toBe("announce"); + expect(normalized.payload).toEqual({ + kind: "command", + argv: ["sh", "-lc", "echo ok"], + cwd: "/srv/example", + env: { FOO: "bar" }, + timeoutSeconds: 30, + noOutputTimeoutSeconds: 5, + outputMaxBytes: 4096, + }); + expect(validateCronAddParams(normalized)).toBe(true); + }); + + it("preserves command argv argument bytes", () => { + const normalized = normalizeCronJobCreate({ + name: "command exact argv", + schedule: { kind: "every", everyMs: 60_000 }, + payload: { + kind: "command", + argv: ["printf", "%s", " padded value "], + }, + }) as unknown as Record; + + expect(normalized.payload).toMatchObject({ + kind: "command", + argv: ["printf", "%s", " padded value "], + }); + expect(validateCronAddParams(normalized)).toBe(true); + }); + it("preserves timeoutSeconds=0 for no-timeout agentTurn payloads", () => { const normalized = normalizeCronJobCreate({ name: "no-timeout", diff --git a/src/cron/normalize.ts b/src/cron/normalize.ts index b6f318673390..a5daa9cccc5d 100644 --- a/src/cron/normalize.ts +++ b/src/cron/normalize.ts @@ -53,6 +53,32 @@ function normalizeTrimmedStringArray( return undefined; } +function normalizeTrimmedStringRecord(value: unknown): Record | undefined { + if (!isRecord(value)) { + return undefined; + } + const entries: Array<[string, string]> = []; + for (const [rawKey, rawValue] of Object.entries(value)) { + const key = normalizeOptionalString(rawKey); + const val = typeof rawValue === "string" ? rawValue : undefined; + if (!key || val === undefined) { + return undefined; + } + entries.push([key, val]); + } + return Object.fromEntries(entries); +} + +function normalizeCommandArgv(value: unknown): string[] | undefined { + if (!Array.isArray(value) || value.length === 0) { + return undefined; + } + if (value.some((entry) => typeof entry !== "string" || entry.length === 0)) { + return undefined; + } + return [...value]; +} + function coerceSchedule(schedule: UnknownRecord) { const next: UnknownRecord = { ...schedule }; const rawKind = normalizeLowercaseStringOrEmpty(schedule.kind); @@ -122,6 +148,8 @@ function coercePayload(payload: UnknownRecord) { next.kind = "agentTurn"; } else if (kindRaw === "systemevent") { next.kind = "systemEvent"; + } else if (kindRaw === "command") { + next.kind = "command"; } else if (kindRaw) { next.kind = kindRaw; } @@ -181,6 +209,52 @@ function coercePayload(payload: UnknownRecord) { delete next.toolsAllow; } } + if ("argv" in next) { + const argv = normalizeCommandArgv(next.argv); + if (Array.isArray(argv) && argv.length > 0) { + next.argv = argv; + } else { + delete next.argv; + } + } + if ("cwd" in next) { + const cwd = parseOptionalField(TrimmedNonEmptyStringFieldSchema, next.cwd); + if (cwd !== undefined) { + next.cwd = cwd; + } else { + delete next.cwd; + } + } + if ("env" in next) { + const env = normalizeTrimmedStringRecord(next.env); + if (env !== undefined) { + next.env = env; + } else { + delete next.env; + } + } + if ("input" in next && typeof next.input !== "string") { + delete next.input; + } + if ("noOutputTimeoutSeconds" in next) { + const noOutputTimeoutSeconds = parseOptionalField( + TimeoutSecondsFieldSchema, + next.noOutputTimeoutSeconds, + ); + if (noOutputTimeoutSeconds !== undefined) { + next.noOutputTimeoutSeconds = noOutputTimeoutSeconds; + } else { + delete next.noOutputTimeoutSeconds; + } + } + if ("outputMaxBytes" in next) { + const outputMaxBytes = parseOptionalField(TimeoutSecondsFieldSchema, next.outputMaxBytes); + if (outputMaxBytes !== undefined && outputMaxBytes > 0) { + next.outputMaxBytes = Math.floor(outputMaxBytes); + } else { + delete next.outputMaxBytes; + } + } if ( "allowUnsafeExternalContent" in next && typeof next.allowUnsafeExternalContent !== "boolean" @@ -196,8 +270,29 @@ function coercePayload(payload: UnknownRecord) { delete next.lightContext; delete next.allowUnsafeExternalContent; delete next.toolsAllow; + delete next.argv; + delete next.cwd; + delete next.env; + delete next.input; + delete next.noOutputTimeoutSeconds; + delete next.outputMaxBytes; } else if (next.kind === "agentTurn") { delete next.text; + delete next.argv; + delete next.cwd; + delete next.env; + delete next.input; + delete next.noOutputTimeoutSeconds; + delete next.outputMaxBytes; + } else if (next.kind === "command") { + delete next.text; + delete next.message; + delete next.model; + delete next.fallbacks; + delete next.thinking; + delete next.lightContext; + delete next.allowUnsafeExternalContent; + delete next.toolsAllow; } return next; } @@ -468,7 +563,12 @@ export function normalizeCronJobInput( ) { next.name = inferCronJobName({ schedule: next.schedule as { kind?: unknown; everyMs?: unknown; expr?: unknown }, - payload: next.payload as { kind?: unknown; text?: unknown; message?: unknown }, + payload: next.payload as { + kind?: unknown; + text?: unknown; + message?: unknown; + argv?: unknown; + }, }); } else if (typeof next.name === "string") { const trimmed = next.name.trim(); @@ -482,7 +582,7 @@ export function normalizeCronJobInput( // turns isolate by default to avoid unbounded token accumulation. if (kind === "systemEvent") { next.sessionTarget = "main"; - } else if (kind === "agentTurn") { + } else if (kind === "agentTurn" || kind === "command") { next.sessionTarget = "isolated"; } } @@ -522,13 +622,17 @@ export function normalizeCronJobInput( const sessionTarget = typeof next.sessionTarget === "string" ? next.sessionTarget : ""; // Resolved "current" and custom session ids still use isolated-agent // delivery semantics, so they get the same default announce behavior. - const isIsolatedAgentTurn = + const isDetachedDeliveryJob = sessionTarget === "isolated" || sessionTarget === "current" || sessionTarget.startsWith("session:") || - (sessionTarget === "" && payloadKind === "agentTurn"); + (sessionTarget === "" && (payloadKind === "agentTurn" || payloadKind === "command")); const hasDelivery = "delivery" in next && next.delivery !== undefined; - if (!hasDelivery && isIsolatedAgentTurn && payloadKind === "agentTurn") { + if ( + !hasDelivery && + isDetachedDeliveryJob && + (payloadKind === "agentTurn" || payloadKind === "command") + ) { next.delivery = { mode: "announce" }; } } diff --git a/src/cron/persisted-shape.ts b/src/cron/persisted-shape.ts index 25f16bb1d445..6293ced3f929 100644 --- a/src/cron/persisted-shape.ts +++ b/src/cron/persisted-shape.ts @@ -58,7 +58,7 @@ export function getInvalidPersistedCronJobReason( } const payloadRecord = payload as Record; const payloadKind = payloadRecord.kind; - if (payloadKind !== "systemEvent" && payloadKind !== "agentTurn") { + if (payloadKind !== "systemEvent" && payloadKind !== "agentTurn" && payloadKind !== "command") { return "invalid-payload"; } if (payloadKind === "systemEvent") { @@ -73,5 +73,15 @@ export function getInvalidPersistedCronJobReason( return "invalid-payload"; } } + if (payloadKind === "command") { + const argv = payloadRecord.argv; + if ( + !Array.isArray(argv) || + argv.length === 0 || + argv.some((value) => typeof value !== "string" || value.length === 0) + ) { + return "invalid-payload"; + } + } return null; } diff --git a/src/cron/service/initial-delivery.ts b/src/cron/service/initial-delivery.ts index 2fbc3c2f7210..f9efd37337f2 100644 --- a/src/cron/service/initial-delivery.ts +++ b/src/cron/service/initial-delivery.ts @@ -6,7 +6,10 @@ export function resolveInitialCronDelivery(input: CronJobCreate): CronDelivery | if (input.delivery) { return input.delivery; } - if (input.sessionTarget === "isolated" && input.payload.kind === "agentTurn") { + if ( + input.sessionTarget === "isolated" && + (input.payload.kind === "agentTurn" || input.payload.kind === "command") + ) { return { mode: "announce" }; } return undefined; diff --git a/src/cron/service/jobs.ts b/src/cron/service/jobs.ts index fd68502a206a..97ab6ffdd126 100644 --- a/src/cron/service/jobs.ts +++ b/src/cron/service/jobs.ts @@ -286,8 +286,10 @@ export function assertSupportedJobSpec(job: Pick = { ...existing }; + if (Array.isArray(patch.argv)) { + next.argv = patch.argv; + } + if (typeof patch.cwd === "string") { + next.cwd = patch.cwd; + } + if (patch.env && typeof patch.env === "object" && !Array.isArray(patch.env)) { + next.env = patch.env; + } + if (typeof patch.input === "string") { + next.input = patch.input; + } + if (typeof patch.timeoutSeconds === "number") { + next.timeoutSeconds = patch.timeoutSeconds; + } + if (typeof patch.noOutputTimeoutSeconds === "number") { + next.noOutputTimeoutSeconds = patch.noOutputTimeoutSeconds; + } + if (typeof patch.outputMaxBytes === "number") { + next.outputMaxBytes = patch.outputMaxBytes; + } + return next; + } + if (existing.kind !== "agentTurn") { return buildPayloadFromPatch(patch); } @@ -924,6 +955,22 @@ function buildPayloadFromPatch(patch: CronPayloadPatch): CronPayload { return { kind: "systemEvent", text: patch.text }; } + if (patch.kind === "command") { + if (!Array.isArray(patch.argv) || patch.argv.length === 0) { + throw new Error('cron.update payload.kind="command" requires argv'); + } + return { + kind: "command", + argv: patch.argv, + cwd: patch.cwd, + env: patch.env, + input: patch.input, + timeoutSeconds: patch.timeoutSeconds, + noOutputTimeoutSeconds: patch.noOutputTimeoutSeconds, + outputMaxBytes: patch.outputMaxBytes, + }; + } + if (typeof patch.message !== "string" || patch.message.length === 0) { throw new Error('cron.update payload.kind="agentTurn" requires message'); } diff --git a/src/cron/service/normalize.ts b/src/cron/service/normalize.ts index 3bf021e7adf9..733d95f49e64 100644 --- a/src/cron/service/normalize.ts +++ b/src/cron/service/normalize.ts @@ -35,14 +35,16 @@ export function normalizeOptionalAgentId(raw: unknown) { /** Infers a compact cron job name from payload text first, then schedule shape. */ export function inferCronJobName(job: { schedule?: { kind?: unknown; everyMs?: unknown; expr?: unknown }; - payload?: { kind?: unknown; text?: unknown; message?: unknown }; + payload?: { kind?: unknown; text?: unknown; message?: unknown; argv?: unknown }; }) { const text = job?.payload?.kind === "systemEvent" && typeof job.payload.text === "string" ? job.payload.text : job?.payload?.kind === "agentTurn" && typeof job.payload.message === "string" ? job.payload.message - : ""; + : job?.payload?.kind === "command" && Array.isArray(job.payload.argv) + ? job.payload.argv.join(" ") + : ""; const firstLine = text .split("\n") @@ -72,5 +74,7 @@ export function normalizePayloadToSystemText(payload: CronPayload) { if (payload.kind === "systemEvent") { return typeof payload.text === "string" ? payload.text.trim() : ""; } - return typeof payload.message === "string" ? payload.message.trim() : ""; + return payload.kind === "agentTurn" && typeof payload.message === "string" + ? payload.message.trim() + : ""; } diff --git a/src/cron/service/state.ts b/src/cron/service/state.ts index e77dea333785..525687ffad5c 100644 --- a/src/cron/service/state.ts +++ b/src/cron/service/state.ts @@ -135,6 +135,13 @@ export type CronServiceDeps = { } & CronRunOutcome & CronRunTelemetry >; + runCommandJob?: (params: { job: CronJob; abortSignal?: AbortSignal }) => Promise< + { + delivered?: boolean; + deliveryAttempted?: boolean; + delivery?: CronDeliveryTrace; + } & CronRunOutcome + >; cleanupTimedOutAgentRun?: (params: { job: CronJob; timeoutMs: number; diff --git a/src/cron/service/timeout-policy.ts b/src/cron/service/timeout-policy.ts index 50810c28ae1b..a949e6356150 100644 --- a/src/cron/service/timeout-policy.ts +++ b/src/cron/service/timeout-policy.ts @@ -15,10 +15,11 @@ export const DEFAULT_JOB_TIMEOUT_MS = 10 * 60_000; // 10 minutes */ export const AGENT_TURN_SAFETY_TIMEOUT_MS = 60 * 60_000; // 60 minutes -/** Resolves the wall-clock timeout for a cron job, including explicit agent-turn overrides. */ +/** Resolves the wall-clock timeout for a cron job, including explicit detached-run overrides. */ export function resolveCronJobTimeoutMs(job: CronJob): number | undefined { const configuredTimeoutMs = - job.payload.kind === "agentTurn" && typeof job.payload.timeoutSeconds === "number" + (job.payload.kind === "agentTurn" || job.payload.kind === "command") && + typeof job.payload.timeoutSeconds === "number" ? (finiteSecondsToTimerSafeMilliseconds(job.payload.timeoutSeconds) ?? 0) : undefined; if (configuredTimeoutMs === undefined) { diff --git a/src/cron/service/timer.test.ts b/src/cron/service/timer.test.ts index 969db66943c5..4d9a7f69ff78 100644 --- a/src/cron/service/timer.test.ts +++ b/src/cron/service/timer.test.ts @@ -47,6 +47,22 @@ function createDueIsolatedAgentJob(params: { now: number }): CronJob { }; } +function createDueCommandJob(params: { now: number }): CronJob { + return { + id: "command-job", + agentId: "finn", + name: "command job", + enabled: true, + createdAtMs: params.now - 60_000, + updatedAtMs: params.now - 60_000, + schedule: { kind: "every", everyMs: 60_000, anchorMs: params.now - 60_000 }, + sessionTarget: "isolated", + wakeMode: "now", + payload: { kind: "command", argv: ["sh", "-lc", "echo ok"] }, + state: { nextRunAtMs: params.now - 1 }, + }; +} + afterEach(() => { resetTaskRegistryForTests(); }); @@ -184,6 +200,36 @@ describe("cron service timer seam coverage", () => { timeoutSpy.mockRestore(); }); + it("runs command cron jobs without isolated agent setup", async () => { + const { storePath } = await makeStorePath(); + const now = Date.parse("2026-03-23T12:00:00.000Z"); + const runIsolatedAgentJob = vi.fn(async () => ({ status: "ok" as const })); + const runCommandJob = vi.fn(async () => ({ + status: "ok" as const, + summary: "command ok", + })); + const state = createCronServiceState({ + storePath, + cronEnabled: true, + log: logger, + nowMs: () => now, + enqueueSystemEvent: vi.fn(), + requestHeartbeat: vi.fn(), + runIsolatedAgentJob, + runCommandJob, + }); + const job = createDueCommandJob({ now }); + + const result = await executeJobCore(state, job); + + expect(result).toMatchObject({ status: "ok", summary: "command ok" }); + expect(runCommandJob).toHaveBeenCalledWith({ + job, + abortSignal: undefined, + }); + expect(runIsolatedAgentJob).not.toHaveBeenCalled(); + }); + it("records isolated cron task runs against the backing cron session", async () => { const { storePath } = await makeStorePath(); const now = Date.parse("2026-03-23T12:00:00.000Z"); diff --git a/src/cron/service/timer.ts b/src/cron/service/timer.ts index 9cbd8c108e9f..ee7baabcd72b 100644 --- a/src/cron/service/timer.ts +++ b/src/cron/service/timer.ts @@ -1516,8 +1516,45 @@ async function executeDetachedCronJob( delivery?: CronDeliveryTrace; } > { + if (job.payload.kind === "command") { + if (!state.deps.runCommandJob) { + const error = "cron command runner is not configured"; + return { + status: "skipped", + error, + diagnostics: createCronRunDiagnosticsFromError("cron-preflight", error, { + severity: "warn", + nowMs: state.deps.nowMs, + }), + }; + } + const res = await state.deps.runCommandJob({ + job, + abortSignal, + }); + if (abortSignal?.aborted) { + const error = abortErrorMessage(abortSignal); + return { + status: "error", + error, + diagnostics: createCronRunDiagnosticsFromError("cron-setup", error, { + nowMs: state.deps.nowMs, + }), + }; + } + return { + status: res.status, + error: res.error, + summary: res.summary, + delivered: res.delivered, + deliveryAttempted: res.deliveryAttempted, + delivery: res.delivery, + diagnostics: res.diagnostics, + }; + } + if (job.payload.kind !== "agentTurn") { - const error = "isolated job requires payload.kind=agentTurn"; + const error = 'isolated job requires payload.kind="agentTurn" or "command"'; return { status: "skipped", error, diff --git a/src/cron/store.test.ts b/src/cron/store.test.ts index 4ace264f876c..a059c6562bd1 100644 --- a/src/cron/store.test.ts +++ b/src/cron/store.test.ts @@ -522,6 +522,35 @@ describe("cron store", () => { }); }); + it("round-trips command payloads through SQLite", async () => { + const store = await makeStorePath(); + const payload = makeStore("command-job", true); + payload.jobs[0].sessionTarget = "isolated"; + payload.jobs[0].payload = { + kind: "command", + argv: ["sh", "-lc", 'printf %s "$1"', " "], + cwd: "/srv/example", + env: { FOO: "bar" }, + input: "stdin", + timeoutSeconds: 45, + noOutputTimeoutSeconds: 10, + outputMaxBytes: 4096, + }; + + await saveCronStore(store.storePath, payload); + + expect((await loadCronStore(store.storePath)).jobs[0]?.payload).toEqual({ + kind: "command", + argv: ["sh", "-lc", 'printf %s "$1"', " "], + cwd: "/srv/example", + env: { FOO: "bar" }, + input: "stdin", + timeoutSeconds: 45, + noOutputTimeoutSeconds: 10, + outputMaxBytes: 4096, + }); + }); + it("round-trips completion destinations through SQLite delivery columns", async () => { const { storePath } = await makeStorePath(); const job = makeStore("sqlite-webhook-delivery-job", true).jobs[0]; diff --git a/src/cron/store/payload-codec.ts b/src/cron/store/payload-codec.ts index f9590bbbca2b..de62fd9fbe20 100644 --- a/src/cron/store/payload-codec.ts +++ b/src/cron/store/payload-codec.ts @@ -15,6 +15,51 @@ function parseExternalContentSource(raw: string | null): "gmail" | "webhook" | u return parsed === "gmail" || parsed === "webhook" ? parsed : undefined; } +function parseCommandPayloadMessage( + raw: string | null, +): Omit, "kind" | "timeoutSeconds"> | null { + const parsed = raw ? parseJsonValue(raw, undefined) : undefined; + if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) { + return null; + } + const record = parsed as Record; + if ( + !Array.isArray(record.argv) || + record.argv.length === 0 || + record.argv.some((value) => typeof value !== "string" || value.length === 0) + ) { + return null; + } + const argv = record.argv.map((value) => String(value)); + const env = + record.env && typeof record.env === "object" && !Array.isArray(record.env) + ? Object.fromEntries( + Object.entries(record.env as Record).filter( + (entry): entry is [string, string] => typeof entry[1] === "string", + ), + ) + : undefined; + const rawNoOutputTimeoutSeconds = + typeof record.noOutputTimeoutSeconds === "number" || + typeof record.noOutputTimeoutSeconds === "bigint" + ? record.noOutputTimeoutSeconds + : null; + const rawOutputMaxBytes = + typeof record.outputMaxBytes === "number" || typeof record.outputMaxBytes === "bigint" + ? record.outputMaxBytes + : null; + const noOutputTimeoutSeconds = normalizeNumber(rawNoOutputTimeoutSeconds); + const outputMaxBytes = normalizeNumber(rawOutputMaxBytes); + return { + argv, + ...(typeof record.cwd === "string" && record.cwd.trim() ? { cwd: record.cwd } : {}), + ...(env && Object.keys(env).length > 0 ? { env } : {}), + ...(typeof record.input === "string" ? { input: record.input } : {}), + ...(noOutputTimeoutSeconds != null ? { noOutputTimeoutSeconds } : {}), + ...(outputMaxBytes != null && outputMaxBytes > 0 ? { outputMaxBytes } : {}), + }; +} + /** Maps cron payload variants into normalized SQLite columns. */ export function bindPayloadColumns( payload: CronPayload, @@ -45,6 +90,21 @@ export function bindPayloadColumns( payload_tools_allow_json: null, }; } + if (payload.kind === "command") { + const { timeoutSeconds: _timeoutSeconds, ...payloadMessage } = payload; + return { + payload_kind: "command", + payload_message: serializeJson(payloadMessage), + payload_model: null, + payload_fallbacks_json: null, + payload_thinking: null, + payload_timeout_seconds: payload.timeoutSeconds ?? null, + payload_allow_unsafe_external_content: null, + payload_external_content_source_json: null, + payload_light_context: null, + payload_tools_allow_json: null, + }; + } return { payload_kind: "agentTurn", payload_message: payload.message, @@ -97,5 +157,17 @@ export function payloadFromRow(row: CronJobRow): CronPayload | null { ...(toolsAllow ? { toolsAllow } : {}), }; } + if (row.payload_kind === "command") { + const command = parseCommandPayloadMessage(row.payload_message); + if (!command) { + return null; + } + const timeoutSeconds = normalizeNumber(row.payload_timeout_seconds); + return { + kind: "command", + ...command, + ...(timeoutSeconds != null ? { timeoutSeconds } : {}), + }; + } return null; } diff --git a/src/cron/types.ts b/src/cron/types.ts index 069e3b279e9d..541bdefdc73b 100644 --- a/src/cron/types.ts +++ b/src/cron/types.ts @@ -221,11 +221,17 @@ export type CronFailureAlert = { accountId?: string; }; -/** Payload variants cron can execute in main-session or isolated-agent modes. */ -export type CronPayload = { kind: "systemEvent"; text: string } | CronAgentTurnPayload; +/** Payload variants cron can execute in main-session or detached modes. */ +export type CronPayload = + | { kind: "systemEvent"; text: string } + | CronAgentTurnPayload + | CronCommandPayload; /** Partial payload update shape used by cron patch/edit flows. */ -export type CronPayloadPatch = { kind: "systemEvent"; text?: string } | CronAgentTurnPayloadPatch; +export type CronPayloadPatch = + | { kind: "systemEvent"; text?: string } + | CronAgentTurnPayloadPatch + | CronCommandPayloadPatch; type CronAgentTurnPayloadFields = { message: string; @@ -253,6 +259,25 @@ type CronAgentTurnPayloadPatch = { } & Partial> & { toolsAllow?: string[] | null; }; + +type CronCommandPayloadFields = { + /** Explicit argv vector to execute. Use a shell wrapper argv for shell syntax. */ + argv: string[]; + cwd?: string; + env?: Record; + input?: string; + timeoutSeconds?: number; + noOutputTimeoutSeconds?: number; + outputMaxBytes?: number; +}; + +type CronCommandPayload = { + kind: "command"; +} & CronCommandPayloadFields; + +type CronCommandPayloadPatch = { + kind: "command"; +} & Partial; /** Mutable runtime state persisted beside the immutable cron job spec. */ export type CronJobState = { nextRunAtMs?: number; diff --git a/src/gateway/server-cron.test.ts b/src/gateway/server-cron.test.ts index 1de83aa24965..ed4d217f0244 100644 --- a/src/gateway/server-cron.test.ts +++ b/src/gateway/server-cron.test.ts @@ -404,6 +404,79 @@ describe("buildGatewayCronService", () => { } }); + it("suppresses command cron NO_REPLY output before announce delivery", async () => { + const cfg = createCronConfig("server-cron-command-no-reply"); + loadConfigMock.mockReturnValue(cfg); + + const state = buildGatewayCronService({ + cfg, + deps: {} as CliDeps, + broadcast: () => {}, + }); + try { + const job = await state.cron.add({ + name: "silent-command", + enabled: true, + deleteAfterRun: false, + schedule: { kind: "at", at: new Date(1).toISOString() }, + sessionTarget: "isolated", + wakeMode: "next-heartbeat", + payload: { + kind: "command", + argv: [process.execPath, "-e", "process.stdout.write('NO_REPLY\\n')"], + }, + delivery: { + mode: "announce", + channel: "telegram", + to: "123", + }, + }); + + await state.cron.run(job.id, "force"); + + expect(state.cron.getJob(job.id)?.state.lastRunStatus).toBe("ok"); + expect(state.cron.getJob(job.id)?.state.lastDeliveryError).toBeUndefined(); + } finally { + state.cron.stop(); + } + }); + + it("suppresses command cron NO_REPLY output before webhook delivery", async () => { + const cfg = createCronConfig("server-cron-command-webhook-no-reply"); + loadConfigMock.mockReturnValue(cfg); + + const state = buildGatewayCronService({ + cfg, + deps: {} as CliDeps, + broadcast: () => {}, + }); + try { + const job = await state.cron.add({ + name: "silent-command-webhook", + enabled: true, + deleteAfterRun: false, + schedule: { kind: "at", at: new Date(1).toISOString() }, + sessionTarget: "isolated", + wakeMode: "next-heartbeat", + payload: { + kind: "command", + argv: [process.execPath, "-e", "process.stdout.write('NO_REPLY\\n')"], + }, + delivery: { + mode: "webhook", + to: "https://example.invalid/cron-finished", + }, + }); + + await state.cron.run(job.id, "force"); + + expect(state.cron.getJob(job.id)?.state.lastRunStatus).toBe("ok"); + expect(fetchWithSsrFGuardMock).not.toHaveBeenCalled(); + } finally { + state.cron.stop(); + } + }); + it("routes global-scope main cron jobs through the global queue for queued wakes", async () => { const cfg = { ...createCronConfig("server-cron-global-queued"), diff --git a/src/gateway/server-cron.ts b/src/gateway/server-cron.ts index a4fbac7d3aeb..391e29d8c3a0 100644 --- a/src/gateway/server-cron.ts +++ b/src/gateway/server-cron.ts @@ -3,6 +3,7 @@ import { retireSessionMcpRuntime } from "../agents/agent-bundle-mcp-tools.js"; import { resolveDefaultAgentId } from "../agents/agent-scope.js"; import { abortAndDrainEmbeddedAgentRun } from "../agents/embedded-agent.js"; +import { isSilentReplyText, SILENT_REPLY_TOKEN } from "../auto-reply/tokens.js"; import { cleanupBrowserSessionsForLifecycleEnd } from "../browser-lifecycle-cleanup.js"; import type { CliDeps } from "../cli/deps.types.js"; import { getRuntimeConfig } from "../config/io.js"; @@ -14,11 +15,16 @@ 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 { runCronCommandJob } from "../cron/command-runner.js"; +import { resolveCronDeliveryPlan, sendCronAnnouncePayloadStrict } from "../cron/delivery.js"; import { runCronIsolatedAgentTurn } from "../cron/isolated-agent.js"; import { appendCronRunLog, resolveCronRunLogPruneOptions } from "../cron/run-log.js"; import type { CronServiceContract } from "../cron/service-contract.js"; import { CronService } from "../cron/service.js"; -import { resolveCronSessionTargetSessionKey } from "../cron/session-target.js"; +import { + resolveCronDeliverySessionKey, + resolveCronSessionTargetSessionKey, +} from "../cron/session-target.js"; import { resolveCronJobsStorePath } from "../cron/store.js"; import type { CronJob } from "../cron/types.js"; import { formatErrorMessage } from "../infra/errors.js"; @@ -382,6 +388,105 @@ export function buildGatewayCronService(params: { }); } }, + runCommandJob: async ({ job, abortSignal }) => { + const result = await runCronCommandJob({ + job, + abortSignal, + nowMs: Date.now, + }); + const plan = resolveCronDeliveryPlan(job); + const deliveryTrace = { + intended: pickDefined( + { + channel: plan.channel, + to: plan.to, + accountId: plan.accountId, + threadId: plan.threadId, + source: "explicit" as const, + }, + ["channel", "to", "accountId", "threadId", "source"], + ), + }; + const summaryIsSilent = + typeof result.summary === "string" && isSilentReplyText(result.summary, SILENT_REPLY_TOKEN); + if (summaryIsSilent) { + const { summary: _summary, ...silentResult } = result; + return { + ...silentResult, + deliveryAttempted: false, + delivered: false, + delivery: deliveryTrace, + }; + } + const shouldAnnounce = + plan.mode === "announce" && typeof result.summary === "string" && result.summary.trim(); + if (!shouldAnnounce) { + return { + ...result, + deliveryAttempted: false, + delivered: false, + delivery: deliveryTrace, + }; + } + const message = result.summary; + if (typeof message !== "string") { + return { + ...result, + deliveryAttempted: false, + delivered: false, + delivery: deliveryTrace, + }; + } + const { agentId, cfg: runtimeConfig } = resolveCronAgent(job.agentId); + try { + await sendCronAnnouncePayloadStrict({ + deps: params.deps, + cfg: runtimeConfig, + agentId, + jobId: job.id, + target: { + channel: plan.channel, + to: plan.to, + accountId: plan.accountId, + sessionKey: resolveCronDeliverySessionKey(job), + }, + message, + abortSignal: abortSignal ?? new AbortController().signal, + }); + return { + ...result, + deliveryAttempted: true, + delivered: true, + delivery: { + ...deliveryTrace, + delivered: true, + }, + }; + } catch (err) { + const error = formatErrorMessage(err); + cronLogger.warn({ jobId: job.id, err: error }, "cron: command delivery failed"); + return { + ...result, + status: job.delivery?.bestEffort ? result.status : "error", + error: job.delivery?.bestEffort ? result.error : error, + deliveryAttempted: true, + delivered: false, + delivery: { + ...deliveryTrace, + delivered: false, + resolved: { + channel: plan.channel, + to: plan.to, + accountId: plan.accountId, + threadId: plan.threadId, + source: "explicit" as const, + ok: false, + error, + }, + }, + }; + } + }, cleanupTimedOutAgentRun: async ({ job, execution }) => { if (!execution?.sessionId) { return; diff --git a/src/process/exec.ts b/src/process/exec.ts index 083fce10e0e0..b34ab00815ed 100644 --- a/src/process/exec.ts +++ b/src/process/exec.ts @@ -13,6 +13,7 @@ import { } from "../infra/windows-encoding.js"; import { getWindowsInstallRoots } from "../infra/windows-install-roots.js"; import { logDebug, logError } from "../logger.js"; +import { killProcessTree as terminateProcessTree } from "./kill-tree.js"; import { resolveCommandStdio } from "./spawn-utils.js"; import { resolveWindowsCommandShim } from "./windows-command.js"; @@ -256,10 +257,12 @@ export type CommandOptions = { noOutputTimeoutMs?: number; signal?: AbortSignal; maxOutputBytes?: number; + killProcessTree?: boolean; }; const WINDOWS_CLOSE_STATE_SETTLE_TIMEOUT_MS = 250; const WINDOWS_CLOSE_STATE_POLL_MS = 10; +const COMMAND_PROCESS_TREE_KILL_GRACE_MS = 300; const DEFAULT_COMMAND_OUTPUT_MAX_BYTES = 16 * 1024 * 1024; type CapturedOutputBuffers = { @@ -368,7 +371,8 @@ export async function runCommandWithTimeout( ): Promise { const options: CommandOptions = typeof optionsOrTimeout === "number" ? { timeoutMs: optionsOrTimeout } : optionsOrTimeout; - const { timeoutMs, cwd, input, baseEnv, env, noOutputTimeoutMs, signal } = options; + const { timeoutMs, cwd, input, baseEnv, env, noOutputTimeoutMs, signal, killProcessTree } = + options; const hasInput = input !== undefined; const resolvedEnv = resolveCommandEnv({ argv, baseEnv, env }); const stdio = resolveCommandStdio({ hasInput, preferInherit: true }); @@ -393,6 +397,9 @@ export async function runCommandWithTimeout( stdio, cwd, env: resolvedEnv, + // Cron shell wrappers need their own process group so timeout/abort kills + // reach foreground children instead of leaving duplicate scheduled work. + ...(killProcessTree && process.platform !== "win32" ? { detached: true } : {}), windowsHide: invocation.windowsHide, windowsVerbatimArguments: invocation.windowsVerbatimArguments, ...(shouldSpawnWithShell({ resolvedCommand: invocation.command, platform: process.platform }) @@ -412,6 +419,7 @@ export async function runCommandWithTimeout( let killIssuedByAbort = false; let childExitState: { code: number | null; signal: NodeJS.Signals | null } | null = null; let closeFallbackTimer: NodeJS.Timeout | null = null; + let processTreeForceKillTimer: NodeJS.Timeout | null = null; let noOutputTimer: NodeJS.Timeout | null = null; const shouldTrackOutputTimeout = typeof noOutputTimeoutMs === "number" && @@ -435,6 +443,14 @@ export async function runCommandWithTimeout( closeFallbackTimer = null; }; + const clearProcessTreeForceKillTimer = () => { + if (!processTreeForceKillTimer) { + return; + } + clearTimeout(processTreeForceKillTimer); + processTreeForceKillTimer = null; + }; + const killChild = (byTimeout = true) => { if (settled || typeof child?.kill !== "function") { return; @@ -444,6 +460,47 @@ export async function runCommandWithTimeout( } else { killIssuedByAbort = true; } + if ( + killProcessTree && + typeof child.pid === "number" && + child.pid > 0 + ) { + if (process.platform === "win32") { + try { + spawn("taskkill", ["/PID", String(child.pid), "/T"], { + stdio: "ignore", + windowsHide: true, + }); + if (!processTreeForceKillTimer) { + processTreeForceKillTimer = setTimeout(() => { + processTreeForceKillTimer = null; + if ( + settled || + childExitState != null || + child.exitCode != null || + child.signalCode != null + ) { + return; + } + try { + spawn("taskkill", ["/PID", String(child.pid), "/T", "/F"], { + stdio: "ignore", + windowsHide: true, + }); + } catch { + child.kill("SIGKILL"); + } + }, COMMAND_PROCESS_TREE_KILL_GRACE_MS); + processTreeForceKillTimer.unref(); + } + return; + } catch { + // Fall through to Node's direct child kill as a last resort. + } + } + terminateProcessTree(child.pid, { graceMs: COMMAND_PROCESS_TREE_KILL_GRACE_MS }); + return; + } if (process.platform === "win32" && typeof child.pid === "number" && child.pid > 0) { try { spawn("taskkill", ["/PID", String(child.pid), "/T", "/F"], { @@ -507,12 +564,14 @@ export async function runCommandWithTimeout( clearTimeout(timer); clearNoOutputTimer(); clearCloseFallbackTimer(); + clearProcessTreeForceKillTimer(); removeAbortListener?.(); removeAbortListener = null; reject(err); }); child.on("exit", (code, signalResult) => { childExitState = { code, signal: signalResult }; + clearProcessTreeForceKillTimer(); if (settled || closeFallbackTimer) { return; } @@ -532,6 +591,7 @@ export async function runCommandWithTimeout( clearTimeout(timer); clearNoOutputTimer(); clearCloseFallbackTimer(); + clearProcessTreeForceKillTimer(); removeAbortListener?.(); removeAbortListener = null; const resolvedSignal = childExitState?.signal ?? signalValue ?? child.signalCode ?? null; diff --git a/ui/src/ui/app-defaults.ts b/ui/src/ui/app-defaults.ts index 9c7f33fb3d6f..29fa6a0f6a01 100644 --- a/ui/src/ui/app-defaults.ts +++ b/ui/src/ui/app-defaults.ts @@ -36,6 +36,7 @@ export const DEFAULT_CRON_FORM: CronFormState = { sessionTarget: "isolated", wakeMode: "now", payloadKind: "agentTurn", + payloadLocked: false, payloadText: "", payloadModel: "", payloadThinking: "", diff --git a/ui/src/ui/controllers/cron.test.ts b/ui/src/ui/controllers/cron.test.ts index 6219fec4c54f..37017e7eca83 100644 --- a/ui/src/ui/controllers/cron.test.ts +++ b/ui/src/ui/controllers/cron.test.ts @@ -600,6 +600,47 @@ describe("cron controller", () => { expect(state.cronForm.deliveryAccountId).toBe("bot-2"); }); + it("preserves command payloads when editing Control UI metadata", async () => { + const request = vi.fn(async (method: string, _payload?: unknown) => { + if (method === "cron.update") { + return { id: "job-command" }; + } + if (method === "cron.list") { + return { jobs: [{ id: "job-command" }] }; + } + if (method === "cron.status") { + return { enabled: true, jobs: 1, nextWakeAtMs: null }; + } + return {}; + }); + const job = { + id: "job-command", + name: "Command", + enabled: true, + createdAtMs: 0, + updatedAtMs: 0, + schedule: { kind: "every" as const, everyMs: 600_000 }, + sessionTarget: "isolated" as const, + wakeMode: "next-heartbeat" as const, + payload: { kind: "command" as const, argv: ["sh", "-lc", "echo ok"] }, + delivery: { mode: "announce" as const, channel: "telegram", to: "123" }, + state: {}, + }; + const state = createState({ + client: { request } as unknown as CronState["client"], + cronJobs: [job], + }); + + startCronEdit(state, job); + state.cronForm.name = "Command renamed"; + await addCronJob(state); + + const updateCall = findRequestCall(request.mock.calls, "cron.update"); + const patch = requestPatch(updateCall); + expect(patch.name).toBe("Command renamed"); + expect(patch).not.toHaveProperty("payload"); + }); + 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") { diff --git a/ui/src/ui/controllers/cron.ts b/ui/src/ui/controllers/cron.ts index 95a83968cdd4..56da1d4970f5 100644 --- a/ui/src/ui/controllers/cron.ts +++ b/ui/src/ui/controllers/cron.ts @@ -97,8 +97,10 @@ export type CronModelSuggestionsState = { cronModelSuggestions: string[]; }; -function supportsAnnounceDelivery(form: Pick) { - return form.sessionTarget !== "main" && form.payloadKind === "agentTurn"; +function supportsAnnounceDelivery( + form: Pick, +) { + return form.sessionTarget !== "main" && (form.payloadKind === "agentTurn" || form.payloadLocked); } export function normalizeCronFormState(form: CronFormState): CronFormState { @@ -143,13 +145,13 @@ export function validateCronForm(form: CronFormState): CronFieldErrors { } } } - if (!form.payloadText.trim()) { + if (!form.payloadLocked && !form.payloadText.trim()) { errors.payloadText = form.payloadKind === "systemEvent" ? "cron.errors.systemTextRequired" : "cron.errors.agentMessageRequired"; } - if (form.payloadKind === "agentTurn") { + if (!form.payloadLocked && form.payloadKind === "agentTurn") { const timeoutRaw = form.timeoutSeconds.trim(); if (timeoutRaw) { const timeout = toNumber(timeoutRaw, 0); @@ -485,6 +487,7 @@ function parseStaggerSchedule( function jobToForm(job: CronJob, prev: CronFormState): CronFormState { const failureAlert = job.failureAlert; const payload = getCronJobPayload(job); + const payloadLocked = payload?.kind === "command"; const next: CronFormState = { ...prev, name: job.name, @@ -505,8 +508,19 @@ function jobToForm(job: CronJob, prev: CronFormState): CronFormState { staggerUnit: "seconds", sessionTarget: job.sessionTarget, wakeMode: job.wakeMode, - payloadKind: payload?.kind ?? DEFAULT_CRON_FORM.payloadKind, - payloadText: payload?.kind === "systemEvent" ? payload.text : (payload?.message ?? ""), + payloadKind: + payload?.kind === "systemEvent" || payload?.kind === "agentTurn" + ? payload.kind + : DEFAULT_CRON_FORM.payloadKind, + payloadLocked, + payloadText: + payload?.kind === "systemEvent" + ? payload.text + : payload?.kind === "agentTurn" + ? payload.message + : payload?.kind === "command" + ? payload.argv.join(" ") + : "", payloadModel: payload?.kind === "agentTurn" ? (payload.model ?? "") : "", payloadThinking: payload?.kind === "agentTurn" ? (payload.thinking ?? "") : "", payloadLightContext: payload?.kind === "agentTurn" ? payload.lightContext === true : false, @@ -699,12 +713,15 @@ export async function addCronJob(state: CronState): Promise { } const schedule = buildCronSchedule(form); - const payload = buildCronPayload(form); const editingJob = state.cronEditingJobId ? state.cronJobs.find((job) => job.id === state.cronEditingJobId) : undefined; const editingPayload = editingJob ? getCronJobPayload(editingJob) : null; - if (payload.kind === "agentTurn") { + const preserveLockedPayload = Boolean( + state.cronEditingJobId && form.payloadLocked && editingPayload?.kind === "command", + ); + const payload = preserveLockedPayload ? undefined : buildCronPayload(form); + if (payload?.kind === "agentTurn") { const existingLightContext = editingPayload?.kind === "agentTurn" ? editingPayload.lightContext : undefined; if ( @@ -743,7 +760,7 @@ export async function addCronJob(state: CronState): Promise { const agentId = form.clearAgent ? null : form.agentId.trim(); const sessionKeyRaw = form.sessionKey.trim(); const sessionKey = sessionKeyRaw || (editingJob?.sessionKey ? null : undefined); - const job = { + const job: Record = { name: form.name.trim(), description: form.description.trim(), agentId: agentId === null ? null : agentId || undefined, @@ -753,10 +770,12 @@ export async function addCronJob(state: CronState): Promise { schedule, sessionTarget: form.sessionTarget, wakeMode: form.wakeMode, - payload, delivery, failureAlert, }; + if (payload) { + job.payload = payload; + } if (!job.name) { throw new Error(t("cron.errors.nameRequiredShort")); } @@ -944,6 +963,11 @@ export function startCronClone(state: CronState, job: CronJob) { ); const cloned = jobToForm(job, state.cronForm); cloned.name = buildCloneName(job.name, existingNames); + if (cloned.payloadLocked) { + cloned.payloadLocked = false; + cloned.payloadKind = DEFAULT_CRON_FORM.payloadKind; + cloned.payloadText = ""; + } state.cronForm = cloned; state.cronFieldErrors = validateCronForm(state.cronForm); } diff --git a/ui/src/ui/cron-payload.ts b/ui/src/ui/cron-payload.ts index 18dc8d0ec884..f61a9f9508e6 100644 --- a/ui/src/ui/cron-payload.ts +++ b/ui/src/ui/cron-payload.ts @@ -15,6 +15,9 @@ export function isCronPayload(value: unknown): value is CronPayload { if (value.kind === "agentTurn") { return typeof value.message === "string"; } + if (value.kind === "command") { + return Array.isArray(value.argv) && value.argv.every((arg) => typeof arg === "string"); + } return false; } diff --git a/ui/src/ui/presenter.ts b/ui/src/ui/presenter.ts index 314dc6e87283..019b3b287a7d 100644 --- a/ui/src/ui/presenter.ts +++ b/ui/src/ui/presenter.ts @@ -79,6 +79,9 @@ export function formatCronPayload(job: CronJob) { if (p.kind === "systemEvent") { return `System: ${p.text}`; } + if (p.kind === "command") { + return `Command: ${p.argv.join(" ")}`; + } const base = `Agent: ${p.message}`; const delivery = job.delivery; if (delivery && delivery.mode !== "none") { diff --git a/ui/src/ui/types.ts b/ui/src/ui/types.ts index 2333f789677f..9a80c12068a8 100644 --- a/ui/src/ui/types.ts +++ b/ui/src/ui/types.ts @@ -548,6 +548,16 @@ export type CronWakeMode = "next-heartbeat" | "now"; export type CronPayload = | { kind: "systemEvent"; text: string } + | { + kind: "command"; + argv: string[]; + cwd?: string; + env?: Record; + input?: string; + timeoutSeconds?: number; + noOutputTimeoutSeconds?: number; + outputMaxBytes?: number; + } | { kind: "agentTurn"; message: string; diff --git a/ui/src/ui/ui-types.ts b/ui/src/ui/ui-types.ts index 186602cbc1c3..180bbb57b82e 100644 --- a/ui/src/ui/ui-types.ts +++ b/ui/src/ui/ui-types.ts @@ -61,6 +61,7 @@ export type CronFormState = { sessionTarget: "main" | "isolated" | "current" | `session:${string}`; wakeMode: "next-heartbeat" | "now"; payloadKind: "systemEvent" | "agentTurn"; + payloadLocked: boolean; payloadText: string; payloadModel: string; payloadThinking: string; diff --git a/ui/src/ui/views/cron.ts b/ui/src/ui/views/cron.ts index e8b476e31f67..12b85a521277 100644 --- a/ui/src/ui/views/cron.ts +++ b/ui/src/ui/views/cron.ts @@ -362,7 +362,8 @@ function renderFieldLabel(text: string, required = false) { export function renderCron(props: CronProps) { const isEditing = Boolean(props.editingJobId); - const isAgentTurn = props.form.payloadKind === "agentTurn"; + const payloadLocked = props.form.payloadLocked; + const isAgentTurn = !payloadLocked && props.form.payloadKind === "agentTurn"; const isCronSchedule = props.form.scheduleKind === "cron"; const channelOptions = buildChannelOptions(props); const selectedJob = @@ -385,7 +386,8 @@ export function renderCron(props: CronProps) { const statusSummary = summarizeSelection(selectedStatusLabels, t("cron.runs.allStatuses")); const deliverySummary = summarizeSelection(selectedDeliveryLabels, t("cron.runs.allDelivery")); const supportsAnnounce = - props.form.sessionTarget !== "main" && props.form.payloadKind === "agentTurn"; + props.form.sessionTarget !== "main" && + (props.form.payloadKind === "agentTurn" || payloadLocked); const selectedDeliveryMode = props.form.deliveryMode === "announce" && !supportsAnnounce ? "none" : props.form.deliveryMode; const formOpen = props.cronFormCollapsed === false || isEditing; @@ -919,26 +921,35 @@ export function renderCron(props: CronProps) {
${t("cron.form.wakeModeHelp")}
- + ${payloadLocked + ? html` + + ` + : html` + + `} ${isAgentTurn ? html`