From 2041b7037baceb3e16bb35401903442ba2e67cae Mon Sep 17 00:00:00 2001 From: viettranx Date: Sat, 28 Feb 2026 14:17:30 +0700 Subject: [PATCH] feat: Implement timezone support for cron schedules, improve cron job patching, and enhance trace visibility for cron-triggered agent runs. --- cmd/gateway_consumer.go | 2 + internal/agent/loop.go | 9 ++- internal/cron/service_execution.go | 10 +++ internal/store/pg/cron.go | 116 +++++++++++++++++++++++++++- internal/store/pg/cron_scan.go | 11 ++- internal/store/pg/cron_scheduler.go | 3 + internal/tools/cron.go | 5 ++ main.go | 6 +- 8 files changed, 155 insertions(+), 7 deletions(-) diff --git a/cmd/gateway_consumer.go b/cmd/gateway_consumer.go index bf82bcba..5c64dde0 100644 --- a/cmd/gateway_consumer.go +++ b/cmd/gateway_consumer.go @@ -669,6 +669,8 @@ func makeCronJobHandler(agents *agent.Router, msgBus *bus.MessageBus, cfg *confi UserID: job.UserID, RunID: fmt.Sprintf("cron-%s", job.ID), Stream: false, + TraceName: fmt.Sprintf("Cron [%s] - %s", job.Name, agentID), + TraceTags: []string{"cron"}, }) if err != nil { return nil, err diff --git a/internal/agent/loop.go b/internal/agent/loop.go index 60df5d0d..5d9e811d 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -239,6 +239,8 @@ type RunRequest struct { HistoryLimit int // max user turns to keep in context (0=unlimited, from channel config) ParentTraceID uuid.UUID // if set, reuse parent trace instead of creating new (announce runs) ParentRootSpanID uuid.UUID // if set, nest announce agent span under this parent span + TraceName string // override trace name (default: "chat ") + TraceTags []string // additional tags for the trace (e.g. "cron") } // RunResult is the output of a completed agent run. @@ -282,17 +284,22 @@ func (l *Loop) Run(ctx context.Context, req RunRequest) (*RunResult, error) { } else if l.traceCollector != nil { traceID = store.GenNewID() now := time.Now().UTC() + traceName := "chat " + l.id + if req.TraceName != "" { + traceName = req.TraceName + } trace := &store.TraceData{ ID: traceID, RunID: req.RunID, SessionKey: req.SessionKey, UserID: req.UserID, Channel: req.Channel, - Name: "chat " + l.id, + Name: traceName, InputPreview: truncateStr(req.Message, 500), Status: store.TraceStatusRunning, StartTime: now, CreatedAt: now, + Tags: req.TraceTags, } if l.agentUUID != uuid.Nil { trace.AgentID = &l.agentUUID diff --git a/internal/cron/service_execution.go b/internal/cron/service_execution.go index 0edf561e..99ba4e2a 100644 --- a/internal/cron/service_execution.go +++ b/internal/cron/service_execution.go @@ -267,6 +267,11 @@ func (cs *Service) computeNextRun(schedule *Schedule, now int64) *int64 { return nil } nowTime := time.UnixMilli(now) + if schedule.TZ != "" { + if loc, err := time.LoadLocation(schedule.TZ); err == nil { + nowTime = nowTime.In(loc) + } + } nextTime, err := gronx.NextTickAfter(schedule.Expr, nowTime, false) if err != nil { slog.Error("cron: failed to compute next run", "expr", schedule.Expr, "error", err) @@ -298,6 +303,11 @@ func (cs *Service) validateSchedule(schedule *Schedule) error { if !gx.IsValid(schedule.Expr) { return fmt.Errorf("invalid cron expression: %s", schedule.Expr) } + if schedule.TZ != "" { + if _, err := time.LoadLocation(schedule.TZ); err != nil { + return fmt.Errorf("invalid timezone: %s", schedule.TZ) + } + } default: return fmt.Errorf("unknown schedule kind: %s", schedule.Kind) } diff --git a/internal/store/pg/cron.go b/internal/store/pg/cron.go index 10633967..f3e0d137 100644 --- a/internal/store/pg/cron.go +++ b/internal/store/pg/cron.go @@ -9,6 +9,7 @@ import ( "sync" "time" + "github.com/adhocore/gronx" "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/cron" @@ -47,6 +48,12 @@ func (s *PGCronStore) SetRetryConfig(cfg cron.RetryConfig) { } func (s *PGCronStore) AddJob(name string, schedule store.CronSchedule, message string, deliver bool, channel, to, agentID, userID string) (*store.CronJob, error) { + if schedule.TZ != "" { + if _, err := time.LoadLocation(schedule.TZ); err != nil { + return nil, fmt.Errorf("invalid timezone: %s", schedule.TZ) + } + } + payload := store.CronPayload{ Kind: "agent_turn", Message: message, Deliver: deliver, Channel: channel, To: to, } @@ -191,10 +198,113 @@ func (s *PGCronStore) UpdateJob(jobID string, patch store.CronJobPatch) (*store. updates["enabled"] = *patch.Enabled } if patch.Schedule != nil { - updates["schedule_kind"] = patch.Schedule.Kind - if patch.Schedule.Expr != "" { - updates["cron_expression"] = patch.Schedule.Expr + // Fetch current schedule to merge with patch (partial updates) + var curKind string + var curExpr, curTZ *string + var curIntervalMS *int64 + var curRunAt *time.Time + s.db.QueryRow( + "SELECT schedule_kind, cron_expression, timezone, interval_ms, run_at FROM cron_jobs WHERE id = $1", id, + ).Scan(&curKind, &curExpr, &curTZ, &curIntervalMS, &curRunAt) + + // Resolve the effective schedule kind + newKind := patch.Schedule.Kind + if newKind == "" { + newKind = curKind // keep current kind if not specified } + updates["schedule_kind"] = newKind + + // Set type-specific fields and clear others + switch newKind { + case "cron": + if patch.Schedule.Expr != "" { + updates["cron_expression"] = patch.Schedule.Expr + } + if patch.Schedule.TZ != "" { + if _, err := time.LoadLocation(patch.Schedule.TZ); err != nil { + return nil, fmt.Errorf("invalid timezone: %s", patch.Schedule.TZ) + } + updates["timezone"] = patch.Schedule.TZ + } + // Clear other type fields when switching to cron + if curKind != "cron" { + updates["interval_ms"] = nil + updates["run_at"] = nil + } + case "every": + if patch.Schedule.EveryMS != nil { + updates["interval_ms"] = *patch.Schedule.EveryMS + } + // Clear other type fields when switching to every + if curKind != "every" { + updates["cron_expression"] = nil + updates["timezone"] = nil + updates["run_at"] = nil + } + case "at": + if patch.Schedule.AtMS != nil { + t := time.UnixMilli(*patch.Schedule.AtMS) + updates["run_at"] = t + } + // Clear other type fields when switching to at + if curKind != "at" { + updates["cron_expression"] = nil + updates["timezone"] = nil + updates["interval_ms"] = nil + } + } + + // Build merged schedule for recomputing next_run_at + merged := store.CronSchedule{Kind: newKind} + switch newKind { + case "cron": + if patch.Schedule.Expr != "" { + merged.Expr = patch.Schedule.Expr + } else if curExpr != nil { + merged.Expr = *curExpr + } + if patch.Schedule.TZ != "" { + merged.TZ = patch.Schedule.TZ + } else if curTZ != nil && newKind == curKind { + merged.TZ = *curTZ + } + case "every": + if patch.Schedule.EveryMS != nil { + merged.EveryMS = patch.Schedule.EveryMS + } else if curIntervalMS != nil { + merged.EveryMS = curIntervalMS + } + case "at": + if patch.Schedule.AtMS != nil { + merged.AtMS = patch.Schedule.AtMS + } else if curRunAt != nil { + ms := curRunAt.UnixMilli() + merged.AtMS = &ms + } + } + + // Validate the merged schedule before applying + switch merged.Kind { + case "cron": + if merged.Expr == "" { + return nil, fmt.Errorf("cron schedule requires expr") + } + gx := gronx.New() + if !gx.IsValid(merged.Expr) { + return nil, fmt.Errorf("invalid cron expression: %s", merged.Expr) + } + case "every": + if merged.EveryMS == nil || *merged.EveryMS <= 0 { + return nil, fmt.Errorf("every schedule requires positive everyMs") + } + case "at": + if merged.AtMS == nil { + return nil, fmt.Errorf("at schedule requires atMs") + } + } + + next := computeNextRun(&merged, time.Now()) + updates["next_run_at"] = next } if patch.DeleteAfterRun != nil { updates["delete_after_run"] = *patch.DeleteAfterRun diff --git a/internal/store/pg/cron_scan.go b/internal/store/pg/cron_scan.go index 1988d392..d2c39af8 100644 --- a/internal/store/pg/cron_scan.go +++ b/internal/store/pg/cron_scan.go @@ -123,11 +123,18 @@ func computeNextRun(schedule *store.CronSchedule, now time.Time) *time.Time { if schedule.Expr == "" { return nil } - nextTime, err := gronx.NextTickAfter(schedule.Expr, now, false) + evalTime := now + if schedule.TZ != "" { + if loc, err := time.LoadLocation(schedule.TZ); err == nil { + evalTime = now.In(loc) + } + } + nextTime, err := gronx.NextTickAfter(schedule.Expr, evalTime, false) if err != nil { return nil } - return &nextTime + utcNext := nextTime.UTC() + return &utcNext default: return nil } diff --git a/internal/store/pg/cron_scheduler.go b/internal/store/pg/cron_scheduler.go index ee2e4d0f..881abe01 100644 --- a/internal/store/pg/cron_scheduler.go +++ b/internal/store/pg/cron_scheduler.go @@ -98,6 +98,9 @@ func (s *PGCronStore) recomputeStaleJobs() { if intervalMS != nil { schedule.EveryMS = intervalMS } + if tz != nil { + schedule.TZ = *tz + } next := computeNextRun(&schedule, now) if next == nil { diff --git a/internal/tools/cron.go b/internal/tools/cron.go index 6aad3e26..ae031b2a 100644 --- a/internal/tools/cron.go +++ b/internal/tools/cron.go @@ -203,6 +203,11 @@ func (t *CronTool) handleAdd(ctx context.Context, args map[string]interface{}, a return ErrorResult("job.schedule.expr is required for 'cron' schedule") } schedule.TZ = stringFromMap(scheduleObj, "tz") + if schedule.TZ != "" { + if _, err := time.LoadLocation(schedule.TZ); err != nil { + return ErrorResult(fmt.Sprintf("invalid timezone '%s': use IANA names like 'Asia/Ho_Chi_Minh', 'America/New_York'", schedule.TZ)) + } + } default: return ErrorResult(fmt.Sprintf("invalid schedule kind: %s (must be at, every, or cron)", schedule.Kind)) } diff --git a/main.go b/main.go index a02df032..0b46fc49 100644 --- a/main.go +++ b/main.go @@ -1,6 +1,10 @@ package main -import "github.com/nextlevelbuilder/goclaw/cmd" +import ( + _ "time/tzdata" // embed IANA timezone database for containers without tzdata + + "github.com/nextlevelbuilder/goclaw/cmd" +) func main() { cmd.Execute()