fix: 代理暂停归入任务暂停语义并为所有成功路径加轮次门控

server/egress_probe.go

@@ -373,6 +373,7 @@ func (s *Server) testEgressProxy(w http.ResponseWriter, r *http.Request) {
 	// 整体上限 40s：单次 Timeout 只兜一次，不兜总量；前端 fetch 没有超时。
 	ctx, cancel := context.WithTimeout(r.Context(), 40*time.Second)
 	defer cancel()
+	proxyGuards := s.snapshotProxyReconnectGuards(false)
 
 	configuredProxy := strings.TrimSpace(req.GlobalProxy)
 	rep := egressReport{
@@ -421,7 +422,7 @@ func (s *Server) testEgressProxy(w http.ResponseWriter, r *http.Request) {
 		// The settings page uses this endpoint for its manual "测试代理" action.
 		// A successful probe is the same signal as an automatic reconnect success,
 		// but it must never consume an automatic attempt.
-		s.proxyReconnectSuccess()
+		s.proxyReconnectSuccessWithGuards(proxyGuards, nil)
 	}
 	writeJSON(w, 200, rep)
 }

server/engine.go

@@ -1357,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/goals.go

@@ -121,13 +121,17 @@ func (s *Server) admitTaskWhenScheduled(t *Task, mode string, requirePaused, sch
 // 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.admitTaskWhenScheduledWithGeneration(t, mode, requirePaused, scheduled, requiredReason, preserveSchedule, 0)
+	return s.admitTaskWhenScheduledWithGenerationChecked(t, mode, requirePaused, scheduled, requiredReason, preserveSchedule, 0, false)
 }
 
-// admitTaskWhenScheduledWithGeneration extends the proxy recovery admission
-// gate with a reconnect-round check. A delayed probe must not admit a task from
-// a newer proxy pause round that happens to have the same attempt number.
+// 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")
 	}
