refactor(cron): normalize payload columns into dedicated DB fields (#33)

Extract wake_heartbeat and stateless from JSON payload into first-class
columns on cron_jobs. Adds migration 000033 with backfill from existing
payload data. Updates PG + SQLite stores, RPC handlers, and UI i18n.
This commit is contained in:
viettranx committed 2026-03-30 15:19:43 +07:00
1 parent d10236241e
commit 24717b0f51
28 files changed
+268 -131

No files matched your search

+16 -13
View File
@@ -49,7 +49,7 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg
}
sessionKey := sessions.BuildCronSessionKey(agentID, job.ID)
channel := job.Payload.Channel
channel := job.DeliverChannel
if channel == "" {
channel = "cron"
}
@@ -63,12 +63,12 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg
// Build cron context so the agent knows delivery target and requester.
var extraPrompt string
if job.Payload.Deliver && job.Payload.Channel != "" && job.Payload.To != "" {
if job.Deliver && job.DeliverChannel != "" && job.DeliverTo != "" {
extraPrompt = fmt.Sprintf(
"[Cron Job]\nThis is scheduled job \"%s\" (ID: %s).\n"+
"Requester: user %s on channel \"%s\" (chat %s).\n"+
"Your response will be automatically delivered to that chat — just produce the content directly.",
job.Name, job.ID, job.UserID, job.Payload.Channel, job.Payload.To,
job.Name, job.ID, job.UserID, job.DeliverChannel, job.DeliverTo,
)
} else {
extraPrompt = fmt.Sprintf(
@@ -84,8 +84,11 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg
// Reset session before each cron run to prevent tool errors from previous
// runs from polluting the context and blocking future executions (#294).
// Save() persists the empty session to DB so stale data won't reload after restart.
sessionMgr.Reset(cronCtx, sessionKey)
sessionMgr.Save(cronCtx, sessionKey)
// Stateless jobs skip this — they intentionally carry no session history.
if !job.Stateless {
sessionMgr.Reset(cronCtx, sessionKey)
sessionMgr.Save(cronCtx, sessionKey)
}
// Schedule through cron lane — scheduler handles agent resolution and concurrency
outCh := sched.Schedule(cronCtx, scheduler.LaneCron, agent.RunRequest{
@@ -93,7 +96,7 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg
Message: job.Payload.Message,
Channel: channel,
ChannelType: channelType,
ChatID: job.Payload.To,
ChatID: job.DeliverTo,
PeerKind: peerKind,
UserID: job.UserID,
RunID: fmt.Sprintf("cron:%s", job.ID),
@@ -112,20 +115,20 @@ 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.Payload.Deliver && job.Payload.Channel != "" && job.Payload.To != "" {
if job.Deliver && job.DeliverChannel != "" && job.DeliverTo != "" {
outMsg := bus.OutboundMessage{
Channel: job.Payload.Channel,
ChatID: job.Payload.To,
Channel: job.DeliverChannel,
ChatID: job.DeliverTo,
Content: result.Content,
}
if peerKind == "group" {
outMsg.Metadata = map[string]string{"group_id": job.Payload.To}
outMsg.Metadata = map[string]string{"group_id": job.DeliverTo}
}
appendMediaToOutbound(&outMsg, result.Media)
msgBus.PublishOutbound(outMsg)
} else if job.Payload.Deliver {
} 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.Payload.Channel, "to", job.Payload.To)
"job_id", job.ID, "job_name", job.Name, "channel", job.DeliverChannel, "to", job.DeliverTo)
}
cronResult := &store.CronJobResult{
@@ -138,7 +141,7 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg
// wakeMode: trigger heartbeat after cron job completes.
// Use original job.AgentID (UUID) — cronHeartbeatWakeFn expects UUID for ticker.Wake().
if job.Payload.WakeHeartbeat && cronHeartbeatWakeFn != nil {
if job.WakeHeartbeat && cronHeartbeatWakeFn != nil {
cronHeartbeatWakeFn(job.AgentID)
}
+14 -8
View File
@@ -114,10 +114,10 @@ func (cs *Service) AddJob(name string, schedule Schedule, message string, delive
Payload: Payload{
Kind: "agent_turn",
Message: message,
Deliver: deliver,
Channel: channel,
To: to,
},
Deliver: deliver,
DeliverChannel: channel,
DeliverTo: to,
CreatedAtMS: now,
UpdatedAtMS: now,
DeleteAfterRun: schedule.Kind == "at",
@@ -236,13 +236,19 @@ func (cs *Service) UpdateJob(jobID string, patch JobPatch) (*Job, error) {
job.Payload.Message = patch.Message
}
if patch.Deliver != nil {
job.Payload.Deliver = *patch.Deliver
job.Deliver = *patch.Deliver
}
if patch.Channel != nil {
job.Payload.Channel = *patch.Channel
if patch.DeliverChannel != nil {
job.DeliverChannel = *patch.DeliverChannel
}
if patch.To != nil {
job.Payload.To = *patch.To
if patch.DeliverTo != nil {
job.DeliverTo = *patch.DeliverTo
}
if patch.WakeHeartbeat != nil {
job.WakeHeartbeat = *patch.WakeHeartbeat
}
if patch.Stateless != nil {
job.Stateless = *patch.Stateless
}
if patch.DeleteAfterRun != nil {
job.DeleteAfterRun = *patch.DeleteAfterRun
+10 -6
View File
@@ -27,9 +27,6 @@ type Payload struct {
Kind string `json:"kind"` // "agent_turn"
Message string `json:"message"` // content to process
Command string `json:"command,omitempty"` // optional shell command
Deliver bool `json:"deliver"` // true = direct chat, false = agent processing
Channel string `json:"channel,omitempty"` // target channel (telegram, discord, etc.)
To string `json:"to,omitempty"` // target chat ID / recipient
}
// JobState tracks runtime state for a job.
@@ -52,6 +49,11 @@ type Job struct {
CreatedAtMS int64 `json:"createdAtMs"`
UpdatedAtMS int64 `json:"updatedAtMs"`
DeleteAfterRun bool `json:"deleteAfterRun,omitempty"`
Stateless bool `json:"stateless"`
Deliver bool `json:"deliver"`
DeliverChannel string `json:"deliverChannel"`
DeliverTo string `json:"deliverTo"`
WakeHeartbeat bool `json:"wakeHeartbeat"`
}
// Store is the persistent store for all cron jobs.
@@ -68,10 +70,12 @@ type JobPatch struct {
Enabled *bool `json:"enabled,omitempty"`
Schedule *Schedule `json:"schedule,omitempty"`
Message string `json:"message,omitempty"`
Deliver *bool `json:"deliver,omitempty"`
Channel *string `json:"channel,omitempty"`
To *string `json:"to,omitempty"`
DeleteAfterRun *bool `json:"deleteAfterRun,omitempty"`
Stateless *bool `json:"stateless,omitempty"`
Deliver *bool `json:"deliver,omitempty"`
DeliverChannel *string `json:"deliverChannel,omitempty"`
DeliverTo *string `json:"deliverTo,omitempty"`
WakeHeartbeat *bool `json:"wakeHeartbeat,omitempty"`
}
// RunLogEntry is an in-memory record of a job execution.
+24 -8
View File
@@ -62,13 +62,15 @@ 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"`
Channel string `json:"channel"`
To string `json:"to"`
AgentID string `json:"agentId"`
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"`
AgentID string `json:"agentId"`
}
if req.Params != nil {
json.Unmarshal(req.Params, &params)
@@ -87,12 +89,26 @@ func (m *CronMethods) handleCreate(ctx context.Context, client *gateway.Client,
return
}
job, err := m.service.AddJob(ctx, params.Name, params.Schedule, params.Message, params.Deliver, params.Channel, params.To, params.AgentID, client.UserID())
job, err := m.service.AddJob(ctx, params.Name, params.Schedule, params.Message, params.Deliver, params.DeliverChannel, params.DeliverTo, params.AgentID, client.UserID())
if err != nil {
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, err.Error()))
return
}
// Apply extra fields not in AddJob signature via an immediate patch.
if params.WakeHeartbeat || params.Stateless {
patch := store.CronJobPatch{}
if params.WakeHeartbeat {
patch.WakeHeartbeat = &params.WakeHeartbeat
}
if params.Stateless {
patch.Stateless = &params.Stateless
}
if updated, pErr := m.service.UpdateJob(ctx, job.ID, patch); pErr == nil {
job = updated
}
}
client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{
"job": job,
}))
+12 -10
View File
@@ -30,6 +30,11 @@ type CronJob struct {
CreatedAtMS int64 `json:"createdAtMs"`
UpdatedAtMS int64 `json:"updatedAtMs"`
DeleteAfterRun bool `json:"deleteAfterRun,omitempty"`
Stateless bool `json:"stateless"`
Deliver bool `json:"deliver"`
DeliverChannel string `json:"deliverChannel"`
DeliverTo string `json:"deliverTo"`
WakeHeartbeat bool `json:"wakeHeartbeat"`
}
// CronSchedule defines when a job should run.
@@ -43,13 +48,9 @@ type CronSchedule struct {
// CronPayload describes what a job does when triggered.
type CronPayload struct {
Kind string `json:"kind"`
Message string `json:"message"`
Command string `json:"command,omitempty"`
Deliver bool `json:"deliver"`
Channel string `json:"channel,omitempty"`
To string `json:"to,omitempty"`
WakeHeartbeat bool `json:"wake_heartbeat,omitempty"` // trigger heartbeat after job completes
Kind string `json:"kind"`
Message string `json:"message"`
Command string `json:"command,omitempty"`
}
// CronJobState tracks runtime state for a job.
@@ -87,10 +88,11 @@ type CronJobPatch struct {
Enabled *bool `json:"enabled,omitempty"`
Schedule *CronSchedule `json:"schedule,omitempty"`
Message string `json:"message,omitempty"`
Deliver *bool `json:"deliver,omitempty"`
Channel *string `json:"channel,omitempty"`
To *string `json:"to,omitempty"`
DeleteAfterRun *bool `json:"deleteAfterRun,omitempty"`
Stateless *bool `json:"stateless,omitempty"`
Deliver *bool `json:"deliver,omitempty"`
DeliverChannel *string `json:"deliverChannel,omitempty"`
DeliverTo *string `json:"deliverTo,omitempty"`
WakeHeartbeat *bool `json:"wakeHeartbeat,omitempty"`
}
+6 -5
View File
@@ -24,7 +24,7 @@ func (s *PGCronStore) AddJob(ctx context.Context, name string, schedule store.Cr
}
payload := store.CronPayload{
Kind: "agent_turn", Message: message, Deliver: deliver, Channel: channel, To: to,
Kind: "agent_turn", Message: message,
}
payloadJSON, _ := json.Marshal(payload)
@@ -68,10 +68,10 @@ func (s *PGCronStore) AddJob(ctx context.Context, name string, schedule store.Cr
_, err := s.db.ExecContext(ctx,
`INSERT INTO cron_jobs (id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
interval_ms, payload, delete_after_run, next_run_at, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, true, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)`,
interval_ms, payload, delete_after_run, deliver, deliver_channel, deliver_to, wake_heartbeat, next_run_at, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, true, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)`,
id, tenantIDForInsert(ctx), agentUUID, userIDPtr, name, scheduleKind, cronExpr, runAt, tz,
intervalMS, payloadJSON, deleteAfterRun, nextRun, now, now,
intervalMS, payloadJSON, deleteAfterRun, deliver, channel, to, false, nextRun, now, now,
)
if err != nil {
return nil, fmt.Errorf("create cron job: %w", err)
@@ -97,7 +97,8 @@ func (s *PGCronStore) GetJob(ctx context.Context, jobID string) (*store.CronJob,
func (s *PGCronStore) ListJobs(ctx context.Context, includeDisabled bool, agentID, userID string) []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, next_run_at, last_run_at, last_status, last_error,
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`
var args []any
+11 -2
View File
@@ -15,7 +15,8 @@ import (
// scanJob fetches a single cron job by ID with tenant filtering.
func (s *PGCronStore) scanJob(ctx context.Context, id uuid.UUID) (*store.CronJob, error) {
q := `SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
interval_ms, payload, delete_after_run, next_run_at, last_run_at, last_status, last_error,
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`
args := []any{id}
@@ -45,6 +46,8 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
var userID *string
var name, scheduleKind string
var enabled, deleteAfterRun bool
var stateless, deliver, wakeHeartbeat bool
var deliverChannel, deliverTo string
var cronExpr, tz, lastStatus, lastError *string
var runAt, nextRunAt, lastRunAt *time.Time
var intervalMS *int64
@@ -52,7 +55,8 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
var createdAt, updatedAt time.Time
err := row.Scan(&id, &tenantID, &agentID, &userID, &name, &enabled, &scheduleKind, &cronExpr, &runAt, &tz,
&intervalMS, &payloadJSON, &deleteAfterRun, &nextRunAt, &lastRunAt, &lastStatus, &lastError,
&intervalMS, &payloadJSON, &deleteAfterRun, &stateless, &deliver, &deliverChannel, &deliverTo, &wakeHeartbeat,
&nextRunAt, &lastRunAt, &lastStatus, &lastError,
&createdAt, &updatedAt)
if err != nil {
return nil, err
@@ -77,6 +81,11 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
CreatedAtMS: createdAt.UnixMilli(),
UpdatedAtMS: updatedAt.UnixMilli(),
DeleteAfterRun: deleteAfterRun,
Stateless: stateless,
Deliver: deliver,
DeliverChannel: deliverChannel,
DeliverTo: deliverTo,
WakeHeartbeat: wakeHeartbeat,
}
if agentID != nil {
+20 -8
View File
@@ -91,6 +91,11 @@ func validCronRow(payloadJSON []byte) []any {
(*int64)(nil), // interval_ms
payloadJSON, // payload
false, // delete_after_run
false, // stateless
false, // deliver
"", // deliver_channel
"", // deliver_to
false, // wake_heartbeat
(*time.Time)(nil), // next_run_at
(*time.Time)(nil), // last_run_at
(*string)(nil), // last_status
@@ -120,12 +125,16 @@ func TestScanCronRow_ValidPayload(t *testing.T) {
payload := store.CronPayload{
Kind: "message",
Message: "hello world",
Deliver: true,
Channel: "telegram",
To: "user123",
}
payloadJSON, _ := json.Marshal(payload)
row := &mockRowScanner{values: validCronRow(payloadJSON)}
// Build row with deliver/deliver_channel/deliver_to set as dedicated columns.
rowVals := validCronRow(payloadJSON)
// Indices: stateless=13, deliver=14, deliver_channel=15, deliver_to=16, wake_heartbeat=17
rowVals[14] = true // deliver
rowVals[15] = "telegram" // deliver_channel
rowVals[16] = "user123" // deliver_to
row := &mockRowScanner{values: rowVals}
job, err := scanCronRow(row)
if err != nil {
@@ -135,11 +144,14 @@ func TestScanCronRow_ValidPayload(t *testing.T) {
if job.Payload.Message != "hello world" {
t.Errorf("expected Message 'hello world', got %q", job.Payload.Message)
}
if job.Payload.Channel != "telegram" {
t.Errorf("expected Channel 'telegram', got %q", job.Payload.Channel)
if job.DeliverChannel != "telegram" {
t.Errorf("expected DeliverChannel 'telegram', got %q", job.DeliverChannel)
}
if job.Payload.To != "user123" {
t.Errorf("expected To 'user123', got %q", job.Payload.To)
if job.DeliverTo != "user123" {
t.Errorf("expected DeliverTo 'user123', got %q", job.DeliverTo)
}
if !job.Deliver {
t.Errorf("expected Deliver true")
}
}
+4 -2
View File
@@ -37,7 +37,8 @@ func (s *PGCronStore) GetDueJobs(now time.Time) []store.CronJob {
func (s *PGCronStore) refreshJobCache() {
rows, err := s.db.QueryContext(s.baseCtx,
`SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
interval_ms, payload, delete_after_run, next_run_at, last_run_at, last_status, last_error,
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`)
if err != nil {
return
@@ -296,7 +297,8 @@ func (s *PGCronStore) loadClaimedJob(id uuid.UUID) (*store.CronJob, bool) {
row := s.db.QueryRowContext(
s.baseCtx,
`SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
interval_ms, payload, delete_after_run, next_run_at, last_run_at, last_status, last_error,
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 AND enabled = true AND next_run_at IS NULL`,
+18 -18
View File
@@ -74,25 +74,25 @@ func (s *PGCronStore) UpdateJob(ctx context.Context, jobID string, patch store.C
updates["next_run_at"] = nextRun
}
needsPayloadUpdate := patch.Message != "" || patch.Deliver != nil || patch.Channel != nil || patch.To != nil || patch.WakeHeartbeat != nil
if needsPayloadUpdate {
payload := current.Payload
if patch.Message != "" {
payload.Message = patch.Message
}
if patch.Deliver != nil {
payload.Deliver = *patch.Deliver
}
if patch.Channel != nil {
payload.Channel = *patch.Channel
}
if patch.To != nil {
payload.To = *patch.To
}
if patch.WakeHeartbeat != nil {
payload.WakeHeartbeat = *patch.WakeHeartbeat
}
if patch.Stateless != nil {
updates["stateless"] = *patch.Stateless
}
if patch.Deliver != nil {
updates["deliver"] = *patch.Deliver
}
if patch.DeliverChannel != nil {
updates["deliver_channel"] = *patch.DeliverChannel
}
if patch.DeliverTo != nil {
updates["deliver_to"] = *patch.DeliverTo
}
if patch.WakeHeartbeat != nil {
updates["wake_heartbeat"] = *patch.WakeHeartbeat
}
if patch.Message != "" {
payload := current.Payload
payload.Message = patch.Message
mergedPayload, err := json.Marshal(payload)
if err != nil {
return nil, fmt.Errorf("failed to marshal payload for job %s: %w", jobID, err)
+2 -13
View File
@@ -52,7 +52,7 @@ func TestScheduleKindCorruption_LogicPath(t *testing.T) {
// The fix: UpdateJob now returns an error when payload Unmarshal fails,
// so existing fields are never overwritten with zero values.
func TestPayloadUnmarshal_CorruptedJSON(t *testing.T) {
corruptedJSON := []byte(`{"kind":"message","message":"daily report","deliver":true,CORRUPT`)
corruptedJSON := []byte(`{"kind":"message","message":"daily report",CORRUPT`)
var payload store.CronPayload
err := json.Unmarshal(corruptedJSON, &payload)
@@ -66,9 +66,6 @@ func TestPayloadUnmarshal_CorruptedJSON(t *testing.T) {
if payload.Message != "" {
t.Error("expected empty Message after failed Unmarshal")
}
if payload.Deliver {
t.Error("expected false Deliver after failed Unmarshal")
}
// If we ignored this error and applied patch fields, we'd wipe existing data.
// The fix returns an error instead.
@@ -76,13 +73,11 @@ func TestPayloadUnmarshal_CorruptedJSON(t *testing.T) {
}
// TestPayloadUnmarshal_ValidJSON verifies normal payload round-trip works.
// Deliver/DeliverChannel/DeliverTo are now dedicated columns on CronJob, not payload fields.
func TestPayloadUnmarshal_ValidJSON(t *testing.T) {
original := store.CronPayload{
Kind: "message",
Message: "daily report",
Deliver: true,
Channel: "telegram",
To: "admin-group",
}
data, _ := json.Marshal(original)
@@ -94,10 +89,4 @@ func TestPayloadUnmarshal_ValidJSON(t *testing.T) {
if decoded.Message != original.Message {
t.Errorf("Message mismatch: got %q, want %q", decoded.Message, original.Message)
}
if decoded.Channel != original.Channel {
t.Errorf("Channel mismatch: got %q, want %q", decoded.Channel, original.Channel)
}
if decoded.To != original.To {
t.Errorf("To mismatch: got %q, want %q", decoded.To, original.To)
}
}
+11 -2
View File
@@ -118,6 +118,8 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
var userID *string
var name, scheduleKind string
var enabled, deleteAfterRun bool
var stateless, deliver, wakeHeartbeat bool
var deliverChannel, deliverTo string
var cronExpr, tz, lastStatus, lastError *string
var runAt, nextRunAt, lastRunAt nullSqliteTime
var intervalMS *int64
@@ -125,7 +127,8 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
createdAt, updatedAt := scanTimePair()
err := row.Scan(&id, &tenantID, &agentID, &userID, &name, &enabled, &scheduleKind, &cronExpr, &runAt, &tz,
&intervalMS, &payloadJSON, &deleteAfterRun, &nextRunAt, &lastRunAt, &lastStatus, &lastError,
&intervalMS, &payloadJSON, &deleteAfterRun, &stateless, &deliver, &deliverChannel, &deliverTo, &wakeHeartbeat,
&nextRunAt, &lastRunAt, &lastStatus, &lastError,
createdAt, updatedAt)
if err != nil {
return nil, err
@@ -148,6 +151,11 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
CreatedAtMS: createdAt.Time.UnixMilli(),
UpdatedAtMS: updatedAt.Time.UnixMilli(),
DeleteAfterRun: deleteAfterRun,
Stateless: stateless,
Deliver: deliver,
DeliverChannel: deliverChannel,
DeliverTo: deliverTo,
WakeHeartbeat: wakeHeartbeat,
}
if agentID != nil {
@@ -194,7 +202,8 @@ func computeNextRun(schedule *store.CronSchedule, now time.Time, defaultTZ strin
func (s *SQLiteCronStore) scanJob(ctx context.Context, id uuid.UUID) (*store.CronJob, error) {
q := `SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
interval_ms, payload, delete_after_run, next_run_at, last_run_at, last_status, last_error,
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 = ?`
args := []any{id}
+24 -23
View File
@@ -28,7 +28,7 @@ func (s *SQLiteCronStore) AddJob(ctx context.Context, name string, schedule stor
}
payload := store.CronPayload{
Kind: "agent_turn", Message: message, Deliver: deliver, Channel: channel, To: to,
Kind: "agent_turn", Message: message,
}
payloadJSON, _ := json.Marshal(payload)
@@ -71,10 +71,10 @@ func (s *SQLiteCronStore) AddJob(ctx context.Context, name string, schedule stor
_, err := s.db.ExecContext(ctx,
`INSERT INTO cron_jobs (id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
interval_ms, payload, delete_after_run, next_run_at, created_at, updated_at)
VALUES (?,?,?,?,?,1,?,?,?,?,?,?,?,?,?,?)`,
interval_ms, payload, delete_after_run, deliver, deliver_channel, deliver_to, wake_heartbeat, next_run_at, created_at, updated_at)
VALUES (?,?,?,?,?,1,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
id, tenantIDForInsert(ctx), agentUUID, userIDPtr, name, scheduleKind, cronExpr, runAt, tz,
intervalMS, payloadJSON, deleteAfterRun, nextRun, now, now,
intervalMS, payloadJSON, deleteAfterRun, deliver, channel, to, false, nextRun, now, now,
)
if err != nil {
return nil, fmt.Errorf("create cron job: %w", err)
@@ -99,7 +99,8 @@ func (s *SQLiteCronStore) GetJob(ctx context.Context, jobID string) (*store.Cron
func (s *SQLiteCronStore) ListJobs(ctx context.Context, includeDisabled bool, agentID, userID string) []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, next_run_at, last_run_at, last_status, last_error,
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`
var args []any
@@ -285,25 +286,25 @@ func (s *SQLiteCronStore) UpdateJob(ctx context.Context, jobID string, patch sto
updates["next_run_at"] = nextRun
}
needsPayloadUpdate := patch.Message != "" || patch.Deliver != nil || patch.Channel != nil || patch.To != nil || patch.WakeHeartbeat != nil
if needsPayloadUpdate {
payload := current.Payload
if patch.Message != "" {
payload.Message = patch.Message
}
if patch.Deliver != nil {
payload.Deliver = *patch.Deliver
}
if patch.Channel != nil {
payload.Channel = *patch.Channel
}
if patch.To != nil {
payload.To = *patch.To
}
if patch.WakeHeartbeat != nil {
payload.WakeHeartbeat = *patch.WakeHeartbeat
}
if patch.Stateless != nil {
updates["stateless"] = *patch.Stateless
}
if patch.Deliver != nil {
updates["deliver"] = *patch.Deliver
}
if patch.DeliverChannel != nil {
updates["deliver_channel"] = *patch.DeliverChannel
}
if patch.DeliverTo != nil {
updates["deliver_to"] = *patch.DeliverTo
}
if patch.WakeHeartbeat != nil {
updates["wake_heartbeat"] = *patch.WakeHeartbeat
}
if patch.Message != "" {
payload := current.Payload
payload.Message = patch.Message
mergedPayload, err := json.Marshal(payload)
if err != nil {
return nil, fmt.Errorf("failed to marshal payload for job %s: %w", jobID, err)
+4 -2
View File
@@ -39,7 +39,8 @@ func (s *SQLiteCronStore) GetDueJobs(now time.Time) []store.CronJob {
func (s *SQLiteCronStore) refreshJobCache() {
rows, err := s.db.QueryContext(s.baseCtx,
`SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
interval_ms, payload, delete_after_run, next_run_at, last_run_at, last_status, last_error,
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`)
if err != nil {
return
@@ -296,7 +297,8 @@ func (s *SQLiteCronStore) loadClaimedJob(id uuid.UUID) (*store.CronJob, bool) {
row := s.db.QueryRowContext(
s.baseCtx,
`SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
interval_ms, payload, delete_after_run, next_run_at, last_run_at, last_status, last_error,
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 = ? AND enabled = 1 AND next_run_at IS NULL`,
+13 -1
View File
@@ -14,7 +14,7 @@ var schemaSQL string
// SchemaVersion is the current SQLite schema version.
// Bump this when adding new migration steps below.
const SchemaVersion = 2
const SchemaVersion = 3
// 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,18 @@ const SchemaVersion = 2
var migrations = map[int]string{
// 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: `ALTER TABLE cron_jobs ADD COLUMN stateless INTEGER NOT NULL DEFAULT 0;
ALTER TABLE cron_jobs ADD COLUMN deliver INTEGER NOT NULL DEFAULT 0;
ALTER TABLE cron_jobs ADD COLUMN deliver_channel TEXT NOT NULL DEFAULT '';
ALTER TABLE cron_jobs ADD COLUMN deliver_to TEXT NOT NULL DEFAULT '';
ALTER TABLE cron_jobs ADD COLUMN wake_heartbeat INTEGER NOT NULL DEFAULT 0;
UPDATE cron_jobs SET
deliver = COALESCE(json_extract(payload, '$.deliver'), 0),
deliver_channel = COALESCE(json_extract(payload, '$.channel'), ''),
deliver_to = COALESCE(json_extract(payload, '$.to'), ''),
wake_heartbeat = COALESCE(json_extract(payload, '$.wake_heartbeat'), 0)
WHERE payload IS NOT NULL;`,
}
// EnsureSchema creates tables if they don't exist and applies incremental migrations.
+5
View File
@@ -449,6 +449,11 @@ CREATE TABLE IF NOT EXISTS cron_jobs (
timezone VARCHAR(50),
payload TEXT NOT NULL,
delete_after_run BOOLEAN NOT NULL DEFAULT 0,
stateless INTEGER NOT NULL DEFAULT 0,
deliver INTEGER NOT NULL DEFAULT 0,
deliver_channel TEXT NOT NULL DEFAULT '',
deliver_to TEXT NOT NULL DEFAULT '',
wake_heartbeat INTEGER NOT NULL DEFAULT 0,
next_run_at TEXT,
last_run_at TEXT,
last_status VARCHAR(20),
+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 = 32
const RequiredSchemaVersion uint = 33
@@ -0,0 +1,16 @@
UPDATE cron_jobs SET payload = jsonb_set(
jsonb_set(
jsonb_set(
jsonb_set(payload, '{deliver}', to_jsonb(deliver)),
'{channel}', to_jsonb(deliver_channel)
),
'{to}', to_jsonb(deliver_to)
),
'{wake_heartbeat}', to_jsonb(wake_heartbeat)
);
ALTER TABLE cron_jobs DROP COLUMN stateless;
ALTER TABLE cron_jobs DROP COLUMN deliver;
ALTER TABLE cron_jobs DROP COLUMN deliver_channel;
ALTER TABLE cron_jobs DROP COLUMN deliver_to;
ALTER TABLE cron_jobs DROP COLUMN wake_heartbeat;
@@ -0,0 +1,15 @@
ALTER TABLE cron_jobs ADD COLUMN stateless BOOLEAN NOT NULL DEFAULT false;
ALTER TABLE cron_jobs ADD COLUMN deliver BOOLEAN NOT NULL DEFAULT false;
ALTER TABLE cron_jobs ADD COLUMN deliver_channel TEXT NOT NULL DEFAULT '';
ALTER TABLE cron_jobs ADD COLUMN deliver_to TEXT NOT NULL DEFAULT '';
ALTER TABLE cron_jobs ADD COLUMN wake_heartbeat BOOLEAN NOT NULL DEFAULT false;
UPDATE cron_jobs SET
deliver = COALESCE((payload->>'deliver')::boolean, false),
deliver_channel = COALESCE(payload->>'channel', ''),
deliver_to = COALESCE(payload->>'to', ''),
wake_heartbeat = COALESCE((payload->>'wake_heartbeat')::boolean, false)
WHERE payload IS NOT NULL;
UPDATE cron_jobs SET payload = payload - 'deliver' - 'channel' - 'to' - 'wake_heartbeat'
WHERE payload IS NOT NULL;
@@ -130,6 +130,8 @@
"timezone": "Timezone",
"timezoneDesc": "Timezone for cron expressions"
},
"stateless": "Stateless",
"statelessHelp": "Skip session persistence. Each run starts fresh without loading previous messages. Saves tokens for crons that don't need past context.",
"toast": {
"created": "Cron job created",
"createdDesc": "{{name}} has been added",
@@ -130,6 +130,8 @@
"timezone": "Múi giờ",
"timezoneDesc": "Múi giờ cho biểu thức cron"
},
"stateless": "Không trạng thái",
"statelessHelp": "Bỏ qua lưu session. Mỗi lần chạy bắt đầu mới. Tiết kiệm token cho các cron không cần context cũ.",
"toast": {
"created": "Đã tạo tác vụ định kỳ",
"createdDesc": "{{name}} đã được thêm",
@@ -130,6 +130,8 @@
"timezone": "时区",
"timezoneDesc": "Cron表达式的时区"
},
"stateless": "无状态",
"statelessHelp": "跳过会话持久化。每次运行从零开始。为不需要历史上下文的定时任务节省令牌。",
"toast": {
"created": "定时任务已创建",
"createdDesc": "{{name}} 已添加",
+6
View File
@@ -6,6 +6,11 @@ export interface CronJob {
schedule: CronSchedule
payload: CronPayload
state: CronState
deliver?: boolean
deliverChannel?: string
deliverTo?: string
wakeHeartbeat?: boolean
stateless?: boolean
deleteAfterRun: boolean
createdAtMs: number
updatedAtMs: number
@@ -22,6 +27,7 @@ export interface CronSchedule {
export interface CronPayload {
kind: string
message: string
command?: string
}
export interface CronState {
+2
View File
@@ -130,6 +130,8 @@
"timezone": "Timezone",
"timezoneDesc": "Timezone for cron expressions"
},
"stateless": "Stateless",
"statelessHelp": "Skip session persistence. Each run starts fresh without loading previous messages. Saves tokens for crons that don't need past context.",
"toast": {
"created": "Cron job created",
"createdDesc": "{{name}} has been added",
+2
View File
@@ -130,6 +130,8 @@
"timezone": "Múi giờ",
"timezoneDesc": "Múi giờ cho biểu thức cron"
},
"stateless": "Không trạng thái",
"statelessHelp": "Bỏ qua lưu session. Mỗi lần chạy bắt đầu mới. Tiết kiệm token cho các cron không cần context cũ.",
"toast": {
"created": "Đã tạo tác vụ định kỳ",
"createdDesc": "{{name}} đã được thêm",
+2
View File
@@ -130,6 +130,8 @@
"timezone": "时区",
"timezoneDesc": "Cron表达式的时区"
},
"stateless": "无状态",
"statelessHelp": "跳过会话持久化。每次运行从零开始。为不需要历史上下文的定时任务节省令牌。",
"toast": {
"created": "定时任务已创建",
"createdDesc": "{{name}} 已添加",
@@ -21,11 +21,12 @@ interface CronAdvancedDialogProps {
function deriveState(job: CronJob) {
return {
timezone: job.schedule.tz ?? "UTC",
deliver: job.payload?.deliver ?? false,
channel: job.payload?.channel ?? "",
to: job.payload?.to ?? "",
wakeHeartbeat: Boolean((job as unknown as Record<string, unknown>).wakeHeartbeat),
deliver: job.deliver ?? false,
channel: job.deliverChannel ?? "",
to: job.deliverTo ?? "",
wakeHeartbeat: job.wakeHeartbeat ?? false,
deleteAfterRun: job.deleteAfterRun ?? false,
stateless: job.stateless ?? false,
};
}
@@ -40,6 +41,7 @@ export function CronAdvancedDialog({ open, onOpenChange, job, onUpdate }: CronAd
const [to, setTo] = useState(init.to);
const [wakeHeartbeat, setWakeHeartbeat] = useState(init.wakeHeartbeat);
const [deleteAfterRun, setDeleteAfterRun] = useState(init.deleteAfterRun);
const [stateless, setStateless] = useState(init.stateless);
const [saving, setSaving] = useState(false);
// Re-sync when dialog opens
@@ -52,6 +54,7 @@ export function CronAdvancedDialog({ open, onOpenChange, job, onUpdate }: CronAd
setTo(s.to);
setWakeHeartbeat(s.wakeHeartbeat);
setDeleteAfterRun(s.deleteAfterRun);
setStateless(s.stateless);
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [open]);
@@ -68,10 +71,11 @@ export function CronAdvancedDialog({ open, onOpenChange, job, onUpdate }: CronAd
tz: timezone !== "UTC" ? timezone : undefined,
},
deliver,
channel: deliver ? channel.trim() || undefined : undefined,
to: deliver ? to.trim() || undefined : undefined,
deliverChannel: deliver ? channel.trim() || undefined : undefined,
deliverTo: deliver ? to.trim() || undefined : undefined,
wakeHeartbeat,
deleteAfterRun,
stateless,
});
onOpenChange(false);
} catch { // toast shown by hook
@@ -173,6 +177,14 @@ export function CronAdvancedDialog({ open, onOpenChange, job, onUpdate }: CronAd
<Switch checked={deleteAfterRun} onCheckedChange={setDeleteAfterRun} />
</div>
<div className="flex items-center justify-between gap-4 rounded-md border px-3 py-2.5">
<div>
<p className="text-sm font-medium">{t("stateless")}</p>
<p className="text-xs text-muted-foreground">{t("statelessHelp")}</p>
</div>
<Switch checked={stateless} onCheckedChange={setStateless} />
</div>
</div>
{/* Footer */}
+6 -3
View File
@@ -18,9 +18,7 @@ export interface CronSchedule {
export interface CronPayload {
kind: string;
message: string;
deliver: boolean;
channel: string;
to: string;
command?: string;
}
export interface CronJob {
@@ -30,6 +28,11 @@ export interface CronJob {
enabled: boolean;
schedule: CronSchedule;
payload: CronPayload;
deliver?: boolean;
deliverChannel?: string;
deliverTo?: string;
wakeHeartbeat?: boolean;
stateless?: boolean;
createdAtMs: number;
updatedAtMs: number;
deleteAfterRun?: boolean;