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 593ef41..ca78a98 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 ac5c19e..4d88586 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 @@ -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 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/engine.go b/server/engine.go index 04f610c..58b8464 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 @@ -962,6 +965,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 会优先展示该终态）。 @@ -1124,6 +1130,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、任务 // 未暂停/未终止/未取消【且未进入收尾】时重试；否则让位给对应分支处理(收尾期不 diff --git a/server/manager.go b/server/manager.go index d6a6545..0aaeda4 100644 --- a/server/manager.go +++ b/server/manager.go @@ -28,19 +28,25 @@ 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:"-"` + 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 +94,12 @@ type taskLifecycleState struct { PinnedAt int64 Status string Paused bool + PauseReason string + ReconnectAttempt int + ReconnectMax int + ReconnectNextAt int64 + ReconnectExhausted bool + ReconnectLastError string Queued bool QueuedAt int64 QueueMode string @@ -113,11 +125,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, Queued: t.Queued, QueuedAt: t.QueuedAt, QueueMode: t.QueueMode, @@ -145,6 +159,12 @@ 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.Queued = state.Queued t.QueuedAt = state.QueuedAt t.QueueMode = state.QueueMode @@ -1073,7 +1093,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, @@ -1420,7 +1440,7 @@ func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued b WHEN $2 IN ('done','failed','timeout') THEN COALESCE(completed_at, now()) ELSE NULL END, - paused=false, schedule_paused=false, + paused=false, pause_reason='', schedule_paused=false, schedule_manual_until=COALESCE($7::timestamptz, schedule_manual_until), queued=$3, queued_at=CASE @@ -1458,6 +1478,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 @@ -1494,6 +1515,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 +1531,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 +1547,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 +1567,60 @@ 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 { + m.taskStateMu.Lock() + defer m.taskStateMu.Unlock() + n, err := strconv.ParseInt(id, 10, 64) + if err != nil { + return 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, + schedule_paused=false, 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).Scan(&mode, &active, &started) + if errors.Is(err, sql.ErrNoRows) { + return nil + } + if err != nil { + return 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 + }) + } + return 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 = "" }) + } + return nil +} + // EnqueueTask persists the concurrency hold and syncs the in-memory handle. func (m *Manager) EnqueueTask(id, mode string) error { m.taskStateMu.Lock() diff --git a/server/proxy_autoresume.go b/server/proxy_autoresume.go new file mode 100644 index 0000000..1fbb101 --- /dev/null +++ b/server/proxy_autoresume.go @@ -0,0 +1,242 @@ +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 deliberately requires a configured proxy at the call site and +// only matches transport failures. Provider quota/auth errors must keep their +// existing retry/failover behavior. +func isProxyFailure(err error) bool { + if err == nil { + return false + } + s := strings.ToLower(err.Error()) + for _, part := range []string{"proxyconnect", "proxy connection", "proxy error", "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{} +} + +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) handleProxyFailure(taskID string, cause error) { + if strings.TrimSpace(s.m.GlobalProxy()) == "" { + return + } + if err := s.m.ApplyProxyPause(taskID); err != nil { + log.Printf("[proxy] pause task %s: %v", taskID, err) + return + } + if s.engine != nil { + if t, ok := s.m.Task(taskID); ok && t.lifecycleSnapshot().PauseReason == "paused_by_proxy" { + s.engine.Pause(taskID, agent.AbortPausedByProxy) + } + } + s.cancelTaskChat(taskID, agent.AbortChatPausedWithTask) + max, interval := s.proxyReconnectConfig() + now := time.Now().Add(interval).Unix() + for _, t := range s.m.List() { + if t.ID != taskID && t.lifecycleSnapshot().PauseReason != "paused_by_proxy" { + continue + } + t.updateLifecycle(func(st *taskLifecycleState) { + st.ReconnectMax = max + st.ReconnectNextAt = now + st.ReconnectExhausted = false + st.ReconnectLastError = truncateProxyError(cause) + }) + } + s.ensureProxySupervisor() + s.signalProxySupervisor() +} + +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 st.PauseReason != "paused_by_proxy" || st.ReconnectExhausted || (st.ReconnectNextAt > 0 && st.ReconnectNextAt > now) { + continue + } + st.ReconnectAttempt++ + st.ReconnectMax = max + t.updateLifecycle(func(v *taskLifecycleState) { + v.ReconnectAttempt = st.ReconnectAttempt + v.ReconnectMax = max + v.ReconnectNextAt = 0 + }) + res := probeEgress(context.Background(), s.m.GlobalProxy(), "", defaultEgressProbes, 10*time.Second) + if res.OK { + s.proxyReconnectSuccess() + return + } + err := res.Error + exhausted := st.ReconnectAttempt >= max + next := time.Now().Add(interval).Unix() + t.updateLifecycle(func(v *taskLifecycleState) { + v.ReconnectNextAt = next + v.ReconnectExhausted = exhausted + v.ReconnectLastError = err + }) + log.Printf("[proxy] reconnect attempt %d/%d failed: %s", st.ReconnectAttempt, max, err) + } +} + +func (s *Server) proxyReconnectSuccess() { + for _, t := range s.m.List() { + st := t.lifecycleSnapshot() + if st.PauseReason != "paused_by_proxy" { + continue + } + var err error + if s.engine != nil { + _, err = s.admitTaskWhenScheduled(t, "resume", true, false) + } else { + err = s.m.ClearPauseReason(t.ID) + } + if err != nil { + log.Printf("[proxy] resume task %s: %v", t.ID, err) + continue + } + if s.engine != nil { + s.engine.Resume(t) + } + t.updateLifecycle(func(v *taskLifecycleState) { + 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()) + if proxy == "" { + writeJSON(w, 200, map[string]any{"ok": true, "direct": true}) + s.proxyReconnectSuccess() + return + } + res := probeEgress(r.Context(), proxy, "", defaultEgressProbes, 10*time.Second) + if res.OK { + s.proxyReconnectSuccess() + } + 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..33137a7 --- /dev/null +++ b/server/proxy_autoresume_test.go @@ -0,0 +1,54 @@ +package server + +import ( + "context" + "errors" + "testing" + + "github.com/Autumn-27/artex/agent" +) + +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"), true}, + {"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 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) + } + } +} diff --git a/server/server.go b/server/server.go index efc4375..5cbc675 100644 --- a/server/server.go +++ b/server/server.go @@ -128,6 +128,7 @@ type Server struct { archiveWake chan struct{} archiveWG sync.WaitGroup credentialWake chan struct{} + proxyReconnect proxyReconnectSupervisor side *sideQuestionState } @@ -192,6 +193,7 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi defer s.cfgMu.Unlock() return s.llmOn }) + s.engine.SetProxyFailureHandler(s.handleProxyFailure) // 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 +282,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 +930,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 +3544,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 +3622,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"` @@ -3842,6 +3850,15 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) { return } changed = true // capture-off egress is baked into agents at build time → rebuild + if strings.TrimSpace(*req.GlobalProxy) == "" { + s.proxyReconnectSuccess() + } + } + 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 b851ca7..a4fbdf5 100644 --- a/server/server_mgmt.go +++ b/server/server_mgmt.go @@ -101,7 +101,12 @@ 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. + _ = 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..740fc72 100644 --- a/server/task_control.go +++ b/server/task_control.go @@ -116,9 +116,17 @@ 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 _, err := s.m.pg.Exec(`UPDATE tasks SET schedule_paused=false, pause_reason='paused_by_user' WHERE id=$1`, t.ID); err != nil { return out, err } + t.updateLifecycle(func(state *taskLifecycleState) { + state.PauseReason = "paused_by_user" + state.ReconnectAttempt = 0 + state.ReconnectMax = 0 + state.ReconnectNextAt = 0 + state.ReconnectExhausted = false + state.ReconnectLastError = "" + }) out.Paused = true out.Status = "paused" return out, nil @@ -131,7 +139,17 @@ 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 { + 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 { if !wasEnginePaused && !wasQueued { s.engine.Resume(t) } diff --git a/web/src/app/(main)/function/tasks/detail/page.tsx b/web/src/app/(main)/function/tasks/detail/page.tsx index be52acc..6753359 100644 --- a/web/src/app/(main)/function/tasks/detail/page.tsx +++ b/web/src/app/(main)/function/tasks/detail/page.tsx @@ -419,6 +419,13 @@ 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.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 2f63a6c..9c2e25d 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; @@ -757,6 +767,12 @@ export default function SystemSettingsPage() {

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

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

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