diff --git a/cmd/cron_cmd.go b/cmd/cron_cmd.go index b7edd573..87ed4aea 100644 --- a/cmd/cron_cmd.go +++ b/cmd/cron_cmd.go @@ -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() diff --git a/cmd/gateway.go b/cmd/gateway.go index 1893d33b..2b80a17a 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -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. diff --git a/cmd/gateway_cron.go b/cmd/gateway_cron.go index af2f6155..79c12c84 100644 --- a/cmd/gateway_cron.go +++ b/cmd/gateway_cron.go @@ -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 == "" { diff --git a/cmd/gateway_cron_command_test.go b/cmd/gateway_cron_command_test.go new file mode 100644 index 00000000..bfa46075 --- /dev/null +++ b/cmd/gateway_cron_command_test.go @@ -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") + } +} diff --git a/cmd/gateway_tools_wiring.go b/cmd/gateway_tools_wiring.go index 547fdc46..43d11ec8 100644 --- a/cmd/gateway_tools_wiring.go +++ b/cmd/gateway_tools_wiring.go @@ -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") diff --git a/docs/08-scheduling-cron.md b/docs/08-scheduling-cron.md index e3800e2f..dbe0dda7 100644 --- a/docs/08-scheduling-cron.md +++ b/docs/08-scheduling-cron.md @@ -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) | diff --git a/internal/config/config.go b/internal/config/config.go index 8eb54db7..362e0c88 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -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() diff --git a/internal/cronexec/pgid_unix.go b/internal/cronexec/pgid_unix.go new file mode 100644 index 00000000..2580e5d4 --- /dev/null +++ b/internal/cronexec/pgid_unix.go @@ -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) +} diff --git a/internal/cronexec/pgid_windows.go b/internal/cronexec/pgid_windows.go new file mode 100644 index 00000000..d366f100 --- /dev/null +++ b/internal/cronexec/pgid_windows.go @@ -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() +} diff --git a/internal/cronexec/runner.go b/internal/cronexec/runner.go new file mode 100644 index 00000000..709703fa --- /dev/null +++ b/internal/cronexec/runner.go @@ -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 +} diff --git a/internal/cronexec/runner_test.go b/internal/cronexec/runner_test.go new file mode 100644 index 00000000..aec1c474 --- /dev/null +++ b/internal/cronexec/runner_test.go @@ -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) + } +} diff --git a/internal/gateway/methods/cron.go b/internal/gateway/methods/cron.go index 1eab7dd9..06c46407 100644 --- a/internal/gateway/methods/cron.go +++ b/internal/gateway/methods/cron.go @@ -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, ¶ms) @@ -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 = ¶ms.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 diff --git a/internal/gateway/methods/cron_test.go b/internal/gateway/methods/cron_test.go index 4e5315c9..a366212a 100644 --- a/internal/gateway/methods/cron_test.go +++ b/internal/gateway/methods/cron_test.go @@ -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) + } +} diff --git a/internal/i18n/catalog_en.go b/internal/i18n/catalog_en.go index 07082241..c91ca2b9 100644 --- a/internal/i18n/catalog_en.go +++ b/internal/i18n/catalog_en.go @@ -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)", diff --git a/internal/i18n/catalog_vi.go b/internal/i18n/catalog_vi.go index c13530fe..5e5cc17b 100644 --- a/internal/i18n/catalog_vi.go +++ b/internal/i18n/catalog_vi.go @@ -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)", diff --git a/internal/i18n/catalog_zh.go b/internal/i18n/catalog_zh.go index 76f33417..fa086acd 100644 --- a/internal/i18n/catalog_zh.go +++ b/internal/i18n/catalog_zh.go @@ -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 不匹配）", diff --git a/internal/i18n/keys.go b/internal/i18n/keys.go index e5ced893..d40ae8d2 100644 --- a/internal/i18n/keys.go +++ b/internal/i18n/keys.go @@ -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)" diff --git a/internal/store/cron_store.go b/internal/store/cron_store.go index 4e3204b3..97dcd019 100644 --- a/internal/store/cron_store.go +++ b/internal/store/cron_store.go @@ -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", "