fix: 清空运行时代理后转发器直连并门控恢复、跳过空地址成员、补测

Proxy failure pause, timed reconnect, and guarded auto-resume.

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 暂停了任务",

agent/chat.go

@@ -29,6 +29,7 @@ type ChatAgent struct {
 	tx             *transcript.Store
 	window         int
 	proxyAddr      string
+	proxyAddrFn    func() string
 	proxyCACert    string
 	webSearch      WebSearchOpts
 	guard          *guard.Guard // optional; nil disables intercept hooks for chat
@@ -65,7 +66,8 @@ func (c *ChatAgent) maxTokens() int {
 
 // SetProxy points the chat agent's WebFetch/Bash at the recording proxy plus the
 // CA cert it trusts (empty addr = direct). Kept for parity with the other agents.
-func (c *ChatAgent) SetProxy(addr, caCert string) { c.proxyAddr, c.proxyCACert = addr, caCert }
+func (c *ChatAgent) SetProxy(addr, caCert string)      { c.proxyAddr, c.proxyCACert = addr, caCert }
+func (c *ChatAgent) SetProxyProvider(fn func() string) { c.proxyAddrFn = fn }
 
 // SetWebSearch selects the web_search backend for the chat agent (off by default).
 func (c *ChatAgent) SetWebSearch(o WebSearchOpts) { c.webSearch = o }
@@ -124,6 +126,10 @@ func (c *ChatAgent) Chat(ctx context.Context, agentKey, sessionID, message strin
 	defer cleanup()
 
 	system, boundary := deferredSystem(chatSystem(agentKey, c.workDir, sessionWorkDir), def)
+	proxyAddr := c.proxyAddr
+	if c.proxyAddrFn != nil {
+		proxyAddr = c.proxyAddrFn()
+	}
 	opts := agentcore.Options{
 		Provider:        c.prov,
 		SystemPrompt:    system,
@@ -133,7 +139,7 @@ func (c *ChatAgent) Chat(ctx context.Context, agentKey, sessionID, message strin
 		UnlockSet:       def.Unlock,
 		PermissionMode:  permission.ModeBypass,
 		EnableWebFetch:  true, // 走记录代理留痕；载入代理 CA 验证 MITM 重签的 HTTPS 证书
-		WebFetchProxy:   c.proxyAddr,
+		WebFetchProxy:   proxyAddr,
 		WebFetchCACert:  c.proxyCACert,
 		// 联网搜索(可选)。ddgs 无需 key；brave-free 需 BraveKey；tavily 需 TavilyKey。
 		// WebSearchProxy 是独立出口代理(http/https/socks5)，与记录流量的 MITM 代理无关；空则直连。
@@ -145,7 +151,7 @@ func (c *ChatAgent) Chat(ctx context.Context, agentKey, sessionID, message strin
 		DeepSeekSearchAPIKey:  ws.DeepSeekAPIKey,
 		DeepSeekSearchModel:   ws.DeepSeekModel,
 		WebSearchProxy:        ws.Proxy,
-		BashEnv:               proxyEnv(c.proxyAddr, c.proxyCACert), // Bash 子命令默认走代理+信任 CA
+		BashEnv:               proxyEnv(proxyAddr, c.proxyCACert), // Bash 子命令默认走代理+信任 CA
 		WorkingDir:            sessionWorkDir,
 		MaxTurns:              maxTurns,
 		MaxDuration:           maxDuration,

agent/mainagent.go

@@ -21,11 +21,12 @@ type MainAgent struct {
 	findingRecorder FindingRecorder
 	prov            llm.Provider
 	model           string
-	tx              *transcript.Store                      // raw LLM conversation persistence (nil = off)
+	tx              *transcript.Store // raw LLM conversation persistence (nil = off)
-	window          int                                    // context window in tokens (for compaction)
+	window          int               // context window in tokens (for compaction)
-	windowFn        func() int                             // optional dynamic task-chain minimum
+	windowFn        func() int        // optional dynamic task-chain minimum
-	maxTurns        int                                    // max agent turns per run (0 = unlimited)
+	maxTurns        int               // max agent turns per run (0 = unlimited)
-	proxyAddr       string                                 // recording proxy for WebFetch (empty = direct)
+	proxyAddr       string            // recording proxy for WebFetch (empty = direct)
+	proxyAddrFn     func() string
 	proxyCACert     string                                 // recording proxy's CA cert path (HTTPS verify)
 	webSearch       WebSearchOpts                          // web_search tool backend selection (off by default)
 	workDir         string                                 // shared work dir (surfaced in prompt as artifact-output target)
@@ -72,7 +73,8 @@ func (m *MainAgent) compactionWindow() int {
 
 // SetProxy points the main agent's WebFetch at the recording proxy plus the CA
 // cert it trusts to verify HTTPS through it (empty addr = direct).
-func (m *MainAgent) SetProxy(addr, caCert string) { m.proxyAddr, m.proxyCACert = addr, caCert }
+func (m *MainAgent) SetProxy(addr, caCert string)      { m.proxyAddr, m.proxyCACert = addr, caCert }
+func (m *MainAgent) SetProxyProvider(fn func() string) { m.proxyAddrFn = fn }
 
 // SetWebSearch selects the web_search backend for the main agent (off by default).
 func (m *MainAgent) SetWebSearch(o WebSearchOpts) { m.webSearch = o }
@@ -132,6 +134,10 @@ func (m *MainAgent) Chat(ctx context.Context, taskID int64, mainSeg int, as *db.
 	mainDir := ensureRunDir(m.workDir, taskID, 0)
 	ctx = intercept.WithReviewWorkingDirectory(ctx, mainDir)
 	system, boundary := deferredSystem(mainAgentSystem(goal, m.workDir, mainDir), def)
+	proxyAddr := m.proxyAddr
+	if m.proxyAddrFn != nil {
+		proxyAddr = m.proxyAddrFn()
+	}
 	opts := agentcore.Options{
 		Provider:        m.prov,
 		SystemPrompt:    system,
@@ -141,7 +147,7 @@ func (m *MainAgent) Chat(ctx context.Context, taskID int64, mainSeg int, as *db.
 		UnlockSet:       def.Unlock,
 		PermissionMode:  permission.ModeBypass,
 		EnableWebFetch:  true, // 走记录代理留痕；载入代理 CA 验证 MITM 重签的 HTTPS 证书
-		WebFetchProxy:   m.proxyAddr,
+		WebFetchProxy:   proxyAddr,
 		WebFetchCACert:  m.proxyCACert,
 		// 联网搜索(可选)。ddgs 无需 key；brave-free 需 BraveKey；tavily 需 TavilyKey。
 		// WebSearchProxy 是独立出口代理(http/https/socks5)，与记录流量的 MITM 代理无关；空则直连。
@@ -153,8 +159,8 @@ func (m *MainAgent) Chat(ctx context.Context, taskID int64, mainSeg int, as *db.
 		DeepSeekSearchAPIKey:  m.webSearch.DeepSeekAPIKey,
 		DeepSeekSearchModel:   m.webSearch.DeepSeekModel,
 		WebSearchProxy:        m.webSearch.Proxy,
-		BashEnv:               proxyEnv(m.proxyAddr, m.proxyCACert), // Bash 子命令默认走代理+信任 CA
+		BashEnv:               proxyEnv(proxyAddr, m.proxyCACert), // Bash 子命令默认走代理+信任 CA
-		WorkingDir:            mainDir,                              // 本任务工作目录 <workDir>/tasks/<taskID>
+		WorkingDir:            mainDir,                            // 本任务工作目录 <workDir>/tasks/<taskID>
 		ToolOutputDir:         cmdOutDir(mainDir),
 		MaxTurns:              m.maxTurns,                             // 0 = unlimited (configurable in agent management)
 		Compaction:            compactionConfig(m.compactionWindow()), // long chats stay within the window

agent/planner.go

@@ -31,14 +31,15 @@ type Planner struct {
 	killWork          func(intentID int64) error             // engine callback to terminate a running work (nil = off)
 	steerWork         func(intentID int64, msg string) error // engine callback to steer a running work mid-run (nil = off)
 	proxyAddr         string                                 // recording proxy for WebFetch (empty = direct)
-	proxyCACert       string                                 // recording proxy's CA cert path (HTTPS verify)
+	proxyAddrFn       func() string
-	webSearch         WebSearchOpts                          // web_search tool backend selection (off by default)
+	proxyCACert       string        // recording proxy's CA cert path (HTTPS verify)
-	workDir           string                                 // shared work dir (surfaced in prompt as artifact-output target)
+	webSearch         WebSearchOpts // web_search tool backend selection (off by default)
-	injectConstraints func() bool                            // resolver: inject task operation constraints into system prompt? (nil = yes)
+	workDir           string        // shared work dir (surfaced in prompt as artifact-output target)
-	nonStreamingFn    func() bool                            // resolver: use non-streaming (Complete) path? (nil = streaming)
+	injectConstraints func() bool   // resolver: inject task operation constraints into system prompt? (nil = yes)
-	noaEnabledFn      func() bool                            // resolver: use experimental noa compaction? (nil = off)
+	nonStreamingFn    func() bool   // resolver: use non-streaming (Complete) path? (nil = streaming)
-	maxTokensFn       func() int                             // resolver: per-reply output cap (nil/0 = send no cap)
+	noaEnabledFn      func() bool   // resolver: use experimental noa compaction? (nil = off)
-	compactor         *Compactor                             // cold-node compaction (§7); nil = disabled
+	maxTokensFn       func() int    // resolver: per-reply output cap (nil/0 = send no cap)
+	compactor         *Compactor    // cold-node compaction (§7); nil = disabled
 
 	// todos keeps ONE plan-scratchpad per task (keyed by exploration id) so the
 	// planner's multi-step plan survives across wake-ups — each Plan() is a fresh
@@ -90,7 +91,8 @@ func (p *Planner) compactionWindow() int {
 
 // SetProxy points the planner's WebFetch at the recording proxy plus the CA cert
 // it trusts to verify HTTPS through it (empty addr = direct).
-func (p *Planner) SetProxy(addr, caCert string) { p.proxyAddr, p.proxyCACert = addr, caCert }
+func (p *Planner) SetProxy(addr, caCert string)      { p.proxyAddr, p.proxyCACert = addr, caCert }
+func (p *Planner) SetProxyProvider(fn func() string) { p.proxyAddrFn = fn }
 
 // SetWebSearch selects the web_search backend for the planner (off by default).
 func (p *Planner) SetWebSearch(o WebSearchOpts) { p.webSearch = o }
@@ -402,6 +404,10 @@ func (p *Planner) Plan(ctx context.Context, taskID int64, as *db.AssetStore, ts
 	if tc.DeadlineUnix > 0 {
 		settle = wrapupSettlementForTask("planner", nil, clamped)
 	}
+	proxyAddr := p.proxyAddr
+	if p.proxyAddrFn != nil {
+		proxyAddr = p.proxyAddrFn()
+	}
 	opts := agentcore.Options{
 		Provider:        p.prov,
 		SystemPrompt:    system,
@@ -411,7 +417,7 @@ func (p *Planner) Plan(ctx context.Context, taskID int64, as *db.AssetStore, ts
 		UnlockSet:       def.Unlock,
 		PermissionMode:  permission.ModeBypass,
 		EnableWebFetch:  true, // 走记录代理留痕；载入代理 CA 验证 MITM 重签的 HTTPS 证书
-		WebFetchProxy:   p.proxyAddr,
+		WebFetchProxy:   proxyAddr,
 		WebFetchCACert:  p.proxyCACert,
 		// 联网搜索(可选)。ddgs 无需 key；brave-free 需 BraveKey；tavily 需 TavilyKey。
 		// WebSearchProxy 是独立出口代理(http/https/socks5)，与记录流量的 MITM 代理无关；空则直连。
@@ -423,8 +429,8 @@ func (p *Planner) Plan(ctx context.Context, taskID int64, as *db.AssetStore, ts
 		DeepSeekSearchAPIKey:  p.webSearch.DeepSeekAPIKey,
 		DeepSeekSearchModel:   p.webSearch.DeepSeekModel,
 		WebSearchProxy:        p.webSearch.Proxy,
-		BashEnv:               proxyEnv(p.proxyAddr, p.proxyCACert), // Bash 子命令默认走代理+信任 CA
+		BashEnv:               proxyEnv(proxyAddr, p.proxyCACert), // Bash 子命令默认走代理+信任 CA
-		WorkingDir:            taskDir,                              // 本任务工作目录 <workDir>/tasks/<taskID>
+		WorkingDir:            taskDir,                            // 本任务工作目录 <workDir>/tasks/<taskID>
 		ToolOutputDir:         cmdOutDir(taskDir),
 		MaxTurns:              p.maxTurns, // 0 = unlimited (configurable in agent management)
 		MaxDuration:           maxDur,     // 0=不限;有 deadline 时=距 deadline 剩余

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,

agent/worker.go

@@ -57,6 +57,7 @@ type Worker struct {
 	model           string
 	workDir         string
 	proxyAddr       string
+	proxyAddrFn     func() string
 	proxyCACert     string            // recording proxy's CA cert path (for WebFetch HTTPS verify)
 	webSearch       WebSearchOpts     // web_search tool backend selection (off by default)
 	tx              *transcript.Store // raw LLM conversation persistence (nil = off)
@@ -188,6 +189,8 @@ func (w *Worker) compactionWindow() int {
 // that MITM proxy. Empty addr disables the hint.
 func (w *Worker) SetProxy(addr, caCert string) { w.proxyAddr, w.proxyCACert = addr, caCert }
 
+func (w *Worker) SetProxyProvider(fn func() string) { w.proxyAddrFn = fn }
+
 // SetWebSearch selects the web_search backend for this worker (off by default).
 func (w *Worker) SetWebSearch(o WebSearchOpts) { w.webSearch = o }
 
@@ -393,7 +396,11 @@ func (w *Worker) execute(ctx context.Context, name string, taskID int64, as *db.
 	// inherit a parent run's background into the action reviewer.
 	ctx = intercept.WithReviewContext(ctx, runDir, intercept.ReviewBackground{})
 	overview := renderWorkerGraphOverview(tsx.graphOverviewData())
-	sysBody := workerSystem(w.proxyAddr, w.proxyCACert, w.workDir, runDir)
+	proxyAddr := w.proxyAddr
+	if w.proxyAddrFn != nil {
+		proxyAddr = w.proxyAddrFn()
+	}
+	sysBody := workerSystem(proxyAddr, w.proxyCACert, w.workDir, runDir)
 	if w.wantConstraints() {
 		sysBody += constraintBlock(ts) // 操作约束(若有)注入系统提示,worker 执行时严格遵守
 	}
@@ -437,7 +444,7 @@ func (w *Worker) execute(ctx context.Context, name string, taskID int64, as *db.
 		// WebFetch 走记录代理，其 HTTP 与 curl 一样被留痕；载入代理 CA 让经 MITM
 		// 重签的 HTTPS 证书能【正常校验通过】（而非关掉校验）。proxy 空则直连。
 		EnableWebFetch: true,
-		WebFetchProxy:  w.proxyAddr,
+		WebFetchProxy:  proxyAddr,
 		WebFetchCACert: w.proxyCACert,
 		// 联网搜索(可选)。ddgs 无需 key；brave-free 需 BraveKey；tavily 需 TavilyKey。
 		// WebSearchProxy 是独立的出口代理(http/https/socks5)，与记录流量的 MITM 代理无关；空则直连。
@@ -450,7 +457,7 @@ func (w *Worker) execute(ctx context.Context, name string, taskID int64, as *db.
 		DeepSeekSearchModel:   w.webSearch.DeepSeekModel,
 		WebSearchProxy:        w.webSearch.Proxy,
 		// Bash 子命令的 HTTP 默认走记录代理 + 信任其 CA（工具无需 -x/-k）。
-		BashEnv:    proxyEnv(w.proxyAddr, w.proxyCACert),
+		BashEnv:    proxyEnv(proxyAddr, w.proxyCACert),
 		WorkingDir: runDir,
 		MaxTurns:   w.maxTurns, // 0 = unlimited (configurable in agent management)
 		// 墙钟预算,轮边界判,不打断半路;0 = 不限。有任务级 deadline 时夹逼到 min(自身预算,

cmd/artex/main.go

@@ -113,6 +113,7 @@ func run() int {
 	}
 	log.Printf("[config] skill 目录: %s", skillDir)
 	srv := server.New(ctx, mgr, skillDir, *dataDir, config.BaseDir())
+	defer srv.Close()
 	httpSrv := &http.Server{
 		Addr:              *addr,
 		Handler:           srv.Handler(),

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;
 

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
@@ -612,6 +613,7 @@ func (d *DB) SetStatusRuntime(id int64, status string) (int64, *time.Time, error
 	err := d.QueryRow(`
 UPDATE tasks
    SET status = $1,
+       pause_reason = CASE WHEN $1 IN ('done','failed','timeout') THEN '' ELSE pause_reason END,
        completed_at = CASE WHEN $1 IN ('done','failed','timeout') THEN COALESCE(completed_at, now()) ELSE NULL END,
        active_run_seconds = active_run_seconds + CASE WHEN $1 IN ('done','failed','timeout') AND run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
        run_segment_started_at = CASE WHEN $1 IN ('done','failed','timeout') THEN NULL ELSE run_segment_started_at END
@@ -638,6 +640,7 @@ func (d *DB) SetTerminalStatusGuardedRuntime(id int64, status string) (active in
 	err = d.QueryRow(`
 UPDATE tasks
    SET status = $1,
+       pause_reason = '',
        completed_at = COALESCE(completed_at, now()),
        active_run_seconds = active_run_seconds + CASE WHEN run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
        run_segment_started_at = NULL

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"

server/customtool.go

@@ -368,7 +368,7 @@ func (s *Server) runScriptTool(ctx context.Context, key string, execRaw json.Raw
 	}
 	body, err := execPython(ctx, interp, key, spec.Code, params, workDir, sessionEnv, timeoutOr(spec.TimeoutMs, 120000))
 	if err != nil {
-		return actool.Errorf(err.Error()), nil
+		return actool.Errorf(maskSecrets(err.Error())), nil
 	}
 	return actool.Text(actool.Capture(tc, body)), nil
 }
@@ -436,7 +436,7 @@ func (s *Server) runHTTPTool(ctx context.Context, execRaw json.RawMessage, param
 	defer cancel()
 	req, err := http.NewRequestWithContext(runCtx, method, rawURL, bodyReader)
 	if err != nil {
-		return actool.Errorf(err.Error()), nil
+		return actool.Errorf(maskSecrets(err.Error())), nil
 	}
 	for k, v := range spec.Headers {
 		req.Header.Set(k, renderTemplate(v, params, identity))
@@ -447,7 +447,10 @@ func (s *Server) runHTTPTool(ctx context.Context, execRaw json.RawMessage, param
 	}
 	resp, err := client.Do(req)
 	if err != nil {
-		return actool.Errorf("请求失败: " + err.Error()), nil
+		if spec.UseRecordingProxy || strings.TrimSpace(spec.Proxy) == strings.TrimSpace(s.m.GlobalProxy()) {
+			s.handleProxyToolFailure(runCtx, err)
+		}
+		return actool.Errorf("请求失败: " + maskSecrets(err.Error())), nil
 	}
 	defer resp.Body.Close()
 	respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))

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,
+		ID:                t.ID,
-		ExplorationID:      t.ExpID,
+		ExplorationID:     t.ExpID,
-		Name:               lifecycle.Name,
+		Name:              lifecycle.Name,
-		Remark:             lifecycle.Remark,
+		Remark:            lifecycle.Remark,
-		CategoryID:         lifecycle.CategoryID,
+		CategoryID:        lifecycle.CategoryID,
-		CategoryName:       lifecycle.CategoryName,
+		CategoryName:      lifecycle.CategoryName,
-		Pinned:             lifecycle.PinnedAt > 0,
+		Pinned:            lifecycle.PinnedAt > 0,
-		PinnedAt:           completedRFC(lifecycle.PinnedAt),
+		PinnedAt:          completedRFC(lifecycle.PinnedAt),
-		Description:        t.Description,
+		Description:       t.Description,
-		Goal:               t.Goal,
+		Goal:              t.Goal,
-		Status:             status,
+		Status:            status,
-		CreatedAt:          rfc3339(time.Unix(t.CreatedAt, 0)),
+		CreatedAt:         rfc3339(time.Unix(t.CreatedAt, 0)),
-		CreatedUnix:        t.CreatedAt,
+		CreatedUnix:       t.CreatedAt,
-		RunSeconds:         runSeconds,
+		RunSeconds:        runSeconds,
-		ActiveRunSeconds:   lifecycle.ActiveRunSeconds,
+		ActiveRunSeconds:  lifecycle.ActiveRunSeconds,
-		RunSegmentStarted:  lifecycle.RunSegmentStartedAt,
+		RunSegmentStarted: lifecycle.RunSegmentStartedAt,
-		CompletedAt:        completedRFC(lifecycle.CompletedAt),
+		CompletedAt:       completedRFC(lifecycle.CompletedAt),
-		CompletedUnix:      lifecycle.CompletedAt,
+		CompletedUnix:     lifecycle.CompletedAt,
-		Paused:             lifecycle.Paused,
+		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,

server/egress_probe.go

@@ -341,6 +341,7 @@ func (s *Server) testEgressProxy(w http.ResponseWriter, r *http.Request) {
 	var req struct {
 		GlobalProxy string `json:"global_proxy"`
 		ProbeURL    string `json:"probe_url"` // 可选：覆盖默认探测地址
+		Resume      bool   `json:"resume"`    // manual settings probe may release paused_by_proxy tasks
 	}
 	// 请求体只有一个地址，给个上限，免得被塞大 payload。
 	r.Body = http.MaxBytesReader(w, r.Body, 64<<10)
@@ -366,15 +367,21 @@ func (s *Server) testEgressProxy(w http.ResponseWriter, r *http.Request) {
 	agentCA := s.m.ProxyCACert()    // 捕获开时是 MITM 的 CA 路径；关时为空
 	savedProxy := s.m.GlobalProxy() // 已保存的 global_proxy
 
-	// 表单值 ≠ 已保存值时，② 仍按表单值测，但 agent 走的是已保存值——必须提示。
-	// 归一化再比：尾斜杠不算"未保存"。
-	pending := normalizeProxy(req.GlobalProxy) != normalizeProxy(savedProxy)
-
 	// 整体上限 40s：单次 Timeout 只兜一次，不兜总量；前端 fetch 没有超时。
 	ctx, cancel := context.WithTimeout(r.Context(), 40*time.Second)
 	defer cancel()
+	var proxyGuards map[string]proxyReconnectGuard
+	if req.Resume {
+		proxyGuards = s.snapshotProxyReconnectGuardsWithExhausted(false, true)
+	}
 
 	configuredProxy := strings.TrimSpace(req.GlobalProxy)
+	if strings.Contains(configuredProxy, "***") || strings.Contains(configuredProxy, "******") {
+		configuredProxy = s.m.GlobalProxy()
+	}
+	// 表单值 ≠ 已保存值时，② 仍按表单值测，但 agent 走的是已保存值——必须提示。
+	// 归一化再比：尾斜杠不算"未保存"。
+	pending := normalizeProxy(configuredProxy) != normalizeProxy(savedProxy)
 	rep := egressReport{
 		Capture:    capture,
 		Configured: maskSecrets(configuredProxy),
@@ -417,6 +424,12 @@ func (s *Server) testEgressProxy(w http.ResponseWriter, r *http.Request) {
 
 	rep.OK = judgeEgress(rep)
 	rep.Warnings = egressWarnings(rep, pending, savedProxy)
+	if req.Resume && rep.OK && !pending {
+		// The navbar uses the same probe endpoint for display-only polling but
+		// omits resume. Only an explicit settings-page test can release the
+		// guarded paused_by_proxy tasks.
+		s.proxyReconnectSuccessWithGuards(proxyGuards, nil)
+	}
 	writeJSON(w, 200, rep)
 }
 

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:

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")
 	}

server/forward_proxy.go

@@ -0,0 +1,294 @@
+package server
+
+import (
+	"bufio"
+	"context"
+	"crypto/tls"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"strings"
+	"sync"
+	"time"
+
+	"golang.org/x/net/proxy"
+)
+
+// dynamicForwardProxy is a credential-free, stable local proxy. Each request
+// resolves the current upstream, so pool switches do not rebuild agents or MCP.
+type dynamicForwardProxy struct {
+	ln      net.Listener
+	addr    string
+	current func() string
+	direct  func() bool
+	server  *http.Server
+	once    sync.Once
+	mu      sync.Mutex
+	conns   map[net.Conn]struct{}
+	ctx     context.Context
+	cancel  context.CancelFunc
+}
+
+func newDynamicForwardProxy(current func() string) (*dynamicForwardProxy, error) {
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		return nil, err
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	p := &dynamicForwardProxy{ln: ln, addr: "http://" + ln.Addr().String(), current: current, direct: func() bool { return true }, conns: map[net.Conn]struct{}{}, ctx: ctx, cancel: cancel}
+	p.server = &http.Server{Handler: http.HandlerFunc(p.serveHTTP), BaseContext: func(net.Listener) context.Context { return ctx }}
+	go p.server.Serve(ln)
+	return p, nil
+}
+
+func (p *dynamicForwardProxy) Addr() string {
+	if p == nil {
+		return ""
+	}
+	return p.addr
+}
+func (p *dynamicForwardProxy) Close() error {
+	if p == nil {
+		return nil
+	}
+	p.once.Do(func() {
+		p.cancel()
+		p.mu.Lock()
+		for c := range p.conns {
+			_ = c.Close()
+		}
+		p.conns = nil
+		p.mu.Unlock()
+		_ = p.server.Close()
+	})
+	return nil
+}
+
+func (p *dynamicForwardProxy) track(c net.Conn) func() {
+	if c == nil {
+		return func() {}
+	}
+	p.mu.Lock()
+	if p.conns != nil {
+		p.conns[c] = struct{}{}
+	}
+	p.mu.Unlock()
+	return func() { p.mu.Lock(); delete(p.conns, c); p.mu.Unlock() }
+}
+
+func (p *dynamicForwardProxy) serveHTTP(w http.ResponseWriter, r *http.Request) {
+	if r.Method == http.MethodConnect {
+		p.serveConnect(w, r)
+		return
+	}
+	up := strings.TrimSpace(p.current())
+	if up == "" && p.direct != nil && !p.direct() {
+		http.Error(w, "proxy unavailable", 502)
+		return
+	}
+	tr := &http.Transport{Proxy: nil}
+	if up != "" {
+		u, err := url.Parse(up)
+		if err != nil {
+			http.Error(w, "proxy unavailable", 502)
+			return
+		}
+		if strings.HasPrefix(strings.ToLower(u.Scheme), "socks5") {
+			d, err := socksDialer(u)
+			if err != nil {
+				http.Error(w, "proxy unavailable", 502)
+				return
+			}
+			tr.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) {
+				return dialProxyContext(ctx, d, network, addr)
+			}
+		} else {
+			tr.Proxy = http.ProxyURL(u)
+		}
+	}
+	tr.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12}
+	defer tr.CloseIdleConnections()
+	rr := r.Clone(r.Context())
+	if rr.URL.IsAbs() == false {
+		rr.URL.Scheme = "http"
+		rr.URL.Host = r.Host
+	}
+	rr.RequestURI = ""
+	for k := range rr.Header {
+		if strings.EqualFold(k, "Proxy-Authorization") {
+			rr.Header.Del(k)
+		}
+	}
+	resp, err := tr.RoundTrip(rr)
+	if err != nil {
+		http.Error(w, "proxy unavailable", 502)
+		return
+	}
+	defer resp.Body.Close()
+	for k, vv := range resp.Header {
+		for _, v := range vv {
+			w.Header().Add(k, v)
+		}
+	}
+	w.WriteHeader(resp.StatusCode)
+	_, _ = io.Copy(w, resp.Body)
+}
+
+func (p *dynamicForwardProxy) serveConnect(w http.ResponseWriter, r *http.Request) {
+	up := strings.TrimSpace(p.current())
+	if up == "" && p.direct != nil && !p.direct() {
+		http.Error(w, "proxy unavailable", 502)
+		return
+	}
+	target := r.Host
+	var conn net.Conn
+	var err error
+	if up == "" {
+		conn, err = (&net.Dialer{Timeout: 10 * time.Second}).DialContext(r.Context(), "tcp", target)
+	} else {
+		u, e := url.Parse(up)
+		if e != nil {
+			err = e
+		} else if strings.HasPrefix(strings.ToLower(u.Scheme), "socks5") {
+			d, e := socksDialer(u)
+			if e != nil {
+				err = e
+			} else {
+				conn, err = dialProxyContext(r.Context(), d, "tcp", target)
+			}
+		} else {
+			conn, err = dialHTTPConnect(r.Context(), u, target)
+		}
+	}
+	if err != nil {
+		http.Error(w, "proxy unavailable", 502)
+		return
+	}
+	h, ok := w.(http.Hijacker)
+	if !ok {
+		http.Error(w, "proxy unavailable", 500)
+		return
+	}
+	client, rw, err := h.Hijack()
+	if err != nil {
+		return
+	}
+	removeClient := p.track(client)
+	defer removeClient()
+	removeUpstream := p.track(conn)
+	defer removeUpstream()
+	defer client.Close()
+	defer conn.Close()
+	_, _ = rw.WriteString("HTTP/1.1 200 Connection Established\r\n\r\n")
+	_ = rw.Flush()
+	clientReader := io.Reader(client)
+	if rw.Reader != nil && rw.Reader.Buffered() > 0 {
+		clientReader = io.MultiReader(rw.Reader, client)
+	}
+	done := make(chan struct{}, 2)
+	go func() { _, _ = io.Copy(conn, clientReader); _ = conn.Close(); done <- struct{}{} }()
+	go func() { _, _ = io.Copy(client, conn); _ = client.Close(); done <- struct{}{} }()
+	<-done
+}
+
+func socksDialer(u *url.URL) (proxy.Dialer, error) {
+	var auth *proxy.Auth
+	if u.User != nil {
+		pw, _ := u.User.Password()
+		auth = &proxy.Auth{User: u.User.Username(), Password: pw}
+	}
+	return proxy.SOCKS5("tcp", u.Host, auth, &net.Dialer{Timeout: 10 * time.Second})
+}
+
+func dialProxyContext(ctx context.Context, d proxy.Dialer, network, addr string) (net.Conn, error) {
+	if cd, ok := d.(proxy.ContextDialer); ok {
+		return cd.DialContext(ctx, network, addr)
+	}
+	type result struct {
+		c   net.Conn
+		err error
+	}
+	ch := make(chan result, 1)
+	go func() { c, err := d.Dial(network, addr); ch <- result{c, err} }()
+	select {
+	case r := <-ch:
+		return r.c, r.err
+	case <-ctx.Done():
+		go func() {
+			r := <-ch
+			if r.c != nil {
+				_ = r.c.Close()
+			}
+		}()
+		return nil, ctx.Err()
+	}
+}
+
+type bufferedConn struct {
+	net.Conn
+	reader *bufio.Reader
+}
+
+func (c *bufferedConn) Read(p []byte) (int, error) { return c.reader.Read(p) }
+
+func dialHTTPConnect(ctx context.Context, u *url.URL, target string) (net.Conn, error) {
+	d := &net.Dialer{Timeout: 10 * time.Second}
+	addr := u.Host
+	if !strings.Contains(addr, ":") {
+		addr += ":80"
+	}
+	conn, err := d.DialContext(ctx, "tcp", addr)
+	if err != nil {
+		return nil, err
+	}
+	if strings.EqualFold(u.Scheme, "https") {
+		conn = tls.Client(conn, &tls.Config{ServerName: u.Hostname(), MinVersion: tls.VersionTLS12})
+		if err := conn.(*tls.Conn).HandshakeContext(ctx); err != nil {
+			_ = conn.Close()
+			return nil, err
+		}
+	}
+	br := bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn))
+	fmt.Fprintf(br, "CONNECT %s HTTP/1.1\r\nHost: %s\r\n", target, target)
+	if u.User != nil {
+		req := &http.Request{Header: http.Header{}}
+		req.SetBasicAuth(u.User.Username(), func() string { p, _ := u.User.Password(); return p }())
+		fmt.Fprintf(br, "Proxy-Authorization: %s\r\n", req.Header.Get("Authorization"))
+	}
+	fmt.Fprint(br, "\r\n")
+	if err := br.Flush(); err != nil {
+		_ = conn.Close()
+		return nil, err
+	}
+	readCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+	type responseResult struct {
+		resp *http.Response
+		err  error
+	}
+	readDone := make(chan responseResult, 1)
+	go func() {
+		resp, err := http.ReadResponse(br.Reader, &http.Request{Method: http.MethodConnect})
+		readDone <- responseResult{resp: resp, err: err}
+	}()
+	var rr responseResult
+	select {
+	case rr = <-readDone:
+	case <-readCtx.Done():
+		_ = conn.Close()
+		<-readDone
+		return nil, readCtx.Err()
+	}
+	resp, err := rr.resp, rr.err
+	if err != nil {
+		_ = conn.Close()
+		return nil, err
+	}
+	if resp.StatusCode/100 != 2 {
+		_ = conn.Close()
+		return nil, fmt.Errorf("upstream CONNECT: %s", resp.Status)
+	}
+	return &bufferedConn{Conn: conn, reader: br.Reader}, nil
+}

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)
 		}

server/manager.go

@@ -13,6 +13,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/Autumn-27/artex/agent"
@@ -28,19 +29,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"`
+	ID                  string `json:"id"`
-	ExpID        int64  `json:"exploration_id"`
+	ExpID               int64  `json:"exploration_id"`
-	Remark       string `json:"remark"`
+	Remark              string `json:"remark"`
-	Name         string `json:"name"` // 可选任务名称;空=未命名
+	Name                string `json:"name"` // 可选任务名称;空=未命名
-	CategoryID   *int64 `json:"category_id,omitempty"`
+	CategoryID          *int64 `json:"category_id,omitempty"`
-	CategoryName string `json:"category_name,omitempty"`
+	CategoryName        string `json:"category_name,omitempty"`
-	PinnedAt     int64  `json:"pinned_at,omitempty"`
+	PinnedAt            int64  `json:"pinned_at,omitempty"`
-	Description  string `json:"description"`
+	Description         string `json:"description"`
-	Goal         string `json:"goal"`
+	Goal                string `json:"goal"`
-	CreatedAt    int64  `json:"created_at"`
+	CreatedAt           int64  `json:"created_at"`
-	CompletedAt  int64  `json:"completed_at,omitempty"` // 进入终态的 unix 秒;0=未完成
+	CompletedAt         int64  `json:"completed_at,omitempty"` // 进入终态的 unix 秒;0=未完成
-	Paused       bool   `json:"paused"`
+	Paused              bool   `json:"paused"`
-	Queued       bool   `json:"queued"` // 因并发上限被挂起、等待空位自动启动;true=尚未开跑
+	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 +96,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 +128,13 @@ func (t *Task) lifecycleSnapshot() taskLifecycleState {
 
 func (t *Task) lifecycleSnapshotLocked() taskLifecycleState {
 	return taskLifecycleState{
-		Name:                t.Name,
+		Name:             t.Name,
-		Remark:              t.Remark,
+		Remark:           t.Remark,
-		PinnedAt:            t.PinnedAt,
+		PinnedAt:         t.PinnedAt,
-		Status:              t.Status,
+		Status:           t.Status,
-		Paused:              t.Paused,
+		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 +162,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 +266,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
@@ -258,9 +283,16 @@ type Manager struct {
 	// (http/https/socks5, optional user:pass). Empty = direct. When traffic
 	// capture is on it becomes the MITM's upstream; when capture is off it is
 	// injected into agent bash env / WebFetch directly. See ProxyAddr.
-	globalProxy string
+	globalProxy         string
+	proxyPool           *proxyPoolState
+	forwardProxy        *dynamicForwardProxy
+	proxyPoolConfigured atomic.Bool
 }
 
+// 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"
@@ -554,6 +586,14 @@ func NewManager(dir, proxyAddr string) (*Manager, error) {
 		log.Printf("[llmusage] create table: %v", err)
 	}
 	m := &Manager{dir: dir, pg: pg, assets: pg.Assets(), tasks: map[string]*Task{}, interceptor: intercept.New(pg)}
+	if fp, ferr := newDynamicForwardProxy(func() string { return m.GlobalProxy() }); ferr == nil {
+		m.forwardProxy = fp
+		// An empty runtime route means direct egress, even when the persisted
+		// pool still contains disabled or blank entries.
+		fp.direct = func() bool { return strings.TrimSpace(m.GlobalProxy()) == "" }
+	} else {
+		log.Printf("[proxy] local forwarder disabled: %v", ferr)
+	}
 	if proxyAddr != "" {
 		tr, err := traffic.Open(filepath.Join(dir, "traffic"), proxyAddr)
 		if err != nil {
@@ -614,6 +654,7 @@ func NewManager(dir, proxyAddr string) (*Manager, error) {
 	if v, ok, _ := pg.GetSetting(settingGlobalProxy); ok {
 		m.globalProxy = strings.TrimSpace(v)
 	}
+	m.initProxyPool()
 	if m.traffic != nil {
 		if err := m.traffic.SetUpstreamProxy(m.globalProxy); err != nil {
 			log.Printf("[proxy] 全局代理 %q 无效，已忽略: %v", m.globalProxy, err)
@@ -988,8 +1029,12 @@ func (m *Manager) ProxyAddr() string {
 		return m.traffic.ProxyAddr()
 	}
 	m.mu.RLock()
-	defer m.mu.RUnlock()
+	global := m.globalProxy
-	return m.globalProxy
+	m.mu.RUnlock()
+	if (global != "" || m.proxyPoolConfigured.Load()) && m.forwardProxy != nil {
+		return m.forwardProxy.Addr()
+	}
+	return maskSecrets(global)
 }
 
 // ProxyCACert returns the CA cert path agents must trust to verify HTTPS through
@@ -1019,7 +1064,30 @@ func (m *Manager) SetGlobalProxy(raw string) error {
 	raw = strings.TrimSpace(raw)
 	if raw != "" {
 		if _, err := traffic.ValidateProxyURL(raw); err != nil {
-			return err
+			return fmt.Errorf("%s", maskSecrets(err.Error()))
+		}
+	}
+	if m.proxyPool != nil {
+		m.proxyPool.mu.Lock()
+		managed := len(m.proxyPool.entries) > 0
+		if !managed && raw != "" {
+			m.proxyPool.entries = []proxyPoolEntry{{ProxyPoolEntry: ProxyPoolEntry{ID: "proxy-legacy", URL: maskSecrets(raw), Enabled: true, Health: "healthy"}, RawURL: raw}}
+			m.proxyPool.active = "proxy-legacy"
+			m.proxyPoolConfigured.Store(true)
+			if err := m.applyActiveProxyLocked("proxy-legacy", "manual"); err != nil {
+				m.proxyPool.entries = nil
+				m.proxyPool.active = ""
+				m.proxyPool.mu.Unlock()
+				return err
+			}
+			m.proxyPool.mu.Unlock()
+			return nil
+		}
+		m.proxyPool.mu.Unlock()
+		if managed {
+			// The legacy field is a read-only echo once a pool exists. Pool API
+			// operations are the sole authority for health, enablement and switching.
+			return nil
 		}
 	}
 	if err := m.pg.SetSetting(settingGlobalProxy, raw); err != nil {
@@ -1033,12 +1101,14 @@ func (m *Manager) SetGlobalProxy(raw string) error {
 			return err
 		}
 	}
-	// Keep the browser MCP's egress in sync with the new global proxy too.
 	m.syncBrowserMCPProxy()
 	return nil
 }
 
 func (m *Manager) Close() error {
+	if m.forwardProxy != nil {
+		_ = m.forwardProxy.Close()
+	}
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	if m.traffic != nil {
@@ -1073,7 +1143,7 @@ func taskFromPG(pt *pgdb.Task, store *pgdb.ExplorationStore, ic *intercept.Inter
 		Remark:     pt.Remark,
 		CategoryID: cloneInt64Ptr(pt.CategoryID), CategoryName: pt.CategoryName,
 		PinnedAt:    unixOrZero(pt.PinnedAt),
-		Description: pt.Description, Goal: pt.Goal, CreatedAt: pt.CreatedAt.Unix(), Paused: pt.Paused, Queued: pt.Queued,
+		Description: pt.Description, Goal: pt.Goal, CreatedAt: pt.CreatedAt.Unix(), Paused: pt.Paused, PauseReason: pt.PauseReason, Queued: pt.Queued,
 		QueuedAt: unixNanoOrZero(pt.QueuedAt), QueueMode: pt.QueueMode,
 		CompletedAt: unixOrZero(pt.CompletedAt), Status: pt.Status, ParentRef: pt.ParentRef,
 		LLMProfileID:  pt.LLMProfileID,
@@ -1391,12 +1461,16 @@ func (m *Manager) SetTaskPaused(id string, paused bool) error {
 // admission keeps its FIFO timestamp; a task that was explicitly paused and is
 // now re-queued receives a fresh tail position.
 func (m *Manager) ApplyTaskAdmission(id, expectedStatus, status string, queued bool, mode string, preservePosition bool) error {
-	return m.applyTaskAdmission(id, expectedStatus, status, queued, mode, preservePosition, nil)
+	return m.applyTaskAdmissionWithSchedule(id, expectedStatus, status, queued, mode, preservePosition, nil, false)
 }
 
 // Manual task control records its schedule override in the admission commit, so
 // failed resumes cannot bypass a calendar and FIFO promotion preserves it.
 func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued bool, mode string, preservePosition bool, manualUntil *time.Time) error {
+	return m.applyTaskAdmissionWithSchedule(id, expectedStatus, status, queued, mode, preservePosition, manualUntil, false)
+}
+
+func (m *Manager) applyTaskAdmissionWithSchedule(id, expectedStatus, status string, queued bool, mode string, preservePosition bool, manualUntil *time.Time, preserveSchedule bool) error {
 	m.taskStateMu.Lock()
 	defer m.taskStateMu.Unlock()
 	n, err := strconv.ParseInt(id, 10, 64)
@@ -1420,8 +1494,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,
+	    paused=false, pause_reason='',
-	    schedule_manual_until=COALESCE($7::timestamptz, schedule_manual_until),
+	    schedule_paused=CASE WHEN $8 THEN schedule_paused ELSE false END,
+	    schedule_manual_until=CASE WHEN $8 THEN schedule_manual_until ELSE COALESCE($7::timestamptz, schedule_manual_until) END,
 	    queued=$3,
 	    queued_at=CASE
 	        WHEN NOT $3 THEN NULL
@@ -1444,7 +1519,7 @@ func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued b
 	    ,active_run_seconds=active_run_seconds + CASE WHEN ($3 OR $2 IN ('done','failed','timeout')) AND run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END
 	    ,run_segment_started_at=CASE WHEN $3 OR $2 IN ('done','failed','timeout') THEN NULL ELSE run_segment_started_at END
 	WHERE id=$1 AND deleted_at IS NULL AND status=$6
-	RETURNING queued_at, queue_mode, completed_at, first_run_at, deadline_at, active_run_seconds, run_segment_started_at`, n, status, queued, mode, preservePosition, expectedStatus, manualUntil).
+	RETURNING queued_at, queue_mode, completed_at, first_run_at, deadline_at, active_run_seconds, run_segment_started_at`, n, status, queued, mode, preservePosition, expectedStatus, manualUntil, preserveSchedule).
 		Scan(&queuedAt, &committedMode, &completedAt, &firstRunAt, &deadlineAt, &activeRunSeconds, &segmentStartedAt)
 	if errors.Is(err, sql.ErrNoRows) {
 		return fmt.Errorf("task %s lifecycle changed before admission (expected status %q)", id, expectedStatus)
@@ -1458,6 +1533,7 @@ func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued b
 		t.updateLifecycle(func(state *taskLifecycleState) {
 			state.Status = status
 			state.Paused = false
+			state.PauseReason = ""
 			state.Queued = queued
 			state.QueueMode = committedMode
 			state.QueuedAt = 0
@@ -1481,6 +1557,11 @@ func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued b
 			if segmentStartedAt.Valid {
 				state.RunSegmentStartedAt = segmentStartedAt.Time.Unix()
 			}
+			state.ReconnectAttempt = 0
+			state.ReconnectMax = 0
+			state.ReconnectNextAt = 0
+			state.ReconnectExhausted = false
+			state.ReconnectLastError = ""
 		})
 	}
 	m.mu.Unlock()
@@ -1494,6 +1575,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 +1591,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 +1607,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 +1627,110 @@ func (m *Manager) ApplyTaskPauseOrigin(id string, scheduled bool) error {
 	return nil
 }
 
+// ApplyProxyPause atomically pauses a running task and marks it eligible for
+// reconnect auto-resume. Already-paused tasks retain their original reason.
+func (m *Manager) ApplyProxyPause(id string) error {
+	_, err := m.ApplyProxyPauseWithCallback(id, nil)
+	return err
+}
+
+// ApplyProxyPauseWithCallback is the locked proxy-pause transition. When the
+// task actually changes from running to paused, callback runs while the
+// Server's task-control mutex is still held, after the DB and live lifecycle
+// state have committed. This lets callers install the Engine barrier before a
+// concurrent user/orchestrator pause can observe the new paused state.
+func (m *Manager) ApplyProxyPauseWithCallback(id string, callback func()) (transitioned bool, err error) {
+	if m.controlMu != nil {
+		m.controlMu.Lock()
+		defer m.controlMu.Unlock()
+	}
+	m.taskStateMu.Lock()
+	defer m.taskStateMu.Unlock()
+	n, err := strconv.ParseInt(id, 10, 64)
+	if err != nil {
+		return false, err
+	}
+	var mode string
+	var active int64
+	var started sql.NullTime
+	err = m.pg.QueryRow(`UPDATE tasks SET paused=true, pause_reason='paused_by_proxy', queued=false, queued_at=NULL,
+ active_run_seconds=active_run_seconds + CASE WHEN run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
+ run_segment_started_at=NULL WHERE id=$1 AND deleted_at IS NULL AND paused=false AND status NOT IN ('done','failed','timeout')
+ RETURNING COALESCE(queue_mode,''), active_run_seconds, run_segment_started_at`, n).Scan(&mode, &active, &started)
+	if errors.Is(err, sql.ErrNoRows) {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+	m.mu.RLock()
+	t := m.tasks[id]
+	m.mu.RUnlock()
+	if t != nil {
+		t.updateLifecycle(func(s *taskLifecycleState) {
+			s.Paused = true
+			s.PauseReason = "paused_by_proxy"
+			s.Queued = false
+			s.QueuedAt = 0
+			s.QueueMode = mode
+			s.ActiveRunSeconds = active
+			s.RunSegmentStartedAt = 0
+		})
+	}
+	if callback != nil {
+		callback()
+	}
+	return true, nil
+}
+
+// ClearPauseReason removes auto-resume eligibility without changing lifecycle.
+func (m *Manager) ClearPauseReason(id string) error {
+	if _, err := m.pg.Exec(`UPDATE tasks SET pause_reason='' WHERE id=$1 AND pause_reason='paused_by_proxy'`, id); err != nil {
+		return err
+	}
+	m.mu.RLock()
+	t := m.tasks[id]
+	m.mu.RUnlock()
+	if t != nil && t.lifecycleSnapshot().PauseReason == "paused_by_proxy" {
+		t.updateLifecycle(func(s *taskLifecycleState) {
+			s.PauseReason = ""
+			s.ReconnectAttempt, s.ReconnectMax, s.ReconnectNextAt = 0, 0, 0
+			s.ReconnectExhausted, s.ReconnectLastError = false, ""
+		})
+	}
+	return nil
+}
+
+// ConvertProxyPauseToSchedule records that a calendar window ended while a
+// task was waiting for proxy recovery. It preserves the Engine pause barrier
+// and removes proxy auto-resume eligibility until the next window.
+func (m *Manager) ConvertProxyPauseToSchedule(id string) error {
+	m.taskStateMu.Lock()
+	defer m.taskStateMu.Unlock()
+	n, err := strconv.ParseInt(id, 10, 64)
+	if err != nil {
+		return err
+	}
+	res, err := m.pg.Exec(`UPDATE tasks SET pause_reason='paused_by_schedule', schedule_paused=true WHERE id=$1 AND paused=true AND pause_reason='paused_by_proxy' AND status NOT IN ('done','failed','timeout')`, n)
+	if err != nil {
+		return err
+	}
+	if rows, _ := res.RowsAffected(); rows == 0 {
+		return nil
+	}
+	m.mu.RLock()
+	t := m.tasks[id]
+	m.mu.RUnlock()
+	if t != nil {
+		t.updateLifecycle(func(v *taskLifecycleState) {
+			v.PauseReason = "paused_by_schedule"
+			v.ReconnectAttempt, v.ReconnectMax, v.ReconnectNextAt = 0, 0, 0
+			v.ReconnectExhausted, v.ReconnectLastError = false, ""
+		})
+	}
+	return nil
+}
+
 // EnqueueTask persists the concurrency hold and syncs the in-memory handle.
 func (m *Manager) EnqueueTask(id, mode string) error {
 	m.taskStateMu.Lock()
@@ -1748,6 +1945,10 @@ func (m *Manager) SetTaskStatusGuarded(id, status string) (won bool, err error)
 			state.RunSegmentStartedAt = unixOrZero(started)
 			if pgdb.IsTerminal(status) {
 				state.RunSegmentStartedAt = 0
+				state.Paused = false
+				state.PauseReason = ""
+				state.ReconnectAttempt, state.ReconnectMax, state.ReconnectNextAt = 0, 0, 0
+				state.ReconnectExhausted, state.ReconnectLastError = false, ""
 				if state.CompletedAt == 0 {
 					state.CompletedAt = time.Now().Unix()
 				}
@@ -1784,6 +1985,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()
 				}

server/proxy_autoresume.go

@@ -0,0 +1,462 @@
+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 {
+	return s.snapshotProxyReconnectGuardsWithExhausted(checkAttempt, false)
+}
+
+func (s *Server) snapshotProxyReconnectGuardsWithExhausted(checkAttempt, includeExhausted 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
+		}
+		if st.ReconnectExhausted && !includeExhausted {
+			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 s.handleProxyPoolFailure(cause) {
+		return
+	}
+	if probe := s.runProxyProbe(s.ctx, 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 s.handleProxyPoolFailure(cause) {
+		return
+	}
+	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 s.handleProxyPoolFailure(cause) {
+		return
+	}
+	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 := maskSecrets(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()) == "" {
+		// Clearing the runtime route is a successful direct transition. Reuse the
+		// generation gate so only eligible, non-exhausted proxy pauses resume.
+		s.proxyReconnectSuccess()
+		return
+	}
+	if s.m.proxyPool != nil && len(s.m.proxyPoolEntries()) > 0 {
+		// A pool tick may confirm the active route or switch to another available
+		// member. If it cannot confirm either, retain the original #106 per-task
+		// attempt/backoff/exhaustion loop (especially for a one-member legacy
+		// pool).
+		outcome := s.proxyPoolTick()
+		if outcome.confirmed || outcome.switched {
+			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.snapshotProxyReconnectGuardsWithExhausted(false, true)
+	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
+}

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)
+	}
+}

server/proxy_pool_test.go

@@ -0,0 +1,791 @@
+package server
+
+import (
+	"bufio"
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/Autumn-27/artex/agent"
+	"github.com/Autumn-27/artex/traffic"
+)
+
+func TestParseProxyBatchValidationAndDedup(t *testing.T) {
+	entries, errs := ParseProxyBatch("http://user:[email protected]:8080|one\nhttp://user:[email protected]:8080\nftp://bad:1\n# comment\n")
+	if len(entries) != 1 || len(errs) != 2 {
+		t.Fatalf("entries=%d errs=%d", len(entries), len(errs))
+	}
+	if entries[0].URL == "" || !strings.Contains(entries[0].URL, "pass") {
+		t.Fatalf("parser lost stored credential")
+	}
+}
+
+func TestProxyPoolSnapshotMasksCredentialsAndDisabledHealth(t *testing.T) {
+	m := &Manager{proxyPool: &proxyPoolState{entries: []proxyPoolEntry{{ProxyPoolEntry: ProxyPoolEntry{ID: "1", Enabled: false, Health: "healthy"}, RawURL: "http://user:secret@host:8080"}}, logs: []ProxySwitchLog{{From: "http://user:secret@host:8080", To: "http://user:secret2@host:8081", Reason: "timed"}}}}
+	entries, _, logs := m.ProxyPoolSnapshot()
+	if strings.Contains(entries[0].URL, "secret") || entries[0].Health != "disabled" {
+		t.Fatalf("entry leaked or wrong state: %+v", entries[0])
+	}
+	if strings.Contains(logs[0].From, "secret") || strings.Contains(logs[0].To, "secret2") {
+		t.Fatalf("switch log leaked credentials: %+v", logs[0])
+	}
+}
+
+func TestProxyPoolSequentialSkipsUnhealthyAndDisabled(t *testing.T) {
+	m := &Manager{proxyPool: &proxyPoolState{active: "a", entries: []proxyPoolEntry{
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"},
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: false, Health: "healthy"}, RawURL: "http://b"},
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "c", Enabled: true, Health: "unhealthy-cooling"}, RawURL: "http://c"},
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "d", Enabled: true, Health: "healthy"}, RawURL: "http://d"},
+	}}}
+	if got := m.selectHealthyLocked("a"); got != "d" {
+		t.Fatalf("selected %q", got)
+	}
+}
+
+func TestProxyPoolIntervalLowerBound(t *testing.T) {
+	if minProxyRotateInterval >= 5 {
+		t.Fatalf("bound must allow the documented 5s setting")
+	}
+	if minProxyRotateInterval != 3 {
+		t.Fatalf("bound=%d, rationale expects 3s", minProxyRotateInterval)
+	}
+}
+
+func TestProxyPoolPersistenceKeepsRawURL(t *testing.T) {
+	b, err := json.Marshal(proxyPoolPersist{Entries: []proxyPoolEntry{{ProxyPoolEntry: ProxyPoolEntry{ID: "1", URL: "http://***@host:1"}, RawURL: "http://user:pass@host:1"}}})
+	if err != nil || !strings.Contains(string(b), "raw_url") || !strings.Contains(string(b), "pass") {
+		t.Fatalf("bad persistence encoding: %s", b)
+	}
+}
+
+func TestProxyPoolURLKeyNormalizesHostCase(t *testing.T) {
+	a, _ := traffic.ValidateProxyURL("http://user:pass@HOST:8080")
+	b, _ := traffic.ValidateProxyURL("http://user:pass@host:8080")
+	if proxyURLKey(a) != proxyURLKey(b) {
+		t.Fatalf("keys differ: %q %q", proxyURLKey(a), proxyURLKey(b))
+	}
+}
+
+func TestProxyPoolHealthThresholdCooldownAndRecovery(t *testing.T) {
+	m := &Manager{proxyPool: &proxyPoolState{entries: []proxyPoolEntry{{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"}}}}
+	fail := egressProbeResult{Error: "http://user:pass@proxy: connection refused"}
+	for i := 0; i < defaultProxyHealthThreshold-1; i++ {
+		m.markProxyProbe("a", fail)
+		if got := m.proxyPool.entries[0].Health; got != "healthy" {
+			t.Fatalf("failure %d moved to %s before threshold", i+1, got)
+		}
+	}
+	m.markProxyProbe("a", fail)
+	if got := m.proxyPool.entries[0]; got.Health != "unhealthy-cooling" || got.ConsecutiveFailures != defaultProxyHealthThreshold {
+		t.Fatalf("threshold state=%+v", got)
+	}
+	if got := m.proxyPool.entries[0].LastError; strings.Contains(got, "pass") || !strings.Contains(got, "***@") {
+		t.Fatalf("failure was not masked: %q", got)
+	}
+	m.markProxyProbe("a", egressProbeResult{OK: true, ExitIP: "203.0.113.9"})
+	if got := m.proxyPool.entries[0]; got.Health != "healthy" || got.ConsecutiveFailures != 0 || got.EgressIP != "203.0.113.9" {
+		t.Fatalf("recovery state=%+v", got)
+	}
+}
+
+func TestProxyPoolUncheckedMemberDoesNotPauseBeforeFailureThreshold(t *testing.T) {
+	m := proxyTestManager(t)
+	task, err := m.CreateTask("unchecked proxy", "goal", nil, 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
+	m.globalProxy = "http://a"
+	m.proxyPool = &proxyPoolState{active: "a", entries: []proxyPoolEntry{{
+		ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "unknown"}, RawURL: "http://a",
+	}}}
+	s := &Server{m: m, ctx: context.Background(), proxyPoolProbeSem: make(chan struct{}, proxyPoolMaxProbeConcurrency)}
+	s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
+		return egressProbeResult{Error: "proxy down"}
+	}
+	s.proxyPoolTickLocked(true)
+	if st := task.lifecycleSnapshot(); st.Paused || st.PauseReason == "paused_by_proxy" {
+		t.Fatalf("unchecked member paused the task: %+v", st)
+	}
+	if got := m.proxyPool.entries[0].Health; got != "unknown" {
+		t.Fatalf("single failure changed unchecked member to %q", got)
+	}
+}
+
+func TestProxyPoolCoolingWindowSkipsProbeUntilExpiry(t *testing.T) {
+	m := proxyTestManager(t)
+	m.globalProxy = "http://a"
+	m.proxyPool = &proxyPoolState{active: "a", entries: []proxyPoolEntry{{
+		ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "unhealthy-cooling", ConsecutiveFailures: 3, LastCheckedAt: time.Now()}, RawURL: "http://a",
+	}}}
+	s := &Server{m: m, ctx: context.Background(), proxyPoolProbeSem: make(chan struct{}, proxyPoolMaxProbeConcurrency)}
+	calls := 0
+	s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
+		calls++
+		return egressProbeResult{OK: true, ExitIP: "203.0.113.5"}
+	}
+	s.proxyPoolTickLocked(false)
+	if calls != 0 {
+		t.Fatalf("cooling member was probed before cooldown expiry: calls=%d", calls)
+	}
+	m.proxyPool.entries[0].LastCheckedAt = time.Now().Add(-2 * time.Minute)
+	s.proxyPoolTickLocked(false)
+	if calls != 1 || m.proxyPool.entries[0].Health != "healthy" {
+		t.Fatalf("expired cooling member did not recover: calls=%d entry=%+v", calls, m.proxyPool.entries[0])
+	}
+}
+
+func TestProxyPoolTimedRotationUsesAvailableMembersAndRecoversTasks(t *testing.T) {
+	m := proxyTestManager(t)
+	if err := m.SetProxyPoolSettings(true, minProxyRotateInterval, true, "sequential", 3, 60, 30); err != nil {
+		t.Fatal(err)
+	}
+	m.globalProxy = "http://a"
+	m.proxyPool = &proxyPoolState{active: "a", lastTimed: time.Now().Add(-time.Minute), lastHealth: time.Now(), entries: []proxyPoolEntry{
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"},
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: true, Health: "healthy"}, RawURL: "http://b"},
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "c", Enabled: true, Health: "unhealthy-cooling"}, RawURL: "http://c"},
+	}}
+	s := &Server{m: m, ctx: context.Background()}
+	s.proxyPoolTickLocked(false)
+	if got := m.GlobalProxy(); got != "http://b" {
+		t.Fatalf("timed rotation selected %q", got)
+	}
+	if len(m.proxyPool.logs) == 0 || m.proxyPool.logs[len(m.proxyPool.logs)-1].Reason != "timed" {
+		t.Fatalf("timed rotation was not logged: %+v", m.proxyPool.logs)
+	}
+}
+
+func TestProxyPoolFailureSwitchRecoversOnlyEligibleTasks(t *testing.T) {
+	m := proxyTestManager(t)
+	if err := m.SetProxyPoolSettings(false, minProxyRotateInterval, true, "sequential", 3, 60, 30); err != nil {
+		t.Fatal(err)
+	}
+	m.globalProxy = "http://a"
+	m.proxyPool = &proxyPoolState{active: "a", entries: []proxyPoolEntry{
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"},
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: true, Health: "healthy"}, RawURL: "http://b"},
+	}}
+	eligible, err := m.CreateTask("eligible", "goal", nil, 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	exhausted, err := m.CreateTask("exhausted", "goal", nil, 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		_, _ = m.DeleteTask(eligible.ID, DeleteTaskOptions{})
+		_, _ = m.DeleteTask(exhausted.ID, DeleteTaskOptions{})
+	}()
+	for _, task := range []*Task{eligible, exhausted} {
+		if err := m.ApplyProxyPause(task.ID); err != nil {
+			t.Fatal(err)
+		}
+		task.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectGeneration = 7; v.ReconnectMax = 3 })
+	}
+	exhausted.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectAttempt, v.ReconnectExhausted = 3, true })
+	s := newAdmissionTestServer(m, nil)
+	m.SetTaskControlMutex(&s.concMu)
+	s.proxyPoolWake = make(chan struct{}, 1)
+	s.proxyPoolProbeSem = make(chan struct{}, proxyPoolMaxProbeConcurrency)
+	s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
+		return egressProbeResult{Error: "proxy down"}
+	}
+	s.engine.Pause(eligible.ID, agent.AbortPausedByProxy)
+	s.engine.Pause(exhausted.ID, agent.AbortPausedByProxy)
+	if !s.handleProxyPoolFailure(errors.New("proxyconnect tcp: connection refused")) {
+		t.Fatal("confirmed failure did not switch to the available member")
+	}
+	if got := m.GlobalProxy(); got != "http://b" {
+		t.Fatalf("failure switch selected %q", got)
+	}
+	if st := eligible.lifecycleSnapshot(); st.PauseReason != "" || st.Paused {
+		t.Fatalf("eligible task did not recover: %+v", st)
+	}
+	if st := exhausted.lifecycleSnapshot(); st.PauseReason != "paused_by_proxy" || !st.ReconnectExhausted {
+		t.Fatalf("exhausted task was automatically resumed: %+v", st)
+	}
+}
+
+func TestProxyPoolReconnectKeepsAttemptsAndDoesNotAutoResumeExhausted(t *testing.T) {
+	m := proxyTestManager(t)
+	if err := m.pg.SetSetting(settingProxyReconnectMax, "2"); err != nil {
+		t.Fatal(err)
+	}
+	if err := m.pg.SetSetting(settingProxyReconnectInterval, "1"); err != nil {
+		t.Fatal(err)
+	}
+	m.globalProxy = "http://legacy"
+	m.proxyPool = &proxyPoolState{active: "legacy", entries: []proxyPoolEntry{{
+		ProxyPoolEntry: ProxyPoolEntry{ID: "legacy", Enabled: true, Health: "healthy"}, RawURL: "http://legacy",
+	}}}
+	task, err := m.CreateTask("legacy reconnect", "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.ReconnectGeneration = 3 })
+	s := &Server{m: m, ctx: context.Background(), proxyPoolProbeSem: make(chan struct{}, proxyPoolMaxProbeConcurrency)}
+	s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
+		return egressProbeResult{Error: "proxy down"}
+	}
+	s.proxyReconnectTick()
+	first := task.lifecycleSnapshot()
+	if first.ReconnectAttempt != 1 || first.ReconnectExhausted {
+		t.Fatalf("first pool reconnect attempt=%+v", first)
+	}
+	task.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectNextAt = 0 })
+	s.proxyReconnectTick()
+	second := task.lifecycleSnapshot()
+	if second.ReconnectAttempt != 2 || !second.ReconnectExhausted || second.PauseReason != "paused_by_proxy" {
+		t.Fatalf("second pool reconnect did not exhaust: %+v", second)
+	}
+	s.proxyReconnectTick()
+	third := task.lifecycleSnapshot()
+	if third.ReconnectAttempt != second.ReconnectAttempt || !third.ReconnectExhausted || third.PauseReason != "paused_by_proxy" {
+		t.Fatalf("automatic tick revived exhausted task: %+v", third)
+	}
+	guards := s.snapshotProxyReconnectGuardsWithExhausted(false, true)
+	s.proxyReconnectSuccessWithGuards(guards, nil)
+	if got := task.lifecycleSnapshot(); got.PauseReason != "" {
+		t.Fatalf("manual guarded resume did not release exhausted task: %+v", got)
+	}
+}
+
+func TestProxyPoolDisableOrClearActiveRoutesAway(t *testing.T) {
+	newServer := func(entries []proxyPoolEntry) (*Server, *Manager) {
+		m := &Manager{globalProxy: "http://a", proxyPool: &proxyPoolState{active: "a", entries: entries}}
+		return &Server{m: m}, m
+	}
+	t.Run("disable switches to healthy member", func(t *testing.T) {
+		s, m := newServer([]proxyPoolEntry{
+			{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"},
+			{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: true, Health: "healthy"}, RawURL: "http://b"},
+		})
+		r := httptest.NewRequest(http.MethodPatch, "/api/settings/proxy-pool/a", strings.NewReader(`{"enabled":false}`))
+		r.SetPathValue("id", "a")
+		w := httptest.NewRecorder()
+		s.updateProxyPool(w, r)
+		if w.Code != http.StatusOK || m.GlobalProxy() != "http://b" || m.proxyPoolActiveID() != "b" {
+			t.Fatalf("disable route status=%d proxy=%q active=%q body=%s", w.Code, m.GlobalProxy(), m.proxyPoolActiveID(), w.Body.String())
+		}
+	})
+	t.Run("clear last member clears runtime", func(t *testing.T) {
+		s, m := newServer([]proxyPoolEntry{{
+			ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a",
+		}})
+		r := httptest.NewRequest(http.MethodPatch, "/api/settings/proxy-pool/a", strings.NewReader(`{"url":""}`))
+		r.SetPathValue("id", "a")
+		w := httptest.NewRecorder()
+		s.updateProxyPool(w, r)
+		if w.Code != http.StatusOK || m.GlobalProxy() != "" || m.proxyPoolActiveID() != "" {
+			t.Fatalf("clear route status=%d proxy=%q active=%q body=%s", w.Code, m.GlobalProxy(), m.proxyPoolActiveID(), w.Body.String())
+		}
+	})
+}
+
+func TestProxyPoolClearOrDeleteLastMemberRecoversOnlyEligibleProxyPauses(t *testing.T) {
+	run := func(t *testing.T, deleteLast bool) {
+		m := proxyTestManager(t)
+		m.globalProxy = "http://a"
+		m.proxyPool = &proxyPoolState{active: "a", entries: []proxyPoolEntry{{
+			ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a",
+		}}}
+		eligible, err := m.CreateTask("eligible", "goal", nil, 0, 0)
+		if err != nil {
+			t.Fatal(err)
+		}
+		exhausted, err := m.CreateTask("exhausted", "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)
+		}
+		defer func() {
+			_, _ = m.DeleteTask(eligible.ID, DeleteTaskOptions{})
+			_, _ = m.DeleteTask(exhausted.ID, DeleteTaskOptions{})
+			_, _ = m.DeleteTask(userPaused.ID, DeleteTaskOptions{})
+		}()
+		for _, task := range []*Task{eligible, exhausted} {
+			if err := m.ApplyProxyPause(task.ID); err != nil {
+				t.Fatal(err)
+			}
+			task.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectGeneration = 7 })
+		}
+		exhausted.updateLifecycle(func(v *taskLifecycleState) {
+			v.ReconnectAttempt, v.ReconnectExhausted = 3, true
+		})
+		if err := m.ApplyTaskPauseOrigin(userPaused.ID, false); err != nil {
+			t.Fatal(err)
+		}
+		s := newAdmissionTestServer(m, nil)
+		for _, task := range []*Task{eligible, exhausted, userPaused} {
+			s.engine.Pause(task.ID, agent.AbortPausedByProxy)
+		}
+		if deleteLast {
+			r := httptest.NewRequest(http.MethodDelete, "/api/settings/proxy-pool/a", nil)
+			r.SetPathValue("id", "a")
+			w := httptest.NewRecorder()
+			s.deleteProxyPool(w, r)
+			if w.Code != http.StatusOK {
+				t.Fatalf("delete status=%d body=%s", w.Code, w.Body.String())
+			}
+		} else {
+			r := httptest.NewRequest(http.MethodPatch, "/api/settings/proxy-pool/a", strings.NewReader(`{"url":""}`))
+			r.SetPathValue("id", "a")
+			w := httptest.NewRecorder()
+			s.updateProxyPool(w, r)
+			if w.Code != http.StatusOK {
+				t.Fatalf("clear status=%d body=%s", w.Code, w.Body.String())
+			}
+		}
+		if got := eligible.lifecycleSnapshot(); got.PauseReason != "" || got.Paused {
+			t.Fatalf("eligible proxy pause was not recovered: %+v", got)
+		}
+		if got := exhausted.lifecycleSnapshot(); got.PauseReason != "paused_by_proxy" || !got.ReconnectExhausted {
+			t.Fatalf("exhausted proxy pause was recovered: %+v", got)
+		}
+		if got := userPaused.lifecycleSnapshot(); got.PauseReason != "paused_by_user" {
+			t.Fatalf("user pause was recovered: %+v", got)
+		}
+	}
+	t.Run("clear", func(t *testing.T) { run(t, false) })
+	t.Run("delete", func(t *testing.T) { run(t, true) })
+}
+
+func TestProxyPoolCaptureOffUsesLocalForwarder(t *testing.T) {
+	fp, err := newDynamicForwardProxy(func() string { return "" })
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer fp.Close()
+	m := &Manager{forwardProxy: fp}
+	m.proxyPoolConfigured.Store(true)
+	got := m.ProxyAddr()
+	if !strings.HasPrefix(got, "http://127.0.0.1:") {
+		t.Fatalf("capture-off proxy address=%q, want local forwarder", got)
+	}
+}
+
+func TestProxyPoolClearedRuntimeForwarderFallsBackToDirect(t *testing.T) {
+	target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.WriteString(w, "direct-ok")
+	}))
+	defer target.Close()
+	m := proxyTestManager(t)
+	// A configured pool can remain persisted after its last usable member is
+	// disabled or cleared. The runtime route is still direct in that state.
+	m.proxyPoolConfigured.Store(true)
+	proxyURL, err := url.Parse(m.ProxyAddr())
+	if err != nil {
+		t.Fatal(err)
+	}
+	client := &http.Client{Transport: &http.Transport{Proxy: http.ProxyURL(proxyURL)}}
+	resp, err := client.Get(target.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	body, _ := io.ReadAll(resp.Body)
+	if resp.StatusCode != http.StatusOK || string(body) != "direct-ok" {
+		t.Fatalf("cleared runtime route status=%d body=%q", resp.StatusCode, body)
+	}
+}
+
+func TestProxyReconnectGuardsGateExhaustedAutomaticResume(t *testing.T) {
+	task := &Task{ID: "exhausted", PauseReason: "paused_by_proxy", ReconnectGeneration: 4, ReconnectAttempt: 3, ReconnectExhausted: true}
+	m := &Manager{tasks: map[string]*Task{task.ID: task}}
+	s := &Server{m: m}
+	if got := s.snapshotProxyReconnectGuards(false); len(got) != 0 {
+		t.Fatalf("automatic guards included exhausted task: %+v", got)
+	}
+	if got := s.snapshotProxyReconnectGuardsWithExhausted(false, true); len(got) != 1 {
+		t.Fatalf("manual guards did not include exhausted task: %+v", got)
+	}
+}
+
+func TestProxyPoolSettingsRejectTooShortInterval(t *testing.T) {
+	m := &Manager{}
+	if err := m.SetProxyPoolSettings(true, minProxyRotateInterval-1, true, "sequential", 3, 60, 30); err == nil {
+		t.Fatal("interval below lower bound was accepted")
+	}
+}
+
+func TestProxyPoolRandomSelectionStillExcludesDisabledAndCooling(t *testing.T) {
+	m := proxyTestManager(t)
+	if err := m.pg.SetSetting(settingProxyRotateStrategy, "random"); err != nil {
+		t.Fatal(err)
+	}
+	m.proxyPool = &proxyPoolState{entries: []proxyPoolEntry{
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "disabled", Enabled: false, Health: "healthy"}, RawURL: "http://disabled"},
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "cooling", Enabled: true, Health: "unhealthy-cooling"}, RawURL: "http://cooling"},
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "empty", Enabled: true, Health: "healthy"}},
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "good", Enabled: true, Health: "healthy"}, RawURL: "http://good"},
+	}}
+	for i := 0; i < 20; i++ {
+		if got := m.selectHealthyLocked(""); got != "good" {
+			t.Fatalf("random selected %q", got)
+		}
+	}
+}
+
+func TestProxyPoolProbeSkipsEnabledEmptyURL(t *testing.T) {
+	m := proxyTestManager(t)
+	m.globalProxy = "http://good"
+	m.proxyPool = &proxyPoolState{active: "good", entries: []proxyPoolEntry{
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "empty", Enabled: true, Health: "healthy"}},
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "good", Enabled: true, Health: "healthy"}, RawURL: "http://good"},
+	}}
+	s := &Server{m: m, ctx: context.Background(), proxyPoolProbeSem: make(chan struct{}, proxyPoolMaxProbeConcurrency)}
+	calls := 0
+	s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
+		calls++
+		return egressProbeResult{OK: true}
+	}
+	s.proxyPoolTickLocked(true)
+	if calls != 1 {
+		t.Fatalf("probe calls=%d, want only non-empty enabled member", calls)
+	}
+}
+
+func TestProxyPoolFailureMarksOriginalMemberAfterConcurrentSwitch(t *testing.T) {
+	m := &Manager{globalProxy: "http://a", proxyPool: &proxyPoolState{active: "a", entries: []proxyPoolEntry{
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"},
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: true, Health: "healthy"}, RawURL: "http://b"},
+	}}}
+	s := &Server{m: m, ctx: context.Background(), proxyPoolWake: make(chan struct{}, 1)}
+	s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
+		m.mu.Lock()
+		m.globalProxy = "http://b"
+		m.mu.Unlock()
+		return egressProbeResult{Error: "proxy connection refused"}
+	}
+	if !s.handleProxyPoolFailure(fmt.Errorf("proxyconnect tcp: connection refused")) {
+		t.Fatal("failure after concurrent switch was not consumed")
+	}
+	if got := m.proxyPool.entries[0].ConsecutiveFailures; got != 1 {
+		t.Fatalf("original member failure count=%d", got)
+	}
+}
+
+func TestProxyPoolAllMembersUnhealthyKeepsProxyConfigured(t *testing.T) {
+	m := &Manager{globalProxy: "http://a", proxyPool: &proxyPoolState{active: "a", entries: []proxyPoolEntry{
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "unhealthy-cooling", ConsecutiveFailures: defaultProxyHealthThreshold, LastCheckedAt: time.Now().Add(-2 * time.Minute)}, RawURL: "http://a"},
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: true, Health: "unhealthy-cooling", ConsecutiveFailures: defaultProxyHealthThreshold, LastCheckedAt: time.Now().Add(-2 * time.Minute)}, RawURL: "http://b"},
+	}}}
+	s := &Server{m: m, ctx: context.Background(), proxyPoolProbeSem: make(chan struct{}, proxyPoolMaxProbeConcurrency)}
+	s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
+		return egressProbeResult{Error: "proxy down"}
+	}
+	s.proxyPoolTickLocked(true)
+	if got := m.GlobalProxy(); got != "http://a" {
+		t.Fatalf("all unhealthy members fell back to direct: %q", got)
+	}
+	if got := m.proxyPoolActiveID(); got != "a" {
+		t.Fatalf("active member changed without a healthy replacement: %q", got)
+	}
+}
+
+func TestProxyPoolMaskedEditHotCutsCoolingActiveMember(t *testing.T) {
+	m := &Manager{globalProxy: "http://old:[email protected]:8080", proxyPool: &proxyPoolState{active: "a", entries: []proxyPoolEntry{{
+		ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "unhealthy-cooling", ConsecutiveFailures: 3},
+		RawURL:         "http://old:[email protected]:8080",
+	}}}}
+	s := &Server{m: m}
+	r := httptest.NewRequest(http.MethodPatch, "/api/settings/proxy-pool/a", strings.NewReader(`{"url":"http://***@new.example:9090"}`))
+	r.SetPathValue("id", "a")
+	w := httptest.NewRecorder()
+	s.updateProxyPool(w, r)
+	if w.Code != http.StatusOK {
+		t.Fatalf("status=%d body=%s", w.Code, w.Body.String())
+	}
+	if got := m.GlobalProxy(); got != "http://old:[email protected]:9090" {
+		t.Fatalf("active proxy=%q", got)
+	}
+	if got := m.proxyPool.entries[0].Health; got != "healthy" {
+		t.Fatalf("edited active member health=%q", got)
+	}
+}
+
+func TestProxyPoolDeleteNeverRetainsDeletedRuntimeURL(t *testing.T) {
+	m := &Manager{globalProxy: "http://a", proxyPool: &proxyPoolState{active: "a", entries: []proxyPoolEntry{
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"},
+		{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: true, Health: "healthy"}, RawURL: "http://b"},
+	}}}
+	s := &Server{m: m}
+	deleteEntry := func(id string) {
+		r := httptest.NewRequest(http.MethodDelete, "/api/settings/proxy-pool/"+id, nil)
+		r.SetPathValue("id", id)
+		w := httptest.NewRecorder()
+		s.deleteProxyPool(w, r)
+		if w.Code != http.StatusOK {
+			t.Fatalf("delete %s status=%d body=%s", id, w.Code, w.Body.String())
+		}
+	}
+	deleteEntry("a")
+	if got := m.GlobalProxy(); got != "http://b" {
+		t.Fatalf("surviving pool member was not activated: %q", got)
+	}
+	deleteEntry("b")
+	if got := m.GlobalProxy(); got != "" || m.proxyPool.active != "" {
+		t.Fatalf("deleted last member left runtime proxy=%q active=%q", got, m.proxyPool.active)
+	}
+}
+
+func TestSetGlobalProxyIsReadOnlyWhenPoolManaged(t *testing.T) {
+	m := &Manager{globalProxy: "http://pool.example:8080", proxyPool: &proxyPoolState{entries: []proxyPoolEntry{{
+		ProxyPoolEntry: ProxyPoolEntry{ID: "pool", Enabled: true, Health: "healthy"}, RawURL: "http://pool.example:8080",
+	}}}}
+	if err := m.SetGlobalProxy("http://other.example:8080"); err != nil {
+		t.Fatal(err)
+	}
+	if got := m.GlobalProxy(); got != "http://pool.example:8080" {
+		t.Fatalf("managed pool runtime changed through legacy setting: %q", got)
+	}
+}
+
+func TestProxyPoolEmptyPersistenceDoesNotResurrectLegacyProxy(t *testing.T) {
+	dir := t.TempDir()
+	m, err := NewManager(dir, "")
+	if err != nil {
+		t.Skipf("postgres unavailable (%v) - skipping", err)
+	}
+	m.proxyPool = &proxyPoolState{active: "a", entries: []proxyPoolEntry{{
+		ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://deleted.example:8080",
+	}}}
+	m.globalProxy = "http://deleted.example:8080"
+	if err := m.persistProxyPoolLocked(); err != nil {
+		_ = m.Close()
+		t.Fatal(err)
+	}
+	s := &Server{m: m}
+	r := httptest.NewRequest(http.MethodDelete, "/api/settings/proxy-pool/a", nil)
+	r.SetPathValue("id", "a")
+	w := httptest.NewRecorder()
+	s.deleteProxyPool(w, r)
+	if w.Code != http.StatusOK {
+		_ = m.Close()
+		t.Fatalf("delete status=%d body=%s", w.Code, w.Body.String())
+	}
+	if got := m.GlobalProxy(); got != "" {
+		t.Fatalf("delete left runtime proxy=%q", got)
+	}
+	if err := m.Close(); err != nil {
+		t.Fatal(err)
+	}
+	m2, err := NewManager(dir, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer m2.Close()
+	if got := m2.GlobalProxy(); got != "" || len(m2.proxyPoolEntries()) != 0 {
+		t.Fatalf("empty persisted pool resurrected proxy=%q entries=%d", got, len(m2.proxyPoolEntries()))
+	}
+}
+
+func TestMergeMaskedProxyURLPreservesOnlyPassword(t *testing.T) {
+	got, err := mergeMaskedProxyURL("http://***@NEW.example:9090", "http://old:[email protected]:8080")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got != "http://old:[email protected]:9090" {
+		t.Fatalf("got %q", got)
+	}
+}
+
+func TestMergeMaskedProxyURLErrorIsMasked(t *testing.T) {
+	_, err := mergeMaskedProxyURL("http://user:secret@", "")
+	if err == nil || strings.Contains(err.Error(), "secret") {
+		t.Fatalf("unmasked error: %v", err)
+	}
+}
+
+func TestDynamicForwardProxyUsesCurrentUpstream(t *testing.T) {
+	target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte("ok")) }))
+	defer target.Close()
+	up := ""
+	fp, err := newDynamicForwardProxy(func() string { return up })
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer fp.Close()
+	pu, _ := url.Parse(fp.Addr())
+	c := &http.Client{Transport: &http.Transport{Proxy: http.ProxyURL(pu)}}
+	resp, err := c.Get(target.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != 200 {
+		t.Fatalf("status %d", resp.StatusCode)
+	}
+}
+
+func TestDynamicForwardProxyConnectAuthenticatedUpstreamAndBufferedBytes(t *testing.T) {
+	const user, pass = "alice", "s3cret"
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodConnect {
+			http.Error(w, "connect required", http.StatusMethodNotAllowed)
+			return
+		}
+		want := "Basic " + base64.StdEncoding.EncodeToString([]byte(user+":"+pass))
+		if r.Header.Get("Proxy-Authorization") != want {
+			w.Header().Set("Proxy-Authenticate", `Basic realm="test"`)
+			w.WriteHeader(http.StatusProxyAuthRequired)
+			return
+		}
+		h, ok := w.(http.Hijacker)
+		if !ok {
+			http.Error(w, "no hijack", 500)
+			return
+		}
+		c, rw, err := h.Hijack()
+		if err != nil {
+			return
+		}
+		defer c.Close()
+		_, _ = rw.WriteString("HTTP/1.1 200 Connection Established\r\n\r\nupstream-buffer")
+		_ = rw.Flush()
+	}))
+	defer upstream.Close()
+	up, _ := url.Parse(upstream.URL)
+	up.User = url.UserPassword(user, pass)
+	fp, err := newDynamicForwardProxy(func() string { return up.String() })
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer fp.Close()
+
+	c, err := net.Dial("tcp", strings.TrimPrefix(fp.Addr(), "http://"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close()
+	_, _ = fmt.Fprintf(c, "CONNECT example.test:443 HTTP/1.1\r\nHost: example.test:443\r\n\r\n")
+	rd := bufio.NewReader(c)
+	resp, err := http.ReadResponse(rd, &http.Request{Method: http.MethodConnect})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("local CONNECT status %s", resp.Status)
+	}
+	b, err := io.ReadAll(io.LimitReader(rd, int64(len("upstream-buffer"))))
+	if err != nil || string(b) != "upstream-buffer" {
+		t.Fatalf("buffered bytes=%q err=%v", b, err)
+	}
+}
+
+func TestDynamicForwardProxyConnectForwardsHijackBufferedClientBytes(t *testing.T) {
+	targetLn, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer targetLn.Close()
+	go func() {
+		c, err := targetLn.Accept()
+		if err != nil {
+			return
+		}
+		defer c.Close()
+		buf := make([]byte, len("hello"))
+		if _, err := io.ReadFull(c, buf); err == nil {
+			_, _ = c.Write([]byte("target"))
+		}
+	}()
+	proxyServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodConnect {
+			http.Error(w, "connect required", 405)
+			return
+		}
+		c, err := net.Dial("tcp", targetLn.Addr().String())
+		if err != nil {
+			http.Error(w, err.Error(), 502)
+			return
+		}
+		h, _ := w.(http.Hijacker)
+		client, rw, err := h.Hijack()
+		if err != nil {
+			_ = c.Close()
+			return
+		}
+		defer client.Close()
+		defer c.Close()
+		_, _ = rw.WriteString("HTTP/1.1 200 Connection Established\r\n\r\n")
+		_ = rw.Flush()
+		buf := make([]byte, len("hello"))
+		if _, err := io.ReadFull(rw.Reader, buf); err != nil {
+			return
+		}
+		_, _ = c.Write(buf)
+		response := make([]byte, len("target"))
+		if _, err := io.ReadFull(c, response); err == nil {
+			_, _ = client.Write(response)
+		}
+	}))
+	defer proxyServer.Close()
+	fp, err := newDynamicForwardProxy(func() string { return proxyServer.URL })
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer fp.Close()
+	c, err := net.Dial("tcp", strings.TrimPrefix(fp.Addr(), "http://"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close()
+	_, _ = fmt.Fprintf(c, "CONNECT %s HTTP/1.1\r\nHost: %s\r\n\r\nhello", targetLn.Addr().String(), targetLn.Addr().String())
+	rd := bufio.NewReader(c)
+	resp, err := http.ReadResponse(rd, &http.Request{Method: http.MethodConnect})
+	if err != nil || resp.StatusCode != http.StatusOK {
+		t.Fatalf("CONNECT response=%v status=%v", err, resp)
+	}
+	_ = c.SetReadDeadline(time.Now().Add(2 * time.Second))
+	b, err := io.ReadAll(io.LimitReader(rd, int64(len("target"))))
+	if err != nil || string(b) != "target" {
+		t.Fatalf("target response=%q err=%v", b, err)
+	}
+}
+
+func TestDialHTTPConnectBlackholeHonorsContext(t *testing.T) {
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ln.Close()
+	go func() {
+		c, err := ln.Accept()
+		if err == nil {
+			defer c.Close()
+			_, _ = io.Copy(io.Discard, c)
+		}
+	}()
+	u, _ := url.Parse("http://" + ln.Addr().String())
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	started := time.Now()
+	_, err = dialHTTPConnect(ctx, u, "example.test:443")
+	if err == nil {
+		t.Fatal("blackholed CONNECT unexpectedly succeeded")
+	}
+	if elapsed := time.Since(started); elapsed > time.Second {
+		t.Fatalf("context cancellation took %s", elapsed)
+	}
+}

server/server.go

@@ -125,10 +125,18 @@ type Server struct {
 
 	// Cold task archives run through one persistent FIFO worker. The buffered wake
 	// channel coalesces enqueue bursts; the database remains the source of truth.
-	archiveWake    chan struct{}
+	archiveWake       chan struct{}
-	archiveWG      sync.WaitGroup
+	archiveWG         sync.WaitGroup
-	credentialWake chan struct{}
+	credentialWake    chan struct{}
-	side           *sideQuestionState
+	proxyReconnect    proxyReconnectSupervisor
+	proxyPoolWake     chan struct{}
+	proxyPoolWG       sync.WaitGroup
+	proxyPoolProbeSem chan struct{}
+	proxyPoolTickMu   sync.Mutex
+	// 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.
@@ -159,10 +167,11 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
 	}
 	s := &Server{m: m, engine: NewEngine(m), ctx: ctx, skillDir: skillDir, dataDir: dataDir, jwtKey: key, chatBusy: map[string]bool{},
 		chatCancel: map[string]context.CancelCauseFunc{}, triggerQ: map[string][]triggeredRun{},
-		triggerActive: map[string]int{}, triggerCfg: map[string]triggerBehavior{},
+		triggerActive: map[string]int{}, triggerCfg: map[string]triggerBehavior{}, proxyPoolWake: make(chan struct{}, 1), proxyPoolProbeSem: make(chan struct{}, proxyPoolMaxProbeConcurrency),
 		profChatAgents: map[int64]*agent.ChatAgent{},
 		provByProfile:  map[int64]*provEntry{}, llmHealth: newLLMHealthRegistry(m.pg),
 		taskAgents: map[string]*taskAgentBundle{}, archiveWake: make(chan struct{}, 1), credentialWake: make(chan struct{}, 1)}
+	m.SetTaskControlMutex(&s.concMu)
 	s.initSideQuestions()
 	if m.pg != nil {
 		if err := s.basicAuth.load(m.pg); err != nil {
@@ -192,6 +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 {
@@ -280,6 +293,9 @@ 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()
+	s.proxyPoolWG.Add(1)
+	go func() { defer s.proxyPoolWG.Done(); s.runProxyPoolSupervisor(s.ctx) }()
 	go s.reconcileConcurrency()
 	s.startTaskArchiveWorker()
 	if m.pg != nil {
@@ -296,6 +312,13 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
 	return s
 }
 
+// Close waits for proxy pool supervision to stop before Manager closes the DB.
+func (s *Server) Close() {
+	if s != nil {
+		s.proxyPoolWG.Wait()
+	}
+}
+
 // Restored deadline and worker loops must inherit the same context as new tasks,
 // including the side-question checkpoint publisher installed during startup.
 func (s *Server) restoreTaskRuntimes() {
@@ -509,6 +532,7 @@ func (s *Server) applyLLM(cfg agent.Config) error {
 	// (backend/key) and gates Enabled per-conversation-agent at Chat time. 对话始终用激活配置。
 	s.chatAgent = agent.NewChatAgent(prov, cfg.Model, s.m.dir, tx, win) // chat page runner
 	s.chatAgent.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
+	s.chatAgent.SetProxyProvider(s.m.ProxyAddr)
 	s.chatAgent.SetWebSearch(s.m.WebSearchOpts())
 	s.chatAgent.SetGuard(s.chatGuard())
 	s.chatAgent.SetNonStreaming(nonStreamingResolver(cfg))
@@ -671,6 +695,7 @@ func (s *Server) chatAgentForProfile(id int64) *agent.ChatAgent {
 	tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts"))
 	ca := agent.NewChatAgent(s.poolForBinding(id, prov, cfg), cfg.Model, s.m.dir, tx, cfg.CompactionWindow())
 	ca.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
+	ca.SetProxyProvider(s.m.ProxyAddr)
 	ca.SetWebSearch(s.m.WebSearchOpts())
 	ca.SetGuard(s.chatGuard())
 	ca.SetNonStreaming(nonStreamingResolver(cfg))
@@ -927,6 +952,14 @@ 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("GET /api/settings/proxy-pool", s.getProxyPool)
+	mux.HandleFunc("POST /api/settings/proxy-pool/import", s.importProxyPool)
+	mux.HandleFunc("PUT /api/settings/proxy-pool/{id}", s.updateProxyPool)
+	mux.HandleFunc("DELETE /api/settings/proxy-pool/{id}", s.deleteProxyPool)
+	mux.HandleFunc("POST /api/settings/proxy-pool/{id}/test", s.testProxyPoolEntry)
+	mux.HandleFunc("POST /api/settings/proxy-pool/test-all", s.testAllProxyPool)
+	mux.HandleFunc("POST /api/settings/proxy-pool/{id}/activate", s.activateProxyPool)
+	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)
@@ -2332,7 +2365,10 @@ func (s *Server) findingsExport(w http.ResponseWriter, r *http.Request) {
 	stamp := now.Format("20060102-150405")
 	setDownload := func(contentType, filename string) {
 		w.Header().Set("Content-Type", contentType)
-		w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"`, filename))
+		// Keep an ASCII fallback for older clients and provide the real UTF-8
+		// name through RFC 5987. Browsers otherwise commonly decode Chinese
+		// names using the platform legacy code page.
+		w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"; filename*=UTF-8''%s`, sanitizeFilename(filename), url.PathEscape(filename)))
 	}
 
 	switch format {
@@ -3535,14 +3571,32 @@ func (s *Server) settingsPayload() map[string]any {
 		"web_search_backend":                   backend,
 		"brave_key_set":                        strings.TrimSpace(braveKey) != "",
 		"tavily_key_set":                       strings.TrimSpace(tavilyKey) != "",
-		"web_search_proxy":                     proxy,             // 独立出口代理(http/https/socks5)，空=直连
+		"web_search_proxy":                     maskSecrets(proxy),             // 独立出口代理(http/https/socks5)，空=直连
-		"global_proxy":                         s.m.GlobalProxy(), // 全局出口代理(http/https/socks5)，所有目标流量走它，空=直连
+		"global_proxy":                         maskSecrets(s.m.GlobalProxy()), // 全局出口代理(http/https/socks5)，所有目标流量走它，空=直连
-		"show_egress_ip":                       s.m.pg.GetBool(settingShowEgressIP, false),
+		"proxy_pool":                           func() []ProxyPoolEntry { e, _, _ := s.m.ProxyPoolSnapshot(); return e }(),
-		"python_interpreter":                   strings.TrimSpace(pyStored), // 用户/自动设的值(空=用运行时检测)
+		"proxy_pool_active_id":                 func() string { _, a, _ := s.m.ProxyPoolSnapshot(); return a }(),
-		"workers":                              s.m.Workers(),               // 并发工作 agent 数(默认3)；对之后启动的任务生效
+		"proxy_pool_switch_log":                func() []ProxySwitchLog { _, _, l := s.m.ProxyPoolSnapshot(); return l }(),
-		"worker_pools":                         s.m.WorkerPools(),           // 非空时按池顺序绑定 worker 会话
+		"proxy_rotate_timed_enabled":           s.m.pg.GetBool(settingProxyRotateTimed, false),
-		"task_concurrency_enabled":             concOn,                      // 任务并发上限开关(默认关)
+		"proxy_rotate_interval_seconds":        proxyIntSetting(s.m.pg, settingProxyRotateInterval, defaultProxyRotateInterval),
-		"task_concurrency_limit":               concLimit,                   // 同时运行任务上限(开启后默认5)
+		"proxy_rotate_on_failure":              s.m.pg.GetBool(settingProxyRotateFailure, false),
+		"proxy_rotate_strategy": func() string {
+			v, ok, _ := s.m.pg.GetSetting(settingProxyRotateStrategy)
+			if ok && (v == "random" || v == "sequential") {
+				return v
+			}
+			return "sequential"
+		}(),
+		"proxy_health_fail_threshold":         proxyIntSetting(s.m.pg, settingProxyHealthThreshold, defaultProxyHealthThreshold),
+		"proxy_health_cooldown_seconds":       proxyIntSetting(s.m.pg, settingProxyHealthCooldown, defaultProxyHealthCooldown),
+		"proxy_health_check_interval_seconds": proxyIntSetting(s.m.pg, settingProxyHealthInterval, defaultProxyHealthInterval),
+		"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)；对之后启动的任务生效
+		"worker_pools":                        s.m.WorkerPools(),           // 非空时按池顺序绑定 worker 会话
+		"task_concurrency_enabled":            concOn,                      // 任务并发上限开关(默认关)
+		"task_concurrency_limit":              concLimit,                   // 同时运行任务上限(开启后默认5)
 		// LLM 轮询(故障转移)。默认关；开启后走全局激活配置的 agent 在当前配置不可用时
 		// 自动切到下一个配置。bind_fallback 仅在轮询开启时有意义(默认关)。
 		"llm_pool_enabled":       s.m.LLMPoolEnabled(),
@@ -3613,16 +3667,25 @@ 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"`
+		WebSearchEnabled       *bool         `json:"web_search_enabled"`
-		WebSearchBackend *string       `json:"web_search_backend"`
+		WebSearchBackend       *string       `json:"web_search_backend"`
-		BraveKey         *string       `json:"brave_search_api_key"`
+		BraveKey               *string       `json:"brave_search_api_key"`
-		TavilyKey        *string       `json:"tavily_search_api_key"`
+		TavilyKey              *string       `json:"tavily_search_api_key"`
-		WebSearchProxy   *string       `json:"web_search_proxy"`   // 独立出口代理(http/https/socks5)；null=不改，""=清空
+		WebSearchProxy         *string       `json:"web_search_proxy"` // 独立出口代理(http/https/socks5)；null=不改，""=清空
-		GlobalProxy      *string       `json:"global_proxy"`       // 全局出口代理(http/https/socks5)；null=不改，""=清空(直连)
+		GlobalProxy            *string       `json:"global_proxy"`     // 全局出口代理(http/https/socks5)；null=不改，""=清空(直连)
-		ShowEgressIP     *bool         `json:"show_egress_ip"`     // 在顶栏显示直连/全局代理出口 IP；默认关
+		ProxyRotateTimed       *bool         `json:"proxy_rotate_timed_enabled"`
-		PythonInterp     *string       `json:"python_interpreter"` // 自定义脚本工具的 python 解释器路径
+		ProxyRotateInterval    *int          `json:"proxy_rotate_interval_seconds"`
-		Workers          *int          `json:"workers"`            // 并发工作 agent 数(>0)；对之后启动的任务生效
+		ProxyRotateFailure     *bool         `json:"proxy_rotate_on_failure"`
-		WorkerPools      *[]WorkerPool `json:"worker_pools"`
+		ProxyRotateStrategy    *string       `json:"proxy_rotate_strategy"`
+		ProxyHealthThreshold   *int          `json:"proxy_health_fail_threshold"`
+		ProxyHealthCooldown    *int          `json:"proxy_health_cooldown_seconds"`
+		ProxyHealthInterval    *int          `json:"proxy_health_check_interval_seconds"`
+		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"`
@@ -3833,12 +3896,61 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
 		changed = true
 	}
 	if req.GlobalProxy != nil {
+		globalProxyValue := strings.TrimSpace(*req.GlobalProxy)
+		if strings.Contains(globalProxyValue, "***") || strings.Contains(globalProxyValue, "******") {
+			globalProxyValue = s.m.GlobalProxy()
+		}
+		var proxyGuards map[string]proxyReconnectGuard
+		if globalProxyValue == "" {
+			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 {
+		if err := s.m.SetGlobalProxy(globalProxyValue); err != nil {
 			writeErr(w, 400, err.Error())
 			return
 		}
-		changed = true // capture-off egress is baked into agents at build time → rebuild
+		if globalProxyValue == "" && (s.m.proxyPool == nil || len(s.m.proxyPoolEntries()) == 0) {
+			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.ProxyRotateTimed != nil || req.ProxyRotateInterval != nil || req.ProxyRotateFailure != nil || req.ProxyRotateStrategy != nil || req.ProxyHealthThreshold != nil || req.ProxyHealthCooldown != nil || req.ProxyHealthInterval != nil {
+		timed, interval, onFailure, strategy := s.m.proxyRotateConfig()
+		threshold, cooldown, healthInterval := s.m.proxyHealthConfig()
+		if req.ProxyRotateTimed != nil {
+			timed = *req.ProxyRotateTimed
+		}
+		if req.ProxyRotateInterval != nil {
+			interval = time.Duration(*req.ProxyRotateInterval) * time.Second
+		}
+		if req.ProxyRotateFailure != nil {
+			onFailure = *req.ProxyRotateFailure
+		}
+		if req.ProxyRotateStrategy != nil {
+			strategy = strings.TrimSpace(*req.ProxyRotateStrategy)
+		}
+		if req.ProxyHealthThreshold != nil {
+			threshold = *req.ProxyHealthThreshold
+		}
+		if req.ProxyHealthCooldown != nil {
+			cooldown = time.Duration(*req.ProxyHealthCooldown) * time.Second
+		}
+		if req.ProxyHealthInterval != nil {
+			healthInterval = time.Duration(*req.ProxyHealthInterval) * time.Second
+		}
+		if err := s.m.SetProxyPoolSettings(timed, int(interval/time.Second), onFailure, strategy, threshold, int(cooldown/time.Second), int(healthInterval/time.Second)); err != nil {
+			writeErr(w, 400, err.Error())
+			return
+		}
+		select {
+		case s.proxyPoolWake <- struct{}{}:
+		default:
+		}
 	}
 	if req.ShowEgressIP != nil {
 		if err := s.m.pg.SetBool(settingShowEgressIP, *req.ShowEgressIP); err != nil {
@@ -3855,6 +3967,10 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
 		if req.WebSearchBackend != nil {
 			backend = *req.WebSearchBackend
 		}
+		if req.WebSearchProxy != nil && (strings.Contains(*req.WebSearchProxy, "***") || strings.Contains(*req.WebSearchProxy, "******")) {
+			_, _, _, _, savedSearchProxy := s.m.WebSearch()
+			req.WebSearchProxy = &savedSearchProxy
+		}
 		if err := s.m.SetWebSearch(on, backend, req.BraveKey, req.TavilyKey, req.WebSearchProxy); err != nil {
 			writeErr(w, 500, err.Error())
 			return

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,

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 {
+		reason := "paused_by_user"
-			if !wasEnginePaused && !wasQueued {
+		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

server/task_llm.go

@@ -592,6 +592,7 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle {
 	wk.SetNoaEnabled(s.m.NoaCompactionEnabled)     // 实验功能:noa 上下文压缩(平台级开关,每 run 读)
 	wk.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second)
 	wk.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
+	wk.SetProxyProvider(s.m.ProxyAddr)
 	wk.SetWebSearch(s.webSearchFor("worker"))
 	wk.SetConstraintInject(s.constraintInjectWorker) // 操作约束注入 worker(可配置,默认开)
 	pl := agent.NewPlanner(plannerRuntime, "task-router", s.m.dir, tx, plannerRuntime.CompactionWindow(), s.agentMaxTurns("planner"))
@@ -603,6 +604,7 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle {
 	pl.SetKillWork(s.engine.KillWork)
 	pl.SetSteerWork(s.engine.SteerWork)
 	pl.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
+	pl.SetProxyProvider(s.m.ProxyAddr)
 	pl.SetWebSearch(s.webSearchFor("planner"))
 	pl.SetConstraintInject(s.constraintInjectPlanner) // 操作约束注入 planner(可配置,默认开)
 	// cold-digest §7: 冷节点后台压缩。引擎经权威解析器实际驱动的就是这套 per-task planner
@@ -616,6 +618,7 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle {
 	main.SetMaxTokens(mainRuntime.maxTokens)
 	main.SetNoaEnabled(s.m.NoaCompactionEnabled) // 实验功能:noa 上下文压缩(平台级开关,每 run 读)
 	main.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
+	main.SetProxyProvider(s.m.ProxyAddr)
 	main.SetWebSearch(s.webSearchFor("mainagent"))
 	main.SetSteerWork(s.engine.SteerWork) // steer_work：人对运行中 work 实时纠偏
 	bundle := &taskAgentBundle{
@@ -662,6 +665,7 @@ func (s *Server) workerForTaskSlot(t *Task, slot WorkerSlot) (*agent.Worker, err
 	worker.SetNoaEnabled(s.m.NoaCompactionEnabled)
 	worker.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second)
 	worker.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
+	worker.SetProxyProvider(s.m.ProxyAddr)
 	worker.SetWebSearch(s.webSearchFor("worker"))
 	worker.SetConstraintInject(s.constraintInjectWorker)
 	bundle.pinnedWorkers[id] = worker

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

web/src/app/(main)/_components/egress-ip-banner.tsx

@@ -61,6 +61,14 @@ export function EgressIpProvider({ children }: { children: React.ReactNode }) {
     };
   }, [probeForProxy]);
 
+  React.useEffect(() => {
+    if (!enabled) return;
+    const timer = window.setInterval(() => {
+      void api.settings().then((settings) => probeForProxy(settings.global_proxy ?? "")).catch(() => undefined);
+    }, 10000);
+    return () => window.clearInterval(timer);
+  }, [enabled, probeForProxy]);
+
   const setEnabled = React.useCallback(
     async (nextEnabled: boolean) => {
       const settings = await api.setSettings({ show_egress_ip: nextEnabled });

web/src/app/(main)/function/tasks/detail/page.tsx

@@ -419,6 +419,14 @@ function TaskDetailInner() {
           </Button>
         </div>
         <p className="truncate text-xs text-muted-foreground">{task.goal}</p>
+        {task.paused && task.pause_reason && (
+          <div className="flex flex-wrap items-center gap-2 text-xs text-amber-600">
+            <span>暂停原因：{task.pause_reason === "paused_by_proxy" ? "全局代理异常" : task.pause_reason}</span>
+            {task.reconnect && <span>重连 {task.reconnect.attempt}/{task.reconnect.max} {task.reconnect.exhausted ? "已耗尽" : task.reconnect.next_at ? `下次 ${new Date(task.reconnect.next_at * 1000).toLocaleTimeString()}` : "探测中"}</span>}
+            {task.reconnect?.last_error && <span className="max-w-full truncate" title={task.reconnect.last_error}>原因：{task.reconnect.last_error}</span>}
+            {task.pause_reason === "paused_by_proxy" && <Button size="sm" variant="outline" onClick={() => api.testTaskProxy(task.id).then((r) => r.ok ? (toast.success("代理已恢复，任务将自动继续"), load()) : toast.error(r.error || "代理探测失败")).catch((e) => toast.error((e as Error).message))}>测试代理</Button>}
+          </div>
+        )}
         {task.remark ? (
           <p className="truncate text-xs text-muted-foreground" title={task.remark}>
             备注：{task.remark}

web/src/app/(main)/system/settings/page.tsx

@@ -18,12 +18,13 @@ import { Button } from "@/components/ui/button";
 import { Card, CardContent, CardDescription, CardHeader, CardTitle } from "@/components/ui/card";
 import { Field, FieldContent, FieldDescription, FieldLabel } from "@/components/ui/field";
 import { Input } from "@/components/ui/input";
+import { Textarea } from "@/components/ui/textarea";
 import { Label } from "@/components/ui/label";
 import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select";
 import { Switch } from "@/components/ui/switch";
 import { api } from "@/lib/api";
 import { CHAT_SEND_MODE_OPTIONS, type ChatSendMode, setChatSendMode, useChatSendMode } from "@/lib/chat-send-mode";
-import type { EgressReport, LLMProfile, Settings, WorkerPool } from "@/lib/types";
+import type { EgressReport, LLMProfile, ProxyPoolEntry, Settings, WorkerPool } from "@/lib/types";
 
 import { useEgressIp } from "../../_components/egress-ip-banner";
 import { BasicAuthCard } from "./_components/basic-auth-card";
@@ -55,6 +56,19 @@ export default function SystemSettingsPage() {
   const [savingProxy, setSavingProxy] = React.useState(false);
   const [globalProxyInput, setGlobalProxyInput] = React.useState("");
   const [savingGlobalProxy, setSavingGlobalProxy] = React.useState(false);
+  const [proxyPool, setProxyPool] = React.useState<ProxyPoolEntry[]>([]);
+  const [proxyPoolActiveId, setProxyPoolActiveId] = React.useState("");
+  const [proxyPoolText, setProxyPoolText] = React.useState("");
+  const [proxyPoolTimed, setProxyPoolTimed] = React.useState(false);
+  const [proxyPoolInterval, setProxyPoolInterval] = React.useState("30");
+  const [proxyPoolFailure, setProxyPoolFailure] = React.useState(false);
+  const [proxyPoolStrategy, setProxyPoolStrategy] = React.useState<"sequential" | "random">("sequential");
+  const [proxyPoolThreshold, setProxyPoolThreshold] = React.useState("3");
+  const [proxyPoolCooldown, setProxyPoolCooldown] = React.useState("60");
+  const [proxyPoolHealthInterval, setProxyPoolHealthInterval] = React.useState("30");
+  const [proxyPoolLogs, setProxyPoolLogs] = React.useState<Array<{ at: string; from: string; to: string; reason: string }>>([]);
+  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 +125,18 @@ export default function SystemSettingsPage() {
     setProxyInput(s.web_search_proxy ?? "");
     const savedProxy = s.global_proxy ?? "";
     setGlobalProxyInput(savedProxy);
+    setProxyPool(s.proxy_pool ?? []);
+    setProxyPoolActiveId(s.proxy_pool_active_id ?? "");
+    setProxyPoolTimed(!!s.proxy_rotate_timed_enabled);
+    setProxyPoolInterval(String(s.proxy_rotate_interval_seconds ?? 30));
+    setProxyPoolFailure(!!s.proxy_rotate_on_failure);
+    setProxyPoolStrategy(s.proxy_rotate_strategy === "random" ? "random" : "sequential");
+    setProxyPoolThreshold(String(s.proxy_health_fail_threshold ?? 3));
+    setProxyPoolCooldown(String(s.proxy_health_cooldown_seconds ?? 60));
+    setProxyPoolHealthInterval(String(s.proxy_health_check_interval_seconds ?? 30));
+    setProxyPoolLogs(s.proxy_pool_switch_log ?? []);
+    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 ?? []);
@@ -201,6 +227,11 @@ export default function SystemSettingsPage() {
       .then((s) => {
         apply(s);
         setSettingsLoaded(true);
+        void api.proxyPool().then((pool) => {
+          setProxyPool(pool.entries);
+          setProxyPoolActiveId(pool.active_id);
+          setProxyPoolLogs(pool.switch_log);
+        }).catch(() => undefined);
       })
       .catch(() => {
         setSettingsLoaded(false);
@@ -414,6 +445,10 @@ export default function SystemSettingsPage() {
   };
 
   const saveGlobalProxy = () => {
+    if (proxyPool.length > 0) {
+      toast.info("代理池非空时，全局代理输入框仅用于回显，请在代理池中编辑或切换");
+      return;
+    }
     setSavingGlobalProxy(true);
     api
       .setSettings({ global_proxy: globalProxyInput.trim() })
@@ -428,6 +463,29 @@ export default function SystemSettingsPage() {
       .finally(() => setSavingGlobalProxy(false));
   };
 
+  const importProxyPool = () => {
+    api.importProxyPool({ text: proxyPoolText }).then((r) => { setProxyPool(r.entries); setProxyPoolActiveId(r.active_id); setProxyPoolLogs(r.switch_log); setProxyPoolText(""); if (r.errors.length) toast.error(r.errors.map((e) => `第${e.line}行：${e.error}`).join("；")); else toast.success("代理池已导入"); }).catch((e) => toast.error("导入失败：" + (e as Error).message));
+  };
+  const saveProxyPoolStrategy = () => {
+    const interval = Number(proxyPoolInterval);
+    if (!Number.isFinite(interval) || !Number.isInteger(interval) || interval < 3) {
+      toast.error("定时轮换间隔必须是大于等于 3 秒的整数");
+      return;
+    }
+    api.setSettings({ proxy_rotate_timed_enabled: proxyPoolTimed, proxy_rotate_interval_seconds: Number(proxyPoolInterval), proxy_rotate_on_failure: proxyPoolFailure, proxy_rotate_strategy: proxyPoolStrategy, proxy_health_fail_threshold: Number(proxyPoolThreshold), proxy_health_cooldown_seconds: Number(proxyPoolCooldown), proxy_health_check_interval_seconds: Number(proxyPoolHealthInterval) }).then((s) => { apply(s); toast.success("代理池轮换策略已保存"); }).catch((e) => toast.error("保存失败：" + (e as Error).message));
+  };
+  const testAllProxyPool = () => api.testAllProxyPool().then((r) => { setProxyPool(r.entries); setProxyPoolActiveId(r.active_id); setProxyPoolLogs(r.switch_log); toast.success("已完成全部代理探测"); }).catch((e) => toast.error("探测失败：" + (e as Error).message));
+  const testOneProxyPool = (id: string) => api.testProxyPoolEntry(id).then(() => api.proxyPool()).then((r) => { setProxyPool(r.entries); setProxyPoolActiveId(r.active_id); setProxyPoolLogs(r.switch_log); }).catch((e) => toast.error("探测失败：" + (e as Error).message));
+  const refreshProxyPool = () => api.proxyPool().then((r) => { setProxyPool(r.entries); setProxyPoolActiveId(r.active_id); setProxyPoolLogs(r.switch_log); });
+  const toggleProxyPoolEntry = (entry: ProxyPoolEntry) => api.updateProxyPool(entry.id, { enabled: !entry.enabled }).then(() => refreshProxyPool()).catch((e) => toast.error("更新失败：" + (e as Error).message));
+  const deleteProxyPoolEntry = (id: string) => api.deleteProxyPool(id).then(() => refreshProxyPool()).catch((e) => toast.error("删除失败：" + (e as Error).message));
+
+  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;
@@ -436,6 +494,7 @@ export default function SystemSettingsPage() {
     api
       .testEgressProxy({
         global_proxy: globalProxyInput.trim(),
+        resume: true,
 
         ...(egressProbeUrl.trim() ? { probe_url: egressProbeUrl.trim() } : {}),
       })
@@ -751,12 +810,18 @@ export default function SystemSettingsPage() {
                   </Button>
                   {/* 测表单里当前的值，不要求先保存。 */}
                   <Button type="button" variant="outline" onClick={testEgress} disabled={!loaded || egressTesting}>
-                    {egressTesting ? "测试中…" : "测试"}
+                    {egressTesting ? "测试中…" : "测试代理"}
                   </Button>
                 </div>
                 <p className="text-muted-foreground text-xs">
                   {globalProxyInput.trim() ? "已配置 · 所有目标流量经此代理出网" : "未配置 · 目标流量直连出网"}
                 </p>
+                <div className="grid grid-cols-1 gap-2 sm:grid-cols-3 items-end">
+                  <div><Label htmlFor="proxy-reconnect-max">最大重连次数</Label><Input id="proxy-reconnect-max" type="number" min={1} value={proxyReconnectMax} onChange={(e) => setProxyReconnectMax(e.target.value)} /></div>
+                  <div><Label htmlFor="proxy-reconnect-interval">重连间隔（秒）</Label><Input id="proxy-reconnect-interval" type="number" min={1} value={proxyReconnectInterval} onChange={(e) => setProxyReconnectInterval(e.target.value)} /></div>
+                  <Button type="button" variant="outline" onClick={saveProxyReconnect}>保存重连策略</Button>
+                </div>
+                <p className="text-muted-foreground text-xs">代理异常暂停的任务会按此策略探测；次数耗尽后保持暂停，可手动恢复。</p>
                 <div className="flex items-center justify-between gap-4 rounded-md border px-3 py-2">
                   <div className="min-w-0">
                     <Label htmlFor="show-egress-ip" className="text-sm font-normal">
@@ -863,6 +928,40 @@ export default function SystemSettingsPage() {
                     ))}
                   </div>
                 )}
+                <div className="mt-3 rounded-md border p-3">
+                  <div className="flex flex-wrap items-center justify-between gap-2">
+                    <div>
+                      <p className="text-sm font-medium">代理池</p>
+                      <p className="text-muted-foreground text-xs">当前在用：{proxyPool.find((entry) => entry.id === proxyPoolActiveId)?.name ?? (proxyPoolActiveId ? "已选择代理" : "直连")} · 每行一个代理；可用 URL|名称|备注。账号密码只会以掩码返回。</p>
+                    </div>
+                    <Button type="button" variant="outline" size="sm" onClick={testAllProxyPool}>测试全部</Button>
+                  </div>
+                  <Textarea className="mt-2 font-mono text-xs" rows={3} placeholder="http://user:pass@host:8080|香港 1|备用" value={proxyPoolText} onChange={(e) => setProxyPoolText(e.target.value)} />
+                  <div className="mt-2 flex flex-wrap gap-2"><Button type="button" size="sm" onClick={importProxyPool} disabled={!proxyPoolText.trim()}>批量导入</Button></div>
+                  <div className="mt-3 space-y-2">
+                    {proxyPool.length === 0 ? <p className="text-muted-foreground text-xs">暂无代理池条目；单代理仍由上面的全局代理输入框兼容。</p> : proxyPool.map((entry) => (
+                      <div key={entry.id} className="flex flex-wrap items-center gap-2 rounded border px-2 py-2 text-xs">
+                        <Switch checked={entry.enabled} onCheckedChange={() => toggleProxyPoolEntry(entry)} aria-label="启用代理" />
+                        <span className="min-w-0 flex-1 font-mono">{entry.name ? `${entry.name} · ` : ""}{entry.url}{entry.id === proxyPoolActiveId ? " · 当前" : ""}</span>
+                        <span className={entry.health === "healthy" ? "text-emerald-600" : entry.health === "disabled" ? "text-muted-foreground" : "text-amber-600"}>{entry.enabled ? entry.health : "disabled"}</span>
+                        <span className="text-muted-foreground">{entry.egress_ip ?? "—"} · {entry.latency_ms ?? "—"}ms · 失败 {entry.consecutive_failures}{entry.last_error ? ` · ${entry.last_error}` : ""}</span>
+                        <Button type="button" variant="ghost" size="sm" onClick={() => testOneProxyPool(entry.id)}>测试</Button>
+                        <Button type="button" variant="ghost" size="sm" onClick={() => deleteProxyPoolEntry(entry.id)}>删除</Button>
+                      </div>
+                    ))}
+                  </div>
+                  <div className="mt-3 grid grid-cols-1 gap-2 sm:grid-cols-3">
+                    <div className="flex items-center justify-between gap-2 rounded border px-2 py-2"><Label htmlFor="proxy-pool-timed">定时轮换</Label><Switch id="proxy-pool-timed" checked={proxyPoolTimed} onCheckedChange={setProxyPoolTimed} /></div>
+                    <div><Label htmlFor="proxy-pool-interval">间隔（秒，≥3）</Label><Input id="proxy-pool-interval" type="number" min={3} value={proxyPoolInterval} onChange={(e) => setProxyPoolInterval(e.target.value)} /></div>
+                    <div className="flex items-center justify-between gap-2 rounded border px-2 py-2"><Label htmlFor="proxy-pool-failure">失效轮换</Label><Switch id="proxy-pool-failure" checked={proxyPoolFailure} onCheckedChange={setProxyPoolFailure} /></div>
+                    <div><Label>策略</Label><Select value={proxyPoolStrategy} onValueChange={(v) => setProxyPoolStrategy(v as "sequential" | "random")}><SelectTrigger><SelectValue /></SelectTrigger><SelectContent><SelectItem value="sequential">顺序</SelectItem><SelectItem value="random">随机</SelectItem></SelectContent></Select></div>
+                    <div><Label htmlFor="proxy-pool-threshold">失败阈值</Label><Input id="proxy-pool-threshold" type="number" min={1} value={proxyPoolThreshold} onChange={(e) => setProxyPoolThreshold(e.target.value)} /></div>
+                    <div><Label htmlFor="proxy-pool-cooldown">冷却（秒）</Label><Input id="proxy-pool-cooldown" type="number" min={1} value={proxyPoolCooldown} onChange={(e) => setProxyPoolCooldown(e.target.value)} /></div>
+                    <div><Label htmlFor="proxy-pool-health-interval">健康检查（秒）</Label><Input id="proxy-pool-health-interval" type="number" min={1} value={proxyPoolHealthInterval} onChange={(e) => setProxyPoolHealthInterval(e.target.value)} /></div>
+                    <Button type="button" variant="outline" onClick={saveProxyPoolStrategy}>保存轮换策略</Button>
+                  </div>
+                  {proxyPoolLogs.length > 0 && <div className="mt-3 text-xs"><p className="font-medium">最近切换</p>{proxyPoolLogs.slice(-5).reverse().map((log) => <p key={`${log.at}-${log.to}`} className="text-muted-foreground">{new Date(log.at).toLocaleTimeString()} · {log.from || "直连"} → {log.to || "直连"} · {log.reason}</p>)}</div>}
+                </div>
               </CardContent>
             </Card>
 

web/src/lib/api.ts

@@ -34,6 +34,8 @@ import type {
   DeleteTaskResult,
   Edge,
   EgressReport,
+  ProxyPoolEntry,
+  ProxySwitchLog,
   EvidenceBodyPreview,
   ExplorationNodePage,
   ExplorationNodeQuery,
@@ -119,6 +121,24 @@ import type {
   SystemMetrics,
 } from "@/lib/types";
 
+/**
+ * Read a download name from Content-Disposition. RFC 5987's filename* is
+ * preferred because it preserves UTF-8 names; older filename remains the
+ * fallback for clients that do not understand it.
+ */
+export function downloadFilename(contentDisposition: string, fallback: string): string {
+  const encoded = contentDisposition.match(/(?:^|;)\s*filename\*\s*=\s*UTF-8''([^;]+)/i)?.[1];
+  if (encoded) {
+    try {
+      return decodeURIComponent(encoded.replace(/^"|"$/g, ""));
+    } catch {
+      // Fall through to the legacy token when a malformed header is received.
+    }
+  }
+  const legacy = contentDisposition.match(/(?:^|;)\s*filename\s*=\s*(?:"([^"]*)"|([^;\s]+))/i);
+  return (legacy?.[1] ?? legacy?.[2] ?? fallback).trim() || fallback;
+}
+
 function getToken(): string | null {
   if (typeof window === "undefined") return null;
   return localStorage.getItem("artex_token");
@@ -534,6 +554,7 @@ export const api = {
   },
   workspaceDownload: async (path: string) => {
     let blob: Blob;
+    let contentDisposition = "";
     if (MOCK) {
       blob = new Blob([`（demo）${path} 的下载内容示例。`], { type: "text/plain" });
     } else {
@@ -542,12 +563,13 @@ export const api = {
         headers: { ...(token ? { Authorization: `Bearer ${token}` } : {}) },
       });
       if (!r.ok) throw new Error(`下载失败: ${r.status}`);
+      contentDisposition = r.headers.get("Content-Disposition") ?? "";
       blob = await r.blob();
     }
     const objUrl = URL.createObjectURL(blob);
     const a = document.createElement("a");
     a.href = objUrl;
-    a.download = path.split("/").pop() || "download";
+    a.download = downloadFilename(contentDisposition, path.split("/").pop() || "download");
     document.body.appendChild(a);
     a.click();
     a.remove();
@@ -703,8 +725,7 @@ export const api = {
     const blob = await r.blob();
     // 文件名优先取后端 Content-Disposition,取不到则用默认名。
     const disp = r.headers.get("Content-Disposition") ?? "";
-    const m = disp.match(/filename="?([^"]+)"?/);
+    const filename = downloadFilename(disp, `findings-export`);
-    const filename = m?.[1] ?? `findings-export`;
     const url = URL.createObjectURL(blob);
     const a = document.createElement("a");
     a.href = url;
@@ -971,8 +992,16 @@ export const api = {
   }) => post<{ ok: boolean; error?: string; count?: number; backend?: string }>(`/settings/web-search/test`, patch),
 
   // global_proxy 传空 = 测直连；probe_url 可覆盖默认探测地址（隔离网用）。
-  testEgressProxy: (patch: { global_proxy?: string; probe_url?: string }) =>
+  testEgressProxy: (patch: { global_proxy?: string; probe_url?: string; resume?: boolean }) =>
     post<EgressReport>(`/settings/proxy/test`, patch),
+  proxyPool: () => get<{ entries: ProxyPoolEntry[]; active_id: string; active_proxy: string; switch_log: ProxySwitchLog[] }>(`/settings/proxy-pool`),
+  importProxyPool: (payload: Record<string, unknown>) => post<{ entries: ProxyPoolEntry[]; active_id: string; switch_log: ProxySwitchLog[]; errors: Array<{ line: number; input: string; error: string }> }>(`/settings/proxy-pool/import`, payload),
+  updateProxyPool: (id: string, payload: Partial<Pick<ProxyPoolEntry, "url" | "name" | "remark" | "enabled">>) => put<{ entry: ProxyPoolEntry }>(`/settings/proxy-pool/${encodeURIComponent(id)}`, payload),
+  deleteProxyPool: (id: string) => del<{ ok: boolean }>(`/settings/proxy-pool/${encodeURIComponent(id)}`),
+  testProxyPoolEntry: (id: string) => post<{ ok: boolean; result?: EgressReport["checks"][number]["result"] }>(`/settings/proxy-pool/${encodeURIComponent(id)}/test`, {}),
+  testAllProxyPool: () => post<{ entries: ProxyPoolEntry[]; active_id: string; switch_log: ProxySwitchLog[] }>(`/settings/proxy-pool/test-all`, {}),
+  activateProxyPool: (id: string) => post<{ ok: boolean; active_id: string; active_proxy: string }>(`/settings/proxy-pool/${encodeURIComponent(id)}/activate`, {}),
+  testTaskProxy: (taskID: string) => post<{ ok: boolean; error?: string }>(`/tasks/${taskID}/proxy/test`, {}),
 
   // ---- 漏洞 IM 推送 ----
   // 渠道是多实例资源（同一类型可配多个机器人、各有过滤规则），因此独立成组，

web/src/lib/mock/data.ts

@@ -3350,6 +3350,15 @@ export const settings: Settings = {
   tavily_key_set: true,
   web_search_proxy: "",
   global_proxy: "",
+	proxy_pool: [],
+	proxy_pool_active_id: "",
+	proxy_rotate_timed_enabled: false,
+	proxy_rotate_interval_seconds: 30,
+	proxy_rotate_on_failure: false,
+	proxy_rotate_strategy: "sequential",
+	proxy_health_fail_threshold: 3,
+	proxy_health_cooldown_seconds: 60,
+	proxy_health_check_interval_seconds: 30,
 	show_egress_ip: false,
   python_interpreter: "/usr/bin/python3",
   workers: 3,

web/src/lib/mock/handler.ts

@@ -2406,6 +2406,12 @@ function route(m: string, path: string, seg: string[], q: URLSearchParams, b: Re
       ],
       warnings: [],
     };
+  if (path === "/settings/proxy-pool" && m === "GET") return { entries: D.settings.proxy_pool ?? [], active_id: D.settings.proxy_pool_active_id ?? "", active_proxy: D.settings.global_proxy ?? "", switch_log: [] };
+  if (path === "/settings/proxy-pool/import" && m === "POST") return { entries: D.settings.proxy_pool ?? [], active_id: D.settings.proxy_pool_active_id ?? "", switch_log: [], errors: [] };
+  if (path === "/settings/proxy-pool/test-all" && m === "POST") return { entries: D.settings.proxy_pool ?? [], active_id: D.settings.proxy_pool_active_id ?? "", switch_log: [] };
+  if (path.startsWith("/settings/proxy-pool/") && m === "PUT") return { entry: { ...(D.settings.proxy_pool ?? []).find((x) => x.id === path.split("/").at(-1)), ...b } };
+  if (path.startsWith("/settings/proxy-pool/") && m === "DELETE") return { ok: true };
+  if (path.startsWith("/settings/proxy-pool/") && m === "POST") return { ok: true, active_id: path.split("/").at(-1), active_proxy: "" };
   if (path === "/settings/python/detect") return { python_interpreter: "/usr/bin/python3" };
   if (path === "/chat")
     return { reply: "（demo）我已把该建议注入为一条高优意图，work agent 会尽快执行。", mode: "hint" };

web/src/lib/types.ts

@@ -42,6 +42,8 @@ export interface Task {
   active_run_seconds?: number; // completed running segments in seconds
   run_segment_started_unix?: number; // current running segment start; 0 when inactive
   paused?: boolean;
+  pause_reason?: string;
+  reconnect?: { attempt: number; max: number; next_at?: number; exhausted: boolean; last_error?: string };
   queued?: boolean;
   active?: boolean;
   in_flight?: number;
@@ -1002,6 +1004,18 @@ export interface Settings {
   // 全局出口代理(http/https/socks5，可带 user:pass)，所有目标流量走它。开启流量捕获时作为
   // MITM 上游；关闭捕获时直接注入 agent 的 bash/WebFetch。空=直连。
   global_proxy?: string;
+  proxy_pool?: ProxyPoolEntry[];
+  proxy_pool_active_id?: string;
+  proxy_pool_switch_log?: ProxySwitchLog[];
+  proxy_rotate_timed_enabled?: boolean;
+  proxy_rotate_interval_seconds?: number;
+  proxy_rotate_on_failure?: boolean;
+  proxy_rotate_strategy?: "sequential" | "random";
+  proxy_health_fail_threshold?: number;
+  proxy_health_cooldown_seconds?: number;
+  proxy_health_check_interval_seconds?: number;
+  proxy_reconnect_max_attempts?: number;
+  proxy_reconnect_interval_seconds?: number;
   show_egress_ip?: boolean; // 在顶栏显示直连 / 全局代理出口 IP；默认关闭
   python_interpreter?: string; // 自定义脚本工具的 python 解释器路径(空=运行时检测)
   workers?: number; // 并发工作 agent 数(默认3)；对之后启动的任务生效
@@ -1027,6 +1041,22 @@ export interface Settings {
   notify_digest_interval_min?: number; // 汇总模式周期（分钟），默认 30
 }
 
+export interface ProxyPoolEntry {
+  id: string;
+  url: string;
+  name?: string;
+  remark?: string;
+  enabled: boolean;
+  egress_ip?: string;
+  latency_ms?: number;
+  last_error?: string;
+  last_checked_at?: string;
+  consecutive_failures: number;
+  health: "healthy" | "unhealthy-cooling" | "disabled" | string;
+}
+
+export interface ProxySwitchLog { at: string; from: string; to: string; reason: string }
+
 export interface WorkerPool {
   profile_id: string;
   workers: number;