Compare commits

...
Author SHA1 Message Date
RuoJi6 e74ab49f72 fix: 清空运行时代理后转发器直连并门控恢复、跳过空地址成员、补测 2026-11-01 22:53:11 +08:00
RuoJi6 5dcfd9fd39 fix: 代理池未探测不暂停、切换成功统一门控恢复、池模式保留重连次数与耗尽、禁用当前成员即切换、补关键测试 2026-11-01 22:23:35 +08:00
RuoJi6 2b30c47414 fix: 代理池恢复需探测确认并受耗尽门控、转发器认证与缓冲、掩码回写与删除收尾、补测 2026-11-01 21:24:02 +08:00
RuoJi6 45514b3586 fix: 代理池无健康成员走暂停不直连、单飞探测、热切覆盖 capture-off 与浏览器 MCP、脱敏与 UI 补全 2026-11-01 20:20:02 +08:00
RuoJi6 0bb954e0a4 feat: 全局代理池(健康检查、定时/失效轮换可同时开启、热切换与脱敏) 2026-11-01 19:37:18 +08:00
RuoJi6 008a9ced0d fix: 代理暂停归入任务暂停语义并为所有成功路径加轮次门控 2026-10-31 19:30:38 +08:00
RuoJi6 d93032fbea fix: 新一轮代理暂停重置重连计数并丢弃过期探测写回 2026-10-31 18:58:15 +08:00
RuoJi6 97a4f2b236 fix: 代理暂停/恢复屏障在锁内安装并防止复活已耗尽任务 2026-10-31 18:23:40 +08:00
RuoJi6 e878020c4d fix: 收紧代理故障判定并修复自动恢复门控/日历/终态问题 2026-10-31 16:23:28 +08:00
RuoJi6 de6a0fa125 feat: 代理异常暂停与自动重连恢复
Proxy failure pause, timed reconnect, and guarded auto-resume.
2026-10-31 15:54:41 +08:00
RuoJi6 0ed68d9143 Merge pull request #100 from RuoJi6/codex/fix-export-utf8-filenames
修复导出中文文件名乱码
2026-10-30 00:08:44 -07:00
ji ruo 542d7e9ca6 fix: preserve UTF-8 filenames for exports 2026-10-30 15:06:37 +08:00
RuoJi6 35a33b20f9 Merge pull request #98 from RuoJi6/codex/fix-image-drag-ghost
修复截图拖动时的原生拖拽重影
2026-10-29 22:13:34 -07:00
ji ruo 74d8b1efa2 fix: 禁用截图原生拖拽重影 2026-10-30 13:12:35 +08:00
RuoJi6 92e58d3de7 Merge pull request #97 from RuoJi6/codex/image-viewer-pan-zoom
修复截图放大拖动并显示缩放比例
2026-10-29 22:08:42 -07:00
ji ruo 21057b1a6b fix: 支持截图放大拖动并显示缩放比例 2026-10-30 13:07:49 +08:00
RuoJi6 136782f48e Merge pull request #96 from RuoJi6/codex/report-dialog-history-scroll
修复报告历史对话绑定与滚动位置
2026-10-29 22:04:54 -07:00
ji ruo 3af131272c fix: 修复报告对话绑定与滚动位置 2026-10-30 13:01:24 +08:00
RuoJi6 07c0c8ee35 Merge pull request #95 from RuoJi6/codex/backfill-report-conversation-binding
修复历史报告撰写对话回溯绑定
2026-10-29 21:36:50 -07:00
ji ruo 3f2c44be91 fix: 回溯绑定历史报告撰写对话 2026-10-30 12:34:04 +08:00
RuoJi6 57424249ce Merge pull request #94 from RuoJi6/codex/fix-report-conversation-binding
修复工具触发报告对话绑定
2026-10-29 21:29:23 -07:00
37 changed files with 4191 additions and 153 deletions

No files matched your search

