diff --git a/cmd/gateway_cron.go b/cmd/gateway_cron.go index 3e60482a..8d186947 100644 --- a/cmd/gateway_cron.go +++ b/cmd/gateway_cron.go @@ -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) } diff --git a/internal/cron/service.go b/internal/cron/service.go index 839c5e1a..f9cf32d0 100644 --- a/internal/cron/service.go +++ b/internal/cron/service.go @@ -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 diff --git a/internal/cron/types.go b/internal/cron/types.go index f2dbec35..41c61dd0 100644 --- a/internal/cron/types.go +++ b/internal/cron/types.go @@ -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. diff --git a/internal/gateway/methods/cron.go b/internal/gateway/methods/cron.go index 42d8608d..1855ff51 100644 --- a/internal/gateway/methods/cron.go +++ b/internal/gateway/methods/cron.go @@ -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, ¶ms) @@ -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 = ¶ms.WakeHeartbeat + } + if params.Stateless { + patch.Stateless = ¶ms.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, })) diff --git a/internal/store/cron_store.go b/internal/store/cron_store.go index a8ea96ff..d84d1f27 100644 --- a/internal/store/cron_store.go +++ b/internal/store/cron_store.go @@ -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"` } diff --git a/internal/store/pg/cron_crud.go b/internal/store/pg/cron_crud.go index 932da59a..aad26ee3 100644 --- a/internal/store/pg/cron_crud.go +++ b/internal/store/pg/cron_crud.go @@ -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 diff --git a/internal/store/pg/cron_scan.go b/internal/store/pg/cron_scan.go index a4e3906a..7926fb01 100644 --- a/internal/store/pg/cron_scan.go +++ b/internal/store/pg/cron_scan.go @@ -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 { diff --git a/internal/store/pg/cron_scan_test.go b/internal/store/pg/cron_scan_test.go index d122f12d..8246511d 100644 --- a/internal/store/pg/cron_scan_test.go +++ b/internal/store/pg/cron_scan_test.go @@ -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") } } diff --git a/internal/store/pg/cron_scheduler.go b/internal/store/pg/cron_scheduler.go index 73452b42..9101a601 100644 --- a/internal/store/pg/cron_scheduler.go +++ b/internal/store/pg/cron_scheduler.go @@ -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`, diff --git a/internal/store/pg/cron_update.go b/internal/store/pg/cron_update.go index 46ade3e1..a7c85970 100644 --- a/internal/store/pg/cron_update.go +++ b/internal/store/pg/cron_update.go @@ -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) diff --git a/internal/store/pg/cron_update_test.go b/internal/store/pg/cron_update_test.go index 7cac5a98..a32c6117 100644 --- a/internal/store/pg/cron_update_test.go +++ b/internal/store/pg/cron_update_test.go @@ -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) - } } diff --git a/internal/store/sqlitestore/cron.go b/internal/store/sqlitestore/cron.go index 5a47031b..12351dd1 100644 --- a/internal/store/sqlitestore/cron.go +++ b/internal/store/sqlitestore/cron.go @@ -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} diff --git a/internal/store/sqlitestore/cron_crud.go b/internal/store/sqlitestore/cron_crud.go index 0dfd86ab..cefcdaca 100644 --- a/internal/store/sqlitestore/cron_crud.go +++ b/internal/store/sqlitestore/cron_crud.go @@ -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) diff --git a/internal/store/sqlitestore/cron_scheduler.go b/internal/store/sqlitestore/cron_scheduler.go index b62f9e0e..653e1a6f 100644 --- a/internal/store/sqlitestore/cron_scheduler.go +++ b/internal/store/sqlitestore/cron_scheduler.go @@ -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`, diff --git a/internal/store/sqlitestore/schema.go b/internal/store/sqlitestore/schema.go index 528ecb00..84bcb808 100644 --- a/internal/store/sqlitestore/schema.go +++ b/internal/store/sqlitestore/schema.go @@ -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. diff --git a/internal/store/sqlitestore/schema.sql b/internal/store/sqlitestore/schema.sql index ef164d14..d31c3003 100644 --- a/internal/store/sqlitestore/schema.sql +++ b/internal/store/sqlitestore/schema.sql @@ -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), diff --git a/internal/upgrade/version.go b/internal/upgrade/version.go index a075e8dd..fde61142 100644 --- a/internal/upgrade/version.go +++ b/internal/upgrade/version.go @@ -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 diff --git a/migrations/000033_cron_payload_columns.down.sql b/migrations/000033_cron_payload_columns.down.sql new file mode 100644 index 00000000..6fb106be --- /dev/null +++ b/migrations/000033_cron_payload_columns.down.sql @@ -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; diff --git a/migrations/000033_cron_payload_columns.up.sql b/migrations/000033_cron_payload_columns.up.sql new file mode 100644 index 00000000..09ed3c7c --- /dev/null +++ b/migrations/000033_cron_payload_columns.up.sql @@ -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; diff --git a/ui/desktop/frontend/src/i18n/locales/en/cron.json b/ui/desktop/frontend/src/i18n/locales/en/cron.json index 44ee6d30..521cb4c4 100644 --- a/ui/desktop/frontend/src/i18n/locales/en/cron.json +++ b/ui/desktop/frontend/src/i18n/locales/en/cron.json @@ -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", diff --git a/ui/desktop/frontend/src/i18n/locales/vi/cron.json b/ui/desktop/frontend/src/i18n/locales/vi/cron.json index 9b642eb6..85c23ba0 100644 --- a/ui/desktop/frontend/src/i18n/locales/vi/cron.json +++ b/ui/desktop/frontend/src/i18n/locales/vi/cron.json @@ -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", diff --git a/ui/desktop/frontend/src/i18n/locales/zh/cron.json b/ui/desktop/frontend/src/i18n/locales/zh/cron.json index 7fb61d23..0e5b232f 100644 --- a/ui/desktop/frontend/src/i18n/locales/zh/cron.json +++ b/ui/desktop/frontend/src/i18n/locales/zh/cron.json @@ -130,6 +130,8 @@ "timezone": "时区", "timezoneDesc": "Cron表达式的时区" }, + "stateless": "无状态", + "statelessHelp": "跳过会话持久化。每次运行从零开始。为不需要历史上下文的定时任务节省令牌。", "toast": { "created": "定时任务已创建", "createdDesc": "{{name}} 已添加", diff --git a/ui/desktop/frontend/src/types/cron.ts b/ui/desktop/frontend/src/types/cron.ts index 65d9f789..549f14b5 100644 --- a/ui/desktop/frontend/src/types/cron.ts +++ b/ui/desktop/frontend/src/types/cron.ts @@ -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 { diff --git a/ui/web/src/i18n/locales/en/cron.json b/ui/web/src/i18n/locales/en/cron.json index 44ee6d30..521cb4c4 100644 --- a/ui/web/src/i18n/locales/en/cron.json +++ b/ui/web/src/i18n/locales/en/cron.json @@ -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", diff --git a/ui/web/src/i18n/locales/vi/cron.json b/ui/web/src/i18n/locales/vi/cron.json index 9b642eb6..85c23ba0 100644 --- a/ui/web/src/i18n/locales/vi/cron.json +++ b/ui/web/src/i18n/locales/vi/cron.json @@ -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", diff --git a/ui/web/src/i18n/locales/zh/cron.json b/ui/web/src/i18n/locales/zh/cron.json index 7fb61d23..0e5b232f 100644 --- a/ui/web/src/i18n/locales/zh/cron.json +++ b/ui/web/src/i18n/locales/zh/cron.json @@ -130,6 +130,8 @@ "timezone": "时区", "timezoneDesc": "Cron表达式的时区" }, + "stateless": "无状态", + "statelessHelp": "跳过会话持久化。每次运行从零开始。为不需要历史上下文的定时任务节省令牌。", "toast": { "created": "定时任务已创建", "createdDesc": "{{name}} 已添加", diff --git a/ui/web/src/pages/cron/cron-detail/cron-advanced-dialog.tsx b/ui/web/src/pages/cron/cron-detail/cron-advanced-dialog.tsx index 139df0df..f99bd842 100644 --- a/ui/web/src/pages/cron/cron-detail/cron-advanced-dialog.tsx +++ b/ui/web/src/pages/cron/cron-detail/cron-advanced-dialog.tsx @@ -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).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 +
+
+

{t("stateless")}

+

{t("statelessHelp")}

+
+ +
+ {/* Footer */} diff --git a/ui/web/src/pages/cron/hooks/use-cron.ts b/ui/web/src/pages/cron/hooks/use-cron.ts index 2c97671f..d7475f31 100644 --- a/ui/web/src/pages/cron/hooks/use-cron.ts +++ b/ui/web/src/pages/cron/hooks/use-cron.ts @@ -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;