mirror of
https://github.com/RuoJi6/ARTEX.git
synced 2026-11-01 10:43:49 +08:00
Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
008a9ced0d | ||
|
|
d93032fbea | ||
|
|
97a4f2b236 | ||
|
|
e878020c4d | ||
|
|
de6a0fa125 |
No files matched your search
@@ -28,6 +28,7 @@ func Causef(code, short, format string, args ...any) *AbortCause {
|
||||
var (
|
||||
// Task-level execution context.
|
||||
AbortPausedBySchedule = cause("paused_by_schedule", "计划窗口结束,暂停任务", "任务将在下一个计划窗口恢复,本次执行已停止")
|
||||
AbortPausedByProxy = cause("paused_by_proxy", "全局代理异常,暂停任务", "全局出口代理连接失败,任务已暂停;代理恢复后将自动继续")
|
||||
AbortPausedByUser = cause("paused_by_user", "用户暂停了任务",
|
||||
"用户通过任务控制接口(POST /api/tasks/{id}/control,action=pause)暂停了任务。本次 Planner/Worker 运行被主动取消;运行中的意图会退回 frontier(open),恢复任务后重新领取并从头执行")
|
||||
AbortPausedByOrchestrator = cause("paused_by_orchestrator", "编排 Agent 暂停了任务",
|
||||
|
||||
@@ -139,7 +139,7 @@ func TestTerminalTextNonAbortReasons(t *testing.T) {
|
||||
|
||||
func TestAbortCausesAreWellFormed(t *testing.T) {
|
||||
all := []*AbortCause{
|
||||
AbortPausedByUser, AbortPausedByOrchestrator, AbortTaskDeleted, AbortPausedOnReload,
|
||||
AbortPausedByUser, AbortPausedByProxy, AbortPausedByOrchestrator, AbortTaskDeleted, AbortPausedOnReload,
|
||||
AbortGoalMet, AbortSettleDrainTimeout, AbortKilledByPlanner, AbortWorkPausedByUser,
|
||||
AbortWorkCancelledByUser, AbortWorkFinished, AbortPausedRaceGuard,
|
||||
AbortChatStoppedByUser, AbortChatPausedWithTask, AbortChatTurnFinished,
|
||||
|
||||
@@ -513,6 +513,7 @@ CREATE TABLE IF NOT EXISTS tasks (
|
||||
status TEXT NOT NULL DEFAULT 'created'
|
||||
CHECK (status IN ('created','running','paused','done','failed','timeout')),
|
||||
paused BOOLEAN NOT NULL DEFAULT false,
|
||||
pause_reason TEXT NOT NULL DEFAULT '',
|
||||
queued BOOLEAN NOT NULL DEFAULT false,
|
||||
queued_at TIMESTAMPTZ,
|
||||
queue_mode TEXT NOT NULL DEFAULT '',
|
||||
@@ -566,6 +567,7 @@ ALTER TABLE tasks ADD COLUMN IF NOT EXISTS llm_chain_revision BIGINT NOT NULL DE
|
||||
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS archived_at TIMESTAMPTZ;
|
||||
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS active_run_seconds BIGINT NOT NULL DEFAULT 0;
|
||||
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS run_segment_started_at TIMESTAMPTZ;
|
||||
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS pause_reason TEXT NOT NULL DEFAULT '';
|
||||
CREATE INDEX IF NOT EXISTS idx_tasks_category ON tasks(category_id, created_at DESC)
|
||||
WHERE deleted_at IS NULL AND category_id IS NOT NULL;
|
||||
CREATE INDEX IF NOT EXISTS idx_tasks_pinned ON tasks(pinned_at DESC)
|
||||
@@ -1084,6 +1086,7 @@ CREATE TABLE IF NOT EXISTS task_schedule_tasks (
|
||||
);
|
||||
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_paused BOOLEAN NOT NULL DEFAULT false;
|
||||
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_manual_until TIMESTAMPTZ;
|
||||
UPDATE tasks SET pause_reason='paused_by_schedule' WHERE paused AND schedule_paused AND pause_reason='';
|
||||
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;
|
||||
|
||||
|
||||
+6
-3
@@ -25,6 +25,7 @@ type Task struct {
|
||||
ExplorationID int64 `json:"exploration_id"`
|
||||
Status string `json:"status"`
|
||||
Paused bool `json:"paused"`
|
||||
PauseReason string `json:"pause_reason,omitempty"`
|
||||
Queued bool `json:"queued"`
|
||||
QueuedAt *time.Time `json:"queued_at,omitempty"`
|
||||
QueueMode string `json:"queue_mode,omitempty"`
|
||||
@@ -302,7 +303,7 @@ RETURNING id, status, paused, created_at`, opts.Name, opts.Remark, opts.Category
|
||||
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 {
|
||||
if _, err = tx.Exec(`UPDATE tasks SET paused=true,pause_reason='paused_by_schedule',schedule_paused=true,queue_mode='bootstrap' WHERE id=$1`, t.ID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
t.Paused = true
|
||||
@@ -405,11 +406,11 @@ func insertTaskLLMProfiles(tx *sql.Tx, taskID int64, profileIDs []int64) error {
|
||||
|
||||
const taskCols = `id, COALESCE(name,''), remark, category_id,
|
||||
COALESCE((SELECT category.name FROM task_categories category WHERE category.id=tasks.category_id),''),
|
||||
description, goal, exploration_id, status, paused, queued, queued_at, COALESCE(queue_mode,''), llm_profile_id, active_llm_profile_id, COALESCE(parent_ref,''), pinned_at, created_at, completed_at, COALESCE(timeout_seconds,0), COALESCE(plan_heartbeat_seconds,300), COALESCE(coverage_enabled,true), COALESCE(worker_count,0), first_run_at, deadline_at, COALESCE(active_run_seconds,0), run_segment_started_at`
|
||||
description, goal, exploration_id, status, paused, COALESCE(pause_reason,''), queued, queued_at, COALESCE(queue_mode,''), llm_profile_id, active_llm_profile_id, COALESCE(parent_ref,''), pinned_at, created_at, completed_at, COALESCE(timeout_seconds,0), COALESCE(plan_heartbeat_seconds,300), COALESCE(coverage_enabled,true), COALESCE(worker_count,0), first_run_at, deadline_at, COALESCE(active_run_seconds,0), run_segment_started_at`
|
||||
|
||||
func scanTask(sc interface{ Scan(...any) error }) (*Task, error) {
|
||||
var t Task
|
||||
if err := sc.Scan(&t.ID, &t.Name, &t.Remark, &t.CategoryID, &t.CategoryName, &t.Description, &t.Goal, &t.ExplorationID, &t.Status, &t.Paused, &t.Queued, &t.QueuedAt, &t.QueueMode, &t.LLMProfileID, &t.ActiveLLMProfileID, &t.ParentRef, &t.PinnedAt, &t.CreatedAt, &t.CompletedAt, &t.TimeoutSeconds, &t.PlanHeartbeatSeconds, &t.CoverageEnabled, &t.WorkerCount, &t.FirstRunAt, &t.DeadlineAt, &t.ActiveRunSeconds, &t.RunSegmentStartedAt); err != nil {
|
||||
if err := sc.Scan(&t.ID, &t.Name, &t.Remark, &t.CategoryID, &t.CategoryName, &t.Description, &t.Goal, &t.ExplorationID, &t.Status, &t.Paused, &t.PauseReason, &t.Queued, &t.QueuedAt, &t.QueueMode, &t.LLMProfileID, &t.ActiveLLMProfileID, &t.ParentRef, &t.PinnedAt, &t.CreatedAt, &t.CompletedAt, &t.TimeoutSeconds, &t.PlanHeartbeatSeconds, &t.CoverageEnabled, &t.WorkerCount, &t.FirstRunAt, &t.DeadlineAt, &t.ActiveRunSeconds, &t.RunSegmentStartedAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
t.Pinned = t.PinnedAt != nil
|
||||
@@ -612,6 +613,7 @@ func (d *DB) SetStatusRuntime(id int64, status string) (int64, *time.Time, error
|
||||
err := d.QueryRow(`
|
||||
UPDATE tasks
|
||||
SET status = $1,
|
||||
pause_reason = CASE WHEN $1 IN ('done','failed','timeout') THEN '' ELSE pause_reason END,
|
||||
completed_at = CASE WHEN $1 IN ('done','failed','timeout') THEN COALESCE(completed_at, now()) ELSE NULL END,
|
||||
active_run_seconds = active_run_seconds + CASE WHEN $1 IN ('done','failed','timeout') AND run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
|
||||
run_segment_started_at = CASE WHEN $1 IN ('done','failed','timeout') THEN NULL ELSE run_segment_started_at END
|
||||
@@ -638,6 +640,7 @@ func (d *DB) SetTerminalStatusGuardedRuntime(id int64, status string) (active in
|
||||
err = d.QueryRow(`
|
||||
UPDATE tasks
|
||||
SET status = $1,
|
||||
pause_reason = '',
|
||||
completed_at = COALESCE(completed_at, now()),
|
||||
active_run_seconds = active_run_seconds + CASE WHEN run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
|
||||
run_segment_started_at = NULL
|
||||
|
||||
@@ -188,6 +188,43 @@ func (s *Server) taskManualScheduleEnd(taskID string, now time.Time) (*time.Time
|
||||
return until, nil
|
||||
}
|
||||
|
||||
// proxyResumeInsideSchedule reports whether a task attached to any enabled
|
||||
// calendar is currently allowed to run. Tasks without a calendar remain
|
||||
// eligible. Manual schedule releases are honored through schedule_manual_until.
|
||||
func (s *Server) proxyResumeInsideSchedule(taskID string, now time.Time) (bool, error) {
|
||||
id, err := strconv.ParseInt(taskID, 10, 64)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
items, err := s.m.pg.ListTaskSchedules()
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
attached := false
|
||||
open := false
|
||||
for _, item := range items {
|
||||
for _, boundID := range item.TaskIDs {
|
||||
if boundID != id {
|
||||
continue
|
||||
}
|
||||
attached = true
|
||||
if item.Enabled && item.Status != "completed" {
|
||||
w := evaluateScheduleWindow(item, now)
|
||||
open = open || w.active || item.LastWindowKey == "manual"
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
if !attached {
|
||||
return true, nil
|
||||
}
|
||||
var manualUntil time.Time
|
||||
if err := s.m.pg.QueryRow(`SELECT COALESCE(schedule_manual_until, 'epoch'::timestamptz) FROM tasks WHERE id=$1`, id).Scan(&manualUntil); err == nil && manualUntil.After(now) {
|
||||
open = true
|
||||
}
|
||||
return open, nil
|
||||
}
|
||||
|
||||
func (sc *Scheduler) fireTaskSchedules() {
|
||||
sc.fireTaskSchedulesAt(time.Now())
|
||||
}
|
||||
@@ -248,6 +285,15 @@ func (sc *Scheduler) fireTaskSchedulesAt(now time.Time) {
|
||||
if db.IsTerminal(state.Status) {
|
||||
continue
|
||||
}
|
||||
// A proxy-paused task can remain paused when a calendar window ends
|
||||
// during reconnect. Convert it to a schedule pause so a later probe
|
||||
// cannot resume work outside the configured window.
|
||||
if !open[id] && state.Paused && state.PauseReason == "paused_by_proxy" {
|
||||
if err := sc.s.m.ConvertProxyPauseToSchedule(strconv.FormatInt(id, 10)); err != nil {
|
||||
errors = append(errors, fmt.Sprintf("任务 #%d: %s", id, err))
|
||||
}
|
||||
continue
|
||||
}
|
||||
action := ""
|
||||
if open[id] && state.Paused {
|
||||
action = "resume"
|
||||
|
||||
@@ -447,6 +447,9 @@ func (s *Server) runHTTPTool(ctx context.Context, execRaw json.RawMessage, param
|
||||
}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
if spec.UseRecordingProxy || strings.TrimSpace(spec.Proxy) == strings.TrimSpace(s.m.GlobalProxy()) {
|
||||
s.handleProxyToolFailure(runCtx, err)
|
||||
}
|
||||
return actool.Errorf("请求失败: " + err.Error()), nil
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
+36
-19
@@ -52,6 +52,8 @@ type TaskDTO struct {
|
||||
ActiveRunSeconds int64 `json:"active_run_seconds"`
|
||||
RunSegmentStarted int64 `json:"run_segment_started_unix"`
|
||||
Paused bool `json:"paused"`
|
||||
PauseReason string `json:"pause_reason,omitempty"`
|
||||
Reconnect *ProxyReconnectDTO `json:"reconnect,omitempty"`
|
||||
Queued bool `json:"queued"`
|
||||
Tokens TokenTotalDTO `json:"tokens"` // whole-task token consumption
|
||||
GoalsTotal int `json:"goals_total"`
|
||||
@@ -70,6 +72,14 @@ type TaskDTO struct {
|
||||
WorkerCount int `json:"worker_count"` // 本任务 Worker 数;0=用全局默认
|
||||
}
|
||||
|
||||
type ProxyReconnectDTO struct {
|
||||
Attempt int `json:"attempt"`
|
||||
Max int `json:"max"`
|
||||
NextAt int64 `json:"next_at,omitempty"`
|
||||
Exhausted bool `json:"exhausted"`
|
||||
LastError string `json:"last_error,omitempty"`
|
||||
}
|
||||
|
||||
// FindingSeverityDTO 是任务列表里按严重度分档的漏洞计数(严重/高/中/低)。
|
||||
type FindingSeverityDTO struct {
|
||||
Critical int `json:"critical"`
|
||||
@@ -118,25 +128,32 @@ func taskDTO(t *Task, status string) TaskDTO {
|
||||
}
|
||||
profileIDs := append(make([]int64, 0, len(llmState.ProfileIDs)), llmState.ProfileIDs...)
|
||||
return TaskDTO{
|
||||
ID: t.ID,
|
||||
ExplorationID: t.ExpID,
|
||||
Name: lifecycle.Name,
|
||||
Remark: lifecycle.Remark,
|
||||
CategoryID: lifecycle.CategoryID,
|
||||
CategoryName: lifecycle.CategoryName,
|
||||
Pinned: lifecycle.PinnedAt > 0,
|
||||
PinnedAt: completedRFC(lifecycle.PinnedAt),
|
||||
Description: t.Description,
|
||||
Goal: t.Goal,
|
||||
Status: status,
|
||||
CreatedAt: rfc3339(time.Unix(t.CreatedAt, 0)),
|
||||
CreatedUnix: t.CreatedAt,
|
||||
RunSeconds: runSeconds,
|
||||
ActiveRunSeconds: lifecycle.ActiveRunSeconds,
|
||||
RunSegmentStarted: lifecycle.RunSegmentStartedAt,
|
||||
CompletedAt: completedRFC(lifecycle.CompletedAt),
|
||||
CompletedUnix: lifecycle.CompletedAt,
|
||||
Paused: lifecycle.Paused,
|
||||
ID: t.ID,
|
||||
ExplorationID: t.ExpID,
|
||||
Name: lifecycle.Name,
|
||||
Remark: lifecycle.Remark,
|
||||
CategoryID: lifecycle.CategoryID,
|
||||
CategoryName: lifecycle.CategoryName,
|
||||
Pinned: lifecycle.PinnedAt > 0,
|
||||
PinnedAt: completedRFC(lifecycle.PinnedAt),
|
||||
Description: t.Description,
|
||||
Goal: t.Goal,
|
||||
Status: status,
|
||||
CreatedAt: rfc3339(time.Unix(t.CreatedAt, 0)),
|
||||
CreatedUnix: t.CreatedAt,
|
||||
RunSeconds: runSeconds,
|
||||
ActiveRunSeconds: lifecycle.ActiveRunSeconds,
|
||||
RunSegmentStarted: lifecycle.RunSegmentStartedAt,
|
||||
CompletedAt: completedRFC(lifecycle.CompletedAt),
|
||||
CompletedUnix: lifecycle.CompletedAt,
|
||||
Paused: lifecycle.Paused,
|
||||
PauseReason: lifecycle.PauseReason,
|
||||
Reconnect: func() *ProxyReconnectDTO {
|
||||
if lifecycle.PauseReason != "paused_by_proxy" && lifecycle.ReconnectAttempt == 0 && !lifecycle.ReconnectExhausted {
|
||||
return nil
|
||||
}
|
||||
return &ProxyReconnectDTO{Attempt: lifecycle.ReconnectAttempt, Max: lifecycle.ReconnectMax, NextAt: lifecycle.ReconnectNextAt, Exhausted: lifecycle.ReconnectExhausted, LastError: lifecycle.ReconnectLastError}
|
||||
}(),
|
||||
Queued: lifecycle.Queued,
|
||||
LLMProfileID: llmState.ProfileID,
|
||||
LLMProfileIDs: profileIDs,
|
||||
|
||||
@@ -373,6 +373,7 @@ func (s *Server) testEgressProxy(w http.ResponseWriter, r *http.Request) {
|
||||
// 整体上限 40s:单次 Timeout 只兜一次,不兜总量;前端 fetch 没有超时。
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 40*time.Second)
|
||||
defer cancel()
|
||||
proxyGuards := s.snapshotProxyReconnectGuards(false)
|
||||
|
||||
configuredProxy := strings.TrimSpace(req.GlobalProxy)
|
||||
rep := egressReport{
|
||||
@@ -417,6 +418,12 @@ func (s *Server) testEgressProxy(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
rep.OK = judgeEgress(rep)
|
||||
rep.Warnings = egressWarnings(rep, pending, savedProxy)
|
||||
if rep.OK && !pending {
|
||||
// The settings page uses this endpoint for its manual "测试代理" action.
|
||||
// A successful probe is the same signal as an automatic reconnect success,
|
||||
// but it must never consume an automatic attempt.
|
||||
s.proxyReconnectSuccessWithGuards(proxyGuards, nil)
|
||||
}
|
||||
writeJSON(w, 200, rep)
|
||||
}
|
||||
|
||||
|
||||
+23
-4
@@ -153,9 +153,12 @@ type Engine struct {
|
||||
slotPlans sync.Map // task -> map[slot name]profile ID; immutable for a run
|
||||
// readiness reports whether a global LLM provider is configured — the signal behind
|
||||
// Ready()/the llm_configured indicator. Wired once at startup; nil → not ready.
|
||||
readiness func() bool
|
||||
readiness func() bool
|
||||
proxyFailure func(taskID string, err error)
|
||||
}
|
||||
|
||||
func (e *Engine) SetProxyFailureHandler(fn func(string, error)) { e.proxyFailure = fn }
|
||||
|
||||
type slotAvailability struct {
|
||||
available bool
|
||||
lastEmit time.Time
|
||||
@@ -942,7 +945,12 @@ func (e *Engine) plannerLoop(ctx context.Context, t *Task) {
|
||||
}
|
||||
e.stampFirstRun(t) // 首次真正规划 → 盖 first_run_at + 算 deadline(仅带 timeout 的任务)
|
||||
e.touch(t.ID)
|
||||
emit := func(r db.Activity) { e.emitActivity(t, r) }
|
||||
emit := func(r db.Activity) {
|
||||
if r.Kind == "tool_result" && r.IsError && e.proxyFailure != nil && isGlobalProxyDialFailure(errors.New(r.Detail+" "+r.Summary)) {
|
||||
e.proxyFailure(t.ID, errors.New(r.Detail+" "+r.Summary))
|
||||
}
|
||||
e.emitActivity(t, r)
|
||||
}
|
||||
ectx := e.clockCtx(e.execContextFor(ctx, t.ID), t, false) // cancellable by Pause; 带任务 deadline
|
||||
if ectx.Err() != nil || e.IsDeleting(t.ID) {
|
||||
return
|
||||
@@ -962,6 +970,9 @@ func (e *Engine) plannerLoop(ctx context.Context, t *Task) {
|
||||
switch {
|
||||
case err != nil && ectx.Err() == nil:
|
||||
log.Printf("[planner] task %s 规划出错: %v", t.ID, err)
|
||||
if e.proxyFailure != nil && isProxyFailure(err) {
|
||||
e.proxyFailure(t.ID, err)
|
||||
}
|
||||
case met:
|
||||
log.Printf("[planner] task %s 判定目标达成: %s", t.ID, reason)
|
||||
// 所有目标达成 → 持久化任务状态为 done(前端 DTO 会优先展示该终态)。
|
||||
@@ -1081,7 +1092,12 @@ func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *ag
|
||||
hasChatMessage := message != ""
|
||||
e.stampFirstRun(t) // 首次真正执行 → 盖 first_run_at + 算 deadline(仅带 timeout 的任务)
|
||||
e.touch(t.ID)
|
||||
emit := func(r db.Activity) { e.emitActivity(t, r) }
|
||||
emit := func(r db.Activity) {
|
||||
if r.Kind == "tool_result" && r.IsError && e.proxyFailure != nil && isGlobalProxyDialFailure(errors.New(r.Detail+" "+r.Summary)) {
|
||||
e.proxyFailure(t.ID, errors.New(r.Detail+" "+r.Summary))
|
||||
}
|
||||
e.emitActivity(t, r)
|
||||
}
|
||||
ectx := e.clockCtx(e.execContextFor(ctx, t.ID), t, false) // cancellable by Pause; 带任务 deadline
|
||||
if ectx.Err() != nil || e.IsDeleting(t.ID) {
|
||||
if err := transitionIntentState(t.Store, intent.ID, "running", "open"); err != nil {
|
||||
@@ -1124,6 +1140,9 @@ func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *ag
|
||||
} else {
|
||||
reason, wrote, err = worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding)
|
||||
}
|
||||
if err != nil && e.proxyFailure != nil && isProxyFailure(err) {
|
||||
e.proxyFailure(t.ID, err)
|
||||
}
|
||||
e.EndLLMCall(t.ID)
|
||||
// model_error 收场 → 额外重跑几次(退避后再试)。仅在意图仍属本 work、任务
|
||||
// 未暂停/未终止/未取消【且未进入收尾】时重试;否则让位给对应分支处理(收尾期不
|
||||
@@ -1338,7 +1357,7 @@ func taskExecutionPaused(cause error) bool {
|
||||
return false
|
||||
}
|
||||
switch abort.Code {
|
||||
case "paused_by_schedule", "paused_by_user", "paused_by_orchestrator", "paused_on_reload", "paused_race_guard",
|
||||
case "paused_by_schedule", "paused_by_user", "paused_by_orchestrator", "paused_by_proxy", "paused_on_reload", "paused_race_guard",
|
||||
"queued_for_admission", "llm_unavailable_queued", "task_deleted":
|
||||
return true
|
||||
default:
|
||||
|
||||
@@ -177,6 +177,11 @@ func TestTaskPauseCauseSurvivesImmediateResume(t *testing.T) {
|
||||
if old.Err() == nil || !taskExecutionPaused(context.Cause(old)) {
|
||||
t.Fatalf("old execution cause=%v was not retained as a task pause", context.Cause(old))
|
||||
}
|
||||
proxyCtx, proxyCancel := context.WithCancelCause(context.Background())
|
||||
proxyCancel(agent.AbortPausedByProxy)
|
||||
if !taskExecutionPaused(context.Cause(proxyCtx)) {
|
||||
t.Fatalf("proxy pause cause=%v was not classified as a task pause", context.Cause(proxyCtx))
|
||||
}
|
||||
if taskExecutionPaused(agent.AbortGoalMet) {
|
||||
t.Fatal("goal completion was misclassified as a task pause")
|
||||
}
|
||||
|
||||
+28
-2
@@ -112,10 +112,30 @@ func (s *Server) admitTaskWhen(t *Task, mode string, requirePaused bool) (bool,
|
||||
return s.admitTaskWhenScheduled(t, mode, requirePaused, false)
|
||||
}
|
||||
func (s *Server) admitTaskWhenScheduled(t *Task, mode string, requirePaused, scheduled bool) (queued bool, err error) {
|
||||
return s.admitTaskWhenScheduledWithReason(t, mode, requirePaused, scheduled, "", false)
|
||||
}
|
||||
|
||||
// admitTaskWhenScheduledWithReason is the shared admission gate. A non-empty
|
||||
// requiredReason is checked after concMu is acquired, alongside the paused and
|
||||
// terminal checks, so proxy recovery cannot revive a task that was re-paused by
|
||||
// a user while the probe was in flight. preserveSchedule keeps calendar state
|
||||
// and never creates schedule_manual_until (used by proxy recovery).
|
||||
func (s *Server) admitTaskWhenScheduledWithReason(t *Task, mode string, requirePaused, scheduled bool, requiredReason string, preserveSchedule bool) (queued bool, err error) {
|
||||
return s.admitTaskWhenScheduledWithGenerationChecked(t, mode, requirePaused, scheduled, requiredReason, preserveSchedule, 0, false)
|
||||
}
|
||||
|
||||
// admitTaskWhenScheduledWithGeneration checks the supplied reconnect round.
|
||||
// Generation zero is valid; callers that do not have a round use the explicit
|
||||
// unchecked form through admitTaskWhenScheduledWithReason.
|
||||
func (s *Server) admitTaskWhenScheduledWithGeneration(t *Task, mode string, requirePaused, scheduled bool, requiredReason string, preserveSchedule bool, requiredGeneration uint64) (queued bool, err error) {
|
||||
return s.admitTaskWhenScheduledWithGenerationChecked(t, mode, requirePaused, scheduled, requiredReason, preserveSchedule, requiredGeneration, true)
|
||||
}
|
||||
|
||||
func (s *Server) admitTaskWhenScheduledWithGenerationChecked(t *Task, mode string, requirePaused, scheduled bool, requiredReason string, preserveSchedule bool, requiredGeneration uint64, checkGeneration bool) (queued bool, err error) {
|
||||
if t == nil {
|
||||
return false, fmt.Errorf("task not found")
|
||||
}
|
||||
manualSchedule := requirePaused && !scheduled
|
||||
manualSchedule := requirePaused && !scheduled && !preserveSchedule
|
||||
if manualSchedule {
|
||||
// Match the calendar tick's lock order: scheduleMu before concMu. The
|
||||
// override must commit before another tick can decide to pause this task.
|
||||
@@ -155,6 +175,12 @@ func (s *Server) admitTaskWhenScheduled(t *Task, mode string, requirePaused, sch
|
||||
if !lifecycle.Paused {
|
||||
return false, fmt.Errorf("仅已暂停的任务可以继续")
|
||||
}
|
||||
if requiredReason != "" && lifecycle.PauseReason != requiredReason {
|
||||
return false, fmt.Errorf("任务暂停原因已改变")
|
||||
}
|
||||
if requiredReason == "paused_by_proxy" && checkGeneration && lifecycle.ReconnectGeneration != requiredGeneration {
|
||||
return false, fmt.Errorf("任务代理重连轮次已改变")
|
||||
}
|
||||
mode = s.resumeAdmissionMode(t)
|
||||
}
|
||||
var manualUntil *time.Time
|
||||
@@ -212,7 +238,7 @@ func (s *Server) admitTaskWhenScheduled(t *Task, mode string, requirePaused, sch
|
||||
if wasTerminal {
|
||||
status = "running"
|
||||
}
|
||||
if err := s.m.applyTaskAdmission(t.ID, lifecycle.Status, status, shouldQueue, mode, wasQueued, manualUntil); err != nil {
|
||||
if err := s.m.applyTaskAdmissionWithSchedule(t.ID, lifecycle.Status, status, shouldQueue, mode, wasQueued, manualUntil, preserveSchedule); err != nil {
|
||||
if !engineWasPaused && !wasPaused && !wasQueued {
|
||||
s.engine.Resume(t)
|
||||
}
|
||||
|
||||
+188
-25
@@ -28,19 +28,26 @@ import (
|
||||
// sharing the process-wide asset store. ID is the PG task id as a string; ExpID
|
||||
// is the exploration the task owns.
|
||||
type Task struct {
|
||||
ID string `json:"id"`
|
||||
ExpID int64 `json:"exploration_id"`
|
||||
Remark string `json:"remark"`
|
||||
Name string `json:"name"` // 可选任务名称;空=未命名
|
||||
CategoryID *int64 `json:"category_id,omitempty"`
|
||||
CategoryName string `json:"category_name,omitempty"`
|
||||
PinnedAt int64 `json:"pinned_at,omitempty"`
|
||||
Description string `json:"description"`
|
||||
Goal string `json:"goal"`
|
||||
CreatedAt int64 `json:"created_at"`
|
||||
CompletedAt int64 `json:"completed_at,omitempty"` // 进入终态的 unix 秒;0=未完成
|
||||
Paused bool `json:"paused"`
|
||||
Queued bool `json:"queued"` // 因并发上限被挂起、等待空位自动启动;true=尚未开跑
|
||||
ID string `json:"id"`
|
||||
ExpID int64 `json:"exploration_id"`
|
||||
Remark string `json:"remark"`
|
||||
Name string `json:"name"` // 可选任务名称;空=未命名
|
||||
CategoryID *int64 `json:"category_id,omitempty"`
|
||||
CategoryName string `json:"category_name,omitempty"`
|
||||
PinnedAt int64 `json:"pinned_at,omitempty"`
|
||||
Description string `json:"description"`
|
||||
Goal string `json:"goal"`
|
||||
CreatedAt int64 `json:"created_at"`
|
||||
CompletedAt int64 `json:"completed_at,omitempty"` // 进入终态的 unix 秒;0=未完成
|
||||
Paused bool `json:"paused"`
|
||||
PauseReason string `json:"pause_reason,omitempty"`
|
||||
ReconnectAttempt int `json:"-"`
|
||||
ReconnectMax int `json:"-"`
|
||||
ReconnectNextAt int64 `json:"-"`
|
||||
ReconnectExhausted bool `json:"-"`
|
||||
ReconnectLastError string `json:"-"`
|
||||
ReconnectGeneration uint64 `json:"-"`
|
||||
Queued bool `json:"queued"` // 因并发上限被挂起、等待空位自动启动;true=尚未开跑
|
||||
// QueuedAt is an internal Unix-nanosecond ordering key. It is deliberately
|
||||
// finer than CreatedAt so several tasks enqueued in the same second retain
|
||||
// their real FIFO order.
|
||||
@@ -88,6 +95,13 @@ type taskLifecycleState struct {
|
||||
PinnedAt int64
|
||||
Status string
|
||||
Paused bool
|
||||
PauseReason string
|
||||
ReconnectAttempt int
|
||||
ReconnectMax int
|
||||
ReconnectNextAt int64
|
||||
ReconnectExhausted bool
|
||||
ReconnectLastError string
|
||||
ReconnectGeneration uint64
|
||||
Queued bool
|
||||
QueuedAt int64
|
||||
QueueMode string
|
||||
@@ -113,11 +127,13 @@ func (t *Task) lifecycleSnapshot() taskLifecycleState {
|
||||
|
||||
func (t *Task) lifecycleSnapshotLocked() taskLifecycleState {
|
||||
return taskLifecycleState{
|
||||
Name: t.Name,
|
||||
Remark: t.Remark,
|
||||
PinnedAt: t.PinnedAt,
|
||||
Status: t.Status,
|
||||
Paused: t.Paused,
|
||||
Name: t.Name,
|
||||
Remark: t.Remark,
|
||||
PinnedAt: t.PinnedAt,
|
||||
Status: t.Status,
|
||||
Paused: t.Paused,
|
||||
PauseReason: t.PauseReason,
|
||||
ReconnectAttempt: t.ReconnectAttempt, ReconnectMax: t.ReconnectMax, ReconnectNextAt: t.ReconnectNextAt, ReconnectExhausted: t.ReconnectExhausted, ReconnectLastError: t.ReconnectLastError, ReconnectGeneration: t.ReconnectGeneration,
|
||||
Queued: t.Queued,
|
||||
QueuedAt: t.QueuedAt,
|
||||
QueueMode: t.QueueMode,
|
||||
@@ -145,6 +161,13 @@ func (t *Task) updateLifecycle(update func(*taskLifecycleState)) {
|
||||
t.PinnedAt = state.PinnedAt
|
||||
t.Status = state.Status
|
||||
t.Paused = state.Paused
|
||||
t.PauseReason = state.PauseReason
|
||||
t.ReconnectAttempt = state.ReconnectAttempt
|
||||
t.ReconnectMax = state.ReconnectMax
|
||||
t.ReconnectNextAt = state.ReconnectNextAt
|
||||
t.ReconnectExhausted = state.ReconnectExhausted
|
||||
t.ReconnectLastError = state.ReconnectLastError
|
||||
t.ReconnectGeneration = state.ReconnectGeneration
|
||||
t.Queued = state.Queued
|
||||
t.QueuedAt = state.QueuedAt
|
||||
t.QueueMode = state.QueueMode
|
||||
@@ -242,6 +265,7 @@ type Manager struct {
|
||||
// their in-memory mirrors. lifecycleMu makes snapshots race-free, but without
|
||||
// this outer write lock an older request could commit first and publish last.
|
||||
taskStateMu sync.Mutex
|
||||
controlMu *sync.Mutex // Server's admission/task-control mutex, when attached.
|
||||
mu sync.RWMutex
|
||||
tasks map[string]*Task
|
||||
active string
|
||||
@@ -261,6 +285,10 @@ type Manager struct {
|
||||
globalProxy string
|
||||
}
|
||||
|
||||
// SetTaskControlMutex lets lifecycle operations that originate in Manager
|
||||
// participate in the Server's admission/task-control critical section.
|
||||
func (m *Manager) SetTaskControlMutex(mu *sync.Mutex) { m.controlMu = mu }
|
||||
|
||||
// Settings keys the UI toggles at runtime.
|
||||
const (
|
||||
settingTrafficCapture = "traffic_capture"
|
||||
@@ -1073,7 +1101,7 @@ func taskFromPG(pt *pgdb.Task, store *pgdb.ExplorationStore, ic *intercept.Inter
|
||||
Remark: pt.Remark,
|
||||
CategoryID: cloneInt64Ptr(pt.CategoryID), CategoryName: pt.CategoryName,
|
||||
PinnedAt: unixOrZero(pt.PinnedAt),
|
||||
Description: pt.Description, Goal: pt.Goal, CreatedAt: pt.CreatedAt.Unix(), Paused: pt.Paused, Queued: pt.Queued,
|
||||
Description: pt.Description, Goal: pt.Goal, CreatedAt: pt.CreatedAt.Unix(), Paused: pt.Paused, PauseReason: pt.PauseReason, Queued: pt.Queued,
|
||||
QueuedAt: unixNanoOrZero(pt.QueuedAt), QueueMode: pt.QueueMode,
|
||||
CompletedAt: unixOrZero(pt.CompletedAt), Status: pt.Status, ParentRef: pt.ParentRef,
|
||||
LLMProfileID: pt.LLMProfileID,
|
||||
@@ -1391,12 +1419,16 @@ func (m *Manager) SetTaskPaused(id string, paused bool) error {
|
||||
// admission keeps its FIFO timestamp; a task that was explicitly paused and is
|
||||
// now re-queued receives a fresh tail position.
|
||||
func (m *Manager) ApplyTaskAdmission(id, expectedStatus, status string, queued bool, mode string, preservePosition bool) error {
|
||||
return m.applyTaskAdmission(id, expectedStatus, status, queued, mode, preservePosition, nil)
|
||||
return m.applyTaskAdmissionWithSchedule(id, expectedStatus, status, queued, mode, preservePosition, nil, false)
|
||||
}
|
||||
|
||||
// Manual task control records its schedule override in the admission commit, so
|
||||
// failed resumes cannot bypass a calendar and FIFO promotion preserves it.
|
||||
func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued bool, mode string, preservePosition bool, manualUntil *time.Time) error {
|
||||
return m.applyTaskAdmissionWithSchedule(id, expectedStatus, status, queued, mode, preservePosition, manualUntil, false)
|
||||
}
|
||||
|
||||
func (m *Manager) applyTaskAdmissionWithSchedule(id, expectedStatus, status string, queued bool, mode string, preservePosition bool, manualUntil *time.Time, preserveSchedule bool) error {
|
||||
m.taskStateMu.Lock()
|
||||
defer m.taskStateMu.Unlock()
|
||||
n, err := strconv.ParseInt(id, 10, 64)
|
||||
@@ -1420,8 +1452,9 @@ 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, schedule_paused=false,
|
||||
schedule_manual_until=COALESCE($7::timestamptz, schedule_manual_until),
|
||||
paused=false, pause_reason='',
|
||||
schedule_paused=CASE WHEN $8 THEN schedule_paused ELSE false END,
|
||||
schedule_manual_until=CASE WHEN $8 THEN schedule_manual_until ELSE COALESCE($7::timestamptz, schedule_manual_until) END,
|
||||
queued=$3,
|
||||
queued_at=CASE
|
||||
WHEN NOT $3 THEN NULL
|
||||
@@ -1444,7 +1477,7 @@ func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued b
|
||||
,active_run_seconds=active_run_seconds + CASE WHEN ($3 OR $2 IN ('done','failed','timeout')) AND run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END
|
||||
,run_segment_started_at=CASE WHEN $3 OR $2 IN ('done','failed','timeout') THEN NULL ELSE run_segment_started_at END
|
||||
WHERE id=$1 AND deleted_at IS NULL AND status=$6
|
||||
RETURNING queued_at, queue_mode, completed_at, first_run_at, deadline_at, active_run_seconds, run_segment_started_at`, n, status, queued, mode, preservePosition, expectedStatus, manualUntil).
|
||||
RETURNING queued_at, queue_mode, completed_at, first_run_at, deadline_at, active_run_seconds, run_segment_started_at`, n, status, queued, mode, preservePosition, expectedStatus, manualUntil, preserveSchedule).
|
||||
Scan(&queuedAt, &committedMode, &completedAt, &firstRunAt, &deadlineAt, &activeRunSeconds, &segmentStartedAt)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return fmt.Errorf("task %s lifecycle changed before admission (expected status %q)", id, expectedStatus)
|
||||
@@ -1458,6 +1491,7 @@ func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued b
|
||||
t.updateLifecycle(func(state *taskLifecycleState) {
|
||||
state.Status = status
|
||||
state.Paused = false
|
||||
state.PauseReason = ""
|
||||
state.Queued = queued
|
||||
state.QueueMode = committedMode
|
||||
state.QueuedAt = 0
|
||||
@@ -1481,6 +1515,11 @@ func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued b
|
||||
if segmentStartedAt.Valid {
|
||||
state.RunSegmentStartedAt = segmentStartedAt.Time.Unix()
|
||||
}
|
||||
state.ReconnectAttempt = 0
|
||||
state.ReconnectMax = 0
|
||||
state.ReconnectNextAt = 0
|
||||
state.ReconnectExhausted = false
|
||||
state.ReconnectLastError = ""
|
||||
})
|
||||
}
|
||||
m.mu.Unlock()
|
||||
@@ -1494,6 +1533,12 @@ func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued b
|
||||
func (m *Manager) ApplyTaskPause(id string) error { return m.ApplyTaskPauseOrigin(id, false) }
|
||||
|
||||
func (m *Manager) ApplyTaskPauseOrigin(id string, scheduled bool) error {
|
||||
return m.ApplyTaskPauseOriginReason(id, map[bool]string{true: "paused_by_schedule", false: "paused_by_user"}[scheduled], scheduled)
|
||||
}
|
||||
|
||||
// ApplyTaskPauseOriginReason persists a pause cause and keeps it on the live handle.
|
||||
// reason is a stable API value such as user, schedule, orchestrator, or proxy.
|
||||
func (m *Manager) ApplyTaskPauseOriginReason(id, reason string, scheduled bool) error {
|
||||
m.taskStateMu.Lock()
|
||||
defer m.taskStateMu.Unlock()
|
||||
n, err := strconv.ParseInt(id, 10, 64)
|
||||
@@ -1504,12 +1549,12 @@ func (m *Manager) ApplyTaskPauseOrigin(id string, scheduled bool) error {
|
||||
var activeRunSeconds int64
|
||||
var segmentStartedAt sql.NullTime
|
||||
err = m.pg.QueryRow(`UPDATE tasks
|
||||
SET paused=true, queued=false, queued_at=NULL, schedule_paused=$2, schedule_manual_until=NULL,
|
||||
SET paused=true, pause_reason=$3, queued=false, queued_at=NULL, schedule_paused=$2, schedule_manual_until=NULL,
|
||||
active_run_seconds=active_run_seconds + CASE WHEN run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
|
||||
run_segment_started_at=NULL
|
||||
WHERE id=$1 AND deleted_at IS NULL AND paused=false
|
||||
AND status NOT IN ('done','failed','timeout')
|
||||
RETURNING COALESCE(queue_mode,''), active_run_seconds, run_segment_started_at`, n, scheduled).Scan(&mode, &activeRunSeconds, &segmentStartedAt)
|
||||
RETURNING COALESCE(queue_mode,''), active_run_seconds, run_segment_started_at`, n, scheduled, reason).Scan(&mode, &activeRunSeconds, &segmentStartedAt)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return fmt.Errorf("task %s is unavailable for pause", id)
|
||||
}
|
||||
@@ -1520,6 +1565,12 @@ func (m *Manager) ApplyTaskPauseOrigin(id string, scheduled bool) error {
|
||||
if t := m.tasks[id]; t != nil {
|
||||
t.updateLifecycle(func(state *taskLifecycleState) {
|
||||
state.Paused = true
|
||||
state.PauseReason = reason
|
||||
state.ReconnectAttempt = 0
|
||||
state.ReconnectMax = 0
|
||||
state.ReconnectNextAt = 0
|
||||
state.ReconnectExhausted = false
|
||||
state.ReconnectLastError = ""
|
||||
state.Queued = false
|
||||
state.QueuedAt = 0
|
||||
state.QueueMode = mode
|
||||
@@ -1534,6 +1585,110 @@ func (m *Manager) ApplyTaskPauseOrigin(id string, scheduled bool) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// ApplyProxyPause atomically pauses a running task and marks it eligible for
|
||||
// reconnect auto-resume. Already-paused tasks retain their original reason.
|
||||
func (m *Manager) ApplyProxyPause(id string) error {
|
||||
_, err := m.ApplyProxyPauseWithCallback(id, nil)
|
||||
return err
|
||||
}
|
||||
|
||||
// ApplyProxyPauseWithCallback is the locked proxy-pause transition. When the
|
||||
// task actually changes from running to paused, callback runs while the
|
||||
// Server's task-control mutex is still held, after the DB and live lifecycle
|
||||
// state have committed. This lets callers install the Engine barrier before a
|
||||
// concurrent user/orchestrator pause can observe the new paused state.
|
||||
func (m *Manager) ApplyProxyPauseWithCallback(id string, callback func()) (transitioned bool, err error) {
|
||||
if m.controlMu != nil {
|
||||
m.controlMu.Lock()
|
||||
defer m.controlMu.Unlock()
|
||||
}
|
||||
m.taskStateMu.Lock()
|
||||
defer m.taskStateMu.Unlock()
|
||||
n, err := strconv.ParseInt(id, 10, 64)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
var mode string
|
||||
var active int64
|
||||
var started sql.NullTime
|
||||
err = m.pg.QueryRow(`UPDATE tasks SET paused=true, pause_reason='paused_by_proxy', queued=false, queued_at=NULL,
|
||||
active_run_seconds=active_run_seconds + CASE WHEN run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
|
||||
run_segment_started_at=NULL WHERE id=$1 AND deleted_at IS NULL AND paused=false AND status NOT IN ('done','failed','timeout')
|
||||
RETURNING COALESCE(queue_mode,''), active_run_seconds, run_segment_started_at`, n).Scan(&mode, &active, &started)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
m.mu.RLock()
|
||||
t := m.tasks[id]
|
||||
m.mu.RUnlock()
|
||||
if t != nil {
|
||||
t.updateLifecycle(func(s *taskLifecycleState) {
|
||||
s.Paused = true
|
||||
s.PauseReason = "paused_by_proxy"
|
||||
s.Queued = false
|
||||
s.QueuedAt = 0
|
||||
s.QueueMode = mode
|
||||
s.ActiveRunSeconds = active
|
||||
s.RunSegmentStartedAt = 0
|
||||
})
|
||||
}
|
||||
if callback != nil {
|
||||
callback()
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// ClearPauseReason removes auto-resume eligibility without changing lifecycle.
|
||||
func (m *Manager) ClearPauseReason(id string) error {
|
||||
if _, err := m.pg.Exec(`UPDATE tasks SET pause_reason='' WHERE id=$1 AND pause_reason='paused_by_proxy'`, id); err != nil {
|
||||
return err
|
||||
}
|
||||
m.mu.RLock()
|
||||
t := m.tasks[id]
|
||||
m.mu.RUnlock()
|
||||
if t != nil && t.lifecycleSnapshot().PauseReason == "paused_by_proxy" {
|
||||
t.updateLifecycle(func(s *taskLifecycleState) {
|
||||
s.PauseReason = ""
|
||||
s.ReconnectAttempt, s.ReconnectMax, s.ReconnectNextAt = 0, 0, 0
|
||||
s.ReconnectExhausted, s.ReconnectLastError = false, ""
|
||||
})
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ConvertProxyPauseToSchedule records that a calendar window ended while a
|
||||
// task was waiting for proxy recovery. It preserves the Engine pause barrier
|
||||
// and removes proxy auto-resume eligibility until the next window.
|
||||
func (m *Manager) ConvertProxyPauseToSchedule(id string) error {
|
||||
m.taskStateMu.Lock()
|
||||
defer m.taskStateMu.Unlock()
|
||||
n, err := strconv.ParseInt(id, 10, 64)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
res, err := m.pg.Exec(`UPDATE tasks SET pause_reason='paused_by_schedule', schedule_paused=true WHERE id=$1 AND paused=true AND pause_reason='paused_by_proxy' AND status NOT IN ('done','failed','timeout')`, n)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if rows, _ := res.RowsAffected(); rows == 0 {
|
||||
return nil
|
||||
}
|
||||
m.mu.RLock()
|
||||
t := m.tasks[id]
|
||||
m.mu.RUnlock()
|
||||
if t != nil {
|
||||
t.updateLifecycle(func(v *taskLifecycleState) {
|
||||
v.PauseReason = "paused_by_schedule"
|
||||
v.ReconnectAttempt, v.ReconnectMax, v.ReconnectNextAt = 0, 0, 0
|
||||
v.ReconnectExhausted, v.ReconnectLastError = false, ""
|
||||
})
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// EnqueueTask persists the concurrency hold and syncs the in-memory handle.
|
||||
func (m *Manager) EnqueueTask(id, mode string) error {
|
||||
m.taskStateMu.Lock()
|
||||
@@ -1748,6 +1903,10 @@ func (m *Manager) SetTaskStatusGuarded(id, status string) (won bool, err error)
|
||||
state.RunSegmentStartedAt = unixOrZero(started)
|
||||
if pgdb.IsTerminal(status) {
|
||||
state.RunSegmentStartedAt = 0
|
||||
state.Paused = false
|
||||
state.PauseReason = ""
|
||||
state.ReconnectAttempt, state.ReconnectMax, state.ReconnectNextAt = 0, 0, 0
|
||||
state.ReconnectExhausted, state.ReconnectLastError = false, ""
|
||||
if state.CompletedAt == 0 {
|
||||
state.CompletedAt = time.Now().Unix()
|
||||
}
|
||||
@@ -1784,6 +1943,10 @@ func (m *Manager) SetTaskStatus(id, status string) error {
|
||||
state.RunSegmentStartedAt = unixOrZero(started)
|
||||
if pgdb.IsTerminal(status) {
|
||||
state.RunSegmentStartedAt = 0
|
||||
state.Paused = false
|
||||
state.PauseReason = ""
|
||||
state.ReconnectAttempt, state.ReconnectMax, state.ReconnectNextAt = 0, 0, 0
|
||||
state.ReconnectExhausted, state.ReconnectLastError = false, ""
|
||||
if state.CompletedAt == 0 {
|
||||
state.CompletedAt = time.Now().Unix()
|
||||
}
|
||||
|
||||
@@ -0,0 +1,433 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/Autumn-27/artex/agent"
|
||||
)
|
||||
|
||||
const (
|
||||
settingProxyReconnectMax = "proxy_reconnect_max_attempts"
|
||||
settingProxyReconnectInterval = "proxy_reconnect_interval_seconds"
|
||||
defaultProxyReconnectMax = 5
|
||||
defaultProxyReconnectInterval = 30
|
||||
)
|
||||
|
||||
// isProxyFailure is used for LLM/planner errors, where the error has no trusted
|
||||
// indication that it came from the global target proxy. Keep this deliberately
|
||||
// narrow: generic EOF/reset/refused errors belong to the model_error retry path.
|
||||
func isProxyFailure(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
s := strings.ToLower(err.Error())
|
||||
for _, part := range []string{"proxyconnect", "proxy connection", "proxy error", "proxy handshake", "socks handshake", "failed to connect to proxy", "connect to proxy"} {
|
||||
if strings.Contains(s, part) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// isGlobalProxyDialFailure is only used by the MITM upstream callback. That
|
||||
// callback is on the actual configured global-proxy path; the saved proxy is
|
||||
// still probed before any task is paused, so a target-side RequestError cannot
|
||||
// pause tasks while the proxy itself remains healthy.
|
||||
func isGlobalProxyDialFailure(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
s := strings.ToLower(err.Error())
|
||||
for _, part := range []string{"proxyconnect", "proxy connection", "proxy error", "proxy handshake", "socks", "connect to proxy", "upstream proxy", "connection refused", "connection reset", "network is unreachable", "broken pipe", "unexpected eof", "eof"} {
|
||||
if strings.Contains(s, part) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
type proxyReconnectSupervisor struct {
|
||||
mu sync.Mutex
|
||||
wake chan struct{}
|
||||
}
|
||||
|
||||
type proxyReconnectGuard struct {
|
||||
generation uint64
|
||||
attempt int
|
||||
checkAttempt bool
|
||||
}
|
||||
|
||||
func (s *Server) snapshotProxyReconnectGuards(checkAttempt bool) map[string]proxyReconnectGuard {
|
||||
guards := make(map[string]proxyReconnectGuard)
|
||||
if s == nil || s.m == nil {
|
||||
return guards
|
||||
}
|
||||
for _, t := range s.m.List() {
|
||||
st := t.lifecycleSnapshot()
|
||||
if st.PauseReason != "paused_by_proxy" {
|
||||
continue
|
||||
}
|
||||
guards[t.ID] = proxyReconnectGuard{
|
||||
generation: st.ReconnectGeneration,
|
||||
attempt: st.ReconnectAttempt,
|
||||
checkAttempt: checkAttempt,
|
||||
}
|
||||
}
|
||||
return guards
|
||||
}
|
||||
|
||||
func proxyReconnectGuardMatches(st taskLifecycleState, guard proxyReconnectGuard) bool {
|
||||
return st.PauseReason == "paused_by_proxy" && st.ReconnectGeneration == guard.generation &&
|
||||
(!guard.checkAttempt || st.ReconnectAttempt == guard.attempt)
|
||||
}
|
||||
|
||||
func (s *Server) ensureProxySupervisor() {
|
||||
s.proxyReconnect.mu.Lock()
|
||||
if s.proxyReconnect.wake == nil {
|
||||
s.proxyReconnect.wake = make(chan struct{}, 1)
|
||||
go s.runProxyReconnect(s.ctx)
|
||||
}
|
||||
s.proxyReconnect.mu.Unlock()
|
||||
}
|
||||
|
||||
func (s *Server) signalProxySupervisor() {
|
||||
s.ensureProxySupervisor()
|
||||
select {
|
||||
case s.proxyReconnect.wake <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) proxyReconnectConfig() (int, time.Duration) {
|
||||
max := defaultProxyReconnectMax
|
||||
interval := defaultProxyReconnectInterval
|
||||
if s == nil || s.m == nil || s.m.pg == nil {
|
||||
return max, time.Duration(interval) * time.Second
|
||||
}
|
||||
if v, ok, _ := s.m.pg.GetSetting(settingProxyReconnectMax); ok {
|
||||
if n, err := fmt.Sscanf(strings.TrimSpace(v), "%d", &max); err != nil || n != 1 || max <= 0 {
|
||||
max = defaultProxyReconnectMax
|
||||
}
|
||||
}
|
||||
if v, ok, _ := s.m.pg.GetSetting(settingProxyReconnectInterval); ok {
|
||||
if n, err := fmt.Sscanf(strings.TrimSpace(v), "%d", &interval); err != nil || n != 1 || interval <= 0 {
|
||||
interval = defaultProxyReconnectInterval
|
||||
}
|
||||
}
|
||||
return max, time.Duration(interval) * time.Second
|
||||
}
|
||||
|
||||
func (s *Server) runProxyProbe(ctx context.Context, proxyRaw, caPEMPath string, probes []string, timeout time.Duration) egressProbeResult {
|
||||
if s != nil && s.proxyProbe != nil {
|
||||
return s.proxyProbe(ctx, proxyRaw, caPEMPath, probes, timeout)
|
||||
}
|
||||
return probeEgress(ctx, proxyRaw, caPEMPath, probes, timeout)
|
||||
}
|
||||
|
||||
func (s *Server) handleProxyFailure(taskID string, cause error) {
|
||||
if strings.TrimSpace(s.m.GlobalProxy()) == "" {
|
||||
return
|
||||
}
|
||||
if probe := probeEgress(context.Background(), s.m.GlobalProxy(), "", defaultEgressProbes, 10*time.Second); probe.OK {
|
||||
return
|
||||
}
|
||||
s.pauseProxyTask(taskID, cause)
|
||||
}
|
||||
|
||||
func (s *Server) pauseProxyTask(taskID string, cause error) {
|
||||
if strings.TrimSpace(s.m.GlobalProxy()) == "" {
|
||||
return
|
||||
}
|
||||
max, interval := s.proxyReconnectConfig()
|
||||
now := time.Now().Add(interval).Unix()
|
||||
lastError := truncateProxyError(cause)
|
||||
transitioned, err := s.m.ApplyProxyPauseWithCallback(taskID, func() {
|
||||
if s.engine != nil {
|
||||
s.engine.Pause(taskID, agent.AbortPausedByProxy)
|
||||
}
|
||||
if t, ok := s.m.Task(taskID); ok {
|
||||
t.updateLifecycle(func(st *taskLifecycleState) {
|
||||
// A successful admission clears the prior round's counters. This
|
||||
// transition starts a fresh round, so make that reset atomic with
|
||||
// the persisted proxy pause and install a new generation.
|
||||
st.ReconnectGeneration++
|
||||
st.ReconnectAttempt = 0
|
||||
st.ReconnectMax = max
|
||||
st.ReconnectNextAt = now
|
||||
st.ReconnectExhausted = false
|
||||
st.ReconnectLastError = lastError
|
||||
})
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
log.Printf("[proxy] pause task %s: %v", taskID, err)
|
||||
return
|
||||
}
|
||||
if !transitioned {
|
||||
// A no-op means the task was already paused (possibly by a user,
|
||||
// schedule, or an exhausted proxy attempt). Preserve its reason and
|
||||
// reconnect state exactly as-is.
|
||||
return
|
||||
}
|
||||
s.cancelTaskChat(taskID, agent.AbortChatPausedWithTask)
|
||||
s.ensureProxySupervisor()
|
||||
s.signalProxySupervisor()
|
||||
}
|
||||
|
||||
func (s *Server) handleProxyTrafficFailure(cause error) {
|
||||
if !isGlobalProxyDialFailure(cause) || strings.TrimSpace(s.m.GlobalProxy()) == "" {
|
||||
return
|
||||
}
|
||||
// Confirm the saved global proxy itself before pausing. MITM RequestError can
|
||||
// also represent a target-side failure, and those must not pause a task.
|
||||
if probe := probeEgress(context.Background(), s.m.GlobalProxy(), "", defaultEgressProbes, 10*time.Second); probe.OK {
|
||||
return
|
||||
}
|
||||
for _, t := range s.m.List() {
|
||||
st := t.lifecycleSnapshot()
|
||||
if isTerminalStatus(st.Status) || st.Paused || s.engine.IsDeleting(t.ID) {
|
||||
continue
|
||||
}
|
||||
s.pauseProxyTask(t.ID, cause)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) handleProxyToolFailure(ctx context.Context, cause error) {
|
||||
if !isGlobalProxyDialFailure(cause) || strings.TrimSpace(s.m.GlobalProxy()) == "" {
|
||||
return
|
||||
}
|
||||
if probe := probeEgress(context.Background(), s.m.GlobalProxy(), "", defaultEgressProbes, 10*time.Second); probe.OK {
|
||||
return
|
||||
}
|
||||
if info := agent.RunInfoFrom(ctx); info.TaskID > 0 {
|
||||
s.pauseProxyTask(fmt.Sprint(info.TaskID), cause)
|
||||
}
|
||||
}
|
||||
|
||||
func truncateProxyError(err error) string {
|
||||
if err == nil {
|
||||
return ""
|
||||
}
|
||||
s := err.Error()
|
||||
if len(s) > 240 {
|
||||
return s[:240] + "…"
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func (s *Server) runProxyReconnect(ctx context.Context) {
|
||||
ticker := time.NewTicker(500 * time.Millisecond)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
s.proxyReconnectTick()
|
||||
case <-s.proxyReconnect.wake:
|
||||
s.proxyReconnectTick()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) proxyReconnectTick() {
|
||||
if strings.TrimSpace(s.m.GlobalProxy()) == "" {
|
||||
return
|
||||
}
|
||||
now := time.Now().Unix()
|
||||
max, interval := s.proxyReconnectConfig()
|
||||
for _, t := range s.m.List() {
|
||||
st := t.lifecycleSnapshot()
|
||||
if isTerminalStatus(st.Status) {
|
||||
t.updateLifecycle(func(v *taskLifecycleState) {
|
||||
if v.PauseReason == "paused_by_proxy" {
|
||||
v.PauseReason = ""
|
||||
}
|
||||
v.ReconnectAttempt, v.ReconnectMax, v.ReconnectNextAt = 0, 0, 0
|
||||
v.ReconnectExhausted, v.ReconnectLastError = false, ""
|
||||
})
|
||||
continue
|
||||
}
|
||||
if st.PauseReason != "paused_by_proxy" || st.ReconnectExhausted || (st.ReconnectNextAt > 0 && st.ReconnectNextAt > now) {
|
||||
continue
|
||||
}
|
||||
attempt := st.ReconnectAttempt + 1
|
||||
round := st.ReconnectGeneration
|
||||
claimed := false
|
||||
t.updateLifecycle(func(v *taskLifecycleState) {
|
||||
if v.PauseReason != "paused_by_proxy" || v.ReconnectGeneration != round || v.ReconnectAttempt != st.ReconnectAttempt {
|
||||
return
|
||||
}
|
||||
v.ReconnectAttempt = attempt
|
||||
v.ReconnectMax = max
|
||||
v.ReconnectNextAt = 0
|
||||
claimed = true
|
||||
})
|
||||
if !claimed {
|
||||
continue
|
||||
}
|
||||
res := s.runProxyProbe(context.Background(), s.m.GlobalProxy(), "", defaultEgressProbes, 10*time.Second)
|
||||
if res.OK {
|
||||
s.proxyReconnectSuccessWithGuards(map[string]proxyReconnectGuard{
|
||||
t.ID: {generation: round, attempt: attempt, checkAttempt: true},
|
||||
}, nil)
|
||||
return
|
||||
}
|
||||
err := res.Error
|
||||
exhausted := attempt >= max
|
||||
next := time.Now().Add(interval).Unix()
|
||||
t.updateLifecycle(func(v *taskLifecycleState) {
|
||||
if v.PauseReason != "paused_by_proxy" || v.ReconnectGeneration != round || v.ReconnectAttempt != attempt {
|
||||
return
|
||||
}
|
||||
v.ReconnectNextAt = next
|
||||
v.ReconnectExhausted = exhausted
|
||||
v.ReconnectLastError = err
|
||||
})
|
||||
log.Printf("[proxy] reconnect attempt %d/%d failed: %s", attempt, max, err)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) proxyReconnectSuccess() {
|
||||
s.proxyReconnectSuccessWithGuards(s.snapshotProxyReconnectGuards(false), nil)
|
||||
}
|
||||
|
||||
func (s *Server) proxyReconnectSuccessFor(taskID string, generation uint64, attempt int) {
|
||||
s.proxyReconnectSuccessWithGuards(map[string]proxyReconnectGuard{
|
||||
taskID: {generation: generation, attempt: attempt, checkAttempt: true},
|
||||
}, nil)
|
||||
}
|
||||
|
||||
// proxyReconnectSuccessWithHook is kept small and private so tests can
|
||||
// deterministically place a competing pause immediately after admission has
|
||||
// released concMu. Production callers use proxyReconnectSuccess without a
|
||||
// hook.
|
||||
func (s *Server) proxyReconnectSuccessWithHook(afterAdmission func()) {
|
||||
s.proxyReconnectSuccessWithGuards(s.snapshotProxyReconnectGuards(false), afterAdmission)
|
||||
}
|
||||
|
||||
func (s *Server) proxyReconnectSuccessWithGuard(expectedID string, expectedGeneration uint64, expectedAttempt int, afterAdmission func()) {
|
||||
guards := s.snapshotProxyReconnectGuards(false)
|
||||
if expectedID != "" {
|
||||
guards = map[string]proxyReconnectGuard{
|
||||
expectedID: {generation: expectedGeneration, attempt: expectedAttempt, checkAttempt: true},
|
||||
}
|
||||
}
|
||||
s.proxyReconnectSuccessWithGuards(guards, afterAdmission)
|
||||
}
|
||||
|
||||
func (s *Server) proxyReconnectSuccessWithGuards(guards map[string]proxyReconnectGuard, afterAdmission func()) {
|
||||
max, interval := s.proxyReconnectConfig()
|
||||
s.scheduleMu.Lock()
|
||||
defer s.scheduleMu.Unlock()
|
||||
for _, t := range s.m.List() {
|
||||
guard, ok := guards[t.ID]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if !proxyReconnectGuardMatches(t.lifecycleSnapshot(), guard) {
|
||||
continue
|
||||
}
|
||||
if s.m.pg != nil {
|
||||
inside, scheduleErr := s.proxyResumeInsideSchedule(t.ID, time.Now())
|
||||
if scheduleErr != nil {
|
||||
log.Printf("[proxy] schedule gate for task %s: %v", t.ID, scheduleErr)
|
||||
continue
|
||||
}
|
||||
if !inside {
|
||||
if err := s.m.ConvertProxyPauseToSchedule(t.ID); err != nil {
|
||||
log.Printf("[proxy] schedule pause task %s: %v", t.ID, err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
}
|
||||
var err error
|
||||
if s.engine != nil {
|
||||
_, err = s.admitTaskWhenScheduledWithGenerationChecked(t, "resume", true, false, "paused_by_proxy", true, guard.generation, true)
|
||||
if afterAdmission != nil {
|
||||
afterAdmission()
|
||||
}
|
||||
} else {
|
||||
err = s.m.ClearPauseReason(t.ID)
|
||||
}
|
||||
if err != nil {
|
||||
// Admission re-checks the persisted reason under concMu. A user pause,
|
||||
// terminal transition, delete, or archive therefore cannot be revived;
|
||||
// only keep probing if the proxy pause is still eligible.
|
||||
st := t.lifecycleSnapshot()
|
||||
if proxyReconnectGuardMatches(st, guard) && !isTerminalStatus(st.Status) {
|
||||
t.updateLifecycle(func(v *taskLifecycleState) {
|
||||
v.ReconnectMax = max
|
||||
v.ReconnectNextAt = time.Now().Add(interval).Unix()
|
||||
})
|
||||
}
|
||||
log.Printf("[proxy] resume task %s: %v", t.ID, err)
|
||||
continue
|
||||
}
|
||||
// The non-queued path's startAdmittedTask already resumes the Engine
|
||||
// while concMu is held. Do not Resume again after releasing the lock:
|
||||
// a concurrent user/orchestrator pause could have installed a newer
|
||||
// barrier in the meantime.
|
||||
t.updateLifecycle(func(v *taskLifecycleState) {
|
||||
if !proxyReconnectGuardMatches(*v, guard) {
|
||||
return
|
||||
}
|
||||
v.ReconnectAttempt = 0
|
||||
v.ReconnectMax = 0
|
||||
v.ReconnectNextAt = 0
|
||||
v.ReconnectExhausted = false
|
||||
v.ReconnectLastError = ""
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) testTaskProxy(w http.ResponseWriter, r *http.Request) {
|
||||
id := r.PathValue("id")
|
||||
t, ok := s.m.Task(id)
|
||||
if !ok {
|
||||
writeErr(w, 404, "task not found")
|
||||
return
|
||||
}
|
||||
proxy := strings.TrimSpace(s.m.GlobalProxy())
|
||||
guards := s.snapshotProxyReconnectGuards(false)
|
||||
if proxy == "" {
|
||||
writeJSON(w, 200, map[string]any{"ok": true, "direct": true})
|
||||
s.proxyReconnectSuccessWithGuards(guards, nil)
|
||||
return
|
||||
}
|
||||
res := s.runProxyProbe(r.Context(), proxy, "", defaultEgressProbes, 10*time.Second)
|
||||
if res.OK {
|
||||
s.proxyReconnectSuccessWithGuards(guards, nil)
|
||||
}
|
||||
if !res.OK && res.Error == "" {
|
||||
res.Error = "代理探测失败"
|
||||
}
|
||||
writeJSON(w, 200, map[string]any{"ok": res.OK, "error": res.Error, "task_id": t.ID})
|
||||
}
|
||||
|
||||
func (s *Server) setProxyReconnectSettings(max, interval *int) error {
|
||||
if max != nil {
|
||||
if *max < 1 || *max > 1000 {
|
||||
return errors.New("proxy reconnect max must be 1..1000")
|
||||
}
|
||||
if err := s.m.pg.SetSetting(settingProxyReconnectMax, fmt.Sprint(*max)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if interval != nil {
|
||||
if *interval < 1 || *interval > 86400 {
|
||||
return errors.New("proxy reconnect interval must be 1..86400 seconds")
|
||||
}
|
||||
if err := s.m.pg.SetSetting(settingProxyReconnectInterval, fmt.Sprint(*interval)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,500 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/Autumn-27/artex/agent"
|
||||
)
|
||||
|
||||
func proxyTestManager(t *testing.T) *Manager {
|
||||
t.Helper()
|
||||
m, err := NewManager(t.TempDir(), "")
|
||||
if err != nil {
|
||||
t.Skipf("postgres unavailable (%v) - skipping", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = m.Close() })
|
||||
return m
|
||||
}
|
||||
|
||||
func TestProxyFailureClassification(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
err error
|
||||
want bool
|
||||
}{
|
||||
{"proxy connect", errors.New("proxyconnect tcp: connection refused"), true},
|
||||
{"reset", errors.New("dial tcp: connection reset by peer"), false},
|
||||
{"eof", errors.New("unexpected EOF"), false},
|
||||
{"quota", errors.New("status 429: insufficient_quota"), false},
|
||||
{"auth", errors.New("status 401: invalid api key"), false},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
if got := isProxyFailure(tc.err); got != tc.want {
|
||||
t.Fatalf("isProxyFailure(%v)=%v, want %v", tc.err, got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestGlobalProxyTrafficFailureClassification(t *testing.T) {
|
||||
if !isGlobalProxyDialFailure(errors.New("dial tcp 127.0.0.1:8080: connect: connection refused")) {
|
||||
t.Fatal("upstream dial refusal should be eligible for probe confirmation")
|
||||
}
|
||||
if isGlobalProxyDialFailure(errors.New("status 404: target not found")) {
|
||||
t.Fatal("ordinary target failure must not be treated as proxy failure")
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyPauseCauseIsDistinct(t *testing.T) {
|
||||
ctx, cancel := context.WithCancelCause(context.Background())
|
||||
cancel(agent.AbortPausedByProxy)
|
||||
code, short, _, ok := agent.AbortReason(ctx)
|
||||
if !ok || code != "paused_by_proxy" || short == "" {
|
||||
t.Fatalf("proxy cause = %q/%q/%v", code, short, ok)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyReconnectReasonGate(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
reason string
|
||||
want bool
|
||||
}{
|
||||
{"paused_by_proxy", true},
|
||||
{"paused_by_user", false},
|
||||
{"paused_by_schedule", false},
|
||||
{"paused_by_orchestrator", false},
|
||||
} {
|
||||
got := tc.reason == "paused_by_proxy"
|
||||
if got != tc.want {
|
||||
t.Errorf("reason %q gate=%v, want %v", tc.reason, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyAdmissionRechecksReasonAfterSnapshot(t *testing.T) {
|
||||
m := proxyTestManager(t)
|
||||
task, err := m.CreateTask("proxy gate", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
|
||||
if err := m.ApplyProxyPause(task.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s := newAdmissionTestServer(m, nil)
|
||||
s.engine.Pause(task.ID, agent.AbortPausedByProxy)
|
||||
// Simulate a user re-pause winning after proxyReconnectSuccess took its
|
||||
// snapshot but before the admission lock was acquired.
|
||||
if _, err := m.pg.Exec(`UPDATE tasks SET pause_reason='paused_by_user' WHERE id=$1`, task.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
task.updateLifecycle(func(v *taskLifecycleState) { v.PauseReason = "paused_by_user" })
|
||||
if _, err := s.admitTaskWhenScheduledWithReason(task, "resume", true, false, "paused_by_proxy", true); err == nil {
|
||||
t.Fatal("proxy admission passed after user changed pause reason")
|
||||
}
|
||||
if !s.engine.IsPaused(task.ID) {
|
||||
t.Fatal("user re-pause barrier was cleared")
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyQueuedAdmissionDoesNotResumeEngine(t *testing.T) {
|
||||
m := proxyTestManager(t)
|
||||
if err := m.SetConcurrency(true, 1); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
holder, err := m.CreateTask("holder", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
proxyTask, err := m.CreateTask("queued proxy", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() {
|
||||
_, _ = m.DeleteTask(holder.ID, DeleteTaskOptions{})
|
||||
_, _ = m.DeleteTask(proxyTask.ID, DeleteTaskOptions{})
|
||||
}()
|
||||
if err := m.ApplyProxyPause(proxyTask.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s := newAdmissionTestServer(m, nil)
|
||||
s.engine.started.Store(holder.ID, true)
|
||||
s.engine.Pause(proxyTask.ID, agent.AbortPausedByProxy)
|
||||
queued, err := s.admitTaskWhenScheduledWithReason(proxyTask, "resume", true, false, "paused_by_proxy", true)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !queued {
|
||||
t.Fatal("proxy task was expected to queue behind the running holder")
|
||||
}
|
||||
if !s.engine.IsPaused(proxyTask.ID) {
|
||||
t.Fatal("queued proxy task was resumed")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTerminalStatusClearsProxyReconnectState(t *testing.T) {
|
||||
m := proxyTestManager(t)
|
||||
task, err := m.CreateTask("terminal proxy", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
|
||||
if err := m.ApplyProxyPause(task.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
task.updateLifecycle(func(v *taskLifecycleState) {
|
||||
v.ReconnectAttempt, v.ReconnectMax, v.ReconnectNextAt = 2, 5, 123
|
||||
v.ReconnectExhausted, v.ReconnectLastError = true, "proxy down"
|
||||
})
|
||||
if err := m.SetTaskStatus(task.ID, "timeout"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
stored, err := m.pg.GetTask(mustTaskID(t, task.ID))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if stored.PauseReason != "" {
|
||||
t.Fatalf("terminal row retained pause reason %q", stored.PauseReason)
|
||||
}
|
||||
st := task.lifecycleSnapshot()
|
||||
if st.PauseReason != "" || st.ReconnectAttempt != 0 || st.ReconnectNextAt != 0 || st.ReconnectExhausted {
|
||||
t.Fatalf("terminal lifecycle retained reconnect state: %+v", st)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyAdmissionPreservesCalendarOverride(t *testing.T) {
|
||||
m := proxyTestManager(t)
|
||||
task, err := m.CreateTask("calendar proxy", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
|
||||
if _, err := m.pg.Exec(`UPDATE tasks SET schedule_paused=true, schedule_manual_until='2099-01-01 00:00:00+00' WHERE id=$1`, task.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := m.ApplyProxyPause(task.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s := newAdmissionTestServer(m, nil)
|
||||
s.engine.Pause(task.ID, agent.AbortPausedByProxy)
|
||||
if queued, err := s.admitTaskWhenScheduledWithReason(task, "resume", true, false, "paused_by_proxy", true); err != nil || queued {
|
||||
t.Fatalf("proxy admission queued=%v err=%v", queued, err)
|
||||
}
|
||||
var schedulePaused bool
|
||||
var manualUntil time.Time
|
||||
if err := m.pg.QueryRow(`SELECT schedule_paused, schedule_manual_until FROM tasks WHERE id=$1`, task.ID).Scan(&schedulePaused, &manualUntil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !schedulePaused || !manualUntil.Equal(time.Date(2099, 1, 1, 0, 0, 0, 0, time.UTC)) {
|
||||
t.Fatalf("proxy admission changed calendar state: paused=%v until=%v", schedulePaused, manualUntil)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyReconnectSuccessKeepsCompetingUserPauseBarrier(t *testing.T) {
|
||||
m := proxyTestManager(t)
|
||||
task, err := m.CreateTask("proxy success race", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
|
||||
s := newAdmissionTestServer(m, nil)
|
||||
m.SetTaskControlMutex(&s.concMu)
|
||||
m.globalProxy = "http://proxy.invalid"
|
||||
if _, err := m.ApplyProxyPauseWithCallback(task.ID, func() { s.engine.Pause(task.ID, agent.AbortPausedByProxy) }); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
var pauseErr error
|
||||
s.proxyReconnectSuccessWithHook(func() {
|
||||
_, pauseErr = s.applyTaskControl(task, "pause")
|
||||
})
|
||||
if pauseErr != nil {
|
||||
t.Fatalf("competing user pause failed: %v", pauseErr)
|
||||
}
|
||||
state := task.lifecycleSnapshot()
|
||||
if state.PauseReason != "paused_by_user" || !state.Paused {
|
||||
t.Fatalf("user pause was overwritten: %+v", state)
|
||||
}
|
||||
if !s.engine.IsPaused(task.ID) {
|
||||
t.Fatal("user pause barrier was cleared by a post-admission Resume")
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyPauseBarrierInstalledBeforeControlUnlock(t *testing.T) {
|
||||
m := proxyTestManager(t)
|
||||
task, err := m.CreateTask("proxy barrier", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
|
||||
s := newAdmissionTestServer(m, nil)
|
||||
m.SetTaskControlMutex(&s.concMu)
|
||||
m.globalProxy = "http://proxy.invalid"
|
||||
|
||||
userDone := make(chan error, 1)
|
||||
transitioned, err := m.ApplyProxyPauseWithCallback(task.ID, func() {
|
||||
s.engine.Pause(task.ID, agent.AbortPausedByProxy)
|
||||
if !s.engine.IsPaused(task.ID) {
|
||||
t.Fatal("proxy barrier was not installed in the locked callback")
|
||||
}
|
||||
go func() {
|
||||
_, pauseErr := s.applyTaskControl(task, "pause")
|
||||
userDone <- pauseErr
|
||||
}()
|
||||
select {
|
||||
case <-userDone:
|
||||
t.Fatal("user pause acquired concMu before proxy callback returned")
|
||||
case <-time.After(20 * time.Millisecond):
|
||||
}
|
||||
})
|
||||
if err != nil || !transitioned {
|
||||
t.Fatalf("proxy transition=%v err=%v", transitioned, err)
|
||||
}
|
||||
if err := <-userDone; err != nil {
|
||||
t.Fatalf("concurrent user pause failed: %v", err)
|
||||
}
|
||||
if !s.engine.IsPaused(task.ID) {
|
||||
t.Fatal("concurrent user pause left the engine running")
|
||||
}
|
||||
if got := task.lifecycleSnapshot().PauseReason; got != "paused_by_user" {
|
||||
t.Fatalf("concurrent user pause reason=%q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyPauseDoesNotReviveExhaustedOrNonProxyTasks(t *testing.T) {
|
||||
m := proxyTestManager(t)
|
||||
exhausted, err := m.CreateTask("exhausted proxy", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
userPaused, err := m.CreateTask("user paused", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
other, err := m.CreateTask("new proxy failure", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() {
|
||||
_, _ = m.DeleteTask(exhausted.ID, DeleteTaskOptions{})
|
||||
_, _ = m.DeleteTask(userPaused.ID, DeleteTaskOptions{})
|
||||
_, _ = m.DeleteTask(other.ID, DeleteTaskOptions{})
|
||||
}()
|
||||
s := newAdmissionTestServer(m, nil)
|
||||
m.SetTaskControlMutex(&s.concMu)
|
||||
m.globalProxy = "http://proxy.invalid"
|
||||
if err := m.ApplyProxyPause(exhausted.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := m.ApplyTaskPauseOriginReason(userPaused.ID, "paused_by_user", false); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
exhausted.updateLifecycle(func(v *taskLifecycleState) {
|
||||
v.ReconnectAttempt, v.ReconnectMax, v.ReconnectNextAt = 5, 5, 12345
|
||||
v.ReconnectExhausted, v.ReconnectLastError = true, "proxy down"
|
||||
})
|
||||
userPaused.updateLifecycle(func(v *taskLifecycleState) {
|
||||
v.ReconnectAttempt, v.ReconnectMax, v.ReconnectNextAt = 3, 7, 23456
|
||||
v.ReconnectExhausted, v.ReconnectLastError = true, "manual pause state"
|
||||
})
|
||||
exhaustedBefore := exhausted.lifecycleSnapshot()
|
||||
userBefore := userPaused.lifecycleSnapshot()
|
||||
|
||||
// A no-op proxy pause on an already paused task must not reset either state.
|
||||
s.pauseProxyTask(exhausted.ID, errors.New("another proxy failure"))
|
||||
s.pauseProxyTask(userPaused.ID, errors.New("proxy failure while user paused"))
|
||||
if got := exhausted.lifecycleSnapshot(); got.ReconnectNextAt != exhaustedBefore.ReconnectNextAt || !got.ReconnectExhausted || got.ReconnectLastError != exhaustedBefore.ReconnectLastError {
|
||||
t.Fatalf("exhausted task was revived: before=%+v after=%+v", exhaustedBefore, got)
|
||||
}
|
||||
if got := userPaused.lifecycleSnapshot(); got.PauseReason != "paused_by_user" || got.ReconnectNextAt != userBefore.ReconnectNextAt || !got.ReconnectExhausted {
|
||||
t.Fatalf("user-paused task was mutated: before=%+v after=%+v", userBefore, got)
|
||||
}
|
||||
|
||||
// A genuinely new proxy pause of another running task initializes only that
|
||||
// task's reconnect round.
|
||||
s.pauseProxyTask(other.ID, errors.New("new proxy failure"))
|
||||
if got := exhausted.lifecycleSnapshot(); got.ReconnectNextAt != exhaustedBefore.ReconnectNextAt || !got.ReconnectExhausted {
|
||||
t.Fatalf("subsequent proxy pause revived exhausted task: %+v", got)
|
||||
}
|
||||
if got := other.lifecycleSnapshot(); got.PauseReason != "paused_by_proxy" || got.ReconnectMax == 0 || got.ReconnectExhausted {
|
||||
t.Fatalf("new proxy task did not initialize reconnect state: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyReconnectRoundResetsAfterManualResume(t *testing.T) {
|
||||
m := proxyTestManager(t)
|
||||
task, err := m.CreateTask("proxy retry round", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
|
||||
s := newAdmissionTestServer(m, nil)
|
||||
m.SetTaskControlMutex(&s.concMu)
|
||||
m.globalProxy = "http://proxy.invalid"
|
||||
if err := m.ApplyProxyPause(task.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s.engine.Pause(task.ID, agent.AbortPausedByProxy)
|
||||
task.updateLifecycle(func(v *taskLifecycleState) {
|
||||
v.ReconnectAttempt, v.ReconnectMax, v.ReconnectNextAt = 5, 5, 12345
|
||||
v.ReconnectExhausted, v.ReconnectLastError = true, "proxy exhausted"
|
||||
})
|
||||
if _, err := s.applyTaskControl(task, "resume"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
resumed := task.lifecycleSnapshot()
|
||||
if resumed.PauseReason != "" || resumed.ReconnectAttempt != 0 || resumed.ReconnectMax != 0 || resumed.ReconnectNextAt != 0 || resumed.ReconnectExhausted || resumed.ReconnectLastError != "" {
|
||||
t.Fatalf("manual resume retained reconnect state: %+v", resumed)
|
||||
}
|
||||
s.pauseProxyTask(task.ID, errors.New("proxy failed again"))
|
||||
paused := task.lifecycleSnapshot()
|
||||
if paused.PauseReason != "paused_by_proxy" || paused.ReconnectAttempt != 0 || paused.ReconnectExhausted {
|
||||
t.Fatalf("new proxy round was not reset: %+v", paused)
|
||||
}
|
||||
if paused.ReconnectMax == 0 || paused.ReconnectNextAt == 0 || paused.ReconnectLastError == "" {
|
||||
t.Fatalf("new proxy round did not initialize reconnect state: %+v", paused)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyReconnectDropsStaleProbeFailureAfterNewRound(t *testing.T) {
|
||||
m := proxyTestManager(t)
|
||||
task, err := m.CreateTask("proxy stale probe", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
|
||||
s := newAdmissionTestServer(m, nil)
|
||||
m.SetTaskControlMutex(&s.concMu)
|
||||
m.globalProxy = "http://proxy.invalid"
|
||||
// Keep the canceled test server from starting its supervisor. The test owns
|
||||
// the reconnect tick and injects a blocking probe below.
|
||||
s.proxyReconnect.mu.Lock()
|
||||
s.proxyReconnect.wake = make(chan struct{}, 1)
|
||||
s.proxyReconnect.mu.Unlock()
|
||||
oldMax, hadMax, err := m.pg.GetSetting(settingProxyReconnectMax)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() {
|
||||
if hadMax {
|
||||
_ = m.pg.SetSetting(settingProxyReconnectMax, oldMax)
|
||||
} else {
|
||||
_, _ = m.pg.Exec(`DELETE FROM settings WHERE key=$1`, settingProxyReconnectMax)
|
||||
}
|
||||
}()
|
||||
if err := m.pg.SetSetting(settingProxyReconnectMax, "1"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
|
||||
return egressProbeResult{Error: "initial probe failure"}
|
||||
}
|
||||
s.pauseProxyTask(task.ID, errors.New("initial proxy failure"))
|
||||
task.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectNextAt = 0 })
|
||||
oldRound := task.lifecycleSnapshot().ReconnectGeneration
|
||||
probeStarted := make(chan struct{})
|
||||
releaseProbe := make(chan struct{})
|
||||
var probeStartedOnce sync.Once
|
||||
s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
|
||||
probeStartedOnce.Do(func() { close(probeStarted) })
|
||||
<-releaseProbe
|
||||
return egressProbeResult{Error: "stale probe failure"}
|
||||
}
|
||||
tickDone := make(chan struct{})
|
||||
go func() {
|
||||
s.proxyReconnectTick()
|
||||
close(tickDone)
|
||||
}()
|
||||
select {
|
||||
case <-probeStarted:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("reconnect tick did not start its probe")
|
||||
}
|
||||
if _, err := s.applyTaskControl(task, "resume"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s.pauseProxyTask(task.ID, errors.New("new proxy failure"))
|
||||
newRound := task.lifecycleSnapshot()
|
||||
if newRound.ReconnectGeneration == oldRound || newRound.ReconnectAttempt != 0 || newRound.ReconnectExhausted {
|
||||
t.Fatalf("new proxy round was not established: old=%d new=%+v", oldRound, newRound)
|
||||
}
|
||||
close(releaseProbe)
|
||||
select {
|
||||
case <-tickDone:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("reconnect tick did not finish")
|
||||
}
|
||||
after := task.lifecycleSnapshot()
|
||||
if after.PauseReason != "paused_by_proxy" || after.ReconnectGeneration != newRound.ReconnectGeneration || after.ReconnectAttempt != 0 || after.ReconnectExhausted || after.ReconnectNextAt != newRound.ReconnectNextAt || after.ReconnectLastError != newRound.ReconnectLastError {
|
||||
t.Fatalf("stale probe mutated the new round: before=%+v after=%+v", newRound, after)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyReconnectManualSuccessDoesNotAdmitNewRound(t *testing.T) {
|
||||
m := proxyTestManager(t)
|
||||
task, err := m.CreateTask("proxy stale manual success", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
|
||||
s := newAdmissionTestServer(m, nil)
|
||||
m.SetTaskControlMutex(&s.concMu)
|
||||
m.globalProxy = "http://proxy.invalid"
|
||||
if err := m.ApplyProxyPause(task.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s.engine.Pause(task.ID, agent.AbortPausedByProxy)
|
||||
guards := s.snapshotProxyReconnectGuards(false)
|
||||
task.updateLifecycle(func(v *taskLifecycleState) {
|
||||
v.ReconnectGeneration++
|
||||
v.ReconnectNextAt = 987654
|
||||
v.ReconnectLastError = "new round still failing"
|
||||
})
|
||||
before := task.lifecycleSnapshot()
|
||||
s.proxyReconnectSuccessWithGuards(guards, nil)
|
||||
after := task.lifecycleSnapshot()
|
||||
if after.PauseReason != "paused_by_proxy" || !s.engine.IsPaused(task.ID) {
|
||||
t.Fatalf("stale manual success admitted new round: %+v", after)
|
||||
}
|
||||
if after.ReconnectGeneration != before.ReconnectGeneration || after.ReconnectNextAt != before.ReconnectNextAt || after.ReconnectLastError != before.ReconnectLastError {
|
||||
t.Fatalf("stale manual success wiped new round state: before=%+v after=%+v", before, after)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyReconnectSettingsSuccessGuardsGenerationZero(t *testing.T) {
|
||||
m := proxyTestManager(t)
|
||||
task, err := m.CreateTask("proxy generation zero", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
|
||||
s := newAdmissionTestServer(m, nil)
|
||||
m.SetTaskControlMutex(&s.concMu)
|
||||
m.globalProxy = "http://proxy.invalid"
|
||||
if err := m.ApplyProxyPause(task.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s.engine.Pause(task.ID, agent.AbortPausedByProxy)
|
||||
initial := task.lifecycleSnapshot()
|
||||
if initial.ReconnectGeneration != 0 {
|
||||
t.Fatalf("fresh persisted proxy pause generation=%d, want zero", initial.ReconnectGeneration)
|
||||
}
|
||||
guards := s.snapshotProxyReconnectGuards(false)
|
||||
task.updateLifecycle(func(v *taskLifecycleState) {
|
||||
v.ReconnectGeneration = 1
|
||||
v.ReconnectNextAt = 13579
|
||||
v.ReconnectLastError = "generation one failure"
|
||||
})
|
||||
before := task.lifecycleSnapshot()
|
||||
s.proxyReconnectSuccessWithGuards(guards, nil)
|
||||
after := task.lifecycleSnapshot()
|
||||
if after.PauseReason != "paused_by_proxy" || !s.engine.IsPaused(task.ID) {
|
||||
t.Fatalf("generation-zero stale settings success admitted new round: %+v", after)
|
||||
}
|
||||
if after.ReconnectGeneration != before.ReconnectGeneration || after.ReconnectNextAt != before.ReconnectNextAt || after.ReconnectLastError != before.ReconnectLastError {
|
||||
t.Fatalf("generation-zero stale settings success wiped state: before=%+v after=%+v", before, after)
|
||||
}
|
||||
}
|
||||
+39
-11
@@ -128,7 +128,11 @@ type Server struct {
|
||||
archiveWake chan struct{}
|
||||
archiveWG sync.WaitGroup
|
||||
credentialWake chan struct{}
|
||||
side *sideQuestionState
|
||||
proxyReconnect proxyReconnectSupervisor
|
||||
// proxyProbe is injectable for deterministic reconnect race tests. Production
|
||||
// servers leave it nil and use probeEgress directly.
|
||||
proxyProbe func(context.Context, string, string, []string, time.Duration) egressProbeResult
|
||||
side *sideQuestionState
|
||||
}
|
||||
|
||||
// provEntry is a cached provider + its config for one LLM profile id.
|
||||
@@ -163,6 +167,7 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
|
||||
profChatAgents: map[int64]*agent.ChatAgent{},
|
||||
provByProfile: map[int64]*provEntry{}, llmHealth: newLLMHealthRegistry(m.pg),
|
||||
taskAgents: map[string]*taskAgentBundle{}, archiveWake: make(chan struct{}, 1), credentialWake: make(chan struct{}, 1)}
|
||||
m.SetTaskControlMutex(&s.concMu)
|
||||
s.initSideQuestions()
|
||||
if m.pg != nil {
|
||||
if err := s.basicAuth.load(m.pg); err != nil {
|
||||
@@ -192,6 +197,10 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
|
||||
defer s.cfgMu.Unlock()
|
||||
return s.llmOn
|
||||
})
|
||||
s.engine.SetProxyFailureHandler(s.handleProxyFailure)
|
||||
if m.traffic != nil {
|
||||
m.traffic.SetUpstreamFailureHandler(s.handleProxyTrafficFailure)
|
||||
}
|
||||
// Wire DB-stored prompt templates into the agents (新版方案 §3.3 / §5a). With no
|
||||
// override row, agents keep their built-in defaults — behavior is unchanged.
|
||||
if m.pg != nil {
|
||||
@@ -280,6 +289,7 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
|
||||
log.Printf("[engine] no LLM provider configured — engine idle until set via /api/llm or env")
|
||||
}
|
||||
s.restoreTaskRuntimes()
|
||||
s.ensureProxySupervisor()
|
||||
go s.reconcileConcurrency()
|
||||
s.startTaskArchiveWorker()
|
||||
if m.pg != nil {
|
||||
@@ -927,6 +937,7 @@ func (s *Server) Handler() http.Handler {
|
||||
mux.HandleFunc("POST /api/settings/web-search/test", s.testWebSearch)
|
||||
// 出口代理连通性测试:发真实请求验证 global_proxy 是否可用,并回显出口 IP。
|
||||
mux.HandleFunc("POST /api/settings/proxy/test", s.testEgressProxy)
|
||||
mux.HandleFunc("POST /api/tasks/{id}/proxy/test", s.testTaskProxy)
|
||||
mux.HandleFunc("GET /api/report", s.getReport)
|
||||
mux.HandleFunc("GET /api/chat/mentions", s.searchChatMentions)
|
||||
mux.HandleFunc("POST /api/chat", s.chat)
|
||||
@@ -3540,6 +3551,8 @@ func (s *Server) settingsPayload() map[string]any {
|
||||
"tavily_key_set": strings.TrimSpace(tavilyKey) != "",
|
||||
"web_search_proxy": proxy, // 独立出口代理(http/https/socks5),空=直连
|
||||
"global_proxy": s.m.GlobalProxy(), // 全局出口代理(http/https/socks5),所有目标流量走它,空=直连
|
||||
"proxy_reconnect_max_attempts": func() int { n, _ := s.proxyReconnectConfig(); return n }(),
|
||||
"proxy_reconnect_interval_seconds": func() int { _, d := s.proxyReconnectConfig(); return int(d / time.Second) }(),
|
||||
"show_egress_ip": s.m.pg.GetBool(settingShowEgressIP, false),
|
||||
"python_interpreter": strings.TrimSpace(pyStored), // 用户/自动设的值(空=用运行时检测)
|
||||
"workers": s.m.Workers(), // 并发工作 agent 数(默认3);对之后启动的任务生效
|
||||
@@ -3616,16 +3629,18 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
|
||||
LLMRecord *bool `json:"llm_record"` // LLM 录制开关(默认关);即时生效,无需重建 agent
|
||||
// Web search. WebSearchEnabled/Backend toggle the tool + backend; BraveKey/TavilyKey
|
||||
// are optional — omit (null) to leave a stored key untouched, send "" to clear.
|
||||
WebSearchEnabled *bool `json:"web_search_enabled"`
|
||||
WebSearchBackend *string `json:"web_search_backend"`
|
||||
BraveKey *string `json:"brave_search_api_key"`
|
||||
TavilyKey *string `json:"tavily_search_api_key"`
|
||||
WebSearchProxy *string `json:"web_search_proxy"` // 独立出口代理(http/https/socks5);null=不改,""=清空
|
||||
GlobalProxy *string `json:"global_proxy"` // 全局出口代理(http/https/socks5);null=不改,""=清空(直连)
|
||||
ShowEgressIP *bool `json:"show_egress_ip"` // 在顶栏显示直连/全局代理出口 IP;默认关
|
||||
PythonInterp *string `json:"python_interpreter"` // 自定义脚本工具的 python 解释器路径
|
||||
Workers *int `json:"workers"` // 并发工作 agent 数(>0);对之后启动的任务生效
|
||||
WorkerPools *[]WorkerPool `json:"worker_pools"`
|
||||
WebSearchEnabled *bool `json:"web_search_enabled"`
|
||||
WebSearchBackend *string `json:"web_search_backend"`
|
||||
BraveKey *string `json:"brave_search_api_key"`
|
||||
TavilyKey *string `json:"tavily_search_api_key"`
|
||||
WebSearchProxy *string `json:"web_search_proxy"` // 独立出口代理(http/https/socks5);null=不改,""=清空
|
||||
GlobalProxy *string `json:"global_proxy"` // 全局出口代理(http/https/socks5);null=不改,""=清空(直连)
|
||||
ProxyReconnectMax *int `json:"proxy_reconnect_max_attempts"`
|
||||
ProxyReconnectInterval *int `json:"proxy_reconnect_interval_seconds"`
|
||||
ShowEgressIP *bool `json:"show_egress_ip"` // 在顶栏显示直连/全局代理出口 IP;默认关
|
||||
PythonInterp *string `json:"python_interpreter"` // 自定义脚本工具的 python 解释器路径
|
||||
Workers *int `json:"workers"` // 并发工作 agent 数(>0);对之后启动的任务生效
|
||||
WorkerPools *[]WorkerPool `json:"worker_pools"`
|
||||
// 任务并发上限:同时「运行中」的任务数上限。关闭=不限;开启后新建任务超限则排队,有空位自动启动。
|
||||
ConcurrencyEnabled *bool `json:"task_concurrency_enabled"`
|
||||
ConcurrencyLimit *int `json:"task_concurrency_limit"`
|
||||
@@ -3836,12 +3851,25 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
|
||||
changed = true
|
||||
}
|
||||
if req.GlobalProxy != nil {
|
||||
var proxyGuards map[string]proxyReconnectGuard
|
||||
if strings.TrimSpace(*req.GlobalProxy) == "" {
|
||||
proxyGuards = s.snapshotProxyReconnectGuards(false)
|
||||
}
|
||||
// Validation failure (bad scheme/host) is a client error, not a 500.
|
||||
if err := s.m.SetGlobalProxy(*req.GlobalProxy); err != nil {
|
||||
writeErr(w, 400, err.Error())
|
||||
return
|
||||
}
|
||||
changed = true // capture-off egress is baked into agents at build time → rebuild
|
||||
if strings.TrimSpace(*req.GlobalProxy) == "" {
|
||||
s.proxyReconnectSuccessWithGuards(proxyGuards, nil)
|
||||
}
|
||||
}
|
||||
if req.ProxyReconnectMax != nil || req.ProxyReconnectInterval != nil {
|
||||
if err := s.setProxyReconnectSettings(req.ProxyReconnectMax, req.ProxyReconnectInterval); err != nil {
|
||||
writeErr(w, 400, err.Error())
|
||||
return
|
||||
}
|
||||
}
|
||||
if req.ShowEgressIP != nil {
|
||||
if err := s.m.pg.SetBool(settingShowEgressIP, *req.ShowEgressIP); err != nil {
|
||||
|
||||
@@ -101,7 +101,14 @@ func (s *Server) beginTaskDelete(taskID string) bool {
|
||||
if s.engine.IsDeleting(taskID) {
|
||||
return false
|
||||
}
|
||||
return s.engine.BeginDelete(taskID)
|
||||
ok := s.engine.BeginDelete(taskID)
|
||||
if ok {
|
||||
// A user delete/archive action is an explicit override of proxy recovery.
|
||||
if s.m != nil && s.m.pg != nil {
|
||||
_ = s.m.ClearPauseReason(taskID)
|
||||
}
|
||||
}
|
||||
return ok
|
||||
}
|
||||
|
||||
// abortTaskDelete restores the execution barrier from the committed task row,
|
||||
|
||||
+39
-3
@@ -116,9 +116,31 @@ func (s *Server) applyTaskControlWithCause(t *Task, action string, pauseCause er
|
||||
}
|
||||
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 {
|
||||
if pauseCause == nil {
|
||||
pauseCause = agent.AbortPausedByUser
|
||||
}
|
||||
reason := "paused_by_user"
|
||||
if pauseCause == agent.AbortPausedByOrchestrator {
|
||||
reason = "paused_by_orchestrator"
|
||||
}
|
||||
if pauseCause == agent.AbortPausedByProxy {
|
||||
reason = "paused_by_proxy"
|
||||
}
|
||||
if _, err := s.m.pg.Exec(`UPDATE tasks SET schedule_paused=false, pause_reason=$2 WHERE id=$1`, t.ID, reason); err != nil {
|
||||
return out, err
|
||||
}
|
||||
t.updateLifecycle(func(state *taskLifecycleState) {
|
||||
state.PauseReason = reason
|
||||
state.ReconnectAttempt = 0
|
||||
state.ReconnectMax = 0
|
||||
state.ReconnectNextAt = 0
|
||||
state.ReconnectExhausted = false
|
||||
state.ReconnectLastError = ""
|
||||
})
|
||||
// The lifecycle row may already have been paused by another
|
||||
// actor. Reinstall the requested cause barrier while holding
|
||||
// concMu instead of assuming the existing barrier is sufficient.
|
||||
s.engine.Pause(t.ID, pauseCause)
|
||||
out.Paused = true
|
||||
out.Status = "paused"
|
||||
return out, nil
|
||||
@@ -131,8 +153,22 @@ func (s *Server) applyTaskControlWithCause(t *Task, action string, pauseCause er
|
||||
pauseCause = agent.AbortPausedByUser
|
||||
}
|
||||
s.engine.Pause(t.ID, pauseCause)
|
||||
if err := s.m.ApplyTaskPauseOrigin(t.ID, pauseCause == agent.AbortPausedBySchedule); err != nil {
|
||||
if !wasEnginePaused && !wasQueued {
|
||||
reason := "paused_by_user"
|
||||
if pauseCause == agent.AbortPausedBySchedule {
|
||||
reason = "paused_by_schedule"
|
||||
}
|
||||
if pauseCause == agent.AbortPausedByOrchestrator {
|
||||
reason = "paused_by_orchestrator"
|
||||
}
|
||||
if pauseCause == agent.AbortPausedByProxy {
|
||||
reason = "paused_by_proxy"
|
||||
}
|
||||
if err := s.m.ApplyTaskPauseOriginReason(t.ID, reason, pauseCause == agent.AbortPausedBySchedule); err != nil {
|
||||
// Do not compensate over a proxy pause that may have committed while
|
||||
// this control request was failing. The persisted reason is the source
|
||||
// of truth for which Engine barrier must remain installed.
|
||||
stillProxyPaused := t.lifecycleSnapshot().PauseReason == "paused_by_proxy"
|
||||
if !stillProxyPaused && !wasEnginePaused && !wasQueued {
|
||||
s.engine.Resume(t)
|
||||
}
|
||||
return out, err
|
||||
|
||||
+32
-2
@@ -169,7 +169,9 @@ type Traffic struct {
|
||||
// through (nil = dial targets directly). Both the intercepted and the
|
||||
// transparent-passthrough paths honor it (go-mitmproxy's getUpstreamConn),
|
||||
// so no host escapes it. Hot-swappable at runtime via SetUpstreamProxy.
|
||||
upstream atomic.Pointer[url.URL]
|
||||
upstream atomic.Pointer[url.URL]
|
||||
upstreamMu sync.RWMutex
|
||||
upstreamFailure func(error)
|
||||
}
|
||||
|
||||
// Open initializes the traffic tree, blob store and SQLite index under dir.
|
||||
@@ -327,6 +329,27 @@ func (t *Traffic) SetUpstreamProxy(raw string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// SetUpstreamFailureHandler observes failures from the configured global
|
||||
// upstream proxy. The callback is intentionally best-effort and must not block
|
||||
// the MITM request path.
|
||||
func (t *Traffic) SetUpstreamFailureHandler(fn func(error)) {
|
||||
t.upstreamMu.Lock()
|
||||
t.upstreamFailure = fn
|
||||
t.upstreamMu.Unlock()
|
||||
}
|
||||
|
||||
func (t *Traffic) reportUpstreamFailure(err error) {
|
||||
if err == nil || t.upstream.Load() == nil {
|
||||
return
|
||||
}
|
||||
t.upstreamMu.RLock()
|
||||
fn := t.upstreamFailure
|
||||
t.upstreamMu.RUnlock()
|
||||
if fn != nil {
|
||||
go fn(err)
|
||||
}
|
||||
}
|
||||
|
||||
// ValidateProxyURL parses and checks a proxy URL (http/https/socks5, optional
|
||||
// user:pass), returning the parsed URL. Exposed so callers can validate a global
|
||||
// proxy before persisting it even when the traffic proxy itself is disabled.
|
||||
@@ -400,7 +423,14 @@ func (s *sink) Response(f *mproxy.Flow) {
|
||||
// the failure looks proxy/protocol-caused (h2 quirks, HEAD-with-body, protocol
|
||||
// errors) — not a plain target-unreachable error — we flag the host for
|
||||
// transparent passthrough so future requests to it succeed instead of dying.
|
||||
func (s *sink) RequestError(f *mproxy.Flow, err error) { s.t.maybePassthrough(f, err) }
|
||||
func (s *sink) RequestError(f *mproxy.Flow, err error) {
|
||||
s.t.reportUpstreamFailure(err)
|
||||
s.t.maybePassthrough(f, err)
|
||||
}
|
||||
|
||||
func (s *sink) HTTPConnectError(_ *mproxy.Flow, err error) {
|
||||
s.t.reportUpstreamFailure(err)
|
||||
}
|
||||
|
||||
// maybePassthrough marks a host to be tunneled transparently on the next
|
||||
// connection, but only for errors the proxy itself caused — a target that is
|
||||
|
||||
@@ -419,6 +419,14 @@ function TaskDetailInner() {
|
||||
</Button>
|
||||
</div>
|
||||
<p className="truncate text-xs text-muted-foreground">{task.goal}</p>
|
||||
{task.paused && task.pause_reason && (
|
||||
<div className="flex flex-wrap items-center gap-2 text-xs text-amber-600">
|
||||
<span>暂停原因:{task.pause_reason === "paused_by_proxy" ? "全局代理异常" : task.pause_reason}</span>
|
||||
{task.reconnect && <span>重连 {task.reconnect.attempt}/{task.reconnect.max} {task.reconnect.exhausted ? "已耗尽" : task.reconnect.next_at ? `下次 ${new Date(task.reconnect.next_at * 1000).toLocaleTimeString()}` : "探测中"}</span>}
|
||||
{task.reconnect?.last_error && <span className="max-w-full truncate" title={task.reconnect.last_error}>原因:{task.reconnect.last_error}</span>}
|
||||
{task.pause_reason === "paused_by_proxy" && <Button size="sm" variant="outline" onClick={() => api.testTaskProxy(task.id).then((r) => r.ok ? (toast.success("代理已恢复,任务将自动继续"), load()) : toast.error(r.error || "代理探测失败")).catch((e) => toast.error((e as Error).message))}>测试代理</Button>}
|
||||
</div>
|
||||
)}
|
||||
{task.remark ? (
|
||||
<p className="truncate text-xs text-muted-foreground" title={task.remark}>
|
||||
备注:{task.remark}
|
||||
|
||||
@@ -55,6 +55,8 @@ export default function SystemSettingsPage() {
|
||||
const [savingProxy, setSavingProxy] = React.useState(false);
|
||||
const [globalProxyInput, setGlobalProxyInput] = React.useState("");
|
||||
const [savingGlobalProxy, setSavingGlobalProxy] = React.useState(false);
|
||||
const [proxyReconnectMax, setProxyReconnectMax] = React.useState("5");
|
||||
const [proxyReconnectInterval, setProxyReconnectInterval] = React.useState("30");
|
||||
const [savingEgressVisibility, setSavingEgressVisibility] = React.useState(false);
|
||||
// 出口探测:测一次真实请求,拿到"直连 / 经全局代理"两条出口 IP。
|
||||
const [egressTesting, setEgressTesting] = React.useState(false);
|
||||
@@ -111,6 +113,8 @@ export default function SystemSettingsPage() {
|
||||
setProxyInput(s.web_search_proxy ?? "");
|
||||
const savedProxy = s.global_proxy ?? "";
|
||||
setGlobalProxyInput(savedProxy);
|
||||
setProxyReconnectMax(String(s.proxy_reconnect_max_attempts ?? 5));
|
||||
setProxyReconnectInterval(String(s.proxy_reconnect_interval_seconds ?? 30));
|
||||
setPyInterp(s.python_interpreter ?? "");
|
||||
setWorkers(String(s.workers ?? 3));
|
||||
setWorkerPools(s.worker_pools ?? []);
|
||||
@@ -428,6 +432,12 @@ export default function SystemSettingsPage() {
|
||||
.finally(() => setSavingGlobalProxy(false));
|
||||
};
|
||||
|
||||
const saveProxyReconnect = () => {
|
||||
api.setSettings({ proxy_reconnect_max_attempts: Number(proxyReconnectMax), proxy_reconnect_interval_seconds: Number(proxyReconnectInterval) })
|
||||
.then((s) => { apply(s); toast.success("已保存代理重连策略"); })
|
||||
.catch((e) => toast.error("保存失败:" + (e as Error).message));
|
||||
};
|
||||
|
||||
// 测表单里当前的值,不要求先保存;后端返回直连 / 经代理两条。
|
||||
const testEgress = () => {
|
||||
const seq = ++egressSeq.current;
|
||||
@@ -751,12 +761,18 @@ export default function SystemSettingsPage() {
|
||||
</Button>
|
||||
{/* 测表单里当前的值,不要求先保存。 */}
|
||||
<Button type="button" variant="outline" onClick={testEgress} disabled={!loaded || egressTesting}>
|
||||
{egressTesting ? "测试中…" : "测试"}
|
||||
{egressTesting ? "测试中…" : "测试代理"}
|
||||
</Button>
|
||||
</div>
|
||||
<p className="text-muted-foreground text-xs">
|
||||
{globalProxyInput.trim() ? "已配置 · 所有目标流量经此代理出网" : "未配置 · 目标流量直连出网"}
|
||||
</p>
|
||||
<div className="grid grid-cols-1 gap-2 sm:grid-cols-3 items-end">
|
||||
<div><Label htmlFor="proxy-reconnect-max">最大重连次数</Label><Input id="proxy-reconnect-max" type="number" min={1} value={proxyReconnectMax} onChange={(e) => setProxyReconnectMax(e.target.value)} /></div>
|
||||
<div><Label htmlFor="proxy-reconnect-interval">重连间隔(秒)</Label><Input id="proxy-reconnect-interval" type="number" min={1} value={proxyReconnectInterval} onChange={(e) => setProxyReconnectInterval(e.target.value)} /></div>
|
||||
<Button type="button" variant="outline" onClick={saveProxyReconnect}>保存重连策略</Button>
|
||||
</div>
|
||||
<p className="text-muted-foreground text-xs">代理异常暂停的任务会按此策略探测;次数耗尽后保持暂停,可手动恢复。</p>
|
||||
<div className="flex items-center justify-between gap-4 rounded-md border px-3 py-2">
|
||||
<div className="min-w-0">
|
||||
<Label htmlFor="show-egress-ip" className="text-sm font-normal">
|
||||
|
||||
@@ -992,6 +992,7 @@ export const api = {
|
||||
// global_proxy 传空 = 测直连;probe_url 可覆盖默认探测地址(隔离网用)。
|
||||
testEgressProxy: (patch: { global_proxy?: string; probe_url?: string }) =>
|
||||
post<EgressReport>(`/settings/proxy/test`, patch),
|
||||
testTaskProxy: (taskID: string) => post<{ ok: boolean; error?: string }>(`/tasks/${taskID}/proxy/test`, {}),
|
||||
|
||||
// ---- 漏洞 IM 推送 ----
|
||||
// 渠道是多实例资源(同一类型可配多个机器人、各有过滤规则),因此独立成组,
|
||||
|
||||
@@ -42,6 +42,8 @@ export interface Task {
|
||||
active_run_seconds?: number; // completed running segments in seconds
|
||||
run_segment_started_unix?: number; // current running segment start; 0 when inactive
|
||||
paused?: boolean;
|
||||
pause_reason?: string;
|
||||
reconnect?: { attempt: number; max: number; next_at?: number; exhausted: boolean; last_error?: string };
|
||||
queued?: boolean;
|
||||
active?: boolean;
|
||||
in_flight?: number;
|
||||
@@ -1002,6 +1004,8 @@ export interface Settings {
|
||||
// 全局出口代理(http/https/socks5,可带 user:pass),所有目标流量走它。开启流量捕获时作为
|
||||
// MITM 上游;关闭捕获时直接注入 agent 的 bash/WebFetch。空=直连。
|
||||
global_proxy?: string;
|
||||
proxy_reconnect_max_attempts?: number;
|
||||
proxy_reconnect_interval_seconds?: number;
|
||||
show_egress_ip?: boolean; // 在顶栏显示直连 / 全局代理出口 IP;默认关闭
|
||||
python_interpreter?: string; // 自定义脚本工具的 python 解释器路径(空=运行时检测)
|
||||
workers?: number; // 并发工作 agent 数(默认3);对之后启动的任务生效
|
||||
|
||||
Reference in new issue
Block a user