+1
View File
@@ -28,6 +28,7 @@ func Causef(code, short, format string, args ...any) *AbortCause {
var ( var (
// Task-level execution context. // Task-level execution context.
AbortPausedBySchedule = cause("paused_by_schedule", "计划窗口结束,暂停任务", "任务将在下一个计划窗口恢复,本次执行已停止") AbortPausedBySchedule = cause("paused_by_schedule", "计划窗口结束,暂停任务", "任务将在下一个计划窗口恢复,本次执行已停止")
AbortPausedByProxy = cause("paused_by_proxy", "全局代理异常,暂停任务", "全局出口代理连接失败,任务已暂停;代理恢复后将自动继续")
AbortPausedByUser = cause("paused_by_user", "用户暂停了任务", AbortPausedByUser = cause("paused_by_user", "用户暂停了任务",
"用户通过任务控制接口(POST /api/tasks/{id}/control,action=pause)暂停了任务。本次 Planner/Worker 运行被主动取消;运行中的意图会退回 frontier(open),恢复任务后重新领取并从头执行") "用户通过任务控制接口(POST /api/tasks/{id}/control,action=pause)暂停了任务。本次 Planner/Worker 运行被主动取消;运行中的意图会退回 frontier(open),恢复任务后重新领取并从头执行")
AbortPausedByOrchestrator = cause("paused_by_orchestrator", "编排 Agent 暂停了任务", AbortPausedByOrchestrator = cause("paused_by_orchestrator", "编排 Agent 暂停了任务",
+9 -3
View File
@@ -29,6 +29,7 @@ type ChatAgent struct {
tx *transcript.Store tx *transcript.Store
window int window int
proxyAddr string proxyAddr string
proxyAddrFn func() string
proxyCACert string proxyCACert string
webSearch WebSearchOpts webSearch WebSearchOpts
guard *guard.Guard // optional; nil disables intercept hooks for chat 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 // 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. // 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). // SetWebSearch selects the web_search backend for the chat agent (off by default).
func (c *ChatAgent) SetWebSearch(o WebSearchOpts) { c.webSearch = o } 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() defer cleanup()
system, boundary := deferredSystem(chatSystem(agentKey, c.workDir, sessionWorkDir), def) system, boundary := deferredSystem(chatSystem(agentKey, c.workDir, sessionWorkDir), def)
proxyAddr := c.proxyAddr
if c.proxyAddrFn != nil {
proxyAddr = c.proxyAddrFn()
}
opts := agentcore.Options{ opts := agentcore.Options{
Provider: c.prov, Provider: c.prov,
SystemPrompt: system, SystemPrompt: system,
@@ -133,7 +139,7 @@ func (c *ChatAgent) Chat(ctx context.Context, agentKey, sessionID, message strin
UnlockSet: def.Unlock, UnlockSet: def.Unlock,
PermissionMode: permission.ModeBypass, PermissionMode: permission.ModeBypass,
EnableWebFetch: true, // 走记录代理留痕;载入代理 CA 验证 MITM 重签的 HTTPS 证书 EnableWebFetch: true, // 走记录代理留痕;载入代理 CA 验证 MITM 重签的 HTTPS 证书
WebFetchProxy: c.proxyAddr, WebFetchProxy: proxyAddr,
WebFetchCACert: c.proxyCACert, WebFetchCACert: c.proxyCACert,
// 联网搜索(可选)。ddgs 无需 key;brave-free 需 BraveKey;tavily 需 TavilyKey。 // 联网搜索(可选)。ddgs 无需 key;brave-free 需 BraveKey;tavily 需 TavilyKey。
// WebSearchProxy 是独立出口代理(http/https/socks5),与记录流量的 MITM 代理无关;空则直连。 // WebSearchProxy 是独立出口代理(http/https/socks5),与记录流量的 MITM 代理无关;空则直连。
@@ -145,7 +151,7 @@ func (c *ChatAgent) Chat(ctx context.Context, agentKey, sessionID, message strin
DeepSeekSearchAPIKey: ws.DeepSeekAPIKey, DeepSeekSearchAPIKey: ws.DeepSeekAPIKey,
DeepSeekSearchModel: ws.DeepSeekModel, DeepSeekSearchModel: ws.DeepSeekModel,
WebSearchProxy: ws.Proxy, WebSearchProxy: ws.Proxy,
BashEnv: proxyEnv(c.proxyAddr, c.proxyCACert), // Bash 子命令默认走代理+信任 CA BashEnv: proxyEnv(proxyAddr, c.proxyCACert), // Bash 子命令默认走代理+信任 CA
WorkingDir: sessionWorkDir, WorkingDir: sessionWorkDir,
MaxTurns: maxTurns, MaxTurns: maxTurns,
MaxDuration: maxDuration, MaxDuration: maxDuration,
+15 -9
View File
@@ -21,11 +21,12 @@ type MainAgent struct {
findingRecorder FindingRecorder findingRecorder FindingRecorder
prov llm.Provider prov llm.Provider
model string model string
tx *transcript.Store // raw LLM conversation persistence (nil = off) tx *transcript.Store // raw LLM conversation persistence (nil = off)
window int // context window in tokens (for compaction) window int // context window in tokens (for compaction)
windowFn func() int // optional dynamic task-chain minimum windowFn func() int // optional dynamic task-chain minimum
maxTurns int // max agent turns per run (0 = unlimited) maxTurns int // max agent turns per run (0 = unlimited)
proxyAddr string // recording proxy for WebFetch (empty = direct) proxyAddr string // recording proxy for WebFetch (empty = direct)
proxyAddrFn func() string
proxyCACert string // recording proxy's CA cert path (HTTPS verify) proxyCACert string // recording proxy's CA cert path (HTTPS verify)
webSearch WebSearchOpts // web_search tool backend selection (off by default) webSearch WebSearchOpts // web_search tool backend selection (off by default)
workDir string // shared work dir (surfaced in prompt as artifact-output target) 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 // 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). // 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). // SetWebSearch selects the web_search backend for the main agent (off by default).
func (m *MainAgent) SetWebSearch(o WebSearchOpts) { m.webSearch = o } 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) mainDir := ensureRunDir(m.workDir, taskID, 0)
ctx = intercept.WithReviewWorkingDirectory(ctx, mainDir) ctx = intercept.WithReviewWorkingDirectory(ctx, mainDir)
system, boundary := deferredSystem(mainAgentSystem(goal, m.workDir, mainDir), def) system, boundary := deferredSystem(mainAgentSystem(goal, m.workDir, mainDir), def)
proxyAddr := m.proxyAddr
if m.proxyAddrFn != nil {
proxyAddr = m.proxyAddrFn()
}
opts := agentcore.Options{ opts := agentcore.Options{
Provider: m.prov, Provider: m.prov,
SystemPrompt: system, SystemPrompt: system,
@@ -141,7 +147,7 @@ func (m *MainAgent) Chat(ctx context.Context, taskID int64, mainSeg int, as *db.
UnlockSet: def.Unlock, UnlockSet: def.Unlock,
PermissionMode: permission.ModeBypass, PermissionMode: permission.ModeBypass,
EnableWebFetch: true, // 走记录代理留痕;载入代理 CA 验证 MITM 重签的 HTTPS 证书 EnableWebFetch: true, // 走记录代理留痕;载入代理 CA 验证 MITM 重签的 HTTPS 证书
WebFetchProxy: m.proxyAddr, WebFetchProxy: proxyAddr,
WebFetchCACert: m.proxyCACert, WebFetchCACert: m.proxyCACert,
// 联网搜索(可选)。ddgs 无需 key;brave-free 需 BraveKey;tavily 需 TavilyKey。 // 联网搜索(可选)。ddgs 无需 key;brave-free 需 BraveKey;tavily 需 TavilyKey。
// WebSearchProxy 是独立出口代理(http/https/socks5),与记录流量的 MITM 代理无关;空则直连。 // 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, DeepSeekSearchAPIKey: m.webSearch.DeepSeekAPIKey,
DeepSeekSearchModel: m.webSearch.DeepSeekModel, DeepSeekSearchModel: m.webSearch.DeepSeekModel,
WebSearchProxy: m.webSearch.Proxy, WebSearchProxy: m.webSearch.Proxy,
BashEnv: proxyEnv(m.proxyAddr, m.proxyCACert), // Bash 子命令默认走代理+信任 CA BashEnv: proxyEnv(proxyAddr, m.proxyCACert), // Bash 子命令默认走代理+信任 CA
WorkingDir: mainDir, // 本任务工作目录 <workDir>/tasks/<taskID> WorkingDir: mainDir, // 本任务工作目录 <workDir>/tasks/<taskID>
ToolOutputDir: cmdOutDir(mainDir), ToolOutputDir: cmdOutDir(mainDir),
MaxTurns: m.maxTurns, // 0 = unlimited (configurable in agent management) MaxTurns: m.maxTurns, // 0 = unlimited (configurable in agent management)
Compaction: compactionConfig(m.compactionWindow()), // long chats stay within the window 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) 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) 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) proxyAddr string // recording proxy for WebFetch (empty = direct)
proxyCACert string // recording proxy's CA cert path (HTTPS verify) proxyAddrFn func() string
webSearch WebSearchOpts // web_search tool backend selection (off by default) proxyCACert string // recording proxy's CA cert path (HTTPS verify)
workDir string // shared work dir (surfaced in prompt as artifact-output target) webSearch WebSearchOpts // web_search tool backend selection (off by default)
injectConstraints func() bool // resolver: inject task operation constraints into system prompt? (nil = yes) workDir string // shared work dir (surfaced in prompt as artifact-output target)
nonStreamingFn func() bool // resolver: use non-streaming (Complete) path? (nil = streaming) injectConstraints func() bool // resolver: inject task operation constraints into system prompt? (nil = yes)
noaEnabledFn func() bool // resolver: use experimental noa compaction? (nil = off) nonStreamingFn func() bool // resolver: use non-streaming (Complete) path? (nil = streaming)
maxTokensFn func() int // resolver: per-reply output cap (nil/0 = send no cap) noaEnabledFn func() bool // resolver: use experimental noa compaction? (nil = off)
compactor *Compactor // cold-node compaction (§7); nil = disabled 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 // 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 // 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 // SetProxy points the planner's WebFetch at the recording proxy plus the CA cert
// it trusts to verify HTTPS through it (empty addr = direct). // 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). // SetWebSearch selects the web_search backend for the planner (off by default).
func (p *Planner) SetWebSearch(o WebSearchOpts) { p.webSearch = o } 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 { if tc.DeadlineUnix > 0 {
settle = wrapupSettlementForTask("planner", nil, clamped) settle = wrapupSettlementForTask("planner", nil, clamped)
} }
proxyAddr := p.proxyAddr
if p.proxyAddrFn != nil {
proxyAddr = p.proxyAddrFn()
}
opts := agentcore.Options{ opts := agentcore.Options{
Provider: p.prov, Provider: p.prov,
SystemPrompt: system, SystemPrompt: system,
@@ -411,7 +417,7 @@ func (p *Planner) Plan(ctx context.Context, taskID int64, as *db.AssetStore, ts
UnlockSet: def.Unlock, UnlockSet: def.Unlock,
PermissionMode: permission.ModeBypass, PermissionMode: permission.ModeBypass,
EnableWebFetch: true, // 走记录代理留痕;载入代理 CA 验证 MITM 重签的 HTTPS 证书 EnableWebFetch: true, // 走记录代理留痕;载入代理 CA 验证 MITM 重签的 HTTPS 证书
WebFetchProxy: p.proxyAddr, WebFetchProxy: proxyAddr,
WebFetchCACert: p.proxyCACert, WebFetchCACert: p.proxyCACert,
// 联网搜索(可选)。ddgs 无需 key;brave-free 需 BraveKey;tavily 需 TavilyKey。 // 联网搜索(可选)。ddgs 无需 key;brave-free 需 BraveKey;tavily 需 TavilyKey。
// WebSearchProxy 是独立出口代理(http/https/socks5),与记录流量的 MITM 代理无关;空则直连。 // 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, DeepSeekSearchAPIKey: p.webSearch.DeepSeekAPIKey,
DeepSeekSearchModel: p.webSearch.DeepSeekModel, DeepSeekSearchModel: p.webSearch.DeepSeekModel,
WebSearchProxy: p.webSearch.Proxy, WebSearchProxy: p.webSearch.Proxy,
BashEnv: proxyEnv(p.proxyAddr, p.proxyCACert), // Bash 子命令默认走代理+信任 CA BashEnv: proxyEnv(proxyAddr, p.proxyCACert), // Bash 子命令默认走代理+信任 CA
WorkingDir: taskDir, // 本任务工作目录 <workDir>/tasks/<taskID> WorkingDir: taskDir, // 本任务工作目录 <workDir>/tasks/<taskID>
ToolOutputDir: cmdOutDir(taskDir), ToolOutputDir: cmdOutDir(taskDir),
MaxTurns: p.maxTurns, // 0 = unlimited (configurable in agent management) MaxTurns: p.maxTurns, // 0 = unlimited (configurable in agent management)
MaxDuration: maxDur, // 0=不限;有 deadline 时=距 deadline 剩余 MaxDuration: maxDur, // 0=不限;有 deadline 时=距 deadline 剩余
+1 -1
View File
@@ -139,7 +139,7 @@ func TestTerminalTextNonAbortReasons(t *testing.T) {
func TestAbortCausesAreWellFormed(t *testing.T) { func TestAbortCausesAreWellFormed(t *testing.T) {
all := []*AbortCause{ all := []*AbortCause{
AbortPausedByUser, AbortPausedByOrchestrator, AbortTaskDeleted, AbortPausedOnReload, AbortPausedByUser, AbortPausedByProxy, AbortPausedByOrchestrator, AbortTaskDeleted, AbortPausedOnReload,
AbortGoalMet, AbortSettleDrainTimeout, AbortKilledByPlanner, AbortWorkPausedByUser, AbortGoalMet, AbortSettleDrainTimeout, AbortKilledByPlanner, AbortWorkPausedByUser,
AbortWorkCancelledByUser, AbortWorkFinished, AbortPausedRaceGuard, AbortWorkCancelledByUser, AbortWorkFinished, AbortPausedRaceGuard,
AbortChatStoppedByUser, AbortChatPausedWithTask, AbortChatTurnFinished, AbortChatStoppedByUser, AbortChatPausedWithTask, AbortChatTurnFinished,
+10 -3
View File
@@ -57,6 +57,7 @@ type Worker struct {
model string model string
workDir string workDir string
proxyAddr string proxyAddr string
proxyAddrFn func() string
proxyCACert string // recording proxy's CA cert path (for WebFetch HTTPS verify) proxyCACert string // recording proxy's CA cert path (for WebFetch HTTPS verify)
webSearch WebSearchOpts // web_search tool backend selection (off by default) webSearch WebSearchOpts // web_search tool backend selection (off by default)
tx *transcript.Store // raw LLM conversation persistence (nil = off) 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. // that MITM proxy. Empty addr disables the hint.
func (w *Worker) SetProxy(addr, caCert string) { w.proxyAddr, w.proxyCACert = addr, caCert } 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). // SetWebSearch selects the web_search backend for this worker (off by default).
func (w *Worker) SetWebSearch(o WebSearchOpts) { w.webSearch = o } func (w *Worker) SetWebSearch(o WebSearchOpts) { w.webSearch = o }
@@ -393,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. // inherit a parent run's background into the action reviewer.
ctx = intercept.WithReviewContext(ctx, runDir, intercept.ReviewBackground{}) ctx = intercept.WithReviewContext(ctx, runDir, intercept.ReviewBackground{})
overview := renderWorkerGraphOverview(tsx.graphOverviewData()) 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() { if w.wantConstraints() {
sysBody += constraintBlock(ts) // 操作约束(若有)注入系统提示,worker 执行时严格遵守 sysBody += constraintBlock(ts) // 操作约束(若有)注入系统提示,worker 执行时严格遵守
} }
@@ -437,7 +444,7 @@ func (w *Worker) execute(ctx context.Context, name string, taskID int64, as *db.
// WebFetch 走记录代理,其 HTTP 与 curl 一样被留痕;载入代理 CA 让经 MITM // WebFetch 走记录代理,其 HTTP 与 curl 一样被留痕;载入代理 CA 让经 MITM
// 重签的 HTTPS 证书能【正常校验通过】(而非关掉校验)。proxy 空则直连。 // 重签的 HTTPS 证书能【正常校验通过】(而非关掉校验)。proxy 空则直连。
EnableWebFetch: true, EnableWebFetch: true,
WebFetchProxy: w.proxyAddr, WebFetchProxy: proxyAddr,
WebFetchCACert: w.proxyCACert, WebFetchCACert: w.proxyCACert,
// 联网搜索(可选)。ddgs 无需 key;brave-free 需 BraveKey;tavily 需 TavilyKey。 // 联网搜索(可选)。ddgs 无需 key;brave-free 需 BraveKey;tavily 需 TavilyKey。
// WebSearchProxy 是独立的出口代理(http/https/socks5),与记录流量的 MITM 代理无关;空则直连。 // 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, DeepSeekSearchModel: w.webSearch.DeepSeekModel,
WebSearchProxy: w.webSearch.Proxy, WebSearchProxy: w.webSearch.Proxy,
// Bash 子命令的 HTTP 默认走记录代理 + 信任其 CA(工具无需 -x/-k)。 // Bash 子命令的 HTTP 默认走记录代理 + 信任其 CA(工具无需 -x/-k)。
BashEnv: proxyEnv(w.proxyAddr, w.proxyCACert), BashEnv: proxyEnv(proxyAddr, w.proxyCACert),
WorkingDir: runDir, WorkingDir: runDir,
MaxTurns: w.maxTurns, // 0 = unlimited (configurable in agent management) MaxTurns: w.maxTurns, // 0 = unlimited (configurable in agent management)
// 墙钟预算,轮边界判,不打断半路;0 = 不限。有任务级 deadline 时夹逼到 min(自身预算, // 墙钟预算,轮边界判,不打断半路;0 = 不限。有任务级 deadline 时夹逼到 min(自身预算,
+1
View File
@@ -113,6 +113,7 @@ func run() int {
} }
log.Printf("[config] skill 目录: %s", skillDir) log.Printf("[config] skill 目录: %s", skillDir)
srv := server.New(ctx, mgr, skillDir, *dataDir, config.BaseDir()) srv := server.New(ctx, mgr, skillDir, *dataDir, config.BaseDir())
defer srv.Close()
httpSrv := &http.Server{ httpSrv := &http.Server{
Addr: *addr, Addr: *addr,
Handler: srv.Handler(), Handler: srv.Handler(),
+34
View File
@@ -680,6 +680,11 @@ func (d *DB) GetFinding(id int64) (*DBFinding, error) {
if reportConv.Valid { if reportConv.Valid {
f.ReportConversationID = &reportConv.Int64 f.ReportConversationID = &reportConv.Int64
} }
if f.ReportConversationID == nil && strings.TrimSpace(f.Report) != "" {
if resolved, resolveErr := d.ResolveFindingReportConversation(f.ID); resolveErr == nil {
f.ReportConversationID = resolved
}
}
return f, nil return f, nil
} }
@@ -688,6 +693,35 @@ func (d *DB) SetFindingReportConversationID(findingID, conversationID int64) err
return err return err
} }
// ResolveFindingReportConversation backfills links for reports generated before
// report_conversation_id was persisted.
func (d *DB) ResolveFindingReportConversation(findingID int64) (*int64, error) {
var id int64
err := d.QueryRow(`SELECT c.id FROM conversations c
JOIN conversation_activities a ON a.conversation_id=c.id
WHERE c.agent_key='reporter' AND (
c.title = '漏洞报告 · finding#' || $1::text OR
c.title ILIKE '%finding#' || $1::text || '%' OR
a.detail ILIKE '%finding_id=' || $1::text || '%' OR
a.detail ILIKE '%"finding_id":' || $1::text || '%'
)
ORDER BY CASE WHEN EXISTS (
SELECT 1 FROM conversation_activities r
WHERE r.conversation_id=c.id AND (r.tool='update_finding_report' OR r.summary ILIKE 'finding % report updated%')
) THEN 0 ELSE 1 END,
c.created_at ASC, c.id ASC LIMIT 1`, findingID).Scan(&id)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
if err := d.SetFindingReportConversationID(findingID, id); err != nil {
return nil, err
}
return &id, nil
}
// DeleteFinding removes a finding entirely: the standalone findings row and its // DeleteFinding removes a finding entirely: the standalone findings row and its
// originating exploration node (kind='finding'), so it disappears from the findings // originating exploration node (kind='finding'), so it disappears from the findings
// list, the per-task 发现 Tab, and the exploration graph alike. Deleting the node // list, the per-task 发现 Tab, and the exploration graph alike. Deleting the node
+3
View File
@@ -513,6 +513,7 @@ CREATE TABLE IF NOT EXISTS tasks (
status TEXT NOT NULL DEFAULT 'created' status TEXT NOT NULL DEFAULT 'created'
CHECK (status IN ('created','running','paused','done','failed','timeout')), CHECK (status IN ('created','running','paused','done','failed','timeout')),
paused BOOLEAN NOT NULL DEFAULT false, paused BOOLEAN NOT NULL DEFAULT false,
pause_reason TEXT NOT NULL DEFAULT '',
queued BOOLEAN NOT NULL DEFAULT false, queued BOOLEAN NOT NULL DEFAULT false,
queued_at TIMESTAMPTZ, queued_at TIMESTAMPTZ,
queue_mode TEXT NOT NULL DEFAULT '', queue_mode TEXT NOT NULL DEFAULT '',
@@ -566,6 +567,7 @@ ALTER TABLE tasks ADD COLUMN IF NOT EXISTS llm_chain_revision BIGINT NOT NULL DE
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS archived_at TIMESTAMPTZ; ALTER TABLE tasks ADD COLUMN IF NOT EXISTS archived_at TIMESTAMPTZ;
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS active_run_seconds BIGINT NOT NULL DEFAULT 0; ALTER TABLE tasks ADD COLUMN IF NOT EXISTS active_run_seconds BIGINT NOT NULL DEFAULT 0;
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS run_segment_started_at TIMESTAMPTZ; ALTER TABLE tasks ADD COLUMN IF NOT EXISTS run_segment_started_at TIMESTAMPTZ;
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS pause_reason TEXT NOT NULL DEFAULT '';
CREATE INDEX IF NOT EXISTS idx_tasks_category ON tasks(category_id, created_at DESC) CREATE INDEX IF NOT EXISTS idx_tasks_category ON tasks(category_id, created_at DESC)
WHERE deleted_at IS NULL AND category_id IS NOT NULL; WHERE deleted_at IS NULL AND category_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_tasks_pinned ON tasks(pinned_at DESC) CREATE INDEX IF NOT EXISTS idx_tasks_pinned ON tasks(pinned_at DESC)
@@ -1084,6 +1086,7 @@ CREATE TABLE IF NOT EXISTS task_schedule_tasks (
); );
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_paused BOOLEAN NOT NULL DEFAULT false; ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_paused BOOLEAN NOT NULL DEFAULT false;
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_manual_until TIMESTAMPTZ; ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_manual_until TIMESTAMPTZ;
UPDATE tasks SET pause_reason='paused_by_schedule' WHERE paused AND schedule_paused AND pause_reason='';
CREATE INDEX IF NOT EXISTS idx_task_schedule_tasks_task ON task_schedule_tasks(task_id); CREATE INDEX IF NOT EXISTS idx_task_schedule_tasks_task ON task_schedule_tasks(task_id);
ALTER TABLE task_schedule_tasks ADD COLUMN IF NOT EXISTS paused_by_schedule BOOLEAN NOT NULL DEFAULT false; ALTER TABLE task_schedule_tasks ADD COLUMN IF NOT EXISTS paused_by_schedule BOOLEAN NOT NULL DEFAULT false;
+6 -3
View File
@@ -25,6 +25,7 @@ type Task struct {
ExplorationID int64 `json:"exploration_id"` ExplorationID int64 `json:"exploration_id"`
Status string `json:"status"` Status string `json:"status"`
Paused bool `json:"paused"` Paused bool `json:"paused"`
PauseReason string `json:"pause_reason,omitempty"`
Queued bool `json:"queued"` Queued bool `json:"queued"`
QueuedAt *time.Time `json:"queued_at,omitempty"` QueuedAt *time.Time `json:"queued_at,omitempty"`
QueueMode string `json:"queue_mode,omitempty"` QueueMode string `json:"queue_mode,omitempty"`
@@ -302,7 +303,7 @@ RETURNING id, status, paused, created_at`, opts.Name, opts.Remark, opts.Category
return nil, err return nil, err
} }
if !in.StartImmediately { if !in.StartImmediately {
if _, err = tx.Exec(`UPDATE tasks SET paused=true,schedule_paused=true,queue_mode='bootstrap' WHERE id=$1`, t.ID); err != nil { if _, err = tx.Exec(`UPDATE tasks SET paused=true,pause_reason='paused_by_schedule',schedule_paused=true,queue_mode='bootstrap' WHERE id=$1`, t.ID); err != nil {
return nil, err return nil, err
} }
t.Paused = true t.Paused = true
@@ -405,11 +406,11 @@ func insertTaskLLMProfiles(tx *sql.Tx, taskID int64, profileIDs []int64) error {
const taskCols = `id, COALESCE(name,''), remark, category_id, const taskCols = `id, COALESCE(name,''), remark, category_id,
COALESCE((SELECT category.name FROM task_categories category WHERE category.id=tasks.category_id),''), COALESCE((SELECT category.name FROM task_categories category WHERE category.id=tasks.category_id),''),
description, goal, exploration_id, status, paused, queued, queued_at, COALESCE(queue_mode,''), llm_profile_id, active_llm_profile_id, COALESCE(parent_ref,''), pinned_at, created_at, completed_at, COALESCE(timeout_seconds,0), COALESCE(plan_heartbeat_seconds,300), COALESCE(coverage_enabled,true), COALESCE(worker_count,0), first_run_at, deadline_at, COALESCE(active_run_seconds,0), run_segment_started_at` description, goal, exploration_id, status, paused, COALESCE(pause_reason,''), queued, queued_at, COALESCE(queue_mode,''), llm_profile_id, active_llm_profile_id, COALESCE(parent_ref,''), pinned_at, created_at, completed_at, COALESCE(timeout_seconds,0), COALESCE(plan_heartbeat_seconds,300), COALESCE(coverage_enabled,true), COALESCE(worker_count,0), first_run_at, deadline_at, COALESCE(active_run_seconds,0), run_segment_started_at`
func scanTask(sc interface{ Scan(...any) error }) (*Task, error) { func scanTask(sc interface{ Scan(...any) error }) (*Task, error) {
var t Task var t Task
if err := sc.Scan(&t.ID, &t.Name, &t.Remark, &t.CategoryID, &t.CategoryName, &t.Description, &t.Goal, &t.ExplorationID, &t.Status, &t.Paused, &t.Queued, &t.QueuedAt, &t.QueueMode, &t.LLMProfileID, &t.ActiveLLMProfileID, &t.ParentRef, &t.PinnedAt, &t.CreatedAt, &t.CompletedAt, &t.TimeoutSeconds, &t.PlanHeartbeatSeconds, &t.CoverageEnabled, &t.WorkerCount, &t.FirstRunAt, &t.DeadlineAt, &t.ActiveRunSeconds, &t.RunSegmentStartedAt); err != nil { if err := sc.Scan(&t.ID, &t.Name, &t.Remark, &t.CategoryID, &t.CategoryName, &t.Description, &t.Goal, &t.ExplorationID, &t.Status, &t.Paused, &t.PauseReason, &t.Queued, &t.QueuedAt, &t.QueueMode, &t.LLMProfileID, &t.ActiveLLMProfileID, &t.ParentRef, &t.PinnedAt, &t.CreatedAt, &t.CompletedAt, &t.TimeoutSeconds, &t.PlanHeartbeatSeconds, &t.CoverageEnabled, &t.WorkerCount, &t.FirstRunAt, &t.DeadlineAt, &t.ActiveRunSeconds, &t.RunSegmentStartedAt); err != nil {
return nil, err return nil, err
} }
t.Pinned = t.PinnedAt != nil t.Pinned = t.PinnedAt != nil
@@ -612,6 +613,7 @@ func (d *DB) SetStatusRuntime(id int64, status string) (int64, *time.Time, error
err := d.QueryRow(` err := d.QueryRow(`
UPDATE tasks UPDATE tasks
SET status = $1, SET status = $1,
pause_reason = CASE WHEN $1 IN ('done','failed','timeout') THEN '' ELSE pause_reason END,
completed_at = CASE WHEN $1 IN ('done','failed','timeout') THEN COALESCE(completed_at, now()) ELSE NULL END, completed_at = CASE WHEN $1 IN ('done','failed','timeout') THEN COALESCE(completed_at, now()) ELSE NULL END,
active_run_seconds = active_run_seconds + CASE WHEN $1 IN ('done','failed','timeout') AND run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END, active_run_seconds = active_run_seconds + CASE WHEN $1 IN ('done','failed','timeout') AND run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
run_segment_started_at = CASE WHEN $1 IN ('done','failed','timeout') THEN NULL ELSE run_segment_started_at END run_segment_started_at = CASE WHEN $1 IN ('done','failed','timeout') THEN NULL ELSE run_segment_started_at END
@@ -638,6 +640,7 @@ func (d *DB) SetTerminalStatusGuardedRuntime(id int64, status string) (active in
err = d.QueryRow(` err = d.QueryRow(`
UPDATE tasks UPDATE tasks
SET status = $1, SET status = $1,
pause_reason = '',
completed_at = COALESCE(completed_at, now()), completed_at = COALESCE(completed_at, now()),
active_run_seconds = active_run_seconds + CASE WHEN run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END, active_run_seconds = active_run_seconds + CASE WHEN run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
run_segment_started_at = NULL run_segment_started_at = NULL
+46
View File
@@ -188,6 +188,43 @@ func (s *Server) taskManualScheduleEnd(taskID string, now time.Time) (*time.Time
return until, nil return until, nil
} }
// proxyResumeInsideSchedule reports whether a task attached to any enabled
// calendar is currently allowed to run. Tasks without a calendar remain
// eligible. Manual schedule releases are honored through schedule_manual_until.
func (s *Server) proxyResumeInsideSchedule(taskID string, now time.Time) (bool, error) {
id, err := strconv.ParseInt(taskID, 10, 64)
if err != nil {
return false, err
}
items, err := s.m.pg.ListTaskSchedules()
if err != nil {
return false, err
}
attached := false
open := false
for _, item := range items {
for _, boundID := range item.TaskIDs {
if boundID != id {
continue
}
attached = true
if item.Enabled && item.Status != "completed" {
w := evaluateScheduleWindow(item, now)
open = open || w.active || item.LastWindowKey == "manual"
}
break
}
}
if !attached {
return true, nil
}
var manualUntil time.Time
if err := s.m.pg.QueryRow(`SELECT COALESCE(schedule_manual_until, 'epoch'::timestamptz) FROM tasks WHERE id=$1`, id).Scan(&manualUntil); err == nil && manualUntil.After(now) {
open = true
}
return open, nil
}
func (sc *Scheduler) fireTaskSchedules() { func (sc *Scheduler) fireTaskSchedules() {
sc.fireTaskSchedulesAt(time.Now()) sc.fireTaskSchedulesAt(time.Now())
} }
@@ -248,6 +285,15 @@ func (sc *Scheduler) fireTaskSchedulesAt(now time.Time) {
if db.IsTerminal(state.Status) { if db.IsTerminal(state.Status) {
continue continue
} }
// A proxy-paused task can remain paused when a calendar window ends
// during reconnect. Convert it to a schedule pause so a later probe
// cannot resume work outside the configured window.
if !open[id] && state.Paused && state.PauseReason == "paused_by_proxy" {
if err := sc.s.m.ConvertProxyPauseToSchedule(strconv.FormatInt(id, 10)); err != nil {
errors = append(errors, fmt.Sprintf("任务 #%d: %s", id, err))
}
continue
}
action := "" action := ""
if open[id] && state.Paused { if open[id] && state.Paused {
action = "resume" action = "resume"
+6 -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)) body, err := execPython(ctx, interp, key, spec.Code, params, workDir, sessionEnv, timeoutOr(spec.TimeoutMs, 120000))
if err != nil { if err != nil {
return actool.Errorf(err.Error()), nil return actool.Errorf(maskSecrets(err.Error())), nil
} }
return actool.Text(actool.Capture(tc, body)), 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() defer cancel()
req, err := http.NewRequestWithContext(runCtx, method, rawURL, bodyReader) req, err := http.NewRequestWithContext(runCtx, method, rawURL, bodyReader)
if err != nil { if err != nil {
return actool.Errorf(err.Error()), nil return actool.Errorf(maskSecrets(err.Error())), nil
} }
for k, v := range spec.Headers { for k, v := range spec.Headers {
req.Header.Set(k, renderTemplate(v, params, identity)) req.Header.Set(k, renderTemplate(v, params, identity))
@@ -447,7 +447,10 @@ func (s *Server) runHTTPTool(ctx context.Context, execRaw json.RawMessage, param
} }
resp, err := client.Do(req) resp, err := client.Do(req)
if err != nil { if err != nil {
return actool.Errorf("请求失败: " + err.Error()), nil if spec.UseRecordingProxy || strings.TrimSpace(spec.Proxy) == strings.TrimSpace(s.m.GlobalProxy()) {
s.handleProxyToolFailure(runCtx, err)
}
return actool.Errorf("请求失败: " + maskSecrets(err.Error())), nil
} }
defer resp.Body.Close() defer resp.Body.Close()
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
+55 -36
View File
@@ -52,6 +52,8 @@ type TaskDTO struct {
ActiveRunSeconds int64 `json:"active_run_seconds"` ActiveRunSeconds int64 `json:"active_run_seconds"`
RunSegmentStarted int64 `json:"run_segment_started_unix"` RunSegmentStarted int64 `json:"run_segment_started_unix"`
Paused bool `json:"paused"` Paused bool `json:"paused"`
PauseReason string `json:"pause_reason,omitempty"`
Reconnect *ProxyReconnectDTO `json:"reconnect,omitempty"`
Queued bool `json:"queued"` Queued bool `json:"queued"`
Tokens TokenTotalDTO `json:"tokens"` // whole-task token consumption Tokens TokenTotalDTO `json:"tokens"` // whole-task token consumption
GoalsTotal int `json:"goals_total"` GoalsTotal int `json:"goals_total"`
@@ -70,6 +72,14 @@ type TaskDTO struct {
WorkerCount int `json:"worker_count"` // 本任务 Worker 数;0=用全局默认 WorkerCount int `json:"worker_count"` // 本任务 Worker 数;0=用全局默认
} }
type ProxyReconnectDTO struct {
Attempt int `json:"attempt"`
Max int `json:"max"`
NextAt int64 `json:"next_at,omitempty"`
Exhausted bool `json:"exhausted"`
LastError string `json:"last_error,omitempty"`
}
// FindingSeverityDTO 是任务列表里按严重度分档的漏洞计数(严重/高/中/低)。 // FindingSeverityDTO 是任务列表里按严重度分档的漏洞计数(严重/高/中/低)。
type FindingSeverityDTO struct { type FindingSeverityDTO struct {
Critical int `json:"critical"` Critical int `json:"critical"`
@@ -118,25 +128,32 @@ func taskDTO(t *Task, status string) TaskDTO {
} }
profileIDs := append(make([]int64, 0, len(llmState.ProfileIDs)), llmState.ProfileIDs...) profileIDs := append(make([]int64, 0, len(llmState.ProfileIDs)), llmState.ProfileIDs...)
return TaskDTO{ return TaskDTO{
ID: t.ID, ID: t.ID,
ExplorationID: t.ExpID, ExplorationID: t.ExpID,
Name: lifecycle.Name, Name: lifecycle.Name,
Remark: lifecycle.Remark, Remark: lifecycle.Remark,
CategoryID: lifecycle.CategoryID, CategoryID: lifecycle.CategoryID,
CategoryName: lifecycle.CategoryName, CategoryName: lifecycle.CategoryName,
Pinned: lifecycle.PinnedAt > 0, Pinned: lifecycle.PinnedAt > 0,
PinnedAt: completedRFC(lifecycle.PinnedAt), PinnedAt: completedRFC(lifecycle.PinnedAt),
Description: t.Description, Description: t.Description,
Goal: t.Goal, Goal: t.Goal,
Status: status, Status: status,
CreatedAt: rfc3339(time.Unix(t.CreatedAt, 0)), CreatedAt: rfc3339(time.Unix(t.CreatedAt, 0)),
CreatedUnix: t.CreatedAt, CreatedUnix: t.CreatedAt,
RunSeconds: runSeconds, RunSeconds: runSeconds,
ActiveRunSeconds: lifecycle.ActiveRunSeconds, ActiveRunSeconds: lifecycle.ActiveRunSeconds,
RunSegmentStarted: lifecycle.RunSegmentStartedAt, RunSegmentStarted: lifecycle.RunSegmentStartedAt,
CompletedAt: completedRFC(lifecycle.CompletedAt), CompletedAt: completedRFC(lifecycle.CompletedAt),
CompletedUnix: lifecycle.CompletedAt, CompletedUnix: lifecycle.CompletedAt,
Paused: lifecycle.Paused, Paused: lifecycle.Paused,
PauseReason: lifecycle.PauseReason,
Reconnect: func() *ProxyReconnectDTO {
if lifecycle.PauseReason != "paused_by_proxy" && lifecycle.ReconnectAttempt == 0 && !lifecycle.ReconnectExhausted {
return nil
}
return &ProxyReconnectDTO{Attempt: lifecycle.ReconnectAttempt, Max: lifecycle.ReconnectMax, NextAt: lifecycle.ReconnectNextAt, Exhausted: lifecycle.ReconnectExhausted, LastError: lifecycle.ReconnectLastError}
}(),
Queued: lifecycle.Queued, Queued: lifecycle.Queued,
LLMProfileID: llmState.ProfileID, LLMProfileID: llmState.ProfileID,
LLMProfileIDs: profileIDs, LLMProfileIDs: profileIDs,
@@ -325,15 +342,16 @@ type FindingDTO struct {
ReportStale bool `json:"report_stale"` ReportStale bool `json:"report_stale"`
TrafficBindings []db.FindingTrafficBinding `json:"traffic_bindings,omitempty"` TrafficBindings []db.FindingTrafficBinding `json:"traffic_bindings,omitempty"`
ID string `json:"id"` ID string `json:"id"`
FindingID string `json:"finding_id,omitempty"` // standalone findings-table id — the handle for status updates FindingID string `json:"finding_id,omitempty"` // standalone findings-table id — the handle for status updates
VulnClass string `json:"vulnclass"` VulnClass string `json:"vulnclass"`
Name string `json:"name,omitempty"` // 漏洞名称;为空时前端回退展示 vulnclass Name string `json:"name,omitempty"` // 漏洞名称;为空时前端回退展示 vulnclass
Severity string `json:"severity"` // critical | high | medium | low Severity string `json:"severity"` // critical | high | medium | low
Status string `json:"status"` // pending | in_progress | confirmed | resolved | fixed | false_positive | ignored | duplicate | risk_accepted Status string `json:"status"` // pending | in_progress | confirmed | resolved | fixed | false_positive | ignored | duplicate | risk_accepted
Summary string `json:"summary"` Summary string `json:"summary"`
Evidence string `json:"evidence"` Evidence string `json:"evidence"`
Report string `json:"report,omitempty"` // 详细报告(Markdown);仅详情接口返回,列表为空 Report string `json:"report,omitempty"` // 详细报告(Markdown);仅详情接口返回,列表为空
ReportConversationID *int64 `json:"report_conversation_id,omitempty"`
IntentID string `json:"intent_id,omitempty"` IntentID string `json:"intent_id,omitempty"`
ParamID string `json:"param_id,omitempty"` ParamID string `json:"param_id,omitempty"`
@@ -464,14 +482,15 @@ func findingFromDB(f *db.DBFinding, assets map[int64]*db.Asset) FindingDTO {
FindingID: i64s(f.ID), FindingID: i64s(f.ID),
TrafficCount: f.TrafficCount, EvidenceVersion: f.EvidenceVersion, ReportEvidenceVersion: f.ReportEvidenceVersion, TrafficCount: f.TrafficCount, EvidenceVersion: f.EvidenceVersion, ReportEvidenceVersion: f.ReportEvidenceVersion,
ReportStale: f.Report != "" && f.EvidenceVersion != f.ReportEvidenceVersion, TrafficBindings: f.TrafficBindings, ReportStale: f.Report != "" && f.EvidenceVersion != f.ReportEvidenceVersion, TrafficBindings: f.TrafficBindings,
VulnClass: f.VulnClass, VulnClass: f.VulnClass,
Name: f.Name, Name: f.Name,
Severity: f.Severity, Severity: f.Severity,
Status: status, Status: status,
Summary: f.Summary, Summary: f.Summary,
Evidence: f.Evidence, Evidence: f.Evidence,
Report: f.Report, Report: f.Report,
TS: rfc3339(f.CreatedAt), ReportConversationID: f.ReportConversationID,
TS: rfc3339(f.CreatedAt),
} }
if f.ExportCaseID > 0 { if f.ExportCaseID > 0 {
d.ID = "case:" + i64s(f.ExportCaseID) d.ID = "case:" + i64s(f.ExportCaseID)
+17 -4
View File
@@ -341,6 +341,7 @@ func (s *Server) testEgressProxy(w http.ResponseWriter, r *http.Request) {
var req struct { var req struct {
GlobalProxy string `json:"global_proxy"` GlobalProxy string `json:"global_proxy"`
ProbeURL string `json:"probe_url"` // 可选:覆盖默认探测地址 ProbeURL string `json:"probe_url"` // 可选:覆盖默认探测地址
Resume bool `json:"resume"` // manual settings probe may release paused_by_proxy tasks
} }
// 请求体只有一个地址,给个上限,免得被塞大 payload。 // 请求体只有一个地址,给个上限,免得被塞大 payload。
r.Body = http.MaxBytesReader(w, r.Body, 64<<10) r.Body = http.MaxBytesReader(w, r.Body, 64<<10)
@@ -366,15 +367,21 @@ func (s *Server) testEgressProxy(w http.ResponseWriter, r *http.Request) {
agentCA := s.m.ProxyCACert() // 捕获开时是 MITM 的 CA 路径;关时为空 agentCA := s.m.ProxyCACert() // 捕获开时是 MITM 的 CA 路径;关时为空
savedProxy := s.m.GlobalProxy() // 已保存的 global_proxy savedProxy := s.m.GlobalProxy() // 已保存的 global_proxy
// 表单值 ≠ 已保存值时,② 仍按表单值测,但 agent 走的是已保存值——必须提示。
// 归一化再比:尾斜杠不算"未保存"。
pending := normalizeProxy(req.GlobalProxy) != normalizeProxy(savedProxy)
// 整体上限 40s:单次 Timeout 只兜一次,不兜总量;前端 fetch 没有超时。 // 整体上限 40s:单次 Timeout 只兜一次,不兜总量;前端 fetch 没有超时。
ctx, cancel := context.WithTimeout(r.Context(), 40*time.Second) ctx, cancel := context.WithTimeout(r.Context(), 40*time.Second)
defer cancel() defer cancel()
var proxyGuards map[string]proxyReconnectGuard
if req.Resume {
proxyGuards = s.snapshotProxyReconnectGuardsWithExhausted(false, true)
}
configuredProxy := strings.TrimSpace(req.GlobalProxy) configuredProxy := strings.TrimSpace(req.GlobalProxy)
if strings.Contains(configuredProxy, "***") || strings.Contains(configuredProxy, "******") {
configuredProxy = s.m.GlobalProxy()
}
// 表单值 ≠ 已保存值时,② 仍按表单值测,但 agent 走的是已保存值——必须提示。
// 归一化再比:尾斜杠不算"未保存"。
pending := normalizeProxy(configuredProxy) != normalizeProxy(savedProxy)
rep := egressReport{ rep := egressReport{
Capture: capture, Capture: capture,
Configured: maskSecrets(configuredProxy), Configured: maskSecrets(configuredProxy),
@@ -417,6 +424,12 @@ func (s *Server) testEgressProxy(w http.ResponseWriter, r *http.Request) {
rep.OK = judgeEgress(rep) rep.OK = judgeEgress(rep)
rep.Warnings = egressWarnings(rep, pending, savedProxy) rep.Warnings = egressWarnings(rep, pending, savedProxy)
if req.Resume && rep.OK && !pending {
// The navbar uses the same probe endpoint for display-only polling but
// omits resume. Only an explicit settings-page test can release the
// guarded paused_by_proxy tasks.
s.proxyReconnectSuccessWithGuards(proxyGuards, nil)
}
writeJSON(w, 200, rep) writeJSON(w, 200, rep)
} }
+23 -4
View File
@@ -153,9 +153,12 @@ type Engine struct {
slotPlans sync.Map // task -> map[slot name]profile ID; immutable for a run slotPlans sync.Map // task -> map[slot name]profile ID; immutable for a run
// readiness reports whether a global LLM provider is configured — the signal behind // readiness reports whether a global LLM provider is configured — the signal behind
// Ready()/the llm_configured indicator. Wired once at startup; nil → not ready. // Ready()/the llm_configured indicator. Wired once at startup; nil → not ready.
readiness func() bool readiness func() bool
proxyFailure func(taskID string, err error)
} }
func (e *Engine) SetProxyFailureHandler(fn func(string, error)) { e.proxyFailure = fn }
type slotAvailability struct { type slotAvailability struct {
available bool available bool
lastEmit time.Time lastEmit time.Time
@@ -942,7 +945,12 @@ func (e *Engine) plannerLoop(ctx context.Context, t *Task) {
} }
e.stampFirstRun(t) // 首次真正规划 → 盖 first_run_at + 算 deadline(仅带 timeout 的任务) e.stampFirstRun(t) // 首次真正规划 → 盖 first_run_at + 算 deadline(仅带 timeout 的任务)
e.touch(t.ID) e.touch(t.ID)
emit := func(r db.Activity) { e.emitActivity(t, r) } emit := func(r db.Activity) {
if r.Kind == "tool_result" && r.IsError && e.proxyFailure != nil && isGlobalProxyDialFailure(errors.New(r.Detail+" "+r.Summary)) {
e.proxyFailure(t.ID, errors.New(r.Detail+" "+r.Summary))
}
e.emitActivity(t, r)
}
ectx := e.clockCtx(e.execContextFor(ctx, t.ID), t, false) // cancellable by Pause; 带任务 deadline ectx := e.clockCtx(e.execContextFor(ctx, t.ID), t, false) // cancellable by Pause; 带任务 deadline
if ectx.Err() != nil || e.IsDeleting(t.ID) { if ectx.Err() != nil || e.IsDeleting(t.ID) {
return return
@@ -962,6 +970,9 @@ func (e *Engine) plannerLoop(ctx context.Context, t *Task) {
switch { switch {
case err != nil && ectx.Err() == nil: case err != nil && ectx.Err() == nil:
log.Printf("[planner] task %s 规划出错: %v", t.ID, err) log.Printf("[planner] task %s 规划出错: %v", t.ID, err)
if e.proxyFailure != nil && isProxyFailure(err) {
e.proxyFailure(t.ID, err)
}
case met: case met:
log.Printf("[planner] task %s 判定目标达成: %s", t.ID, reason) log.Printf("[planner] task %s 判定目标达成: %s", t.ID, reason)
// 所有目标达成 → 持久化任务状态为 done(前端 DTO 会优先展示该终态)。 // 所有目标达成 → 持久化任务状态为 done(前端 DTO 会优先展示该终态)。
@@ -1081,7 +1092,12 @@ func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *ag
hasChatMessage := message != "" hasChatMessage := message != ""
e.stampFirstRun(t) // 首次真正执行 → 盖 first_run_at + 算 deadline(仅带 timeout 的任务) e.stampFirstRun(t) // 首次真正执行 → 盖 first_run_at + 算 deadline(仅带 timeout 的任务)
e.touch(t.ID) e.touch(t.ID)
emit := func(r db.Activity) { e.emitActivity(t, r) } emit := func(r db.Activity) {
if r.Kind == "tool_result" && r.IsError && e.proxyFailure != nil && isGlobalProxyDialFailure(errors.New(r.Detail+" "+r.Summary)) {
e.proxyFailure(t.ID, errors.New(r.Detail+" "+r.Summary))
}
e.emitActivity(t, r)
}
ectx := e.clockCtx(e.execContextFor(ctx, t.ID), t, false) // cancellable by Pause; 带任务 deadline ectx := e.clockCtx(e.execContextFor(ctx, t.ID), t, false) // cancellable by Pause; 带任务 deadline
if ectx.Err() != nil || e.IsDeleting(t.ID) { if ectx.Err() != nil || e.IsDeleting(t.ID) {
if err := transitionIntentState(t.Store, intent.ID, "running", "open"); err != nil { if err := transitionIntentState(t.Store, intent.ID, "running", "open"); err != nil {
@@ -1124,6 +1140,9 @@ func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *ag
} else { } else {
reason, wrote, err = worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding) reason, wrote, err = worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding)
} }
if err != nil && e.proxyFailure != nil && isProxyFailure(err) {
e.proxyFailure(t.ID, err)
}
e.EndLLMCall(t.ID) e.EndLLMCall(t.ID)
// model_error 收场 → 额外重跑几次(退避后再试)。仅在意图仍属本 work、任务 // model_error 收场 → 额外重跑几次(退避后再试)。仅在意图仍属本 work、任务
// 未暂停/未终止/未取消【且未进入收尾】时重试;否则让位给对应分支处理(收尾期不 // 未暂停/未终止/未取消【且未进入收尾】时重试;否则让位给对应分支处理(收尾期不
@@ -1338,7 +1357,7 @@ func taskExecutionPaused(cause error) bool {
return false return false
} }
switch abort.Code { switch abort.Code {
case "paused_by_schedule", "paused_by_user", "paused_by_orchestrator", "paused_on_reload", "paused_race_guard", case "paused_by_schedule", "paused_by_user", "paused_by_orchestrator", "paused_by_proxy", "paused_on_reload", "paused_race_guard",
"queued_for_admission", "llm_unavailable_queued", "task_deleted": "queued_for_admission", "llm_unavailable_queued", "task_deleted":
return true return true
default: default:
+5
View File
@@ -177,6 +177,11 @@ func TestTaskPauseCauseSurvivesImmediateResume(t *testing.T) {
if old.Err() == nil || !taskExecutionPaused(context.Cause(old)) { if old.Err() == nil || !taskExecutionPaused(context.Cause(old)) {
t.Fatalf("old execution cause=%v was not retained as a task pause", context.Cause(old)) t.Fatalf("old execution cause=%v was not retained as a task pause", context.Cause(old))
} }
proxyCtx, proxyCancel := context.WithCancelCause(context.Background())
proxyCancel(agent.AbortPausedByProxy)
if !taskExecutionPaused(context.Cause(proxyCtx)) {
t.Fatalf("proxy pause cause=%v was not classified as a task pause", context.Cause(proxyCtx))
}
if taskExecutionPaused(agent.AbortGoalMet) { if taskExecutionPaused(agent.AbortGoalMet) {
t.Fatal("goal completion was misclassified as a task pause") t.Fatal("goal completion was misclassified as a task pause")
} }
+294
View File
@@ -0,0 +1,294 @@
package server
import (
"bufio"
"context"
"crypto/tls"
"fmt"
"io"
"net"
"net/http"
"net/url"
"strings"
"sync"
"time"
"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
mu sync.Mutex
conns map[net.Conn]struct{}
ctx context.Context
cancel context.CancelFunc
}
func newDynamicForwardProxy(current func() string) (*dynamicForwardProxy, error) {
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
return nil, err
}
ctx, cancel := context.WithCancel(context.Background())
p := &dynamicForwardProxy{ln: ln, addr: "http://" + ln.Addr().String(), current: current, direct: func() bool { return true }, conns: map[net.Conn]struct{}{}, ctx: ctx, cancel: cancel}
p.server = &http.Server{Handler: http.HandlerFunc(p.serveHTTP), BaseContext: func(net.Listener) context.Context { return ctx }}
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.cancel()
p.mu.Lock()
for c := range p.conns {
_ = c.Close()
}
p.conns = nil
p.mu.Unlock()
_ = p.server.Close()
})
return nil
}
func (p *dynamicForwardProxy) track(c net.Conn) func() {
if c == nil {
return func() {}
}
p.mu.Lock()
if p.conns != nil {
p.conns[c] = struct{}{}
}
p.mu.Unlock()
return func() { p.mu.Lock(); delete(p.conns, c); p.mu.Unlock() }
}
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 dialProxyContext(ctx, d, 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{Timeout: 10 * time.Second}).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 = dialProxyContext(r.Context(), d, "tcp", target)
}
} else {
conn, err = dialHTTPConnect(r.Context(), u, target)
}
}
if err != nil {
http.Error(w, "proxy unavailable", 502)
return
}
h, ok := w.(http.Hijacker)
if !ok {
http.Error(w, "proxy unavailable", 500)
return
}
client, rw, err := h.Hijack()
if err != nil {
return
}
removeClient := p.track(client)
defer removeClient()
removeUpstream := p.track(conn)
defer removeUpstream()
defer client.Close()
defer conn.Close()
_, _ = rw.WriteString("HTTP/1.1 200 Connection Established\r\n\r\n")
_ = rw.Flush()
clientReader := io.Reader(client)
if rw.Reader != nil && rw.Reader.Buffered() > 0 {
clientReader = io.MultiReader(rw.Reader, client)
}
done := make(chan struct{}, 2)
go func() { _, _ = io.Copy(conn, clientReader); _ = conn.Close(); done <- struct{}{} }()
go func() { _, _ = io.Copy(client, conn); _ = client.Close(); done <- struct{}{} }()
<-done
}
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{Timeout: 10 * time.Second})
}
func dialProxyContext(ctx context.Context, d proxy.Dialer, network, addr string) (net.Conn, error) {
if cd, ok := d.(proxy.ContextDialer); ok {
return cd.DialContext(ctx, network, addr)
}
type result struct {
c net.Conn
err error
}
ch := make(chan result, 1)
go func() { c, err := d.Dial(network, addr); ch <- result{c, err} }()
select {
case r := <-ch:
return r.c, r.err
case <-ctx.Done():
go func() {
r := <-ch
if r.c != nil {
_ = r.c.Close()
}
}()
return nil, ctx.Err()
}
}
type bufferedConn struct {
net.Conn
reader *bufio.Reader
}
func (c *bufferedConn) Read(p []byte) (int, error) { return c.reader.Read(p) }
func dialHTTPConnect(ctx context.Context, u *url.URL, target string) (net.Conn, error) {
d := &net.Dialer{Timeout: 10 * time.Second}
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("Authorization"))
}
fmt.Fprint(br, "\r\n")
if err := br.Flush(); err != nil {
_ = conn.Close()
return nil, err
}
readCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
type responseResult struct {
resp *http.Response
err error
}
readDone := make(chan responseResult, 1)
go func() {
resp, err := http.ReadResponse(br.Reader, &http.Request{Method: http.MethodConnect})
readDone <- responseResult{resp: resp, err: err}
}()
var rr responseResult
select {
case rr = <-readDone:
case <-readCtx.Done():
_ = conn.Close()
<-readDone
return nil, readCtx.Err()
}
resp, err := rr.resp, rr.err
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 &bufferedConn{Conn: conn, reader: br.Reader}, nil
}
+28 -2
View File
@@ -112,10 +112,30 @@ func (s *Server) admitTaskWhen(t *Task, mode string, requirePaused bool) (bool,
return s.admitTaskWhenScheduled(t, mode, requirePaused, false) return s.admitTaskWhenScheduled(t, mode, requirePaused, false)
} }
func (s *Server) admitTaskWhenScheduled(t *Task, mode string, requirePaused, scheduled bool) (queued bool, err error) { func (s *Server) admitTaskWhenScheduled(t *Task, mode string, requirePaused, scheduled bool) (queued bool, err error) {
return s.admitTaskWhenScheduledWithReason(t, mode, requirePaused, scheduled, "", false)
}
// admitTaskWhenScheduledWithReason is the shared admission gate. A non-empty
// requiredReason is checked after concMu is acquired, alongside the paused and
// terminal checks, so proxy recovery cannot revive a task that was re-paused by
// a user while the probe was in flight. preserveSchedule keeps calendar state
// and never creates schedule_manual_until (used by proxy recovery).
func (s *Server) admitTaskWhenScheduledWithReason(t *Task, mode string, requirePaused, scheduled bool, requiredReason string, preserveSchedule bool) (queued bool, err error) {
return s.admitTaskWhenScheduledWithGenerationChecked(t, mode, requirePaused, scheduled, requiredReason, preserveSchedule, 0, false)
}
// admitTaskWhenScheduledWithGeneration checks the supplied reconnect round.
// Generation zero is valid; callers that do not have a round use the explicit
// unchecked form through admitTaskWhenScheduledWithReason.
func (s *Server) admitTaskWhenScheduledWithGeneration(t *Task, mode string, requirePaused, scheduled bool, requiredReason string, preserveSchedule bool, requiredGeneration uint64) (queued bool, err error) {
return s.admitTaskWhenScheduledWithGenerationChecked(t, mode, requirePaused, scheduled, requiredReason, preserveSchedule, requiredGeneration, true)
}
func (s *Server) admitTaskWhenScheduledWithGenerationChecked(t *Task, mode string, requirePaused, scheduled bool, requiredReason string, preserveSchedule bool, requiredGeneration uint64, checkGeneration bool) (queued bool, err error) {
if t == nil { if t == nil {
return false, fmt.Errorf("task not found") return false, fmt.Errorf("task not found")
} }
manualSchedule := requirePaused && !scheduled manualSchedule := requirePaused && !scheduled && !preserveSchedule
if manualSchedule { if manualSchedule {
// Match the calendar tick's lock order: scheduleMu before concMu. The // Match the calendar tick's lock order: scheduleMu before concMu. The
// override must commit before another tick can decide to pause this task. // override must commit before another tick can decide to pause this task.
@@ -155,6 +175,12 @@ func (s *Server) admitTaskWhenScheduled(t *Task, mode string, requirePaused, sch
if !lifecycle.Paused { if !lifecycle.Paused {
return false, fmt.Errorf("仅已暂停的任务可以继续") return false, fmt.Errorf("仅已暂停的任务可以继续")
} }
if requiredReason != "" && lifecycle.PauseReason != requiredReason {
return false, fmt.Errorf("任务暂停原因已改变")
}
if requiredReason == "paused_by_proxy" && checkGeneration && lifecycle.ReconnectGeneration != requiredGeneration {
return false, fmt.Errorf("任务代理重连轮次已改变")
}
mode = s.resumeAdmissionMode(t) mode = s.resumeAdmissionMode(t)
} }
var manualUntil *time.Time var manualUntil *time.Time
@@ -212,7 +238,7 @@ func (s *Server) admitTaskWhenScheduled(t *Task, mode string, requirePaused, sch
if wasTerminal { if wasTerminal {
status = "running" status = "running"
} }
if err := s.m.applyTaskAdmission(t.ID, lifecycle.Status, status, shouldQueue, mode, wasQueued, manualUntil); err != nil { if err := s.m.applyTaskAdmissionWithSchedule(t.ID, lifecycle.Status, status, shouldQueue, mode, wasQueued, manualUntil, preserveSchedule); err != nil {
if !engineWasPaused && !wasPaused && !wasQueued { if !engineWasPaused && !wasPaused && !wasQueued {
s.engine.Resume(t) s.engine.Resume(t)
} }
+235 -30
View File
@@ -13,6 +13,7 @@ import (
"strconv" "strconv"
"strings" "strings"
"sync" "sync"
"sync/atomic"
"time" "time"
"github.com/Autumn-27/artex/agent" "github.com/Autumn-27/artex/agent"
@@ -28,19 +29,26 @@ import (
// sharing the process-wide asset store. ID is the PG task id as a string; ExpID // sharing the process-wide asset store. ID is the PG task id as a string; ExpID
// is the exploration the task owns. // is the exploration the task owns.
type Task struct { type Task struct {
ID string `json:"id"` ID string `json:"id"`
ExpID int64 `json:"exploration_id"` ExpID int64 `json:"exploration_id"`
Remark string `json:"remark"` Remark string `json:"remark"`
Name string `json:"name"` // 可选任务名称;空=未命名 Name string `json:"name"` // 可选任务名称;空=未命名
CategoryID *int64 `json:"category_id,omitempty"` CategoryID *int64 `json:"category_id,omitempty"`
CategoryName string `json:"category_name,omitempty"` CategoryName string `json:"category_name,omitempty"`
PinnedAt int64 `json:"pinned_at,omitempty"` PinnedAt int64 `json:"pinned_at,omitempty"`
Description string `json:"description"` Description string `json:"description"`
Goal string `json:"goal"` Goal string `json:"goal"`
CreatedAt int64 `json:"created_at"` CreatedAt int64 `json:"created_at"`
CompletedAt int64 `json:"completed_at,omitempty"` // 进入终态的 unix 秒;0=未完成 CompletedAt int64 `json:"completed_at,omitempty"` // 进入终态的 unix 秒;0=未完成
Paused bool `json:"paused"` Paused bool `json:"paused"`
Queued bool `json:"queued"` // 因并发上限被挂起、等待空位自动启动;true=尚未开跑 PauseReason string `json:"pause_reason,omitempty"`
ReconnectAttempt int `json:"-"`
ReconnectMax int `json:"-"`
ReconnectNextAt int64 `json:"-"`
ReconnectExhausted bool `json:"-"`
ReconnectLastError string `json:"-"`
ReconnectGeneration uint64 `json:"-"`
Queued bool `json:"queued"` // 因并发上限被挂起、等待空位自动启动;true=尚未开跑
// QueuedAt is an internal Unix-nanosecond ordering key. It is deliberately // QueuedAt is an internal Unix-nanosecond ordering key. It is deliberately
// finer than CreatedAt so several tasks enqueued in the same second retain // finer than CreatedAt so several tasks enqueued in the same second retain
// their real FIFO order. // their real FIFO order.
@@ -88,6 +96,13 @@ type taskLifecycleState struct {
PinnedAt int64 PinnedAt int64
Status string Status string
Paused bool Paused bool
PauseReason string
ReconnectAttempt int
ReconnectMax int
ReconnectNextAt int64
ReconnectExhausted bool
ReconnectLastError string
ReconnectGeneration uint64
Queued bool Queued bool
QueuedAt int64 QueuedAt int64
QueueMode string QueueMode string
@@ -113,11 +128,13 @@ func (t *Task) lifecycleSnapshot() taskLifecycleState {
func (t *Task) lifecycleSnapshotLocked() taskLifecycleState { func (t *Task) lifecycleSnapshotLocked() taskLifecycleState {
return taskLifecycleState{ return taskLifecycleState{
Name: t.Name, Name: t.Name,
Remark: t.Remark, Remark: t.Remark,
PinnedAt: t.PinnedAt, PinnedAt: t.PinnedAt,
Status: t.Status, Status: t.Status,
Paused: t.Paused, Paused: t.Paused,
PauseReason: t.PauseReason,
ReconnectAttempt: t.ReconnectAttempt, ReconnectMax: t.ReconnectMax, ReconnectNextAt: t.ReconnectNextAt, ReconnectExhausted: t.ReconnectExhausted, ReconnectLastError: t.ReconnectLastError, ReconnectGeneration: t.ReconnectGeneration,
Queued: t.Queued, Queued: t.Queued,
QueuedAt: t.QueuedAt, QueuedAt: t.QueuedAt,
QueueMode: t.QueueMode, QueueMode: t.QueueMode,
@@ -145,6 +162,13 @@ func (t *Task) updateLifecycle(update func(*taskLifecycleState)) {
t.PinnedAt = state.PinnedAt t.PinnedAt = state.PinnedAt
t.Status = state.Status t.Status = state.Status
t.Paused = state.Paused t.Paused = state.Paused
t.PauseReason = state.PauseReason
t.ReconnectAttempt = state.ReconnectAttempt
t.ReconnectMax = state.ReconnectMax
t.ReconnectNextAt = state.ReconnectNextAt
t.ReconnectExhausted = state.ReconnectExhausted
t.ReconnectLastError = state.ReconnectLastError
t.ReconnectGeneration = state.ReconnectGeneration
t.Queued = state.Queued t.Queued = state.Queued
t.QueuedAt = state.QueuedAt t.QueuedAt = state.QueuedAt
t.QueueMode = state.QueueMode t.QueueMode = state.QueueMode
@@ -242,6 +266,7 @@ type Manager struct {
// their in-memory mirrors. lifecycleMu makes snapshots race-free, but without // their in-memory mirrors. lifecycleMu makes snapshots race-free, but without
// this outer write lock an older request could commit first and publish last. // this outer write lock an older request could commit first and publish last.
taskStateMu sync.Mutex taskStateMu sync.Mutex
controlMu *sync.Mutex // Server's admission/task-control mutex, when attached.
mu sync.RWMutex mu sync.RWMutex
tasks map[string]*Task tasks map[string]*Task
active string active string
@@ -258,9 +283,16 @@ type Manager struct {
// (http/https/socks5, optional user:pass). Empty = direct. When traffic // (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 // capture is on it becomes the MITM's upstream; when capture is off it is
// injected into agent bash env / WebFetch directly. See ProxyAddr. // injected into agent bash env / WebFetch directly. See ProxyAddr.
globalProxy string globalProxy string
proxyPool *proxyPoolState
forwardProxy *dynamicForwardProxy
proxyPoolConfigured atomic.Bool
} }
// SetTaskControlMutex lets lifecycle operations that originate in Manager
// participate in the Server's admission/task-control critical section.
func (m *Manager) SetTaskControlMutex(mu *sync.Mutex) { m.controlMu = mu }
// Settings keys the UI toggles at runtime. // Settings keys the UI toggles at runtime.
const ( const (
settingTrafficCapture = "traffic_capture" settingTrafficCapture = "traffic_capture"
@@ -554,6 +586,14 @@ func NewManager(dir, proxyAddr string) (*Manager, error) {
log.Printf("[llmusage] create table: %v", err) log.Printf("[llmusage] create table: %v", err)
} }
m := &Manager{dir: dir, pg: pg, assets: pg.Assets(), tasks: map[string]*Task{}, interceptor: intercept.New(pg)} 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
// An empty runtime route means direct egress, even when the persisted
// pool still contains disabled or blank entries.
fp.direct = func() bool { return strings.TrimSpace(m.GlobalProxy()) == "" }
} else {
log.Printf("[proxy] local forwarder disabled: %v", ferr)
}
if proxyAddr != "" { if proxyAddr != "" {
tr, err := traffic.Open(filepath.Join(dir, "traffic"), proxyAddr) tr, err := traffic.Open(filepath.Join(dir, "traffic"), proxyAddr)
if err != nil { if err != nil {
@@ -614,6 +654,7 @@ func NewManager(dir, proxyAddr string) (*Manager, error) {
if v, ok, _ := pg.GetSetting(settingGlobalProxy); ok { if v, ok, _ := pg.GetSetting(settingGlobalProxy); ok {
m.globalProxy = strings.TrimSpace(v) m.globalProxy = strings.TrimSpace(v)
} }
m.initProxyPool()
if m.traffic != nil { if m.traffic != nil {
if err := m.traffic.SetUpstreamProxy(m.globalProxy); err != nil { if err := m.traffic.SetUpstreamProxy(m.globalProxy); err != nil {
log.Printf("[proxy] 全局代理 %q 无效,已忽略: %v", m.globalProxy, err) log.Printf("[proxy] 全局代理 %q 无效,已忽略: %v", m.globalProxy, err)
@@ -988,8 +1029,12 @@ func (m *Manager) ProxyAddr() string {
return m.traffic.ProxyAddr() return m.traffic.ProxyAddr()
} }
m.mu.RLock() m.mu.RLock()
defer m.mu.RUnlock() global := m.globalProxy
return 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 // ProxyCACert returns the CA cert path agents must trust to verify HTTPS through
@@ -1019,7 +1064,30 @@ func (m *Manager) SetGlobalProxy(raw string) error {
raw = strings.TrimSpace(raw) raw = strings.TrimSpace(raw)
if raw != "" { if raw != "" {
if _, err := traffic.ValidateProxyURL(raw); err != nil { if _, err := traffic.ValidateProxyURL(raw); err != nil {
return err return fmt.Errorf("%s", maskSecrets(err.Error()))
}
}
if m.proxyPool != nil {
m.proxyPool.mu.Lock()
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 { if err := m.pg.SetSetting(settingGlobalProxy, raw); err != nil {
@@ -1033,12 +1101,14 @@ func (m *Manager) SetGlobalProxy(raw string) error {
return err return err
} }
} }
// Keep the browser MCP's egress in sync with the new global proxy too.
m.syncBrowserMCPProxy() m.syncBrowserMCPProxy()
return nil return nil
} }
func (m *Manager) Close() error { func (m *Manager) Close() error {
if m.forwardProxy != nil {
_ = m.forwardProxy.Close()
}
m.mu.Lock() m.mu.Lock()
defer m.mu.Unlock() defer m.mu.Unlock()
if m.traffic != nil { if m.traffic != nil {
@@ -1073,7 +1143,7 @@ func taskFromPG(pt *pgdb.Task, store *pgdb.ExplorationStore, ic *intercept.Inter
Remark: pt.Remark, Remark: pt.Remark,
CategoryID: cloneInt64Ptr(pt.CategoryID), CategoryName: pt.CategoryName, CategoryID: cloneInt64Ptr(pt.CategoryID), CategoryName: pt.CategoryName,
PinnedAt: unixOrZero(pt.PinnedAt), PinnedAt: unixOrZero(pt.PinnedAt),
Description: pt.Description, Goal: pt.Goal, CreatedAt: pt.CreatedAt.Unix(), Paused: pt.Paused, Queued: pt.Queued, Description: pt.Description, Goal: pt.Goal, CreatedAt: pt.CreatedAt.Unix(), Paused: pt.Paused, PauseReason: pt.PauseReason, Queued: pt.Queued,
QueuedAt: unixNanoOrZero(pt.QueuedAt), QueueMode: pt.QueueMode, QueuedAt: unixNanoOrZero(pt.QueuedAt), QueueMode: pt.QueueMode,
CompletedAt: unixOrZero(pt.CompletedAt), Status: pt.Status, ParentRef: pt.ParentRef, CompletedAt: unixOrZero(pt.CompletedAt), Status: pt.Status, ParentRef: pt.ParentRef,
LLMProfileID: pt.LLMProfileID, LLMProfileID: pt.LLMProfileID,
@@ -1391,12 +1461,16 @@ func (m *Manager) SetTaskPaused(id string, paused bool) error {
// admission keeps its FIFO timestamp; a task that was explicitly paused and is // admission keeps its FIFO timestamp; a task that was explicitly paused and is
// now re-queued receives a fresh tail position. // now re-queued receives a fresh tail position.
func (m *Manager) ApplyTaskAdmission(id, expectedStatus, status string, queued bool, mode string, preservePosition bool) error { func (m *Manager) ApplyTaskAdmission(id, expectedStatus, status string, queued bool, mode string, preservePosition bool) error {
return m.applyTaskAdmission(id, expectedStatus, status, queued, mode, preservePosition, nil) return m.applyTaskAdmissionWithSchedule(id, expectedStatus, status, queued, mode, preservePosition, nil, false)
} }
// Manual task control records its schedule override in the admission commit, so // Manual task control records its schedule override in the admission commit, so
// failed resumes cannot bypass a calendar and FIFO promotion preserves it. // failed resumes cannot bypass a calendar and FIFO promotion preserves it.
func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued bool, mode string, preservePosition bool, manualUntil *time.Time) error { func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued bool, mode string, preservePosition bool, manualUntil *time.Time) error {
return m.applyTaskAdmissionWithSchedule(id, expectedStatus, status, queued, mode, preservePosition, manualUntil, false)
}
func (m *Manager) applyTaskAdmissionWithSchedule(id, expectedStatus, status string, queued bool, mode string, preservePosition bool, manualUntil *time.Time, preserveSchedule bool) error {
m.taskStateMu.Lock() m.taskStateMu.Lock()
defer m.taskStateMu.Unlock() defer m.taskStateMu.Unlock()
n, err := strconv.ParseInt(id, 10, 64) n, err := strconv.ParseInt(id, 10, 64)
@@ -1420,8 +1494,9 @@ func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued b
WHEN $2 IN ('done','failed','timeout') THEN COALESCE(completed_at, now()) WHEN $2 IN ('done','failed','timeout') THEN COALESCE(completed_at, now())
ELSE NULL ELSE NULL
END, END,
paused=false, schedule_paused=false, paused=false, pause_reason='',
schedule_manual_until=COALESCE($7::timestamptz, schedule_manual_until), schedule_paused=CASE WHEN $8 THEN schedule_paused ELSE false END,
schedule_manual_until=CASE WHEN $8 THEN schedule_manual_until ELSE COALESCE($7::timestamptz, schedule_manual_until) END,
queued=$3, queued=$3,
queued_at=CASE queued_at=CASE
WHEN NOT $3 THEN NULL WHEN NOT $3 THEN NULL
@@ -1444,7 +1519,7 @@ func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued b
,active_run_seconds=active_run_seconds + CASE WHEN ($3 OR $2 IN ('done','failed','timeout')) AND run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END ,active_run_seconds=active_run_seconds + CASE WHEN ($3 OR $2 IN ('done','failed','timeout')) AND run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END
,run_segment_started_at=CASE WHEN $3 OR $2 IN ('done','failed','timeout') THEN NULL ELSE run_segment_started_at END ,run_segment_started_at=CASE WHEN $3 OR $2 IN ('done','failed','timeout') THEN NULL ELSE run_segment_started_at END
WHERE id=$1 AND deleted_at IS NULL AND status=$6 WHERE id=$1 AND deleted_at IS NULL AND status=$6
RETURNING queued_at, queue_mode, completed_at, first_run_at, deadline_at, active_run_seconds, run_segment_started_at`, n, status, queued, mode, preservePosition, expectedStatus, manualUntil). RETURNING queued_at, queue_mode, completed_at, first_run_at, deadline_at, active_run_seconds, run_segment_started_at`, n, status, queued, mode, preservePosition, expectedStatus, manualUntil, preserveSchedule).
Scan(&queuedAt, &committedMode, &completedAt, &firstRunAt, &deadlineAt, &activeRunSeconds, &segmentStartedAt) Scan(&queuedAt, &committedMode, &completedAt, &firstRunAt, &deadlineAt, &activeRunSeconds, &segmentStartedAt)
if errors.Is(err, sql.ErrNoRows) { if errors.Is(err, sql.ErrNoRows) {
return fmt.Errorf("task %s lifecycle changed before admission (expected status %q)", id, expectedStatus) return fmt.Errorf("task %s lifecycle changed before admission (expected status %q)", id, expectedStatus)
@@ -1458,6 +1533,7 @@ func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued b
t.updateLifecycle(func(state *taskLifecycleState) { t.updateLifecycle(func(state *taskLifecycleState) {
state.Status = status state.Status = status
state.Paused = false state.Paused = false
state.PauseReason = ""
state.Queued = queued state.Queued = queued
state.QueueMode = committedMode state.QueueMode = committedMode
state.QueuedAt = 0 state.QueuedAt = 0
@@ -1481,6 +1557,11 @@ func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued b
if segmentStartedAt.Valid { if segmentStartedAt.Valid {
state.RunSegmentStartedAt = segmentStartedAt.Time.Unix() state.RunSegmentStartedAt = segmentStartedAt.Time.Unix()
} }
state.ReconnectAttempt = 0
state.ReconnectMax = 0
state.ReconnectNextAt = 0
state.ReconnectExhausted = false
state.ReconnectLastError = ""
}) })
} }
m.mu.Unlock() m.mu.Unlock()
@@ -1494,6 +1575,12 @@ func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued b
func (m *Manager) ApplyTaskPause(id string) error { return m.ApplyTaskPauseOrigin(id, false) } func (m *Manager) ApplyTaskPause(id string) error { return m.ApplyTaskPauseOrigin(id, false) }
func (m *Manager) ApplyTaskPauseOrigin(id string, scheduled bool) error { func (m *Manager) ApplyTaskPauseOrigin(id string, scheduled bool) error {
return m.ApplyTaskPauseOriginReason(id, map[bool]string{true: "paused_by_schedule", false: "paused_by_user"}[scheduled], scheduled)
}
// ApplyTaskPauseOriginReason persists a pause cause and keeps it on the live handle.
// reason is a stable API value such as user, schedule, orchestrator, or proxy.
func (m *Manager) ApplyTaskPauseOriginReason(id, reason string, scheduled bool) error {
m.taskStateMu.Lock() m.taskStateMu.Lock()
defer m.taskStateMu.Unlock() defer m.taskStateMu.Unlock()
n, err := strconv.ParseInt(id, 10, 64) n, err := strconv.ParseInt(id, 10, 64)
@@ -1504,12 +1591,12 @@ func (m *Manager) ApplyTaskPauseOrigin(id string, scheduled bool) error {
var activeRunSeconds int64 var activeRunSeconds int64
var segmentStartedAt sql.NullTime var segmentStartedAt sql.NullTime
err = m.pg.QueryRow(`UPDATE tasks err = m.pg.QueryRow(`UPDATE tasks
SET paused=true, queued=false, queued_at=NULL, schedule_paused=$2, schedule_manual_until=NULL, SET paused=true, pause_reason=$3, queued=false, queued_at=NULL, schedule_paused=$2, schedule_manual_until=NULL,
active_run_seconds=active_run_seconds + CASE WHEN run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END, active_run_seconds=active_run_seconds + CASE WHEN run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
run_segment_started_at=NULL run_segment_started_at=NULL
WHERE id=$1 AND deleted_at IS NULL AND paused=false WHERE id=$1 AND deleted_at IS NULL AND paused=false
AND status NOT IN ('done','failed','timeout') AND status NOT IN ('done','failed','timeout')
RETURNING COALESCE(queue_mode,''), active_run_seconds, run_segment_started_at`, n, scheduled).Scan(&mode, &activeRunSeconds, &segmentStartedAt) RETURNING COALESCE(queue_mode,''), active_run_seconds, run_segment_started_at`, n, scheduled, reason).Scan(&mode, &activeRunSeconds, &segmentStartedAt)
if errors.Is(err, sql.ErrNoRows) { if errors.Is(err, sql.ErrNoRows) {
return fmt.Errorf("task %s is unavailable for pause", id) return fmt.Errorf("task %s is unavailable for pause", id)
} }
@@ -1520,6 +1607,12 @@ func (m *Manager) ApplyTaskPauseOrigin(id string, scheduled bool) error {
if t := m.tasks[id]; t != nil { if t := m.tasks[id]; t != nil {
t.updateLifecycle(func(state *taskLifecycleState) { t.updateLifecycle(func(state *taskLifecycleState) {
state.Paused = true state.Paused = true
state.PauseReason = reason
state.ReconnectAttempt = 0
state.ReconnectMax = 0
state.ReconnectNextAt = 0
state.ReconnectExhausted = false
state.ReconnectLastError = ""
state.Queued = false state.Queued = false
state.QueuedAt = 0 state.QueuedAt = 0
state.QueueMode = mode state.QueueMode = mode
@@ -1534,6 +1627,110 @@ func (m *Manager) ApplyTaskPauseOrigin(id string, scheduled bool) error {
return nil return nil
} }
// ApplyProxyPause atomically pauses a running task and marks it eligible for
// reconnect auto-resume. Already-paused tasks retain their original reason.
func (m *Manager) ApplyProxyPause(id string) error {
_, err := m.ApplyProxyPauseWithCallback(id, nil)
return err
}
// ApplyProxyPauseWithCallback is the locked proxy-pause transition. When the
// task actually changes from running to paused, callback runs while the
// Server's task-control mutex is still held, after the DB and live lifecycle
// state have committed. This lets callers install the Engine barrier before a
// concurrent user/orchestrator pause can observe the new paused state.
func (m *Manager) ApplyProxyPauseWithCallback(id string, callback func()) (transitioned bool, err error) {
if m.controlMu != nil {
m.controlMu.Lock()
defer m.controlMu.Unlock()
}
m.taskStateMu.Lock()
defer m.taskStateMu.Unlock()
n, err := strconv.ParseInt(id, 10, 64)
if err != nil {
return false, err
}
var mode string
var active int64
var started sql.NullTime
err = m.pg.QueryRow(`UPDATE tasks SET paused=true, pause_reason='paused_by_proxy', queued=false, queued_at=NULL,
active_run_seconds=active_run_seconds + CASE WHEN run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
run_segment_started_at=NULL WHERE id=$1 AND deleted_at IS NULL AND paused=false AND status NOT IN ('done','failed','timeout')
RETURNING COALESCE(queue_mode,''), active_run_seconds, run_segment_started_at`, n).Scan(&mode, &active, &started)
if errors.Is(err, sql.ErrNoRows) {
return false, nil
}
if err != nil {
return false, err
}
m.mu.RLock()
t := m.tasks[id]
m.mu.RUnlock()
if t != nil {
t.updateLifecycle(func(s *taskLifecycleState) {
s.Paused = true
s.PauseReason = "paused_by_proxy"
s.Queued = false
s.QueuedAt = 0
s.QueueMode = mode
s.ActiveRunSeconds = active
s.RunSegmentStartedAt = 0
})
}
if callback != nil {
callback()
}
return true, nil
}
// ClearPauseReason removes auto-resume eligibility without changing lifecycle.
func (m *Manager) ClearPauseReason(id string) error {
if _, err := m.pg.Exec(`UPDATE tasks SET pause_reason='' WHERE id=$1 AND pause_reason='paused_by_proxy'`, id); err != nil {
return err
}
m.mu.RLock()
t := m.tasks[id]
m.mu.RUnlock()
if t != nil && t.lifecycleSnapshot().PauseReason == "paused_by_proxy" {
t.updateLifecycle(func(s *taskLifecycleState) {
s.PauseReason = ""
s.ReconnectAttempt, s.ReconnectMax, s.ReconnectNextAt = 0, 0, 0
s.ReconnectExhausted, s.ReconnectLastError = false, ""
})
}
return nil
}
// ConvertProxyPauseToSchedule records that a calendar window ended while a
// task was waiting for proxy recovery. It preserves the Engine pause barrier
// and removes proxy auto-resume eligibility until the next window.
func (m *Manager) ConvertProxyPauseToSchedule(id string) error {
m.taskStateMu.Lock()
defer m.taskStateMu.Unlock()
n, err := strconv.ParseInt(id, 10, 64)
if err != nil {
return err
}
res, err := m.pg.Exec(`UPDATE tasks SET pause_reason='paused_by_schedule', schedule_paused=true WHERE id=$1 AND paused=true AND pause_reason='paused_by_proxy' AND status NOT IN ('done','failed','timeout')`, n)
if err != nil {
return err
}
if rows, _ := res.RowsAffected(); rows == 0 {
return nil
}
m.mu.RLock()
t := m.tasks[id]
m.mu.RUnlock()
if t != nil {
t.updateLifecycle(func(v *taskLifecycleState) {
v.PauseReason = "paused_by_schedule"
v.ReconnectAttempt, v.ReconnectMax, v.ReconnectNextAt = 0, 0, 0
v.ReconnectExhausted, v.ReconnectLastError = false, ""
})
}
return nil
}
// EnqueueTask persists the concurrency hold and syncs the in-memory handle. // EnqueueTask persists the concurrency hold and syncs the in-memory handle.
func (m *Manager) EnqueueTask(id, mode string) error { func (m *Manager) EnqueueTask(id, mode string) error {
m.taskStateMu.Lock() m.taskStateMu.Lock()
@@ -1748,6 +1945,10 @@ func (m *Manager) SetTaskStatusGuarded(id, status string) (won bool, err error)
state.RunSegmentStartedAt = unixOrZero(started) state.RunSegmentStartedAt = unixOrZero(started)
if pgdb.IsTerminal(status) { if pgdb.IsTerminal(status) {
state.RunSegmentStartedAt = 0 state.RunSegmentStartedAt = 0
state.Paused = false
state.PauseReason = ""
state.ReconnectAttempt, state.ReconnectMax, state.ReconnectNextAt = 0, 0, 0
state.ReconnectExhausted, state.ReconnectLastError = false, ""
if state.CompletedAt == 0 { if state.CompletedAt == 0 {
state.CompletedAt = time.Now().Unix() state.CompletedAt = time.Now().Unix()
} }
@@ -1784,6 +1985,10 @@ func (m *Manager) SetTaskStatus(id, status string) error {
state.RunSegmentStartedAt = unixOrZero(started) state.RunSegmentStartedAt = unixOrZero(started)
if pgdb.IsTerminal(status) { if pgdb.IsTerminal(status) {
state.RunSegmentStartedAt = 0 state.RunSegmentStartedAt = 0
state.Paused = false
state.PauseReason = ""
state.ReconnectAttempt, state.ReconnectMax, state.ReconnectNextAt = 0, 0, 0
state.ReconnectExhausted, state.ReconnectLastError = false, ""
if state.CompletedAt == 0 { if state.CompletedAt == 0 {
state.CompletedAt = time.Now().Unix() state.CompletedAt = time.Now().Unix()
} }
+462
View File
@@ -0,0 +1,462 @@
package server
import (
"context"
"errors"
"fmt"
"log"
"net/http"
"strings"
"sync"
"time"
"github.com/Autumn-27/artex/agent"
)
const (
settingProxyReconnectMax = "proxy_reconnect_max_attempts"
settingProxyReconnectInterval = "proxy_reconnect_interval_seconds"
defaultProxyReconnectMax = 5
defaultProxyReconnectInterval = 30
)
// isProxyFailure is used for LLM/planner errors, where the error has no trusted
// indication that it came from the global target proxy. Keep this deliberately
// narrow: generic EOF/reset/refused errors belong to the model_error retry path.
func isProxyFailure(err error) bool {
if err == nil {
return false
}
s := strings.ToLower(err.Error())
for _, part := range []string{"proxyconnect", "proxy connection", "proxy error", "proxy handshake", "socks handshake", "failed to connect to proxy", "connect to proxy"} {
if strings.Contains(s, part) {
return true
}
}
return false
}
// isGlobalProxyDialFailure is only used by the MITM upstream callback. That
// callback is on the actual configured global-proxy path; the saved proxy is
// still probed before any task is paused, so a target-side RequestError cannot
// pause tasks while the proxy itself remains healthy.
func isGlobalProxyDialFailure(err error) bool {
if err == nil {
return false
}
s := strings.ToLower(err.Error())
for _, part := range []string{"proxyconnect", "proxy connection", "proxy error", "proxy handshake", "socks", "connect to proxy", "upstream proxy", "connection refused", "connection reset", "network is unreachable", "broken pipe", "unexpected eof", "eof"} {
if strings.Contains(s, part) {
return true
}
}
return false
}
type proxyReconnectSupervisor struct {
mu sync.Mutex
wake chan struct{}
}
type proxyReconnectGuard struct {
generation uint64
attempt int
checkAttempt bool
}
func (s *Server) snapshotProxyReconnectGuards(checkAttempt bool) map[string]proxyReconnectGuard {
return s.snapshotProxyReconnectGuardsWithExhausted(checkAttempt, false)
}
func (s *Server) snapshotProxyReconnectGuardsWithExhausted(checkAttempt, includeExhausted bool) map[string]proxyReconnectGuard {
guards := make(map[string]proxyReconnectGuard)
if s == nil || s.m == nil {
return guards
}
for _, t := range s.m.List() {
st := t.lifecycleSnapshot()
if st.PauseReason != "paused_by_proxy" {
continue
}
if st.ReconnectExhausted && !includeExhausted {
continue
}
guards[t.ID] = proxyReconnectGuard{
generation: st.ReconnectGeneration,
attempt: st.ReconnectAttempt,
checkAttempt: checkAttempt,
}
}
return guards
}
func proxyReconnectGuardMatches(st taskLifecycleState, guard proxyReconnectGuard) bool {
return st.PauseReason == "paused_by_proxy" && st.ReconnectGeneration == guard.generation &&
(!guard.checkAttempt || st.ReconnectAttempt == guard.attempt)
}
func (s *Server) ensureProxySupervisor() {
s.proxyReconnect.mu.Lock()
if s.proxyReconnect.wake == nil {
s.proxyReconnect.wake = make(chan struct{}, 1)
go s.runProxyReconnect(s.ctx)
}
s.proxyReconnect.mu.Unlock()
}
func (s *Server) signalProxySupervisor() {
s.ensureProxySupervisor()
select {
case s.proxyReconnect.wake <- struct{}{}:
default:
}
}
func (s *Server) proxyReconnectConfig() (int, time.Duration) {
max := defaultProxyReconnectMax
interval := defaultProxyReconnectInterval
if s == nil || s.m == nil || s.m.pg == nil {
return max, time.Duration(interval) * time.Second
}
if v, ok, _ := s.m.pg.GetSetting(settingProxyReconnectMax); ok {
if n, err := fmt.Sscanf(strings.TrimSpace(v), "%d", &max); err != nil || n != 1 || max <= 0 {
max = defaultProxyReconnectMax
}
}
if v, ok, _ := s.m.pg.GetSetting(settingProxyReconnectInterval); ok {
if n, err := fmt.Sscanf(strings.TrimSpace(v), "%d", &interval); err != nil || n != 1 || interval <= 0 {
interval = defaultProxyReconnectInterval
}
}
return max, time.Duration(interval) * time.Second
}
func (s *Server) runProxyProbe(ctx context.Context, proxyRaw, caPEMPath string, probes []string, timeout time.Duration) egressProbeResult {
if s != nil && s.proxyProbe != nil {
return s.proxyProbe(ctx, proxyRaw, caPEMPath, probes, timeout)
}
return probeEgress(ctx, proxyRaw, caPEMPath, probes, timeout)
}
func (s *Server) handleProxyFailure(taskID string, cause error) {
if strings.TrimSpace(s.m.GlobalProxy()) == "" {
return
}
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)
}
func (s *Server) pauseProxyTask(taskID string, cause error) {
if strings.TrimSpace(s.m.GlobalProxy()) == "" {
return
}
max, interval := s.proxyReconnectConfig()
now := time.Now().Add(interval).Unix()
lastError := truncateProxyError(cause)
transitioned, err := s.m.ApplyProxyPauseWithCallback(taskID, func() {
if s.engine != nil {
s.engine.Pause(taskID, agent.AbortPausedByProxy)
}
if t, ok := s.m.Task(taskID); ok {
t.updateLifecycle(func(st *taskLifecycleState) {
// A successful admission clears the prior round's counters. This
// transition starts a fresh round, so make that reset atomic with
// the persisted proxy pause and install a new generation.
st.ReconnectGeneration++
st.ReconnectAttempt = 0
st.ReconnectMax = max
st.ReconnectNextAt = now
st.ReconnectExhausted = false
st.ReconnectLastError = lastError
})
}
})
if err != nil {
log.Printf("[proxy] pause task %s: %v", taskID, err)
return
}
if !transitioned {
// A no-op means the task was already paused (possibly by a user,
// schedule, or an exhausted proxy attempt). Preserve its reason and
// reconnect state exactly as-is.
return
}
s.cancelTaskChat(taskID, agent.AbortChatPausedWithTask)
s.ensureProxySupervisor()
s.signalProxySupervisor()
}
func (s *Server) handleProxyTrafficFailure(cause error) {
if s.handleProxyPoolFailure(cause) {
return
}
if !isGlobalProxyDialFailure(cause) || strings.TrimSpace(s.m.GlobalProxy()) == "" {
return
}
// Confirm the saved global proxy itself before pausing. MITM RequestError can
// also represent a target-side failure, and those must not pause a task.
if probe := probeEgress(context.Background(), s.m.GlobalProxy(), "", defaultEgressProbes, 10*time.Second); probe.OK {
return
}
for _, t := range s.m.List() {
st := t.lifecycleSnapshot()
if isTerminalStatus(st.Status) || st.Paused || s.engine.IsDeleting(t.ID) {
continue
}
s.pauseProxyTask(t.ID, cause)
}
}
func (s *Server) handleProxyToolFailure(ctx context.Context, cause error) {
if s.handleProxyPoolFailure(cause) {
return
}
if !isGlobalProxyDialFailure(cause) || strings.TrimSpace(s.m.GlobalProxy()) == "" {
return
}
if probe := probeEgress(context.Background(), s.m.GlobalProxy(), "", defaultEgressProbes, 10*time.Second); probe.OK {
return
}
if info := agent.RunInfoFrom(ctx); info.TaskID > 0 {
s.pauseProxyTask(fmt.Sprint(info.TaskID), cause)
}
}
func truncateProxyError(err error) string {
if err == nil {
return ""
}
s := maskSecrets(err.Error())
if len(s) > 240 {
return s[:240] + "…"
}
return s
}
func (s *Server) runProxyReconnect(ctx context.Context) {
ticker := time.NewTicker(500 * time.Millisecond)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
s.proxyReconnectTick()
case <-s.proxyReconnect.wake:
s.proxyReconnectTick()
}
}
}
func (s *Server) proxyReconnectTick() {
if strings.TrimSpace(s.m.GlobalProxy()) == "" {
// Clearing the runtime route is a successful direct transition. Reuse the
// generation gate so only eligible, non-exhausted proxy pauses resume.
s.proxyReconnectSuccess()
return
}
if s.m.proxyPool != nil && len(s.m.proxyPoolEntries()) > 0 {
// A pool tick may confirm the active route or switch to another available
// member. If it cannot confirm either, retain the original #106 per-task
// attempt/backoff/exhaustion loop (especially for a one-member legacy
// pool).
outcome := s.proxyPoolTick()
if outcome.confirmed || outcome.switched {
return
}
}
now := time.Now().Unix()
max, interval := s.proxyReconnectConfig()
for _, t := range s.m.List() {
st := t.lifecycleSnapshot()
if isTerminalStatus(st.Status) {
t.updateLifecycle(func(v *taskLifecycleState) {
if v.PauseReason == "paused_by_proxy" {
v.PauseReason = ""
}
v.ReconnectAttempt, v.ReconnectMax, v.ReconnectNextAt = 0, 0, 0
v.ReconnectExhausted, v.ReconnectLastError = false, ""
})
continue
}
if st.PauseReason != "paused_by_proxy" || st.ReconnectExhausted || (st.ReconnectNextAt > 0 && st.ReconnectNextAt > now) {
continue
}
attempt := st.ReconnectAttempt + 1
round := st.ReconnectGeneration
claimed := false
t.updateLifecycle(func(v *taskLifecycleState) {
if v.PauseReason != "paused_by_proxy" || v.ReconnectGeneration != round || v.ReconnectAttempt != st.ReconnectAttempt {
return
}
v.ReconnectAttempt = attempt
v.ReconnectMax = max
v.ReconnectNextAt = 0
claimed = true
})
if !claimed {
continue
}
res := s.runProxyProbe(context.Background(), s.m.GlobalProxy(), "", defaultEgressProbes, 10*time.Second)
if res.OK {
s.proxyReconnectSuccessWithGuards(map[string]proxyReconnectGuard{
t.ID: {generation: round, attempt: attempt, checkAttempt: true},
}, nil)
return
}
err := res.Error
exhausted := attempt >= max
next := time.Now().Add(interval).Unix()
t.updateLifecycle(func(v *taskLifecycleState) {
if v.PauseReason != "paused_by_proxy" || v.ReconnectGeneration != round || v.ReconnectAttempt != attempt {
return
}
v.ReconnectNextAt = next
v.ReconnectExhausted = exhausted
v.ReconnectLastError = err
})
log.Printf("[proxy] reconnect attempt %d/%d failed: %s", attempt, max, err)
}
}
func (s *Server) proxyReconnectSuccess() {
s.proxyReconnectSuccessWithGuards(s.snapshotProxyReconnectGuards(false), nil)
}
func (s *Server) proxyReconnectSuccessFor(taskID string, generation uint64, attempt int) {
s.proxyReconnectSuccessWithGuards(map[string]proxyReconnectGuard{
taskID: {generation: generation, attempt: attempt, checkAttempt: true},
}, nil)
}
// proxyReconnectSuccessWithHook is kept small and private so tests can
// deterministically place a competing pause immediately after admission has
// released concMu. Production callers use proxyReconnectSuccess without a
// hook.
func (s *Server) proxyReconnectSuccessWithHook(afterAdmission func()) {
s.proxyReconnectSuccessWithGuards(s.snapshotProxyReconnectGuards(false), afterAdmission)
}
func (s *Server) proxyReconnectSuccessWithGuard(expectedID string, expectedGeneration uint64, expectedAttempt int, afterAdmission func()) {
guards := s.snapshotProxyReconnectGuards(false)
if expectedID != "" {
guards = map[string]proxyReconnectGuard{
expectedID: {generation: expectedGeneration, attempt: expectedAttempt, checkAttempt: true},
}
}
s.proxyReconnectSuccessWithGuards(guards, afterAdmission)
}
func (s *Server) proxyReconnectSuccessWithGuards(guards map[string]proxyReconnectGuard, afterAdmission func()) {
max, interval := s.proxyReconnectConfig()
s.scheduleMu.Lock()
defer s.scheduleMu.Unlock()
for _, t := range s.m.List() {
guard, ok := guards[t.ID]
if !ok {
continue
}
if !proxyReconnectGuardMatches(t.lifecycleSnapshot(), guard) {
continue
}
if s.m.pg != nil {
inside, scheduleErr := s.proxyResumeInsideSchedule(t.ID, time.Now())
if scheduleErr != nil {
log.Printf("[proxy] schedule gate for task %s: %v", t.ID, scheduleErr)
continue
}
if !inside {
if err := s.m.ConvertProxyPauseToSchedule(t.ID); err != nil {
log.Printf("[proxy] schedule pause task %s: %v", t.ID, err)
}
continue
}
}
var err error
if s.engine != nil {
_, err = s.admitTaskWhenScheduledWithGenerationChecked(t, "resume", true, false, "paused_by_proxy", true, guard.generation, true)
if afterAdmission != nil {
afterAdmission()
}
} else {
err = s.m.ClearPauseReason(t.ID)
}
if err != nil {
// Admission re-checks the persisted reason under concMu. A user pause,
// terminal transition, delete, or archive therefore cannot be revived;
// only keep probing if the proxy pause is still eligible.
st := t.lifecycleSnapshot()
if proxyReconnectGuardMatches(st, guard) && !isTerminalStatus(st.Status) {
t.updateLifecycle(func(v *taskLifecycleState) {
v.ReconnectMax = max
v.ReconnectNextAt = time.Now().Add(interval).Unix()
})
}
log.Printf("[proxy] resume task %s: %v", t.ID, err)
continue
}
// The non-queued path's startAdmittedTask already resumes the Engine
// while concMu is held. Do not Resume again after releasing the lock:
// a concurrent user/orchestrator pause could have installed a newer
// barrier in the meantime.
t.updateLifecycle(func(v *taskLifecycleState) {
if !proxyReconnectGuardMatches(*v, guard) {
return
}
v.ReconnectAttempt = 0
v.ReconnectMax = 0
v.ReconnectNextAt = 0
v.ReconnectExhausted = false
v.ReconnectLastError = ""
})
}
}
func (s *Server) testTaskProxy(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
t, ok := s.m.Task(id)
if !ok {
writeErr(w, 404, "task not found")
return
}
proxy := strings.TrimSpace(s.m.GlobalProxy())
guards := s.snapshotProxyReconnectGuardsWithExhausted(false, true)
if proxy == "" {
writeJSON(w, 200, map[string]any{"ok": true, "direct": true})
s.proxyReconnectSuccessWithGuards(guards, nil)
return
}
res := s.runProxyProbe(r.Context(), proxy, "", defaultEgressProbes, 10*time.Second)
if res.OK {
s.proxyReconnectSuccessWithGuards(guards, nil)
}
if !res.OK && res.Error == "" {
res.Error = "代理探测失败"
}
writeJSON(w, 200, map[string]any{"ok": res.OK, "error": res.Error, "task_id": t.ID})
}
func (s *Server) setProxyReconnectSettings(max, interval *int) error {
if max != nil {
if *max < 1 || *max > 1000 {
return errors.New("proxy reconnect max must be 1..1000")
}
if err := s.m.pg.SetSetting(settingProxyReconnectMax, fmt.Sprint(*max)); err != nil {
return err
}
}
if interval != nil {
if *interval < 1 || *interval > 86400 {
return errors.New("proxy reconnect interval must be 1..86400 seconds")
}
if err := s.m.pg.SetSetting(settingProxyReconnectInterval, fmt.Sprint(*interval)); err != nil {
return err
}
}
return nil
}
+500
View File
@@ -0,0 +1,500 @@
package server
import (
"context"
"errors"
"sync"
"testing"
"time"
"github.com/Autumn-27/artex/agent"
)
func proxyTestManager(t *testing.T) *Manager {
t.Helper()
m, err := NewManager(t.TempDir(), "")
if err != nil {
t.Skipf("postgres unavailable (%v) - skipping", err)
}
t.Cleanup(func() { _ = m.Close() })
return m
}
func TestProxyFailureClassification(t *testing.T) {
for _, tc := range []struct {
name string
err error
want bool
}{
{"proxy connect", errors.New("proxyconnect tcp: connection refused"), true},
{"reset", errors.New("dial tcp: connection reset by peer"), false},
{"eof", errors.New("unexpected EOF"), false},
{"quota", errors.New("status 429: insufficient_quota"), false},
{"auth", errors.New("status 401: invalid api key"), false},
} {
t.Run(tc.name, func(t *testing.T) {
if got := isProxyFailure(tc.err); got != tc.want {
t.Fatalf("isProxyFailure(%v)=%v, want %v", tc.err, got, tc.want)
}
})
}
}
func TestGlobalProxyTrafficFailureClassification(t *testing.T) {
if !isGlobalProxyDialFailure(errors.New("dial tcp 127.0.0.1:8080: connect: connection refused")) {
t.Fatal("upstream dial refusal should be eligible for probe confirmation")
}
if isGlobalProxyDialFailure(errors.New("status 404: target not found")) {
t.Fatal("ordinary target failure must not be treated as proxy failure")
}
}
func TestProxyPauseCauseIsDistinct(t *testing.T) {
ctx, cancel := context.WithCancelCause(context.Background())
cancel(agent.AbortPausedByProxy)
code, short, _, ok := agent.AbortReason(ctx)
if !ok || code != "paused_by_proxy" || short == "" {
t.Fatalf("proxy cause = %q/%q/%v", code, short, ok)
}
}
func TestProxyReconnectReasonGate(t *testing.T) {
for _, tc := range []struct {
reason string
want bool
}{
{"paused_by_proxy", true},
{"paused_by_user", false},
{"paused_by_schedule", false},
{"paused_by_orchestrator", false},
} {
got := tc.reason == "paused_by_proxy"
if got != tc.want {
t.Errorf("reason %q gate=%v, want %v", tc.reason, got, tc.want)
}
}
}
func TestProxyAdmissionRechecksReasonAfterSnapshot(t *testing.T) {
m := proxyTestManager(t)
task, err := m.CreateTask("proxy gate", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
if err := m.ApplyProxyPause(task.ID); err != nil {
t.Fatal(err)
}
s := newAdmissionTestServer(m, nil)
s.engine.Pause(task.ID, agent.AbortPausedByProxy)
// Simulate a user re-pause winning after proxyReconnectSuccess took its
// snapshot but before the admission lock was acquired.
if _, err := m.pg.Exec(`UPDATE tasks SET pause_reason='paused_by_user' WHERE id=$1`, task.ID); err != nil {
t.Fatal(err)
}
task.updateLifecycle(func(v *taskLifecycleState) { v.PauseReason = "paused_by_user" })
if _, err := s.admitTaskWhenScheduledWithReason(task, "resume", true, false, "paused_by_proxy", true); err == nil {
t.Fatal("proxy admission passed after user changed pause reason")
}
if !s.engine.IsPaused(task.ID) {
t.Fatal("user re-pause barrier was cleared")
}
}
func TestProxyQueuedAdmissionDoesNotResumeEngine(t *testing.T) {
m := proxyTestManager(t)
if err := m.SetConcurrency(true, 1); err != nil {
t.Fatal(err)
}
holder, err := m.CreateTask("holder", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
proxyTask, err := m.CreateTask("queued proxy", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
defer func() {
_, _ = m.DeleteTask(holder.ID, DeleteTaskOptions{})
_, _ = m.DeleteTask(proxyTask.ID, DeleteTaskOptions{})
}()
if err := m.ApplyProxyPause(proxyTask.ID); err != nil {
t.Fatal(err)
}
s := newAdmissionTestServer(m, nil)
s.engine.started.Store(holder.ID, true)
s.engine.Pause(proxyTask.ID, agent.AbortPausedByProxy)
queued, err := s.admitTaskWhenScheduledWithReason(proxyTask, "resume", true, false, "paused_by_proxy", true)
if err != nil {
t.Fatal(err)
}
if !queued {
t.Fatal("proxy task was expected to queue behind the running holder")
}
if !s.engine.IsPaused(proxyTask.ID) {
t.Fatal("queued proxy task was resumed")
}
}
func TestTerminalStatusClearsProxyReconnectState(t *testing.T) {
m := proxyTestManager(t)
task, err := m.CreateTask("terminal proxy", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
if err := m.ApplyProxyPause(task.ID); err != nil {
t.Fatal(err)
}
task.updateLifecycle(func(v *taskLifecycleState) {
v.ReconnectAttempt, v.ReconnectMax, v.ReconnectNextAt = 2, 5, 123
v.ReconnectExhausted, v.ReconnectLastError = true, "proxy down"
})
if err := m.SetTaskStatus(task.ID, "timeout"); err != nil {
t.Fatal(err)
}
stored, err := m.pg.GetTask(mustTaskID(t, task.ID))
if err != nil {
t.Fatal(err)
}
if stored.PauseReason != "" {
t.Fatalf("terminal row retained pause reason %q", stored.PauseReason)
}
st := task.lifecycleSnapshot()
if st.PauseReason != "" || st.ReconnectAttempt != 0 || st.ReconnectNextAt != 0 || st.ReconnectExhausted {
t.Fatalf("terminal lifecycle retained reconnect state: %+v", st)
}
}
func TestProxyAdmissionPreservesCalendarOverride(t *testing.T) {
m := proxyTestManager(t)
task, err := m.CreateTask("calendar proxy", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
if _, err := m.pg.Exec(`UPDATE tasks SET schedule_paused=true, schedule_manual_until='2099-01-01 00:00:00+00' WHERE id=$1`, task.ID); err != nil {
t.Fatal(err)
}
if err := m.ApplyProxyPause(task.ID); err != nil {
t.Fatal(err)
}
s := newAdmissionTestServer(m, nil)
s.engine.Pause(task.ID, agent.AbortPausedByProxy)
if queued, err := s.admitTaskWhenScheduledWithReason(task, "resume", true, false, "paused_by_proxy", true); err != nil || queued {
t.Fatalf("proxy admission queued=%v err=%v", queued, err)
}
var schedulePaused bool
var manualUntil time.Time
if err := m.pg.QueryRow(`SELECT schedule_paused, schedule_manual_until FROM tasks WHERE id=$1`, task.ID).Scan(&schedulePaused, &manualUntil); err != nil {
t.Fatal(err)
}
if !schedulePaused || !manualUntil.Equal(time.Date(2099, 1, 1, 0, 0, 0, 0, time.UTC)) {
t.Fatalf("proxy admission changed calendar state: paused=%v until=%v", schedulePaused, manualUntil)
}
}
func TestProxyReconnectSuccessKeepsCompetingUserPauseBarrier(t *testing.T) {
m := proxyTestManager(t)
task, err := m.CreateTask("proxy success race", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
s := newAdmissionTestServer(m, nil)
m.SetTaskControlMutex(&s.concMu)
m.globalProxy = "http://proxy.invalid"
if _, err := m.ApplyProxyPauseWithCallback(task.ID, func() { s.engine.Pause(task.ID, agent.AbortPausedByProxy) }); err != nil {
t.Fatal(err)
}
var pauseErr error
s.proxyReconnectSuccessWithHook(func() {
_, pauseErr = s.applyTaskControl(task, "pause")
})
if pauseErr != nil {
t.Fatalf("competing user pause failed: %v", pauseErr)
}
state := task.lifecycleSnapshot()
if state.PauseReason != "paused_by_user" || !state.Paused {
t.Fatalf("user pause was overwritten: %+v", state)
}
if !s.engine.IsPaused(task.ID) {
t.Fatal("user pause barrier was cleared by a post-admission Resume")
}
}
func TestProxyPauseBarrierInstalledBeforeControlUnlock(t *testing.T) {
m := proxyTestManager(t)
task, err := m.CreateTask("proxy barrier", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
s := newAdmissionTestServer(m, nil)
m.SetTaskControlMutex(&s.concMu)
m.globalProxy = "http://proxy.invalid"
userDone := make(chan error, 1)
transitioned, err := m.ApplyProxyPauseWithCallback(task.ID, func() {
s.engine.Pause(task.ID, agent.AbortPausedByProxy)
if !s.engine.IsPaused(task.ID) {
t.Fatal("proxy barrier was not installed in the locked callback")
}
go func() {
_, pauseErr := s.applyTaskControl(task, "pause")
userDone <- pauseErr
}()
select {
case <-userDone:
t.Fatal("user pause acquired concMu before proxy callback returned")
case <-time.After(20 * time.Millisecond):
}
})
if err != nil || !transitioned {
t.Fatalf("proxy transition=%v err=%v", transitioned, err)
}
if err := <-userDone; err != nil {
t.Fatalf("concurrent user pause failed: %v", err)
}
if !s.engine.IsPaused(task.ID) {
t.Fatal("concurrent user pause left the engine running")
}
if got := task.lifecycleSnapshot().PauseReason; got != "paused_by_user" {
t.Fatalf("concurrent user pause reason=%q", got)
}
}
func TestProxyPauseDoesNotReviveExhaustedOrNonProxyTasks(t *testing.T) {
m := proxyTestManager(t)
exhausted, err := m.CreateTask("exhausted proxy", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
userPaused, err := m.CreateTask("user paused", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
other, err := m.CreateTask("new proxy failure", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
defer func() {
_, _ = m.DeleteTask(exhausted.ID, DeleteTaskOptions{})
_, _ = m.DeleteTask(userPaused.ID, DeleteTaskOptions{})
_, _ = m.DeleteTask(other.ID, DeleteTaskOptions{})
}()
s := newAdmissionTestServer(m, nil)
m.SetTaskControlMutex(&s.concMu)
m.globalProxy = "http://proxy.invalid"
if err := m.ApplyProxyPause(exhausted.ID); err != nil {
t.Fatal(err)
}
if err := m.ApplyTaskPauseOriginReason(userPaused.ID, "paused_by_user", false); err != nil {
t.Fatal(err)
}
exhausted.updateLifecycle(func(v *taskLifecycleState) {
v.ReconnectAttempt, v.ReconnectMax, v.ReconnectNextAt = 5, 5, 12345
v.ReconnectExhausted, v.ReconnectLastError = true, "proxy down"
})
userPaused.updateLifecycle(func(v *taskLifecycleState) {
v.ReconnectAttempt, v.ReconnectMax, v.ReconnectNextAt = 3, 7, 23456
v.ReconnectExhausted, v.ReconnectLastError = true, "manual pause state"
})
exhaustedBefore := exhausted.lifecycleSnapshot()
userBefore := userPaused.lifecycleSnapshot()
// A no-op proxy pause on an already paused task must not reset either state.
s.pauseProxyTask(exhausted.ID, errors.New("another proxy failure"))
s.pauseProxyTask(userPaused.ID, errors.New("proxy failure while user paused"))
if got := exhausted.lifecycleSnapshot(); got.ReconnectNextAt != exhaustedBefore.ReconnectNextAt || !got.ReconnectExhausted || got.ReconnectLastError != exhaustedBefore.ReconnectLastError {
t.Fatalf("exhausted task was revived: before=%+v after=%+v", exhaustedBefore, got)
}
if got := userPaused.lifecycleSnapshot(); got.PauseReason != "paused_by_user" || got.ReconnectNextAt != userBefore.ReconnectNextAt || !got.ReconnectExhausted {
t.Fatalf("user-paused task was mutated: before=%+v after=%+v", userBefore, got)
}
// A genuinely new proxy pause of another running task initializes only that
// task's reconnect round.
s.pauseProxyTask(other.ID, errors.New("new proxy failure"))
if got := exhausted.lifecycleSnapshot(); got.ReconnectNextAt != exhaustedBefore.ReconnectNextAt || !got.ReconnectExhausted {
t.Fatalf("subsequent proxy pause revived exhausted task: %+v", got)
}
if got := other.lifecycleSnapshot(); got.PauseReason != "paused_by_proxy" || got.ReconnectMax == 0 || got.ReconnectExhausted {
t.Fatalf("new proxy task did not initialize reconnect state: %+v", got)
}
}
func TestProxyReconnectRoundResetsAfterManualResume(t *testing.T) {
m := proxyTestManager(t)
task, err := m.CreateTask("proxy retry round", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
s := newAdmissionTestServer(m, nil)
m.SetTaskControlMutex(&s.concMu)
m.globalProxy = "http://proxy.invalid"
if err := m.ApplyProxyPause(task.ID); err != nil {
t.Fatal(err)
}
s.engine.Pause(task.ID, agent.AbortPausedByProxy)
task.updateLifecycle(func(v *taskLifecycleState) {
v.ReconnectAttempt, v.ReconnectMax, v.ReconnectNextAt = 5, 5, 12345
v.ReconnectExhausted, v.ReconnectLastError = true, "proxy exhausted"
})
if _, err := s.applyTaskControl(task, "resume"); err != nil {
t.Fatal(err)
}
resumed := task.lifecycleSnapshot()
if resumed.PauseReason != "" || resumed.ReconnectAttempt != 0 || resumed.ReconnectMax != 0 || resumed.ReconnectNextAt != 0 || resumed.ReconnectExhausted || resumed.ReconnectLastError != "" {
t.Fatalf("manual resume retained reconnect state: %+v", resumed)
}
s.pauseProxyTask(task.ID, errors.New("proxy failed again"))
paused := task.lifecycleSnapshot()
if paused.PauseReason != "paused_by_proxy" || paused.ReconnectAttempt != 0 || paused.ReconnectExhausted {
t.Fatalf("new proxy round was not reset: %+v", paused)
}
if paused.ReconnectMax == 0 || paused.ReconnectNextAt == 0 || paused.ReconnectLastError == "" {
t.Fatalf("new proxy round did not initialize reconnect state: %+v", paused)
}
}
func TestProxyReconnectDropsStaleProbeFailureAfterNewRound(t *testing.T) {
m := proxyTestManager(t)
task, err := m.CreateTask("proxy stale probe", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
s := newAdmissionTestServer(m, nil)
m.SetTaskControlMutex(&s.concMu)
m.globalProxy = "http://proxy.invalid"
// Keep the canceled test server from starting its supervisor. The test owns
// the reconnect tick and injects a blocking probe below.
s.proxyReconnect.mu.Lock()
s.proxyReconnect.wake = make(chan struct{}, 1)
s.proxyReconnect.mu.Unlock()
oldMax, hadMax, err := m.pg.GetSetting(settingProxyReconnectMax)
if err != nil {
t.Fatal(err)
}
defer func() {
if hadMax {
_ = m.pg.SetSetting(settingProxyReconnectMax, oldMax)
} else {
_, _ = m.pg.Exec(`DELETE FROM settings WHERE key=$1`, settingProxyReconnectMax)
}
}()
if err := m.pg.SetSetting(settingProxyReconnectMax, "1"); err != nil {
t.Fatal(err)
}
s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
return egressProbeResult{Error: "initial probe failure"}
}
s.pauseProxyTask(task.ID, errors.New("initial proxy failure"))
task.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectNextAt = 0 })
oldRound := task.lifecycleSnapshot().ReconnectGeneration
probeStarted := make(chan struct{})
releaseProbe := make(chan struct{})
var probeStartedOnce sync.Once
s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
probeStartedOnce.Do(func() { close(probeStarted) })
<-releaseProbe
return egressProbeResult{Error: "stale probe failure"}
}
tickDone := make(chan struct{})
go func() {
s.proxyReconnectTick()
close(tickDone)
}()
select {
case <-probeStarted:
case <-time.After(time.Second):
t.Fatal("reconnect tick did not start its probe")
}
if _, err := s.applyTaskControl(task, "resume"); err != nil {
t.Fatal(err)
}
s.pauseProxyTask(task.ID, errors.New("new proxy failure"))
newRound := task.lifecycleSnapshot()
if newRound.ReconnectGeneration == oldRound || newRound.ReconnectAttempt != 0 || newRound.ReconnectExhausted {
t.Fatalf("new proxy round was not established: old=%d new=%+v", oldRound, newRound)
}
close(releaseProbe)
select {
case <-tickDone:
case <-time.After(time.Second):
t.Fatal("reconnect tick did not finish")
}
after := task.lifecycleSnapshot()
if after.PauseReason != "paused_by_proxy" || after.ReconnectGeneration != newRound.ReconnectGeneration || after.ReconnectAttempt != 0 || after.ReconnectExhausted || after.ReconnectNextAt != newRound.ReconnectNextAt || after.ReconnectLastError != newRound.ReconnectLastError {
t.Fatalf("stale probe mutated the new round: before=%+v after=%+v", newRound, after)
}
}
func TestProxyReconnectManualSuccessDoesNotAdmitNewRound(t *testing.T) {
m := proxyTestManager(t)
task, err := m.CreateTask("proxy stale manual success", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
s := newAdmissionTestServer(m, nil)
m.SetTaskControlMutex(&s.concMu)
m.globalProxy = "http://proxy.invalid"
if err := m.ApplyProxyPause(task.ID); err != nil {
t.Fatal(err)
}
s.engine.Pause(task.ID, agent.AbortPausedByProxy)
guards := s.snapshotProxyReconnectGuards(false)
task.updateLifecycle(func(v *taskLifecycleState) {
v.ReconnectGeneration++
v.ReconnectNextAt = 987654
v.ReconnectLastError = "new round still failing"
})
before := task.lifecycleSnapshot()
s.proxyReconnectSuccessWithGuards(guards, nil)
after := task.lifecycleSnapshot()
if after.PauseReason != "paused_by_proxy" || !s.engine.IsPaused(task.ID) {
t.Fatalf("stale manual success admitted new round: %+v", after)
}
if after.ReconnectGeneration != before.ReconnectGeneration || after.ReconnectNextAt != before.ReconnectNextAt || after.ReconnectLastError != before.ReconnectLastError {
t.Fatalf("stale manual success wiped new round state: before=%+v after=%+v", before, after)
}
}
func TestProxyReconnectSettingsSuccessGuardsGenerationZero(t *testing.T) {
m := proxyTestManager(t)
task, err := m.CreateTask("proxy generation zero", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
s := newAdmissionTestServer(m, nil)
m.SetTaskControlMutex(&s.concMu)
m.globalProxy = "http://proxy.invalid"
if err := m.ApplyProxyPause(task.ID); err != nil {
t.Fatal(err)
}
s.engine.Pause(task.ID, agent.AbortPausedByProxy)
initial := task.lifecycleSnapshot()
if initial.ReconnectGeneration != 0 {
t.Fatalf("fresh persisted proxy pause generation=%d, want zero", initial.ReconnectGeneration)
}
guards := s.snapshotProxyReconnectGuards(false)
task.updateLifecycle(func(v *taskLifecycleState) {
v.ReconnectGeneration = 1
v.ReconnectNextAt = 13579
v.ReconnectLastError = "generation one failure"
})
before := task.lifecycleSnapshot()
s.proxyReconnectSuccessWithGuards(guards, nil)
after := task.lifecycleSnapshot()
if after.PauseReason != "paused_by_proxy" || !s.engine.IsPaused(task.ID) {
t.Fatalf("generation-zero stale settings success admitted new round: %+v", after)
}
if after.ReconnectGeneration != before.ReconnectGeneration || after.ReconnectNextAt != before.ReconnectNextAt || after.ReconnectLastError != before.ReconnectLastError {
t.Fatalf("generation-zero stale settings success wiped state: before=%+v after=%+v", before, after)
}
}
+1160
View File
File diff suppressed because it is too large. Load diff
+791
View File
@@ -0,0 +1,791 @@
package server
import (
"bufio"
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
"net"
"net/http"
"net/http/httptest"
"net/url"
"strings"
"testing"
"time"
"github.com/Autumn-27/artex/agent"
"github.com/Autumn-27/artex/traffic"
)
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"}, RawURL: "http://a"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: false, Health: "healthy"}, RawURL: "http://b"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "c", Enabled: true, Health: "unhealthy-cooling"}, RawURL: "http://c"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "d", Enabled: true, Health: "healthy"}, RawURL: "http://d"},
}}}
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)
}
}
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 TestProxyPoolHealthThresholdCooldownAndRecovery(t *testing.T) {
m := &Manager{proxyPool: &proxyPoolState{entries: []proxyPoolEntry{{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"}}}}
fail := egressProbeResult{Error: "http://user:pass@proxy: connection refused"}
for i := 0; i < defaultProxyHealthThreshold-1; i++ {
m.markProxyProbe("a", fail)
if got := m.proxyPool.entries[0].Health; got != "healthy" {
t.Fatalf("failure %d moved to %s before threshold", i+1, got)
}
}
m.markProxyProbe("a", fail)
if got := m.proxyPool.entries[0]; got.Health != "unhealthy-cooling" || got.ConsecutiveFailures != defaultProxyHealthThreshold {
t.Fatalf("threshold state=%+v", got)
}
if got := m.proxyPool.entries[0].LastError; strings.Contains(got, "pass") || !strings.Contains(got, "***@") {
t.Fatalf("failure was not masked: %q", got)
}
m.markProxyProbe("a", egressProbeResult{OK: true, ExitIP: "203.0.113.9"})
if got := m.proxyPool.entries[0]; got.Health != "healthy" || got.ConsecutiveFailures != 0 || got.EgressIP != "203.0.113.9" {
t.Fatalf("recovery state=%+v", got)
}
}
func TestProxyPoolUncheckedMemberDoesNotPauseBeforeFailureThreshold(t *testing.T) {
m := proxyTestManager(t)
task, err := m.CreateTask("unchecked proxy", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
m.globalProxy = "http://a"
m.proxyPool = &proxyPoolState{active: "a", entries: []proxyPoolEntry{{
ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "unknown"}, RawURL: "http://a",
}}}
s := &Server{m: m, ctx: context.Background(), proxyPoolProbeSem: make(chan struct{}, proxyPoolMaxProbeConcurrency)}
s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
return egressProbeResult{Error: "proxy down"}
}
s.proxyPoolTickLocked(true)
if st := task.lifecycleSnapshot(); st.Paused || st.PauseReason == "paused_by_proxy" {
t.Fatalf("unchecked member paused the task: %+v", st)
}
if got := m.proxyPool.entries[0].Health; got != "unknown" {
t.Fatalf("single failure changed unchecked member to %q", got)
}
}
func TestProxyPoolCoolingWindowSkipsProbeUntilExpiry(t *testing.T) {
m := proxyTestManager(t)
m.globalProxy = "http://a"
m.proxyPool = &proxyPoolState{active: "a", entries: []proxyPoolEntry{{
ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "unhealthy-cooling", ConsecutiveFailures: 3, LastCheckedAt: time.Now()}, RawURL: "http://a",
}}}
s := &Server{m: m, ctx: context.Background(), proxyPoolProbeSem: make(chan struct{}, proxyPoolMaxProbeConcurrency)}
calls := 0
s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
calls++
return egressProbeResult{OK: true, ExitIP: "203.0.113.5"}
}
s.proxyPoolTickLocked(false)
if calls != 0 {
t.Fatalf("cooling member was probed before cooldown expiry: calls=%d", calls)
}
m.proxyPool.entries[0].LastCheckedAt = time.Now().Add(-2 * time.Minute)
s.proxyPoolTickLocked(false)
if calls != 1 || m.proxyPool.entries[0].Health != "healthy" {
t.Fatalf("expired cooling member did not recover: calls=%d entry=%+v", calls, m.proxyPool.entries[0])
}
}
func TestProxyPoolTimedRotationUsesAvailableMembersAndRecoversTasks(t *testing.T) {
m := proxyTestManager(t)
if err := m.SetProxyPoolSettings(true, minProxyRotateInterval, true, "sequential", 3, 60, 30); err != nil {
t.Fatal(err)
}
m.globalProxy = "http://a"
m.proxyPool = &proxyPoolState{active: "a", lastTimed: time.Now().Add(-time.Minute), lastHealth: time.Now(), entries: []proxyPoolEntry{
{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: true, Health: "healthy"}, RawURL: "http://b"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "c", Enabled: true, Health: "unhealthy-cooling"}, RawURL: "http://c"},
}}
s := &Server{m: m, ctx: context.Background()}
s.proxyPoolTickLocked(false)
if got := m.GlobalProxy(); got != "http://b" {
t.Fatalf("timed rotation selected %q", got)
}
if len(m.proxyPool.logs) == 0 || m.proxyPool.logs[len(m.proxyPool.logs)-1].Reason != "timed" {
t.Fatalf("timed rotation was not logged: %+v", m.proxyPool.logs)
}
}
func TestProxyPoolFailureSwitchRecoversOnlyEligibleTasks(t *testing.T) {
m := proxyTestManager(t)
if err := m.SetProxyPoolSettings(false, minProxyRotateInterval, true, "sequential", 3, 60, 30); err != nil {
t.Fatal(err)
}
m.globalProxy = "http://a"
m.proxyPool = &proxyPoolState{active: "a", entries: []proxyPoolEntry{
{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: true, Health: "healthy"}, RawURL: "http://b"},
}}
eligible, err := m.CreateTask("eligible", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
exhausted, err := m.CreateTask("exhausted", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
defer func() {
_, _ = m.DeleteTask(eligible.ID, DeleteTaskOptions{})
_, _ = m.DeleteTask(exhausted.ID, DeleteTaskOptions{})
}()
for _, task := range []*Task{eligible, exhausted} {
if err := m.ApplyProxyPause(task.ID); err != nil {
t.Fatal(err)
}
task.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectGeneration = 7; v.ReconnectMax = 3 })
}
exhausted.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectAttempt, v.ReconnectExhausted = 3, true })
s := newAdmissionTestServer(m, nil)
m.SetTaskControlMutex(&s.concMu)
s.proxyPoolWake = make(chan struct{}, 1)
s.proxyPoolProbeSem = make(chan struct{}, proxyPoolMaxProbeConcurrency)
s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
return egressProbeResult{Error: "proxy down"}
}
s.engine.Pause(eligible.ID, agent.AbortPausedByProxy)
s.engine.Pause(exhausted.ID, agent.AbortPausedByProxy)
if !s.handleProxyPoolFailure(errors.New("proxyconnect tcp: connection refused")) {
t.Fatal("confirmed failure did not switch to the available member")
}
if got := m.GlobalProxy(); got != "http://b" {
t.Fatalf("failure switch selected %q", got)
}
if st := eligible.lifecycleSnapshot(); st.PauseReason != "" || st.Paused {
t.Fatalf("eligible task did not recover: %+v", st)
}
if st := exhausted.lifecycleSnapshot(); st.PauseReason != "paused_by_proxy" || !st.ReconnectExhausted {
t.Fatalf("exhausted task was automatically resumed: %+v", st)
}
}
func TestProxyPoolReconnectKeepsAttemptsAndDoesNotAutoResumeExhausted(t *testing.T) {
m := proxyTestManager(t)
if err := m.pg.SetSetting(settingProxyReconnectMax, "2"); err != nil {
t.Fatal(err)
}
if err := m.pg.SetSetting(settingProxyReconnectInterval, "1"); err != nil {
t.Fatal(err)
}
m.globalProxy = "http://legacy"
m.proxyPool = &proxyPoolState{active: "legacy", entries: []proxyPoolEntry{{
ProxyPoolEntry: ProxyPoolEntry{ID: "legacy", Enabled: true, Health: "healthy"}, RawURL: "http://legacy",
}}}
task, err := m.CreateTask("legacy reconnect", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }()
if err := m.ApplyProxyPause(task.ID); err != nil {
t.Fatal(err)
}
task.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectGeneration = 3 })
s := &Server{m: m, ctx: context.Background(), proxyPoolProbeSem: make(chan struct{}, proxyPoolMaxProbeConcurrency)}
s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
return egressProbeResult{Error: "proxy down"}
}
s.proxyReconnectTick()
first := task.lifecycleSnapshot()
if first.ReconnectAttempt != 1 || first.ReconnectExhausted {
t.Fatalf("first pool reconnect attempt=%+v", first)
}
task.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectNextAt = 0 })
s.proxyReconnectTick()
second := task.lifecycleSnapshot()
if second.ReconnectAttempt != 2 || !second.ReconnectExhausted || second.PauseReason != "paused_by_proxy" {
t.Fatalf("second pool reconnect did not exhaust: %+v", second)
}
s.proxyReconnectTick()
third := task.lifecycleSnapshot()
if third.ReconnectAttempt != second.ReconnectAttempt || !third.ReconnectExhausted || third.PauseReason != "paused_by_proxy" {
t.Fatalf("automatic tick revived exhausted task: %+v", third)
}
guards := s.snapshotProxyReconnectGuardsWithExhausted(false, true)
s.proxyReconnectSuccessWithGuards(guards, nil)
if got := task.lifecycleSnapshot(); got.PauseReason != "" {
t.Fatalf("manual guarded resume did not release exhausted task: %+v", got)
}
}
func TestProxyPoolDisableOrClearActiveRoutesAway(t *testing.T) {
newServer := func(entries []proxyPoolEntry) (*Server, *Manager) {
m := &Manager{globalProxy: "http://a", proxyPool: &proxyPoolState{active: "a", entries: entries}}
return &Server{m: m}, m
}
t.Run("disable switches to healthy member", func(t *testing.T) {
s, m := newServer([]proxyPoolEntry{
{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: true, Health: "healthy"}, RawURL: "http://b"},
})
r := httptest.NewRequest(http.MethodPatch, "/api/settings/proxy-pool/a", strings.NewReader(`{"enabled":false}`))
r.SetPathValue("id", "a")
w := httptest.NewRecorder()
s.updateProxyPool(w, r)
if w.Code != http.StatusOK || m.GlobalProxy() != "http://b" || m.proxyPoolActiveID() != "b" {
t.Fatalf("disable route status=%d proxy=%q active=%q body=%s", w.Code, m.GlobalProxy(), m.proxyPoolActiveID(), w.Body.String())
}
})
t.Run("clear last member clears runtime", func(t *testing.T) {
s, m := newServer([]proxyPoolEntry{{
ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a",
}})
r := httptest.NewRequest(http.MethodPatch, "/api/settings/proxy-pool/a", strings.NewReader(`{"url":""}`))
r.SetPathValue("id", "a")
w := httptest.NewRecorder()
s.updateProxyPool(w, r)
if w.Code != http.StatusOK || m.GlobalProxy() != "" || m.proxyPoolActiveID() != "" {
t.Fatalf("clear route status=%d proxy=%q active=%q body=%s", w.Code, m.GlobalProxy(), m.proxyPoolActiveID(), w.Body.String())
}
})
}
func TestProxyPoolClearOrDeleteLastMemberRecoversOnlyEligibleProxyPauses(t *testing.T) {
run := func(t *testing.T, deleteLast bool) {
m := proxyTestManager(t)
m.globalProxy = "http://a"
m.proxyPool = &proxyPoolState{active: "a", entries: []proxyPoolEntry{{
ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a",
}}}
eligible, err := m.CreateTask("eligible", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
exhausted, err := m.CreateTask("exhausted", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
userPaused, err := m.CreateTask("user paused", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
defer func() {
_, _ = m.DeleteTask(eligible.ID, DeleteTaskOptions{})
_, _ = m.DeleteTask(exhausted.ID, DeleteTaskOptions{})
_, _ = m.DeleteTask(userPaused.ID, DeleteTaskOptions{})
}()
for _, task := range []*Task{eligible, exhausted} {
if err := m.ApplyProxyPause(task.ID); err != nil {
t.Fatal(err)
}
task.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectGeneration = 7 })
}
exhausted.updateLifecycle(func(v *taskLifecycleState) {
v.ReconnectAttempt, v.ReconnectExhausted = 3, true
})
if err := m.ApplyTaskPauseOrigin(userPaused.ID, false); err != nil {
t.Fatal(err)
}
s := newAdmissionTestServer(m, nil)
for _, task := range []*Task{eligible, exhausted, userPaused} {
s.engine.Pause(task.ID, agent.AbortPausedByProxy)
}
if deleteLast {
r := httptest.NewRequest(http.MethodDelete, "/api/settings/proxy-pool/a", nil)
r.SetPathValue("id", "a")
w := httptest.NewRecorder()
s.deleteProxyPool(w, r)
if w.Code != http.StatusOK {
t.Fatalf("delete status=%d body=%s", w.Code, w.Body.String())
}
} else {
r := httptest.NewRequest(http.MethodPatch, "/api/settings/proxy-pool/a", strings.NewReader(`{"url":""}`))
r.SetPathValue("id", "a")
w := httptest.NewRecorder()
s.updateProxyPool(w, r)
if w.Code != http.StatusOK {
t.Fatalf("clear status=%d body=%s", w.Code, w.Body.String())
}
}
if got := eligible.lifecycleSnapshot(); got.PauseReason != "" || got.Paused {
t.Fatalf("eligible proxy pause was not recovered: %+v", got)
}
if got := exhausted.lifecycleSnapshot(); got.PauseReason != "paused_by_proxy" || !got.ReconnectExhausted {
t.Fatalf("exhausted proxy pause was recovered: %+v", got)
}
if got := userPaused.lifecycleSnapshot(); got.PauseReason != "paused_by_user" {
t.Fatalf("user pause was recovered: %+v", got)
}
}
t.Run("clear", func(t *testing.T) { run(t, false) })
t.Run("delete", func(t *testing.T) { run(t, true) })
}
func TestProxyPoolCaptureOffUsesLocalForwarder(t *testing.T) {
fp, err := newDynamicForwardProxy(func() string { return "" })
if err != nil {
t.Fatal(err)
}
defer fp.Close()
m := &Manager{forwardProxy: fp}
m.proxyPoolConfigured.Store(true)
got := m.ProxyAddr()
if !strings.HasPrefix(got, "http://127.0.0.1:") {
t.Fatalf("capture-off proxy address=%q, want local forwarder", got)
}
}
func TestProxyPoolClearedRuntimeForwarderFallsBackToDirect(t *testing.T) {
target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_, _ = io.WriteString(w, "direct-ok")
}))
defer target.Close()
m := proxyTestManager(t)
// A configured pool can remain persisted after its last usable member is
// disabled or cleared. The runtime route is still direct in that state.
m.proxyPoolConfigured.Store(true)
proxyURL, err := url.Parse(m.ProxyAddr())
if err != nil {
t.Fatal(err)
}
client := &http.Client{Transport: &http.Transport{Proxy: http.ProxyURL(proxyURL)}}
resp, err := client.Get(target.URL)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
if resp.StatusCode != http.StatusOK || string(body) != "direct-ok" {
t.Fatalf("cleared runtime route status=%d body=%q", resp.StatusCode, body)
}
}
func TestProxyReconnectGuardsGateExhaustedAutomaticResume(t *testing.T) {
task := &Task{ID: "exhausted", PauseReason: "paused_by_proxy", ReconnectGeneration: 4, ReconnectAttempt: 3, ReconnectExhausted: true}
m := &Manager{tasks: map[string]*Task{task.ID: task}}
s := &Server{m: m}
if got := s.snapshotProxyReconnectGuards(false); len(got) != 0 {
t.Fatalf("automatic guards included exhausted task: %+v", got)
}
if got := s.snapshotProxyReconnectGuardsWithExhausted(false, true); len(got) != 1 {
t.Fatalf("manual guards did not include exhausted task: %+v", got)
}
}
func TestProxyPoolSettingsRejectTooShortInterval(t *testing.T) {
m := &Manager{}
if err := m.SetProxyPoolSettings(true, minProxyRotateInterval-1, true, "sequential", 3, 60, 30); err == nil {
t.Fatal("interval below lower bound was accepted")
}
}
func TestProxyPoolRandomSelectionStillExcludesDisabledAndCooling(t *testing.T) {
m := proxyTestManager(t)
if err := m.pg.SetSetting(settingProxyRotateStrategy, "random"); err != nil {
t.Fatal(err)
}
m.proxyPool = &proxyPoolState{entries: []proxyPoolEntry{
{ProxyPoolEntry: ProxyPoolEntry{ID: "disabled", Enabled: false, Health: "healthy"}, RawURL: "http://disabled"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "cooling", Enabled: true, Health: "unhealthy-cooling"}, RawURL: "http://cooling"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "empty", Enabled: true, Health: "healthy"}},
{ProxyPoolEntry: ProxyPoolEntry{ID: "good", Enabled: true, Health: "healthy"}, RawURL: "http://good"},
}}
for i := 0; i < 20; i++ {
if got := m.selectHealthyLocked(""); got != "good" {
t.Fatalf("random selected %q", got)
}
}
}
func TestProxyPoolProbeSkipsEnabledEmptyURL(t *testing.T) {
m := proxyTestManager(t)
m.globalProxy = "http://good"
m.proxyPool = &proxyPoolState{active: "good", entries: []proxyPoolEntry{
{ProxyPoolEntry: ProxyPoolEntry{ID: "empty", Enabled: true, Health: "healthy"}},
{ProxyPoolEntry: ProxyPoolEntry{ID: "good", Enabled: true, Health: "healthy"}, RawURL: "http://good"},
}}
s := &Server{m: m, ctx: context.Background(), proxyPoolProbeSem: make(chan struct{}, proxyPoolMaxProbeConcurrency)}
calls := 0
s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
calls++
return egressProbeResult{OK: true}
}
s.proxyPoolTickLocked(true)
if calls != 1 {
t.Fatalf("probe calls=%d, want only non-empty enabled member", calls)
}
}
func TestProxyPoolFailureMarksOriginalMemberAfterConcurrentSwitch(t *testing.T) {
m := &Manager{globalProxy: "http://a", proxyPool: &proxyPoolState{active: "a", entries: []proxyPoolEntry{
{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: true, Health: "healthy"}, RawURL: "http://b"},
}}}
s := &Server{m: m, ctx: context.Background(), proxyPoolWake: make(chan struct{}, 1)}
s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
m.mu.Lock()
m.globalProxy = "http://b"
m.mu.Unlock()
return egressProbeResult{Error: "proxy connection refused"}
}
if !s.handleProxyPoolFailure(fmt.Errorf("proxyconnect tcp: connection refused")) {
t.Fatal("failure after concurrent switch was not consumed")
}
if got := m.proxyPool.entries[0].ConsecutiveFailures; got != 1 {
t.Fatalf("original member failure count=%d", got)
}
}
func TestProxyPoolAllMembersUnhealthyKeepsProxyConfigured(t *testing.T) {
m := &Manager{globalProxy: "http://a", proxyPool: &proxyPoolState{active: "a", entries: []proxyPoolEntry{
{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "unhealthy-cooling", ConsecutiveFailures: defaultProxyHealthThreshold, LastCheckedAt: time.Now().Add(-2 * time.Minute)}, RawURL: "http://a"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: true, Health: "unhealthy-cooling", ConsecutiveFailures: defaultProxyHealthThreshold, LastCheckedAt: time.Now().Add(-2 * time.Minute)}, RawURL: "http://b"},
}}}
s := &Server{m: m, ctx: context.Background(), proxyPoolProbeSem: make(chan struct{}, proxyPoolMaxProbeConcurrency)}
s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult {
return egressProbeResult{Error: "proxy down"}
}
s.proxyPoolTickLocked(true)
if got := m.GlobalProxy(); got != "http://a" {
t.Fatalf("all unhealthy members fell back to direct: %q", got)
}
if got := m.proxyPoolActiveID(); got != "a" {
t.Fatalf("active member changed without a healthy replacement: %q", got)
}
}
func TestProxyPoolMaskedEditHotCutsCoolingActiveMember(t *testing.T) {
m := &Manager{globalProxy: "http://old:[email protected]:8080", proxyPool: &proxyPoolState{active: "a", entries: []proxyPoolEntry{{
ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "unhealthy-cooling", ConsecutiveFailures: 3},
RawURL: "http://old:[email protected]:8080",
}}}}
s := &Server{m: m}
r := httptest.NewRequest(http.MethodPatch, "/api/settings/proxy-pool/a", strings.NewReader(`{"url":"http://***@new.example:9090"}`))
r.SetPathValue("id", "a")
w := httptest.NewRecorder()
s.updateProxyPool(w, r)
if w.Code != http.StatusOK {
t.Fatalf("status=%d body=%s", w.Code, w.Body.String())
}
if got := m.GlobalProxy(); got != "http://old:[email protected]:9090" {
t.Fatalf("active proxy=%q", got)
}
if got := m.proxyPool.entries[0].Health; got != "healthy" {
t.Fatalf("edited active member health=%q", got)
}
}
func TestProxyPoolDeleteNeverRetainsDeletedRuntimeURL(t *testing.T) {
m := &Manager{globalProxy: "http://a", proxyPool: &proxyPoolState{active: "a", entries: []proxyPoolEntry{
{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: true, Health: "healthy"}, RawURL: "http://b"},
}}}
s := &Server{m: m}
deleteEntry := func(id string) {
r := httptest.NewRequest(http.MethodDelete, "/api/settings/proxy-pool/"+id, nil)
r.SetPathValue("id", id)
w := httptest.NewRecorder()
s.deleteProxyPool(w, r)
if w.Code != http.StatusOK {
t.Fatalf("delete %s status=%d body=%s", id, w.Code, w.Body.String())
}
}
deleteEntry("a")
if got := m.GlobalProxy(); got != "http://b" {
t.Fatalf("surviving pool member was not activated: %q", got)
}
deleteEntry("b")
if got := m.GlobalProxy(); got != "" || m.proxyPool.active != "" {
t.Fatalf("deleted last member left runtime proxy=%q active=%q", got, m.proxyPool.active)
}
}
func TestSetGlobalProxyIsReadOnlyWhenPoolManaged(t *testing.T) {
m := &Manager{globalProxy: "http://pool.example:8080", proxyPool: &proxyPoolState{entries: []proxyPoolEntry{{
ProxyPoolEntry: ProxyPoolEntry{ID: "pool", Enabled: true, Health: "healthy"}, RawURL: "http://pool.example:8080",
}}}}
if err := m.SetGlobalProxy("http://other.example:8080"); err != nil {
t.Fatal(err)
}
if got := m.GlobalProxy(); got != "http://pool.example:8080" {
t.Fatalf("managed pool runtime changed through legacy setting: %q", got)
}
}
func TestProxyPoolEmptyPersistenceDoesNotResurrectLegacyProxy(t *testing.T) {
dir := t.TempDir()
m, err := NewManager(dir, "")
if err != nil {
t.Skipf("postgres unavailable (%v) - skipping", err)
}
m.proxyPool = &proxyPoolState{active: "a", entries: []proxyPoolEntry{{
ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://deleted.example:8080",
}}}
m.globalProxy = "http://deleted.example:8080"
if err := m.persistProxyPoolLocked(); err != nil {
_ = m.Close()
t.Fatal(err)
}
s := &Server{m: m}
r := httptest.NewRequest(http.MethodDelete, "/api/settings/proxy-pool/a", nil)
r.SetPathValue("id", "a")
w := httptest.NewRecorder()
s.deleteProxyPool(w, r)
if w.Code != http.StatusOK {
_ = m.Close()
t.Fatalf("delete status=%d body=%s", w.Code, w.Body.String())
}
if got := m.GlobalProxy(); got != "" {
t.Fatalf("delete left runtime proxy=%q", got)
}
if err := m.Close(); err != nil {
t.Fatal(err)
}
m2, err := NewManager(dir, "")
if err != nil {
t.Fatal(err)
}
defer m2.Close()
if got := m2.GlobalProxy(); got != "" || len(m2.proxyPoolEntries()) != 0 {
t.Fatalf("empty persisted pool resurrected proxy=%q entries=%d", got, len(m2.proxyPoolEntries()))
}
}
func TestMergeMaskedProxyURLPreservesOnlyPassword(t *testing.T) {
got, err := mergeMaskedProxyURL("http://***@NEW.example:9090", "http://old:[email protected]:8080")
if err != nil {
t.Fatal(err)
}
if got != "http://old:[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)
}
}
func TestDynamicForwardProxyConnectAuthenticatedUpstreamAndBufferedBytes(t *testing.T) {
const user, pass = "alice", "s3cret"
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodConnect {
http.Error(w, "connect required", http.StatusMethodNotAllowed)
return
}
want := "Basic " + base64.StdEncoding.EncodeToString([]byte(user+":"+pass))
if r.Header.Get("Proxy-Authorization") != want {
w.Header().Set("Proxy-Authenticate", `Basic realm="test"`)
w.WriteHeader(http.StatusProxyAuthRequired)
return
}
h, ok := w.(http.Hijacker)
if !ok {
http.Error(w, "no hijack", 500)
return
}
c, rw, err := h.Hijack()
if err != nil {
return
}
defer c.Close()
_, _ = rw.WriteString("HTTP/1.1 200 Connection Established\r\n\r\nupstream-buffer")
_ = rw.Flush()
}))
defer upstream.Close()
up, _ := url.Parse(upstream.URL)
up.User = url.UserPassword(user, pass)
fp, err := newDynamicForwardProxy(func() string { return up.String() })
if err != nil {
t.Fatal(err)
}
defer fp.Close()
c, err := net.Dial("tcp", strings.TrimPrefix(fp.Addr(), "http://"))
if err != nil {
t.Fatal(err)
}
defer c.Close()
_, _ = fmt.Fprintf(c, "CONNECT example.test:443 HTTP/1.1\r\nHost: example.test:443\r\n\r\n")
rd := bufio.NewReader(c)
resp, err := http.ReadResponse(rd, &http.Request{Method: http.MethodConnect})
if err != nil {
t.Fatal(err)
}
if resp.StatusCode != http.StatusOK {
t.Fatalf("local CONNECT status %s", resp.Status)
}
b, err := io.ReadAll(io.LimitReader(rd, int64(len("upstream-buffer"))))
if err != nil || string(b) != "upstream-buffer" {
t.Fatalf("buffered bytes=%q err=%v", b, err)
}
}
func TestDynamicForwardProxyConnectForwardsHijackBufferedClientBytes(t *testing.T) {
targetLn, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
defer targetLn.Close()
go func() {
c, err := targetLn.Accept()
if err != nil {
return
}
defer c.Close()
buf := make([]byte, len("hello"))
if _, err := io.ReadFull(c, buf); err == nil {
_, _ = c.Write([]byte("target"))
}
}()
proxyServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodConnect {
http.Error(w, "connect required", 405)
return
}
c, err := net.Dial("tcp", targetLn.Addr().String())
if err != nil {
http.Error(w, err.Error(), 502)
return
}
h, _ := w.(http.Hijacker)
client, rw, err := h.Hijack()
if err != nil {
_ = c.Close()
return
}
defer client.Close()
defer c.Close()
_, _ = rw.WriteString("HTTP/1.1 200 Connection Established\r\n\r\n")
_ = rw.Flush()
buf := make([]byte, len("hello"))
if _, err := io.ReadFull(rw.Reader, buf); err != nil {
return
}
_, _ = c.Write(buf)
response := make([]byte, len("target"))
if _, err := io.ReadFull(c, response); err == nil {
_, _ = client.Write(response)
}
}))
defer proxyServer.Close()
fp, err := newDynamicForwardProxy(func() string { return proxyServer.URL })
if err != nil {
t.Fatal(err)
}
defer fp.Close()
c, err := net.Dial("tcp", strings.TrimPrefix(fp.Addr(), "http://"))
if err != nil {
t.Fatal(err)
}
defer c.Close()
_, _ = fmt.Fprintf(c, "CONNECT %s HTTP/1.1\r\nHost: %s\r\n\r\nhello", targetLn.Addr().String(), targetLn.Addr().String())
rd := bufio.NewReader(c)
resp, err := http.ReadResponse(rd, &http.Request{Method: http.MethodConnect})
if err != nil || resp.StatusCode != http.StatusOK {
t.Fatalf("CONNECT response=%v status=%v", err, resp)
}
_ = c.SetReadDeadline(time.Now().Add(2 * time.Second))
b, err := io.ReadAll(io.LimitReader(rd, int64(len("target"))))
if err != nil || string(b) != "target" {
t.Fatalf("target response=%q err=%v", b, err)
}
}
func TestDialHTTPConnectBlackholeHonorsContext(t *testing.T) {
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
defer ln.Close()
go func() {
c, err := ln.Accept()
if err == nil {
defer c.Close()
_, _ = io.Copy(io.Discard, c)
}
}()
u, _ := url.Parse("http://" + ln.Addr().String())
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()
started := time.Now()
_, err = dialHTTPConnect(ctx, u, "example.test:443")
if err == nil {
t.Fatal("blackholed CONNECT unexpectedly succeeded")
}
if elapsed := time.Since(started); elapsed > time.Second {
t.Fatalf("context cancellation took %s", elapsed)
}
}
+142 -26
View File
@@ -125,10 +125,18 @@ type Server struct {
// Cold task archives run through one persistent FIFO worker. The buffered wake // Cold task archives run through one persistent FIFO worker. The buffered wake
// channel coalesces enqueue bursts; the database remains the source of truth. // channel coalesces enqueue bursts; the database remains the source of truth.
archiveWake chan struct{} archiveWake chan struct{}
archiveWG sync.WaitGroup archiveWG sync.WaitGroup
credentialWake chan struct{} credentialWake chan struct{}
side *sideQuestionState proxyReconnect proxyReconnectSupervisor
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
side *sideQuestionState
} }
// provEntry is a cached provider + its config for one LLM profile id. // provEntry is a cached provider + its config for one LLM profile id.
@@ -159,10 +167,11 @@ 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{}, 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{}, 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{}, profChatAgents: map[int64]*agent.ChatAgent{},
provByProfile: map[int64]*provEntry{}, llmHealth: newLLMHealthRegistry(m.pg), provByProfile: map[int64]*provEntry{}, llmHealth: newLLMHealthRegistry(m.pg),
taskAgents: map[string]*taskAgentBundle{}, archiveWake: make(chan struct{}, 1), credentialWake: make(chan struct{}, 1)} taskAgents: map[string]*taskAgentBundle{}, archiveWake: make(chan struct{}, 1), credentialWake: make(chan struct{}, 1)}
m.SetTaskControlMutex(&s.concMu)
s.initSideQuestions() s.initSideQuestions()
if m.pg != nil { if m.pg != nil {
if err := s.basicAuth.load(m.pg); err != nil { if err := s.basicAuth.load(m.pg); err != nil {
@@ -192,6 +201,10 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
defer s.cfgMu.Unlock() defer s.cfgMu.Unlock()
return s.llmOn return s.llmOn
}) })
s.engine.SetProxyFailureHandler(s.handleProxyFailure)
if m.traffic != nil {
m.traffic.SetUpstreamFailureHandler(s.handleProxyTrafficFailure)
}
// Wire DB-stored prompt templates into the agents (新版方案 §3.3 / §5a). With no // Wire DB-stored prompt templates into the agents (新版方案 §3.3 / §5a). With no
// override row, agents keep their built-in defaults — behavior is unchanged. // override row, agents keep their built-in defaults — behavior is unchanged.
if m.pg != nil { if m.pg != nil {
@@ -280,6 +293,9 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
log.Printf("[engine] no LLM provider configured — engine idle until set via /api/llm or env") log.Printf("[engine] no LLM provider configured — engine idle until set via /api/llm or env")
} }
s.restoreTaskRuntimes() s.restoreTaskRuntimes()
s.ensureProxySupervisor()
s.proxyPoolWG.Add(1)
go func() { defer s.proxyPoolWG.Done(); s.runProxyPoolSupervisor(s.ctx) }()
go s.reconcileConcurrency() go s.reconcileConcurrency()
s.startTaskArchiveWorker() s.startTaskArchiveWorker()
if m.pg != nil { if m.pg != nil {
@@ -296,6 +312,13 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
return s 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, // Restored deadline and worker loops must inherit the same context as new tasks,
// including the side-question checkpoint publisher installed during startup. // including the side-question checkpoint publisher installed during startup.
func (s *Server) restoreTaskRuntimes() { func (s *Server) restoreTaskRuntimes() {
@@ -509,6 +532,7 @@ func (s *Server) applyLLM(cfg agent.Config) error {
// (backend/key) and gates Enabled per-conversation-agent at Chat time. 对话始终用激活配置。 // (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 = agent.NewChatAgent(prov, cfg.Model, s.m.dir, tx, win) // chat page runner
s.chatAgent.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) s.chatAgent.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
s.chatAgent.SetProxyProvider(s.m.ProxyAddr)
s.chatAgent.SetWebSearch(s.m.WebSearchOpts()) s.chatAgent.SetWebSearch(s.m.WebSearchOpts())
s.chatAgent.SetGuard(s.chatGuard()) s.chatAgent.SetGuard(s.chatGuard())
s.chatAgent.SetNonStreaming(nonStreamingResolver(cfg)) s.chatAgent.SetNonStreaming(nonStreamingResolver(cfg))
@@ -671,6 +695,7 @@ func (s *Server) chatAgentForProfile(id int64) *agent.ChatAgent {
tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts")) 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 := agent.NewChatAgent(s.poolForBinding(id, prov, cfg), cfg.Model, s.m.dir, tx, cfg.CompactionWindow())
ca.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) ca.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
ca.SetProxyProvider(s.m.ProxyAddr)
ca.SetWebSearch(s.m.WebSearchOpts()) ca.SetWebSearch(s.m.WebSearchOpts())
ca.SetGuard(s.chatGuard()) ca.SetGuard(s.chatGuard())
ca.SetNonStreaming(nonStreamingResolver(cfg)) ca.SetNonStreaming(nonStreamingResolver(cfg))
@@ -927,6 +952,14 @@ func (s *Server) Handler() http.Handler {
mux.HandleFunc("POST /api/settings/web-search/test", s.testWebSearch) mux.HandleFunc("POST /api/settings/web-search/test", s.testWebSearch)
// 出口代理连通性测试:发真实请求验证 global_proxy 是否可用,并回显出口 IP。 // 出口代理连通性测试:发真实请求验证 global_proxy 是否可用,并回显出口 IP。
mux.HandleFunc("POST /api/settings/proxy/test", s.testEgressProxy) 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/report", s.getReport)
mux.HandleFunc("GET /api/chat/mentions", s.searchChatMentions) mux.HandleFunc("GET /api/chat/mentions", s.searchChatMentions)
mux.HandleFunc("POST /api/chat", s.chat) mux.HandleFunc("POST /api/chat", s.chat)
@@ -2332,7 +2365,10 @@ func (s *Server) findingsExport(w http.ResponseWriter, r *http.Request) {
stamp := now.Format("20060102-150405") stamp := now.Format("20060102-150405")
setDownload := func(contentType, filename string) { setDownload := func(contentType, filename string) {
w.Header().Set("Content-Type", contentType) w.Header().Set("Content-Type", contentType)
w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"`, filename)) // Keep an ASCII fallback for older clients and provide the real UTF-8
// name through RFC 5987. Browsers otherwise commonly decode Chinese
// names using the platform legacy code page.
w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"; filename*=UTF-8''%s`, sanitizeFilename(filename), url.PathEscape(filename)))
} }
switch format { switch format {
@@ -3535,14 +3571,32 @@ func (s *Server) settingsPayload() map[string]any {
"web_search_backend": backend, "web_search_backend": backend,
"brave_key_set": strings.TrimSpace(braveKey) != "", "brave_key_set": strings.TrimSpace(braveKey) != "",
"tavily_key_set": strings.TrimSpace(tavilyKey) != "", "tavily_key_set": strings.TrimSpace(tavilyKey) != "",
"web_search_proxy": proxy, // 独立出口代理(http/https/socks5),空=直连 "web_search_proxy": maskSecrets(proxy), // 独立出口代理(http/https/socks5),空=直连
"global_proxy": s.m.GlobalProxy(), // 全局出口代理(http/https/socks5),所有目标流量走它,空=直连 "global_proxy": maskSecrets(s.m.GlobalProxy()), // 全局出口代理(http/https/socks5),所有目标流量走它,空=直连
"show_egress_ip": s.m.pg.GetBool(settingShowEgressIP, false), "proxy_pool": func() []ProxyPoolEntry { e, _, _ := s.m.ProxyPoolSnapshot(); return e }(),
"python_interpreter": strings.TrimSpace(pyStored), // 用户/自动设的值(空=用运行时检测) "proxy_pool_active_id": func() string { _, a, _ := s.m.ProxyPoolSnapshot(); return a }(),
"workers": s.m.Workers(), // 并发工作 agent 数(默认3);对之后启动的任务生效 "proxy_pool_switch_log": func() []ProxySwitchLog { _, _, l := s.m.ProxyPoolSnapshot(); return l }(),
"worker_pools": s.m.WorkerPools(), // 非空时按池顺序绑定 worker 会话 "proxy_rotate_timed_enabled": s.m.pg.GetBool(settingProxyRotateTimed, false),
"task_concurrency_enabled": concOn, // 任务并发上限开关(默认关) "proxy_rotate_interval_seconds": proxyIntSetting(s.m.pg, settingProxyRotateInterval, defaultProxyRotateInterval),
"task_concurrency_limit": concLimit, // 同时运行任务上限(开启后默认5) "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 在当前配置不可用时 // LLM 轮询(故障转移)。默认关;开启后走全局激活配置的 agent 在当前配置不可用时
// 自动切到下一个配置。bind_fallback 仅在轮询开启时有意义(默认关)。 // 自动切到下一个配置。bind_fallback 仅在轮询开启时有意义(默认关)。
"llm_pool_enabled": s.m.LLMPoolEnabled(), "llm_pool_enabled": s.m.LLMPoolEnabled(),
@@ -3613,16 +3667,25 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
LLMRecord *bool `json:"llm_record"` // LLM 录制开关(默认关);即时生效,无需重建 agent LLMRecord *bool `json:"llm_record"` // LLM 录制开关(默认关);即时生效,无需重建 agent
// Web search. WebSearchEnabled/Backend toggle the tool + backend; BraveKey/TavilyKey // Web search. WebSearchEnabled/Backend toggle the tool + backend; BraveKey/TavilyKey
// are optional — omit (null) to leave a stored key untouched, send "" to clear. // are optional — omit (null) to leave a stored key untouched, send "" to clear.
WebSearchEnabled *bool `json:"web_search_enabled"` WebSearchEnabled *bool `json:"web_search_enabled"`
WebSearchBackend *string `json:"web_search_backend"` WebSearchBackend *string `json:"web_search_backend"`
BraveKey *string `json:"brave_search_api_key"` BraveKey *string `json:"brave_search_api_key"`
TavilyKey *string `json:"tavily_search_api_key"` TavilyKey *string `json:"tavily_search_api_key"`
WebSearchProxy *string `json:"web_search_proxy"` // 独立出口代理(http/https/socks5);null=不改,""=清空 WebSearchProxy *string `json:"web_search_proxy"` // 独立出口代理(http/https/socks5);null=不改,""=清空
GlobalProxy *string `json:"global_proxy"` // 全局出口代理(http/https/socks5);null=不改,""=清空(直连) GlobalProxy *string `json:"global_proxy"` // 全局出口代理(http/https/socks5);null=不改,""=清空(直连)
ShowEgressIP *bool `json:"show_egress_ip"` // 在顶栏显示直连/全局代理出口 IP;默认关 ProxyRotateTimed *bool `json:"proxy_rotate_timed_enabled"`
PythonInterp *string `json:"python_interpreter"` // 自定义脚本工具的 python 解释器路径 ProxyRotateInterval *int `json:"proxy_rotate_interval_seconds"`
Workers *int `json:"workers"` // 并发工作 agent 数(>0);对之后启动的任务生效 ProxyRotateFailure *bool `json:"proxy_rotate_on_failure"`
WorkerPools *[]WorkerPool `json:"worker_pools"` 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;默认关
PythonInterp *string `json:"python_interpreter"` // 自定义脚本工具的 python 解释器路径
Workers *int `json:"workers"` // 并发工作 agent 数(>0);对之后启动的任务生效
WorkerPools *[]WorkerPool `json:"worker_pools"`
// 任务并发上限:同时「运行中」的任务数上限。关闭=不限;开启后新建任务超限则排队,有空位自动启动。 // 任务并发上限:同时「运行中」的任务数上限。关闭=不限;开启后新建任务超限则排队,有空位自动启动。
ConcurrencyEnabled *bool `json:"task_concurrency_enabled"` ConcurrencyEnabled *bool `json:"task_concurrency_enabled"`
ConcurrencyLimit *int `json:"task_concurrency_limit"` ConcurrencyLimit *int `json:"task_concurrency_limit"`
@@ -3833,12 +3896,61 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
changed = true changed = true
} }
if req.GlobalProxy != nil { 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 globalProxyValue == "" {
proxyGuards = s.snapshotProxyReconnectGuards(false)
}
// Validation failure (bad scheme/host) is a client error, not a 500. // 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()) writeErr(w, 400, err.Error())
return return
} }
changed = true // capture-off egress is baked into agents at build time → rebuild if globalProxyValue == "" && (s.m.proxyPool == nil || len(s.m.proxyPoolEntries()) == 0) {
s.proxyReconnectSuccessWithGuards(proxyGuards, nil)
}
}
if req.ProxyReconnectMax != nil || req.ProxyReconnectInterval != nil {
if err := s.setProxyReconnectSettings(req.ProxyReconnectMax, req.ProxyReconnectInterval); err != nil {
writeErr(w, 400, err.Error())
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 req.ShowEgressIP != nil {
if err := s.m.pg.SetBool(settingShowEgressIP, *req.ShowEgressIP); err != nil { if err := s.m.pg.SetBool(settingShowEgressIP, *req.ShowEgressIP); err != nil {
@@ -3855,6 +3967,10 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
if req.WebSearchBackend != nil { if req.WebSearchBackend != nil {
backend = *req.WebSearchBackend 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 { if err := s.m.SetWebSearch(on, backend, req.BraveKey, req.TavilyKey, req.WebSearchProxy); err != nil {
writeErr(w, 500, err.Error()) writeErr(w, 500, err.Error())
return return
+8 -1
View File
@@ -101,7 +101,14 @@ func (s *Server) beginTaskDelete(taskID string) bool {
if s.engine.IsDeleting(taskID) { if s.engine.IsDeleting(taskID) {
return false return false
} }
return s.engine.BeginDelete(taskID) ok := s.engine.BeginDelete(taskID)
if ok {
// A user delete/archive action is an explicit override of proxy recovery.
if s.m != nil && s.m.pg != nil {
_ = s.m.ClearPauseReason(taskID)
}
}
return ok
} }
// abortTaskDelete restores the execution barrier from the committed task row, // abortTaskDelete restores the execution barrier from the committed task row,
+39 -3
View File
@@ -116,9 +116,31 @@ func (s *Server) applyTaskControlWithCause(t *Task, action string, pauseCause er
} }
if lifecycle.Paused { if lifecycle.Paused {
if pauseCause != agent.AbortPausedBySchedule { if pauseCause != agent.AbortPausedBySchedule {
if _, err := s.m.pg.Exec(`UPDATE tasks SET schedule_paused=false WHERE id=$1`, t.ID); err != nil { if pauseCause == nil {
pauseCause = agent.AbortPausedByUser
}
reason := "paused_by_user"
if pauseCause == agent.AbortPausedByOrchestrator {
reason = "paused_by_orchestrator"
}
if pauseCause == agent.AbortPausedByProxy {
reason = "paused_by_proxy"
}
if _, err := s.m.pg.Exec(`UPDATE tasks SET schedule_paused=false, pause_reason=$2 WHERE id=$1`, t.ID, reason); err != nil {
return out, err return out, err
} }
t.updateLifecycle(func(state *taskLifecycleState) {
state.PauseReason = reason
state.ReconnectAttempt = 0
state.ReconnectMax = 0
state.ReconnectNextAt = 0
state.ReconnectExhausted = false
state.ReconnectLastError = ""
})
// The lifecycle row may already have been paused by another
// actor. Reinstall the requested cause barrier while holding
// concMu instead of assuming the existing barrier is sufficient.
s.engine.Pause(t.ID, pauseCause)
out.Paused = true out.Paused = true
out.Status = "paused" out.Status = "paused"
return out, nil return out, nil
@@ -131,8 +153,22 @@ func (s *Server) applyTaskControlWithCause(t *Task, action string, pauseCause er
pauseCause = agent.AbortPausedByUser pauseCause = agent.AbortPausedByUser
} }
s.engine.Pause(t.ID, pauseCause) s.engine.Pause(t.ID, pauseCause)
if err := s.m.ApplyTaskPauseOrigin(t.ID, pauseCause == agent.AbortPausedBySchedule); err != nil { reason := "paused_by_user"
if !wasEnginePaused && !wasQueued { if pauseCause == agent.AbortPausedBySchedule {
reason = "paused_by_schedule"
}
if pauseCause == agent.AbortPausedByOrchestrator {
reason = "paused_by_orchestrator"
}
if pauseCause == agent.AbortPausedByProxy {
reason = "paused_by_proxy"
}
if err := s.m.ApplyTaskPauseOriginReason(t.ID, reason, pauseCause == agent.AbortPausedBySchedule); err != nil {
// Do not compensate over a proxy pause that may have committed while
// this control request was failing. The persisted reason is the source
// of truth for which Engine barrier must remain installed.
stillProxyPaused := t.lifecycleSnapshot().PauseReason == "paused_by_proxy"
if !stillProxyPaused && !wasEnginePaused && !wasQueued {
s.engine.Resume(t) s.engine.Resume(t)
} }
return out, err return out, err
+4
View File
@@ -592,6 +592,7 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle {
wk.SetNoaEnabled(s.m.NoaCompactionEnabled) // 实验功能:noa 上下文压缩(平台级开关,每 run 读) wk.SetNoaEnabled(s.m.NoaCompactionEnabled) // 实验功能:noa 上下文压缩(平台级开关,每 run 读)
wk.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second) wk.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second)
wk.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) wk.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
wk.SetProxyProvider(s.m.ProxyAddr)
wk.SetWebSearch(s.webSearchFor("worker")) wk.SetWebSearch(s.webSearchFor("worker"))
wk.SetConstraintInject(s.constraintInjectWorker) // 操作约束注入 worker(可配置,默认开) wk.SetConstraintInject(s.constraintInjectWorker) // 操作约束注入 worker(可配置,默认开)
pl := agent.NewPlanner(plannerRuntime, "task-router", s.m.dir, tx, plannerRuntime.CompactionWindow(), s.agentMaxTurns("planner")) 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.SetKillWork(s.engine.KillWork)
pl.SetSteerWork(s.engine.SteerWork) pl.SetSteerWork(s.engine.SteerWork)
pl.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) pl.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
pl.SetProxyProvider(s.m.ProxyAddr)
pl.SetWebSearch(s.webSearchFor("planner")) pl.SetWebSearch(s.webSearchFor("planner"))
pl.SetConstraintInject(s.constraintInjectPlanner) // 操作约束注入 planner(可配置,默认开) pl.SetConstraintInject(s.constraintInjectPlanner) // 操作约束注入 planner(可配置,默认开)
// cold-digest §7: 冷节点后台压缩。引擎经权威解析器实际驱动的就是这套 per-task planner // cold-digest §7: 冷节点后台压缩。引擎经权威解析器实际驱动的就是这套 per-task planner
@@ -616,6 +618,7 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle {
main.SetMaxTokens(mainRuntime.maxTokens) main.SetMaxTokens(mainRuntime.maxTokens)
main.SetNoaEnabled(s.m.NoaCompactionEnabled) // 实验功能:noa 上下文压缩(平台级开关,每 run 读) main.SetNoaEnabled(s.m.NoaCompactionEnabled) // 实验功能:noa 上下文压缩(平台级开关,每 run 读)
main.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) main.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
main.SetProxyProvider(s.m.ProxyAddr)
main.SetWebSearch(s.webSearchFor("mainagent")) main.SetWebSearch(s.webSearchFor("mainagent"))
main.SetSteerWork(s.engine.SteerWork) // steer_work:人对运行中 work 实时纠偏 main.SetSteerWork(s.engine.SteerWork) // steer_work:人对运行中 work 实时纠偏
bundle := &taskAgentBundle{ bundle := &taskAgentBundle{
@@ -662,6 +665,7 @@ func (s *Server) workerForTaskSlot(t *Task, slot WorkerSlot) (*agent.Worker, err
worker.SetNoaEnabled(s.m.NoaCompactionEnabled) worker.SetNoaEnabled(s.m.NoaCompactionEnabled)
worker.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second) worker.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second)
worker.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) worker.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
worker.SetProxyProvider(s.m.ProxyAddr)
worker.SetWebSearch(s.webSearchFor("worker")) worker.SetWebSearch(s.webSearchFor("worker"))
worker.SetConstraintInject(s.constraintInjectWorker) worker.SetConstraintInject(s.constraintInjectWorker)
bundle.pinnedWorkers[id] = worker bundle.pinnedWorkers[id] = worker
+32 -2
View File
@@ -169,7 +169,9 @@ type Traffic struct {
// through (nil = dial targets directly). Both the intercepted and the // through (nil = dial targets directly). Both the intercepted and the
// transparent-passthrough paths honor it (go-mitmproxy's getUpstreamConn), // transparent-passthrough paths honor it (go-mitmproxy's getUpstreamConn),
// so no host escapes it. Hot-swappable at runtime via SetUpstreamProxy. // so no host escapes it. Hot-swappable at runtime via SetUpstreamProxy.
upstream atomic.Pointer[url.URL] upstream atomic.Pointer[url.URL]
upstreamMu sync.RWMutex
upstreamFailure func(error)
} }
// Open initializes the traffic tree, blob store and SQLite index under dir. // Open initializes the traffic tree, blob store and SQLite index under dir.
@@ -327,6 +329,27 @@ func (t *Traffic) SetUpstreamProxy(raw string) error {
return nil return nil
} }
// SetUpstreamFailureHandler observes failures from the configured global
// upstream proxy. The callback is intentionally best-effort and must not block
// the MITM request path.
func (t *Traffic) SetUpstreamFailureHandler(fn func(error)) {
t.upstreamMu.Lock()
t.upstreamFailure = fn
t.upstreamMu.Unlock()
}
func (t *Traffic) reportUpstreamFailure(err error) {
if err == nil || t.upstream.Load() == nil {
return
}
t.upstreamMu.RLock()
fn := t.upstreamFailure
t.upstreamMu.RUnlock()
if fn != nil {
go fn(err)
}
}
// ValidateProxyURL parses and checks a proxy URL (http/https/socks5, optional // ValidateProxyURL parses and checks a proxy URL (http/https/socks5, optional
// user:pass), returning the parsed URL. Exposed so callers can validate a global // user:pass), returning the parsed URL. Exposed so callers can validate a global
// proxy before persisting it even when the traffic proxy itself is disabled. // proxy before persisting it even when the traffic proxy itself is disabled.
@@ -400,7 +423,14 @@ func (s *sink) Response(f *mproxy.Flow) {
// the failure looks proxy/protocol-caused (h2 quirks, HEAD-with-body, protocol // the failure looks proxy/protocol-caused (h2 quirks, HEAD-with-body, protocol
// errors) — not a plain target-unreachable error — we flag the host for // errors) — not a plain target-unreachable error — we flag the host for
// transparent passthrough so future requests to it succeed instead of dying. // transparent passthrough so future requests to it succeed instead of dying.
func (s *sink) RequestError(f *mproxy.Flow, err error) { s.t.maybePassthrough(f, err) } func (s *sink) RequestError(f *mproxy.Flow, err error) {
s.t.reportUpstreamFailure(err)
s.t.maybePassthrough(f, err)
}
func (s *sink) HTTPConnectError(_ *mproxy.Flow, err error) {
s.t.reportUpstreamFailure(err)
}
// maybePassthrough marks a host to be tunneled transparently on the next // maybePassthrough marks a host to be tunneled transparently on the next
// connection, but only for errors the proxy itself caused — a target that is // connection, but only for errors the proxy itself caused — a target that is
@@ -61,6 +61,14 @@ export function EgressIpProvider({ children }: { children: React.ReactNode }) {
}; };
}, [probeForProxy]); }, [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( const setEnabled = React.useCallback(
async (nextEnabled: boolean) => { async (nextEnabled: boolean) => {
const settings = await api.setSettings({ show_egress_ip: nextEnabled }); const settings = await api.setSettings({ show_egress_ip: nextEnabled });
@@ -419,6 +419,14 @@ function TaskDetailInner() {
</Button> </Button>
</div> </div>
<p className="truncate text-xs text-muted-foreground">{task.goal}</p> <p className="truncate text-xs text-muted-foreground">{task.goal}</p>
{task.paused && task.pause_reason && (
<div className="flex flex-wrap items-center gap-2 text-xs text-amber-600">
<span>暂停原因:{task.pause_reason === "paused_by_proxy" ? "全局代理异常" : task.pause_reason}</span>
{task.reconnect && <span>重连 {task.reconnect.attempt}/{task.reconnect.max} {task.reconnect.exhausted ? "已耗尽" : task.reconnect.next_at ? `下次 ${new Date(task.reconnect.next_at * 1000).toLocaleTimeString()}` : "探测中"}</span>}
{task.reconnect?.last_error && <span className="max-w-full truncate" title={task.reconnect.last_error}>原因:{task.reconnect.last_error}</span>}
{task.pause_reason === "paused_by_proxy" && <Button size="sm" variant="outline" onClick={() => api.testTaskProxy(task.id).then((r) => r.ok ? (toast.success("代理已恢复,任务将自动继续"), load()) : toast.error(r.error || "代理探测失败")).catch((e) => toast.error((e as Error).message))}>测试代理</Button>}
</div>
)}
{task.remark ? ( {task.remark ? (
<p className="truncate text-xs text-muted-foreground" title={task.remark}> <p className="truncate text-xs text-muted-foreground" title={task.remark}>
备注:{task.remark} 备注:{task.remark}
+101 -2
View File
@@ -18,12 +18,13 @@ import { Button } from "@/components/ui/button";
import { Card, CardContent, CardDescription, CardHeader, CardTitle } from "@/components/ui/card"; import { Card, CardContent, CardDescription, CardHeader, CardTitle } from "@/components/ui/card";
import { Field, FieldContent, FieldDescription, FieldLabel } from "@/components/ui/field"; import { Field, FieldContent, FieldDescription, FieldLabel } from "@/components/ui/field";
import { Input } from "@/components/ui/input"; import { Input } from "@/components/ui/input";
import { Textarea } from "@/components/ui/textarea";
import { Label } from "@/components/ui/label"; import { Label } from "@/components/ui/label";
import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select"; import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select";
import { Switch } from "@/components/ui/switch"; import { Switch } from "@/components/ui/switch";
import { api } from "@/lib/api"; import { api } from "@/lib/api";
import { CHAT_SEND_MODE_OPTIONS, type ChatSendMode, setChatSendMode, useChatSendMode } from "@/lib/chat-send-mode"; 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 { useEgressIp } from "../../_components/egress-ip-banner";
import { BasicAuthCard } from "./_components/basic-auth-card"; import { BasicAuthCard } from "./_components/basic-auth-card";
@@ -55,6 +56,19 @@ export default function SystemSettingsPage() {
const [savingProxy, setSavingProxy] = React.useState(false); const [savingProxy, setSavingProxy] = React.useState(false);
const [globalProxyInput, setGlobalProxyInput] = React.useState(""); const [globalProxyInput, setGlobalProxyInput] = React.useState("");
const [savingGlobalProxy, setSavingGlobalProxy] = React.useState(false); 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); const [savingEgressVisibility, setSavingEgressVisibility] = React.useState(false);
// 出口探测:测一次真实请求,拿到"直连 / 经全局代理"两条出口 IP。 // 出口探测:测一次真实请求,拿到"直连 / 经全局代理"两条出口 IP。
const [egressTesting, setEgressTesting] = React.useState(false); const [egressTesting, setEgressTesting] = React.useState(false);
@@ -111,6 +125,18 @@ export default function SystemSettingsPage() {
setProxyInput(s.web_search_proxy ?? ""); setProxyInput(s.web_search_proxy ?? "");
const savedProxy = s.global_proxy ?? ""; const savedProxy = s.global_proxy ?? "";
setGlobalProxyInput(savedProxy); 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));
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 ?? ""); setPyInterp(s.python_interpreter ?? "");
setWorkers(String(s.workers ?? 3)); setWorkers(String(s.workers ?? 3));
setWorkerPools(s.worker_pools ?? []); setWorkerPools(s.worker_pools ?? []);
@@ -201,6 +227,11 @@ export default function SystemSettingsPage() {
.then((s) => { .then((s) => {
apply(s); apply(s);
setSettingsLoaded(true); setSettingsLoaded(true);
void api.proxyPool().then((pool) => {
setProxyPool(pool.entries);
setProxyPoolActiveId(pool.active_id);
setProxyPoolLogs(pool.switch_log);
}).catch(() => undefined);
}) })
.catch(() => { .catch(() => {
setSettingsLoaded(false); setSettingsLoaded(false);
@@ -414,6 +445,10 @@ export default function SystemSettingsPage() {
}; };
const saveGlobalProxy = () => { const saveGlobalProxy = () => {
if (proxyPool.length > 0) {
toast.info("代理池非空时,全局代理输入框仅用于回显,请在代理池中编辑或切换");
return;
}
setSavingGlobalProxy(true); setSavingGlobalProxy(true);
api api
.setSettings({ global_proxy: globalProxyInput.trim() }) .setSettings({ global_proxy: globalProxyInput.trim() })
@@ -428,6 +463,29 @@ export default function SystemSettingsPage() {
.finally(() => setSavingGlobalProxy(false)); .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 = () => {
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 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) })
.then((s) => { apply(s); toast.success("已保存代理重连策略"); })
.catch((e) => toast.error("保存失败:" + (e as Error).message));
};
// 测表单里当前的值,不要求先保存;后端返回直连 / 经代理两条。 // 测表单里当前的值,不要求先保存;后端返回直连 / 经代理两条。
const testEgress = () => { const testEgress = () => {
const seq = ++egressSeq.current; const seq = ++egressSeq.current;
@@ -436,6 +494,7 @@ export default function SystemSettingsPage() {
api api
.testEgressProxy({ .testEgressProxy({
global_proxy: globalProxyInput.trim(), global_proxy: globalProxyInput.trim(),
resume: true,
...(egressProbeUrl.trim() ? { probe_url: egressProbeUrl.trim() } : {}), ...(egressProbeUrl.trim() ? { probe_url: egressProbeUrl.trim() } : {}),
}) })
@@ -751,12 +810,18 @@ export default function SystemSettingsPage() {
</Button> </Button>
{/* 测表单里当前的值,不要求先保存。 */} {/* 测表单里当前的值,不要求先保存。 */}
<Button type="button" variant="outline" onClick={testEgress} disabled={!loaded || egressTesting}> <Button type="button" variant="outline" onClick={testEgress} disabled={!loaded || egressTesting}>
{egressTesting ? "测试中…" : "测试"} {egressTesting ? "测试中…" : "测试代理"}
</Button> </Button>
</div> </div>
<p className="text-muted-foreground text-xs"> <p className="text-muted-foreground text-xs">
{globalProxyInput.trim() ? "已配置 · 所有目标流量经此代理出网" : "未配置 · 目标流量直连出网"} {globalProxyInput.trim() ? "已配置 · 所有目标流量经此代理出网" : "未配置 · 目标流量直连出网"}
</p> </p>
<div className="grid grid-cols-1 gap-2 sm:grid-cols-3 items-end">
<div><Label htmlFor="proxy-reconnect-max">最大重连次数</Label><Input id="proxy-reconnect-max" type="number" min={1} value={proxyReconnectMax} onChange={(e) => setProxyReconnectMax(e.target.value)} /></div>
<div><Label htmlFor="proxy-reconnect-interval">重连间隔(秒)</Label><Input id="proxy-reconnect-interval" type="number" min={1} value={proxyReconnectInterval} onChange={(e) => setProxyReconnectInterval(e.target.value)} /></div>
<Button type="button" variant="outline" onClick={saveProxyReconnect}>保存重连策略</Button>
</div>
<p className="text-muted-foreground text-xs">代理异常暂停的任务会按此策略探测;次数耗尽后保持暂停,可手动恢复。</p>
<div className="flex items-center justify-between gap-4 rounded-md border px-3 py-2"> <div className="flex items-center justify-between gap-4 rounded-md border px-3 py-2">
<div className="min-w-0"> <div className="min-w-0">
<Label htmlFor="show-egress-ip" className="text-sm font-normal"> <Label htmlFor="show-egress-ip" className="text-sm font-normal">
@@ -863,6 +928,40 @@ export default function SystemSettingsPage() {
))} ))}
</div> </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.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}{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>
))}
</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> </CardContent>
</Card> </Card>
+20 -1
View File
@@ -51,6 +51,7 @@ export function FindingAutoChat({
const [historyOpen, setHistoryOpen] = React.useState(false); const [historyOpen, setHistoryOpen] = React.useState(false);
const [history, setHistory] = React.useState<Conversation[]>([]); const [history, setHistory] = React.useState<Conversation[]>([]);
const cursor = React.useRef(0); const cursor = React.useRef(0);
const chatViewportRef = React.useRef<HTMLDivElement | null>(null);
const safeTitle = findingTitle.replace(/[\[\]\r\n]/g, " ").trim().slice(0, 100); const safeTitle = findingTitle.replace(/[\[\]\r\n]/g, " ").trim().slice(0, 100);
const storageKey = `artex.finding.${agentKey}-conversation.${findingId}`; const storageKey = `artex.finding.${agentKey}-conversation.${findingId}`;
@@ -136,6 +137,18 @@ export function FindingAutoChat({
}; };
}, [conversation, open, storageKey]); }, [conversation, open, storageKey]);
React.useEffect(() => {
if (!open || !conversation || loading || !messages.length) return;
const viewport = chatViewportRef.current;
if (!viewport) return;
const key = `artex.conversation-scroll.${conversation.id}`;
const saved = window.sessionStorage.getItem(key);
requestAnimationFrame(() => {
if (saved != null) viewport.scrollTop = Number(saved) || 0;
else viewport.scrollTop = viewport.scrollHeight;
});
}, [conversation, loading, messages.length, open]);
React.useEffect(() => { React.useEffect(() => {
if (!open || !conversation || !running) return; if (!open || !conversation || !running) return;
let live = true; let live = true;
@@ -271,7 +284,13 @@ export function FindingAutoChat({
<XIcon /> <XIcon />
</Button> </Button>
</header> </header>
<div className="min-h-0 min-w-0 flex-1 overflow-x-hidden overflow-y-auto p-3"> <div
ref={chatViewportRef}
onScroll={(event) => {
if (conversation) window.sessionStorage.setItem(`artex.conversation-scroll.${conversation.id}`, String(event.currentTarget.scrollTop));
}}
className="min-h-0 min-w-0 flex-1 overflow-x-hidden overflow-y-auto p-3"
>
<div className="w-full min-w-0 max-w-full"> <div className="w-full min-w-0 max-w-full">
{loading ? ( {loading ? (
<div className="flex items-center justify-center py-10 text-muted-foreground text-sm"> <div className="flex items-center justify-center py-10 text-muted-foreground text-sm">
+31 -4
View File
@@ -18,6 +18,8 @@ function AuthenticatedImage({
const [captureTimestamp, setCaptureTimestamp] = React.useState<string | null>(null); const [captureTimestamp, setCaptureTimestamp] = React.useState<string | null>(null);
const [zoomed, setZoomed] = React.useState(false); const [zoomed, setZoomed] = React.useState(false);
const [zoomScale, setZoomScale] = React.useState(1); const [zoomScale, setZoomScale] = React.useState(1);
const [pan, setPan] = React.useState({ x: 0, y: 0 });
const dragRef = React.useRef<{ x: number; y: number; panX: number; panY: number } | null>(null);
const imageSrc = typeof src === "string" ? src : undefined; const imageSrc = typeof src === "string" ? src : undefined;
React.useEffect(() => { React.useEffect(() => {
@@ -29,6 +31,7 @@ function AuthenticatedImage({
setCaptureTimestamp(null); setCaptureTimestamp(null);
setZoomed(false); setZoomed(false);
setZoomScale(1); setZoomScale(1);
setPan({ x: 0, y: 0 });
return () => { return () => {
cancelled = true; cancelled = true;
}; };
@@ -75,31 +78,55 @@ function AuthenticatedImage({
onClick={() => renderedSrc && setZoomed(true)} onClick={() => renderedSrc && setZoomed(true)}
aria-label="放大查看报告截图" aria-label="放大查看报告截图"
> >
<img {...props} src={renderedSrc} alt={alt || "报告截图,点击放大"} /> <img {...props} src={renderedSrc} alt={alt || "报告截图,点击放大"} draggable={false} onDragStart={(event) => event.preventDefault()} />
{timestamp} {timestamp}
</button> </button>
<Dialog <Dialog
open={zoomed} open={zoomed}
onOpenChange={(next) => { onOpenChange={(next) => {
setZoomed(next); setZoomed(next);
if (!next) setZoomScale(1); if (!next) {
setZoomScale(1);
setPan({ x: 0, y: 0 });
}
}} }}
> >
<DialogContent <DialogContent
showCloseButton showCloseButton
className="flex max-h-[96dvh] max-w-[96vw] items-center justify-center overflow-hidden bg-black/85 p-3 backdrop-blur-sm sm:max-w-[96vw]" className="flex max-h-[96dvh] max-w-[96vw] items-center justify-center overflow-hidden bg-black/85 p-3 backdrop-blur-sm sm:max-w-[96vw]"
> >
<div className="relative flex max-h-[90dvh] max-w-full items-center justify-center overflow-auto"> <div
<div className="relative inline-block origin-center transition-transform duration-150" style={{ transform: `scale(${zoomScale})` }}> className="relative flex h-[90dvh] w-[90vw] max-w-full items-center justify-center overflow-hidden"
onPointerDown={(event) => {
if (zoomScale <= 1 || event.button !== 0) return;
event.currentTarget.setPointerCapture(event.pointerId);
dragRef.current = { x: event.clientX, y: event.clientY, panX: pan.x, panY: pan.y };
}}
onPointerMove={(event) => {
const drag = dragRef.current;
if (!drag) return;
setPan({ x: drag.panX + event.clientX - drag.x, y: drag.panY + event.clientY - drag.y });
}}
onPointerUp={(event) => {
dragRef.current = null;
event.currentTarget.releasePointerCapture?.(event.pointerId);
}}
onPointerCancel={() => { dragRef.current = null; }}
style={{ cursor: zoomScale > 1 ? (dragRef.current ? "grabbing" : "grab") : "default" }}
>
<div className="relative inline-block origin-center" style={{ transform: `translate(${pan.x}px, ${pan.y}px) scale(${zoomScale})` }}>
<img <img
src={renderedSrc} src={renderedSrc}
alt={alt || "报告截图"} alt={alt || "报告截图"}
draggable={false}
onDragStart={(event) => event.preventDefault()}
className="max-h-[90dvh] max-w-[90vw] rounded-md object-contain" className="max-h-[90dvh] max-w-[90vw] rounded-md object-contain"
/> />
{timestamp} {timestamp}
</div> </div>
</div> </div>
<div className="absolute top-3 left-3 z-10 flex items-center gap-1 rounded-lg bg-black/60 p-1 backdrop-blur-sm"> <div className="absolute top-3 left-3 z-10 flex items-center gap-1 rounded-lg bg-black/60 p-1 backdrop-blur-sm">
<span className="px-2 text-xs font-medium text-white tabular-nums">{Math.round(zoomScale * 100)}%</span>
<Button <Button
variant="ghost" variant="ghost"
size="icon-sm" size="icon-sm"
+33 -4
View File
@@ -34,6 +34,8 @@ import type {
DeleteTaskResult, DeleteTaskResult,
Edge, Edge,
EgressReport, EgressReport,
ProxyPoolEntry,
ProxySwitchLog,
EvidenceBodyPreview, EvidenceBodyPreview,
ExplorationNodePage, ExplorationNodePage,
ExplorationNodeQuery, ExplorationNodeQuery,
@@ -119,6 +121,24 @@ import type {
SystemMetrics, SystemMetrics,
} from "@/lib/types"; } from "@/lib/types";
/**
* Read a download name from Content-Disposition. RFC 5987's filename* is
* preferred because it preserves UTF-8 names; older filename remains the
* fallback for clients that do not understand it.
*/
export function downloadFilename(contentDisposition: string, fallback: string): string {
const encoded = contentDisposition.match(/(?:^|;)\s*filename\*\s*=\s*UTF-8''([^;]+)/i)?.[1];
if (encoded) {
try {
return decodeURIComponent(encoded.replace(/^"|"$/g, ""));
} catch {
// Fall through to the legacy token when a malformed header is received.
}
}
const legacy = contentDisposition.match(/(?:^|;)\s*filename\s*=\s*(?:"([^"]*)"|([^;\s]+))/i);
return (legacy?.[1] ?? legacy?.[2] ?? fallback).trim() || fallback;
}
function getToken(): string | null { function getToken(): string | null {
if (typeof window === "undefined") return null; if (typeof window === "undefined") return null;
return localStorage.getItem("artex_token"); return localStorage.getItem("artex_token");
@@ -534,6 +554,7 @@ export const api = {
}, },
workspaceDownload: async (path: string) => { workspaceDownload: async (path: string) => {
let blob: Blob; let blob: Blob;
let contentDisposition = "";
if (MOCK) { if (MOCK) {
blob = new Blob([`(demo)${path} 的下载内容示例。`], { type: "text/plain" }); blob = new Blob([`(demo)${path} 的下载内容示例。`], { type: "text/plain" });
} else { } else {
@@ -542,12 +563,13 @@ export const api = {
headers: { ...(token ? { Authorization: `Bearer ${token}` } : {}) }, headers: { ...(token ? { Authorization: `Bearer ${token}` } : {}) },
}); });
if (!r.ok) throw new Error(`下载失败: ${r.status}`); if (!r.ok) throw new Error(`下载失败: ${r.status}`);
contentDisposition = r.headers.get("Content-Disposition") ?? "";
blob = await r.blob(); blob = await r.blob();
} }
const objUrl = URL.createObjectURL(blob); const objUrl = URL.createObjectURL(blob);
const a = document.createElement("a"); const a = document.createElement("a");
a.href = objUrl; a.href = objUrl;
a.download = path.split("/").pop() || "download"; a.download = downloadFilename(contentDisposition, path.split("/").pop() || "download");
document.body.appendChild(a); document.body.appendChild(a);
a.click(); a.click();
a.remove(); a.remove();
@@ -703,8 +725,7 @@ export const api = {
const blob = await r.blob(); const blob = await r.blob();
// 文件名优先取后端 Content-Disposition,取不到则用默认名。 // 文件名优先取后端 Content-Disposition,取不到则用默认名。
const disp = r.headers.get("Content-Disposition") ?? ""; const disp = r.headers.get("Content-Disposition") ?? "";
const m = disp.match(/filename="?([^"]+)"?/); const filename = downloadFilename(disp, `findings-export`);
const filename = m?.[1] ?? `findings-export`;
const url = URL.createObjectURL(blob); const url = URL.createObjectURL(blob);
const a = document.createElement("a"); const a = document.createElement("a");
a.href = url; a.href = url;
@@ -971,8 +992,16 @@ export const api = {
}) => post<{ ok: boolean; error?: string; count?: number; backend?: string }>(`/settings/web-search/test`, patch), }) => post<{ ok: boolean; error?: string; count?: number; backend?: string }>(`/settings/web-search/test`, patch),
// global_proxy 传空 = 测直连;probe_url 可覆盖默认探测地址(隔离网用)。 // global_proxy 传空 = 测直连;probe_url 可覆盖默认探测地址(隔离网用)。
testEgressProxy: (patch: { global_proxy?: string; probe_url?: string }) => testEgressProxy: (patch: { global_proxy?: string; probe_url?: string; resume?: boolean }) =>
post<EgressReport>(`/settings/proxy/test`, patch), 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 推送 ---- // ---- 漏洞 IM 推送 ----
// 渠道是多实例资源(同一类型可配多个机器人、各有过滤规则),因此独立成组, // 渠道是多实例资源(同一类型可配多个机器人、各有过滤规则),因此独立成组,
+9
View File
@@ -3350,6 +3350,15 @@ export const settings: Settings = {
tavily_key_set: true, tavily_key_set: true,
web_search_proxy: "", web_search_proxy: "",
global_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, show_egress_ip: false,
python_interpreter: "/usr/bin/python3", python_interpreter: "/usr/bin/python3",
workers: 3, workers: 3,
+6
View File
@@ -2406,6 +2406,12 @@ function route(m: string, path: string, seg: string[], q: URLSearchParams, b: Re
], ],
warnings: [], 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 === "/settings/python/detect") return { python_interpreter: "/usr/bin/python3" };
if (path === "/chat") if (path === "/chat")
return { reply: "(demo)我已把该建议注入为一条高优意图,work agent 会尽快执行。", mode: "hint" }; return { reply: "(demo)我已把该建议注入为一条高优意图,work agent 会尽快执行。", mode: "hint" };
+30
View File
@@ -42,6 +42,8 @@ export interface Task {
active_run_seconds?: number; // completed running segments in seconds active_run_seconds?: number; // completed running segments in seconds
run_segment_started_unix?: number; // current running segment start; 0 when inactive run_segment_started_unix?: number; // current running segment start; 0 when inactive
paused?: boolean; paused?: boolean;
pause_reason?: string;
reconnect?: { attempt: number; max: number; next_at?: number; exhausted: boolean; last_error?: string };
queued?: boolean; queued?: boolean;
active?: boolean; active?: boolean;
in_flight?: number; in_flight?: number;
@@ -1002,6 +1004,18 @@ export interface Settings {
// 全局出口代理(http/https/socks5,可带 user:pass),所有目标流量走它。开启流量捕获时作为 // 全局出口代理(http/https/socks5,可带 user:pass),所有目标流量走它。开启流量捕获时作为
// MITM 上游;关闭捕获时直接注入 agent 的 bash/WebFetch。空=直连。 // MITM 上游;关闭捕获时直接注入 agent 的 bash/WebFetch。空=直连。
global_proxy?: string; 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;
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;默认关闭 show_egress_ip?: boolean; // 在顶栏显示直连 / 全局代理出口 IP;默认关闭
python_interpreter?: string; // 自定义脚本工具的 python 解释器路径(空=运行时检测) python_interpreter?: string; // 自定义脚本工具的 python 解释器路径(空=运行时检测)
workers?: number; // 并发工作 agent 数(默认3);对之后启动的任务生效 workers?: number; // 并发工作 agent 数(默认3);对之后启动的任务生效
@@ -1027,6 +1041,22 @@ export interface Settings {
notify_digest_interval_min?: number; // 汇总模式周期(分钟),默认 30 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 { export interface WorkerPool {
profile_id: string; profile_id: string;
workers: number; workers: number;