feat(cron): deterministic command payloads (run a shell command, no LLM) (#1279)

* feat(cron): deterministic command payloads (run a shell command, no LLM)

Cron jobs always run an agent turn today, so deterministic work (health
probes, backups, syncs) pays model tokens on every fire. This adds a
"command" payload kind that runs a shell command directly in the gateway
process with zero model tokens, mirroring openclaw's command cron.

- store: CronPayload.Command (*CronCommandSpec — argv/cwd/env/input/
  timeouts/output cap). Persists in the existing payload JSON blob, so
  there is NO migration and no schema version bump.
- internal/cronexec: in-process runner with wall-clock + no-output
  timeouts, per-stream output capping, and process-group termination so a
  timed-out command's forked children are also killed.
- gateway_cron handler: command jobs run in-process and deliver stdout on
  success (honoring the NO_REPLY sentinel). A non-zero exit / timeout
  returns an error so the run is recorded as error and retried per
  cron.max_retries; failures are NOT delivered, mirroring the agent path
  (only successful output is announced — no channel spam).
- surfaces: cron.create RPC, the agent `cron` tool, and a new
  `goclaw cron create` CLI all accept command payloads.
- security: gated by cron.command_enabled (default false). Commands run
  with the gateway process's privileges, so the feature is opt-in per
  gateway; when disabled the RPC and tool reject command payloads and the
  handler refuses to run them.
- i18n (en/vi/zh), docs (08-scheduling-cron.md), and tests for the runner
  and the handler command path.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(cron): gate command payloads on the update surfaces too

handleUpdate (RPC + agent tool) passed CronJobPatch.Command straight to
UpdateJob, which switches the payload to command kind for any non-nil
Command — without the command_enabled gate or ValidateCronCommandSpec that
create enforces. A normal job could therefore be mutated into a command job
(or persisted with an invalid spec, e.g. empty argv) on a gateway where
command cron is disabled, breaking the disabled-gateway contract.

Both update surfaces now require cron.command_enabled and validate the spec
before UpdateJob, matching create. The agent tool parses the command via the
same path as add and drops the raw keys so a shell-string command can't break
the generic patch unmarshal. Regression tests added for RPC and tool update
(command disabled + invalid argv), plus a positive enabled-valid case.

Addresses review feedback from @mrgoonie on #1279.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Zezae OhandClaude Opus 4.8 authored and GitHub committed 2026-06-25 10:23:55 +07:00
1 parent 0240a36b05
commit 4a79c8a208
22 files changed
+1215 -71

No files matched your search

+152
View File
@@ -4,6 +4,7 @@ import (
"encoding/json"
"fmt"
"os"
"strings"
"text/tabwriter"
"time"
@@ -19,11 +20,130 @@ func cronCmd() *cobra.Command {
Short: "Manage scheduled cron jobs",
}
cmd.AddCommand(cronListCmd())
cmd.AddCommand(cronCreateCmd())
cmd.AddCommand(cronDeleteCmd())
cmd.AddCommand(cronToggleCmd())
return cmd
}
func cronCreateCmd() *cobra.Command {
var (
name string
cronExpr string
every string
at string
tz string
command string
argvJSON string
cwd string
timeout string
envPairs []string
deliver bool
channel string
to string
)
cmd := &cobra.Command{
Use: "create",
Short: "Create a deterministic command cron job (runs a shell command, no LLM)",
Long: "Create a cron job whose payload is a shell command executed in the gateway\n" +
"process WITHOUT an LLM turn (zero model tokens). Requires the gateway to have\n" +
"cron.command_enabled=true.\n\n" +
"Examples:\n" +
" goclaw cron create --name disk-probe --cron '*/15 * * * *' --command 'df -h /'\n" +
" goclaw cron create --name backup --at 2026-07-01T09:00:00Z --argv '[\"/opt/backup.sh\"]' --timeout 5m",
Run: func(cmd *cobra.Command, args []string) {
if name == "" {
fmt.Fprintln(os.Stderr, "Error: --name is required")
os.Exit(1)
}
schedule := map[string]any{}
switch {
case cronExpr != "":
schedule["kind"] = "cron"
schedule["expr"] = cronExpr
if tz != "" {
schedule["tz"] = tz
}
case every != "":
d, err := time.ParseDuration(every)
if err != nil || d <= 0 {
fmt.Fprintf(os.Stderr, "Error: invalid --every duration %q\n", every)
os.Exit(1)
}
schedule["kind"] = "every"
schedule["everyMs"] = d.Milliseconds()
case at != "":
ts, err := time.Parse(time.RFC3339, at)
if err != nil {
fmt.Fprintf(os.Stderr, "Error: invalid --at time %q (use RFC3339, e.g. 2026-07-01T09:00:00Z)\n", at)
os.Exit(1)
}
schedule["kind"] = "at"
schedule["atMs"] = ts.UnixMilli()
default:
fmt.Fprintln(os.Stderr, "Error: one of --cron, --every, or --at is required")
os.Exit(1)
}
var argv []string
switch {
case argvJSON != "":
if err := json.Unmarshal([]byte(argvJSON), &argv); err != nil {
fmt.Fprintf(os.Stderr, "Error: --argv must be a JSON array of strings: %v\n", err)
os.Exit(1)
}
case command != "":
argv = []string{"sh", "-c", command}
default:
fmt.Fprintln(os.Stderr, "Error: one of --command or --argv is required")
os.Exit(1)
}
commandSpec := map[string]any{"argv": argv}
if cwd != "" {
commandSpec["cwd"] = cwd
}
if timeout != "" {
d, err := time.ParseDuration(timeout)
if err != nil || d <= 0 {
fmt.Fprintf(os.Stderr, "Error: invalid --timeout duration %q\n", timeout)
os.Exit(1)
}
commandSpec["timeoutSeconds"] = int(d.Seconds())
}
if len(envPairs) > 0 {
env := map[string]string{}
for _, kv := range envPairs {
k, v, ok := strings.Cut(kv, "=")
if !ok {
fmt.Fprintf(os.Stderr, "Error: --env must be KEY=VALUE, got %q\n", kv)
os.Exit(1)
}
env[k] = v
}
commandSpec["env"] = env
}
cronCreateCommandRPC(name, schedule, commandSpec, deliver, channel, to)
},
}
cmd.Flags().StringVar(&name, "name", "", "job name (lowercase slug, required)")
cmd.Flags().StringVar(&cronExpr, "cron", "", "cron expression (5-field), e.g. '*/15 * * * *'")
cmd.Flags().StringVar(&every, "every", "", "fixed interval as a Go duration, e.g. 15m")
cmd.Flags().StringVar(&at, "at", "", "one-shot time (RFC3339), e.g. 2026-07-01T09:00:00Z")
cmd.Flags().StringVar(&tz, "tz", "", "IANA timezone for --cron (e.g. Asia/Seoul)")
cmd.Flags().StringVar(&command, "command", "", "shell command (run as sh -c)")
cmd.Flags().StringVar(&argvJSON, "argv", "", `explicit argv as a JSON array, e.g. '["node","x.js"]'`)
cmd.Flags().StringVar(&cwd, "cwd", "", "working directory")
cmd.Flags().StringVar(&timeout, "timeout", "", "per-command timeout as a Go duration, e.g. 30s")
cmd.Flags().StringArrayVar(&envPairs, "env", nil, "environment override KEY=VALUE (repeatable)")
cmd.Flags().BoolVar(&deliver, "deliver", false, "deliver command output to a channel")
cmd.Flags().StringVar(&channel, "channel", "", "delivery channel")
cmd.Flags().StringVar(&to, "to", "", "delivery chat/target ID")
return cmd
}
func cronListCmd() *cobra.Command {
var jsonOutput bool
var showDisabled bool
@@ -90,6 +210,38 @@ func cronListRPC(showDisabled, jsonOutput bool) {
printCronJobs(result.Jobs, jsonOutput)
}
func cronCreateCommandRPC(name string, schedule, command map[string]any, deliver bool, channel, to string) {
requireGateway()
params, _ := json.Marshal(map[string]any{
"name": name,
"schedule": schedule,
"command": command,
"deliver": deliver,
"deliverChannel": channel,
"deliverTo": to,
})
resp, err := gatewayRPC(protocol.MethodCronCreate, params)
if err != nil {
fmt.Fprintf(os.Stderr, "Error: %v\n", err)
os.Exit(1)
}
if !resp.OK {
fmt.Fprintf(os.Stderr, "Failed: %s\n", resp.Error.Message)
os.Exit(1)
}
raw, _ := json.Marshal(resp.Payload)
var result struct {
Job store.CronJob `json:"job"`
}
if err := json.Unmarshal(raw, &result); err == nil && result.Job.ID != "" {
fmt.Printf("Created command cron job %s (%s)\n", result.Job.ID, result.Job.Name)
return
}
fmt.Println("Created command cron job.")
}
func cronDeleteRPC(jobID string) {
requireGateway()
+1 -1
View File
@@ -346,7 +346,7 @@ func runGateway() {
_ = skillSearchTool // used via wireExtras → skillsLoader; kept for type clarity
// Register cron/heartbeat/session/message tools, aliases, allow-paths, store wiring.
heartbeatTool, hasMemory := wireExtraTools(pgStores, toolsReg, msgBus, workspace, dataDir, agentCfg, globalSkillsDir, builtinSkillsDir)
heartbeatTool, hasMemory := wireExtraTools(pgStores, toolsReg, msgBus, workspace, dataDir, agentCfg, globalSkillsDir, builtinSkillsDir, cfg.Cron.CommandEnabled)
// Register workstation_exec + claude_remote tools (Standard edition only; deny-all until Phase 6).
// cleanupWorkstation stops the activity sink retention goroutine and drains the write buffer.
+81 -25
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"log/slog"
"strings"
"time"
"github.com/google/uuid"
@@ -12,6 +13,7 @@ import (
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/channels"
"github.com/nextlevelbuilder/goclaw/internal/config"
"github.com/nextlevelbuilder/goclaw/internal/cronexec"
"github.com/nextlevelbuilder/goclaw/internal/providers"
"github.com/nextlevelbuilder/goclaw/internal/scheduler"
"github.com/nextlevelbuilder/goclaw/internal/sessions"
@@ -67,6 +69,12 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg
// Resolve channel type for system prompt context.
channelType := resolveChannelType(channelMgr, channel)
// Deterministic command payload: run the shell command in-process WITHOUT
// an LLM/agent turn (zero model tokens). Gated by cron.command_enabled.
if job.Payload.IsCommand() {
return runCommandCronJob(cfg, job, msgBus, peerKind)
}
// Build cron context so the agent knows delivery target and requester.
var extraPrompt string
if job.Deliver && job.DeliverChannel != "" && job.DeliverTo != "" {
@@ -165,31 +173,7 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg
result := outcome.Result
// If job wants delivery to a channel, send the agent response to the target chat.
if job.Deliver && job.DeliverChannel != "" && job.DeliverTo != "" {
if cronOutputContainsNoReplySentinel(result.Content) {
slog.Info("cron: suppressed delivery because output contained NO_REPLY",
"job_id", job.ID,
"job_name", job.Name,
"channel", job.DeliverChannel,
"to", job.DeliverTo,
"content_len", len(result.Content),
)
} else {
outMsg := bus.OutboundMessage{
Channel: job.DeliverChannel,
ChatID: job.DeliverTo,
Content: result.Content,
}
if peerKind == "group" {
outMsg.Metadata = map[string]string{"group_id": job.DeliverTo}
}
appendMediaToOutbound(&outMsg, result.Media)
msgBus.PublishOutbound(outMsg)
}
} else if job.Deliver {
slog.Warn("cron: delivery configured but channel/chatID missing — output discarded",
"job_id", job.ID, "job_name", job.Name, "channel", job.DeliverChannel, "to", job.DeliverTo)
}
deliverCronOutput(msgBus, job, result.Content, result.Media, peerKind)
cronResult := &store.CronJobResult{
Content: result.Content,
@@ -209,6 +193,78 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg
}
}
// deliverCronOutput publishes a cron job's output to the configured delivery
// channel, honoring the NO_REPLY sentinel. Shared by the agent-turn and the
// deterministic command-payload paths.
func deliverCronOutput(msgBus *bus.MessageBus, job *store.CronJob, content string, media []agent.MediaResult, peerKind string) {
if job.Deliver && job.DeliverChannel != "" && job.DeliverTo != "" {
if cronOutputContainsNoReplySentinel(content) {
slog.Info("cron: suppressed delivery because output contained NO_REPLY",
"job_id", job.ID,
"job_name", job.Name,
"channel", job.DeliverChannel,
"to", job.DeliverTo,
"content_len", len(content),
)
return
}
outMsg := bus.OutboundMessage{
Channel: job.DeliverChannel,
ChatID: job.DeliverTo,
Content: content,
}
if peerKind == "group" {
outMsg.Metadata = map[string]string{"group_id": job.DeliverTo}
}
appendMediaToOutbound(&outMsg, media)
msgBus.PublishOutbound(outMsg)
return
}
if job.Deliver {
slog.Warn("cron: delivery configured but channel/chatID missing — output discarded",
"job_id", job.ID, "job_name", job.Name, "channel", job.DeliverChannel, "to", job.DeliverTo)
}
}
// runCommandCronJob executes a deterministic command-payload cron job in-process
// without an LLM turn. On success it delivers the command output (stdout, else
// stderr) like an agent turn. On failure it returns an error so the run is
// recorded as "error" and retried per cron.max_retries — failures are NOT
// delivered, mirroring the agent path where only successful output is announced.
func runCommandCronJob(cfg *config.Config, job *store.CronJob, msgBus *bus.MessageBus, peerKind string) (*store.CronJobResult, error) {
if !cfg.Cron.CommandEnabled {
return nil, fmt.Errorf("cron command payloads are disabled; set cron.command_enabled=true to allow them")
}
spec := job.Payload.Command
if err := store.ValidateCronCommandSpec(spec); err != nil {
return nil, err
}
cmdTimeout := cfg.Cron.CommandTimeoutDuration()
if spec.TimeoutSeconds > 0 {
cmdTimeout = time.Duration(spec.TimeoutSeconds) * time.Second
}
// The job timeout is a hard ceiling above the per-command timeout.
ctx, cancel := context.WithTimeout(store.WithTenantID(context.Background(), job.TenantID), cfg.Cron.JobTimeoutDuration())
defer cancel()
res := cronexec.Run(ctx, cronexec.Spec{
Argv: spec.Argv,
Cwd: spec.Cwd,
Env: spec.Env,
Input: spec.Input,
Timeout: cmdTimeout,
NoOutputTimeout: time.Duration(spec.NoOutputTimeoutSeconds) * time.Second,
OutputMaxBytes: spec.OutputMaxBytes,
})
if res.Status != cronexec.StatusOK {
return nil, res.Err
}
deliverCronOutput(msgBus, job, res.Summary, nil, peerKind)
return &store.CronJobResult{Content: res.Summary}, nil
}
func cronOutputContainsNoReplySentinel(content string) bool {
text := strings.TrimSpace(content)
if text == "" {
+104
View File
@@ -0,0 +1,104 @@
//go:build !windows
package cmd
import (
"context"
"testing"
"time"
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/config"
"github.com/nextlevelbuilder/goclaw/internal/store"
)
func commandCronConfig(enabled bool) *config.Config {
c := &config.Config{}
c.Cron.CommandEnabled = enabled
return c
}
func commandCronJob(spec *store.CronCommandSpec, deliver bool) *store.CronJob {
job := &store.CronJob{
ID: uuid.NewString(),
TenantID: uuid.New(),
Name: "probe",
AgentID: "ops",
UserID: "user-1",
Payload: store.CronPayload{Kind: store.CronPayloadKindCommand, Command: spec},
}
if deliver {
job.Deliver = true
job.DeliverChannel = "telegram"
job.DeliverTo = "chat-1"
}
return job
}
// A command payload must be refused unless cron.command_enabled is set.
func TestCronJobHandler_CommandDisabled(t *testing.T) {
handler := makeCronJobHandler(nil, nil, commandCronConfig(false), nil, nil, nil, nil, nil)
if _, err := handler(commandCronJob(&store.CronCommandSpec{Argv: []string{"sh", "-c", "echo hi"}}, false)); err == nil {
t.Fatal("expected error when cron.command_enabled is false")
}
}
// A successful command runs with zero model tokens and its stdout is delivered.
func TestCronJobHandler_CommandSuccessDelivers(t *testing.T) {
mb := bus.New()
defer mb.Close()
handler := makeCronJobHandler(nil, mb, commandCronConfig(true), nil, nil, nil, nil, nil)
result, err := handler(commandCronJob(&store.CronCommandSpec{Argv: []string{"sh", "-c", "printf hello"}}, true))
if err != nil {
t.Fatalf("command cron returned error: %v", err)
}
if result == nil || result.Content != "hello" {
t.Fatalf("result = %#v, want content hello", result)
}
if result.InputTokens != 0 || result.OutputTokens != 0 {
t.Errorf("command cron must report zero tokens, got in=%d out=%d", result.InputTokens, result.OutputTokens)
}
ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
defer cancel()
got, ok := mb.SubscribeOutbound(ctx)
if !ok {
t.Fatal("expected outbound delivery of command output")
}
if got.Content != "hello" || got.Channel != "telegram" || got.ChatID != "chat-1" {
t.Fatalf("outbound = %#v, want telegram/chat-1/hello", got)
}
}
// A non-zero exit returns an error (recorded as a failed run) and is NOT
// delivered — only successful output is announced.
func TestCronJobHandler_CommandFailureNotDelivered(t *testing.T) {
mb := bus.New()
defer mb.Close()
handler := makeCronJobHandler(nil, mb, commandCronConfig(true), nil, nil, nil, nil, nil)
result, err := handler(commandCronJob(&store.CronCommandSpec{Argv: []string{"sh", "-c", "echo boom 1>&2; exit 3"}}, true))
if err == nil {
t.Fatal("expected error for non-zero command exit")
}
if result != nil {
t.Fatalf("failed command should return nil result, got %#v", result)
}
ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
defer cancel()
if got, ok := mb.SubscribeOutbound(ctx); ok {
t.Fatalf("failed command must not deliver, got %#v", got)
}
}
// An empty argv is rejected before execution.
func TestCronJobHandler_CommandInvalidSpec(t *testing.T) {
handler := makeCronJobHandler(nil, nil, commandCronConfig(true), nil, nil, nil, nil, nil)
if _, err := handler(commandCronJob(&store.CronCommandSpec{}, false)); err == nil {
t.Fatal("expected error for empty argv")
}
}
+2
View File
@@ -32,6 +32,7 @@ func wireExtraTools(
agentCfg config.AgentDefaults,
globalSkillsDir string,
builtinSkillsDir string,
cronCommandEnabled bool,
) (heartbeatTool *tools.HeartbeatTool, hasMemory bool) {
// web_search: tenant-scoped resolve requires stores + msgBus — register here.
toolsReg.Register(tools.NewWebSearchTool(pgStores.ConfigSecrets, msgBus))
@@ -44,6 +45,7 @@ func wireExtraTools(
// Cron tool (agent-facing)
cronTool := tools.NewCronTool(pgStores.Cron)
cronTool.SetProviderStore(pgStores.Providers)
cronTool.SetCommandEnabled(cronCommandEnabled)
toolsReg.Register(cronTool)
slog.Info("cron tool registered")
+55
View File
@@ -202,12 +202,67 @@ Both jobs run with 5-minute timeout and tenant-scoped context. Failed analyses l
---
## 6. Command Payloads — Deterministic (No LLM)
Most cron jobs run an **agent turn**: the scheduled `message` is sent to the LLM, which costs model tokens on every fire. For purely deterministic work — health probes, backups, syncs, anything that does not need the model — a job can instead carry a **command payload** that runs a shell command directly in the gateway process, with **zero model tokens**.
A job is a command job when its payload `kind` is `command` and it carries a `command` spec instead of a `message`:
| Field | Meaning |
|-------|---------|
| `argv` | Executable + args (no shell parsing). Wrap as `["sh","-c","…"]` for shell syntax. |
| `cwd` | Working directory (default: gateway process cwd) |
| `env` | Extra environment variables, merged over the gateway env |
| `input` | Written to the command's stdin |
| `timeoutSeconds` | Per-command wall-clock timeout (default: `cron.command_timeout`) |
| `noOutputTimeoutSeconds` | Kill if no output is produced for this long (0 = disabled) |
| `outputMaxBytes` | Cap on captured stdout/stderr per stream |
### Execution Semantics
- Runs in-process via a dedicated runner (`internal/cronexec`) with process-group termination, so a timed-out command's forked children are also killed.
- Output is the command's stdout (preferred), else stderr. On success the output is delivered to the configured channel exactly like an agent turn (honoring the `NO_REPLY` sentinel).
- A non-zero exit, timeout, or no-output timeout records the run as **error** and is retried per `cron.max_retries`. Failures are **not** delivered — only successful output is announced, so a failing job cannot spam a channel.
- Token usage is recorded as `0` input / `0` output.
### Security Gate
Command payloads run host commands with the gateway process's privileges, so they are **disabled by default**. An operator must opt in per gateway:
```jsonc
{
"cron": {
"command_enabled": true, // allow command payloads (default false)
"command_timeout": "5m" // default per-command timeout
}
}
```
When disabled, both the RPC (`cron.create`) and the agent `cron` tool reject command payloads, and a command job that somehow exists will refuse to run.
### Creating a Command Job
Via the CLI (operator):
```bash
goclaw cron create --name disk-probe --cron '*/15 * * * *' \
--command 'df -h /' --deliver --channel telegram --to '-100123'
goclaw cron create --name nightly-backup --at 2026-07-01T18:00:00Z \
--argv '["/opt/backup.sh","--full"]' --timeout 5m
```
Via the agent `cron` tool (`action: "add"`), set `command` (a shell string) or `commandArgv` (an array) on the job object instead of `message`.
---
## File Reference
| Module | Path | Purpose |
|---|---|---|
| Scheduler | `internal/scheduler/` | Lane-based concurrency (lanes, queue, drop policies, debounce, cancel, draining) |
| Cron service | `internal/cron/` | In-memory run loop (1s tick), job CRUD, retry with backoff, schedule parsing, types |
| Command runner | `internal/cronexec/` | Deterministic command-payload execution (timeout, no-output watchdog, output cap, process-group kill) |
| Cron store | `internal/store/pg/cron*.go`, `internal/store/cron_store.go` | CronStore interface + PostgreSQL persistence (create, list, update, delete, execution, scanning) |
| Gateway wiring | `cmd/gateway_cron.go`, `internal/gateway/methods/cron.go` | Scheduler lane routing, RPC handlers (list, create, update, delete, toggle, run, runs) |
+18
View File
@@ -480,11 +480,16 @@ type CronConfig struct {
RetryMaxDelay string `json:"retry_max_delay,omitempty"` // maximum backoff delay (default "30s", Go duration)
DefaultTimezone string `json:"default_timezone,omitempty"` // IANA timezone for cron expressions when not set per-job (e.g. "Asia/Ho_Chi_Minh")
JobTimeout string `json:"job_timeout,omitempty"` // max duration per cron job execution (default "10m", Go duration)
CommandEnabled bool `json:"command_enabled,omitempty"` // allow deterministic shell-command cron payloads (kind="command"). Default false. These run inside the gateway process with its privileges — enable only on trusted deployments.
CommandTimeout string `json:"command_timeout,omitempty"` // default per-command wall-clock timeout when a job sets none (default "5m", Go duration)
}
// DefaultJobTimeout is the fallback timeout for cron job execution.
const DefaultJobTimeout = 10 * time.Minute
// DefaultCommandTimeout is the fallback per-command timeout for command cron payloads.
const DefaultCommandTimeout = 5 * time.Minute
// JobTimeoutDuration returns the configured job timeout or the default (10m).
func (cc CronConfig) JobTimeoutDuration() time.Duration {
if cc.JobTimeout != "" {
@@ -497,6 +502,19 @@ func (cc CronConfig) JobTimeoutDuration() time.Duration {
return DefaultJobTimeout
}
// CommandTimeoutDuration returns the configured default per-command timeout or
// the default (5m). A per-job timeoutSeconds, when set, overrides this.
func (cc CronConfig) CommandTimeoutDuration() time.Duration {
if cc.CommandTimeout != "" {
d, err := time.ParseDuration(cc.CommandTimeout)
if err == nil && d > 0 {
return d
}
slog.Warn("cron: invalid command_timeout, using default", "value", cc.CommandTimeout, "default", DefaultCommandTimeout)
}
return DefaultCommandTimeout
}
// ToRetryConfig converts CronConfig to cron.RetryConfig with defaults applied.
func (cc CronConfig) ToRetryConfig() cron.RetryConfig {
cfg := cron.DefaultRetryConfig()
+28
View File
@@ -0,0 +1,28 @@
//go:build !windows
package cronexec
import (
"os/exec"
"syscall"
)
// setProcessGroup makes the child its own process-group leader so signalGroup
// can reach the whole tree (shell + forked children) with one kill(2).
func setProcessGroup(cmd *exec.Cmd) {
cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
}
// signalGroup sends SIGTERM (or SIGKILL when kill is true) to the process group
// rooted at cmd.Process.Pid. Because Setpgid was set, pgid == pid, so kill(-pid)
// reaches every forked child.
func signalGroup(cmd *exec.Cmd, kill bool) {
if cmd.Process == nil {
return
}
sig := syscall.SIGTERM
if kill {
sig = syscall.SIGKILL
}
_ = syscall.Kill(-cmd.Process.Pid, sig)
}
+19
View File
@@ -0,0 +1,19 @@
//go:build windows
package cronexec
import "os/exec"
// setProcessGroup is a no-op on Windows; process-group semantics differ and we
// fall back to killing the root process directly.
func setProcessGroup(cmd *exec.Cmd) {}
// signalGroup kills the root process. Windows lacks POSIX process-group signals,
// so child processes are not guaranteed to be reaped; cron commands on Windows
// should avoid spawning long-lived background children.
func signalGroup(cmd *exec.Cmd, _ bool) {
if cmd.Process == nil {
return
}
_ = cmd.Process.Kill()
}
+270
View File
@@ -0,0 +1,270 @@
// Package cronexec runs deterministic shell-command cron payloads inside the
// gateway process WITHOUT invoking an LLM. It mirrors openclaw's
// src/cron/command-runner.ts: wall-clock timeout, a no-output watchdog, output
// capping, and process-group termination so a timed-out command's forked
// children do not survive.
package cronexec
import (
"bytes"
"context"
"errors"
"fmt"
"os"
"os/exec"
"strings"
"sync"
"time"
)
const (
// DefaultOutputMaxBytes caps captured stdout/stderr per stream when the spec
// does not set OutputMaxBytes. 16 KiB matches the cron run-log truncation.
DefaultOutputMaxBytes = 16 * 1024
// killGrace is how long to wait after SIGTERM before escalating to SIGKILL.
killGrace = 5 * time.Second
// watchdogInterval is how often the no-output watchdog wakes to check.
watchdogInterval = 250 * time.Millisecond
)
// Run statuses.
const (
StatusOK = "ok"
StatusError = "error"
)
// Termination reasons (empty when the process exited on its own).
const (
TermTimeout = "timeout"
TermNoOutputTimeout = "no-output-timeout"
)
// Spec is a deterministic command to run.
type Spec struct {
Argv []string
Cwd string
Env map[string]string
Input string
Timeout time.Duration // wall-clock; <=0 means no explicit limit (still bounded by ctx)
NoOutputTimeout time.Duration // kill if no stdout/stderr for this long; <=0 disables
OutputMaxBytes int // per-stream cap; <=0 uses DefaultOutputMaxBytes
}
// Result is the outcome of a command run.
type Result struct {
Status string // StatusOK or StatusError
ExitCode int // process exit code (-1 if not started or signalled)
Stdout string
Stderr string
Summary string // stdout (preferred), else stderr, else combined
Termination string // "", TermTimeout, or TermNoOutputTimeout
Err error // non-nil when Status==StatusError
}
// Run executes spec and returns its outcome. It never returns an error itself;
// failures are reported via Result.Status/Result.Err so callers can record a
// run log uniformly.
func Run(ctx context.Context, spec Spec) Result {
if len(spec.Argv) == 0 {
return Result{Status: StatusError, ExitCode: -1, Err: errors.New("command requires non-empty argv")}
}
runCtx := ctx
if spec.Timeout > 0 {
var cancel context.CancelFunc
runCtx, cancel = context.WithTimeout(ctx, spec.Timeout)
defer cancel()
}
cmd := exec.Command(spec.Argv[0], spec.Argv[1:]...)
if spec.Cwd != "" {
cmd.Dir = spec.Cwd
}
if len(spec.Env) > 0 {
cmd.Env = mergedEnv(spec.Env)
}
if spec.Input != "" {
cmd.Stdin = strings.NewReader(spec.Input)
}
maxBytes := spec.OutputMaxBytes
if maxBytes <= 0 {
maxBytes = DefaultOutputMaxBytes
}
stdout := newCappedBuffer(maxBytes)
stderr := newCappedBuffer(maxBytes)
cmd.Stdout = stdout
cmd.Stderr = stderr
setProcessGroup(cmd) // platform-specific: own process group for tree kill
if err := cmd.Start(); err != nil {
return Result{Status: StatusError, ExitCode: -1, Err: fmt.Errorf("command failed to start: %w", err)}
}
waitErr := make(chan error, 1)
go func() { waitErr <- cmd.Wait() }()
start := time.Now()
var watchC <-chan time.Time
if spec.NoOutputTimeout > 0 {
ticker := time.NewTicker(watchdogInterval)
defer ticker.Stop()
watchC = ticker.C
}
for {
select {
case err := <-waitErr:
return finalize(stdout, stderr, err, "")
case <-runCtx.Done():
return finalize(stdout, stderr, terminate(cmd, waitErr), TermTimeout)
case <-watchC:
last := latest(stdout.LastWrite(), stderr.LastWrite())
if last.IsZero() {
last = start
}
if time.Since(last) >= spec.NoOutputTimeout {
return finalize(stdout, stderr, terminate(cmd, waitErr), TermNoOutputTimeout)
}
}
}
}
// terminate signals the process group (SIGTERM, then SIGKILL after killGrace),
// waits for the process to be reaped, and returns cmd.Wait()'s error.
func terminate(cmd *exec.Cmd, waitErr <-chan error) error {
signalGroup(cmd, false) // SIGTERM
t := time.AfterFunc(killGrace, func() { signalGroup(cmd, true) }) // SIGKILL
err := <-waitErr
t.Stop()
return err
}
func finalize(stdout, stderr *cappedBuffer, waitErr error, termination string) Result {
outStr := stdout.String()
errStr := stderr.String()
exitCode := 0
signalled := false
if waitErr != nil {
var ee *exec.ExitError
if errors.As(waitErr, &ee) {
exitCode = ee.ExitCode() // -1 when terminated by a signal
if exitCode < 0 {
signalled = true
}
} else {
exitCode = -1
}
}
res := Result{
ExitCode: exitCode,
Stdout: outStr,
Stderr: errStr,
Summary: buildSummary(outStr, errStr),
Termination: termination,
}
if termination == "" && waitErr == nil && exitCode == 0 && !signalled {
res.Status = StatusOK
return res
}
res.Status = StatusError
res.Err = commandError(termination, exitCode, errStr)
return res
}
func commandError(termination string, exitCode int, stderr string) error {
switch termination {
case TermTimeout:
return errors.New("command timed out")
case TermNoOutputTimeout:
return errors.New("command produced no output before the no-output timeout")
}
msg := fmt.Sprintf("command exited with code %d", exitCode)
if s := strings.TrimSpace(stderr); s != "" {
msg += ": " + truncate(s, 500)
}
return errors.New(msg)
}
// buildSummary mirrors openclaw: prefer stdout, fall back to stderr, and when
// both are present, label them.
func buildSummary(stdout, stderr string) string {
so := strings.TrimSpace(stdout)
se := strings.TrimSpace(stderr)
switch {
case so != "" && se != "":
return "stdout:\n" + so + "\n\nstderr:\n" + se
case so != "":
return so
default:
return se
}
}
func mergedEnv(extra map[string]string) []string {
env := os.Environ()
for k, v := range extra {
env = append(env, k+"="+v)
}
return env
}
func latest(a, b time.Time) time.Time {
if b.After(a) {
return b
}
return a
}
func truncate(s string, max int) string {
if len(s) <= max {
return s
}
return s[:max] + "...[truncated]"
}
// cappedBuffer is an io.Writer that retains at most max bytes and records the
// time of the last write (for the no-output watchdog). Writes never error or
// block, so the child process is never throttled by a full buffer.
type cappedBuffer struct {
mu sync.Mutex
buf bytes.Buffer
max int
last time.Time
}
func newCappedBuffer(max int) *cappedBuffer {
return &cappedBuffer{max: max}
}
func (c *cappedBuffer) Write(p []byte) (int, error) {
c.mu.Lock()
defer c.mu.Unlock()
c.last = time.Now()
if c.max > 0 {
remaining := c.max - c.buf.Len()
if remaining <= 0 {
return len(p), nil // capped: discard but report full write
}
if len(p) > remaining {
c.buf.Write(p[:remaining])
return len(p), nil
}
}
return c.buf.Write(p)
}
func (c *cappedBuffer) String() string {
c.mu.Lock()
defer c.mu.Unlock()
return c.buf.String()
}
func (c *cappedBuffer) LastWrite() time.Time {
c.mu.Lock()
defer c.mu.Unlock()
return c.last
}
+142
View File
@@ -0,0 +1,142 @@
//go:build !windows
package cronexec
import (
"context"
"strings"
"testing"
"time"
)
func TestRun_StdoutOK(t *testing.T) {
res := Run(context.Background(), Spec{Argv: []string{"sh", "-c", "printf 'hello world'"}})
if res.Status != StatusOK {
t.Fatalf("status = %q, want ok (err=%v)", res.Status, res.Err)
}
if res.ExitCode != 0 {
t.Errorf("exit code = %d, want 0", res.ExitCode)
}
if res.Summary != "hello world" {
t.Errorf("summary = %q, want %q", res.Summary, "hello world")
}
}
func TestRun_NonZeroExitIsError(t *testing.T) {
res := Run(context.Background(), Spec{Argv: []string{"sh", "-c", "echo oops 1>&2; exit 7"}})
if res.Status != StatusError {
t.Fatalf("status = %q, want error", res.Status)
}
if res.ExitCode != 7 {
t.Errorf("exit code = %d, want 7", res.ExitCode)
}
if res.Err == nil || !strings.Contains(res.Err.Error(), "code 7") {
t.Errorf("err = %v, want it to mention exit code 7", res.Err)
}
if !strings.Contains(res.Err.Error(), "oops") {
t.Errorf("err = %v, want it to include stderr (oops)", res.Err)
}
if !strings.Contains(res.Summary, "oops") {
t.Errorf("summary = %q, want it to include stderr", res.Summary)
}
}
func TestRun_Timeout(t *testing.T) {
start := time.Now()
res := Run(context.Background(), Spec{
Argv: []string{"sh", "-c", "sleep 10"},
Timeout: 200 * time.Millisecond,
})
if res.Status != StatusError {
t.Fatalf("status = %q, want error", res.Status)
}
if res.Termination != TermTimeout {
t.Errorf("termination = %q, want %q", res.Termination, TermTimeout)
}
if elapsed := time.Since(start); elapsed > 5*time.Second {
t.Errorf("took %s, expected the watchdog to kill it promptly", elapsed)
}
}
func TestRun_NoOutputTimeout(t *testing.T) {
res := Run(context.Background(), Spec{
Argv: []string{"sh", "-c", "sleep 10"},
NoOutputTimeout: 200 * time.Millisecond,
})
if res.Status != StatusError {
t.Fatalf("status = %q, want error", res.Status)
}
if res.Termination != TermNoOutputTimeout {
t.Errorf("termination = %q, want %q", res.Termination, TermNoOutputTimeout)
}
}
func TestRun_NoOutputTimeoutNotTrippedWhenProducingOutput(t *testing.T) {
// Emits a line every 50ms for ~300ms, then exits 0. A 200ms no-output
// timeout must NOT fire because output keeps arriving.
res := Run(context.Background(), Spec{
Argv: []string{"sh", "-c", "for i in 1 2 3 4 5 6; do echo tick; sleep 0.05; done"},
NoOutputTimeout: 200 * time.Millisecond,
Timeout: 5 * time.Second,
})
if res.Status != StatusOK {
t.Fatalf("status = %q, want ok (term=%q err=%v)", res.Status, res.Termination, res.Err)
}
}
func TestRun_Stdin(t *testing.T) {
res := Run(context.Background(), Spec{
Argv: []string{"sh", "-c", "cat"},
Input: "piped-input",
})
if res.Status != StatusOK {
t.Fatalf("status = %q, want ok", res.Status)
}
if res.Summary != "piped-input" {
t.Errorf("summary = %q, want %q", res.Summary, "piped-input")
}
}
func TestRun_Env(t *testing.T) {
res := Run(context.Background(), Spec{
Argv: []string{"sh", "-c", "printf '%s' \"$CRON_TEST_VAR\""},
Env: map[string]string{"CRON_TEST_VAR": "from-env"},
})
if res.Status != StatusOK {
t.Fatalf("status = %q, want ok", res.Status)
}
if res.Summary != "from-env" {
t.Errorf("summary = %q, want %q", res.Summary, "from-env")
}
}
func TestRun_OutputCap(t *testing.T) {
res := Run(context.Background(), Spec{
Argv: []string{"sh", "-c", "yes x | head -c 100000"},
OutputMaxBytes: 1024,
Timeout: 5 * time.Second,
})
if res.Status != StatusOK {
t.Fatalf("status = %q, want ok (err=%v)", res.Status, res.Err)
}
if len(res.Stdout) > 1024 {
t.Errorf("stdout length = %d, want capped at 1024", len(res.Stdout))
}
}
func TestRun_EmptyArgv(t *testing.T) {
res := Run(context.Background(), Spec{})
if res.Status != StatusError || res.Err == nil {
t.Fatalf("empty argv should error, got status=%q err=%v", res.Status, res.Err)
}
}
func TestRun_StdoutPreferredOverStderr(t *testing.T) {
res := Run(context.Background(), Spec{Argv: []string{"sh", "-c", "echo out; echo err 1>&2"}})
if res.Status != StatusOK {
t.Fatalf("status = %q, want ok", res.Status)
}
if !strings.Contains(res.Summary, "stdout:") || !strings.Contains(res.Summary, "stderr:") {
t.Errorf("summary should label both streams, got %q", res.Summary)
}
}
+40 -10
View File
@@ -63,15 +63,16 @@ func (m *CronMethods) handleList(ctx context.Context, client *gateway.Client, re
func (m *CronMethods) handleCreate(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) {
locale := store.LocaleFromContext(ctx)
var params struct {
Name string `json:"name"`
Schedule store.CronSchedule `json:"schedule"`
Message string `json:"message"`
Deliver bool `json:"deliver"`
DeliverChannel string `json:"deliverChannel"`
DeliverTo string `json:"deliverTo"`
WakeHeartbeat bool `json:"wakeHeartbeat"`
Stateless *bool `json:"stateless"` // default true for new crons
AgentID string `json:"agentId"`
Name string `json:"name"`
Schedule store.CronSchedule `json:"schedule"`
Message string `json:"message"`
Command *store.CronCommandSpec `json:"command"` // set → deterministic command payload (no LLM)
Deliver bool `json:"deliver"`
DeliverChannel string `json:"deliverChannel"`
DeliverTo string `json:"deliverTo"`
WakeHeartbeat bool `json:"wakeHeartbeat"`
Stateless *bool `json:"stateless"` // default true for new crons
AgentID string `json:"agentId"`
}
if req.Params != nil {
json.Unmarshal(req.Params, &params)
@@ -85,7 +86,18 @@ func (m *CronMethods) handleCreate(ctx context.Context, client *gateway.Client,
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidSlug, "name")))
return
}
if params.Message == "" {
isCommand := params.Command != nil
if isCommand {
if !m.cfg.Cron.CommandEnabled {
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgCommandCronDisabled)))
return
}
if err := store.ValidateCronCommandSpec(params.Command); err != nil {
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, err.Error()))
return
}
} else if params.Message == "" {
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgMsgRequired)))
return
}
@@ -107,6 +119,9 @@ func (m *CronMethods) handleCreate(ctx context.Context, client *gateway.Client,
if params.WakeHeartbeat {
patch.WakeHeartbeat = &params.WakeHeartbeat
}
if isCommand {
patch.Command = params.Command
}
if updated, pErr := m.service.UpdateJob(ctx, job.ID, patch); pErr == nil {
job = updated
}
@@ -243,6 +258,21 @@ func (m *CronMethods) handleUpdate(ctx context.Context, client *gateway.Client,
return
}
// A command payload on update must clear the same gate as create: command
// cron must be enabled and the spec must be valid. Without this, a normal job
// could be mutated into a command job (or persisted with an invalid spec) on a
// gateway where command cron is disabled.
if params.Patch.Command != nil {
if !m.cfg.Cron.CommandEnabled {
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgCommandCronDisabled)))
return
}
if err := store.ValidateCronCommandSpec(params.Patch.Command); err != nil {
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, err.Error()))
return
}
}
job, err := m.service.UpdateJob(ctx, jobID, params.Patch)
if err != nil {
code := protocol.ErrInvalidRequest
+62
View File
@@ -352,3 +352,65 @@ func TestCronRun_BlocksCredentialBoundJobByDifferentUser(t *testing.T) {
t.Fatalf("RunJob called %d times, want 0", svc.runCnt)
}
}
// ---- Tests: handleUpdate command gate ----
// A normal job must not be mutable into a command job when the gateway has
// command cron disabled.
func TestCronUpdate_BlocksCommandWhenDisabled(t *testing.T) {
svc := newStubCronStore()
svc.jobs["job-1"] = &store.CronJob{ID: "job-1", UserID: ""}
m := buildCronMethods(t, svc) // cfg.Cron.CommandEnabled defaults to false
client := nullClient()
req := cronReqFrame(t, protocol.MethodCronUpdate, map[string]any{
"jobId": "job-1",
"patch": map[string]any{"command": map[string]any{"argv": []any{"echo", "hi"}}},
})
m.handleUpdate(context.Background(), client, req)
if svc.updateCnt != 0 {
t.Fatalf("UpdateJob called %d times, want 0", svc.updateCnt)
}
}
// Update must validate the command spec; an empty argv must be rejected even
// when command cron is enabled.
func TestCronUpdate_RejectsInvalidCommandSpec(t *testing.T) {
svc := newStubCronStore()
svc.jobs["job-1"] = &store.CronJob{ID: "job-1", UserID: ""}
cfg := &config.Config{}
cfg.Cron.CommandEnabled = true
m := NewCronMethods(svc, &stubEventPub{}, cfg)
client := nullClient()
req := cronReqFrame(t, protocol.MethodCronUpdate, map[string]any{
"jobId": "job-1",
"patch": map[string]any{"command": map[string]any{"argv": []any{}}}, // empty argv → invalid
})
m.handleUpdate(context.Background(), client, req)
if svc.updateCnt != 0 {
t.Fatalf("UpdateJob called %d times, want 0", svc.updateCnt)
}
}
// A valid command payload on update is accepted when command cron is enabled.
func TestCronUpdate_EnabledValidCommand_Updates(t *testing.T) {
svc := newStubCronStore()
svc.jobs["job-1"] = &store.CronJob{ID: "job-1", UserID: ""}
cfg := &config.Config{}
cfg.Cron.CommandEnabled = true
m := NewCronMethods(svc, &stubEventPub{}, cfg)
client := nullClient()
req := cronReqFrame(t, protocol.MethodCronUpdate, map[string]any{
"jobId": "job-1",
"patch": map[string]any{"command": map[string]any{"argv": []any{"df", "-h"}}},
})
m.handleUpdate(context.Background(), client, req)
if svc.updateCnt != 1 {
t.Fatalf("UpdateJob called %d times, want 1", svc.updateCnt)
}
}
+3 -2
View File
@@ -52,8 +52,9 @@ func init() {
MsgInstanceNotFound: "instance not found",
// Cron
MsgJobNotFound: "job not found",
MsgInvalidCronExpr: "invalid cron expression: %s",
MsgJobNotFound: "job not found",
MsgInvalidCronExpr: "invalid cron expression: %s",
MsgCommandCronDisabled: "command cron jobs are disabled on this gateway (set cron.command_enabled=true to allow them)",
// Config
MsgConfigHashMismatch: "config has changed (hash mismatch)",
+3 -2
View File
@@ -52,8 +52,9 @@ func init() {
MsgInstanceNotFound: "không tìm thấy phiên bản",
// Cron
MsgJobNotFound: "không tìm thấy tác vụ",
MsgInvalidCronExpr: "biểu thức cron không hợp lệ: %s",
MsgJobNotFound: "không tìm thấy tác vụ",
MsgInvalidCronExpr: "biểu thức cron không hợp lệ: %s",
MsgCommandCronDisabled: "tác vụ cron dạng lệnh đang bị tắt trên gateway này (đặt cron.command_enabled=true để cho phép)",
// Config
MsgConfigHashMismatch: "cấu hình đã thay đổi (hash không khớp)",
+3 -2
View File
@@ -52,8 +52,9 @@ func init() {
MsgInstanceNotFound: "未找到实例",
// Cron
MsgJobNotFound: "未找到任务",
MsgInvalidCronExpr: "无效的 cron 表达式:%s",
MsgJobNotFound: "未找到任务",
MsgInvalidCronExpr: "无效的 cron 表达式:%s",
MsgCommandCronDisabled: "此网关已禁用命令型 cron 任务(设置 cron.command_enabled=true 以允许)",
// Config
MsgConfigHashMismatch: "配置已更改(hash 不匹配)",
+3 -2
View File
@@ -53,8 +53,9 @@ const (
MsgInstanceNotFound = "error.instance_not_found" // "instance not found"
// --- Cron ---
MsgJobNotFound = "error.job_not_found" // "job not found"
MsgInvalidCronExpr = "error.invalid_cron_expr" // "invalid cron expression: %s"
MsgJobNotFound = "error.job_not_found" // "job not found"
MsgInvalidCronExpr = "error.invalid_cron_expr" // "invalid cron expression: %s"
MsgCommandCronDisabled = "error.command_cron_disabled" // "command cron jobs are disabled on this gateway"
// --- Config ---
MsgConfigHashMismatch = "error.config_hash_mismatch" // "config has changed (hash mismatch)"
+69 -17
View File
@@ -15,6 +15,14 @@ var (
ErrCronJobNotFound = errors.New("cron job not found")
ErrCronJobNoFutureRun = errors.New("cron job has no future run")
ErrCronCredentialOwnerMismatch = errors.New("cron credential context belongs to a different user")
ErrCronCommandInvalid = errors.New("invalid cron command payload")
)
// Cron payload kinds. A job either runs an LLM/agent turn ("agent_turn") or a
// deterministic shell command in the gateway process ("command", no model cost).
const (
CronPayloadKindAgentTurn = "agent_turn"
CronPayloadKindCommand = "command"
)
// CronJob represents a scheduled job.
@@ -49,12 +57,55 @@ type CronSchedule struct {
TZ string `json:"tz,omitempty" db:"-"`
}
// CronCommandSpec is a deterministic shell command run by a cron job whose
// payload Kind is "command". It executes inside the gateway process WITHOUT an
// LLM/agent turn (zero model tokens). It is a trusted automation surface gated
// by cron.command_enabled — argv is run with the gateway process's privileges.
type CronCommandSpec struct {
// Argv is the executable plus arguments. No shell parsing is done; wrap as
// ["sh", "-c", "<script>"] for shell syntax.
Argv []string `json:"argv,omitempty"`
// Cwd is the working directory (default: gateway process cwd).
Cwd string `json:"cwd,omitempty"`
// Env adds environment variables, merged over the gateway process env.
Env map[string]string `json:"env,omitempty"`
// Input is written to the command's stdin.
Input string `json:"input,omitempty"`
// TimeoutSeconds is the per-command wall-clock timeout (0 = cron.command_timeout default).
TimeoutSeconds int `json:"timeoutSeconds,omitempty"`
// NoOutputTimeoutSeconds kills the command if it produces no output for this
// long (0 = disabled).
NoOutputTimeoutSeconds int `json:"noOutputTimeoutSeconds,omitempty"`
// OutputMaxBytes caps captured stdout/stderr per stream (0 = default).
OutputMaxBytes int `json:"outputMaxBytes,omitempty"`
}
// CronPayload describes what a job does when triggered.
type CronPayload struct {
Kind string `json:"kind" db:"-"`
Message string `json:"message" db:"-"`
Command string `json:"command,omitempty" db:"-"`
CredentialUserID string `json:"credentialUserId,omitempty" db:"-"`
Kind string `json:"kind" db:"-"`
Message string `json:"message" db:"-"`
Command *CronCommandSpec `json:"command,omitempty" db:"-"` // set when Kind=="command" (deterministic, no LLM)
CredentialUserID string `json:"credentialUserId,omitempty" db:"-"`
}
// IsCommand reports whether this payload runs a deterministic shell command
// (no LLM turn) rather than an agent turn.
func (p CronPayload) IsCommand() bool {
return p.Kind == CronPayloadKindCommand
}
// ValidateCronCommandSpec checks structural validity of a command payload.
func ValidateCronCommandSpec(spec *CronCommandSpec) error {
if spec == nil || len(spec.Argv) == 0 {
return fmt.Errorf("%w: argv must be non-empty", ErrCronCommandInvalid)
}
if spec.Argv[0] == "" {
return fmt.Errorf("%w: argv[0] (the executable) must not be empty", ErrCronCommandInvalid)
}
if spec.TimeoutSeconds < 0 || spec.NoOutputTimeoutSeconds < 0 || spec.OutputMaxBytes < 0 {
return fmt.Errorf("%w: timeout/output limits must be non-negative", ErrCronCommandInvalid)
}
return nil
}
// CheckCronCredentialOwner blocks user-triggered mutation/execution of a
@@ -117,19 +168,20 @@ type CronJobResult struct {
// CronJobPatch holds optional fields for updating a job.
type CronJobPatch struct {
Name string `json:"name,omitempty" db:"-"`
AgentID *string `json:"agentId,omitempty" db:"-"`
Enabled *bool `json:"enabled,omitempty" db:"-"`
Schedule *CronSchedule `json:"schedule,omitempty" db:"-"`
Message string `json:"message,omitempty" db:"-"`
DeleteAfterRun *bool `json:"deleteAfterRun,omitempty" db:"-"`
Stateless *bool `json:"stateless,omitempty" db:"-"`
Deliver *bool `json:"deliver,omitempty" db:"-"`
DeliverChannel *string `json:"deliverChannel,omitempty" db:"-"`
DeliverTo *string `json:"deliverTo,omitempty" db:"-"`
WakeHeartbeat *bool `json:"wakeHeartbeat,omitempty" db:"-"`
ProviderID *uuid.UUID `json:"providerId,omitempty" db:"-"`
Model *string `json:"model,omitempty" db:"-"`
Name string `json:"name,omitempty" db:"-"`
AgentID *string `json:"agentId,omitempty" db:"-"`
Enabled *bool `json:"enabled,omitempty" db:"-"`
Schedule *CronSchedule `json:"schedule,omitempty" db:"-"`
Message string `json:"message,omitempty" db:"-"`
Command *CronCommandSpec `json:"command,omitempty" db:"-"` // set → switches the job to a deterministic command payload
DeleteAfterRun *bool `json:"deleteAfterRun,omitempty" db:"-"`
Stateless *bool `json:"stateless,omitempty" db:"-"`
Deliver *bool `json:"deliver,omitempty" db:"-"`
DeliverChannel *string `json:"deliverChannel,omitempty" db:"-"`
DeliverTo *string `json:"deliverTo,omitempty" db:"-"`
WakeHeartbeat *bool `json:"wakeHeartbeat,omitempty" db:"-"`
ProviderID *uuid.UUID `json:"providerId,omitempty" db:"-"`
Model *string `json:"model,omitempty" db:"-"`
}
// CronEvent represents a job lifecycle event sent to subscribers.
+8 -2
View File
@@ -96,9 +96,15 @@ func (s *PGCronStore) UpdateJob(ctx context.Context, jobID string, patch store.C
updates["model"] = *patch.Model
}
if patch.Message != "" {
if patch.Message != "" || patch.Command != nil {
payload := current.Payload
payload.Message = patch.Message
if patch.Message != "" {
payload.Message = patch.Message
}
if patch.Command != nil {
payload.Kind = store.CronPayloadKindCommand
payload.Command = patch.Command
}
mergedPayload, err := json.Marshal(payload)
if err != nil {
return nil, fmt.Errorf("failed to marshal payload for job %s: %w", jobID, err)
+8 -2
View File
@@ -310,9 +310,15 @@ func (s *SQLiteCronStore) UpdateJob(ctx context.Context, jobID string, patch sto
updates["model"] = *patch.Model
}
if patch.Message != "" {
if patch.Message != "" || patch.Command != nil {
payload := current.Payload
payload.Message = patch.Message
if patch.Message != "" {
payload.Message = patch.Message
}
if patch.Command != nil {
payload.Kind = store.CronPayloadKindCommand
payload.Command = patch.Command
}
mergedPayload, err := json.Marshal(payload)
if err != nil {
return nil, fmt.Errorf("failed to marshal payload for job %s: %w", jobID, err)
+102 -6
View File
@@ -13,9 +13,10 @@ import (
// CronTool lets agents manage Gateway cron jobs.
// Matching OpenClaw src/agents/tools/cron-tool.ts.
type CronTool struct {
cronStore store.CronStore
permStore store.ConfigPermissionStore // nil = no group restriction
providerStore store.ProviderStore // nil = provider override by name unavailable
cronStore store.CronStore
permStore store.ConfigPermissionStore // nil = no group restriction
providerStore store.ProviderStore // nil = provider override by name unavailable
commandEnabled bool // allow deterministic command payloads (mirrors cron.command_enabled)
}
func NewCronTool(cronStore store.CronStore) *CronTool {
@@ -32,6 +33,12 @@ func (t *CronTool) SetProviderStore(s store.ProviderStore) {
t.providerStore = s
}
// SetCommandEnabled allows this tool to create deterministic command-payload
// cron jobs (kind="command", no LLM). Mirrors the gateway's cron.command_enabled.
func (t *CronTool) SetCommandEnabled(enabled bool) {
t.commandEnabled = enabled
}
func (t *CronTool) Name() string { return "cron" }
func (t *CronTool) Description() string {
@@ -103,7 +110,16 @@ RULES:
- "name" must match: lowercase letters, numbers, hyphens only.
- Before creating or updating a scheduled job, call the datetime tool first to get the precise current time and unix_ms timestamp. Never guess timestamps.
- Omit optional fields when unknown; do not invent placeholder values like "", 0, or null unless required.
- Jobs run as isolated agent turns using the provided "message".`
- Jobs run as isolated agent turns using the provided "message".
DETERMINISTIC COMMAND JOBS (no LLM, zero tokens):
- Instead of "message", set "command" to a shell string (run as sh -c) OR
"commandArgv" to an explicit argv array (no shell parsing).
- Optional: "commandCwd", "commandEnv" {"KEY":"VAL"}, "commandTimeoutSeconds".
- The command runs inside the gateway process. Only available when the gateway
has cron.command_enabled=true; otherwise add returns an error.
- Use for scheduled probes/scripts that don't need the model. Output is delivered
like a normal job when "deliver" is set; a non-zero exit records the run as an error.`
}
func (t *CronTool) Parameters() map[string]any {
@@ -222,9 +238,20 @@ func (t *CronTool) handleAdd(ctx context.Context, args map[string]any, agentID,
return ErrorResult("job.schedule is required")
}
// Optional deterministic command payload (runs a shell command, no LLM turn).
cmdSpec := parseCronCommandSpec(jobObj)
if cmdSpec != nil {
if !t.commandEnabled {
return ErrorResult("command cron is disabled on this gateway (set cron.command_enabled=true to allow it)")
}
if err := store.ValidateCronCommandSpec(cmdSpec); err != nil {
return ErrorResult(err.Error())
}
}
message, _ := jobObj["message"].(string)
if message == "" {
return ErrorResult("job.message is required")
if cmdSpec == nil && message == "" {
return ErrorResult("job.message is required (or set job.command/job.commandArgv for a deterministic command job)")
}
// Parse schedule
@@ -332,6 +359,10 @@ func (t *CronTool) handleAdd(ctx context.Context, args map[string]any, agentID,
overridePatch.Model = &modelOverride
needOverride = true
}
if cmdSpec != nil {
overridePatch.Command = cmdSpec
needOverride = true
}
if needOverride {
if updated, uErr := t.cronStore.UpdateJob(ctx, job.ID, overridePatch); uErr == nil {
job = updated
@@ -380,11 +411,30 @@ func (t *CronTool) handleUpdate(ctx context.Context, args map[string]any, agentI
return ErrorResult("patch object is required for update action")
}
// A command payload on update follows the same forms (shell string or argv)
// and the same gate as add. Parse it out first and drop the raw keys so the
// generic patch unmarshal below — whose Command field is a structured spec —
// can't choke on a shell string, and so update can't slip a command payload
// past the command-enabled gate.
cmdSpec := parseCronCommandSpec(patchObj)
delete(patchObj, "command")
delete(patchObj, "commandArgv")
var patch store.CronJobPatch
// Re-marshal and unmarshal to leverage JSON tags
patchJSON, _ := json.Marshal(patchObj)
json.Unmarshal(patchJSON, &patch)
if cmdSpec != nil {
if !t.commandEnabled {
return ErrorResult("command cron is disabled on this gateway (set cron.command_enabled=true to allow it)")
}
if err := store.ValidateCronCommandSpec(cmdSpec); err != nil {
return ErrorResult(err.Error())
}
patch.Command = cmdSpec
}
// Resolve provider override by name (providerId UUID is handled by JSON tags above).
if name, _ := patchObj["provider"].(string); name != "" {
pid, errR := t.resolveProviderID(ctx, patchObj)
@@ -533,3 +583,49 @@ func numberFromMap(m map[string]any, key string) (float64, bool) {
v, ok := m[key].(float64)
return v, ok
}
// parseCronCommandSpec extracts an optional deterministic command payload from a
// cron tool "job" object. It accepts either a shell string ("command", wrapped
// as ["sh","-c",...]) or an explicit argv array ("commandArgv"). Returns nil
// when no command fields are present (i.e. it's a normal agent-turn job).
func parseCronCommandSpec(jobObj map[string]any) *store.CronCommandSpec {
var argv []string
if raw, ok := jobObj["commandArgv"].([]any); ok {
for _, v := range raw {
if s, ok := v.(string); ok {
argv = append(argv, s)
}
}
}
if len(argv) == 0 {
if s, _ := jobObj["command"].(string); s != "" {
argv = []string{"sh", "-c", s}
}
}
if len(argv) == 0 {
return nil
}
spec := &store.CronCommandSpec{Argv: argv}
if cwd, _ := jobObj["commandCwd"].(string); cwd != "" {
spec.Cwd = cwd
}
if v, ok := numberFromMap(jobObj, "commandTimeoutSeconds"); ok {
spec.TimeoutSeconds = int(v)
}
if v, ok := numberFromMap(jobObj, "commandNoOutputTimeoutSeconds"); ok {
spec.NoOutputTimeoutSeconds = int(v)
}
if v, ok := numberFromMap(jobObj, "commandOutputMaxBytes"); ok {
spec.OutputMaxBytes = int(v)
}
if env, ok := jobObj["commandEnv"].(map[string]any); ok && len(env) > 0 {
spec.Env = make(map[string]string, len(env))
for k, v := range env {
if s, ok := v.(string); ok {
spec.Env[k] = s
}
}
}
return spec
}
+42
View File
@@ -151,3 +151,45 @@ func TestCronToolBlocksCredentialBoundRunByDifferentUser(t *testing.T) {
t.Fatalf("RunJob called %d times, want 0", cronStore.runCnt)
}
}
// A command payload must not be slipped in via update when command cron is
// disabled — the disabled-gateway contract is that command jobs cannot be
// created OR mutated into existence.
func TestCronToolUpdateBlocksCommandWhenDisabled(t *testing.T) {
cronStore := newTestCronStore(&store.CronJob{ID: "job-1", Payload: store.CronPayload{Message: "old"}})
tool := NewCronTool(cronStore) // commandEnabled defaults to false
result := tool.Execute(context.Background(), map[string]any{
"action": "update",
"jobId": "job-1",
"patch": map[string]any{"commandArgv": []any{"echo", "hi"}},
})
if !result.IsError || !strings.Contains(result.ForLLM, "disabled") {
t.Fatalf("expected command-disabled error, got %#v", result)
}
if cronStore.updateCnt != 0 {
t.Fatalf("UpdateJob called %d times, want 0", cronStore.updateCnt)
}
}
// Update must validate the command spec; an invalid argv must be rejected even
// when command cron is enabled.
func TestCronToolUpdateRejectsInvalidCommandSpec(t *testing.T) {
cronStore := newTestCronStore(&store.CronJob{ID: "job-1", Payload: store.CronPayload{Message: "old"}})
tool := NewCronTool(cronStore)
tool.SetCommandEnabled(true)
result := tool.Execute(context.Background(), map[string]any{
"action": "update",
"jobId": "job-1",
"patch": map[string]any{"commandArgv": []any{""}}, // argv[0] empty → invalid
})
if !result.IsError {
t.Fatalf("expected invalid command spec error, got %#v", result)
}
if cronStore.updateCnt != 0 {
t.Fatalf("UpdateJob called %d times, want 0", cronStore.updateCnt)
}
}