DBT/clawgo
1
0
Fork 0
mirror of https://github.com/YspCoder/clawgo.git synced 2026-04-14 19:37:31 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix cron

Browse Source
This commit is contained in:
lpf
2026-02-26 12:26:04 +08:00
parent 2615a6b196
commit dea7f361f2
9 changed files with 321 additions and 268 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
pkg/cron/cron_expr.go
View File

@@ -1,53 +0,0 @@
package cron
import (
"fmt"
"strings"
"time"
)
// SimpleCronParser is a minimal cron expression parser (minute hour day month dayOfWeek).
// It approximates standard cron behavior to provide missing Expr scheduling support.
type SimpleCronParser struct {
expr string
}
func NewSimpleCronParser(expr string) *SimpleCronParser {
return &SimpleCronParser{expr: expr}
}
func (p *SimpleCronParser) Next(from time.Time) time.Time {
fields := strings.Fields(p.expr)
if len(fields) != 5 {
return time.Time{} // Invalid format
}
// This minimal implementation only supports "*" and exact numbers.
// For production, use github.com/robfig/cron/v3.
next := from.Add(1 * time.Minute).Truncate(time.Minute)
// Simplified logic: if it is not "*" and does not match, keep incrementing until matched
// (up to one year of search).
for i := 0; i < 525600; i++ {
if p.match(next, fields) {
return next
}
next = next.Add(1 * time.Minute)
}
return time.Time{}
}
func (p *SimpleCronParser) match(t time.Time, fields []string) bool {
return matchField(fmt.Sprintf("%d", t.Minute()), fields[0]) &&
matchField(fmt.Sprintf("%d", t.Hour()), fields[1]) &&
matchField(fmt.Sprintf("%d", t.Day()), fields[2]) &&
matchField(fmt.Sprintf("%d", int(t.Month())), fields[3]) &&
matchField(fmt.Sprintf("%d", int(t.Weekday())), fields[4])
}
func matchField(val, field string) bool {
if field == "*" {
return true
}
return val == field
}

163
pkg/cron/service.go
View File

