diff --git a/agent/cancelcause.go b/agent/cancelcause.go index 0e6b4e8..207a5cc 100644 --- a/agent/cancelcause.go +++ b/agent/cancelcause.go @@ -27,7 +27,8 @@ func Causef(code, short, format string, args ...any) *AbortCause { var ( // Task-level execution context. - AbortPausedByUser = cause("paused_by_user", "用户暂停了任务", + AbortPausedBySchedule = cause("paused_by_schedule", "计划窗口结束，暂停任务", "任务将在下一个计划窗口恢复，本次执行已停止") + AbortPausedByUser = cause("paused_by_user", "用户暂停了任务", "用户通过任务控制接口（POST /api/tasks/{id}/control，action=pause）暂停了任务。本次 Planner/Worker 运行被主动取消；运行中的意图会退回 frontier(open)，恢复任务后重新领取并从头执行") AbortPausedByOrchestrator = cause("paused_by_orchestrator", "编排 Agent 暂停了任务", "编排 Agent 调用了 pause_task 工具暂停本任务。本次 Planner/Worker 运行被主动取消；运行中的意图会退回 frontier(open)，恢复后重新执行") diff --git a/db/schedules.go b/db/schedules.go new file mode 100644 index 0000000..12dec12 --- /dev/null +++ b/db/schedules.go @@ -0,0 +1,358 @@ +package db + +import ( + "database/sql" + "encoding/json" + "fmt" + "sort" + "strings" + "time" +) + +// TaskSchedule is a persisted calendar window controlling one or more tasks. +// Weekdays is kept as JSON text so the API can expose a stable []int shape. +type TaskSchedule struct { + ID int64 `json:"id"` + Name string `json:"name"` + Enabled bool `json:"enabled"` + TimezoneMode string `json:"timezone_mode"` + ScheduleType string `json:"schedule_type"` + RunDate string `json:"run_date,omitempty"` + Weekdays []int `json:"weekdays,omitempty"` + StartTime string `json:"start_time"` + EndTime string `json:"end_time,omitempty"` + StartImmediately bool `json:"start_immediately"` + Status string `json:"status"` + LastWindowKey string `json:"last_window_key,omitempty"` + LastTransitionAt *time.Time `json:"last_transition_at,omitempty"` + LastError string `json:"last_error,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + TaskIDs []int64 `json:"task_ids"` +} + +type TaskScheduleInput struct { + Name string + Enabled bool + TimezoneMode string + ScheduleType string + RunDate string + Weekdays []int + StartTime string + EndTime string + StartImmediately bool + TaskIDs []int64 +} + +const scheduleCols = `id,name,enabled,timezone_mode,schedule_type, +COALESCE(to_char(run_date,'YYYY-MM-DD'),''),weekdays,to_char(start_time,'HH24:MI:SS'), +COALESCE(to_char(end_time,'HH24:MI:SS'),''),start_immediately,status,last_window_key, +last_transition_at,last_error,created_at,updated_at` + +func marshalWeekdays(days []int) string { + if len(days) == 0 { + return "[]" + } + b, _ := json.Marshal(days) + return string(b) +} + +func scanSchedule(sc interface{ Scan(...any) error }) (*TaskSchedule, error) { + var s TaskSchedule + var rawDays string + if err := sc.Scan(&s.ID, &s.Name, &s.Enabled, &s.TimezoneMode, &s.ScheduleType, + &s.RunDate, &rawDays, &s.StartTime, &s.EndTime, &s.StartImmediately, + &s.Status, &s.LastWindowKey, &s.LastTransitionAt, &s.LastError, + &s.CreatedAt, &s.UpdatedAt); err != nil { + return nil, err + } + _ = json.Unmarshal([]byte(rawDays), &s.Weekdays) + if s.Weekdays == nil { + s.Weekdays = []int{} + } + return &s, nil +} + +func normalizeScheduleInput(in TaskScheduleInput) (TaskScheduleInput, error) { + in.Name = strings.TrimSpace(in.Name) + if in.Name == "" { + return in, fmt.Errorf("计划名称不能为空") + } + if in.TimezoneMode == "" { + in.TimezoneMode = "beijing" + } + if in.TimezoneMode != "beijing" && in.TimezoneMode != "system" { + return in, fmt.Errorf("timezone_mode 必须为 beijing 或 system") + } + if in.ScheduleType != "once" && in.ScheduleType != "weekly" { + return in, fmt.Errorf("schedule_type 必须为 once 或 weekly") + } + if _, err := time.Parse("15:04", in.StartTime); err != nil { + return in, fmt.Errorf("start_time 必须为 HH:MM") + } + if in.EndTime != "" && in.EndTime != "24:00" { + if _, err := time.Parse("15:04", in.EndTime); err != nil { + return in, fmt.Errorf("end_time 必须为 HH:MM") + } + } + if in.ScheduleType == "once" { + if _, err := time.Parse("2006-01-02", in.RunDate); err != nil { + return in, fmt.Errorf("run_date 必须为 YYYY-MM-DD") + } + in.Weekdays = nil + } else { + seen := map[int]bool{} + for _, day := range in.Weekdays { + if day < 1 || day > 7 || seen[day] { + return in, fmt.Errorf("weekdays 必须是 1-7 且不能重复") + } + seen[day] = true + } + if len(in.Weekdays) == 0 { + return in, fmt.Errorf("weekly 计划至少选择一天") + } + sort.Ints(in.Weekdays) + in.RunDate = "" + } + seenTasks := map[int64]bool{} + for _, id := range in.TaskIDs { + if id <= 0 || seenTasks[id] { + return in, fmt.Errorf("任务编号无效或重复") + } + seenTasks[id] = true + } + if len(in.TaskIDs) > 100 { + return in, fmt.Errorf("最多选择100个任务") + } + return in, nil +} + +func (d *DB) CreateTaskSchedule(in TaskScheduleInput) (*TaskSchedule, error) { + var err error + if in, err = normalizeScheduleInput(in); err != nil { + return nil, err + } + tx, err := d.Begin() + if err != nil { + return nil, err + } + defer tx.Rollback() + return createTaskScheduleTxCommit(tx, in) +} + +func createTaskScheduleTxCommit(tx *sql.Tx, in TaskScheduleInput) (*TaskSchedule, error) { + s, err := insertTaskSchedule(tx, in) + if err != nil { + return nil, err + } + if err = tx.Commit(); err != nil { + return nil, err + } + return s, nil +} + +func insertTaskSchedule(tx *sql.Tx, in TaskScheduleInput) (*TaskSchedule, error) { + row := tx.QueryRow(`INSERT INTO task_schedules(name,enabled,timezone_mode,schedule_type,run_date,weekdays,start_time,end_time,start_immediately) +VALUES ($1,$2,$3,$4,NULLIF($5,'')::date,$6,$7::time,NULLIF($8,'')::time,$9) RETURNING `+scheduleCols, + in.Name, in.Enabled, in.TimezoneMode, in.ScheduleType, in.RunDate, marshalWeekdays(in.Weekdays), in.StartTime, in.EndTime, in.StartImmediately) + s, err := scanSchedule(row) + if err != nil { + return nil, err + } + for _, taskID := range in.TaskIDs { + if taskID <= 0 { + return nil, fmt.Errorf("task_id 无效") + } + var exists bool + if err := tx.QueryRow(`SELECT EXISTS(SELECT 1 FROM tasks WHERE id=$1 AND deleted_at IS NULL)`, taskID).Scan(&exists); err != nil || !exists { + if err != nil { + return nil, err + } + return nil, fmt.Errorf("任务 #%d 不存在", taskID) + } + if _, err := tx.Exec(`INSERT INTO task_schedule_tasks(schedule_id,task_id) VALUES($1,$2)`, s.ID, taskID); err != nil { + return nil, err + } + } + s.TaskIDs = append([]int64(nil), in.TaskIDs...) + return s, nil +} + +func (d *DB) UpdateTaskSchedule(id int64, in TaskScheduleInput) (*TaskSchedule, error) { + var err error + if in, err = normalizeScheduleInput(in); err != nil { + return nil, err + } + tx, err := d.Begin() + if err != nil { + return nil, err + } + defer tx.Rollback() + row := tx.QueryRow(`UPDATE task_schedules SET name=$2,enabled=$3,timezone_mode=$4,schedule_type=$5,run_date=NULLIF($6,'')::date,weekdays=$7,start_time=$8::time,end_time=NULLIF($9,'')::time,start_immediately=$10,status=CASE WHEN $3 THEN CASE WHEN status='completed' THEN 'waiting' ELSE status END ELSE 'paused' END,last_error='',last_window_key='' WHERE id=$1 RETURNING `+scheduleCols, + id, in.Name, in.Enabled, in.TimezoneMode, in.ScheduleType, in.RunDate, marshalWeekdays(in.Weekdays), in.StartTime, in.EndTime, in.StartImmediately) + s, err := scanSchedule(row) + if err != nil { + return nil, err + } + if _, err := tx.Exec(`DELETE FROM task_schedule_tasks WHERE schedule_id=$1`, id); err != nil { + return nil, err + } + for _, taskID := range in.TaskIDs { + if _, err := tx.Exec(`INSERT INTO task_schedule_tasks(schedule_id,task_id) VALUES($1,$2)`, id, taskID); err != nil { + return nil, err + } + } + s.TaskIDs = append([]int64(nil), in.TaskIDs...) + if err := tx.Commit(); err != nil { + return nil, err + } + return s, nil +} + +func (d *DB) DeleteTaskSchedule(id int64) error { + _, err := d.Exec(`DELETE FROM task_schedules WHERE id=$1`, id) + return err +} + +func (d *DB) SetTaskScheduleEnabled(id int64, enabled bool) error { + _, err := d.Exec(`UPDATE task_schedules SET enabled=$2,status=CASE WHEN $2 THEN 'waiting' ELSE 'paused' END,last_error='' WHERE id=$1`, id, enabled) + return err +} + +func (d *DB) GetTaskSchedule(id int64) (*TaskSchedule, error) { + s, err := scanSchedule(d.QueryRow(`SELECT `+scheduleCols+` FROM task_schedules WHERE id=$1`, id)) + if err != nil { + return nil, err + } + s.TaskIDs, err = d.scheduleTaskIDs(id) + return s, err +} + +func (d *DB) ListTaskSchedules() ([]*TaskSchedule, error) { + rows, err := d.Query(`SELECT ` + scheduleCols + ` FROM task_schedules ORDER BY id DESC`) + if err != nil { + return nil, err + } + defer rows.Close() + out := []*TaskSchedule{} + for rows.Next() { + s, err := scanSchedule(rows) + if err != nil { + return nil, err + } + s.TaskIDs, err = d.scheduleTaskIDs(s.ID) + if err != nil { + return nil, err + } + out = append(out, s) + } + return out, rows.Err() +} + +func (d *DB) ListEnabledTaskSchedules() ([]*TaskSchedule, error) { + rows, err := d.Query(`SELECT ` + scheduleCols + ` FROM task_schedules WHERE enabled ORDER BY id`) + if err != nil { + return nil, err + } + defer rows.Close() + out := []*TaskSchedule{} + for rows.Next() { + s, err := scanSchedule(rows) + if err != nil { + return nil, err + } + s.TaskIDs, err = d.scheduleTaskIDs(s.ID) + if err != nil { + return nil, err + } + out = append(out, s) + } + return out, rows.Err() +} + +func (d *DB) scheduleTaskIDs(id int64) ([]int64, error) { + rows, err := d.Query(`SELECT task_id FROM task_schedule_tasks WHERE schedule_id=$1 ORDER BY task_id`, id) + if err != nil { + return nil, err + } + defer rows.Close() + ids := []int64{} + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} + +func (d *DB) TouchTaskSchedule(id int64, status, windowKey, lastError string) error { + _, err := d.Exec(`UPDATE task_schedules SET status=$2,last_window_key=$3,last_transition_at=now(),last_error=$4 WHERE id=$1`, id, status, windowKey, lastError) + return err +} + +// TaskScheduleTasks returns all schedule IDs that reference a task. +func (d *DB) TaskScheduleTasks(taskID int64) ([]int64, error) { + rows, err := d.Query(`SELECT schedule_id FROM task_schedule_tasks WHERE task_id=$1 ORDER BY schedule_id`, taskID) + if err != nil { + return nil, err + } + defer rows.Close() + ids := []int64{} + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} + +func (d *DB) SetSchedulePaused(scheduleID, taskID int64, paused bool) error { + _, err := d.Exec(`UPDATE task_schedule_tasks SET paused_by_schedule=$3 WHERE schedule_id=$1 AND task_id=$2`, scheduleID, taskID, paused) + return err +} + +func (d *DB) SchedulePausedTasks(taskID int64) (bool, error) { + var paused bool + err := d.QueryRow(`SELECT EXISTS(SELECT 1 FROM task_schedule_tasks WHERE task_id=$1 AND paused_by_schedule)`, taskID).Scan(&paused) + return paused, err +} + +// ValidateTaskScheduleInput shares validation between task creation and schedules. +func ValidateTaskScheduleInput(in TaskScheduleInput) error { + _, err := normalizeScheduleInput(in) + return err +} + +type ScheduleHistory struct { + ID int64 `json:"id"` + TaskID *int64 `json:"task_id,omitempty"` + Action string `json:"action"` + Success bool `json:"success"` + Message string `json:"message"` + CreatedAt time.Time `json:"created_at"` +} + +func (d *DB) RecordScheduleHistory(id, taskID int64, action string, success bool, message string) error { + _, err := d.Exec(`INSERT INTO task_schedule_history(schedule_id,task_id,action,success,message) VALUES($1,NULLIF($2,0),$3,$4,$5)`, id, taskID, action, success, message) + return err +} +func (d *DB) ListScheduleHistory(id int64) ([]ScheduleHistory, error) { + rows, err := d.Query(`SELECT id,task_id,action,success,message,created_at FROM task_schedule_history WHERE schedule_id=$1 ORDER BY id DESC LIMIT 100`, id) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ScheduleHistory{} + for rows.Next() { + var item ScheduleHistory + if err := rows.Scan(&item.ID, &item.TaskID, &item.Action, &item.Success, &item.Message, &item.CreatedAt); err != nil { + return nil, err + } + items = append(items, item) + } + return items, rows.Err() +} diff --git a/db/schema.sql b/db/schema.sql index b784bce..a578e01 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -1027,6 +1027,49 @@ CREATE TABLE IF NOT EXISTS scheduler_state ( value TEXT NOT NULL DEFAULT '' ); +-- ===================================================================== +-- J1. Calendar based task schedules +-- ===================================================================== +CREATE TABLE IF NOT EXISTS task_schedules ( + id BIGSERIAL PRIMARY KEY, + name TEXT NOT NULL, + enabled BOOLEAN NOT NULL DEFAULT true, + timezone_mode TEXT NOT NULL DEFAULT 'beijing' + CHECK (timezone_mode IN ('beijing','system')), + schedule_type TEXT NOT NULL DEFAULT 'weekly' + CHECK (schedule_type IN ('once','weekly')), + run_date DATE, + weekdays TEXT NOT NULL DEFAULT '', + start_time TIME NOT NULL, + end_time TIME, + start_immediately BOOLEAN NOT NULL DEFAULT false, + status TEXT NOT NULL DEFAULT 'waiting' + CHECK (status IN ('waiting','running','outside','completed','error','paused')), + last_window_key TEXT NOT NULL DEFAULT '', + last_transition_at TIMESTAMPTZ, + last_error TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CHECK ((schedule_type = 'once' AND run_date IS NOT NULL) OR + (schedule_type = 'weekly' AND run_date IS NULL)), + CHECK ((schedule_type = 'once') OR weekdays <> '') +); +CREATE INDEX IF NOT EXISTS idx_task_schedules_enabled ON task_schedules(enabled, id); +DROP TRIGGER IF EXISTS trg_task_schedules_upd ON task_schedules; +CREATE TRIGGER trg_task_schedules_upd BEFORE UPDATE ON task_schedules + FOR EACH ROW EXECUTE FUNCTION set_updated_at(); + +CREATE TABLE IF NOT EXISTS task_schedule_tasks ( + schedule_id BIGINT NOT NULL REFERENCES task_schedules(id) ON DELETE CASCADE, + task_id BIGINT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, + paused_by_schedule BOOLEAN NOT NULL DEFAULT false, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (schedule_id, task_id) +); +ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_paused BOOLEAN NOT NULL DEFAULT false; +CREATE INDEX IF NOT EXISTS idx_task_schedule_tasks_task ON task_schedule_tasks(task_id); +ALTER TABLE task_schedule_tasks ADD COLUMN IF NOT EXISTS paused_by_schedule BOOLEAN NOT NULL DEFAULT false; + -- ===================================================================== -- K. 拦截规则 -- ===================================================================== @@ -1475,3 +1518,14 @@ ALTER TABLE finding_case_review_runs ADD COLUMN IF NOT EXISTS finding_ids JSONB ALTER TABLE finding_case_review_runs ADD COLUMN IF NOT EXISTS reviewed_ids JSONB NOT NULL DEFAULT '[]'; ALTER TABLE finding_case_review_runs ADD COLUMN IF NOT EXISTS conclusion JSONB NOT NULL DEFAULT '{}'; CREATE INDEX IF NOT EXISTS idx_finding_case_reviews_active_task ON finding_case_review_runs(task_id) WHERE state IN ('queued','running'); + +CREATE TABLE IF NOT EXISTS task_schedule_history ( + id BIGSERIAL PRIMARY KEY, + schedule_id BIGINT NOT NULL REFERENCES task_schedules(id) ON DELETE CASCADE, + task_id BIGINT, + action TEXT NOT NULL, + success BOOLEAN NOT NULL, + message TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +CREATE INDEX IF NOT EXISTS idx_schedule_history ON task_schedule_history(schedule_id,id DESC); diff --git a/db/tasks.go b/db/tasks.go index 26a82f8..4e05cb9 100644 --- a/db/tasks.go +++ b/db/tasks.go @@ -144,6 +144,7 @@ func (d *DB) CreateTask(description, goal string, llmProfileID *int64, timeoutSe // TaskCreateOptions contains the task data that must be committed atomically // with the task/exploration row. type TaskCreateOptions struct { + Schedule *TaskScheduleInput Name string // 可选任务名称;空=未命名 CategoryID *int64 SourceTaskIDs []int64 @@ -248,6 +249,23 @@ RETURNING id, status, paused, created_at`, opts.Name, opts.CategoryID, descripti } else { t.LLMFailoverState = "ready" } + if opts.Schedule != nil { + in, err := normalizeScheduleInput(*opts.Schedule) + if err != nil { + return nil, err + } + in.TaskIDs = []int64{t.ID} + if _, err = insertTaskSchedule(tx, in); err != nil { + return nil, err + } + if !in.StartImmediately { + if _, err = tx.Exec(`UPDATE tasks SET paused=true,schedule_paused=true,queue_mode='bootstrap' WHERE id=$1`, t.ID); err != nil { + return nil, err + } + t.Paused = true + t.QueueMode = "bootstrap" + } + } return t, tx.Commit() } diff --git a/server/calendar_schedule.go b/server/calendar_schedule.go new file mode 100644 index 0000000..6b2b5d1 --- /dev/null +++ b/server/calendar_schedule.go @@ -0,0 +1,214 @@ +package server + +import ( + "fmt" + "strconv" + "strings" + "time" + _ "time/tzdata" + + "github.com/Autumn-27/artex/agent" + "github.com/Autumn-27/artex/db" +) + +type scheduleWindow struct { + active bool + complete bool + key string + nextStart *time.Time + nextEnd *time.Time +} + +func scheduleLocation(mode string) *time.Location { + if mode == "system" { + return time.Local + } + loc, _ := time.LoadLocation("Asia/Shanghai") + return loc +} + +func scheduleClock(day time.Time, clock string, loc *time.Location) time.Time { + parts := strings.Split(clock, ":") + h, _ := strconv.Atoi(parts[0]) + m, _ := strconv.Atoi(parts[1]) + return time.Date(day.Year(), day.Month(), day.Day(), h, m, 0, 0, loc) +} + +// end before start means next-day end. Equal midnight means a full day; +// 24:00 is accepted for the end and normalised by PostgreSQL. +func windowBounds(day time.Time, s *db.TaskSchedule, loc *time.Location) (time.Time, *time.Time) { + start := scheduleClock(day, s.StartTime, loc) + if s.EndTime == "" { + return start, nil + } + end := scheduleClock(day, s.EndTime, loc) + if !end.After(start) { + end = scheduleClock(day.AddDate(0, 0, 1), s.EndTime, loc) + } + return start, &end +} + +func evaluateScheduleWindow(s *db.TaskSchedule, now time.Time) scheduleWindow { + loc := scheduleLocation(s.TimezoneMode) + now = now.In(loc) + if s.ScheduleType == "once" { + day, err := time.ParseInLocation("2006-01-02", s.RunDate, loc) + if err != nil { + return scheduleWindow{} + } + start, end := windowBounds(day, s, loc) + active := !now.Before(start) && (end == nil || now.Before(*end)) + complete := end != nil && !now.Before(*end) + // An open-ended one-shot remains active after its start until manually + // paused; there is no end boundary at which it could be considered missed. + w := scheduleWindow{active: active, complete: complete, key: "once:" + s.RunDate, nextEnd: end} + if now.Before(start) { + w.key = "" + w.nextStart = &start + } + return w + } + // Once an unbounded weekly plan has started, it remains active. + if s.EndTime == "" && strings.HasPrefix(s.LastWindowKey, "active:") { + return scheduleWindow{active: true, key: s.LastWindowKey} + } + days := map[int]bool{} + for _, d := range s.Weekdays { + days[d] = true + } + w := scheduleWindow{} + for offset := -1; offset <= 7; offset++ { + day := now.AddDate(0, 0, offset) + weekday := int(day.Weekday()) + if weekday == 0 { + weekday = 7 + } + if !days[weekday] { + continue + } + start, end := windowBounds(day, s, loc) + if !now.Before(start) && (end == nil || now.Before(*end)) { + w.active = true + w.key = "active:" + start.Format("2006-01-02T15:04") + w.nextEnd = end + // Do not pause between adjacent windows (e.g. Saturday + Sunday full days). + continue + } + if start.After(now) && (w.nextStart == nil || start.Before(*w.nextStart)) { + w.nextStart = &start + if !w.active { + w.nextEnd = end + } + } + } + return w +} + +func (sc *Scheduler) fireTaskSchedules() { + sc.s.scheduleMu.Lock() + defer sc.s.scheduleMu.Unlock() + items, err := sc.pg.ListTaskSchedules() + if err != nil { + return + } + now := time.Now() + open := map[int64]bool{} + windows := map[int64]scheduleWindow{} + for _, item := range items { + w := evaluateScheduleWindow(item, now) + // A service restart must not replay a window that started while the + // service was down. Newly-created schedules are allowed to enter an + // already-open window because their creation is an explicit user action. + if w.active && item.LastWindowKey == "" && item.CreatedAt.Before(sc.startedAt) { + w.active = false + w.key = "missed:" + w.key + } + if strings.HasPrefix(item.LastWindowKey, "missed:") && item.LastWindowKey == "missed:"+w.key { + w.active = false + w.key = item.LastWindowKey + } + if item.LastWindowKey == "manual" && !w.active { + w.active = true + w.key = "manual" + w.complete = false + } + if !item.Enabled || item.Status == "completed" { + w.active = false + } + windows[item.ID] = w + if w.active { + for _, id := range item.TaskIDs { + open[id] = true + } + } + } + // Evaluate the union first, then control each task at most once per tick. + seen := map[int64]bool{} + for _, item := range items { + w := windows[item.ID] + errors := []string{} + for _, id := range item.TaskIDs { + if seen[id] { + continue + } + seen[id] = true + t, ok := sc.s.m.Task(strconv.FormatInt(id, 10)) + if !ok { + continue + } + state := t.lifecycleSnapshot() + if db.IsTerminal(state.Status) { + continue + } + action := "" + if open[id] && state.Paused { + action = "resume" + } + if !open[id] && !state.Paused { + action = "pause" + } + if action == "" { + continue + } + // Explicit user pauses are never automatically lifted. + if action == "resume" { + var managed bool + if err := sc.pg.QueryRow(`SELECT schedule_paused FROM tasks WHERE id=$1`, id).Scan(&managed); err != nil || !managed { + continue + } + } + _, err := sc.s.applyTaskControlWithCause(t, action, agent.AbortPausedBySchedule) + message := "计划窗口" + if err != nil { + message = err.Error() + errors = append(errors, fmt.Sprintf("任务 #%d: %s", id, message)) + } + _ = sc.pg.RecordScheduleHistory(item.ID, id, action, err == nil, message) + } + status := "outside" + if !item.Enabled { + status = "paused" + } else if w.active { + status = "running" + } else if w.complete { + status = "completed" + } else if item.LastWindowKey == "" { + status = "waiting" + } + if len(errors) > 0 { + status = "error" + } + key := item.LastWindowKey + if w.active { + key = w.key + } else if strings.HasPrefix(w.key, "missed:") { + key = w.key + } + if item.Status != status || key != item.LastWindowKey || len(errors) > 0 { + if err := sc.pg.TouchTaskSchedule(item.ID, status, key, strings.Join(errors, "; ")); err != nil { + continue + } + _ = sc.pg.RecordScheduleHistory(item.ID, 0, status, len(errors) == 0, strings.Join(errors, "; ")) + } + } +} diff --git a/server/calendar_schedule_test.go b/server/calendar_schedule_test.go new file mode 100644 index 0000000..d51ba4d --- /dev/null +++ b/server/calendar_schedule_test.go @@ -0,0 +1,32 @@ +package server + +import ( + "testing" + "time" + + "github.com/Autumn-27/artex/db" +) + +func TestEvaluateScheduleWindowWeeklyCrossMidnight(t *testing.T) { + s := &db.TaskSchedule{ScheduleType: "weekly", TimezoneMode: "beijing", Weekdays: []int{1, 2, 3, 4, 5}, StartTime: "18:00:00", EndTime: "09:00:00"} + // Tuesday 08:00 is still inside Monday's overnight window. + w := evaluateScheduleWindow(s, time.Date(2026, 10, 6, 8, 0, 0, 0, time.FixedZone("CST", 8*3600))) + if !w.active { t.Fatal("expected overnight window to be active") } + // Saturday 10:00 is outside the work-week window. + w = evaluateScheduleWindow(s, time.Date(2026, 10, 10, 10, 0, 0, 0, time.FixedZone("CST", 8*3600))) + if w.active { t.Fatal("expected weekend to be outside") } +} + +func TestEvaluateScheduleWindowOnce(t *testing.T) { + s := &db.TaskSchedule{ScheduleType: "once", TimezoneMode: "beijing", RunDate: "2026-10-09", StartTime: "18:00:00", EndTime: "23:00:00"} + loc := time.FixedZone("CST", 8*3600) + if w := evaluateScheduleWindow(s, time.Date(2026, 10, 9, 17, 59, 0, 0, loc)); w.active { t.Fatal("one-shot fired early") } + if w := evaluateScheduleWindow(s, time.Date(2026, 10, 9, 19, 0, 0, 0, loc)); !w.active { t.Fatal("one-shot should be active") } + if w := evaluateScheduleWindow(s, time.Date(2026, 10, 10, 0, 0, 0, 0, loc)); !w.complete { t.Fatal("one-shot should be complete") } +} + +func TestEvaluateScheduleWindowOpenEnded(t *testing.T) { + s := &db.TaskSchedule{ScheduleType: "weekly", TimezoneMode: "beijing", Weekdays: []int{6}, StartTime: "09:00:00"} + loc := time.FixedZone("CST", 8*3600) + if w := evaluateScheduleWindow(s, time.Date(2026, 10, 10, 10, 0, 0, 0, loc)); !w.active { t.Fatal("open-ended weekly schedule should stay active") } +} diff --git a/server/engine.go b/server/engine.go index 13aca38..b5643b2 100644 --- a/server/engine.go +++ b/server/engine.go @@ -1217,7 +1217,7 @@ func taskExecutionPaused(cause error) bool { return false } switch abort.Code { - case "paused_by_user", "paused_by_orchestrator", "paused_on_reload", "paused_race_guard", + case "paused_by_schedule", "paused_by_user", "paused_by_orchestrator", "paused_on_reload", "paused_race_guard", "queued_for_admission", "llm_unavailable_queued", "task_deleted": return true default: diff --git a/server/goals.go b/server/goals.go index e5ded68..513502e 100644 --- a/server/goals.go +++ b/server/goals.go @@ -107,7 +107,10 @@ func (s *Server) admitPausedTask(t *Task) (queued bool, err error) { return s.admitTaskWhen(t, "resume", true) } -func (s *Server) admitTaskWhen(t *Task, mode string, requirePaused bool) (queued bool, err error) { +func (s *Server) admitTaskWhen(t *Task, mode string, requirePaused bool) (bool, error) { + return s.admitTaskWhenScheduled(t, mode, requirePaused, false) +} +func (s *Server) admitTaskWhenScheduled(t *Task, mode string, requirePaused, scheduled bool) (queued bool, err error) { if t == nil { return false, fmt.Errorf("task not found") } @@ -128,6 +131,15 @@ func (s *Server) admitTaskWhen(t *Task, mode string, requirePaused bool) (queued } defer s.engine.decInflight(t.ID) lifecycle := t.lifecycleSnapshot() + if scheduled { + var managed bool + if err := s.m.pg.QueryRow(`SELECT schedule_paused FROM tasks WHERE id=$1`, t.ID).Scan(&managed); err != nil { + return false, err + } + if !managed { + return false, fmt.Errorf("人工暂停优先，计划不能恢复") + } + } if requirePaused { if isTerminalStatus(lifecycle.Status) { return false, fmt.Errorf("终态任务不能执行继续") diff --git a/server/manager.go b/server/manager.go index 6e6605c..ee9a3d0 100644 --- a/server/manager.go +++ b/server/manager.go @@ -1169,7 +1169,7 @@ func (m *Manager) ApplyTaskAdmission(id, expectedStatus, status string, queued b WHEN $2 IN ('done','failed','timeout') THEN COALESCE(completed_at, now()) ELSE NULL END, - paused=false, + paused=false, schedule_paused=false, queued=$3, queued_at=CASE WHEN NOT $3 THEN NULL @@ -1232,7 +1232,9 @@ func (m *Manager) ApplyTaskAdmission(id, expectedStatus, status string, queued b // the user pause. queue_mode is intentionally retained so resuming a never-run // bootstrap task still performs goal decomposition, but the next enqueue receives // a new queued_at timestamp and therefore moves to the FIFO tail. -func (m *Manager) ApplyTaskPause(id string) error { +func (m *Manager) ApplyTaskPause(id string) error { return m.ApplyTaskPauseOrigin(id, false) } + +func (m *Manager) ApplyTaskPauseOrigin(id string, scheduled bool) error { m.taskStateMu.Lock() defer m.taskStateMu.Unlock() n, err := strconv.ParseInt(id, 10, 64) @@ -1241,10 +1243,10 @@ func (m *Manager) ApplyTaskPause(id string) error { } var mode string err = m.pg.QueryRow(`UPDATE tasks - SET paused=true, queued=false, queued_at=NULL + SET paused=true, queued=false, queued_at=NULL, schedule_paused=$2 WHERE id=$1 AND deleted_at IS NULL AND paused=false AND status NOT IN ('done','failed','timeout') - RETURNING COALESCE(queue_mode,'')`, n).Scan(&mode) + RETURNING COALESCE(queue_mode,'')`, n, scheduled).Scan(&mode) if errors.Is(err, sql.ErrNoRows) { return fmt.Errorf("task %s is unavailable for pause", id) } diff --git a/server/scheduler.go b/server/scheduler.go index 519ed80..9c7eccf 100644 --- a/server/scheduler.go +++ b/server/scheduler.go @@ -20,9 +20,10 @@ import ( // + finding watermark + fired-goal set) so a restart resumes without double-firing. // Triggers only attach to CUSTOM agents. type Scheduler struct { - s *Server - pg *db.DB - tick time.Duration + s *Server + pg *db.DB + tick time.Duration + startedAt time.Time } const ( @@ -34,7 +35,7 @@ const ( ) func newScheduler(s *Server) *Scheduler { - return &Scheduler{s: s, pg: s.m.pg, tick: 5 * time.Second} + return &Scheduler{s: s, pg: s.m.pg, tick: 5 * time.Second, startedAt: time.Now()} } // Run loops until ctx is done, ticking the scheduler. Started once from server New. @@ -115,6 +116,7 @@ func (sc *Scheduler) init() { } func (sc *Scheduler) step() { + sc.fireTaskSchedules() triggers, err := sc.pg.ListEnabledTriggers() if err != nil { return diff --git a/server/schedules.go b/server/schedules.go new file mode 100644 index 0000000..9f53587 --- /dev/null +++ b/server/schedules.go @@ -0,0 +1,225 @@ +package server + +import ( + "encoding/json" + "fmt" + "net/http" + "strconv" + "strings" + "time" + + "github.com/Autumn-27/artex/agent" + "github.com/Autumn-27/artex/db" +) + +type scheduleRequest struct { + Name string `json:"name"` + Enabled *bool `json:"enabled,omitempty"` + TimezoneMode string `json:"timezone_mode"` + ScheduleType string `json:"schedule_type"` + RunDate string `json:"run_date"` + Weekdays []int `json:"weekdays"` + StartTime string `json:"start_time"` + EndTime string `json:"end_time"` + StartImmediately bool `json:"start_immediately"` + TaskIDs []int64 `json:"task_ids"` +} + +func (s *Server) scheduleInput(req scheduleRequest, existing *db.TaskSchedule) db.TaskScheduleInput { + enabled := true + if req.Enabled != nil { + enabled = *req.Enabled + } else if existing != nil { + enabled = existing.Enabled + } + return db.TaskScheduleInput{Name: req.Name, Enabled: enabled, TimezoneMode: req.TimezoneMode, ScheduleType: req.ScheduleType, + RunDate: req.RunDate, Weekdays: req.Weekdays, StartTime: req.StartTime, EndTime: req.EndTime, + StartImmediately: req.StartImmediately, TaskIDs: req.TaskIDs} +} + +func parseScheduleID(r *http.Request) (int64, error) { + return strconv.ParseInt(r.PathValue("id"), 10, 64) +} + +func scheduleDTO(item *db.TaskSchedule) map[string]any { + raw, _ := json.Marshal(item) + var dto map[string]any + _ = json.Unmarshal(raw, &dto) + window := evaluateScheduleWindow(item, time.Now()) + dto["timezone"] = scheduleLocation(item.TimezoneMode).String() + dto["server_time"] = time.Now() + if item.Enabled && item.Status != "completed" { + dto["next_start"] = window.nextStart + dto["next_end"] = window.nextEnd + } + return dto +} +func (s *Server) listSchedules(w http.ResponseWriter, r *http.Request) { + items, err := s.m.pg.ListTaskSchedules() + if err != nil { + writeErr(w, 500, err.Error()) + return + } + dtos := []map[string]any{} + for _, item := range items { + dtos = append(dtos, scheduleDTO(item)) + } + writeJSON(w, 200, map[string]any{"schedules": dtos}) +} + +func (s *Server) getSchedule(w http.ResponseWriter, r *http.Request) { + id, err := parseScheduleID(r) + if err != nil { + writeErr(w, 400, "bad schedule id") + return + } + item, err := s.m.pg.GetTaskSchedule(id) + if err != nil { + writeErr(w, 404, "计划不存在") + return + } + history, err := s.m.pg.ListScheduleHistory(id) + if err != nil { + writeErr(w, 500, err.Error()) + return + } + dto := scheduleDTO(item) + dto["history"] = history + writeJSON(w, 200, dto) +} + +func decodeSchedule(r *http.Request) (scheduleRequest, error) { + var req scheduleRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + return req, fmt.Errorf("bad json: %w", err) + } + if req.Enabled == nil { + v := true + req.Enabled = &v + } + return req, nil +} + +func (s *Server) createSchedule(w http.ResponseWriter, r *http.Request) { + req, err := decodeSchedule(r) + if err != nil { + writeErr(w, 400, err.Error()) + return + } + item, err := s.m.pg.CreateTaskSchedule(s.scheduleInput(req, nil)) + if err != nil { + writeErr(w, 400, err.Error()) + return + } + if req.StartImmediately && item.Enabled { + s.runScheduleTasks(item) + } + writeJSON(w, 201, item) +} + +func (s *Server) updateSchedule(w http.ResponseWriter, r *http.Request) { + id, err := parseScheduleID(r) + if err != nil { + writeErr(w, 400, "bad schedule id") + return + } + old, err := s.m.pg.GetTaskSchedule(id) + if err != nil { + writeErr(w, 404, "计划不存在") + return + } + req, err := decodeSchedule(r) + if err != nil { + writeErr(w, 400, err.Error()) + return + } + item, err := s.m.pg.UpdateTaskSchedule(id, s.scheduleInput(req, old)) + if err != nil { + writeErr(w, 400, err.Error()) + return + } + writeJSON(w, 200, item) +} + +func (s *Server) deleteSchedule(w http.ResponseWriter, r *http.Request) { + id, err := parseScheduleID(r) + if err != nil { + writeErr(w, 400, "bad schedule id") + return + } + if err := s.m.pg.DeleteTaskSchedule(id); err != nil { + writeErr(w, 500, err.Error()) + return + } + writeJSON(w, 200, map[string]any{"ok": true}) +} + +func (s *Server) setScheduleEnabled(w http.ResponseWriter, r *http.Request, enabled bool) { + id, err := parseScheduleID(r) + if err != nil { + writeErr(w, 400, "bad schedule id") + return + } + if err := s.m.pg.SetTaskScheduleEnabled(id, enabled); err != nil { + writeErr(w, 500, err.Error()) + return + } + if enabled { + if item, err := s.m.pg.GetTaskSchedule(id); err == nil { + if item.StartImmediately { + s.runScheduleTasks(item) + } + } + } + writeJSON(w, 200, map[string]any{"ok": true, "enabled": enabled}) +} + +func (s *Server) pauseSchedule(w http.ResponseWriter, r *http.Request) { + s.setScheduleEnabled(w, r, false) +} +func (s *Server) resumeSchedule(w http.ResponseWriter, r *http.Request) { + s.setScheduleEnabled(w, r, true) +} + +func (s *Server) runScheduleNow(w http.ResponseWriter, r *http.Request) { + id, err := parseScheduleID(r) + if err != nil { + writeErr(w, 400, "bad schedule id") + return + } + item, err := s.m.pg.GetTaskSchedule(id) + if err != nil { + writeErr(w, 404, "计划不存在") + return + } + s.runScheduleTasks(item) + writeJSON(w, 200, map[string]any{"ok": true}) +} + +func (s *Server) runScheduleTasks(item *db.TaskSchedule) { + for _, taskID := range item.TaskIDs { + t, ok := s.m.Task(strconv.FormatInt(taskID, 10)) + if !ok { + continue + } + state := t.lifecycleSnapshot() + if db.IsTerminal(state.Status) || !state.Paused { + continue + } + _, _ = s.applyTaskControlWithCause(t, "resume", agent.AbortPausedByOrchestrator) + _ = s.m.pg.SetSchedulePaused(item.ID, taskID, false) + } +} + +// scheduleRequestFromTask is used by the task creation handler without making +// the public task API depend on database internals. +func scheduleRequestFromTask(raw json.RawMessage) (*scheduleRequest, error) { + if len(strings.TrimSpace(string(raw))) == 0 || string(raw) == "null" { + return nil, nil + } + var req scheduleRequest + if err := json.Unmarshal(raw, &req); err != nil { + return nil, err + } + return &req, nil +} diff --git a/server/server.go b/server/server.go index 34b4fb7..313d50c 100644 --- a/server/server.go +++ b/server/server.go @@ -39,9 +39,10 @@ var BuildVersion = "dev" // Server exposes the ARTEX backend over a JSON HTTP API for the shadcn/ui // frontend. type Server struct { - m *Manager - engine *Engine - ctx context.Context + scheduleMu sync.Mutex + m *Manager + engine *Engine + ctx context.Context skillDir string // root directory for skill subdirectories jwtKey []byte // HS256 signing key loaded from / generated into dataDir/jwt.key @@ -234,8 +235,8 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi log.Printf("[retester] seed: %v", err) } go s.evidenceStore().RunGC(s.ctx) - s.seedPythonInterpreter() // 自定义脚本工具:开机检测 python 解释器入库(仅空时) - go newScheduler(s).Run(s.ctx) // P3 触发器调度(定时/finding/目标事件),仅自定义 agent + s.seedPythonInterpreter() // 自定义脚本工具:开机检测 python 解释器入库(仅空时) + // Calendar scheduler starts after persisted tasks are restored. // 漏洞 IM 推送投递引擎。与 Scheduler 并列但独立：推送的实时性要求(3s) // 与触发器的业务节奏不同，且两者失败互不牵连——推送卡住不该影响 agent 触发。 go newNotifier(s).Run(s.ctx) @@ -289,6 +290,7 @@ func (s *Server) restoreTaskRuntimes() { s.engine.startDeadlineCoordinator(s.ctx, t) } } + go newScheduler(s).Run(s.ctx) // Restore every task that had already been admitted before shutdown. Starting // only the active UI task left other non-queued tasks counted as concurrency // occupants without live loops, which could permanently block the persistent @@ -686,6 +688,14 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("GET /api/tasks", s.listTasks) mux.HandleFunc("POST /api/tasks", s.createTask) + mux.HandleFunc("GET /api/schedules", s.listSchedules) + mux.HandleFunc("POST /api/schedules", s.createSchedule) + mux.HandleFunc("GET /api/schedules/{id}", s.getSchedule) + mux.HandleFunc("PATCH /api/schedules/{id}", s.updateSchedule) + mux.HandleFunc("DELETE /api/schedules/{id}", s.deleteSchedule) + mux.HandleFunc("POST /api/schedules/{id}/pause", s.pauseSchedule) + mux.HandleFunc("POST /api/schedules/{id}/resume", s.resumeSchedule) + mux.HandleFunc("POST /api/schedules/{id}/run-now", s.runScheduleNow) mux.HandleFunc("GET /api/task-categories", s.pgListTaskCategories) mux.HandleFunc("POST /api/task-categories", s.pgCreateTaskCategory) mux.HandleFunc("PATCH /api/task-categories/{id}", s.pgRenameTaskCategory) @@ -1469,18 +1479,19 @@ func truncateReply(s string) string { } type createTaskReq struct { - Name string `json:"name,omitempty"` // 可选任务名称;省略/空=未命名 - CategoryID *int64 `json:"category_id,omitempty"` - Description string `json:"description"` - Goal string `json:"goal"` - LLMProfileID *int64 `json:"llm_profile_id,omitempty"` // 指定运行本任务的 LLM 配置;省略/null=用激活配置 - LLMProfileIDs []int64 `json:"llm_profile_ids,omitempty"` // 有序任务级配置链;第一项初始生效 - SourceTaskIDs []string `json:"source_task_ids,omitempty"` // 仅直接、只读继承的来源任务 - CompanyIDs []int64 `json:"company_ids,omitempty"` // 关联企业范围并快照关联当前企业资产;不复制资产或强制生成意图 - TimeoutSeconds int `json:"timeout_seconds"` // 任务级超时(秒);0/省略=不限时 - PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"` // planner 心跳触发间隔(秒);0/省略=默认600(10min);下限=默认=600,低于自动抬到600 - SeedFirstIntent *bool `json:"seed_first_intent,omitempty"` // 创建时直接下发一条种子意图(内容=描述+目标),让 worker 免等首轮 planner 直接开跑;省略/null=默认关闭,走标准先规划再执行。显式传 true 才开(CTF 常一 work 解决时可省掉开跑前的 planner 轮)。 - CoverageEnabled *bool `json:"coverage_enabled,omitempty"` // 资产覆盖度功能;省略/null=默认开(true)。false=关闭覆盖度计算/展示/自动累积范围+隐藏 add_task_scope/list_untested_assets。company 关联不受影响。 + Schedule *scheduleRequest `json:"schedule,omitempty"` + Name string `json:"name,omitempty"` // 可选任务名称;省略/空=未命名 + CategoryID *int64 `json:"category_id,omitempty"` + Description string `json:"description"` + Goal string `json:"goal"` + LLMProfileID *int64 `json:"llm_profile_id,omitempty"` // 指定运行本任务的 LLM 配置;省略/null=用激活配置 + LLMProfileIDs []int64 `json:"llm_profile_ids,omitempty"` // 有序任务级配置链;第一项初始生效 + SourceTaskIDs []string `json:"source_task_ids,omitempty"` // 仅直接、只读继承的来源任务 + CompanyIDs []int64 `json:"company_ids,omitempty"` // 关联企业范围并快照关联当前企业资产;不复制资产或强制生成意图 + TimeoutSeconds int `json:"timeout_seconds"` // 任务级超时(秒);0/省略=不限时 + PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"` // planner 心跳触发间隔(秒);0/省略=默认600(10min);下限=默认=600,低于自动抬到600 + SeedFirstIntent *bool `json:"seed_first_intent,omitempty"` // 创建时直接下发一条种子意图(内容=描述+目标),让 worker 免等首轮 planner 直接开跑;省略/null=默认关闭,走标准先规划再执行。显式传 true 才开(CTF 常一 work 解决时可省掉开跑前的 planner 轮)。 + CoverageEnabled *bool `json:"coverage_enabled,omitempty"` // 资产覆盖度功能;省略/null=默认开(true)。false=关闭覆盖度计算/展示/自动累积范围+隐藏 add_task_scope/list_untested_assets。company 关联不受影响。 // InterceptRules 任务级资产拦截/允许规则(创建时录入,存 task_intercept_rules,不进全局表)。 InterceptRules []taskInterceptRuleReq `json:"intercept_rules,omitempty"` } @@ -1534,8 +1545,19 @@ func (s *Server) createTask(w http.ResponseWriter, r *http.Request) { writeErr(w, 400, "任务级拦截规则无效："+err.Error()) return } + var schedule *db.TaskScheduleInput + if req.Schedule != nil { + input := s.scheduleInput(*req.Schedule, nil) + input.TaskIDs = nil + if err := db.ValidateTaskScheduleInput(input); err != nil { + writeErr(w, 400, err.Error()) + return + } + schedule = &input + } t, err := s.m.CreateTaskWithOptions(req.Description, req.Goal, db.TaskCreateOptions{ - Name: strings.TrimSpace(req.Name), CategoryID: req.CategoryID, + Schedule: schedule, + Name: strings.TrimSpace(req.Name), CategoryID: req.CategoryID, SourceTaskIDs: sourceIDs, CompanyIDs: req.CompanyIDs, LLMProfileIDs: req.LLMProfileIDs, TimeoutSeconds: req.TimeoutSeconds, PlanHeartbeatSeconds: req.PlanHeartbeatSeconds, CoverageEnabled: req.CoverageEnabled, @@ -1556,7 +1578,11 @@ func (s *Server) createTask(w http.ResponseWriter, r *http.Request) { log.Printf("[task] 新建任务 #%s «%s» 目标: %s", t.ID, req.Description, req.Goal) // 共享的建后流程(seed + 种子意图 + 后台目标分解 + engine.Run),与 spawn_task 复用同一段。 // launchTask 内部异步,不阻塞 UI —— 目标分解在后台可见地进行。 - s.launchTask(t, req.Description+" "+req.Goal, req.SeedFirstIntent != nil && *req.SeedFirstIntent) + if schedule == nil || schedule.StartImmediately { + s.launchTask(t, req.Description+" "+req.Goal, req.SeedFirstIntent != nil && *req.SeedFirstIntent) + } else { + s.engine.Pause(t.ID, agent.AbortPausedBySchedule) + } writeJSON(w, 201, taskDTO(t, s.resolvedTaskStatus(t))) } diff --git a/server/task_control.go b/server/task_control.go index ea479e1..1cc0e15 100644 --- a/server/task_control.go +++ b/server/task_control.go @@ -115,6 +115,14 @@ func (s *Server) applyTaskControlWithCause(t *Task, action string, pauseCause er return out, fmt.Errorf("终态任务不能执行暂停") } if lifecycle.Paused { + if pauseCause != agent.AbortPausedBySchedule { + if _, err := s.m.pg.Exec(`UPDATE tasks SET schedule_paused=false WHERE id=$1`, t.ID); err != nil { + return out, err + } + out.Paused = true + out.Status = "paused" + return out, nil + } return out, fmt.Errorf("任务已经暂停") } wasQueued := lifecycle.Queued @@ -123,7 +131,7 @@ func (s *Server) applyTaskControlWithCause(t *Task, action string, pauseCause er pauseCause = agent.AbortPausedByUser } s.engine.Pause(t.ID, pauseCause) - if err := s.m.ApplyTaskPause(t.ID); err != nil { + if err := s.m.ApplyTaskPauseOrigin(t.ID, pauseCause == agent.AbortPausedBySchedule); err != nil { if !wasEnginePaused && !wasQueued { s.engine.Resume(t) } @@ -136,7 +144,7 @@ func (s *Server) applyTaskControlWithCause(t *Task, action string, pauseCause er out.Paused, out.Status = true, "paused" go s.reconcileConcurrency() case "resume": - queued, err := s.admitPausedTask(t) + queued, err := s.admitTaskWhenScheduled(t, "resume", true, pauseCause == agent.AbortPausedBySchedule) if err != nil { return out, err } diff --git a/web/src/app/(main)/function/schedules/page.tsx b/web/src/app/(main)/function/schedules/page.tsx new file mode 100644 index 0000000..49177e8 --- /dev/null +++ b/web/src/app/(main)/function/schedules/page.tsx @@ -0,0 +1,419 @@ +"use client"; + +import * as React from "react"; + +import { CalendarClock, Loader2, Pause, Play, Plus, Search, Trash2, Zap } from "lucide-react"; +import { toast } from "sonner"; + +import { TimeSelect } from "@/components/time-select"; +import { Badge } from "@/components/ui/badge"; +import { Button } from "@/components/ui/button"; +import { Card, CardContent, CardDescription, CardHeader, CardTitle } from "@/components/ui/card"; +import { Checkbox } from "@/components/ui/checkbox"; +import { Input } from "@/components/ui/input"; +import { Label } from "@/components/ui/label"; +import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select"; +import { api, type ScheduleInput } from "@/lib/api"; +import type { Task, TaskSchedule } from "@/lib/types"; + +const weekdays = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]; + +const blank: ScheduleInput = { + name: "", + schedule_type: "weekly", + weekdays: [6, 7], + start_time: "18:00", + end_time: "", + task_ids: [], + timezone_mode: "beijing", + start_immediately: false, +}; + +function statusText(s: TaskSchedule) { + return ( + ( + { + waiting: "等待窗口", + running: "窗口运行中", + outside: "窗口外", + completed: "已完成", + error: "错误", + paused: "已暂停", + } as Record + )[s.status] ?? s.status + ); +} + +function formatTime(value?: string) { + return value ? new Date(value).toLocaleString("zh-CN") : "—"; +} + +function statusVariant(status: TaskSchedule["status"]): "default" | "destructive" | "secondary" { + if (status === "running") return "default"; + if (status === "error") return "destructive"; + return "secondary"; +} + +function ScheduleForm({ + value, + tasks, + onSave, + onCancel, + saving, +}: { + value: ScheduleInput; + tasks: Task[]; + onSave: (v: ScheduleInput) => void; + onCancel?: () => void; + saving: boolean; +}) { + const [draft, setDraft] = React.useState(value); + const [taskQuery, setTaskQuery] = React.useState(""); + const set = (patch: Partial) => setDraft((v) => ({ ...v, ...patch })); + const toggleTask = (id: number, checked: boolean) => + set({ task_ids: checked ? [...(draft.task_ids ?? []), id] : (draft.task_ids ?? []).filter((x) => x !== id) }); + const toggleDay = (day: number) => + set({ + weekdays: (draft.weekdays ?? []).includes(day) + ? (draft.weekdays ?? []).filter((x) => x !== day) + : [...(draft.weekdays ?? []), day].sort(), + }); + const filteredTasks = React.useMemo(() => { + const query = taskQuery.trim().toLowerCase(); + if (!query) return tasks; + return tasks.filter((task) => + [String(task.id), task.name ?? "", task.description, task.goal].some((value) => + value.toLowerCase().includes(query), + ), + ); + }, [taskQuery, tasks]); + let taskListContent: React.ReactNode; + if (tasks.length === 0) { + taskListContent = 暂无任务; + } else if (filteredTasks.length === 0) { + taskListContent = 没有匹配的任务; + } else { + taskListContent = filteredTasks.map((task) => ( +
+ toggleTask(Number(task.id), !!v)} + /> + +
+ )); + } + return ( +
+
+ + set({ name: e.target.value })} placeholder="例如：周末持续扫描" /> +
+
+
+ + +
+
+ + +
+
+ {draft.schedule_type === "once" ? ( +
+ + set({ run_date: e.target.value })} /> +
+ ) : ( +
+ +
+ {weekdays.map((label, index) => ( +
+ toggleDay(index + 1)} + /> + +
+ ))} +
+
+ )} +
+
+ + set({ start_time: value })} + aria-label="开始时间" + /> +
+
+ + set({ end_time: value })} + optional + aria-label="结束时间，可选" + /> +
+
+

