mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-15 00:17:27 +00:00
feat: Implement timezone support for cron schedules, improve cron job patching, and enhance trace visibility for cron-triggered agent runs.
This commit is contained in:
1 parent
e76aabb3a4
commit
2041b7037b
8 files changed
+155
-7
No files matched your search
@@ -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
|
||||
|
||||
@@ -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 <agentID>")
|
||||
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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
+113
-3
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user