feat: 全局代理池(健康检查、定时/失效轮换可同时开启、热切换与脱敏)

This commit is contained in:
RuoJi6 committed 2026-11-01 19:37:18 +08:00
1 parent 008a9ced0d
commit 0bb954e0a4
18 files changed
+1167 -53

No files matched your search

+9 -3
View File
@@ -29,6 +29,7 @@ type ChatAgent struct {
tx *transcript.Store
window int
proxyAddr string
proxyAddrFn func() string
proxyCACert string
webSearch WebSearchOpts
guard *guard.Guard // optional; nil disables intercept hooks for chat
@@ -65,7 +66,8 @@ func (c *ChatAgent) maxTokens() int {
// SetProxy points the chat agent's WebFetch/Bash at the recording proxy plus the
// CA cert it trusts (empty addr = direct). Kept for parity with the other agents.
func (c *ChatAgent) SetProxy(addr, caCert string) { c.proxyAddr, c.proxyCACert = addr, caCert }
func (c *ChatAgent) SetProxy(addr, caCert string) { c.proxyAddr, c.proxyCACert = addr, caCert }
func (c *ChatAgent) SetProxyProvider(fn func() string) { c.proxyAddrFn = fn }
// SetWebSearch selects the web_search backend for the chat agent (off by default).
func (c *ChatAgent) SetWebSearch(o WebSearchOpts) { c.webSearch = o }
@@ -124,6 +126,10 @@ func (c *ChatAgent) Chat(ctx context.Context, agentKey, sessionID, message strin
defer cleanup()
system, boundary := deferredSystem(chatSystem(agentKey, c.workDir, sessionWorkDir), def)
proxyAddr := c.proxyAddr
if c.proxyAddrFn != nil {
proxyAddr = c.proxyAddrFn()
}
opts := agentcore.Options{
Provider: c.prov,
SystemPrompt: system,
@@ -133,7 +139,7 @@ func (c *ChatAgent) Chat(ctx context.Context, agentKey, sessionID, message strin
UnlockSet: def.Unlock,
PermissionMode: permission.ModeBypass,
EnableWebFetch: true, // 走记录代理留痕;载入代理 CA 验证 MITM 重签的 HTTPS 证书
WebFetchProxy: c.proxyAddr,
WebFetchProxy: proxyAddr,
WebFetchCACert: c.proxyCACert,
// 联网搜索(可选)。ddgs 无需 key;brave-free 需 BraveKey;tavily 需 TavilyKey。
// WebSearchProxy 是独立出口代理(http/https/socks5),与记录流量的 MITM 代理无关;空则直连。
@@ -145,7 +151,7 @@ func (c *ChatAgent) Chat(ctx context.Context, agentKey, sessionID, message strin
DeepSeekSearchAPIKey: ws.DeepSeekAPIKey,
DeepSeekSearchModel: ws.DeepSeekModel,
WebSearchProxy: ws.Proxy,
BashEnv: proxyEnv(c.proxyAddr, c.proxyCACert), // Bash 子命令默认走代理+信任 CA
BashEnv: proxyEnv(proxyAddr, c.proxyCACert), // Bash 子命令默认走代理+信任 CA
WorkingDir: sessionWorkDir,
MaxTurns: maxTurns,
MaxDuration: maxDuration,
+15 -9
View File
@@ -21,11 +21,12 @@ type MainAgent struct {
findingRecorder FindingRecorder
prov llm.Provider
model string
tx *transcript.Store // raw LLM conversation persistence (nil = off)
window int // context window in tokens (for compaction)
windowFn func() int // optional dynamic task-chain minimum
maxTurns int // max agent turns per run (0 = unlimited)
proxyAddr string // recording proxy for WebFetch (empty = direct)
tx *transcript.Store // raw LLM conversation persistence (nil = off)
window int // context window in tokens (for compaction)
windowFn func() int // optional dynamic task-chain minimum
maxTurns int // max agent turns per run (0 = unlimited)
proxyAddr string // recording proxy for WebFetch (empty = direct)
proxyAddrFn func() string
proxyCACert string // recording proxy's CA cert path (HTTPS verify)
webSearch WebSearchOpts // web_search tool backend selection (off by default)
workDir string // shared work dir (surfaced in prompt as artifact-output target)
@@ -72,7 +73,8 @@ func (m *MainAgent) compactionWindow() int {
// SetProxy points the main agent's WebFetch at the recording proxy plus the CA
// cert it trusts to verify HTTPS through it (empty addr = direct).
func (m *MainAgent) SetProxy(addr, caCert string) { m.proxyAddr, m.proxyCACert = addr, caCert }
func (m *MainAgent) SetProxy(addr, caCert string) { m.proxyAddr, m.proxyCACert = addr, caCert }
func (m *MainAgent) SetProxyProvider(fn func() string) { m.proxyAddrFn = fn }
// SetWebSearch selects the web_search backend for the main agent (off by default).
func (m *MainAgent) SetWebSearch(o WebSearchOpts) { m.webSearch = o }
@@ -132,6 +134,10 @@ func (m *MainAgent) Chat(ctx context.Context, taskID int64, mainSeg int, as *db.
mainDir := ensureRunDir(m.workDir, taskID, 0)
ctx = intercept.WithReviewWorkingDirectory(ctx, mainDir)
system, boundary := deferredSystem(mainAgentSystem(goal, m.workDir, mainDir), def)
proxyAddr := m.proxyAddr
if m.proxyAddrFn != nil {
proxyAddr = m.proxyAddrFn()
}
opts := agentcore.Options{
Provider: m.prov,
SystemPrompt: system,
@@ -141,7 +147,7 @@ func (m *MainAgent) Chat(ctx context.Context, taskID int64, mainSeg int, as *db.
UnlockSet: def.Unlock,
PermissionMode: permission.ModeBypass,
EnableWebFetch: true, // 走记录代理留痕;载入代理 CA 验证 MITM 重签的 HTTPS 证书
WebFetchProxy: m.proxyAddr,
WebFetchProxy: proxyAddr,
WebFetchCACert: m.proxyCACert,
// 联网搜索(可选)。ddgs 无需 key;brave-free 需 BraveKey;tavily 需 TavilyKey。
// WebSearchProxy 是独立出口代理(http/https/socks5),与记录流量的 MITM 代理无关;空则直连。
@@ -153,8 +159,8 @@ func (m *MainAgent) Chat(ctx context.Context, taskID int64, mainSeg int, as *db.
DeepSeekSearchAPIKey: m.webSearch.DeepSeekAPIKey,
DeepSeekSearchModel: m.webSearch.DeepSeekModel,
WebSearchProxy: m.webSearch.Proxy,
BashEnv: proxyEnv(m.proxyAddr, m.proxyCACert), // Bash 子命令默认走代理+信任 CA
WorkingDir: mainDir, // 本任务工作目录 <workDir>/tasks/<taskID>
BashEnv: proxyEnv(proxyAddr, m.proxyCACert), // Bash 子命令默认走代理+信任 CA
WorkingDir: mainDir, // 本任务工作目录 <workDir>/tasks/<taskID>
ToolOutputDir: cmdOutDir(mainDir),
MaxTurns: m.maxTurns, // 0 = unlimited (configurable in agent management)
Compaction: compactionConfig(m.compactionWindow()), // long chats stay within the window
+18 -12
View File
@@ -31,14 +31,15 @@ type Planner struct {
killWork func(intentID int64) error // engine callback to terminate a running work (nil = off)
steerWork func(intentID int64, msg string) error // engine callback to steer a running work mid-run (nil = off)
proxyAddr string // recording proxy for WebFetch (empty = direct)
proxyCACert string // recording proxy's CA cert path (HTTPS verify)
webSearch WebSearchOpts // web_search tool backend selection (off by default)
workDir string // shared work dir (surfaced in prompt as artifact-output target)
injectConstraints func() bool // resolver: inject task operation constraints into system prompt? (nil = yes)
nonStreamingFn func() bool // resolver: use non-streaming (Complete) path? (nil = streaming)
noaEnabledFn func() bool // resolver: use experimental noa compaction? (nil = off)
maxTokensFn func() int // resolver: per-reply output cap (nil/0 = send no cap)
compactor *Compactor // cold-node compaction (§7); nil = disabled
proxyAddrFn func() string
proxyCACert string // recording proxy's CA cert path (HTTPS verify)
webSearch WebSearchOpts // web_search tool backend selection (off by default)
workDir string // shared work dir (surfaced in prompt as artifact-output target)
injectConstraints func() bool // resolver: inject task operation constraints into system prompt? (nil = yes)
nonStreamingFn func() bool // resolver: use non-streaming (Complete) path? (nil = streaming)
noaEnabledFn func() bool // resolver: use experimental noa compaction? (nil = off)
maxTokensFn func() int // resolver: per-reply output cap (nil/0 = send no cap)
compactor *Compactor // cold-node compaction (§7); nil = disabled
// todos keeps ONE plan-scratchpad per task (keyed by exploration id) so the
// planner's multi-step plan survives across wake-ups — each Plan() is a fresh
@@ -90,7 +91,8 @@ func (p *Planner) compactionWindow() int {
// SetProxy points the planner's WebFetch at the recording proxy plus the CA cert
// it trusts to verify HTTPS through it (empty addr = direct).
func (p *Planner) SetProxy(addr, caCert string) { p.proxyAddr, p.proxyCACert = addr, caCert }
func (p *Planner) SetProxy(addr, caCert string) { p.proxyAddr, p.proxyCACert = addr, caCert }
func (p *Planner) SetProxyProvider(fn func() string) { p.proxyAddrFn = fn }
// SetWebSearch selects the web_search backend for the planner (off by default).
func (p *Planner) SetWebSearch(o WebSearchOpts) { p.webSearch = o }
@@ -402,6 +404,10 @@ func (p *Planner) Plan(ctx context.Context, taskID int64, as *db.AssetStore, ts
if tc.DeadlineUnix > 0 {
settle = wrapupSettlementForTask("planner", nil, clamped)
}
proxyAddr := p.proxyAddr
if p.proxyAddrFn != nil {
proxyAddr = p.proxyAddrFn()
}
opts := agentcore.Options{
Provider: p.prov,
SystemPrompt: system,
@@ -411,7 +417,7 @@ func (p *Planner) Plan(ctx context.Context, taskID int64, as *db.AssetStore, ts
UnlockSet: def.Unlock,
PermissionMode: permission.ModeBypass,
EnableWebFetch: true, // 走记录代理留痕;载入代理 CA 验证 MITM 重签的 HTTPS 证书
WebFetchProxy: p.proxyAddr,
WebFetchProxy: proxyAddr,
WebFetchCACert: p.proxyCACert,
// 联网搜索(可选)。ddgs 无需 key;brave-free 需 BraveKey;tavily 需 TavilyKey。
// WebSearchProxy 是独立出口代理(http/https/socks5),与记录流量的 MITM 代理无关;空则直连。
@@ -423,8 +429,8 @@ func (p *Planner) Plan(ctx context.Context, taskID int64, as *db.AssetStore, ts
DeepSeekSearchAPIKey: p.webSearch.DeepSeekAPIKey,
DeepSeekSearchModel: p.webSearch.DeepSeekModel,
WebSearchProxy: p.webSearch.Proxy,
BashEnv: proxyEnv(p.proxyAddr, p.proxyCACert), // Bash 子命令默认走代理+信任 CA
WorkingDir: taskDir, // 本任务工作目录 <workDir>/tasks/<taskID>
BashEnv: proxyEnv(proxyAddr, p.proxyCACert), // Bash 子命令默认走代理+信任 CA
WorkingDir: taskDir, // 本任务工作目录 <workDir>/tasks/<taskID>
ToolOutputDir: cmdOutDir(taskDir),
MaxTurns: p.maxTurns, // 0 = unlimited (configurable in agent management)
MaxDuration: maxDur, // 0=不限;有 deadline 时=距 deadline 剩余
+9 -2
View File
@@ -57,6 +57,7 @@ type Worker struct {
model string
workDir string
proxyAddr string
proxyAddrFn func() string
proxyCACert string // recording proxy's CA cert path (for WebFetch HTTPS verify)
webSearch WebSearchOpts // web_search tool backend selection (off by default)
tx *transcript.Store // raw LLM conversation persistence (nil = off)
@@ -188,6 +189,8 @@ func (w *Worker) compactionWindow() int {
// that MITM proxy. Empty addr disables the hint.
func (w *Worker) SetProxy(addr, caCert string) { w.proxyAddr, w.proxyCACert = addr, caCert }
func (w *Worker) SetProxyProvider(fn func() string) { w.proxyAddrFn = fn }
// SetWebSearch selects the web_search backend for this worker (off by default).
func (w *Worker) SetWebSearch(o WebSearchOpts) { w.webSearch = o }
@@ -426,6 +429,10 @@ 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,
@@ -437,7 +444,7 @@ func (w *Worker) execute(ctx context.Context, name string, taskID int64, as *db.
// WebFetch 走记录代理,其 HTTP 与 curl 一样被留痕;载入代理 CA 让经 MITM
// 重签的 HTTPS 证书能【正常校验通过】(而非关掉校验)。proxy 空则直连。
EnableWebFetch: true,
WebFetchProxy: w.proxyAddr,
WebFetchProxy: proxyAddr,
WebFetchCACert: w.proxyCACert,
// 联网搜索(可选)。ddgs 无需 key;brave-free 需 BraveKey;tavily 需 TavilyKey。
// WebSearchProxy 是独立的出口代理(http/https/socks5),与记录流量的 MITM 代理无关;空则直连。
@@ -450,7 +457,7 @@ func (w *Worker) execute(ctx context.Context, name string, taskID int64, as *db.
DeepSeekSearchModel: w.webSearch.DeepSeekModel,
WebSearchProxy: w.webSearch.Proxy,
// Bash 子命令的 HTTP 默认走记录代理 + 信任其 CA(工具无需 -x/-k)。
BashEnv: proxyEnv(w.proxyAddr, w.proxyCACert),
BashEnv: proxyEnv(proxyAddr, w.proxyCACert),
WorkingDir: runDir,
MaxTurns: w.maxTurns, // 0 = unlimited (configurable in agent management)
// 墙钟预算,轮边界判,不打断半路;0 = 不限。有任务级 deadline 时夹逼到 min(自身预算,
+3 -3
View File
@@ -368,7 +368,7 @@ func (s *Server) runScriptTool(ctx context.Context, key string, execRaw json.Raw
}
body, err := execPython(ctx, interp, key, spec.Code, params, workDir, sessionEnv, timeoutOr(spec.TimeoutMs, 120000))
if err != nil {
return actool.Errorf(err.Error()), nil
return actool.Errorf(maskSecrets(err.Error())), nil
}
return actool.Text(actool.Capture(tc, body)), nil
}
@@ -436,7 +436,7 @@ func (s *Server) runHTTPTool(ctx context.Context, execRaw json.RawMessage, param
defer cancel()
req, err := http.NewRequestWithContext(runCtx, method, rawURL, bodyReader)
if err != nil {
return actool.Errorf(err.Error()), nil
return actool.Errorf(maskSecrets(err.Error())), nil
}
for k, v := range spec.Headers {
req.Header.Set(k, renderTemplate(v, params, identity))
@@ -450,7 +450,7 @@ func (s *Server) runHTTPTool(ctx context.Context, execRaw json.RawMessage, param
if spec.UseRecordingProxy || strings.TrimSpace(spec.Proxy) == strings.TrimSpace(s.m.GlobalProxy()) {
s.handleProxyToolFailure(runCtx, err)
}
return actool.Errorf("请求失败: " + err.Error()), nil
return actool.Errorf("请求失败: " + maskSecrets(err.Error())), nil
}
defer resp.Body.Close()
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
+6 -4
View File
@@ -366,16 +366,18 @@ func (s *Server) testEgressProxy(w http.ResponseWriter, r *http.Request) {
agentCA := s.m.ProxyCACert() // 捕获开时是 MITM 的 CA 路径;关时为空
savedProxy := s.m.GlobalProxy() // 已保存的 global_proxy
// 表单值 ≠ 已保存值时,② 仍按表单值测,但 agent 走的是已保存值——必须提示。
// 归一化再比:尾斜杠不算"未保存"。
pending := normalizeProxy(req.GlobalProxy) != normalizeProxy(savedProxy)
// 整体上限 40s:单次 Timeout 只兜一次,不兜总量;前端 fetch 没有超时。
ctx, cancel := context.WithTimeout(r.Context(), 40*time.Second)
defer cancel()
proxyGuards := s.snapshotProxyReconnectGuards(false)
configuredProxy := strings.TrimSpace(req.GlobalProxy)
if strings.Contains(configuredProxy, "***") || strings.Contains(configuredProxy, "******") {
configuredProxy = s.m.GlobalProxy()
}
// 表单值 ≠ 已保存值时,② 仍按表单值测,但 agent 走的是已保存值——必须提示。
// 归一化再比:尾斜杠不算"未保存"。
pending := normalizeProxy(configuredProxy) != normalizeProxy(savedProxy)
rep := egressReport{
Capture: capture,
Configured: maskSecrets(configuredProxy),
+39
View File
@@ -283,6 +283,7 @@ type Manager struct {
// 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
}
// SetTaskControlMutex lets lifecycle operations that originate in Manager
@@ -642,6 +643,7 @@ func NewManager(dir, proxyAddr string) (*Manager, error) {
if v, ok, _ := pg.GetSetting(settingGlobalProxy); ok {
m.globalProxy = strings.TrimSpace(v)
}
m.initProxyPool()
if m.traffic != nil {
if err := m.traffic.SetUpstreamProxy(m.globalProxy); err != nil {
log.Printf("[proxy] 全局代理 %q 无效,已忽略: %v", m.globalProxy, err)
@@ -1050,6 +1052,16 @@ func (m *Manager) SetGlobalProxy(raw string) error {
return err
}
}
poolLocked := false
if m.proxyPool != nil {
m.proxyPool.mu.Lock()
poolLocked = true
defer func() {
if poolLocked {
m.proxyPool.mu.Unlock()
}
}()
}
if err := m.pg.SetSetting(settingGlobalProxy, raw); err != nil {
return err
}
@@ -1063,6 +1075,33 @@ func (m *Manager) SetGlobalProxy(raw string) error {
}
// 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
}
+7 -1
View File
@@ -182,6 +182,9 @@ func (s *Server) pauseProxyTask(taskID string, cause error) {
}
func (s *Server) handleProxyTrafficFailure(cause error) {
if s.handleProxyPoolFailure(cause) {
return
}
if !isGlobalProxyDialFailure(cause) || strings.TrimSpace(s.m.GlobalProxy()) == "" {
return
}
@@ -200,6 +203,9 @@ func (s *Server) handleProxyTrafficFailure(cause error) {
}
func (s *Server) handleProxyToolFailure(ctx context.Context, cause error) {
if s.handleProxyPoolFailure(cause) {
return
}
if !isGlobalProxyDialFailure(cause) || strings.TrimSpace(s.m.GlobalProxy()) == "" {
return
}
@@ -215,7 +221,7 @@ func truncateProxyError(err error) string {
if err == nil {
return ""
}
s := err.Error()
s := maskSecrets(err.Error())
if len(s) > 240 {
return s[:240] + "…"
}
+782
View File
@@ -0,0 +1,782 @@
package server
import (
"context"
"encoding/json"
"errors"
"fmt"
"math/rand"
"net/http"
"strings"
"sync"
"time"
"github.com/Autumn-27/artex/traffic"
)
const (
settingProxyPool = "proxy_pool_entries"
settingProxyPoolActive = "proxy_pool_active_id"
settingProxyRotateTimed = "proxy_rotate_timed_enabled"
settingProxyRotateInterval = "proxy_rotate_interval_seconds"
settingProxyRotateFailure = "proxy_rotate_on_failure"
settingProxyRotateStrategy = "proxy_rotate_strategy"
settingProxyHealthThreshold = "proxy_health_fail_threshold"
settingProxyHealthCooldown = "proxy_health_cooldown_seconds"
settingProxyHealthInterval = "proxy_health_check_interval_seconds"
defaultProxyRotateInterval = 30
minProxyRotateInterval = 3
defaultProxyHealthThreshold = 3
defaultProxyHealthCooldown = 60
defaultProxyHealthInterval = 30
proxyPoolProbeTimeout = 10 * time.Second
proxyPoolMaxProbeConcurrency = 4
)
// ProxyPoolEntry is the public shape. URL is always masked in API responses.
type ProxyPoolEntry struct {
ID string `json:"id"`
URL string `json:"url"`
Name string `json:"name,omitempty"`
Remark string `json:"remark,omitempty"`
Enabled bool `json:"enabled"`
EgressIP string `json:"egress_ip,omitempty"`
LatencyMS int64 `json:"latency_ms,omitempty"`
LastError string `json:"last_error,omitempty"`
LastCheckedAt time.Time `json:"last_checked_at,omitempty"`
ConsecutiveFailures int `json:"consecutive_failures"`
Health string `json:"health"`
}
type proxyPoolEntry struct {
ProxyPoolEntry
RawURL string `json:"raw_url"`
}
type ProxySwitchLog struct {
At time.Time `json:"at"`
From string `json:"from"`
To string `json:"to"`
Reason string `json:"reason"`
}
type proxyPoolPersist struct {
Entries []proxyPoolEntry `json:"entries"`
Active string `json:"active"`
}
type ProxyImportError struct {
Line int `json:"line"`
Input string `json:"input"`
Error string `json:"error"`
}
// ParseProxyBatch accepts one proxy URL per line. Optional name and remark may
// follow as URL|name|remark. Duplicates are reported once and skipped.
func ParseProxyBatch(raw string) ([]ProxyPoolEntry, []ProxyImportError) {
var out []ProxyPoolEntry
var errs []ProxyImportError
seen := map[string]bool{}
for n, line := range strings.Split(raw, "\n") {
line = strings.TrimSpace(line)
if line == "" || strings.HasPrefix(line, "#") {
continue
}
parts := strings.SplitN(line, "|", 3)
rawURL := strings.TrimSpace(parts[0])
u, err := traffic.ValidateProxyURL(rawURL)
if err != nil {
errs = append(errs, ProxyImportError{Line: n + 1, Input: maskSecrets(line), Error: maskSecrets(err.Error())})
continue
}
key := strings.ToLower(u.String())
if seen[key] {
errs = append(errs, ProxyImportError{Line: n + 1, Input: maskSecrets(line), Error: "重复代理,已跳过"})
continue
}
seen[key] = true
name, remark := "", ""
if len(parts) > 1 {
name = strings.TrimSpace(parts[1])
}
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"})
}
return out, errs
}
type proxyPoolState struct {
mu sync.Mutex
entries []proxyPoolEntry
active string
logs []ProxySwitchLog
lastTimed time.Time
}
func (m *Manager) initProxyPool() {
m.proxyPool = &proxyPoolState{}
if raw, ok, _ := m.pg.GetSetting(settingProxyPool); ok && strings.TrimSpace(raw) != "" {
var p proxyPoolPersist
if json.Unmarshal([]byte(raw), &p) == nil {
m.proxyPool.entries, m.proxyPool.active = p.Entries, p.Active
}
}
if m.proxyPool.active == "" {
if v, ok, _ := m.pg.GetSetting(settingProxyPoolActive); ok {
m.proxyPool.active = strings.TrimSpace(v)
}
}
if len(m.proxyPool.entries) == 0 && strings.TrimSpace(m.globalProxy) != "" {
m.proxyPool.entries = []proxyPoolEntry{{ProxyPoolEntry: ProxyPoolEntry{ID: "proxy-legacy", URL: maskSecrets(m.globalProxy), Enabled: true, Health: "healthy"}, RawURL: m.globalProxy}}
m.proxyPool.active = "proxy-legacy"
_ = m.persistProxyPoolLocked()
}
if m.proxyPool.active != "" {
for _, e := range m.proxyPool.entries {
if e.ID == m.proxyPool.active {
m.globalProxy = e.RawURL
break
}
}
}
}
func (m *Manager) persistProxyPoolLocked() error {
if m.pg == nil {
return nil
}
b, err := json.Marshal(proxyPoolPersist{Entries: m.proxyPool.entries, Active: m.proxyPool.active})
if err != nil {
return err
}
return m.pg.SetSetting(settingProxyPool, string(b))
}
func (m *Manager) ProxyPoolSnapshot() ([]ProxyPoolEntry, string, []ProxySwitchLog) {
if m.proxyPool == nil {
return nil, "", nil
}
m.proxyPool.mu.Lock()
defer m.proxyPool.mu.Unlock()
entries := make([]ProxyPoolEntry, 0, len(m.proxyPool.entries))
for _, e := range m.proxyPool.entries {
p := e.ProxyPoolEntry
p.URL = maskSecrets(e.RawURL)
p.LastError = maskSecrets(p.LastError)
if !p.Enabled {
p.Health = "disabled"
}
entries = append(entries, p)
}
logs := append([]ProxySwitchLog(nil), m.proxyPool.logs...)
for i := range logs {
logs[i].From, logs[i].To = maskSecrets(logs[i].From), maskSecrets(logs[i].To)
}
return entries, m.proxyPool.active, logs
}
func (m *Manager) ActiveProxy() string { return m.GlobalProxy() }
func (m *Manager) proxyPoolEntries() []proxyPoolEntry {
if m.proxyPool == nil {
return nil
}
m.proxyPool.mu.Lock()
defer m.proxyPool.mu.Unlock()
return append([]proxyPoolEntry(nil), m.proxyPool.entries...)
}
func (m *Manager) setProxyPoolEntries(entries []proxyPoolEntry, active string) error {
if m.proxyPool == nil {
m.proxyPool = &proxyPoolState{}
}
m.proxyPool.mu.Lock()
defer m.proxyPool.mu.Unlock()
m.proxyPool.entries, m.proxyPool.active = entries, active
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 == "" {
return nil
}
for _, e := range entries {
if e.ID == active && e.Enabled && e.Health == "healthy" {
return m.applyActiveProxyLocked(active, "manual")
}
}
return nil
}
func (m *Manager) applyActiveProxyLocked(id, reason string) error {
var next string
for _, e := range m.proxyPool.entries {
if e.ID == id && e.Enabled && e.Health == "healthy" {
next = e.RawURL
break
}
}
if id == "" {
next = ""
}
if id != "" && next == "" {
return errors.New("代理未启用或不健康")
}
m.mu.RLock()
from := m.globalProxy
m.mu.RUnlock()
if m.traffic != nil {
if err := m.traffic.SetUpstreamProxy(next); err != nil {
return err
}
}
m.mu.Lock()
m.globalProxy = next
m.mu.Unlock()
m.proxyPool.active = id
if m.pg != nil {
_ = m.pg.SetSetting(settingGlobalProxy, next)
_ = m.pg.SetSetting(settingProxyPoolActive, id)
}
if from != next {
m.proxyPool.logs = append(m.proxyPool.logs, ProxySwitchLog{At: time.Now(), From: maskSecrets(from), To: maskSecrets(next), Reason: reason})
if len(m.proxyPool.logs) > 20 {
m.proxyPool.logs = m.proxyPool.logs[len(m.proxyPool.logs)-20:]
}
}
// Browser MCP reads the current capture/upstream route on each reconciliation.
if m.pg != nil {
m.syncBrowserMCPProxy()
}
return m.persistProxyPoolLocked()
}
func (m *Manager) selectHealthyLocked(exclude string) string {
strategy := "sequential"
if m.pg != nil {
if v, ok, _ := m.pg.GetSetting(settingProxyRotateStrategy); ok && v == "random" {
strategy = v
}
}
pool := make([]string, 0)
for _, e := range m.proxyPool.entries {
if e.Enabled && e.Health == "healthy" && e.ID != exclude {
pool = append(pool, e.ID)
}
}
if len(pool) == 0 && exclude != "" {
for _, e := range m.proxyPool.entries {
if e.Enabled && e.Health == "healthy" {
pool = append(pool, e.ID)
}
}
}
if len(pool) == 0 {
return ""
}
if strategy == "random" {
return pool[rand.Intn(len(pool))]
}
start := 0
for i, e := range m.proxyPool.entries {
if e.ID == exclude {
start = i + 1
break
}
}
for n := 0; n < len(m.proxyPool.entries); n++ {
e := m.proxyPool.entries[(start+n)%len(m.proxyPool.entries)]
if e.Enabled && e.Health == "healthy" && e.ID != exclude {
return e.ID
}
}
return pool[0]
}
func (m *Manager) rotateProxy(reason string) bool {
if m.proxyPool == nil {
return false
}
m.proxyPool.mu.Lock()
defer m.proxyPool.mu.Unlock()
id := m.selectHealthyLocked(m.proxyPool.active)
if id == "" || id == m.proxyPool.active {
return false
}
return m.applyActiveProxyLocked(id, reason) == nil
}
func (m *Manager) markProxyProbe(id string, res egressProbeResult) {
if m.proxyPool == nil {
return
}
m.proxyPool.mu.Lock()
defer m.proxyPool.mu.Unlock()
for i := range m.proxyPool.entries {
e := &m.proxyPool.entries[i]
if e.ID != id {
continue
}
e.LastCheckedAt, e.LatencyMS, e.EgressIP = time.Now(), res.LatencyMS, res.ExitIP
if res.OK {
e.ConsecutiveFailures, e.LastError, e.Health = 0, "", "healthy"
} else {
e.ConsecutiveFailures++
e.LastError = maskSecrets(res.Error)
threshold := proxyIntSetting(m.pg, settingProxyHealthThreshold, defaultProxyHealthThreshold)
if e.ConsecutiveFailures >= threshold {
e.Health = "unhealthy-cooling"
}
}
_ = m.persistProxyPoolLocked()
return
}
}
func proxyIntSetting(pg interface {
GetSetting(string) (string, bool, error)
}, key string, def int) int {
if pg == nil {
return def
}
if v, ok, _ := pg.GetSetting(key); ok {
var n int
if _, err := fmt.Sscanf(strings.TrimSpace(v), "%d", &n); err == nil && n > 0 {
return n
}
}
return def
}
func (m *Manager) proxyHealthConfig() (threshold int, cooldown, interval time.Duration) {
if m.pg == nil {
return defaultProxyHealthThreshold, defaultProxyHealthCooldown * time.Second, defaultProxyHealthInterval * time.Second
}
return proxyIntSetting(m.pg, settingProxyHealthThreshold, defaultProxyHealthThreshold), time.Duration(proxyIntSetting(m.pg, settingProxyHealthCooldown, defaultProxyHealthCooldown)) * time.Second, time.Duration(proxyIntSetting(m.pg, settingProxyHealthInterval, defaultProxyHealthInterval)) * time.Second
}
func (m *Manager) SetProxyPoolSettings(timed bool, interval int, onFailure bool, strategy string, threshold, cooldown, healthInterval int) error {
if interval < minProxyRotateInterval {
return fmt.Errorf("定时轮换间隔不能小于 %d 秒", minProxyRotateInterval)
}
if strategy != "sequential" && strategy != "random" {
return errors.New("proxy_rotate_strategy 必须是 sequential 或 random")
}
if threshold < 1 || cooldown < 1 || healthInterval < 1 {
return errors.New("健康检查参数必须大于 0")
}
return m.pg.SetSettingsAtomic(context.Background(), map[string]string{settingProxyRotateTimed: fmt.Sprint(timed), settingProxyRotateInterval: fmt.Sprint(interval), settingProxyRotateFailure: fmt.Sprint(onFailure), settingProxyRotateStrategy: strategy, settingProxyHealthThreshold: fmt.Sprint(threshold), settingProxyHealthCooldown: fmt.Sprint(cooldown), settingProxyHealthInterval: fmt.Sprint(healthInterval)})
}
func (m *Manager) proxyRotateConfig() (bool, time.Duration, bool, string) {
if m.pg == nil {
return false, defaultProxyRotateInterval * time.Second, false, "sequential"
}
return m.pg.GetBool(settingProxyRotateTimed, false), time.Duration(proxyIntSetting(m.pg, settingProxyRotateInterval, defaultProxyRotateInterval)) * time.Second, m.pg.GetBool(settingProxyRotateFailure, false), func() string {
v, ok, _ := m.pg.GetSetting(settingProxyRotateStrategy)
if ok && v == "random" {
return v
}
return "sequential"
}()
}
func (s *Server) runProxyPoolSupervisor(ctx context.Context) {
for {
_, interval, _, _ := s.m.proxyRotateConfig()
_, _, healthInterval := s.m.proxyHealthConfig()
if healthInterval > 0 && healthInterval < interval {
interval = healthInterval
}
if interval < minProxyRotateInterval*time.Second {
interval = minProxyRotateInterval * time.Second
}
t := time.NewTimer(interval)
select {
case <-ctx.Done():
if !t.Stop() {
<-t.C
}
return
case <-t.C:
s.proxyPoolTick()
case <-s.proxyPoolWake:
if !t.Stop() {
<-t.C
}
s.proxyPoolTick()
}
}
}
func (s *Server) proxyPoolTick() {
if s == nil || s.m == nil || s.m.proxyPool == nil {
return
}
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
if due {
s.m.proxyPool.lastTimed = time.Now()
}
s.m.proxyPool.mu.Unlock()
if due {
_ = s.m.rotateProxy("timed")
}
}
threshold, cooldown, _ := s.m.proxyHealthConfig()
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 {
continue
}
e := e
wg.Add(1)
go func() {
defer wg.Done()
select {
case sem <- struct{}{}:
case <-s.ctx.Done():
return
}
defer func() { <-sem }()
res := s.runProxyProbe(s.ctx, e.RawURL, "", defaultEgressProbes, proxyPoolProbeTimeout)
s.m.markProxyProbe(e.ID, res)
}()
}
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()
activeID := s.m.proxyPoolActiveID()
activeHealthy := false
for _, e := range entries {
if e.ID == activeID && e.Enabled && e.Health == "healthy" {
activeHealthy = true
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")
}
}
}
func (m *Manager) activateProxyID(id, reason string) error {
if m.proxyPool == nil {
return errors.New("代理池未初始化")
}
m.proxyPool.mu.Lock()
defer m.proxyPool.mu.Unlock()
if err := m.applyActiveProxyLocked(id, reason); err != nil {
return err
}
return nil
}
func (s *Server) handleProxyPoolFailure(cause error) bool {
if s == nil || s.m == nil || s.m.proxyPool == nil {
return false
}
if !isGlobalProxyDialFailure(cause) || strings.TrimSpace(s.m.GlobalProxy()) == "" {
return false
}
cur := s.m.GlobalProxy()
res := s.runProxyProbe(context.Background(), cur, "", defaultEgressProbes, proxyPoolProbeTimeout)
if res.OK {
return false
}
for _, e := range s.m.proxyPoolEntries() {
if e.RawURL == cur {
s.m.markProxyProbe(e.ID, res)
break
}
}
_, _, onFailure, _ := s.m.proxyRotateConfig()
if onFailure && s.m.rotateProxy("failure") {
return true
}
select {
case s.proxyPoolWake <- struct{}{}:
default:
}
return false
}
func (s *Server) getProxyPool(w http.ResponseWriter, r *http.Request) {
entries, active, logs := s.m.ProxyPoolSnapshot()
timed, interval, onFailure, strategy := s.m.proxyRotateConfig()
threshold, cooldown, healthInterval := s.m.proxyHealthConfig()
writeJSON(w, 200, map[string]any{"entries": entries, "active_id": active, "active_proxy": maskSecrets(s.m.GlobalProxy()), "switch_log": logs, "timed_enabled": timed, "interval_seconds": int(interval / time.Second), "on_failure": onFailure, "strategy": strategy, "health_fail_threshold": threshold, "health_cooldown_seconds": int(cooldown / time.Second), "health_check_interval_seconds": int(healthInterval / time.Second)})
}
func (s *Server) importProxyPool(w http.ResponseWriter, r *http.Request) {
var req struct {
Text string `json:"text"`
TimedEnabled *bool `json:"timed_enabled"`
Interval *int `json:"interval_seconds"`
OnFailure *bool `json:"on_failure"`
Strategy *string `json:"strategy"`
Threshold *int `json:"health_fail_threshold"`
Cooldown *int `json:"health_cooldown_seconds"`
HealthInterval *int `json:"health_check_interval_seconds"`
}
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&req); err != nil {
writeErr(w, 400, err.Error())
return
}
parsed, errs := ParseProxyBatch(req.Text)
entries := s.m.proxyPoolEntries()
seen := map[string]bool{}
for _, e := range entries {
seen[e.RawURL] = true
}
for _, p := range parsed {
if seen[p.URL] {
continue
}
u, _ := traffic.ValidateProxyURL(p.URL)
entries = append(entries, proxyPoolEntry{ProxyPoolEntry: p, RawURL: u.String()})
seen[u.String()] = true
}
active := s.m.proxyPoolActiveID()
if active != "" {
foundActive := false
for _, e := range entries {
if e.ID == active {
foundActive = true
break
}
}
if !foundActive {
active = ""
}
}
if err := s.m.setProxyPoolEntries(entries, active); err != nil {
writeErr(w, 500, err.Error())
return
}
if req.Interval != nil || req.TimedEnabled != nil || req.OnFailure != nil || req.Strategy != nil || req.Threshold != nil || req.Cooldown != nil || req.HealthInterval != nil {
timed, interval, onFailure, strategy := s.m.proxyRotateConfig()
threshold, cooldown, healthInterval := s.m.proxyHealthConfig()
if req.TimedEnabled != nil {
timed = *req.TimedEnabled
}
if req.Interval != nil {
interval = time.Duration(*req.Interval) * time.Second
}
if req.OnFailure != nil {
onFailure = *req.OnFailure
}
if req.Strategy != nil {
strategy = *req.Strategy
}
if req.Threshold != nil {
threshold = *req.Threshold
}
if req.Cooldown != nil {
cooldown = time.Duration(*req.Cooldown) * time.Second
}
if req.HealthInterval != nil {
healthInterval = time.Duration(*req.HealthInterval) * time.Second
}
if err := s.m.SetProxyPoolSettings(timed, int(interval/time.Second), onFailure, strategy, threshold, int(cooldown/time.Second), int(healthInterval/time.Second)); err != nil {
writeErr(w, 400, err.Error())
return
}
select {
case s.proxyPoolWake <- struct{}{}:
default:
}
}
entriesOut, activeOut, logs := s.m.ProxyPoolSnapshot()
writeJSON(w, 200, map[string]any{"entries": entriesOut, "active_id": activeOut, "switch_log": logs, "errors": errs})
}
func (m *Manager) proxyPoolActiveID() string {
if m.proxyPool == nil {
return ""
}
m.proxyPool.mu.Lock()
defer m.proxyPool.mu.Unlock()
return m.proxyPool.active
}
func (s *Server) updateProxyPool(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
var req struct {
URL *string `json:"url"`
Name *string `json:"name"`
Remark *string `json:"remark"`
Enabled *bool `json:"enabled"`
}
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10)).Decode(&req); err != nil {
writeErr(w, 400, err.Error())
return
}
if s.m.proxyPool == nil {
writeErr(w, 404, "代理池为空")
return
}
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
}
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 = "", ""
}
}
if req.Name != nil {
e.Name = strings.TrimSpace(*req.Name)
}
if req.Remark != nil {
e.Remark = strings.TrimSpace(*req.Remark)
}
if req.Enabled != nil {
e.Enabled = *req.Enabled
if !e.Enabled && s.m.proxyPool.active == id {
_ = s.m.applyActiveProxyLocked("", "manual")
}
}
if err := s.m.persistProxyPoolLocked(); err != nil {
writeErr(w, 500, err.Error())
return
}
health := e.Health
if !e.Enabled {
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}})
return
}
writeErr(w, 404, "代理不存在")
}
func (s *Server) deleteProxyPool(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
if s.m.proxyPool == nil {
writeErr(w, 404, "代理不存在")
return
}
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 {
if e.ID == id {
found = true
continue
}
out = append(out, e)
}
if !found {
writeErr(w, 404, "代理不存在")
return
}
s.m.proxyPool.entries = out
if s.m.proxyPool.active == id {
s.m.proxyPool.active = ""
if err := s.m.applyActiveProxyLocked("", "manual"); err != nil {
writeErr(w, 500, err.Error())
return
}
}
_ = s.m.persistProxyPoolLocked()
writeJSON(w, 200, map[string]any{"ok": true})
}
func (s *Server) activateProxyPool(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
if err := s.m.activateProxyID(id, "manual"); err != nil {
writeErr(w, 400, err.Error())
return
}
writeJSON(w, 200, map[string]any{"ok": true, "active_id": id, "active_proxy": maskSecrets(s.m.GlobalProxy())})
}
func (s *Server) testProxyPoolEntry(w http.ResponseWriter, r *http.Request) {
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()
}
writeJSON(w, 200, map[string]any{"ok": res.OK, "result": res})
return
}
}
writeErr(w, 404, "代理不存在")
}
func (s *Server) testAllProxyPool(w http.ResponseWriter, r *http.Request) {
select {
case s.proxyPoolWake <- struct{}{}:
default:
}
s.proxyPoolTick()
entries, active, logs := s.m.ProxyPoolSnapshot()
writeJSON(w, 200, map[string]any{"entries": entries, "active_id": active, "switch_log": logs})
}
+56
View File
@@ -0,0 +1,56 @@
package server
import (
"encoding/json"
"strings"
"testing"
)
func TestParseProxyBatchValidationAndDedup(t *testing.T) {
entries, errs := ParseProxyBatch("http://user:[email protected]:8080|one\nhttp://user:[email protected]:8080\nftp://bad:1\n# comment\n")
if len(entries) != 1 || len(errs) != 2 {
t.Fatalf("entries=%d errs=%d", len(entries), len(errs))
}
if entries[0].URL == "" || !strings.Contains(entries[0].URL, "pass") {
t.Fatalf("parser lost stored credential")
}
}
func TestProxyPoolSnapshotMasksCredentialsAndDisabledHealth(t *testing.T) {
m := &Manager{proxyPool: &proxyPoolState{entries: []proxyPoolEntry{{ProxyPoolEntry: ProxyPoolEntry{ID: "1", Enabled: false, Health: "healthy"}, RawURL: "http://user:secret@host:8080"}}, logs: []ProxySwitchLog{{From: "http://user:secret@host:8080", To: "http://user:secret2@host:8081", Reason: "timed"}}}}
entries, _, logs := m.ProxyPoolSnapshot()
if strings.Contains(entries[0].URL, "secret") || entries[0].Health != "disabled" {
t.Fatalf("entry leaked or wrong state: %+v", entries[0])
}
if strings.Contains(logs[0].From, "secret") || strings.Contains(logs[0].To, "secret2") {
t.Fatalf("switch log leaked credentials: %+v", logs[0])
}
}
func TestProxyPoolSequentialSkipsUnhealthyAndDisabled(t *testing.T) {
m := &Manager{proxyPool: &proxyPoolState{active: "a", entries: []proxyPoolEntry{
{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}},
{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: false, Health: "healthy"}},
{ProxyPoolEntry: ProxyPoolEntry{ID: "c", Enabled: true, Health: "unhealthy-cooling"}},
{ProxyPoolEntry: ProxyPoolEntry{ID: "d", Enabled: true, Health: "healthy"}},
}}}
if got := m.selectHealthyLocked("a"); got != "d" {
t.Fatalf("selected %q", got)
}
}
func TestProxyPoolIntervalLowerBound(t *testing.T) {
if minProxyRotateInterval >= 5 {
t.Fatalf("bound must allow the documented 5s setting")
}
if minProxyRotateInterval != 3 {
t.Fatalf("bound=%d, rationale expects 3s", minProxyRotateInterval)
}
}
func TestProxyPoolPersistenceKeepsRawURL(t *testing.T) {
b, err := json.Marshal(proxyPoolPersist{Entries: []proxyPoolEntry{{ProxyPoolEntry: ProxyPoolEntry{ID: "1", URL: "http://***@host:1"}, RawURL: "http://user:pass@host:1"}}})
if err != nil || !strings.Contains(string(b), "raw_url") || !strings.Contains(string(b), "pass") {
t.Fatalf("bad persistence encoding: %s", b)
}
}
+95 -18
View File
@@ -125,10 +125,13 @@ type Server struct {
// Cold task archives run through one persistent FIFO worker. The buffered wake
// channel coalesces enqueue bursts; the database remains the source of truth.
archiveWake chan struct{}
archiveWG sync.WaitGroup
credentialWake chan struct{}
proxyReconnect proxyReconnectSupervisor
archiveWake chan struct{}
archiveWG sync.WaitGroup
credentialWake chan struct{}
proxyReconnect proxyReconnectSupervisor
proxyPoolWake chan struct{}
proxyPoolWG sync.WaitGroup
proxyPoolProbeSem chan struct{}
// 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
@@ -163,7 +166,7 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
}
s := &Server{m: m, engine: NewEngine(m), ctx: ctx, skillDir: skillDir, dataDir: dataDir, jwtKey: key, chatBusy: map[string]bool{},
chatCancel: map[string]context.CancelCauseFunc{}, triggerQ: map[string][]triggeredRun{},
triggerActive: map[string]int{}, triggerCfg: map[string]triggerBehavior{},
triggerActive: map[string]int{}, triggerCfg: map[string]triggerBehavior{}, proxyPoolWake: make(chan struct{}, 1), proxyPoolProbeSem: make(chan struct{}, proxyPoolMaxProbeConcurrency),
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)}
@@ -290,6 +293,8 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
}
s.restoreTaskRuntimes()
s.ensureProxySupervisor()
s.proxyPoolWG.Add(1)
go func() { defer s.proxyPoolWG.Done(); s.runProxyPoolSupervisor(s.ctx) }()
go s.reconcileConcurrency()
s.startTaskArchiveWorker()
if m.pg != nil {
@@ -519,6 +524,7 @@ func (s *Server) applyLLM(cfg agent.Config) error {
// (backend/key) and gates Enabled per-conversation-agent at Chat time. 对话始终用激活配置。
s.chatAgent = agent.NewChatAgent(prov, cfg.Model, s.m.dir, tx, win) // chat page runner
s.chatAgent.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
s.chatAgent.SetProxyProvider(s.m.ProxyAddr)
s.chatAgent.SetWebSearch(s.m.WebSearchOpts())
s.chatAgent.SetGuard(s.chatGuard())
s.chatAgent.SetNonStreaming(nonStreamingResolver(cfg))
@@ -681,6 +687,7 @@ func (s *Server) chatAgentForProfile(id int64) *agent.ChatAgent {
tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts"))
ca := agent.NewChatAgent(s.poolForBinding(id, prov, cfg), cfg.Model, s.m.dir, tx, cfg.CompactionWindow())
ca.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
ca.SetProxyProvider(s.m.ProxyAddr)
ca.SetWebSearch(s.m.WebSearchOpts())
ca.SetGuard(s.chatGuard())
ca.SetNonStreaming(nonStreamingResolver(cfg))
@@ -937,6 +944,13 @@ func (s *Server) Handler() http.Handler {
mux.HandleFunc("POST /api/settings/web-search/test", s.testWebSearch)
// 出口代理连通性测试:发真实请求验证 global_proxy 是否可用,并回显出口 IP。
mux.HandleFunc("POST /api/settings/proxy/test", s.testEgressProxy)
mux.HandleFunc("GET /api/settings/proxy-pool", s.getProxyPool)
mux.HandleFunc("POST /api/settings/proxy-pool/import", s.importProxyPool)
mux.HandleFunc("PUT /api/settings/proxy-pool/{id}", s.updateProxyPool)
mux.HandleFunc("DELETE /api/settings/proxy-pool/{id}", s.deleteProxyPool)
mux.HandleFunc("POST /api/settings/proxy-pool/{id}/test", s.testProxyPoolEntry)
mux.HandleFunc("POST /api/settings/proxy-pool/test-all", s.testAllProxyPool)
mux.HandleFunc("POST /api/settings/proxy-pool/{id}/activate", s.activateProxyPool)
mux.HandleFunc("POST /api/tasks/{id}/proxy/test", s.testTaskProxy)
mux.HandleFunc("GET /api/report", s.getReport)
mux.HandleFunc("GET /api/chat/mentions", s.searchChatMentions)
@@ -3549,16 +3563,31 @@ func (s *Server) settingsPayload() map[string]any {
"web_search_backend": backend,
"brave_key_set": strings.TrimSpace(braveKey) != "",
"tavily_key_set": strings.TrimSpace(tavilyKey) != "",
"web_search_proxy": proxy, // 独立出口代理(http/https/socks5),空=直连
"global_proxy": s.m.GlobalProxy(), // 全局出口代理(http/https/socks5),所有目标流量走它,空=直连
"proxy_reconnect_max_attempts": func() int { n, _ := s.proxyReconnectConfig(); return n }(),
"proxy_reconnect_interval_seconds": func() int { _, d := s.proxyReconnectConfig(); return int(d / time.Second) }(),
"show_egress_ip": s.m.pg.GetBool(settingShowEgressIP, false),
"python_interpreter": strings.TrimSpace(pyStored), // 用户/自动设的值(空=用运行时检测)
"workers": s.m.Workers(), // 并发工作 agent 数(默认3);对之后启动的任务生效
"worker_pools": s.m.WorkerPools(), // 非空时按池顺序绑定 worker 会话
"task_concurrency_enabled": concOn, // 任务并发上限开关(默认关)
"task_concurrency_limit": concLimit, // 同时运行任务上限(开启后默认5)
"web_search_proxy": maskSecrets(proxy), // 独立出口代理(http/https/socks5),空=直连
"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_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),
"proxy_rotate_strategy": func() string {
v, ok, _ := s.m.pg.GetSetting(settingProxyRotateStrategy)
if ok && (v == "random" || v == "sequential") {
return v
}
return "sequential"
}(),
"proxy_health_fail_threshold": proxyIntSetting(s.m.pg, settingProxyHealthThreshold, defaultProxyHealthThreshold),
"proxy_health_cooldown_seconds": proxyIntSetting(s.m.pg, settingProxyHealthCooldown, defaultProxyHealthCooldown),
"proxy_health_check_interval_seconds": proxyIntSetting(s.m.pg, settingProxyHealthInterval, defaultProxyHealthInterval),
"proxy_reconnect_max_attempts": func() int { n, _ := s.proxyReconnectConfig(); return n }(),
"proxy_reconnect_interval_seconds": func() int { _, d := s.proxyReconnectConfig(); return int(d / time.Second) }(),
"show_egress_ip": s.m.pg.GetBool(settingShowEgressIP, false),
"python_interpreter": strings.TrimSpace(pyStored), // 用户/自动设的值(空=用运行时检测)
"workers": s.m.Workers(), // 并发工作 agent 数(默认3);对之后启动的任务生效
"worker_pools": s.m.WorkerPools(), // 非空时按池顺序绑定 worker 会话
"task_concurrency_enabled": concOn, // 任务并发上限开关(默认关)
"task_concurrency_limit": concLimit, // 同时运行任务上限(开启后默认5)
// LLM 轮询(故障转移)。默认关;开启后走全局激活配置的 agent 在当前配置不可用时
// 自动切到下一个配置。bind_fallback 仅在轮询开启时有意义(默认关)。
"llm_pool_enabled": s.m.LLMPoolEnabled(),
@@ -3635,6 +3664,13 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
TavilyKey *string `json:"tavily_search_api_key"`
WebSearchProxy *string `json:"web_search_proxy"` // 独立出口代理(http/https/socks5);null=不改,""=清空
GlobalProxy *string `json:"global_proxy"` // 全局出口代理(http/https/socks5);null=不改,""=清空(直连)
ProxyRotateTimed *bool `json:"proxy_rotate_timed_enabled"`
ProxyRotateInterval *int `json:"proxy_rotate_interval_seconds"`
ProxyRotateFailure *bool `json:"proxy_rotate_on_failure"`
ProxyRotateStrategy *string `json:"proxy_rotate_strategy"`
ProxyHealthThreshold *int `json:"proxy_health_fail_threshold"`
ProxyHealthCooldown *int `json:"proxy_health_cooldown_seconds"`
ProxyHealthInterval *int `json:"proxy_health_check_interval_seconds"`
ProxyReconnectMax *int `json:"proxy_reconnect_max_attempts"`
ProxyReconnectInterval *int `json:"proxy_reconnect_interval_seconds"`
ShowEgressIP *bool `json:"show_egress_ip"` // 在顶栏显示直连/全局代理出口 IP;默认关
@@ -3851,17 +3887,21 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
changed = true
}
if req.GlobalProxy != nil {
globalProxyValue := strings.TrimSpace(*req.GlobalProxy)
if strings.Contains(globalProxyValue, "***") || strings.Contains(globalProxyValue, "******") {
globalProxyValue = s.m.GlobalProxy()
}
var proxyGuards map[string]proxyReconnectGuard
if strings.TrimSpace(*req.GlobalProxy) == "" {
if globalProxyValue == "" {
proxyGuards = s.snapshotProxyReconnectGuards(false)
}
// Validation failure (bad scheme/host) is a client error, not a 500.
if err := s.m.SetGlobalProxy(*req.GlobalProxy); err != nil {
if err := s.m.SetGlobalProxy(globalProxyValue); err != nil {
writeErr(w, 400, err.Error())
return
}
changed = true // capture-off egress is baked into agents at build time → rebuild
if strings.TrimSpace(*req.GlobalProxy) == "" {
if globalProxyValue == "" {
s.proxyReconnectSuccessWithGuards(proxyGuards, nil)
}
}
@@ -3871,6 +3911,39 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
return
}
}
if req.ProxyRotateTimed != nil || req.ProxyRotateInterval != nil || req.ProxyRotateFailure != nil || req.ProxyRotateStrategy != nil || req.ProxyHealthThreshold != nil || req.ProxyHealthCooldown != nil || req.ProxyHealthInterval != nil {
timed, interval, onFailure, strategy := s.m.proxyRotateConfig()
threshold, cooldown, healthInterval := s.m.proxyHealthConfig()
if req.ProxyRotateTimed != nil {
timed = *req.ProxyRotateTimed
}
if req.ProxyRotateInterval != nil {
interval = time.Duration(*req.ProxyRotateInterval) * time.Second
}
if req.ProxyRotateFailure != nil {
onFailure = *req.ProxyRotateFailure
}
if req.ProxyRotateStrategy != nil {
strategy = strings.TrimSpace(*req.ProxyRotateStrategy)
}
if req.ProxyHealthThreshold != nil {
threshold = *req.ProxyHealthThreshold
}
if req.ProxyHealthCooldown != nil {
cooldown = time.Duration(*req.ProxyHealthCooldown) * time.Second
}
if req.ProxyHealthInterval != nil {
healthInterval = time.Duration(*req.ProxyHealthInterval) * time.Second
}
if err := s.m.SetProxyPoolSettings(timed, int(interval/time.Second), onFailure, strategy, threshold, int(cooldown/time.Second), int(healthInterval/time.Second)); err != nil {
writeErr(w, 400, err.Error())
return
}
select {
case s.proxyPoolWake <- struct{}{}:
default:
}
}
if req.ShowEgressIP != nil {
if err := s.m.pg.SetBool(settingShowEgressIP, *req.ShowEgressIP); err != nil {
writeErr(w, 500, err.Error())
@@ -3886,6 +3959,10 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
if req.WebSearchBackend != nil {
backend = *req.WebSearchBackend
}
if req.WebSearchProxy != nil && (strings.Contains(*req.WebSearchProxy, "***") || strings.Contains(*req.WebSearchProxy, "******")) {
_, _, _, _, savedSearchProxy := s.m.WebSearch()
req.WebSearchProxy = &savedSearchProxy
}
if err := s.m.SetWebSearch(on, backend, req.BraveKey, req.TavilyKey, req.WebSearchProxy); err != nil {
writeErr(w, 500, err.Error())
return
+4
View File
@@ -592,6 +592,7 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle {
wk.SetNoaEnabled(s.m.NoaCompactionEnabled) // 实验功能:noa 上下文压缩(平台级开关,每 run 读)
wk.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second)
wk.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
wk.SetProxyProvider(s.m.ProxyAddr)
wk.SetWebSearch(s.webSearchFor("worker"))
wk.SetConstraintInject(s.constraintInjectWorker) // 操作约束注入 worker(可配置,默认开)
pl := agent.NewPlanner(plannerRuntime, "task-router", s.m.dir, tx, plannerRuntime.CompactionWindow(), s.agentMaxTurns("planner"))
@@ -603,6 +604,7 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle {
pl.SetKillWork(s.engine.KillWork)
pl.SetSteerWork(s.engine.SteerWork)
pl.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
pl.SetProxyProvider(s.m.ProxyAddr)
pl.SetWebSearch(s.webSearchFor("planner"))
pl.SetConstraintInject(s.constraintInjectPlanner) // 操作约束注入 planner(可配置,默认开)
// cold-digest §7: 冷节点后台压缩。引擎经权威解析器实际驱动的就是这套 per-task planner
@@ -616,6 +618,7 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle {
main.SetMaxTokens(mainRuntime.maxTokens)
main.SetNoaEnabled(s.m.NoaCompactionEnabled) // 实验功能:noa 上下文压缩(平台级开关,每 run 读)
main.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
main.SetProxyProvider(s.m.ProxyAddr)
main.SetWebSearch(s.webSearchFor("mainagent"))
main.SetSteerWork(s.engine.SteerWork) // steer_work:人对运行中 work 实时纠偏
bundle := &taskAgentBundle{
@@ -662,6 +665,7 @@ func (s *Server) workerForTaskSlot(t *Task, slot WorkerSlot) (*agent.Worker, err
worker.SetNoaEnabled(s.m.NoaCompactionEnabled)
worker.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second)
worker.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
worker.SetProxyProvider(s.m.ProxyAddr)
worker.SetWebSearch(s.webSearchFor("worker"))
worker.SetConstraintInject(s.constraintInjectWorker)
bundle.pinnedWorkers[id] = worker
@@ -61,6 +61,14 @@ export function EgressIpProvider({ children }: { children: React.ReactNode }) {
};
}, [probeForProxy]);
React.useEffect(() => {
if (!enabled) return;
const timer = window.setInterval(() => {
void api.settings().then((settings) => probeForProxy(settings.global_proxy ?? "")).catch(() => undefined);
}, 10000);
return () => window.clearInterval(timer);
}, [enabled, probeForProxy]);
const setEnabled = React.useCallback(
async (nextEnabled: boolean) => {
const settings = await api.setSettings({ show_egress_ip: nextEnabled });
+67 -1
View File
@@ -18,12 +18,13 @@ import { Button } from "@/components/ui/button";
import { Card, CardContent, CardDescription, CardHeader, CardTitle } from "@/components/ui/card";
import { Field, FieldContent, FieldDescription, FieldLabel } from "@/components/ui/field";
import { Input } from "@/components/ui/input";
import { Textarea } from "@/components/ui/textarea";
import { Label } from "@/components/ui/label";
import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select";
import { Switch } from "@/components/ui/switch";
import { api } from "@/lib/api";
import { CHAT_SEND_MODE_OPTIONS, type ChatSendMode, setChatSendMode, useChatSendMode } from "@/lib/chat-send-mode";
import type { EgressReport, LLMProfile, Settings, WorkerPool } from "@/lib/types";
import type { EgressReport, LLMProfile, ProxyPoolEntry, Settings, WorkerPool } from "@/lib/types";
import { useEgressIp } from "../../_components/egress-ip-banner";
import { BasicAuthCard } from "./_components/basic-auth-card";
@@ -55,6 +56,17 @@ export default function SystemSettingsPage() {
const [savingProxy, setSavingProxy] = React.useState(false);
const [globalProxyInput, setGlobalProxyInput] = React.useState("");
const [savingGlobalProxy, setSavingGlobalProxy] = React.useState(false);
const [proxyPool, setProxyPool] = React.useState<ProxyPoolEntry[]>([]);
const [proxyPoolActiveId, setProxyPoolActiveId] = React.useState("");
const [proxyPoolText, setProxyPoolText] = React.useState("");
const [proxyPoolTimed, setProxyPoolTimed] = React.useState(false);
const [proxyPoolInterval, setProxyPoolInterval] = React.useState("30");
const [proxyPoolFailure, setProxyPoolFailure] = React.useState(false);
const [proxyPoolStrategy, setProxyPoolStrategy] = React.useState<"sequential" | "random">("sequential");
const [proxyPoolThreshold, setProxyPoolThreshold] = React.useState("3");
const [proxyPoolCooldown, setProxyPoolCooldown] = React.useState("60");
const [proxyPoolHealthInterval, setProxyPoolHealthInterval] = React.useState("30");
const [proxyPoolLogs, setProxyPoolLogs] = React.useState<Array<{ at: string; from: string; to: string; reason: string }>>([]);
const [proxyReconnectMax, setProxyReconnectMax] = React.useState("5");
const [proxyReconnectInterval, setProxyReconnectInterval] = React.useState("30");
const [savingEgressVisibility, setSavingEgressVisibility] = React.useState(false);
@@ -113,6 +125,15 @@ export default function SystemSettingsPage() {
setProxyInput(s.web_search_proxy ?? "");
const savedProxy = s.global_proxy ?? "";
setGlobalProxyInput(savedProxy);
setProxyPool(s.proxy_pool ?? []);
setProxyPoolActiveId(s.proxy_pool_active_id ?? "");
setProxyPoolTimed(!!s.proxy_rotate_timed_enabled);
setProxyPoolInterval(String(s.proxy_rotate_interval_seconds ?? 30));
setProxyPoolFailure(!!s.proxy_rotate_on_failure);
setProxyPoolStrategy(s.proxy_rotate_strategy === "random" ? "random" : "sequential");
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));
setProxyReconnectMax(String(s.proxy_reconnect_max_attempts ?? 5));
setProxyReconnectInterval(String(s.proxy_reconnect_interval_seconds ?? 30));
setPyInterp(s.python_interpreter ?? "");
@@ -432,6 +453,17 @@ export default function SystemSettingsPage() {
.finally(() => setSavingGlobalProxy(false));
};
const importProxyPool = () => {
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 = () => {
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 saveProxyReconnect = () => {
api.setSettings({ proxy_reconnect_max_attempts: Number(proxyReconnectMax), proxy_reconnect_interval_seconds: Number(proxyReconnectInterval) })
.then((s) => { apply(s); toast.success("已保存代理重连策略"); })
@@ -879,6 +911,40 @@ export default function SystemSettingsPage() {
))}
</div>
)}
<div className="mt-3 rounded-md border p-3">
<div className="flex flex-wrap items-center justify-between gap-2">
<div>
<p className="text-sm font-medium">代理池</p>
<p className="text-muted-foreground text-xs">当前在用:{proxyPool.find((entry) => entry.id === proxyPoolActiveId)?.name ?? (proxyPoolActiveId ? "已选择代理" : "直连")} · 每行一个代理;可用 URL|名称|备注。账号密码只会以掩码返回。</p>
</div>
<Button type="button" variant="outline" size="sm" onClick={testAllProxyPool}>测试全部</Button>
</div>
<Textarea className="mt-2 font-mono text-xs" rows={3} placeholder="http://user:pass@host:8080|香港 1|备用" value={proxyPoolText} onChange={(e) => setProxyPoolText(e.target.value)} />
<div className="mt-2 flex flex-wrap gap-2"><Button type="button" size="sm" onClick={importProxyPool} disabled={!proxyPoolText.trim()}>批量导入</Button></div>
<div className="mt-3 space-y-2">
{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={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>
<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>
))}
</div>
<div className="mt-3 grid grid-cols-1 gap-2 sm:grid-cols-3">
<div className="flex items-center justify-between gap-2 rounded border px-2 py-2"><Label htmlFor="proxy-pool-timed">定时轮换</Label><Switch id="proxy-pool-timed" checked={proxyPoolTimed} onCheckedChange={setProxyPoolTimed} /></div>
<div><Label htmlFor="proxy-pool-interval">间隔(秒,≥3)</Label><Input id="proxy-pool-interval" type="number" min={3} value={proxyPoolInterval} onChange={(e) => setProxyPoolInterval(e.target.value)} /></div>
<div className="flex items-center justify-between gap-2 rounded border px-2 py-2"><Label htmlFor="proxy-pool-failure">失效轮换</Label><Switch id="proxy-pool-failure" checked={proxyPoolFailure} onCheckedChange={setProxyPoolFailure} /></div>
<div><Label>策略</Label><Select value={proxyPoolStrategy} onValueChange={(v) => setProxyPoolStrategy(v as "sequential" | "random")}><SelectTrigger><SelectValue /></SelectTrigger><SelectContent><SelectItem value="sequential">顺序</SelectItem><SelectItem value="random">随机</SelectItem></SelectContent></Select></div>
<div><Label htmlFor="proxy-pool-threshold">失败阈值</Label><Input id="proxy-pool-threshold" type="number" min={1} value={proxyPoolThreshold} onChange={(e) => setProxyPoolThreshold(e.target.value)} /></div>
<div><Label htmlFor="proxy-pool-cooldown">冷却(秒)</Label><Input id="proxy-pool-cooldown" type="number" min={1} value={proxyPoolCooldown} onChange={(e) => setProxyPoolCooldown(e.target.value)} /></div>
<div><Label htmlFor="proxy-pool-health-interval">健康检查(秒)</Label><Input id="proxy-pool-health-interval" type="number" min={1} value={proxyPoolHealthInterval} onChange={(e) => setProxyPoolHealthInterval(e.target.value)} /></div>
<Button type="button" variant="outline" onClick={saveProxyPoolStrategy}>保存轮换策略</Button>
</div>
{proxyPoolLogs.length > 0 && <div className="mt-3 text-xs"><p className="font-medium">最近切换</p>{proxyPoolLogs.slice(-5).reverse().map((log) => <p key={`${log.at}-${log.to}`} className="text-muted-foreground">{new Date(log.at).toLocaleTimeString()} · {log.from || "直连"} → {log.to || "直连"} · {log.reason}</p>)}</div>}
</div>
</CardContent>
</Card>
+9
View File
@@ -34,6 +34,8 @@ import type {
DeleteTaskResult,
Edge,
EgressReport,
ProxyPoolEntry,
ProxySwitchLog,
EvidenceBodyPreview,
ExplorationNodePage,
ExplorationNodeQuery,
@@ -992,6 +994,13 @@ export const api = {
// global_proxy 传空 = 测直连;probe_url 可覆盖默认探测地址(隔离网用)。
testEgressProxy: (patch: { global_proxy?: string; probe_url?: string }) =>
post<EgressReport>(`/settings/proxy/test`, patch),
proxyPool: () => get<{ entries: ProxyPoolEntry[]; active_id: string; active_proxy: string; switch_log: ProxySwitchLog[] }>(`/settings/proxy-pool`),
importProxyPool: (payload: Record<string, unknown>) => post<{ entries: ProxyPoolEntry[]; active_id: string; switch_log: ProxySwitchLog[]; errors: Array<{ line: number; input: string; error: string }> }>(`/settings/proxy-pool/import`, payload),
updateProxyPool: (id: string, payload: Partial<Pick<ProxyPoolEntry, "url" | "name" | "remark" | "enabled">>) => put<{ entry: ProxyPoolEntry }>(`/settings/proxy-pool/${encodeURIComponent(id)}`, payload),
deleteProxyPool: (id: string) => del<{ ok: boolean }>(`/settings/proxy-pool/${encodeURIComponent(id)}`),
testProxyPoolEntry: (id: string) => post<{ ok: boolean; result?: EgressReport["checks"][number]["result"] }>(`/settings/proxy-pool/${encodeURIComponent(id)}/test`, {}),
testAllProxyPool: () => post<{ entries: ProxyPoolEntry[]; active_id: string; switch_log: ProxySwitchLog[] }>(`/settings/proxy-pool/test-all`, {}),
activateProxyPool: (id: string) => post<{ ok: boolean; active_id: string; active_proxy: string }>(`/settings/proxy-pool/${encodeURIComponent(id)}/activate`, {}),
testTaskProxy: (taskID: string) => post<{ ok: boolean; error?: string }>(`/tasks/${taskID}/proxy/test`, {}),
// ---- 漏洞 IM 推送 ----
+9
View File
@@ -3350,6 +3350,15 @@ export const settings: Settings = {
tavily_key_set: true,
web_search_proxy: "",
global_proxy: "",
proxy_pool: [],
proxy_pool_active_id: "",
proxy_rotate_timed_enabled: false,
proxy_rotate_interval_seconds: 30,
proxy_rotate_on_failure: false,
proxy_rotate_strategy: "sequential",
proxy_health_fail_threshold: 3,
proxy_health_cooldown_seconds: 60,
proxy_health_check_interval_seconds: 30,
show_egress_ip: false,
python_interpreter: "/usr/bin/python3",
workers: 3,
+6
View File
@@ -2406,6 +2406,12 @@ function route(m: string, path: string, seg: string[], q: URLSearchParams, b: Re
],
warnings: [],
};
if (path === "/settings/proxy-pool" && m === "GET") return { entries: D.settings.proxy_pool ?? [], active_id: D.settings.proxy_pool_active_id ?? "", active_proxy: D.settings.global_proxy ?? "", switch_log: [] };
if (path === "/settings/proxy-pool/import" && m === "POST") return { entries: D.settings.proxy_pool ?? [], active_id: D.settings.proxy_pool_active_id ?? "", switch_log: [], errors: [] };
if (path === "/settings/proxy-pool/test-all" && m === "POST") return { entries: D.settings.proxy_pool ?? [], active_id: D.settings.proxy_pool_active_id ?? "", switch_log: [] };
if (path.startsWith("/settings/proxy-pool/") && m === "PUT") return { entry: { ...(D.settings.proxy_pool ?? []).find((x) => x.id === path.split("/").at(-1)), ...b } };
if (path.startsWith("/settings/proxy-pool/") && m === "DELETE") return { ok: true };
if (path.startsWith("/settings/proxy-pool/") && m === "POST") return { ok: true, active_id: path.split("/").at(-1), active_proxy: "" };
if (path === "/settings/python/detect") return { python_interpreter: "/usr/bin/python3" };
if (path === "/chat")
return { reply: "(demo)我已把该建议注入为一条高优意图,work agent 会尽快执行。", mode: "hint" };
+25
View File
@@ -1004,6 +1004,15 @@ export interface Settings {
// 全局出口代理(http/https/socks5,可带 user:pass),所有目标流量走它。开启流量捕获时作为
// MITM 上游;关闭捕获时直接注入 agent 的 bash/WebFetch。空=直连。
global_proxy?: string;
proxy_pool?: ProxyPoolEntry[];
proxy_pool_active_id?: string;
proxy_rotate_timed_enabled?: boolean;
proxy_rotate_interval_seconds?: number;
proxy_rotate_on_failure?: boolean;
proxy_rotate_strategy?: "sequential" | "random";
proxy_health_fail_threshold?: number;
proxy_health_cooldown_seconds?: number;
proxy_health_check_interval_seconds?: number;
proxy_reconnect_max_attempts?: number;
proxy_reconnect_interval_seconds?: number;
show_egress_ip?: boolean; // 在顶栏显示直连 / 全局代理出口 IP;默认关闭
@@ -1031,6 +1040,22 @@ export interface Settings {
notify_digest_interval_min?: number; // 汇总模式周期(分钟),默认 30
}
export interface ProxyPoolEntry {
id: string;
url: string;
name?: string;
remark?: string;
enabled: boolean;
egress_ip?: string;
latency_ms?: number;
last_error?: string;
last_checked_at?: string;
consecutive_failures: number;
health: "healthy" | "unhealthy-cooling" | "disabled" | string;
}
export interface ProxySwitchLog { at: string; from: string; to: string; reason: string }
export interface WorkerPool {
profile_id: string;
workers: number;