diff --git a/db/tasks.go b/db/tasks.go index 4d88586..30a2841 100644 --- a/db/tasks.go +++ b/db/tasks.go @@ -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 diff --git a/server/calendar_schedule.go b/server/calendar_schedule.go index 02a85a3..543f63d 100644 --- a/server/calendar_schedule.go +++ b/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" diff --git a/server/customtool.go b/server/customtool.go index ad68517..cfa9706 100644 --- a/server/customtool.go +++ b/server/customtool.go @@ -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() diff --git a/server/egress_probe.go b/server/egress_probe.go index 70e764b..275f738 100644 --- a/server/egress_probe.go +++ b/server/egress_probe.go @@ -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) } diff --git a/server/engine.go b/server/engine.go index 58b8464..2146ea4 100644 --- a/server/engine.go +++ b/server/engine.go @@ -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 { diff --git a/server/goals.go b/server/goals.go index 45b67f6..32c9cd4 100644 --- a/server/goals.go +++ b/server/goals.go @@ -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) } diff --git a/server/manager.go b/server/manager.go index 0aaeda4..30d4ec2 100644 --- a/server/manager.go +++ b/server/manager.go @@ -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() } diff --git a/server/proxy_autoresume.go b/server/proxy_autoresume.go index 1fbb101..f4f33ff 100644 --- a/server/proxy_autoresume.go +++ b/server/proxy_autoresume.go @@ -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) { diff --git a/server/proxy_autoresume_test.go b/server/proxy_autoresume_test.go index 33137a7..43f1c6b 100644 --- a/server/proxy_autoresume_test.go +++ b/server/proxy_autoresume_test.go @@ -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) + } +} diff --git a/server/server.go b/server/server.go index 5cbc675..8b69043 100644 --- a/server/server.go +++ b/server/server.go @@ -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 { diff --git a/server/server_mgmt.go b/server/server_mgmt.go index a4fbdf5..a50b41e 100644 --- a/server/server_mgmt.go +++ b/server/server_mgmt.go @@ -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 } diff --git a/server/task_control.go b/server/task_control.go index 740fc72..060b1a4 100644 --- a/server/task_control.go +++ b/server/task_control.go @@ -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 diff --git a/traffic/traffic.go b/traffic/traffic.go index bebdb1e..72a250a 100644 --- a/traffic/traffic.go +++ b/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 diff --git a/web/src/app/(main)/function/tasks/detail/page.tsx b/web/src/app/(main)/function/tasks/detail/page.tsx index 6753359..27e9e12 100644 --- a/web/src/app/(main)/function/tasks/detail/page.tsx +++ b/web/src/app/(main)/function/tasks/detail/page.tsx @@ -423,6 +423,7 @@ function TaskDetailInner() {
暂停原因：{task.pause_reason === "paused_by_proxy" ? "全局代理异常" : task.pause_reason} {task.reconnect && 重连 {task.reconnect.attempt}/{task.reconnect.max} {task.reconnect.exhausted ? "已耗尽" : task.reconnect.next_at ? `下次 ${new Date(task.reconnect.next_at * 1000).toLocaleTimeString()}` : "探测中"}} + {task.reconnect?.last_error && 原因：{task.reconnect.last_error}} {task.pause_reason === "paused_by_proxy" && }
)} diff --git a/web/src/app/(main)/system/settings/page.tsx b/web/src/app/(main)/system/settings/page.tsx index 9c2e25d..69407cc 100644 --- a/web/src/app/(main)/system/settings/page.tsx +++ b/web/src/app/(main)/system/settings/page.tsx @@ -761,7 +761,7 @@ export default function SystemSettingsPage() { {/* 测表单里当前的值，不要求先保存。 */}