diff --git a/agent/cancelcause.go b/agent/cancelcause.go index 207a5cc..90abe6f 100644 --- a/agent/cancelcause.go +++ b/agent/cancelcause.go @@ -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 暂停了任务", diff --git a/agent/terminalreason_test.go b/agent/terminalreason_test.go index ff53fe4..2a73501 100644 --- a/agent/terminalreason_test.go +++ b/agent/terminalreason_test.go @@ -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, diff --git a/db/schema.sql b/db/schema.sql index dd39b7c..37f4041 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -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; diff --git a/db/tasks.go b/db/tasks.go index c618541..4f0007f 100644 --- a/db/tasks.go +++ b/db/tasks.go @@ -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 @@ -409,11 +410,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 @@ -616,6 +617,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 @@ -642,6 +644,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 diff --git a/server/calendar_schedule.go b/server/calendar_schedule.go index 02a85a3..543f63d 100644 --- a/server/calendar_schedule.go +++ b/server/calendar_schedule.go @@ -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" diff --git a/server/customtool.go b/server/customtool.go index ad68517..cfa9706 100644 --- a/server/customtool.go +++ b/server/customtool.go @@ -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() diff --git a/server/dto.go b/server/dto.go index f1c058e..9eead79 100644 --- a/server/dto.go +++ b/server/dto.go @@ -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, diff --git a/server/egress_probe.go b/server/egress_probe.go index 70e764b..e3ea0e6 100644 --- a/server/egress_probe.go +++ b/server/egress_probe.go @@ -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) } diff --git a/server/engine.go b/server/engine.go index 04f610c..df500ba 100644 --- a/server/engine.go +++ b/server/engine.go @@ -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: diff --git a/server/engine_cancelcause_test.go b/server/engine_cancelcause_test.go index d1f4f7a..a463f6f 100644 --- a/server/engine_cancelcause_test.go +++ b/server/engine_cancelcause_test.go @@ -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") } diff --git a/server/goals.go b/server/goals.go index 45b67f6..6791744 100644 --- a/server/goals.go +++ b/server/goals.go @@ -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) } diff --git a/server/manager.go b/server/manager.go index 83a6ee4..69c1a96 100644 --- a/server/manager.go +++ b/server/manager.go @@ -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" @@ -1076,7 +1104,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, @@ -1394,12 +1422,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) @@ -1423,8 +1455,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 @@ -1447,7 +1480,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) @@ -1461,6 +1494,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 @@ -1484,6 +1518,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() @@ -1497,6 +1536,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) @@ -1507,12 +1552,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) } @@ -1523,6 +1568,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 @@ -1537,6 +1588,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() @@ -1751,6 +1906,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() } @@ -1787,6 +1946,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() } diff --git a/server/proxy_autoresume.go b/server/proxy_autoresume.go new file mode 100644 index 0000000..e17982d --- /dev/null +++ b/server/proxy_autoresume.go @@ -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 +} diff --git a/server/proxy_autoresume_test.go b/server/proxy_autoresume_test.go new file mode 100644 index 0000000..194bae2 --- /dev/null +++ b/server/proxy_autoresume_test.go @@ -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) + } +} diff --git a/server/server.go b/server/server.go index 31c5a72..8cbf3c1 100644 --- a/server/server.go +++ b/server/server.go @@ -129,7 +129,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. @@ -167,6 +171,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 { @@ -196,6 +201,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 { @@ -284,6 +293,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 { @@ -982,6 +992,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) @@ -3637,6 +3648,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)；对之后启动的任务生效 @@ -3713,16 +3726,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"` @@ -3933,12 +3948,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 { diff --git a/server/server_mgmt.go b/server/server_mgmt.go index 1a1a242..cd8d595 100644 --- a/server/server_mgmt.go +++ b/server/server_mgmt.go @@ -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, diff --git a/server/task_control.go b/server/task_control.go index 1cc0e15..005e2f8 100644 --- a/server/task_control.go +++ b/server/task_control.go @@ -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 diff --git a/traffic/traffic.go b/traffic/traffic.go index bebdb1e..72a250a 100644 --- a/traffic/traffic.go +++ b/traffic/traffic.go @@ -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 diff --git a/web/src/app/(main)/function/tasks/detail/page.tsx b/web/src/app/(main)/function/tasks/detail/page.tsx index be52acc..27e9e12 100644 --- a/web/src/app/(main)/function/tasks/detail/page.tsx +++ b/web/src/app/(main)/function/tasks/detail/page.tsx @@ -419,6 +419,14 @@ function TaskDetailInner() {

{task.goal}

+ {task.paused && task.pause_reason && ( +
+ 暂停原因：{task.pause_reason === "paused_by_proxy" ? "全局代理异常" : task.pause_reason} + {task.reconnect && 重连 {task.reconnect.attempt}/{task.reconnect.max} {task.reconnect.exhausted ? "已耗尽" : task.reconnect.next_at ? `下次 ${new Date(task.reconnect.next_at * 1000).toLocaleTimeString()}` : "探测中"}} + {task.reconnect?.last_error && 原因：{task.reconnect.last_error}} + {task.pause_reason === "paused_by_proxy" && } +
+ )} {task.remark ? (

备注：{task.remark} diff --git a/web/src/app/(main)/system/settings/page.tsx b/web/src/app/(main)/system/settings/page.tsx index 618c1ec..8c96d5c 100644 --- a/web/src/app/(main)/system/settings/page.tsx +++ b/web/src/app/(main)/system/settings/page.tsx @@ -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() { {/* 测表单里当前的值，不要求先保存。 */}

{globalProxyInput.trim() ? "已配置 · 所有目标流量经此代理出网" : "未配置 · 目标流量直连出网"}

+
+
setProxyReconnectMax(e.target.value)} />
+
setProxyReconnectInterval(e.target.value)} />
+ +
+

代理异常暂停的任务会按此策略探测；次数耗尽后保持暂停，可手动恢复。