fix: 代理池无健康成员走暂停不直连、单飞探测、热切覆盖 capture-off 与浏览器 MCP、脱敏与 UI 补全

This commit is contained in:
RuoJi6 committed 2026-11-01 20:20:02 +08:00
1 parent 0bb954e0a4
commit 45514b3586
10 files changed
+599 -157

No files matched your search

+5 -5
View File
@@ -396,7 +396,11 @@ func (w *Worker) execute(ctx context.Context, name string, taskID int64, as *db.
// inherit a parent run's background into the action reviewer.
ctx = intercept.WithReviewContext(ctx, runDir, intercept.ReviewBackground{})
overview := renderWorkerGraphOverview(tsx.graphOverviewData())
sysBody := workerSystem(w.proxyAddr, w.proxyCACert, w.workDir, runDir)
proxyAddr := w.proxyAddr
if w.proxyAddrFn != nil {
proxyAddr = w.proxyAddrFn()
}
sysBody := workerSystem(proxyAddr, w.proxyCACert, w.workDir, runDir)
if w.wantConstraints() {
sysBody += constraintBlock(ts) // 操作约束(若有)注入系统提示,worker 执行时严格遵守
}
@@ -429,10 +433,6 @@ func (w *Worker) execute(ctx context.Context, name string, taskID int64, as *db.
if tc.DeadlineUnix > 0 {
settle = wrapupSettlementForTask("worker", []string{"Bash"}, clamped)
}
proxyAddr := w.proxyAddr
if w.proxyAddrFn != nil {
proxyAddr = w.proxyAddrFn()
}
opts := agentcore.Options{
Provider: w.prov,
SystemPrompt: system,
+1
View File
@@ -113,6 +113,7 @@ func run() int {
}
log.Printf("[config] skill 目录: %s", skillDir)
srv := server.New(ctx, mgr, skillDir, *dataDir, config.BaseDir())
defer srv.Close()
httpSrv := &http.Server{
Addr: *addr,
Handler: srv.Handler(),
+205
View File
@@ -0,0 +1,205 @@
package server
import (
"bufio"
"context"
"crypto/tls"
"fmt"
"io"
"net"
"net/http"
"net/url"
"strings"
"sync"
"golang.org/x/net/proxy"
)
// dynamicForwardProxy is a credential-free, stable local proxy. Each request
// resolves the current upstream, so pool switches do not rebuild agents or MCP.
type dynamicForwardProxy struct {
ln net.Listener
addr string
current func() string
direct func() bool
server *http.Server
once sync.Once
}
func newDynamicForwardProxy(current func() string) (*dynamicForwardProxy, error) {
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
return nil, err
}
p := &dynamicForwardProxy{ln: ln, addr: "http://" + ln.Addr().String(), current: current, direct: func() bool { return true }}
p.server = &http.Server{Handler: http.HandlerFunc(p.serveHTTP)}
go p.server.Serve(ln)
return p, nil
}
func (p *dynamicForwardProxy) Addr() string {
if p == nil {
return ""
}
return p.addr
}
func (p *dynamicForwardProxy) Close() error {
if p == nil {
return nil
}
p.once.Do(func() { _ = p.server.Close() })
return nil
}
func (p *dynamicForwardProxy) serveHTTP(w http.ResponseWriter, r *http.Request) {
if r.Method == http.MethodConnect {
p.serveConnect(w, r)
return
}
up := strings.TrimSpace(p.current())
if up == "" && p.direct != nil && !p.direct() {
http.Error(w, "proxy unavailable", 502)
return
}
tr := &http.Transport{Proxy: nil}
if up != "" {
u, err := url.Parse(up)
if err != nil {
http.Error(w, "proxy unavailable", 502)
return
}
if strings.HasPrefix(strings.ToLower(u.Scheme), "socks5") {
d, err := socksDialer(u)
if err != nil {
http.Error(w, "proxy unavailable", 502)
return
}
tr.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) { return d.Dial(network, addr) }
} else {
tr.Proxy = http.ProxyURL(u)
}
}
tr.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12}
defer tr.CloseIdleConnections()
rr := r.Clone(r.Context())
if rr.URL.IsAbs() == false {
rr.URL.Scheme = "http"
rr.URL.Host = r.Host
}
rr.RequestURI = ""
for k := range rr.Header {
if strings.EqualFold(k, "Proxy-Authorization") {
rr.Header.Del(k)
}
}
resp, err := tr.RoundTrip(rr)
if err != nil {
http.Error(w, "proxy unavailable", 502)
return
}
defer resp.Body.Close()
for k, vv := range resp.Header {
for _, v := range vv {
w.Header().Add(k, v)
}
}
w.WriteHeader(resp.StatusCode)
_, _ = io.Copy(w, resp.Body)
}
func (p *dynamicForwardProxy) serveConnect(w http.ResponseWriter, r *http.Request) {
up := strings.TrimSpace(p.current())
if up == "" && p.direct != nil && !p.direct() {
http.Error(w, "proxy unavailable", 502)
return
}
target := r.Host
var conn net.Conn
var err error
if up == "" {
conn, err = (&net.Dialer{}).DialContext(r.Context(), "tcp", target)
} else {
u, e := url.Parse(up)
if e != nil {
err = e
} else if strings.HasPrefix(strings.ToLower(u.Scheme), "socks5") {
d, e := socksDialer(u)
if e != nil {
err = e
} else {
conn, err = d.Dial("tcp", target)
}
} else {
conn, err = dialHTTPConnect(r.Context(), u, target)
}
}
if err != nil {
http.Error(w, "proxy unavailable", 502)
return
}
defer conn.Close()
h, ok := w.(http.Hijacker)
if !ok {
http.Error(w, "proxy unavailable", 500)
return
}
client, rw, err := h.Hijack()
if err != nil {
return
}
defer client.Close()
_, _ = rw.WriteString("HTTP/1.1 200 Connection Established\r\n\r\n")
_ = rw.Flush()
go io.Copy(conn, client)
_, _ = io.Copy(client, conn)
}
func socksDialer(u *url.URL) (proxy.Dialer, error) {
var auth *proxy.Auth
if u.User != nil {
pw, _ := u.User.Password()
auth = &proxy.Auth{User: u.User.Username(), Password: pw}
}
return proxy.SOCKS5("tcp", u.Host, auth, &net.Dialer{})
}
func dialHTTPConnect(ctx context.Context, u *url.URL, target string) (net.Conn, error) {
d := &net.Dialer{}
addr := u.Host
if !strings.Contains(addr, ":") {
addr += ":80"
}
conn, err := d.DialContext(ctx, "tcp", addr)
if err != nil {
return nil, err
}
if strings.EqualFold(u.Scheme, "https") {
conn = tls.Client(conn, &tls.Config{ServerName: u.Hostname(), MinVersion: tls.VersionTLS12})
if err := conn.(*tls.Conn).HandshakeContext(ctx); err != nil {
_ = conn.Close()
return nil, err
}
}
br := bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn))
fmt.Fprintf(br, "CONNECT %s HTTP/1.1\r\nHost: %s\r\n", target, target)
if u.User != nil {
req := &http.Request{Header: http.Header{}}
req.SetBasicAuth(u.User.Username(), func() string { p, _ := u.User.Password(); return p }())
fmt.Fprintf(br, "Proxy-Authorization: %s\r\n", req.Header.Get("Proxy-Authorization"))
}
fmt.Fprint(br, "\r\n")
if err := br.Flush(); err != nil {
_ = conn.Close()
return nil, err
}
resp, err := http.ReadResponse(br.Reader, &http.Request{Method: http.MethodConnect})
if err != nil {
_ = conn.Close()
return nil, err
}
if resp.StatusCode/100 != 2 {
_ = conn.Close()
return nil, fmt.Errorf("upstream CONNECT: %s", resp.Status)
}
return conn, nil
}
+39 -38
View File
@@ -13,6 +13,7 @@ import (
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/Autumn-27/artex/agent"
@@ -282,8 +283,10 @@ type Manager struct {
// (http/https/socks5, optional user:pass). Empty = direct. When traffic
// capture is on it becomes the MITM's upstream; when capture is off it is
// injected into agent bash env / WebFetch directly. See ProxyAddr.
globalProxy string
proxyPool *proxyPoolState
globalProxy string
proxyPool *proxyPoolState
forwardProxy *dynamicForwardProxy
proxyPoolConfigured atomic.Bool
}
// SetTaskControlMutex lets lifecycle operations that originate in Manager
@@ -583,6 +586,12 @@ func NewManager(dir, proxyAddr string) (*Manager, error) {
log.Printf("[llmusage] create table: %v", err)
}
m := &Manager{dir: dir, pg: pg, assets: pg.Assets(), tasks: map[string]*Task{}, interceptor: intercept.New(pg)}
if fp, ferr := newDynamicForwardProxy(func() string { return m.GlobalProxy() }); ferr == nil {
m.forwardProxy = fp
fp.direct = func() bool { return !m.proxyPoolConfigured.Load() }
} else {
log.Printf("[proxy] local forwarder disabled: %v", ferr)
}
if proxyAddr != "" {
tr, err := traffic.Open(filepath.Join(dir, "traffic"), proxyAddr)
if err != nil {
@@ -1018,8 +1027,12 @@ func (m *Manager) ProxyAddr() string {
return m.traffic.ProxyAddr()
}
m.mu.RLock()
defer m.mu.RUnlock()
return m.globalProxy
global := m.globalProxy
m.mu.RUnlock()
if (global != "" || m.proxyPoolConfigured.Load()) && m.forwardProxy != nil {
return m.forwardProxy.Addr()
}
return maskSecrets(global)
}
// ProxyCACert returns the CA cert path agents must trust to verify HTTPS through
@@ -1049,18 +1062,31 @@ func (m *Manager) SetGlobalProxy(raw string) error {
raw = strings.TrimSpace(raw)
if raw != "" {
if _, err := traffic.ValidateProxyURL(raw); err != nil {
return err
return fmt.Errorf("%s", maskSecrets(err.Error()))
}
}
poolLocked := false
if m.proxyPool != nil {
m.proxyPool.mu.Lock()
poolLocked = true
defer func() {
if poolLocked {
managed := len(m.proxyPool.entries) > 0
if !managed && raw != "" {
m.proxyPool.entries = []proxyPoolEntry{{ProxyPoolEntry: ProxyPoolEntry{ID: "proxy-legacy", URL: maskSecrets(raw), Enabled: true, Health: "healthy"}, RawURL: raw}}
m.proxyPool.active = "proxy-legacy"
m.proxyPoolConfigured.Store(true)
if err := m.applyActiveProxyLocked("proxy-legacy", "manual"); err != nil {
m.proxyPool.entries = nil
m.proxyPool.active = ""
m.proxyPool.mu.Unlock()
return err
}
}()
m.proxyPool.mu.Unlock()
return nil
}
m.proxyPool.mu.Unlock()
if managed {
// The legacy field is a read-only echo once a pool exists. Pool API
// operations are the sole authority for health, enablement and switching.
return nil
}
}
if err := m.pg.SetSetting(settingGlobalProxy, raw); err != nil {
return err
@@ -1073,39 +1099,14 @@ func (m *Manager) SetGlobalProxy(raw string) error {
return err
}
}
// Keep the browser MCP's egress in sync with the new global proxy too.
m.syncBrowserMCPProxy()
// Keep the legacy single-proxy setting compatible with the pool. If a pool
// exists, update its active entry; otherwise this remains a pool of one.
if m.proxyPool != nil {
if len(m.proxyPool.entries) == 0 && raw != "" {
m.proxyPool.entries = []proxyPoolEntry{{ProxyPoolEntry: ProxyPoolEntry{ID: "proxy-legacy", URL: maskSecrets(raw), Enabled: true, Health: "healthy"}, RawURL: raw}}
m.proxyPool.active = "proxy-legacy"
} else if len(m.proxyPool.entries) > 0 && raw != "" {
found := false
for i := range m.proxyPool.entries {
if m.proxyPool.entries[i].ID == m.proxyPool.active {
m.proxyPool.entries[i].RawURL = raw
m.proxyPool.entries[i].URL = maskSecrets(raw)
m.proxyPool.entries[i].Enabled = true
m.proxyPool.entries[i].Health = "healthy"
found = true
break
}
}
if !found {
m.proxyPool.entries[0].RawURL, m.proxyPool.entries[0].URL, m.proxyPool.entries[0].Enabled, m.proxyPool.entries[0].Health = raw, maskSecrets(raw), true, "healthy"
m.proxyPool.active = m.proxyPool.entries[0].ID
}
} else if raw == "" && len(m.proxyPool.entries) == 0 {
m.proxyPool.active = ""
}
_ = m.persistProxyPoolLocked()
}
return nil
}
func (m *Manager) Close() error {
if m.forwardProxy != nil {
_ = m.forwardProxy.Close()
}
m.mu.Lock()
defer m.mu.Unlock()
if m.traffic != nil {
+23 -1
View File
@@ -135,7 +135,10 @@ func (s *Server) handleProxyFailure(taskID string, cause error) {
if strings.TrimSpace(s.m.GlobalProxy()) == "" {
return
}
if probe := probeEgress(context.Background(), s.m.GlobalProxy(), "", defaultEgressProbes, 10*time.Second); probe.OK {
if s.handleProxyPoolFailure(cause) {
return
}
if probe := s.runProxyProbe(s.ctx, s.m.GlobalProxy(), "", defaultEgressProbes, 10*time.Second); probe.OK {
return
}
s.pauseProxyTask(taskID, cause)
@@ -247,6 +250,25 @@ func (s *Server) proxyReconnectTick() {
if strings.TrimSpace(s.m.GlobalProxy()) == "" {
return
}
if s.m.proxyPool != nil && len(s.m.proxyPoolEntries()) > 0 {
active := s.m.proxyPoolActiveID()
found := false
for _, e := range s.m.proxyPoolEntries() {
if e.ID == active && e.Enabled && e.Health == "healthy" {
found = true
break
}
if e.ID == active {
found = true
s.proxyPoolTick()
return
}
}
if !found {
s.proxyPoolTick()
return
}
}
now := time.Now().Unix()
max, interval := s.proxyReconnectConfig()
for _, t := range s.m.List() {
+248 -107
View File
@@ -7,6 +7,7 @@ import (
"fmt"
"math/rand"
"net/http"
"net/url"
"strings"
"sync"
"time"
@@ -89,7 +90,7 @@ func ParseProxyBatch(raw string) ([]ProxyPoolEntry, []ProxyImportError) {
errs = append(errs, ProxyImportError{Line: n + 1, Input: maskSecrets(line), Error: maskSecrets(err.Error())})
continue
}
key := strings.ToLower(u.String())
key := proxyURLKey(u)
if seen[key] {
errs = append(errs, ProxyImportError{Line: n + 1, Input: maskSecrets(line), Error: "重复代理,已跳过"})
continue
@@ -102,17 +103,36 @@ func ParseProxyBatch(raw string) ([]ProxyPoolEntry, []ProxyImportError) {
if len(parts) > 2 {
remark = strings.TrimSpace(parts[2])
}
out = append(out, ProxyPoolEntry{ID: fmt.Sprintf("proxy-%d", time.Now().UnixNano()+int64(n)), URL: rawURL, Name: name, Remark: remark, Enabled: true, Health: "unhealthy-cooling"})
out = append(out, ProxyPoolEntry{ID: fmt.Sprintf("proxy-%d-%d", time.Now().UnixNano(), n+1), URL: rawURL, Name: name, Remark: remark, Enabled: true, Health: "unhealthy-cooling"})
}
return out, errs
}
func proxyURLKey(u *url.URL) string {
if u == nil {
return ""
}
v := *u
v.Host = strings.ToLower(v.Host)
v.Scheme = strings.ToLower(v.Scheme)
return v.String()
}
func proxyURLKeyRaw(raw string) string {
u, err := traffic.ValidateProxyURL(raw)
if err != nil {
return strings.ToLower(strings.TrimSpace(raw))
}
return proxyURLKey(u)
}
type proxyPoolState struct {
mu sync.Mutex
entries []proxyPoolEntry
active string
logs []ProxySwitchLog
lastTimed time.Time
mu sync.Mutex
entries []proxyPoolEntry
active string
logs []ProxySwitchLog
lastTimed time.Time
lastHealth time.Time
}
func (m *Manager) initProxyPool() {
@@ -141,6 +161,7 @@ func (m *Manager) initProxyPool() {
}
}
}
m.proxyPoolConfigured.Store(len(m.proxyPool.entries) > 0)
}
func (m *Manager) persistProxyPoolLocked() error {
@@ -195,21 +216,16 @@ func (m *Manager) setProxyPoolEntries(entries []proxyPoolEntry, active string) e
m.proxyPool.mu.Lock()
defer m.proxyPool.mu.Unlock()
m.proxyPool.entries, m.proxyPool.active = entries, active
m.proxyPoolConfigured.Store(len(entries) > 0)
if err := m.persistProxyPoolLocked(); err != nil {
return err
}
for _, e := range entries {
if e.ID == active && e.Enabled && e.Health == "healthy" {
m.mu.Lock()
m.globalProxy = e.RawURL
m.mu.Unlock()
if m.pg != nil {
_ = m.pg.SetSetting(settingGlobalProxy, e.RawURL)
}
break
}
}
if active == "" {
for _, e := range entries {
if e.Enabled && e.Health == "healthy" {
return m.applyActiveProxyLocked(e.ID, "reconnect")
}
}
return nil
}
for _, e := range entries {
@@ -324,6 +340,10 @@ func (m *Manager) markProxyProbe(id string, res egressProbeResult) {
}
m.proxyPool.mu.Lock()
defer m.proxyPool.mu.Unlock()
m.markProxyProbeLocked(id, res)
}
func (m *Manager) markProxyProbeLocked(id string, res egressProbeResult) {
for i := range m.proxyPool.entries {
e := &m.proxyPool.entries[i]
if e.ID != id {
@@ -395,26 +415,39 @@ func (m *Manager) proxyRotateConfig() (bool, time.Duration, bool, string) {
func (s *Server) runProxyPoolSupervisor(ctx context.Context) {
for {
_, interval, _, _ := s.m.proxyRotateConfig()
_, rotateInterval, _, _ := s.m.proxyRotateConfig()
_, _, healthInterval := s.m.proxyHealthConfig()
if healthInterval > 0 && healthInterval < interval {
interval = healthInterval
if rotateInterval <= 0 {
rotateInterval = defaultProxyRotateInterval * time.Second
}
if interval < minProxyRotateInterval*time.Second {
interval = minProxyRotateInterval * time.Second
if healthInterval <= 0 {
healthInterval = defaultProxyHealthInterval * time.Second
}
t := time.NewTimer(interval)
d := healthInterval
if timed, _, _, _ := s.m.proxyRotateConfig(); timed && rotateInterval < d {
d = rotateInterval
}
if d <= 0 {
d = time.Second
}
t := time.NewTimer(d)
select {
case <-ctx.Done():
if !t.Stop() {
<-t.C
select {
case <-t.C:
default:
}
}
return
case <-t.C:
s.proxyPoolTick()
case <-s.proxyPoolWake:
if !t.Stop() {
<-t.C
select {
case <-t.C:
default:
}
}
s.proxyPoolTick()
}
@@ -422,34 +455,53 @@ func (s *Server) runProxyPoolSupervisor(ctx context.Context) {
}
func (s *Server) proxyPoolTick() {
if s == nil || s.m == nil || s.m.proxyPool == nil {
if s == nil {
return
}
s.proxyPoolTickMu.Lock()
defer s.proxyPoolTickMu.Unlock()
s.proxyPoolTickLocked(false)
}
func (s *Server) proxyPoolTickLocked(forceHealth bool) {
if s.m == nil || s.m.proxyPool == nil {
return
}
now := time.Now()
timed, rotateInterval, _, _ := s.m.proxyRotateConfig()
if timed {
s.m.proxyPool.mu.Lock()
due := s.m.proxyPool.lastTimed.IsZero() || time.Since(s.m.proxyPool.lastTimed) >= rotateInterval
due := s.m.proxyPool.lastTimed.IsZero() || now.Sub(s.m.proxyPool.lastTimed) >= rotateInterval
if due {
s.m.proxyPool.lastTimed = time.Now()
s.m.proxyPool.lastTimed = now
}
s.m.proxyPool.mu.Unlock()
if due {
_ = s.m.rotateProxy("timed")
}
}
threshold, cooldown, _ := s.m.proxyHealthConfig()
threshold, cooldown, healthInterval := s.m.proxyHealthConfig()
s.m.proxyPool.mu.Lock()
healthDue := s.m.proxyPool.lastHealth.IsZero() || now.Sub(s.m.proxyPool.lastHealth) >= healthInterval
if healthDue {
s.m.proxyPool.lastHealth = now
}
s.m.proxyPool.mu.Unlock()
if !healthDue && !forceHealth {
s.reconcileProxyPoolAfterProbe()
return
}
entries := s.m.proxyPoolEntries()
sem := s.proxyPoolProbeSem
if sem == nil {
sem = make(chan struct{}, proxyPoolMaxProbeConcurrency)
}
var wg sync.WaitGroup
now := time.Now()
for _, e := range entries {
if !e.Enabled {
continue
}
if e.Health == "unhealthy-cooling" && e.ConsecutiveFailures >= threshold && !e.LastCheckedAt.IsZero() && now.Sub(e.LastCheckedAt) < cooldown {
if !forceHealth && e.Health == "unhealthy-cooling" && e.ConsecutiveFailures >= threshold && !e.LastCheckedAt.IsZero() && now.Sub(e.LastCheckedAt) < cooldown {
continue
}
e := e
@@ -467,9 +519,11 @@ func (s *Server) proxyPoolTick() {
}()
}
wg.Wait()
// A successful probe can become the active proxy after the pool was down or
// after the previously active entry cooled out.
entries = s.m.proxyPoolEntries()
s.reconcileProxyPoolAfterProbe()
}
func (s *Server) reconcileProxyPoolAfterProbe() {
entries := s.m.proxyPoolEntries()
activeID := s.m.proxyPoolActiveID()
activeHealthy := false
for _, e := range entries {
@@ -478,17 +532,27 @@ func (s *Server) proxyPoolTick() {
break
}
}
if !activeHealthy {
for _, e := range entries {
if e.Enabled && e.Health == "healthy" {
_ = s.m.activateProxyID(e.ID, "reconnect")
s.proxyReconnectSuccess()
return
}
}
if activeID != "" {
_ = s.m.activateProxyID("", "reconnect")
}
if activeHealthy {
s.proxyReconnectSuccess()
return
}
s.m.proxyPool.mu.Lock()
id := s.m.selectHealthyLocked(activeID)
var err error
if id != "" {
err = s.m.applyActiveProxyLocked(id, "reconnect")
}
s.m.proxyPool.mu.Unlock()
if err == nil && id != "" {
s.proxyReconnectSuccess()
return
}
if err != nil {
s.pauseAllProxyTasks(err)
return
}
if id == "" && (activeID != "" || strings.TrimSpace(s.m.GlobalProxy()) != "") {
s.pauseAllProxyTasks(errors.New("代理池暂无健康成员"))
}
}
@@ -505,26 +569,38 @@ func (m *Manager) activateProxyID(id, reason string) error {
}
func (s *Server) handleProxyPoolFailure(cause error) bool {
if s == nil || s.m == nil || s.m.proxyPool == nil {
if s == nil || s.m == nil || s.m.proxyPool == nil || !isGlobalProxyDialFailure(cause) {
return false
}
if !isGlobalProxyDialFailure(cause) || strings.TrimSpace(s.m.GlobalProxy()) == "" {
cur := strings.TrimSpace(s.m.GlobalProxy())
if cur == "" {
return false
}
cur := s.m.GlobalProxy()
res := s.runProxyProbe(context.Background(), cur, "", defaultEgressProbes, proxyPoolProbeTimeout)
res := s.runProxyProbe(s.ctx, cur, "", defaultEgressProbes, proxyPoolProbeTimeout)
if res.OK {
return false
}
for _, e := range s.m.proxyPoolEntries() {
if e.RawURL == cur {
s.m.markProxyProbe(e.ID, res)
_, _, onFailure, _ := s.m.proxyRotateConfig()
s.proxyPoolTickMu.Lock()
defer s.proxyPoolTickMu.Unlock()
s.m.proxyPool.mu.Lock()
defer s.m.proxyPool.mu.Unlock()
if s.m.GlobalProxy() != cur {
return false
}
active := s.m.proxyPool.active
for _, e := range s.m.proxyPool.entries {
if e.ID == active && e.RawURL == cur {
s.m.markProxyProbeLocked(e.ID, res)
break
}
}
_, _, onFailure, _ := s.m.proxyRotateConfig()
if onFailure && s.m.rotateProxy("failure") {
return true
if onFailure {
if id := s.m.selectHealthyLocked(active); id != "" && id != active {
if err := s.m.applyActiveProxyLocked(id, "failure"); err == nil {
return true
}
}
}
select {
case s.proxyPoolWake <- struct{}{}:
@@ -533,6 +609,16 @@ func (s *Server) handleProxyPoolFailure(cause error) bool {
return false
}
func (s *Server) pauseAllProxyTasks(cause error) {
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) getProxyPool(w http.ResponseWriter, r *http.Request) {
entries, active, logs := s.m.ProxyPoolSnapshot()
timed, interval, onFailure, strategy := s.m.proxyRotateConfig()
@@ -559,15 +645,21 @@ func (s *Server) importProxyPool(w http.ResponseWriter, r *http.Request) {
entries := s.m.proxyPoolEntries()
seen := map[string]bool{}
for _, e := range entries {
seen[e.RawURL] = true
seen[proxyURLKeyRaw(e.RawURL)] = true
}
for _, p := range parsed {
if seen[p.URL] {
key := proxyURLKeyRaw(p.URL)
if seen[key] {
line := 0
if i := strings.LastIndex(p.ID, "-"); i >= 0 {
_, _ = fmt.Sscanf(p.ID[i+1:], "%d", &line)
}
errs = append(errs, ProxyImportError{Line: line, Input: maskSecrets(p.URL), Error: "重复代理,已跳过"})
continue
}
u, _ := traffic.ValidateProxyURL(p.URL)
entries = append(entries, proxyPoolEntry{ProxyPoolEntry: p, RawURL: u.String()})
seen[u.String()] = true
seen[proxyURLKey(u)] = true
}
active := s.m.proxyPoolActiveID()
if active != "" {
@@ -632,6 +724,28 @@ func (m *Manager) proxyPoolActiveID() string {
return m.proxyPool.active
}
func mergeMaskedProxyURL(raw, old string) (string, error) {
raw = strings.TrimSpace(raw)
if raw == "" {
return "", nil
}
nu, err := traffic.ValidateProxyURL(raw)
if err != nil {
return "", fmt.Errorf("%s", maskSecrets(err.Error()))
}
if ou, _ := traffic.ValidateProxyURL(old); ou != nil && nu.User != nil {
pw, _ := nu.User.Password()
if strings.Contains(pw, "***") || strings.Contains(pw, "******") {
if oldpw, ok := ou.User.Password(); ok {
nu.User = url.UserPassword(nu.User.Username(), oldpw)
} else {
nu.User = url.User(nu.User.Username())
}
}
}
return nu.String(), nil
}
func (s *Server) updateProxyPool(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
var req struct {
@@ -648,43 +762,22 @@ func (s *Server) updateProxyPool(w http.ResponseWriter, r *http.Request) {
writeErr(w, 404, "代理池为空")
return
}
needPause := false
s.m.proxyPool.mu.Lock()
defer s.m.proxyPool.mu.Unlock()
for i := range s.m.proxyPool.entries {
e := &s.m.proxyPool.entries[i]
if e.ID != id {
continue
}
oldRaw := e.RawURL
if req.URL != nil {
raw := strings.TrimSpace(*req.URL)
if strings.Contains(raw, "***") || strings.Contains(raw, "******") {
raw = e.RawURL
}
if e.ID == s.m.proxyPool.active && e.Enabled && e.RawURL != "" {
if s.m.traffic != nil {
if err := s.m.traffic.SetUpstreamProxy(e.RawURL); err != nil {
writeErr(w, 400, err.Error())
return
}
}
s.m.mu.Lock()
s.m.globalProxy = e.RawURL
s.m.mu.Unlock()
if s.m.pg != nil {
_ = s.m.pg.SetSetting(settingGlobalProxy, e.RawURL)
}
}
if raw != "" {
u, err := traffic.ValidateProxyURL(raw)
if err != nil {
writeErr(w, 400, err.Error())
return
}
e.RawURL = u.String()
e.URL = maskSecrets(e.RawURL)
} else {
e.RawURL, e.URL = "", ""
raw, err := mergeMaskedProxyURL(*req.URL, oldRaw)
if err != nil {
s.m.proxyPool.mu.Unlock()
writeErr(w, 400, err.Error())
return
}
e.RawURL, e.URL = raw, maskSecrets(raw)
}
if req.Name != nil {
e.Name = strings.TrimSpace(*req.Name)
@@ -694,21 +787,50 @@ func (s *Server) updateProxyPool(w http.ResponseWriter, r *http.Request) {
}
if req.Enabled != nil {
e.Enabled = *req.Enabled
if !e.Enabled && s.m.proxyPool.active == id {
_ = s.m.applyActiveProxyLocked("", "manual")
if !e.Enabled {
e.Health = "disabled"
} else if e.Health == "disabled" {
e.Health = "healthy"
}
}
if s.m.proxyPool.active == id {
if req.URL != nil || req.Enabled != nil {
if e.Enabled && e.RawURL != "" {
if err := s.m.applyActiveProxyLocked(id, "manual"); err != nil {
e.RawURL, e.URL = oldRaw, maskSecrets(oldRaw)
s.m.proxyPool.mu.Unlock()
writeErr(w, 400, maskSecrets(err.Error()))
return
}
} else if next := s.m.selectHealthyLocked(id); next != "" {
if err := s.m.applyActiveProxyLocked(next, "manual"); err != nil {
s.m.proxyPool.mu.Unlock()
writeErr(w, 400, maskSecrets(err.Error()))
return
}
} else {
needPause = true
}
}
}
if err := s.m.persistProxyPoolLocked(); err != nil {
s.m.proxyPool.mu.Unlock()
writeErr(w, 500, err.Error())
return
}
health := e.Health
if !e.Enabled {
health = "disabled"
out := e.ProxyPoolEntry
out.URL = maskSecrets(e.RawURL)
if !out.Enabled {
out.Health = "disabled"
}
writeJSON(w, 200, map[string]any{"entry": ProxyPoolEntry{ID: e.ID, URL: maskSecrets(e.RawURL), Name: e.Name, Remark: e.Remark, Enabled: e.Enabled, Health: health}})
s.m.proxyPool.mu.Unlock()
if needPause {
s.pauseAllProxyTasks(errors.New("代理池暂无健康成员"))
}
writeJSON(w, 200, map[string]any{"entry": out})
return
}
s.m.proxyPool.mu.Unlock()
writeErr(w, 404, "代理不存在")
}
@@ -718,8 +840,8 @@ func (s *Server) deleteProxyPool(w http.ResponseWriter, r *http.Request) {
writeErr(w, 404, "代理不存在")
return
}
needPause := false
s.m.proxyPool.mu.Lock()
defer s.m.proxyPool.mu.Unlock()
found := false
out := s.m.proxyPool.entries[:0]
for _, e := range s.m.proxyPool.entries {
@@ -730,18 +852,33 @@ func (s *Server) deleteProxyPool(w http.ResponseWriter, r *http.Request) {
out = append(out, e)
}
if !found {
s.m.proxyPool.mu.Unlock()
writeErr(w, 404, "代理不存在")
return
}
s.m.proxyPool.entries = out
s.m.proxyPoolConfigured.Store(len(out) > 0)
if s.m.proxyPool.active == id {
s.m.proxyPool.active = ""
if err := s.m.applyActiveProxyLocked("", "manual"); err != nil {
writeErr(w, 500, err.Error())
return
if next := s.m.selectHealthyLocked(id); next != "" {
if err := s.m.applyActiveProxyLocked(next, "manual"); err != nil {
s.m.proxyPool.mu.Unlock()
writeErr(w, 500, maskSecrets(err.Error()))
return
}
} else {
s.m.proxyPool.active = ""
needPause = true
}
}
_ = s.m.persistProxyPoolLocked()
if err := s.m.persistProxyPoolLocked(); err != nil {
s.m.proxyPool.mu.Unlock()
writeErr(w, 500, err.Error())
return
}
s.m.proxyPool.mu.Unlock()
if needPause {
s.pauseAllProxyTasks(errors.New("代理池暂无健康成员"))
}
writeJSON(w, 200, map[string]any{"ok": true})
}
@@ -755,14 +892,15 @@ func (s *Server) activateProxyPool(w http.ResponseWriter, r *http.Request) {
}
func (s *Server) testProxyPoolEntry(w http.ResponseWriter, r *http.Request) {
s.proxyPoolTickMu.Lock()
defer s.proxyPoolTickMu.Unlock()
id := r.PathValue("id")
for _, e := range s.m.proxyPoolEntries() {
if e.ID == id {
res := s.runProxyProbe(r.Context(), e.RawURL, "", defaultEgressProbes, proxyPoolProbeTimeout)
s.m.markProxyProbe(id, res)
if res.OK && s.m.GlobalProxy() == "" {
_ = s.m.activateProxyID(id, "reconnect")
s.proxyReconnectSuccess()
if res.OK {
s.reconcileProxyPoolAfterProbe()
}
writeJSON(w, 200, map[string]any{"ok": res.OK, "result": res})
return
@@ -772,11 +910,14 @@ func (s *Server) testProxyPoolEntry(w http.ResponseWriter, r *http.Request) {
}
func (s *Server) testAllProxyPool(w http.ResponseWriter, r *http.Request) {
select {
case s.proxyPoolWake <- struct{}{}:
default:
s.proxyPoolTickMu.Lock()
if s.m != nil && s.m.proxyPool != nil {
s.m.proxyPool.mu.Lock()
s.m.proxyPool.lastHealth = time.Time{}
s.m.proxyPool.mu.Unlock()
}
s.proxyPoolTick()
s.proxyPoolTickLocked(true)
s.proxyPoolTickMu.Unlock()
entries, active, logs := s.m.ProxyPoolSnapshot()
writeJSON(w, 200, map[string]any{"entries": entries, "active_id": active, "switch_log": logs})
}
+51
View File
@@ -2,8 +2,13 @@ package server
import (
"encoding/json"
"net/http"
"net/http/httptest"
"net/url"
"strings"
"testing"
"github.com/Autumn-27/artex/traffic"
)
func TestParseProxyBatchValidationAndDedup(t *testing.T) {
@@ -54,3 +59,49 @@ func TestProxyPoolPersistenceKeepsRawURL(t *testing.T) {
t.Fatalf("bad persistence encoding: %s", b)
}
}
func TestProxyPoolURLKeyNormalizesHostCase(t *testing.T) {
a, _ := traffic.ValidateProxyURL("http://user:pass@HOST:8080")
b, _ := traffic.ValidateProxyURL("http://user:pass@host:8080")
if proxyURLKey(a) != proxyURLKey(b) {
t.Fatalf("keys differ: %q %q", proxyURLKey(a), proxyURLKey(b))
}
}
func TestMergeMaskedProxyURLPreservesOnlyPassword(t *testing.T) {
got, err := mergeMaskedProxyURL("http://newuser:***@NEW.example:9090", "http://old:[email protected]:8080")
if err != nil {
t.Fatal(err)
}
if got != "http://newuser:[email protected]:9090" {
t.Fatalf("got %q", got)
}
}
func TestMergeMaskedProxyURLErrorIsMasked(t *testing.T) {
_, err := mergeMaskedProxyURL("http://user:secret@", "")
if err == nil || strings.Contains(err.Error(), "secret") {
t.Fatalf("unmasked error: %v", err)
}
}
func TestDynamicForwardProxyUsesCurrentUpstream(t *testing.T) {
target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte("ok")) }))
defer target.Close()
up := ""
fp, err := newDynamicForwardProxy(func() string { return up })
if err != nil {
t.Fatal(err)
}
defer fp.Close()
pu, _ := url.Parse(fp.Addr())
c := &http.Client{Transport: &http.Transport{Proxy: http.ProxyURL(pu)}}
resp, err := c.Get(target.URL)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
if resp.StatusCode != 200 {
t.Fatalf("status %d", resp.StatusCode)
}
}
+10 -2
View File
@@ -132,6 +132,7 @@ type Server struct {
proxyPoolWake chan struct{}
proxyPoolWG sync.WaitGroup
proxyPoolProbeSem chan struct{}
proxyPoolTickMu sync.Mutex
// 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
@@ -311,6 +312,13 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
return s
}
// Close waits for proxy pool supervision to stop before Manager closes the DB.
func (s *Server) Close() {
if s != nil {
s.proxyPoolWG.Wait()
}
}
// Restored deadline and worker loops must inherit the same context as new tasks,
// including the side-question checkpoint publisher installed during startup.
func (s *Server) restoreTaskRuntimes() {
@@ -3567,6 +3575,7 @@ func (s *Server) settingsPayload() map[string]any {
"global_proxy": maskSecrets(s.m.GlobalProxy()), // 全局出口代理(http/https/socks5),所有目标流量走它,空=直连
"proxy_pool": func() []ProxyPoolEntry { e, _, _ := s.m.ProxyPoolSnapshot(); return e }(),
"proxy_pool_active_id": func() string { _, a, _ := s.m.ProxyPoolSnapshot(); return a }(),
"proxy_pool_switch_log": func() []ProxySwitchLog { _, _, l := s.m.ProxyPoolSnapshot(); return l }(),
"proxy_rotate_timed_enabled": s.m.pg.GetBool(settingProxyRotateTimed, false),
"proxy_rotate_interval_seconds": proxyIntSetting(s.m.pg, settingProxyRotateInterval, defaultProxyRotateInterval),
"proxy_rotate_on_failure": s.m.pg.GetBool(settingProxyRotateFailure, false),
@@ -3900,8 +3909,7 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
writeErr(w, 400, err.Error())
return
}
changed = true // capture-off egress is baked into agents at build time → rebuild
if globalProxyValue == "" {
if globalProxyValue == "" && (s.m.proxyPool == nil || len(s.m.proxyPoolEntries()) == 0) {
s.proxyReconnectSuccessWithGuards(proxyGuards, nil)
}
}
+16 -4
View File
@@ -134,6 +134,7 @@ export default function SystemSettingsPage() {
setProxyPoolThreshold(String(s.proxy_health_fail_threshold ?? 3));
setProxyPoolCooldown(String(s.proxy_health_cooldown_seconds ?? 60));
setProxyPoolHealthInterval(String(s.proxy_health_check_interval_seconds ?? 30));
setProxyPoolLogs(s.proxy_pool_switch_log ?? []);
setProxyReconnectMax(String(s.proxy_reconnect_max_attempts ?? 5));
setProxyReconnectInterval(String(s.proxy_reconnect_interval_seconds ?? 30));
setPyInterp(s.python_interpreter ?? "");
@@ -226,6 +227,11 @@ export default function SystemSettingsPage() {
.then((s) => {
apply(s);
setSettingsLoaded(true);
void api.proxyPool().then((pool) => {
setProxyPool(pool.entries);
setProxyPoolActiveId(pool.active_id);
setProxyPoolLogs(pool.switch_log);
}).catch(() => undefined);
})
.catch(() => {
setSettingsLoaded(false);
@@ -457,12 +463,18 @@ export default function SystemSettingsPage() {
api.importProxyPool({ text: proxyPoolText }).then((r) => { setProxyPool(r.entries); setProxyPoolActiveId(r.active_id); setProxyPoolLogs(r.switch_log); setProxyPoolText(""); if (r.errors.length) toast.error(r.errors.map((e) => `第${e.line}行:${e.error}`).join(";")); else toast.success("代理池已导入"); }).catch((e) => toast.error("导入失败:" + (e as Error).message));
};
const saveProxyPoolStrategy = () => {
const interval = Number(proxyPoolInterval);
if (!Number.isFinite(interval) || !Number.isInteger(interval) || interval < 3) {
toast.error("定时轮换间隔必须是大于等于 3 秒的整数");
return;
}
api.setSettings({ proxy_rotate_timed_enabled: proxyPoolTimed, proxy_rotate_interval_seconds: Number(proxyPoolInterval), proxy_rotate_on_failure: proxyPoolFailure, proxy_rotate_strategy: proxyPoolStrategy, proxy_health_fail_threshold: Number(proxyPoolThreshold), proxy_health_cooldown_seconds: Number(proxyPoolCooldown), proxy_health_check_interval_seconds: Number(proxyPoolHealthInterval) }).then((s) => { apply(s); toast.success("代理池轮换策略已保存"); }).catch((e) => toast.error("保存失败:" + (e as Error).message));
};
const testAllProxyPool = () => api.testAllProxyPool().then((r) => { setProxyPool(r.entries); setProxyPoolActiveId(r.active_id); setProxyPoolLogs(r.switch_log); toast.success("已完成全部代理探测"); }).catch((e) => toast.error("探测失败:" + (e as Error).message));
const testOneProxyPool = (id: string) => api.testProxyPoolEntry(id).then(() => api.proxyPool()).then((r) => { setProxyPool(r.entries); setProxyPoolActiveId(r.active_id); setProxyPoolLogs(r.switch_log); }).catch((e) => toast.error("探测失败:" + (e as Error).message));
const toggleProxyPoolEntry = (entry: ProxyPoolEntry) => api.updateProxyPool(entry.id, { enabled: !entry.enabled }).then((r) => setProxyPool((items) => items.map((x) => x.id === entry.id ? r.entry : x))).catch((e) => toast.error("更新失败:" + (e as Error).message));
const deleteProxyPoolEntry = (id: string) => api.deleteProxyPool(id).then(() => setProxyPool((items) => items.filter((x) => x.id !== id))).catch((e) => toast.error("删除失败:" + (e as Error).message));
const refreshProxyPool = () => api.proxyPool().then((r) => { setProxyPool(r.entries); setProxyPoolActiveId(r.active_id); setProxyPoolLogs(r.switch_log); });
const toggleProxyPoolEntry = (entry: ProxyPoolEntry) => api.updateProxyPool(entry.id, { enabled: !entry.enabled }).then(() => refreshProxyPool()).catch((e) => toast.error("更新失败:" + (e as Error).message));
const deleteProxyPoolEntry = (id: string) => api.deleteProxyPool(id).then(() => refreshProxyPool()).catch((e) => toast.error("删除失败:" + (e as Error).message));
const saveProxyReconnect = () => {
api.setSettings({ proxy_reconnect_max_attempts: Number(proxyReconnectMax), proxy_reconnect_interval_seconds: Number(proxyReconnectInterval) })
@@ -925,9 +937,9 @@ export default function SystemSettingsPage() {
{proxyPool.length === 0 ? <p className="text-muted-foreground text-xs">暂无代理池条目;单代理仍由上面的全局代理输入框兼容。</p> : proxyPool.map((entry) => (
<div key={entry.id} className="flex flex-wrap items-center gap-2 rounded border px-2 py-2 text-xs">
<Switch checked={entry.enabled} onCheckedChange={() => toggleProxyPoolEntry(entry)} aria-label="启用代理" />
<span className="min-w-0 flex-1 font-mono">{entry.name || entry.url}</span>
<span className="min-w-0 flex-1 font-mono">{entry.name ? `${entry.name} · ` : ""}{entry.url}{entry.id === proxyPoolActiveId ? " · 当前" : ""}</span>
<span className={entry.health === "healthy" ? "text-emerald-600" : entry.health === "disabled" ? "text-muted-foreground" : "text-amber-600"}>{entry.enabled ? entry.health : "disabled"}</span>
<span className="text-muted-foreground">{entry.egress_ip ?? "—"} · {entry.latency_ms ?? "—"}ms · 失败 {entry.consecutive_failures}</span>
<span className="text-muted-foreground">{entry.egress_ip ?? "—"} · {entry.latency_ms ?? "—"}ms · 失败 {entry.consecutive_failures}{entry.last_error ? ` · ${entry.last_error}` : ""}</span>
<Button type="button" variant="ghost" size="sm" onClick={() => testOneProxyPool(entry.id)}>测试</Button>
<Button type="button" variant="ghost" size="sm" onClick={() => deleteProxyPoolEntry(entry.id)}>删除</Button>
</div>
+1
View File
@@ -1006,6 +1006,7 @@ export interface Settings {
global_proxy?: string;
proxy_pool?: ProxyPoolEntry[];
proxy_pool_active_id?: string;
proxy_pool_switch_log?: ProxySwitchLog[];
proxy_rotate_timed_enabled?: boolean;
proxy_rotate_interval_seconds?: number;
proxy_rotate_on_failure?: boolean;