fix: 收紧代理故障判定并修复自动恢复门控/日历/终态问题

This commit is contained in:
RuoJi6 committed 2026-10-31 16:23:28 +08:00
1 parent de6a0fa125
commit e878020c4d
15 files changed
+437 -26

No files matched your search

+2
View File
@@ -613,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
@@ -639,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
+46
View File
@@ -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"
+3
View File
@@ -447,6 +447,9 @@ func (s *Server) runHTTPTool(ctx context.Context, execRaw json.RawMessage, param
}
resp, err := client.Do(req)
if err != nil {
if spec.UseRecordingProxy || strings.TrimSpace(spec.Proxy) == strings.TrimSpace(s.m.GlobalProxy()) {
s.handleProxyToolFailure(runCtx, err)
}
return actool.Errorf("请求失败: " + err.Error()), nil
}
defer resp.Body.Close()
+6
View File
@@ -417,6 +417,12 @@ func (s *Server) testEgressProxy(w http.ResponseWriter, r *http.Request) {
rep.OK = judgeEgress(rep)
rep.Warnings = egressWarnings(rep, pending, savedProxy)
if rep.OK && !pending {
// 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()
}
writeJSON(w, 200, rep)
}
+12 -2
View File
@@ -945,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
@@ -1087,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 {
+14 -2
View File
@@ -112,10 +112,19 @@ 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) {
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 +164,9 @@ 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("任务暂停原因已改变")
}
mode = s.resumeAdmissionMode(t)
}
var manualUntil *time.Time
@@ -212,7 +224,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)
}
+61 -6
View File
@@ -262,6 +262,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
@@ -281,6 +282,10 @@ type Manager struct {
globalProxy string
}
// 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"
@@ -1411,12 +1416,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)
@@ -1440,8 +1449,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, pause_reason='', schedule_paused=false,
schedule_manual_until=COALESCE($7::timestamptz, schedule_manual_until),
paused=false, pause_reason='',
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
@@ -1464,7 +1474,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)
@@ -1570,6 +1580,10 @@ 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 {
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)
@@ -1580,7 +1594,6 @@ func (m *Manager) ApplyProxyPause(id string) error {
var active int64
var started sql.NullTime
err = m.pg.QueryRow(`UPDATE tasks SET paused=true, pause_reason='paused_by_proxy', queued=false, queued_at=NULL,
schedule_paused=false, schedule_manual_until=NULL,
active_run_seconds=active_run_seconds + CASE WHEN run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
run_segment_started_at=NULL WHERE id=$1 AND deleted_at IS NULL AND paused=false AND status NOT IN ('done','failed','timeout')
RETURNING COALESCE(queue_mode,''), active_run_seconds, run_segment_started_at`, n).Scan(&mode, &active, &started)
@@ -1616,7 +1629,41 @@ func (m *Manager) ClearPauseReason(id string) error {
t := m.tasks[id]
m.mu.RUnlock()
if t != nil && t.lifecycleSnapshot().PauseReason == "paused_by_proxy" {
t.updateLifecycle(func(s *taskLifecycleState) { s.PauseReason = "" })
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
}
@@ -1835,6 +1882,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()
}
@@ -1871,6 +1922,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()
}
+106 -10
View File
@@ -20,15 +20,32 @@ const (
defaultProxyReconnectInterval = 30
)
// isProxyFailure deliberately requires a configured proxy at the call site and
// only matches transport failures. Provider quota/auth errors must keep their
// existing retry/failover behavior.
// 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", "connection refused", "connection reset", "network is unreachable", "broken pipe", "unexpected eof", "eof"} {
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
}
@@ -81,7 +98,18 @@ func (s *Server) handleProxyFailure(taskID string, cause error) {
if strings.TrimSpace(s.m.GlobalProxy()) == "" {
return
}
if err := s.m.ApplyProxyPause(taskID); err != nil {
if probe := probeEgress(context.Background(), 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
}
err := s.m.ApplyProxyPause(taskID)
if err != nil {
log.Printf("[proxy] pause task %s: %v", taskID, err)
return
}
@@ -108,6 +136,36 @@ func (s *Server) handleProxyFailure(taskID string, cause error) {
s.signalProxySupervisor()
}
func (s *Server) handleProxyTrafficFailure(cause error) {
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 !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 ""
@@ -142,6 +200,16 @@ func (s *Server) proxyReconnectTick() {
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
}
@@ -170,22 +238,50 @@ func (s *Server) proxyReconnectTick() {
}
func (s *Server) proxyReconnectSuccess() {
max, interval := s.proxyReconnectConfig()
s.scheduleMu.Lock()
defer s.scheduleMu.Unlock()
for _, t := range s.m.List() {
st := t.lifecycleSnapshot()
if st.PauseReason != "paused_by_proxy" {
if st := t.lifecycleSnapshot(); st.PauseReason != "paused_by_proxy" {
continue
}
var err error
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 (
queued bool
err error
)
if s.engine != nil {
_, err = s.admitTaskWhenScheduled(t, "resume", true, false)
queued, err = s.admitTaskWhenScheduledWithReason(t, "resume", true, false, "paused_by_proxy", true)
} 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 st.PauseReason == "paused_by_proxy" && !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
}
if s.engine != nil {
if s.engine != nil && !queued {
s.engine.Resume(t)
}
t.updateLifecycle(func(v *taskLifecycleState) {
+141 -1
View File
@@ -4,10 +4,21 @@ import (
"context"
"errors"
"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
@@ -15,7 +26,8 @@ func TestProxyFailureClassification(t *testing.T) {
want bool
}{
{"proxy connect", errors.New("proxyconnect tcp: connection refused"), true},
{"reset", errors.New("dial tcp: connection reset by peer"), 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},
} {
@@ -27,6 +39,15 @@ func TestProxyFailureClassification(t *testing.T) {
}
}
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)
@@ -52,3 +73,122 @@ func TestProxyReconnectReasonGate(t *testing.T) {
}
}
}
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)
}
}
+4
View File
@@ -164,6 +164,7 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
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 {
@@ -194,6 +195,9 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
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 {
+3 -1
View File
@@ -104,7 +104,9 @@ func (s *Server) beginTaskDelete(taskID string) bool {
ok := s.engine.BeginDelete(taskID)
if ok {
// A user delete/archive action is an explicit override of proxy recovery.
_ = s.m.ClearPauseReason(taskID)
if s.m != nil && s.m.pg != nil {
_ = s.m.ClearPauseReason(taskID)
}
}
return ok
}
+5 -1
View File
@@ -150,7 +150,11 @@ func (s *Server) applyTaskControlWithCause(t *Task, action string, pauseCause er
reason = "paused_by_proxy"
}
if err := s.m.ApplyTaskPauseOriginReason(t.ID, reason, pauseCause == agent.AbortPausedBySchedule); err != nil {
if !wasEnginePaused && !wasQueued {
// 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
+32 -2
View File
@@ -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
@@ -423,6 +423,7 @@ function TaskDetailInner() {
<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>
)}
+1 -1
View File
@@ -761,7 +761,7 @@ 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">