mirror of
https://github.com/RuoJi6/ARTEX.git
synced 2026-10-28 01:03:50 +08:00
新增日历式计划任务
This commit is contained in:
1 parent
9adbd54627
commit
3254c90a03
19 files changed
+1745
-33
No files matched your search
@@ -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, "; "))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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") }
|
||||
}
|
||||
+1
-1
@@ -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:
|
||||
|
||||
+13
-1
@@ -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("终态任务不能执行继续")
|
||||
|
||||
+6
-4
@@ -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)
|
||||
}
|
||||
|
||||
+6
-4
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
+45
-19
@@ -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)))
|
||||
}
|
||||
|
||||
|
||||
+10
-2
@@ -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
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user