+ 结束时间为空表示开始后持续运行；结束时间早于开始时间表示跨午夜窗口。 +

+
+ set({ start_immediately: !!v })} + /> + +
+
+
+ + 已选 {(draft.task_ids ?? []).length} 个 +
+
+ + setTaskQuery(event.target.value)} + placeholder="检索任务名称、描述或 ID" + className="h-9 pl-8" + /> +
+
{taskListContent}
+
+
+ + +
+
+ ); +} + +export default function SchedulesPage() { + const [items, setItems] = React.useState([]); + const [tasks, setTasks] = React.useState([]); + const [loading, setLoading] = React.useState(true); + const [editing, setEditing] = React.useState<{ id?: number; value: ScheduleInput } | null>(null); + const [saving, setSaving] = React.useState(false); + const load = React.useCallback(async () => { + setLoading(true); + try { + const [s, t] = await Promise.all([api.schedules(), api.tasks()]); + setItems(s); + setTasks(t.tasks); + } catch (e) { + toast.error((e as Error).message); + } finally { + setLoading(false); + } + }, []); + React.useEffect(() => { + void load(); + }, [load]); + const save = async (value: ScheduleInput) => { + if ( + !value.name.trim() || + !value.start_time || + (value.schedule_type === "once" ? !value.run_date : !value.weekdays?.length) + ) { + toast.error("请补全计划名称、日期/星期和开始时间"); + return; + } + setSaving(true); + try { + if (editing?.id) await api.updateSchedule(editing.id, value); + else await api.createSchedule(value); + toast.success("计划已保存"); + setEditing(null); + await load(); + } catch (e) { + toast.error((e as Error).message); + } finally { + setSaving(false); + } + }; + const remove = async (id: number) => { + if (!window.confirm("确定删除这个计划吗？")) return; + try { + await api.deleteSchedule(id); + await load(); + } catch (e) { + toast.error((e as Error).message); + } + }; + const action = async (id: number, type: "pause" | "resume" | "run") => { + try { + if (type === "pause") await api.pauseSchedule(id); + else if (type === "resume") await api.resumeSchedule(id); + else await api.runScheduleNow(id); + await load(); + } catch (e) { + toast.error((e as Error).message); + } + }; + let scheduleContent: React.ReactNode; + if (loading) { + scheduleContent = ( +
+ + 加载计划中… +
+ ); + } else if (items.length === 0) { + scheduleContent = ( + + + 暂无计划任务，点击“新建计划”开始。 + + + ); + } else { + scheduleContent = ( +
+ {items.map((item) => ( + + +
+
+ {item.name} + + {item.schedule_type === "once" + ? item.run_date + : (item.weekdays ?? []).map((d) => weekdays[d - 1]).join("、")} + +
+ {statusText(item)} +
+ + +
+ {item.start_time.slice(0, 5)} + {item.end_time ? ` - ${item.end_time.slice(0, 5)}` : " 起持续运行"} ·{" "} + {item.timezone_mode === "beijing" ? "北京时间" : item.timezone} +
+
绑定任务：{item.task_ids.length ? item.task_ids.map((id) => `#${id}`).join("、") : "无"}
+
+ 最近变化：{formatTime(item.last_transition_at)} + {item.last_error ? ` · ${item.last_error}` : ""} +
+
+ + {item.enabled ? ( + + ) : ( + + )} + + +
+ + + ))} +
+ ); + } + return ( +
+
+
+

+ + 计划任务 +

+

按北京时间或系统时区控制扫描任务的开始、暂停和恢复。

+
+ +
+ {editing && ( + + + {editing.id ? "编辑计划" : "新建计划"} + 计划结束时会暂停任务，下一次有效窗口自动继续。 + + + setEditing(null)} + onSave={save} + /> + + + )} + {scheduleContent} +
+ ); +} diff --git a/web/src/app/(main)/function/tasks/page.tsx b/web/src/app/(main)/function/tasks/page.tsx index e1df143..cf8b8a4 100644 --- a/web/src/app/(main)/function/tasks/page.tsx +++ b/web/src/app/(main)/function/tasks/page.tsx @@ -117,8 +117,9 @@ import { Switch } from "@/components/ui/switch"; import { Table, TableBody, TableCell, TableHead, TableHeader, TableRow } from "@/components/ui/table"; import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs"; import { Textarea } from "@/components/ui/textarea"; +import { TimeSelect } from "@/components/time-select"; import { Tooltip, TooltipContent, TooltipTrigger } from "@/components/ui/tooltip"; -import { api } from "@/lib/api"; +import { api, type ScheduleInput } from "@/lib/api"; import { getLocalStorageValue, setLocalStorageValue } from "@/lib/local-storage.client"; import { type SortDirection, useStoredSortPreference } from "@/lib/sort-preference"; import type { @@ -3096,6 +3097,17 @@ function CreateTaskSheet({ const [seedFirstIntent, setSeedFirstIntent] = React.useState(false); // 创建时下发种子意图,worker 免等首轮 planner 直接开跑;默认关闭,走标准先规划再执行 const [coverageEnabled, setCoverageEnabled] = React.useState(true); // 资产覆盖度功能;默认开。关闭=不计算/展示覆盖度、不累积范围、隐藏范围类工具(company 关联不受影响) const [interceptRules, setInterceptRules] = React.useState([]); // 任务级资产拦截规则(仅本任务生效,不进全局表) + const [scheduleEnabled, setScheduleEnabled] = React.useState(false); + const [schedule, setSchedule] = React.useState({ + name: "新建任务计划", + schedule_type: "weekly", + weekdays: [6, 7], + start_time: "18:00", + end_time: "", + task_ids: [], + timezone_mode: "beijing", + start_immediately: false, + }); // 方式1 文件上传:建任务前把文件暂存到 drafts//uploads/,拿回绝对路径追加进描述。 const [uploading, setUploading] = React.useState(false); const [uploadCount, setUploadCount] = React.useState(0); @@ -3162,6 +3174,7 @@ function CreateTaskSheet({ seedFirstIntent, planHeartbeatSeconds: heartbeatSec, coverageEnabled, + schedule: scheduleEnabled ? { ...schedule, task_ids: [] } : undefined, interceptRules: interceptRules .map((r) => ({ ...r, pattern: r.pattern.trim() })) .filter((r) => r.pattern !== ""), @@ -3180,6 +3193,17 @@ function CreateTaskSheet({ setSeedFirstIntent(false); setCoverageEnabled(true); setInterceptRules([]); + setScheduleEnabled(false); + setSchedule({ + name: "新建任务计划", + schedule_type: "weekly", + weekdays: [6, 7], + start_time: "18:00", + end_time: "", + task_ids: [], + timezone_mode: "beijing", + start_immediately: false, + }); setUploadCount(0); draftIdRef.current = ""; setOpen(false); @@ -3248,6 +3272,110 @@ function CreateTaskSheet({ /> 可选，单个分类；用于任务列表筛选和归档，不影响 Agent 执行。 +
+ 计划启动（可选） + + {scheduleEnabled && ( + + + 计划名称 + setSchedule({ ...schedule, name: e.target.value })} /> + +
+ + 规则 + + + + 时区 + + +
+ {schedule.schedule_type === "once" ? ( + + 执行日期 + setSchedule({ ...schedule, run_date: e.target.value })} + /> + + ) : ( + + 星期（1=周一，7=周日） + + setSchedule({ + ...schedule, + weekdays: e.target.value + .split(",") + .map(Number) + .filter((n) => n >= 1 && n <= 7), + }) + } + placeholder="例如 6,7" + /> + + )} +
+ + 开始时间 + setSchedule({ ...schedule, start_time: value })} + aria-label="开始时间" + /> + + + 结束时间（可选） + setSchedule({ ...schedule, end_time: value })} + optional + aria-label="结束时间，可选" + /> + +
+ + + )} +