@@ -6,10 +6,12 @@ import (
"math"
"os"
"path/filepath"
"strings"
"sync"
"time"
"clawgo/pkg/lifecycle"
robcron "github.com/robfig/cron/v3"
)
const (
@@ -342,6 +344,10 @@ func (cs *CronService) computeNextRun(schedule *CronSchedule, nowMS int64) *int6
}
func (cs *CronService) computeNextRunAfter(schedule *CronSchedule, baseMS int64, nowMS int64) *int64 {
if schedule == nil {
return nil
}
if schedule.Kind == "at" {
if schedule.AtMS != nil && *schedule.AtMS > nowMS {
return schedule.AtMS
@@ -349,26 +355,126 @@ func (cs *CronService) computeNextRunAfter(schedule *CronSchedule, baseMS int64,
return nil
}
if schedule.Kind == "every" {
if schedule.EveryMS == nil || *schedule.EveryMS <= 0 {
return nil
}
next := computeAlignedEveryNext(baseMS, nowMS, *schedule.EveryMS)
return &next
expr, ok := cronExprFromSchedule(schedule)
if !ok {
return nil
}
compiled, err := parseCronSchedule(expr, strings.TrimSpace(schedule.TZ))
if err != nil {
return nil
}
if schedule.Kind == "cron" && schedule.Expr != "" {
parser := NewSimpleCronParser(schedule.Expr)
next := parser.Next(time.UnixMilli(nowMS))
if !next.IsZero() {
ms := next.UnixMilli()
return &ms
}
base := time.UnixMilli(baseMS)
now := time.UnixMilli(nowMS)
next := compiled.Next(base)
for i := 0; i < 1024 && !next.After(now); i++ {
next = compiled.Next(next)
}
if next.After(now) {
ms := next.UnixMilli()
return &ms
}
return nil
}
func cronExprFromSchedule(schedule *CronSchedule) (string, bool) {
if schedule == nil {
return "", false
}
if expr := strings.TrimSpace(schedule.Expr); expr != "" {
return expr, true
}
if schedule.Kind == "every" && schedule.EveryMS != nil && *schedule.EveryMS > 0 {
d := time.Duration(*schedule.EveryMS) * time.Millisecond
return "@every " + d.String(), true
}
return "", false
}
func parseCronSchedule(expr, tz string) (robcron.Schedule, error) {
spec := strings.TrimSpace(expr)
if spec == "" {
return nil, fmt.Errorf("empty cron expression")
}
tz = strings.TrimSpace(tz)
if tz != "" && !strings.HasPrefix(spec, "CRON_TZ=") && !strings.HasPrefix(spec, "TZ=") {
spec = "CRON_TZ=" + tz + " " + spec
}
parser := robcron.NewParser(
robcron.Minute |
robcron.Hour |
robcron.Dom |
robcron.Month |
robcron.Dow |
robcron.Descriptor,
)
return parser.Parse(spec)
}
func normalizeSchedule(schedule CronSchedule) CronSchedule {
schedule.Kind = strings.ToLower(strings.TrimSpace(schedule.Kind))
schedule.Expr = strings.TrimSpace(schedule.Expr)
schedule.TZ = strings.TrimSpace(schedule.TZ)
if schedule.Expr != "" {
schedule.Kind = "cron"
schedule.AtMS = nil
schedule.EveryMS = nil
return schedule
}
if schedule.Kind == "every" && schedule.EveryMS != nil && *schedule.EveryMS > 0 {
d := time.Duration(*schedule.EveryMS) * time.Millisecond
schedule.Expr = "@every " + d.String()
schedule.Kind = "cron"
schedule.EveryMS = nil
return schedule
}
if schedule.Kind == "at" {
schedule.EveryMS = nil
return schedule
}
if schedule.AtMS != nil {
schedule.Kind = "at"
return schedule
}
if schedule.EveryMS != nil && *schedule.EveryMS > 0 {
d := time.Duration(*schedule.EveryMS) * time.Millisecond
schedule.Expr = "@every " + d.String()
schedule.Kind = "cron"
schedule.EveryMS = nil
return schedule
}
if schedule.Expr != "" {
schedule.Kind = "cron"
}
return schedule
}
func validateSchedule(schedule CronSchedule) error {
if schedule.Kind == "at" {
if schedule.AtMS == nil || *schedule.AtMS <= time.Now().UnixMilli() {
return fmt.Errorf("invalid one-time schedule")
}
return nil
}
expr, ok := cronExprFromSchedule(&schedule)
if !ok {
return fmt.Errorf("cron expression is required")
}
_, err := parseCronSchedule(expr, schedule.TZ)
if err != nil {
return fmt.Errorf("invalid cron expression: %w", err)
}
return nil
}
func (cs *CronService) recomputeNextRuns() bool {
changed := false
now := time.Now().UnixMilli()
@@ -427,7 +533,13 @@ func (cs *CronService) loadStore() error {
return err
}
return json.Unmarshal(data, cs.store)
if err := json.Unmarshal(data, cs.store); err != nil {
return err
}
for i := range cs.store.Jobs {
cs.store.Jobs[i].Schedule = normalizeSchedule(cs.store.Jobs[i].Schedule)
}
return nil
}
func (cs *CronService) saveStore() error {
@@ -479,6 +591,11 @@ func (cs *CronService) AddJob(name string, schedule CronSchedule, message string
cs.mu.Lock()
defer cs.mu.Unlock()
schedule = normalizeSchedule(schedule)
if err := validateSchedule(schedule); err != nil {
return nil, err
}
now := time.Now().UnixMilli()
job := CronJob{
@@ -589,18 +706,6 @@ func (cs *CronService) unmarkJobRunning(jobID string) {
delete(cs.running, jobID)
}
func computeAlignedEveryNext(baseMS, nowMS, intervalMS int64) int64 {
if intervalMS <= 0 {
return nowMS
}
next := baseMS + intervalMS
if next > nowMS {
return next
}
miss := (nowMS-next)/intervalMS + 1
return next + miss*intervalMS
}
func computeRetryBackoff(consecutiveFailures int64, base, max time.Duration) time.Duration {
if base <= 0 {
base = defaultRetryBackoffBase
@@ -688,7 +793,11 @@ func (cs *CronService) UpdateJob(jobID string, in UpdateJobInput) (*CronJob, err
job.Name = *in.Name
}
if in.Schedule != nil {
job.Schedule = *in.Schedule
nextSchedule := normalizeSchedule(*in.Schedule)
if err := validateSchedule(nextSchedule); err != nil {
return nil, err
}
job.Schedule = nextSchedule
if job.Enabled {
now := time.Now().UnixMilli()
job.State.NextRunAtMS = cs.computeNextRun(&job.Schedule, now)

77
pkg/nodes/registry_server.go
View File

@@ -46,9 +46,9 @@ func NewRegistryServer(host string, port int, token string, mgr *Manager) *Regis
return &RegistryServer{addr: fmt.Sprintf("%s:%d", addr, port), token: strings.TrimSpace(token), mgr: mgr}
}
func (s *RegistryServer) SetConfigPath(path string) { s.configPath = strings.TrimSpace(path) }
func (s *RegistryServer) SetConfigPath(path string) { s.configPath = strings.TrimSpace(path) }
func (s *RegistryServer) SetWorkspacePath(path string) { s.workspacePath = strings.TrimSpace(path) }
func (s *RegistryServer) SetLogFilePath(path string) { s.logFilePath = strings.TrimSpace(path) }
func (s *RegistryServer) SetLogFilePath(path string) { s.logFilePath = strings.TrimSpace(path) }
func (s *RegistryServer) SetChatHandler(fn func(ctx context.Context, sessionKey, content string) (string, error)) {
s.onChat = fn
}
@@ -131,7 +131,9 @@ func (s *RegistryServer) handleHeartbeat(w http.ResponseWriter, r *http.Request)
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
var body struct{ ID string `json:"id"` }
var body struct {
ID string `json:"id"`
}
if err := json.NewDecoder(r.Body).Decode(&body); err != nil || strings.TrimSpace(body.ID) == "" {
http.Error(w, "id required", http.StatusBadRequest)
return
@@ -633,17 +635,27 @@ func (s *RegistryServer) handleWebUISkills(w http.ResponseWriter, r *http.Reques
desc, tools, sys := readSkillMeta(filepath.Join(dir, name, "SKILL.md"))
if desc == "" || len(tools) == 0 || sys == "" {
d2, t2, s2 := readSkillMeta(filepath.Join(dir, baseName, "SKILL.md"))
if desc == "" { desc = d2 }
if len(tools) == 0 { tools = t2 }
if sys == "" { sys = s2 }
if desc == "" {
desc = d2
}
if len(tools) == 0 {
tools = t2
}
if sys == "" {
sys = s2
}
}
if tools == nil {
tools = []string{}
}
if tools == nil { tools = []string{} }
it := skillItem{ID: baseName, Name: baseName, Description: desc, Tools: tools, SystemPrompt: sys, Enabled: enabled, UpdateChecked: checkUpdates, Source: dir}
if checkUpdates {
found, version, checkErr := queryClawHubSkillVersion(r.Context(), baseName)
it.RemoteFound = found
it.RemoteVersion = version
if checkErr != nil { it.CheckError = checkErr.Error() }
if checkErr != nil {
it.CheckError = checkErr.Error()
}
}
items = append(items, it)
}
@@ -660,7 +672,9 @@ func (s *RegistryServer) handleWebUISkills(w http.ResponseWriter, r *http.Reques
action = strings.ToLower(strings.TrimSpace(action))
id, _ := body["id"].(string)
name, _ := body["name"].(string)
if strings.TrimSpace(name) == "" { name = id }
if strings.TrimSpace(name) == "" {
name = id
}
name = strings.TrimSpace(name)
if name == "" {
http.Error(w, "name required", http.StatusBadRequest)
@@ -701,7 +715,9 @@ func (s *RegistryServer) handleWebUISkills(w http.ResponseWriter, r *http.Reques
var toolsList []string
if arr, ok := body["tools"].([]interface{}); ok {
for _, v := range arr {
if sv, ok := v.(string); ok && strings.TrimSpace(sv) != "" { toolsList = append(toolsList, strings.TrimSpace(sv)) }
if sv, ok := v.(string); ok && strings.TrimSpace(sv) != "" {
toolsList = append(toolsList, strings.TrimSpace(sv))
}
}
}
if action == "create" {
@@ -733,8 +749,12 @@ func (s *RegistryServer) handleWebUISkills(w http.ResponseWriter, r *http.Reques
pathA := filepath.Join(skillsDir, id)
pathB := pathA + ".disabled"
deleted := false
if err := os.RemoveAll(pathA); err == nil { deleted = true }
if err := os.RemoveAll(pathB); err == nil { deleted = true }
if err := os.RemoveAll(pathA); err == nil {
deleted = true
}
if err := os.RemoveAll(pathB); err == nil {
deleted = true
}
_ = json.NewEncoder(w).Encode(map[string]interface{}{"ok": true, "deleted": deleted, "id": id})
default:
@@ -742,7 +762,6 @@ func (s *RegistryServer) handleWebUISkills(w http.ResponseWriter, r *http.Reques
}
}
func buildSkillMarkdown(name, desc string, tools []string, systemPrompt string) string {
if strings.TrimSpace(desc) == "" {
desc = "No description provided."
@@ -913,16 +932,32 @@ func normalizeCronJob(v interface{}) map[string]interface{} {
out[k] = val
}
if sch, ok := m["schedule"].(map[string]interface{}); ok {
if kind, ok := sch["kind"]; ok { out["kind"] = kind }
if every, ok := sch["everyMs"]; ok { out["everyMs"] = every }
if expr, ok := sch["expr"]; ok { out["expr"] = expr }
if at, ok := sch["atMs"]; ok { out["atMs"] = at }
kind, _ := sch["kind"].(string)
if expr, ok := sch["expr"].(string); ok && strings.TrimSpace(expr) != "" {
out["expr"] = strings.TrimSpace(expr)
} else if strings.EqualFold(strings.TrimSpace(kind), "every") {
if every, ok := sch["everyMs"].(float64); ok && every > 0 {
out["expr"] = fmt.Sprintf("@every %s", (time.Duration(int64(every)) * time.Millisecond).String())
}
} else if strings.EqualFold(strings.TrimSpace(kind), "at") {
if at, ok := sch["atMs"].(float64); ok && at > 0 {
out["expr"] = time.UnixMilli(int64(at)).Format(time.RFC3339)
}
}
}
if payload, ok := m["payload"].(map[string]interface{}); ok {
if msg, ok := payload["message"]; ok { out["message"] = msg }
if d, ok := payload["deliver"]; ok { out["deliver"] = d }
if c, ok := payload["channel"]; ok { out["channel"] = c }
if to, ok := payload["to"]; ok { out["to"] = to }
if msg, ok := payload["message"]; ok {
out["message"] = msg
}
if d, ok := payload["deliver"]; ok {
out["deliver"] = d
}
if c, ok := payload["channel"]; ok {
out["channel"] = c
}
if to, ok := payload["to"]; ok {
out["to"] = to
}
}
return out
}