fix: 新一轮代理暂停重置重连计数并丢弃过期探测写回

server/goals.go

@@ -121,6 +121,13 @@ 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)
+}
+
+// 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.
+func (s *Server) admitTaskWhenScheduledWithGeneration(t *Task, mode string, requirePaused, scheduled bool, requiredReason string, preserveSchedule bool, requiredGeneration uint64) (queued bool, err error) {
 	if t == nil {
 		return false, fmt.Errorf("task not found")
 	}
@@ -167,6 +174,9 @@ func (s *Server) admitTaskWhenScheduledWithReason(t *Task, mode string, requireP
 		if requiredReason != "" && lifecycle.PauseReason != requiredReason {
 			return false, fmt.Errorf("任务暂停原因已改变")
 		}
+		if requiredReason == "paused_by_proxy" && requiredGeneration != 0 && lifecycle.ReconnectGeneration != requiredGeneration {
+			return false, fmt.Errorf("任务代理重连轮次已改变")
+		}
 		mode = s.resumeAdmissionMode(t)
 	}
 	var manualUntil *time.Time

server/manager.go

@@ -28,25 +28,26 @@ import (
 // sharing the process-wide asset store. ID is the PG task id as a string; ExpID
 // is the exploration the task owns.
 type Task struct {
-	ID                 string `json:"id"`
-	ExpID              int64  `json:"exploration_id"`
-	Remark             string `json:"remark"`
-	Name               string `json:"name"` // 可选任务名称;空=未命名
-	CategoryID         *int64 `json:"category_id,omitempty"`
-	CategoryName       string `json:"category_name,omitempty"`
-	PinnedAt           int64  `json:"pinned_at,omitempty"`
-	Description        string `json:"description"`
-	Goal               string `json:"goal"`
-	CreatedAt          int64  `json:"created_at"`
-	CompletedAt        int64  `json:"completed_at,omitempty"` // 进入终态的 unix 秒;0=未完成
-	Paused             bool   `json:"paused"`
-	PauseReason        string `json:"pause_reason,omitempty"`
-	ReconnectAttempt   int    `json:"-"`
-	ReconnectMax       int    `json:"-"`
-	ReconnectNextAt    int64  `json:"-"`
-	ReconnectExhausted bool   `json:"-"`
-	ReconnectLastError string `json:"-"`
-	Queued             bool   `json:"queued"` // 因并发上限被挂起、等待空位自动启动;true=尚未开跑
+	ID                  string `json:"id"`
+	ExpID               int64  `json:"exploration_id"`
+	Remark              string `json:"remark"`
+	Name                string `json:"name"` // 可选任务名称;空=未命名
+	CategoryID          *int64 `json:"category_id,omitempty"`
+	CategoryName        string `json:"category_name,omitempty"`
+	PinnedAt            int64  `json:"pinned_at,omitempty"`
+	Description         string `json:"description"`
+	Goal                string `json:"goal"`
+	CreatedAt           int64  `json:"created_at"`
+	CompletedAt         int64  `json:"completed_at,omitempty"` // 进入终态的 unix 秒;0=未完成
+	Paused              bool   `json:"paused"`
+	PauseReason         string `json:"pause_reason,omitempty"`
+	ReconnectAttempt    int    `json:"-"`
+	ReconnectMax        int    `json:"-"`
+	ReconnectNextAt     int64  `json:"-"`
+	ReconnectExhausted  bool   `json:"-"`
+	ReconnectLastError  string `json:"-"`
+	ReconnectGeneration uint64 `json:"-"`
+	Queued              bool   `json:"queued"` // 因并发上限被挂起、等待空位自动启动;true=尚未开跑
 	// QueuedAt is an internal Unix-nanosecond ordering key. It is deliberately
 	// finer than CreatedAt so several tasks enqueued in the same second retain
 	// their real FIFO order.
@@ -100,6 +101,7 @@ type taskLifecycleState struct {
 	ReconnectNextAt     int64
 	ReconnectExhausted  bool
 	ReconnectLastError  string
+	ReconnectGeneration uint64
 	Queued              bool
 	QueuedAt            int64
 	QueueMode           string
@@ -131,7 +133,7 @@ func (t *Task) lifecycleSnapshotLocked() taskLifecycleState {
 		Status:           t.Status,
 		Paused:           t.Paused,
 		PauseReason:      t.PauseReason,
-		ReconnectAttempt: t.ReconnectAttempt, ReconnectMax: t.ReconnectMax, ReconnectNextAt: t.ReconnectNextAt, ReconnectExhausted: t.ReconnectExhausted, ReconnectLastError: t.ReconnectLastError,
+		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,
@@ -165,6 +167,7 @@ func (t *Task) updateLifecycle(update func(*taskLifecycleState)) {
 	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
@@ -1512,6 +1515,11 @@ func (m *Manager) applyTaskAdmissionWithSchedule(id, expectedStatus, status stri
 			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()

server/proxy_autoresume.go

@@ -94,6 +94,13 @@ func (s *Server) proxyReconnectConfig() (int, time.Duration) {
 	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
@@ -108,10 +115,26 @@ 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)
@@ -124,19 +147,6 @@ func (s *Server) pauseProxyTask(taskID string, cause error) {
 		return
 	}
 	s.cancelTaskChat(taskID, agent.AbortChatPausedWithTask)
-	max, interval := s.proxyReconnectConfig()
-	now := time.Now().Add(interval).Unix()
-	if t, ok := s.m.Task(taskID); ok {
-		t.updateLifecycle(func(st *taskLifecycleState) {
-			// Only the task that transitioned in this call starts a new
-			// reconnect round. A task that exhausted a prior round remains
-			// untouched until it is manually resumed and paused again.
-			st.ReconnectMax = max
-			st.ReconnectNextAt = now
-			st.ReconnectExhausted = false
-			st.ReconnectLastError = truncateProxyError(cause)
-		})
-	}
 	s.ensureProxySupervisor()
 	s.signalProxySupervisor()
 }
@@ -218,32 +228,47 @@ func (s *Server) proxyReconnectTick() {
 		if st.PauseReason != "paused_by_proxy" || st.ReconnectExhausted || (st.ReconnectNextAt > 0 && st.ReconnectNextAt > now) {
 			continue
 		}
-		st.ReconnectAttempt++
-		st.ReconnectMax = max
+		attempt := st.ReconnectAttempt + 1
+		round := st.ReconnectGeneration
+		claimed := false
 		t.updateLifecycle(func(v *taskLifecycleState) {
-			v.ReconnectAttempt = st.ReconnectAttempt
+			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
 		})
-		res := probeEgress(context.Background(), s.m.GlobalProxy(), "", defaultEgressProbes, 10*time.Second)
+		if !claimed {
+			continue
+		}
+		res := s.runProxyProbe(context.Background(), s.m.GlobalProxy(), "", defaultEgressProbes, 10*time.Second)
 		if res.OK {
-			s.proxyReconnectSuccess()
+			s.proxyReconnectSuccessFor(t.ID, round, attempt)
 			return
 		}
 		err := res.Error
-		exhausted := st.ReconnectAttempt >= max
+		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", st.ReconnectAttempt, max, err)
+		log.Printf("[proxy] reconnect attempt %d/%d failed: %s", attempt, max, err)
 	}
 }
 
 func (s *Server) proxyReconnectSuccess() {
-	s.proxyReconnectSuccessWithHook(nil)
+	s.proxyReconnectSuccessWithGuard("", 0, 0, nil)
+}
+
+func (s *Server) proxyReconnectSuccessFor(taskID string, generation uint64, attempt int) {
+	s.proxyReconnectSuccessWithGuard(taskID, generation, attempt, nil)
 }
 
 // proxyReconnectSuccessWithHook is kept small and private so tests can
@@ -251,6 +276,10 @@ func (s *Server) proxyReconnectSuccess() {
 // released concMu. Production callers use proxyReconnectSuccess without a
 // hook.
 func (s *Server) proxyReconnectSuccessWithHook(afterAdmission func()) {
+	s.proxyReconnectSuccessWithGuard("", 0, 0, afterAdmission)
+}
+
+func (s *Server) proxyReconnectSuccessWithGuard(expectedID string, expectedGeneration uint64, expectedAttempt int, afterAdmission func()) {
 	max, interval := s.proxyReconnectConfig()
 	s.scheduleMu.Lock()
 	defer s.scheduleMu.Unlock()
@@ -258,6 +287,12 @@ func (s *Server) proxyReconnectSuccessWithHook(afterAdmission func()) {
 		if st := t.lifecycleSnapshot(); st.PauseReason != "paused_by_proxy" {
 			continue
 		}
+		if expectedID == t.ID {
+			st := t.lifecycleSnapshot()
+			if st.PauseReason != "paused_by_proxy" || st.ReconnectGeneration != expectedGeneration || st.ReconnectAttempt != expectedAttempt {
+				continue
+			}
+		}
 		if s.m.pg != nil {
 			inside, scheduleErr := s.proxyResumeInsideSchedule(t.ID, time.Now())
 			if scheduleErr != nil {
@@ -273,7 +308,11 @@ func (s *Server) proxyReconnectSuccessWithHook(afterAdmission func()) {
 		}
 		var err error
 		if s.engine != nil {
-			_, err = s.admitTaskWhenScheduledWithReason(t, "resume", true, false, "paused_by_proxy", true)
+			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)
+			}
 			if afterAdmission != nil {
 				afterAdmission()
 			}
@@ -285,7 +324,8 @@ func (s *Server) proxyReconnectSuccessWithHook(afterAdmission func()) {
 			// 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) {
+			if st.PauseReason == "paused_by_proxy" && !isTerminalStatus(st.Status) &&
+				(expectedID == "" || (st.ReconnectGeneration == expectedGeneration && st.ReconnectAttempt == expectedAttempt)) {
 				t.updateLifecycle(func(v *taskLifecycleState) {
 					v.ReconnectMax = max
 					v.ReconnectNextAt = time.Now().Add(interval).Unix()
@@ -299,6 +339,9 @@ func (s *Server) proxyReconnectSuccessWithHook(afterAdmission func()) {
 		// 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 {
+				return
+			}
 			v.ReconnectAttempt = 0
 			v.ReconnectMax = 0
 			v.ReconnectNextAt = 0

server/proxy_autoresume_test.go

@@ -323,3 +323,105 @@ func TestProxyPauseDoesNotReviveExhaustedOrNonProxyTasks(t *testing.T) {
 		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"
+	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{})
+	s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
+		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)
+	}
+}

server/server.go

@@ -129,7 +129,10 @@ type Server struct {
 	archiveWG      sync.WaitGroup
 	credentialWake chan struct{}
 	proxyReconnect proxyReconnectSupervisor
-	side           *sideQuestionState
+	// 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.