fix: 代理暂停/恢复屏障在锁内安装并防止复活已耗尽任务

server/manager.go

@@ -1580,6 +1580,16 @@ func (m *Manager) ApplyTaskPauseOriginReason(id, reason string, scheduled bool)
 // 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()
@@ -1588,7 +1598,7 @@ func (m *Manager) ApplyProxyPause(id string) error {
 	defer m.taskStateMu.Unlock()
 	n, err := strconv.ParseInt(id, 10, 64)
 	if err != nil {
-		return err
+		return false, err
 	}
 	var mode string
 	var active int64
@@ -1598,10 +1608,10 @@ func (m *Manager) ApplyProxyPause(id string) error {
  run_segment_started_at=NULL WHERE id=$1 AND deleted_at IS NULL AND paused=false AND status NOT IN ('done','failed','timeout')
  RETURNING COALESCE(queue_mode,''), active_run_seconds, run_segment_started_at`, n).Scan(&mode, &active, &started)
 	if errors.Is(err, sql.ErrNoRows) {
-		return nil
+		return false, nil
 	}
 	if err != nil {
-		return err
+		return false, err
 	}
 	m.mu.RLock()
 	t := m.tasks[id]
@@ -1617,7 +1627,10 @@ func (m *Manager) ApplyProxyPause(id string) error {
 			s.RunSegmentStartedAt = 0
 		})
 	}
-	return nil
+	if callback != nil {
+		callback()
+	}
+	return true, nil
 }
 
 // ClearPauseReason removes auto-resume eligibility without changing lifecycle.
@@ -1633,7 +1646,7 @@ func (m *Manager) ClearPauseReason(id string) error {
 			s.PauseReason = ""
 			s.ReconnectAttempt, s.ReconnectMax, s.ReconnectNextAt = 0, 0, 0
 			s.ReconnectExhausted, s.ReconnectLastError = false, ""
-	})
+		})
 	}
 	return nil
 }

server/proxy_autoresume.go

@@ -108,24 +108,29 @@ func (s *Server) pauseProxyTask(taskID string, cause error) {
 	if strings.TrimSpace(s.m.GlobalProxy()) == "" {
 		return
 	}
-	err := s.m.ApplyProxyPause(taskID)
+	transitioned, err := s.m.ApplyProxyPauseWithCallback(taskID, func() {
+		if s.engine != nil {
+			s.engine.Pause(taskID, agent.AbortPausedByProxy)
+		}
+	})
 	if err != nil {
 		log.Printf("[proxy] pause task %s: %v", taskID, err)
 		return
 	}
-	if s.engine != nil {
-		if t, ok := s.m.Task(taskID); ok && t.lifecycleSnapshot().PauseReason == "paused_by_proxy" {
-			s.engine.Pause(taskID, agent.AbortPausedByProxy)
-		}
+	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)
 	max, interval := s.proxyReconnectConfig()
 	now := time.Now().Add(interval).Unix()
-	for _, t := range s.m.List() {
-		if t.ID != taskID && t.lifecycleSnapshot().PauseReason != "paused_by_proxy" {
-			continue
-		}
+	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
@@ -238,6 +243,14 @@ func (s *Server) proxyReconnectTick() {
 }
 
 func (s *Server) proxyReconnectSuccess() {
+	s.proxyReconnectSuccessWithHook(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()) {
 	max, interval := s.proxyReconnectConfig()
 	s.scheduleMu.Lock()
 	defer s.scheduleMu.Unlock()
@@ -258,12 +271,12 @@ func (s *Server) proxyReconnectSuccess() {
 				continue
 			}
 		}
-		var (
-			queued bool
-			err    error
-		)
+		var err error
 		if s.engine != nil {
-			queued, err = s.admitTaskWhenScheduledWithReason(t, "resume", true, false, "paused_by_proxy", true)
+			_, err = s.admitTaskWhenScheduledWithReason(t, "resume", true, false, "paused_by_proxy", true)
+			if afterAdmission != nil {
+				afterAdmission()
+			}
 		} else {
 			err = s.m.ClearPauseReason(t.ID)
 		}
@@ -281,9 +294,10 @@ func (s *Server) proxyReconnectSuccess() {
 			log.Printf("[proxy] resume task %s: %v", t.ID, err)
 			continue
 		}
-		if s.engine != nil && !queued {
-			s.engine.Resume(t)
-		}
+		// 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) {
 			v.ReconnectAttempt = 0
 			v.ReconnectMax = 0

server/proxy_autoresume_test.go

@@ -192,3 +192,134 @@ func TestProxyAdmissionPreservesCalendarOverride(t *testing.T) {
 		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)
+	}
+}

server/task_control.go

@@ -116,17 +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, pause_reason='paused_by_user' 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 = "paused_by_user"
+					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