feat(cron): per-job LLM provider/model override (#1234)

makeCronJobHandler builds the agent RunRequest with no provider/model, so cron
jobs always run on the agent default provider. High-frequency scheduled jobs
(e.g. triage) can't be routed to a cheaper model the way heartbeats already can
via agent_heartbeats.provider_id/model.

Add provider_id/model to cron_jobs (Postgres migration 000074; SQLite schema +
migration v42→43, both mirroring agent_heartbeats) and resolve them into
RunRequest.ProviderOverride/ModelOverride in the handler, reusing the existing
per-run override plumbing that heartbeats use. NULL/unset → agent default, so
existing jobs are unaffected.

Settable today via UpdateJob patch (CronJobPatch.ProviderID/Model) or directly
in the DB; tool/RPC/dashboard exposure left as a follow-up.

Verified: go build ./... and go build -tags sqlite ./... (exit 0), go vet,
gofmt, and cron unit tests on both pg and sqlite stores.
This commit is contained in:
Zezae Oh authored and GitHub committed 2026-06-21 00:19:39 +07:00
1 parent fc308a3989
commit 2bf0f9e172
18 files changed
+151 -22

No files matched your search

+22 -1
View File
@@ -12,6 +12,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/providers"
"github.com/nextlevelbuilder/goclaw/internal/scheduler"
"github.com/nextlevelbuilder/goclaw/internal/sessions"
"github.com/nextlevelbuilder/goclaw/internal/store"
@@ -24,7 +25,7 @@ import (
// Safe because cron jobs only fire after Start(), well after this is set.
var cronHeartbeatWakeFn func(agentID string)
func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg *config.Config, channelMgr *channels.Manager, sessionMgr store.SessionStore, agentStore store.AgentStore) func(job *store.CronJob) (*store.CronJobResult, error) {
func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg *config.Config, channelMgr *channels.Manager, sessionMgr store.SessionStore, agentStore store.AgentStore, providerStore store.ProviderStore, providerReg *providers.Registry) func(job *store.CronJob) (*store.CronJobResult, error) {
return func(job *store.CronJob) (*store.CronJobResult, error) {
agentID := job.AgentID
if agentID == "" && agentStore != nil {
@@ -97,6 +98,24 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg
sessionMgr.Save(cronCtx, sessionKey)
}
// Resolve per-job provider/model override (mirrors heartbeat). Unset → agent default.
var providerOverride providers.Provider
if job.ProviderID != nil && providerStore != nil && providerReg != nil {
if provData, perr := providerStore.GetProvider(cronCtx, *job.ProviderID); perr == nil {
if prov, gerr := providerReg.GetForTenant(job.TenantID, provData.Name); gerr == nil {
providerOverride = prov
} else {
slog.Warn("cron.provider_not_in_registry", "job", job.ID, "provider_id", job.ProviderID, "error", gerr)
}
} else {
slog.Warn("cron.provider_not_found", "job", job.ID, "provider_id", job.ProviderID, "error", perr)
}
}
var modelOverride string
if job.Model != nil {
modelOverride = *job.Model
}
// Schedule through cron lane — scheduler handles agent resolution and concurrency
outCh := sched.Schedule(cronCtx, scheduler.LaneCron, agent.RunRequest{
SessionKey: sessionKey,
@@ -108,6 +127,8 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg
UserID: job.UserID,
RunID: fmt.Sprintf("cron:%s", job.ID),
Stream: false,
ModelOverride: modelOverride,
ProviderOverride: providerOverride,
ExtraSystemPrompt: extraPrompt,
TraceName: fmt.Sprintf("Cron [%s] - %s", job.Name, agentID),
TraceTags: []string{"cron"},
+4
View File
@@ -41,6 +41,8 @@ func TestCronJobHandlerInjectsPayloadCredentialUserID(t *testing.T) {
nil,
nil,
nil,
nil,
nil,
)
result, err := handler(&store.CronJob{
@@ -135,6 +137,8 @@ func TestCronJobHandlerSuppressesNoReplyDelivery(t *testing.T) {
nil,
nil,
nil,
nil,
nil,
)
result, err := handler(&store.CronJob{
+1 -1
View File
@@ -42,7 +42,7 @@ func startCronAndHeartbeat(
heartbeatMethods *methods.HeartbeatMethods,
) *heartbeat.Ticker {
// Start cron service with job handler (routes through scheduler's cron lane)
pgStores.Cron.SetOnJob(makeCronJobHandler(sched, msgBus, cfg, channelMgr, pgStores.Sessions, pgStores.Agents))
pgStores.Cron.SetOnJob(makeCronJobHandler(sched, msgBus, cfg, channelMgr, pgStores.Sessions, pgStores.Agents, pgStores.Providers, providerRegistry))
pgStores.Cron.SetOnEvent(func(event store.CronEvent) {
server.BroadcastEvent(*protocol.NewEvent(protocol.EventCron, event))
})
+3 -1
View File
@@ -42,7 +42,9 @@ func wireExtraTools(
toolsReg.Register(tools.NewWaitTool())
// Cron tool (agent-facing)
toolsReg.Register(tools.NewCronTool(pgStores.Cron))
cronTool := tools.NewCronTool(pgStores.Cron)
cronTool.SetProviderStore(pgStores.Providers)
toolsReg.Register(cronTool)
slog.Info("cron tool registered")
// Heartbeat tool (agent-facing)
+4
View File
@@ -36,6 +36,8 @@ type CronJob struct {
DeliverChannel string `json:"deliverChannel" db:"deliver_channel"`
DeliverTo string `json:"deliverTo" db:"deliver_to"`
WakeHeartbeat bool `json:"wakeHeartbeat" db:"wake_heartbeat"`
ProviderID *uuid.UUID `json:"providerId,omitempty" db:"provider_id"`
Model *string `json:"model,omitempty" db:"model"`
}
// CronSchedule defines when a job should run.
@@ -126,6 +128,8 @@ type CronJobPatch struct {
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.
+1 -1
View File
@@ -101,7 +101,7 @@ func (s *PGCronStore) ListJobs(ctx context.Context, includeDisabled bool, agentI
q := `SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
interval_ms, payload, delete_after_run, stateless, deliver, deliver_channel, deliver_to, wake_heartbeat,
next_run_at, last_run_at, last_status, last_error,
created_at, updated_at FROM cron_jobs WHERE 1=1`
created_at, updated_at, provider_id, model FROM cron_jobs WHERE 1=1`
var args []any
argIdx := 1
+7 -2
View File
@@ -17,7 +17,7 @@ func (s *PGCronStore) scanJob(ctx context.Context, id uuid.UUID) (*store.CronJob
q := `SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
interval_ms, payload, delete_after_run, stateless, deliver, deliver_channel, deliver_to, wake_heartbeat,
next_run_at, last_run_at, last_status, last_error,
created_at, updated_at FROM cron_jobs WHERE id = $1`
created_at, updated_at, provider_id, model FROM cron_jobs WHERE id = $1`
args := []any{id}
if !store.IsCrossTenant(ctx) {
@@ -53,11 +53,13 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
var intervalMS *int64
var payloadJSON []byte
var createdAt, updatedAt time.Time
var providerID *uuid.UUID
var model *string
err := row.Scan(&id, &tenantID, &agentID, &userID, &name, &enabled, &scheduleKind, &cronExpr, &runAt, &tz,
&intervalMS, &payloadJSON, &deleteAfterRun, &stateless, &deliver, &deliverChannel, &deliverTo, &wakeHeartbeat,
&nextRunAt, &lastRunAt, &lastStatus, &lastError,
&createdAt, &updatedAt)
&createdAt, &updatedAt, &providerID, &model)
if err != nil {
return nil, err
}
@@ -88,6 +90,9 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
WakeHeartbeat: wakeHeartbeat,
}
job.ProviderID = providerID
job.Model = model
if agentID != nil {
job.AgentID = agentID.String()
}
+2 -2
View File
@@ -40,7 +40,7 @@ func (s *PGCronStore) refreshJobCache() {
`SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
interval_ms, payload, delete_after_run, stateless, deliver, deliver_channel, deliver_to, wake_heartbeat,
next_run_at, last_run_at, last_status, last_error,
created_at, updated_at FROM cron_jobs WHERE enabled = true`)
created_at, updated_at, provider_id, model FROM cron_jobs WHERE enabled = true`)
if err != nil {
return
}
@@ -373,7 +373,7 @@ func (s *PGCronStore) loadClaimedJob(id uuid.UUID) (*store.CronJob, bool) {
`SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
interval_ms, payload, delete_after_run, stateless, deliver, deliver_channel, deliver_to, wake_heartbeat,
next_run_at, last_run_at, last_status, last_error,
created_at, updated_at
created_at, updated_at, provider_id, model
FROM cron_jobs
WHERE id = $1 AND enabled = true AND next_run_at IS NULL`,
id,
+6 -2
View File
@@ -89,6 +89,12 @@ func (s *PGCronStore) UpdateJob(ctx context.Context, jobID string, patch store.C
if patch.WakeHeartbeat != nil {
updates["wake_heartbeat"] = *patch.WakeHeartbeat
}
if patch.ProviderID != nil {
updates["provider_id"] = *patch.ProviderID
}
if patch.Model != nil {
updates["model"] = *patch.Model
}
if patch.Message != "" {
payload := current.Payload
@@ -181,7 +187,6 @@ func (s *PGCronStore) lockCronJobForMutation(ctx context.Context, tx *sql.Tx, id
return &state, nil
}
func execCronJobUpdateTx(ctx context.Context, tx *sql.Tx, id uuid.UUID, updates map[string]any) error {
if len(updates) == 0 {
return nil
@@ -219,4 +224,3 @@ func execCronJobUpdateTx(ctx context.Context, tx *sql.Tx, id uuid.UUID, updates
}
return nil
}
+7 -2
View File
@@ -131,11 +131,13 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
var intervalMS *int64
var payloadJSON []byte
createdAt, updatedAt := scanTimePair()
var providerID *uuid.UUID
var model *string
err := row.Scan(&id, &tenantID, &agentID, &userID, &name, &enabled, &scheduleKind, &cronExpr, &runAt, &tz,
&intervalMS, &payloadJSON, &deleteAfterRun, &stateless, &deliver, &deliverChannel, &deliverTo, &wakeHeartbeat,
&nextRunAt, &lastRunAt, &lastStatus, &lastError,
createdAt, updatedAt)
createdAt, updatedAt, &providerID, &model)
if err != nil {
return nil, err
}
@@ -164,6 +166,9 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
WakeHeartbeat: wakeHeartbeat,
}
job.ProviderID = providerID
job.Model = model
if agentID != nil {
job.AgentID = agentID.String()
}
@@ -210,7 +215,7 @@ func (s *SQLiteCronStore) scanJob(ctx context.Context, id uuid.UUID) (*store.Cro
q := `SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
interval_ms, payload, delete_after_run, stateless, deliver, deliver_channel, deliver_to, wake_heartbeat,
next_run_at, last_run_at, last_status, last_error,
created_at, updated_at FROM cron_jobs WHERE id = ?`
created_at, updated_at, provider_id, model FROM cron_jobs WHERE id = ?`
args := []any{id}
if !store.IsCrossTenant(ctx) {
+7 -1
View File
@@ -103,7 +103,7 @@ func (s *SQLiteCronStore) ListJobs(ctx context.Context, includeDisabled bool, ag
q := `SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
interval_ms, payload, delete_after_run, stateless, deliver, deliver_channel, deliver_to, wake_heartbeat,
next_run_at, last_run_at, last_status, last_error,
created_at, updated_at FROM cron_jobs WHERE 1=1`
created_at, updated_at, provider_id, model FROM cron_jobs WHERE 1=1`
var args []any
@@ -303,6 +303,12 @@ func (s *SQLiteCronStore) UpdateJob(ctx context.Context, jobID string, patch sto
if patch.WakeHeartbeat != nil {
updates["wake_heartbeat"] = *patch.WakeHeartbeat
}
if patch.ProviderID != nil {
updates["provider_id"] = *patch.ProviderID
}
if patch.Model != nil {
updates["model"] = *patch.Model
}
if patch.Message != "" {
payload := current.Payload
+2 -2
View File
@@ -42,7 +42,7 @@ func (s *SQLiteCronStore) refreshJobCache() {
`SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
interval_ms, payload, delete_after_run, stateless, deliver, deliver_channel, deliver_to, wake_heartbeat,
next_run_at, last_run_at, last_status, last_error,
created_at, updated_at FROM cron_jobs WHERE enabled = 1`)
created_at, updated_at, provider_id, model FROM cron_jobs WHERE enabled = 1`)
if err != nil {
return
}
@@ -366,7 +366,7 @@ func (s *SQLiteCronStore) loadClaimedJob(id uuid.UUID) (*store.CronJob, bool) {
`SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
interval_ms, payload, delete_after_run, stateless, deliver, deliver_channel, deliver_to, wake_heartbeat,
next_run_at, last_run_at, last_status, last_error,
created_at, updated_at
created_at, updated_at, provider_id, model
FROM cron_jobs
WHERE id = ? AND enabled = 1 AND next_run_at IS NULL`,
id,
+4 -1
View File
@@ -16,7 +16,7 @@ var schemaSQL string
// SchemaVersion is the current SQLite schema version.
// Bump this when adding new migration steps below.
const SchemaVersion = 49
const SchemaVersion = 50
// migrations maps version → SQL to apply when upgrading FROM that version.
// schema.sql always represents the LATEST full schema (for fresh DBs).
@@ -30,6 +30,9 @@ const SchemaVersion = 49
//
// Then bump SchemaVersion to 2.
var migrations = map[int]string{
// Version 49 → 50: per-cron-job LLM provider/model override (mirrors agent_heartbeats).
49: `ALTER TABLE cron_jobs ADD COLUMN provider_id TEXT REFERENCES llm_providers(id) ON DELETE SET NULL;
ALTER TABLE cron_jobs ADD COLUMN model VARCHAR(200);`,
// Version 1 → 2: add contact_type column to channel_contacts.
1: `ALTER TABLE channel_contacts ADD COLUMN contact_type VARCHAR(20) NOT NULL DEFAULT 'user';`,
// Version 2 → 3: promote cron payload fields to dedicated columns + add stateless flag.
+2
View File
@@ -470,6 +470,8 @@ CREATE TABLE IF NOT EXISTS cron_jobs (
deliver_channel TEXT NOT NULL DEFAULT '',
deliver_to TEXT NOT NULL DEFAULT '',
wake_heartbeat INTEGER NOT NULL DEFAULT 0,
provider_id TEXT REFERENCES llm_providers(id) ON DELETE SET NULL,
model VARCHAR(200),
next_run_at TEXT,
last_run_at TEXT,
last_status VARCHAR(20),
+70 -5
View File
@@ -6,14 +6,16 @@ import (
"fmt"
"time"
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/store"
)
// 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
cronStore store.CronStore
permStore store.ConfigPermissionStore // nil = no group restriction
providerStore store.ProviderStore // nil = provider override by name unavailable
}
func NewCronTool(cronStore store.CronStore) *CronTool {
@@ -25,6 +27,11 @@ func (t *CronTool) SetConfigPermStore(s store.ConfigPermissionStore) {
t.permStore = s
}
// SetProviderStore enables resolving per-job LLM provider overrides by name.
func (t *CronTool) SetProviderStore(s store.ProviderStore) {
t.providerStore = s
}
func (t *CronTool) Name() string { return "cron" }
func (t *CronTool) Description() string {
@@ -49,6 +56,8 @@ VALID ACTIONS AND EXACT PAYLOAD SHAPES:
"channel": "string", // optional, auto-filled from current channel context
"to": "string", // optional
"agentId": "string", // optional, defaults to current agent
"provider": "string", // optional, LLM provider NAME (e.g. "groq") to run this job on a cheaper model; unset → agent default
"model": "string", // optional, model id for the override provider
"deleteAfterRun": true|false // optional, default true for schedule.kind="at"
}
}
@@ -65,6 +74,8 @@ VALID ACTIONS AND EXACT PAYLOAD SHAPES:
"channel": "string",
"to": "string",
"agentId": "string",
"provider": "string",
"model": "string",
"deleteAfterRun": true|false,
"disabled": true|false
}
@@ -110,7 +121,7 @@ func (t *CronTool) Parameters() map[string]any {
},
"job": map[string]any{
"type": "object",
"description": "Job definition for add action (name, schedule, message, deliver, channel, to, agentId, deleteAfterRun)",
"description": "Job definition for add action (name, schedule, message, deliver, channel, to, agentId, provider, model, deleteAfterRun)",
"additionalProperties": true,
},
"jobId": map[string]any{
@@ -293,15 +304,36 @@ func (t *CronTool) handleAdd(ctx context.Context, args map[string]any, agentID,
agentID = explicit
}
// Resolve optional per-job provider/model override before creating the job.
providerID, errR := t.resolveProviderID(ctx, jobObj)
if errR != nil {
return errR
}
modelOverride, _ := jobObj["model"].(string)
job, err := t.cronStore.AddJob(ctx, name, schedule, message, deliver, channel, to, agentID, userID)
if err != nil {
return ErrorResult(fmt.Sprintf("failed to create cron job: %v", err))
}
// Set wake_heartbeat if requested (triggers heartbeat after cron job completes)
// Apply post-create overrides (wake_heartbeat, provider/model) in a single patch.
overridePatch := store.CronJobPatch{}
needOverride := false
if wh, _ := jobObj["wake_heartbeat"].(bool); wh {
wakeTrue := true
if updated, uErr := t.cronStore.UpdateJob(ctx, job.ID, store.CronJobPatch{WakeHeartbeat: &wakeTrue}); uErr == nil {
overridePatch.WakeHeartbeat = &wakeTrue
needOverride = true
}
if providerID != nil {
overridePatch.ProviderID = providerID
needOverride = true
}
if modelOverride != "" {
overridePatch.Model = &modelOverride
needOverride = true
}
if needOverride {
if updated, uErr := t.cronStore.UpdateJob(ctx, job.ID, overridePatch); uErr == nil {
job = updated
}
}
@@ -353,6 +385,15 @@ func (t *CronTool) handleUpdate(ctx context.Context, args map[string]any, agentI
patchJSON, _ := json.Marshal(patchObj)
json.Unmarshal(patchJSON, &patch)
// 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)
if errR != nil {
return errR
}
patch.ProviderID = pid
}
// Validate atMs not in the past when updating schedule
if patch.Schedule != nil && patch.Schedule.Kind == "at" && patch.Schedule.AtMS != nil {
if *patch.Schedule.AtMS <= time.Now().UnixMilli() {
@@ -447,6 +488,30 @@ func (t *CronTool) handleRuns(ctx context.Context, args map[string]any, agentID,
return NewResult(string(data))
}
// resolveProviderID resolves an optional per-job provider override from a job/patch
// map. Accepts "providerId" (UUID string) or "provider" (provider name, looked up
// via the provider store). Returns (nil, nil) when neither is present.
func (t *CronTool) resolveProviderID(ctx context.Context, m map[string]any) (*uuid.UUID, *Result) {
if raw, _ := m["providerId"].(string); raw != "" {
id, err := uuid.Parse(raw)
if err != nil {
return nil, ErrorResult(fmt.Sprintf("invalid providerId %q: must be a UUID", raw))
}
return &id, nil
}
if name, _ := m["provider"].(string); name != "" {
if t.providerStore == nil {
return nil, ErrorResult("provider override by name is unavailable; pass providerId instead")
}
pd, err := t.providerStore.GetProviderByName(ctx, name)
if err != nil || pd == nil {
return nil, ErrorResult(fmt.Sprintf("provider %q not found; register it first or pass a valid providerId", name))
}
return &pd.ID, nil
}
return nil, nil
}
// --- helpers ---
func resolveJobID(args map[string]any) string {
+1 -1
View File
@@ -2,4 +2,4 @@ package upgrade
// RequiredSchemaVersion is the schema migration version this binary requires.
// Bump this whenever adding a new SQL migration file.
const RequiredSchemaVersion uint = 80
const RequiredSchemaVersion uint = 81
@@ -0,0 +1,2 @@
ALTER TABLE cron_jobs DROP COLUMN IF EXISTS provider_id;
ALTER TABLE cron_jobs DROP COLUMN IF EXISTS model;
@@ -0,0 +1,6 @@
-- 000081: per-cron-job LLM provider/model override.
-- Mirrors agent_heartbeats.provider_id/model (see 000022) so scheduled jobs can run
-- on a cheaper provider than the agent default. NULL → falls back to agent default.
ALTER TABLE cron_jobs
ADD COLUMN IF NOT EXISTS provider_id UUID REFERENCES llm_providers(id) ON DELETE SET NULL,
ADD COLUMN IF NOT EXISTS model VARCHAR(200);