@@ -174,7 +178,7 @@ func (s *Server) admitTaskWhenScheduledWithGeneration(t *Task, mode string, requ
 		if requiredReason != "" && lifecycle.PauseReason != requiredReason {
 			return false, fmt.Errorf("任务暂停原因已改变")
 		}
-		if requiredReason == "paused_by_proxy" && requiredGeneration != 0 && lifecycle.ReconnectGeneration != requiredGeneration {
+		if requiredReason == "paused_by_proxy" && checkGeneration && lifecycle.ReconnectGeneration != requiredGeneration {
 			return false, fmt.Errorf("任务代理重连轮次已改变")
 		}
 		mode = s.resumeAdmissionMode(t)

server/proxy_autoresume.go

@@ -58,6 +58,36 @@ type proxyReconnectSupervisor struct {
 	wake chan struct{}
 }
 
+type proxyReconnectGuard struct {
+	generation   uint64
+	attempt      int
+	checkAttempt bool
+}
+
+func (s *Server) snapshotProxyReconnectGuards(checkAttempt bool) map[string]proxyReconnectGuard {
+	guards := make(map[string]proxyReconnectGuard)
+	if s == nil || s.m == nil {
+		return guards
+	}
+	for _, t := range s.m.List() {
+		st := t.lifecycleSnapshot()
+		if st.PauseReason != "paused_by_proxy" {
+			continue
+		}
+		guards[t.ID] = proxyReconnectGuard{
+			generation:   st.ReconnectGeneration,
+			attempt:      st.ReconnectAttempt,
+			checkAttempt: checkAttempt,
+		}
+	}
+	return guards
+}
+
+func proxyReconnectGuardMatches(st taskLifecycleState, guard proxyReconnectGuard) bool {
+	return st.PauseReason == "paused_by_proxy" && st.ReconnectGeneration == guard.generation &&
+		(!guard.checkAttempt || st.ReconnectAttempt == guard.attempt)
+}
+
 func (s *Server) ensureProxySupervisor() {
 	s.proxyReconnect.mu.Lock()
 	if s.proxyReconnect.wake == nil {
@@ -245,7 +275,9 @@ func (s *Server) proxyReconnectTick() {
 		}
 		res := s.runProxyProbe(context.Background(), s.m.GlobalProxy(), "", defaultEgressProbes, 10*time.Second)
 		if res.OK {
-			s.proxyReconnectSuccessFor(t.ID, round, attempt)
+			s.proxyReconnectSuccessWithGuards(map[string]proxyReconnectGuard{
+				t.ID: {generation: round, attempt: attempt, checkAttempt: true},
+			}, nil)
 			return
 		}
 		err := res.Error
@@ -264,11 +296,13 @@ func (s *Server) proxyReconnectTick() {
 }
 
 func (s *Server) proxyReconnectSuccess() {
-	s.proxyReconnectSuccessWithGuard("", 0, 0, nil)
+	s.proxyReconnectSuccessWithGuards(s.snapshotProxyReconnectGuards(false), nil)
 }
 
 func (s *Server) proxyReconnectSuccessFor(taskID string, generation uint64, attempt int) {
-	s.proxyReconnectSuccessWithGuard(taskID, generation, attempt, nil)
+	s.proxyReconnectSuccessWithGuards(map[string]proxyReconnectGuard{
+		taskID: {generation: generation, attempt: attempt, checkAttempt: true},
+	}, nil)
 }
 
 // proxyReconnectSuccessWithHook is kept small and private so tests can
@@ -276,22 +310,30 @@ func (s *Server) proxyReconnectSuccessFor(taskID string, generation uint64, atte
 // released concMu. Production callers use proxyReconnectSuccess without a
 // hook.
 func (s *Server) proxyReconnectSuccessWithHook(afterAdmission func()) {
-	s.proxyReconnectSuccessWithGuard("", 0, 0, afterAdmission)
+	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() {
-		if st := t.lifecycleSnapshot(); st.PauseReason != "paused_by_proxy" {
+		guard, ok := guards[t.ID]
+		if !ok {
 			continue
 		}
-		if expectedID == t.ID {
-			st := t.lifecycleSnapshot()
-			if st.PauseReason != "paused_by_proxy" || st.ReconnectGeneration != expectedGeneration || st.ReconnectAttempt != expectedAttempt {
-				continue
-			}
+		if !proxyReconnectGuardMatches(t.lifecycleSnapshot(), guard) {
+			continue
 		}
 		if s.m.pg != nil {
 			inside, scheduleErr := s.proxyResumeInsideSchedule(t.ID, time.Now())
@@ -308,11 +350,7 @@ func (s *Server) proxyReconnectSuccessWithGuard(expectedID string, expectedGener
 		}
 		var err error
 		if s.engine != nil {
-			if expectedID == t.ID {
-				_, err = s.admitTaskWhenScheduledWithGeneration(t, "resume", true, false, "paused_by_proxy", true, expectedGeneration)
-			} else {
-				_, err = s.admitTaskWhenScheduledWithReason(t, "resume", true, false, "paused_by_proxy", true)
-			}
+			_, err = s.admitTaskWhenScheduledWithGenerationChecked(t, "resume", true, false, "paused_by_proxy", true, guard.generation, true)
 			if afterAdmission != nil {
 				afterAdmission()
 			}
@@ -324,8 +362,7 @@ func (s *Server) proxyReconnectSuccessWithGuard(expectedID string, expectedGener
 			// terminal transition, delete, or archive therefore cannot be revived;
 			// only keep probing if the proxy pause is still eligible.
 			st := t.lifecycleSnapshot()
-			if st.PauseReason == "paused_by_proxy" && !isTerminalStatus(st.Status) &&
-				(expectedID == "" || (st.ReconnectGeneration == expectedGeneration && st.ReconnectAttempt == expectedAttempt)) {
+			if proxyReconnectGuardMatches(st, guard) && !isTerminalStatus(st.Status) {
 				t.updateLifecycle(func(v *taskLifecycleState) {
 					v.ReconnectMax = max
 					v.ReconnectNextAt = time.Now().Add(interval).Unix()
@@ -339,7 +376,7 @@ func (s *Server) proxyReconnectSuccessWithGuard(expectedID string, expectedGener
 		// a concurrent user/orchestrator pause could have installed a newer
 		// barrier in the meantime.
 		t.updateLifecycle(func(v *taskLifecycleState) {
-			if expectedID == t.ID && v.ReconnectGeneration != expectedGeneration {
+			if !proxyReconnectGuardMatches(*v, guard) {
 				return
 			}
 			v.ReconnectAttempt = 0
@@ -359,14 +396,15 @@ func (s *Server) testTaskProxy(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	proxy := strings.TrimSpace(s.m.GlobalProxy())
+	guards := s.snapshotProxyReconnectGuards(false)
 	if proxy == "" {
 		writeJSON(w, 200, map[string]any{"ok": true, "direct": true})
-		s.proxyReconnectSuccess()
+		s.proxyReconnectSuccessWithGuards(guards, nil)
 		return
 	}
-	res := probeEgress(r.Context(), proxy, "", defaultEgressProbes, 10*time.Second)
+	res := s.runProxyProbe(r.Context(), proxy, "", defaultEgressProbes, 10*time.Second)
 	if res.OK {
-		s.proxyReconnectSuccess()
+		s.proxyReconnectSuccessWithGuards(guards, nil)
 	}
 	if !res.OK && res.Error == "" {
 		res.Error = "代理探测失败"

server/proxy_autoresume_test.go

@@ -3,6 +3,7 @@ package server
 import (
 	"context"
 	"errors"
+	"sync"
 	"testing"
 	"time"
 
@@ -369,6 +370,11 @@ func TestProxyReconnectDropsStaleProbeFailureAfterNewRound(t *testing.T) {
 	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)
@@ -391,8 +397,9 @@ func TestProxyReconnectDropsStaleProbeFailureAfterNewRound(t *testing.T) {
 	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 {
-		close(probeStarted)
+		probeStartedOnce.Do(func() { close(probeStarted) })
 		<-releaseProbe
 		return egressProbeResult{Error: "stale probe failure"}
 	}
@@ -425,3 +432,69 @@ func TestProxyReconnectDropsStaleProbeFailureAfterNewRound(t *testing.T) {
 		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/server.go

@@ -3851,6 +3851,10 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
 		changed = true
 	}
 	if req.GlobalProxy != nil {
+		var proxyGuards map[string]proxyReconnectGuard
+		if strings.TrimSpace(*req.GlobalProxy) == "" {
+			proxyGuards = s.snapshotProxyReconnectGuards(false)
+		}
 		// Validation failure (bad scheme/host) is a client error, not a 500.
 		if err := s.m.SetGlobalProxy(*req.GlobalProxy); err != nil {
 			writeErr(w, 400, err.Error())
@@ -3858,7 +3862,7 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
 		}
 		changed = true // capture-off egress is baked into agents at build time → rebuild
 		if strings.TrimSpace(*req.GlobalProxy) == "" {
-			s.proxyReconnectSuccess()
+			s.proxyReconnectSuccessWithGuards(proxyGuards, nil)
 		}
 	}
 	if req.ProxyReconnectMax != nil || req.ProxyReconnectInterval != nil {