From 22132a4d21e1e43a654f608ac4160398914a6dae Mon Sep 17 00:00:00 2001 From: pphua Date: Fri, 28 Aug 2026 14:03:07 +0800 Subject: [PATCH] chore: sync upstream v0.3.6 changes --- CHANGELOG.md | 43 +++++++ agent/blackboard_inheritance_test.go | 2 +- agent/capture_usage_test.go | 11 ++ agent/chat.go | 44 +++++--- agent/goals.go | 7 +- agent/mainagent.go | 32 ++++-- agent/planner.go | 38 ++++--- agent/provider.go | 41 ++++++- agent/tools.go | 57 +++++++++- agent/worker.go | 14 ++- db/commands.go | 69 ++++++++++-- db/config.go | 25 +++-- db/exploration.go | 34 ++++++ db/exploration_sources.go | 47 ++++++++ db/schema.sql | 4 + go.mod | 4 +- go.sum | 4 +- llmpool/pool.go | 38 +++++++ llmpool/pool_test.go | 11 ++ llmrec/llmrec.go | 47 +++++++- llmrec/llmrec_test.go | 36 ++++++ server/commands.go | 40 +++++-- server/dto.go | 4 + server/goals.go | 3 +- server/orchestration.go | 10 +- server/server.go | 50 ++++++++- server/server_mgmt.go | 9 +- server/task_llm.go | 106 +++++++++++++++++- server/task_llm_test.go | 21 ++++ web/src/app/(main)/function/commands/page.tsx | 93 ++++++++++++++- web/src/app/(main)/system/llm/page.tsx | 43 +++++-- web/src/lib/api.ts | 14 ++- web/src/lib/mock/handler.ts | 14 ++- web/src/lib/types.ts | 9 ++ 34 files changed, 897 insertions(+), 127 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a3fbdeb..f38f7b0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,49 @@ ## [Unreleased] +## [0.3.6] - 2026-08-27 + +### LLM + +#### 修复的问题 + +- 修复「非流式」开关保存后重新打开又变回「流式」的问题（#69）：配置列表接口的 DTO 漏了 `streaming` 字段，响应从不返回它，前端读到 `undefined` 后一律回落成默认流式；实际写入与 DB 存储本无问题，只是读不回来。DTO 补回该字段（不加 `omitempty`，`false` 也必须出现在响应里）。 +- 测试连接改为校验模型确有回复、并按该配置真实的收发模式来测（#65）：此前只判 HTTP 是否成功，请求通但模型零回复（思考烧光预算 / 正文被安全策略吞掉 / 兼容层丢 `content`）也报「连接成功」，与会话里「不回话」的表现割裂；且测试恒走流式，非流式配置测的其实是另一条通道。现在空回复直接判失败、成功时在提示里展示模型回复，并把当前流式开关一并带入测试，让「测试通过」与「会话跑得通」保持一致。 + +### Agent + +#### 修改的功能 + +- `list_facts` 改为分页 + 关键词过滤，避免事实多时单次调用撑爆上下文（#74）：默认返回最新 20 条，支持 `limit`（上限 100）/ `before` 游标 / `q` 摘要关键词，返回 `{facts, total, has_more, next_before}`；单条 summary 过长按字数截断（全文仍走 `node_detail`）。worker / planner 提示词同步为分页语义。旧库通过一次性迁移把新参数 schema 刷进工具目录表（`SeedTool` 首插入only，否则工具管理页显示「无参数」）。 + +### 工具 + +#### 新增的功能 + +- 「工具执行」页新增工具调用次数统计（#72）：工具栏「统计」按钮打开弹窗，按工具展示调用次数、占比与失败次数（次数降序）；沿用列表的任务 / 关键词筛选，统计整个结果集而非当前页，弹窗打开时才拉取。 + +### 依赖 + +#### 修改的功能 + +- 升级 norma v0.2.9 → v0.3.2：修复 OpenAI 系响应里被静默丢弃的 reasoning/refusal（Responses 补 `reasoning_text`）、reasoning 字段三名去重、空响应（零内容块）有界重试、压缩边界切断工具配对导致的网关 400（同时修复已烤进 transcript 的存量孤儿）。 + +### 贡献者 + +- [@Autumn-27](https://github.com/Autumn-27) + +## [0.3.5] - 2026-08-25 + +### LLM + +#### 新增的功能 + +- 每个 LLM 配置支持流式/非流式切换（默认流式）：开走流式 SSE；关走真·非流式（`stream:false`、一次性返回完整 JSON），可绕开部分网关糟糕的 SSE 实现（空帧、思考字段丢帧），代价是失去运行中的实时进度与实时 Token 计数。worker / planner / mainagent / chat / goals 均按当前激活配置动态取值；`llmpool` / `llmrec` / 任务运行时三层 Provider 包装均兼容非流式；`llm_profiles` 新增 `streaming` 列并 `ALTER` 补旧库（默认 `true`，旧配置无感）。 + +### 贡献者 + +- [@Autumn-27](https://github.com/Autumn-27) + ## [0.3.4] - 2026-08-24 ### UI diff --git a/agent/blackboard_inheritance_test.go b/agent/blackboard_inheritance_test.go index ad4fe4b..4549ff5 100644 --- a/agent/blackboard_inheritance_test.go +++ b/agent/blackboard_inheritance_test.go @@ -326,7 +326,7 @@ func TestBlackboardToolsReadDirectSources(t *testing.T) { t.Fatalf("inherited host context missing: %#v", coverage) } - facts := callReadJSON(t, tools.listFacts(), `{}`).([]any) + facts := callReadJSON(t, tools.listFacts(), `{}`).(map[string]any)["facts"].([]any) seenSource, seenGrand := false, false for _, raw := range facts { item := raw.(map[string]any) diff --git a/agent/capture_usage_test.go b/agent/capture_usage_test.go index c76f79c..5b6389c 100644 --- a/agent/capture_usage_test.go +++ b/agent/capture_usage_test.go @@ -19,6 +19,17 @@ func (p captureUsageProvider) Stream(ctx context.Context, _ llm.CompletionReques return func(yield func(llm.StreamEvent, error) bool) { p.stream(ctx, yield) } } +func (p captureUsageProvider) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) { + acc := llm.NewAccumulator() + for ev, err := range p.Stream(ctx, req) { + if err != nil { + return llm.Message{}, "", llm.Usage{}, err + } + acc.Add(ev) + } + return acc.Message(), acc.StopReason, acc.Usage, nil +} + func TestCaptureRunPersistsUsageOnProviderFailure(t *testing.T) { wantErr := errors.New("provider failed after reporting usage") provider := captureUsageProvider{stream: func(_ context.Context, yield func(llm.StreamEvent, error) bool) { diff --git a/agent/chat.go b/agent/chat.go index 41be93b..0f38582 100644 --- a/agent/chat.go +++ b/agent/chat.go @@ -6,13 +6,13 @@ import ( "path/filepath" "time" + "github.com/Autumn-27/artex/db" + "github.com/Autumn-27/artex/guard" "github.com/Autumn-27/norma/agentcore" "github.com/Autumn-27/norma/llm" "github.com/Autumn-27/norma/permission" actool "github.com/Autumn-27/norma/tool" "github.com/Autumn-27/norma/transcript" - "github.com/Autumn-27/artex/db" - "github.com/Autumn-27/artex/guard" ) // ChatAgent is the generic, task-independent conversational runner behind the chat @@ -22,21 +22,28 @@ import ( // LS/Glob/Grep) plus whatever skills/MCP the key is made visible; NO pentest // graph/task context is injected (that stays exclusive to MainAgent). type ChatAgent struct { - prov llm.Provider - model string - workDir string - tx *transcript.Store - window int - proxyAddr string - proxyCACert string - webSearch WebSearchOpts - guard *guard.Guard // optional; nil disables intercept hooks for chat + prov llm.Provider + model string + workDir string + tx *transcript.Store + window int + proxyAddr string + proxyCACert string + webSearch WebSearchOpts + guard *guard.Guard // optional; nil disables intercept hooks for chat + nonStreamingFn func() bool // resolver: use non-streaming (Complete) path? (nil = streaming) } func NewChatAgent(prov llm.Provider, model, workDir string, tx *transcript.Store, window int) *ChatAgent { return &ChatAgent{prov: prov, model: model, workDir: workDir, tx: tx, window: window} } +// SetNonStreaming wires a resolver deciding whether chat runs use the +// non-streaming model path (true = non-streaming). nil/unset = streaming. +func (c *ChatAgent) SetNonStreaming(fn func() bool) { c.nonStreamingFn = fn } + +func (c *ChatAgent) nonStreaming() bool { return c.nonStreamingFn != nil && c.nonStreamingFn() } + // SetProxy points the chat agent's WebFetch/Bash at the recording proxy plus the // CA cert it trusts (empty addr = direct). Kept for parity with the other agents. func (c *ChatAgent) SetProxy(addr, caCert string) { c.proxyAddr, c.proxyCACert = addr, caCert } @@ -115,18 +122,19 @@ func (c *ChatAgent) Chat(ctx context.Context, agentKey, sessionID, message strin BraveSearchAPIKey: ws.BraveKey, TavilySearchAPIKey: ws.TavilyKey, WebSearchProxy: ws.Proxy, - BashEnv: proxyEnv(c.proxyAddr, c.proxyCACert), // Bash 子命令默认走代理+信任 CA - WorkingDir: sessionWorkDir, - MaxTurns: maxTurns, - MaxDuration: maxDuration, - Compaction: compactionConfig(c.window), - Todos: actool.NewTodoStore(), + BashEnv: proxyEnv(c.proxyAddr, c.proxyCACert), // Bash 子命令默认走代理+信任 CA + WorkingDir: sessionWorkDir, + MaxTurns: maxTurns, + MaxDuration: maxDuration, + Compaction: compactionConfig(c.window), + Todos: actool.NewTodoStore(), // large tool output spills to cmd-output/ under the session dir. // 截断上限用 SDK 默认(tool.Capture 的 30000 字符)。 ToolOutputDir: filepath.Join(sessionWorkDir, "cmd-output"), // 命中预算(步数)→ SDK 跑收尾:输出一句总结。Prompt 与收尾轮数按本 agent key 后台可编辑 // (自定义 agent 各自一份;留空/0 用通用默认:10 轮)。 - Settlement: wrapupSettlement(agentKey, nil), + Settlement: wrapupSettlement(agentKey, nil), + NonStreaming: c.nonStreaming(), // 该 profile 选非流式时走 Provider.Complete } if c.guard != nil { opts.Hooks = c.guard.Hooks() diff --git a/agent/goals.go b/agent/goals.go index 388515a..b98e9fe 100644 --- a/agent/goals.go +++ b/agent/goals.go @@ -98,13 +98,13 @@ func DecomposeGoals(ctx context.Context, prov llm.Provider, dataDir, goalText, d if prov == nil { return nil } - return DecomposeGoalsWithProvider(ctx, prov, dataDir, goalText, desc, as, ts, taskID, emit) + return DecomposeGoalsWithProvider(ctx, prov, dataDir, goalText, desc, as, ts, taskID, false, emit) } // DecomposeGoalsWithProvider is the task-runtime variant used when a task has an // ordered provider chain. It preserves the same tools and write behavior while // letting the caller own provider selection/failover. -func DecomposeGoalsWithProvider(ctx context.Context, prov llm.Provider, dataDir, goalText, desc string, as *db.AssetStore, ts *db.ExplorationStore, taskID int64, emit func(db.Activity)) []GoalSpec { +func DecomposeGoalsWithProvider(ctx context.Context, prov llm.Provider, dataDir, goalText, desc string, as *db.AssetStore, ts *db.ExplorationStore, taskID int64, nonStreaming bool, emit func(db.Activity)) []GoalSpec { if prov == nil { return nil } @@ -145,7 +145,8 @@ func DecomposeGoalsWithProvider(ctx context.Context, prov llm.Provider, dataDir, PermissionMode: acperm.ModeBypass, DisableBackgroundTasks: true, // 3 步(抽约束 → 登记范围 → 拆目标)各需一次工具调用,给足回合避免收尾前漏调 set_goals。 - MaxTurns: 8, + MaxTurns: 8, + NonStreaming: nonStreaming, // 该 profile 选非流式时走 Provider.Complete }, userMsg, captureEmit) // set_goals persisted the goals directly; read them back so the caller sees what // was written (empty slice ⇒ the LLM produced nothing ⇒ caller falls back). diff --git a/agent/mainagent.go b/agent/mainagent.go index 99531df..9c40eca 100644 --- a/agent/mainagent.go +++ b/agent/mainagent.go @@ -17,19 +17,26 @@ import ( // (→planner) or direct high-priority intents (→frontier). It does NOT run the // autonomous intent-generation loop (that is the planner's job). type MainAgent struct { - prov llm.Provider - model string - tx *transcript.Store // raw LLM conversation persistence (nil = off) - window int // context window in tokens (for compaction) - windowFn func() int // optional dynamic task-chain minimum - maxTurns int // max agent turns per run (0 = unlimited) - proxyAddr string // recording proxy for WebFetch (empty = direct) - proxyCACert string // recording proxy's CA cert path (HTTPS verify) - webSearch WebSearchOpts // web_search tool backend selection (off by default) - workDir string // shared work dir (surfaced in prompt as artifact-output target) - steerWork func(intentID int64, msg string) error // engine callback: steer a running work (nil = off) + prov llm.Provider + model string + tx *transcript.Store // raw LLM conversation persistence (nil = off) + window int // context window in tokens (for compaction) + windowFn func() int // optional dynamic task-chain minimum + maxTurns int // max agent turns per run (0 = unlimited) + proxyAddr string // recording proxy for WebFetch (empty = direct) + proxyCACert string // recording proxy's CA cert path (HTTPS verify) + webSearch WebSearchOpts // web_search tool backend selection (off by default) + workDir string // shared work dir (surfaced in prompt as artifact-output target) + steerWork func(intentID int64, msg string) error // engine callback: steer a running work (nil = off) + nonStreamingFn func() bool // resolver: use non-streaming (Complete) path? (nil = streaming) } +// SetNonStreaming wires a resolver deciding whether runs use the non-streaming +// model path (true = non-streaming). nil/unset = streaming (default). +func (m *MainAgent) SetNonStreaming(fn func() bool) { m.nonStreamingFn = fn } + +func (m *MainAgent) nonStreaming() bool { return m.nonStreamingFn != nil && m.nonStreamingFn() } + func NewMainAgent(prov llm.Provider, model, workDir string, tx *transcript.Store, window, maxTurns int) *MainAgent { return &MainAgent{prov: prov, model: model, workDir: workDir, tx: tx, window: window, maxTurns: maxTurns} } @@ -128,7 +135,8 @@ func (m *MainAgent) Chat(ctx context.Context, taskID int64, as *db.AssetStore, t Compaction: compactionConfig(m.compactionWindow()), // long chats stay within the window Todos: actool.NewTodoStore(), // 会话级临时待办（TodoWrite），纯规划用，退出即丢 // 命中预算(步数)→ SDK 跑收尾:向用户输出一句进展总结。Prompt 与收尾轮数可后台编辑(默认 10 轮)。 - Settlement: wrapupSettlement("mainagent", nil), + Settlement: wrapupSettlement("mainagent", nil), + NonStreaming: m.nonStreaming(), // 该 profile 选非流式时走 Provider.Complete } if m.tx != nil { // persist raw human↔AI conversation; one accumulating file per task opts.Transcript = m.tx diff --git a/agent/planner.go b/agent/planner.go index dfacc91..2647314 100644 --- a/agent/planner.go +++ b/agent/planner.go @@ -20,19 +20,20 @@ import ( // assets, judges whether the task goal is met, and emits 0..N exploration intents // into the frontier. It is the sole intent generator. type Planner struct { - prov llm.Provider - model string - tx *transcript.Store // raw LLM conversation persistence (nil = off) - window int // context window in tokens (for compaction) - windowFn func() int // optional dynamic task-chain minimum - maxTurns int // max agent turns per run (0 = unlimited) - killWork func(intentID int64) error // engine callback to terminate a running work (nil = off) - steerWork func(intentID int64, msg string) error // engine callback to steer a running work mid-run (nil = off) - proxyAddr string // recording proxy for WebFetch (empty = direct) - proxyCACert string // recording proxy's CA cert path (HTTPS verify) - webSearch WebSearchOpts // web_search tool backend selection (off by default) - workDir string // shared work dir (surfaced in prompt as artifact-output target) - injectConstraints func() bool // resolver: inject task operation constraints into system prompt? (nil = yes) + prov llm.Provider + model string + tx *transcript.Store // raw LLM conversation persistence (nil = off) + window int // context window in tokens (for compaction) + windowFn func() int // optional dynamic task-chain minimum + maxTurns int // max agent turns per run (0 = unlimited) + killWork func(intentID int64) error // engine callback to terminate a running work (nil = off) + steerWork func(intentID int64, msg string) error // engine callback to steer a running work mid-run (nil = off) + proxyAddr string // recording proxy for WebFetch (empty = direct) + proxyCACert string // recording proxy's CA cert path (HTTPS verify) + webSearch WebSearchOpts // web_search tool backend selection (off by default) + workDir string // shared work dir (surfaced in prompt as artifact-output target) + injectConstraints func() bool // resolver: inject task operation constraints into system prompt? (nil = yes) + nonStreamingFn func() bool // resolver: use non-streaming (Complete) path? (nil = streaming) // 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 @@ -48,6 +49,12 @@ func NewPlanner(prov llm.Provider, model, workDir string, tx *transcript.Store, func (p *Planner) SetCompactionWindowResolver(fn func() int) { p.windowFn = fn } +// SetNonStreaming wires a resolver deciding whether runs use the non-streaming +// model path (true = non-streaming). nil/unset = streaming (default). +func (p *Planner) SetNonStreaming(fn func() bool) { p.nonStreamingFn = fn } + +func (p *Planner) nonStreaming() bool { return p.nonStreamingFn != nil && p.nonStreamingFn() } + func (p *Planner) compactionWindow() int { if p.windowFn != nil { return p.windowFn() @@ -238,7 +245,7 @@ const plannerDefaultTmpl = `你是一个授权渗透测试系统的"规划者" ⚠️ 最重要的原则：**这一轮生成 0 个意图是完全正常、而且是最常见的结果。** 你不是"每次都要产出意图"的机器。绝大多数唤醒，frontier 里已有的意图已经覆盖了所有已知方向，你应当什么都不加、直接结束。重复/换措辞地生成已经存在的意图是严重错误。 决策流程（每次唤醒）： -1. **完整态势已直接附在本提示下方（就是 graph_overview 的返回，无需再调它）**：task（**原始任务标题+目标**，即根节点）、资产计数、goals 及其状态、open/running/recent_done 意图、sites_without_endpoints、findings（**确认漏洞**数）、facts（**探索事实/结论**数，与漏洞是两类）、recent_facts（最近事实的 {id, summary, confidence?}，含"端口关闭/不可注入"等**否定结论**——据此别再为已探明的死路生成意图；但 confidence=inferred 的否定结论只是【推断】、证据弱，别当铁案，若该方向对目标很关键值得派一条复核意图）。**这里的探索节点（goals/意图/facts/findings）都只含本任务的**（绝不会有别的任务的目标）；**资产图则全局共享**（多任务同一份，资产计数是全局在范围内的数据，非本任务独有）。需要更深的细节时才按需调：list_facts 列全部事实、list_findings 列全部漏洞、**node_detail(id)** 取某条的完整证据/详情（列表/recent_facts 只给摘要）。 +1. **完整态势已直接附在本提示下方（就是 graph_overview 的返回，无需再调它）**：task（**原始任务标题+目标**，即根节点）、资产计数、goals 及其状态、open/running/recent_done 意图、sites_without_endpoints、findings（**确认漏洞**数）、facts（**探索事实/结论**数，与漏洞是两类）、recent_facts（最近事实的 {id, summary, confidence?}，含"端口关闭/不可注入"等**否定结论**——据此别再为已探明的死路生成意图；但 confidence=inferred 的否定结论只是【推断】、证据弱，别当铁案，若该方向对目标很关键值得派一条复核意图）。**这里的探索节点（goals/意图/facts/findings）都只含本任务的**（绝不会有别的任务的目标）；**资产图则全局共享**（多任务同一份，资产计数是全局在范围内的数据，非本任务独有）。需要更深的细节时才按需调：list_facts 分页列事实（最新在前，默认 20 条，可传 q 关键词过滤、before 翻页，返回带 total/has_more）、list_findings 列全部漏洞、**node_detail(id)** 取某条的完整证据/详情（列表/recent_facts 只给摘要）。 - open_intents / running_intents / recent_done_intents——"哪些方向已经有意图在覆盖/已尝试"。每个意图还带 **parents（上游：它派生自哪些事实/意图）和 yields（下游：它产生了哪些事实/发现）**——这就是探索图的**血缘关系**，据此理解"哪些事实来自哪个方向、能否综合成新方向"。recent_facts 里每个事实带 **from_intent**（由哪个意图产生）。 - sites_without_endpoints / findings——"哪些方向【可能】需要探索"。 - 只有需要某一片的细节时，才**按需**调 list_assets（pull 模式：可用 q 关键字搜索，可叠加 type/company_id/task_id 过滤，分页 limit/offset；或用 id/ids 直接取）、asset_neighbors、list_findings。资产图全局共享，别默认拉全量。资产中可能包含非本次任务涉及到的资产，所以需要主要出现非本次任务相关的资产时忽略这些资产。 @@ -353,7 +360,8 @@ func (p *Planner) Plan(ctx context.Context, taskID int64, as *db.AssetStore, ts // 命中【本轮】步数预算→ SDK 跑收尾:把本轮已想清楚的结论落地(该派的 add_intent、 // 能证的 prove_goal、串行链记 TodoWrite),而非停止规划——planner 之后仍会被反复唤醒。 // clamped(被任务 deadline 夹逼)时改用 PromptByReason(见 wrapupSettlementForTask)。 - Settlement: settle, + Settlement: settle, + NonStreaming: p.nonStreaming(), // 该 profile 选非流式时走 Provider.Complete } if p.tx != nil { // persist raw LLM conversation; one accumulating file per task's planner opts.Transcript = p.tx diff --git a/agent/provider.go b/agent/provider.go index 6011ada..c66a1b8 100644 --- a/agent/provider.go +++ b/agent/provider.go @@ -51,6 +51,11 @@ type Config struct { // "" = 不发送(默认); "low"/"medium"/"high"/"xhigh"/"max" = 对应强度. // OpenAI 映射为顶层 reasoning_effort;Anthropic 映射为 output_config.effort. ReasoningEffort string + // Stream 控制该 profile 是否使用流式(SSE)接口。true(默认)= 流式;false = 真· + // 非流式(发 stream:false,一次性拿完整 JSON,走 Provider.Complete)。非流式可绕开 + // 某些网关糟糕的 SSE 实现(空帧、思考字段丢帧),代价是失去运行中的实时进度/实时 + // token 计数。映射为 agentcore.Options.NonStreaming = !Stream。 + Stream bool } // compaction window resolution bounds (in K tokens). Below the floor the @@ -116,6 +121,8 @@ func FromEnv() (Config, bool) { BaseURL: os.Getenv("ARTEX_LLM_BASE_URL"), Model: os.Getenv("ARTEX_LLM_MODEL"), Proxy: strings.TrimSpace(os.Getenv("ARTEX_LLM_PROXY")), + // 默认流式;ARTEX_LLM_STREAM=false/0/off 显式关闭走非流式。 + Stream: !isFalsy(os.Getenv("ARTEX_LLM_STREAM")), } switch prov { case "openai": @@ -147,6 +154,7 @@ func ConfigFrom(provider, model, baseURL, apiKey, proxy string) Config { BaseURL: strings.TrimRight(strings.TrimSpace(baseURL), "/"), APIKey: strings.TrimSpace(apiKey), Proxy: strings.TrimSpace(proxy), + Stream: true, // 默认流式;调用方按 profile 覆盖 } switch strings.TrimSpace(provider) { case "openai": @@ -167,6 +175,16 @@ func ConfigFrom(provider, model, baseURL, apiKey, proxy string) Config { return c } +// isFalsy reports whether an env-var string explicitly requests "off". Empty or +// unrecognized → false (so an unset var keeps the streaming default). +func isFalsy(s string) bool { + switch strings.ToLower(strings.TrimSpace(s)) { + case "0", "false", "off", "no": + return true + } + return false +} + // Provider returns the short provider name ("anthropic"/"openai"). func (c Config) Provider() string { if c.Format == llm.FormatOpenAI { @@ -288,11 +306,12 @@ func quotaAwareHTTPClient(proxy string) (*http.Client, error) { } // TestConnection makes a minimal real completion to verify the provider/model/ -// endpoint/key actually work. Returns the round-trip latency. -func TestConnection(ctx context.Context, c Config) (time.Duration, error) { +// endpoint/key actually work. Returns the round-trip latency and the model's +// reply text. +func TestConnection(ctx context.Context, c Config) (time.Duration, string, error) { prov, err := c.NewProvider() if err != nil { - return 0, err + return 0, "", err } ctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() @@ -302,12 +321,24 @@ func TestConnection(ctx context.Context, c Config) (time.Duration, error) { // 就撞到输出上限(finish=length)、被截断,连接测试虽仍算通(err=nil)但显示成 // "已中断/length/resume" 一团糟。给足预算让它把 OK 干净吐完(finish=stop)。 // EscalateMaxTokens 保持 false:不因截断而抬额重试,避免 resume 循环空烧。 - _, err = agentcore.Run(ctx, agentcore.Options{ + reply, err := agentcore.Run(ctx, agentcore.Options{ Provider: prov, SystemPrompt: []string{"你是连接测试。直接输出两个字符 OK 即可，不要思考、不要解释、不要别的。"}, PermissionMode: acperm.ModeBypass, MaxTurns: 1, MaxTokens: 8192, + NonStreaming: !c.Stream, // 用该 profile 的真实收发模式做连接测试 }, "ping") - return time.Since(start), err + lat := time.Since(start) + if err != nil { + return lat, "", err + } + // err==nil 还不够：请求通了但模型一个字都不吐的情况真实存在(思考把预算烧光、 + // 正文被安全策略吞掉、兼容层把 content 丢了)。这种配置在会话里就是"不回话", + // 测试却报成功——正是本项要消除的落差。没有可见正文一律判失败。 + reply = strings.TrimSpace(reply) + if reply == "" { + return lat, "", fmt.Errorf("模型无回复内容（请求已通，但未返回任何文本）") + } + return lat, reply, nil } diff --git a/agent/tools.go b/agent/tools.go index f8493ab..ebed6d8 100644 --- a/agent/tools.go +++ b/agent/tools.go @@ -898,19 +898,64 @@ func (t *ToolSet) listFindings() actool.CoreTool { }) } +// factsPageSize is the default page size for list_facts. Facts pile up on long +// tasks; returning all of them at once (the old behaviour) could blow up the +// context, so default to the newest page and let the agent page/filter for more. +const factsPageSize = 20 + func (t *ToolSet) listFacts() actool.CoreTool { - return readTool("list_facts", "列本任务及直接关联任务的【探索事实/结论】(紧凑：id+摘要+状态)。关联任务条目带 source_task_id/inherited=true 且只读。详情用 node_detail(id)，漏洞看 list_findings。", - obj(map[string]any{}), - func(context.Context, json.RawMessage) (actool.Result, error) { - f, _ := t.ts.ListByKindWithSources(db.KindFact, 500) + return readTool("list_facts", "分页列本任务及直接关联任务的【探索事实/结论】，最新在前(紧凑：id+摘要+状态，摘要过长会截断，全文用 node_detail(id))。参数均可选：limit(默认 20，上限 100)、before(游标，传上一页返回的 next_before 取更旧的一页；省略/0=最新一页)、q(按摘要关键词过滤)。返回 {facts, total, has_more, next_before}：total 是过滤后的总数，has_more=true 时用 next_before 继续翻页。关联任务条目带 source_task_id/inherited=true 且只读。漏洞看 list_findings。", + obj(map[string]any{ + "limit": intp("返回条数，默认 20，上限 100"), + "before": intp("分页游标：只返回 id 小于该值的更旧事实；省略或 0 = 最新一页"), + "q": str("按事实摘要关键词过滤（不区分大小写）；省略 = 不过滤"), + }), + func(_ context.Context, in json.RawMessage) (actool.Result, error) { + var a struct { + Limit int `json:"limit"` + Before int64 `json:"before"` + Q string `json:"q"` + } + _ = json.Unmarshal(in, &a) + limit := a.Limit + if limit <= 0 { + limit = factsPageSize + } + if limit > 100 { + limit = 100 + } + f, hasMore, total, err := t.ts.ListByKindPageWithSources(db.KindFact, a.Before, limit, strings.TrimSpace(a.Q)) + if err != nil { + return actool.Result{}, err + } out := make([]map[string]any, 0, len(f)) for _, n := range f { - out = append(out, compactNode(n)) + out = append(out, compactFact(n)) } - return jsonResult(out) + res := map[string]any{"facts": out, "total": total, "has_more": hasMore} + if hasMore && len(f) > 0 { + res["next_before"] = f[len(f)-1].ID // 传回它取下一页(更旧的) + } + return jsonResult(res) }) } +// factSummaryMax caps a fact summary in list_facts output. Facts carry one-line +// conclusions, but nothing enforces brevity; a runaway summary must not bloat a +// whole page. Full text stays available via node_detail(id). +const factSummaryMax = 160 + +// compactFact is compactNode with the summary rune-capped for list_facts, so a +// page of facts stays bounded regardless of how long any single summary grew. +func compactFact(n *db.Node) map[string]any { + m := compactNode(n) + if s, ok := m["summary"].(string); ok && len([]rune(s)) > factSummaryMax { + m["summary"] = string([]rune(s)[:factSummaryMax]) + "…" + m["summary_truncated"] = true + } + return m +} + func (t *ToolSet) nodeDetail() actool.CoreTool { return readTool("node_detail", "按 id 取本任务或直接关联任务的【探索图节点】完整内容。继承节点带 source_task_id/inherited=true 且只读。仅限 list_facts/list_findings/graph_overview 返回的探索节点 id；资产请用 list_assets/asset_neighbors。", obj(map[string]any{"id": idp("探索图节点 id(非资产 id)")}, "id"), diff --git a/agent/worker.go b/agent/worker.go index f2e5ce8..b7a9e7a 100644 --- a/agent/worker.go +++ b/agent/worker.go @@ -62,8 +62,19 @@ type Worker struct { // injected into the worker system prompt. Read per run so the settings toggle // takes effect without rebuilding the agent. nil = inject (default). injectConstraints func() bool + // nonStreamingFn resolves whether this run uses the non-streaming (Complete) + // path. Read per run so a profile/task toggle takes effect without rebuilding + // the agent. nil = streaming (default). + nonStreamingFn func() bool } +// SetNonStreaming wires a resolver deciding whether runs use the non-streaming +// model path (true = non-streaming). nil/unset = streaming (default). Read per +// run so a profile or task-chain toggle takes effect without rebuilding. +func (w *Worker) SetNonStreaming(fn func() bool) { w.nonStreamingFn = fn } + +func (w *Worker) nonStreaming() bool { return w.nonStreamingFn != nil && w.nonStreamingFn() } + // SetConstraintInject wires a resolver deciding whether this task's operation // constraints get injected into the worker system prompt. nil = inject (default). func (w *Worker) SetConstraintInject(fn func() bool) { w.injectConstraints = fn } @@ -160,7 +171,7 @@ const workerDefaultTmpl = `你是一个授权渗透测试系统的"执行者"(wo - insert_assets：登记新资产（资产图）。**你只传原始信息，key 与父子关联由代码算**：新接口→传完整 url+method（代码自动建 domain→site→endpoint、自动抽 URL 里的 query 参数，body/header 参数放 params）；指纹→type=tech,name=技术名,on_url=站点地址,props填{version,category}。多个资产放 assets 数组一次批量登记。属性写在资产自己的 props 上，探索结论不要写这里。 - record_fact：把探索【事实/结论】写入探索图并连到意图（传 intent_id）。正向/否定结论、观察、判断用它；**一次探索的多个观察汇总成一条事实**（summary 总结一句话 + detail 放细节），不要一个属性一条。真有多条不同结论才用 facts 数组。**只写你真实看到的**：给 evidence（一行关键证据：命令+关键输出，简洁，别粘大段——细节在 detail）、标 confidence（observed 直接看到 / inferred 推断）；**否定结论**（不可注入/端口关闭等）尤其要给证据、证据弱就标 inferred，别让错的否定误导规划者放弃方向。 - report_finding：确认漏洞 → 记录(含 PoC，传 intent_id=你领到的意图id)。**只有你在本次运行里真实触发过该漏洞、拿到了可复现的证据（请求/响应或命令输出）才用它。** 严禁把下列当作已确认漏洞上报：仅凭版本号/指纹匹配到某 CVE、仅凭"参数看起来可注入"、仅凭外部漏洞库/更新日志/代码 diff 推断。**不要用查 CVE 库或"对比补丁版本"替代实际触发。** 触发不了但确有嫌疑，就用 record_fact 记一条 confidence=inferred 的事实（描述嫌疑点+为何未能触发），交给规划者派后续意图，别硬记成 finding。 -- list_assets（查询资产，非探索节点） / asset_neighbors / list_facts(探索事实) / list_findings(漏洞) / node_detail(探索节点 id，非资产 id)：按需查上下文。 +- list_assets（查询资产，非探索节点） / asset_neighbors / list_facts(探索事实，分页最新在前；默认 20 条，可传 q 关键词过滤、before 翻页) / list_findings(漏洞) / node_detail(探索节点 id，非资产 id)：按需查上下文。 只在授权范围内操作；若系统提示顶部附有【操作约束】，那是最高优先级红线——任何命令/探测在执行前先自检是否违反，违反即不做（哪怕它落在你领到的意图里）。完成本意图后用一句话总结你做了什么、写回了哪些事实。务实、克制、聚焦这一条意图。` @@ -337,6 +348,7 @@ func (w *Worker) Execute(ctx context.Context, name string, taskID int64, as *db. ToolOutputDir: cmdOutDir(runDir), Compaction: compactionConfig(w.compactionWindow()), // long tool-heavy runs stay within the window Todos: actool.NewTodoStore(), // 会话级临时待办（TodoWrite），纯规划用，退出即丢 + NonStreaming: w.nonStreaming(), // 该 profile 选非流式时走 Provider.Complete } if hooks != nil { // typed-nil guard: only set when concrete (avoids harness panic) opts.Hooks = hooks diff --git a/db/commands.go b/db/commands.go index 27172da..22fe14e 100644 --- a/db/commands.go +++ b/db/commands.go @@ -19,18 +19,10 @@ type CommandRecord struct { CreatedAt time.Time `json:"created_at"` } -// ListCommands returns tool executions (tool_use + paired tool_result) across all -// explorations, with optional filtering and pagination. Covers every tool, not -// just Bash; q matches the tool name or its input. -func (d *DB) ListCommands(expID *int64, q string, page, size int) ([]CommandRecord, int, error) { - if size <= 0 { - size = 50 - } - if page < 0 { - page = 0 - } - offset := page * size - +// commandFilter builds the WHERE clause shared by the tool-execution list and +// its per-tool tally, so the summary always describes exactly the rows the table +// pages through. Returns the clause, its args, and the next placeholder index. +func commandFilter(expID *int64, q string) (string, []any, int) { where := `WHERE u.kind = 'tool_use'` args := []any{} argN := 1 @@ -45,6 +37,59 @@ func (d *DB) ListCommands(expID *int64, q string, page, size int) ([]CommandReco args = append(args, "%"+q+"%") argN++ } + return where, args, argN +} + +// ToolStat is one tool's execution tally for the usage summary. +type ToolStat struct { + Tool string `json:"tool"` + Total int `json:"total"` + Errors int `json:"errors"` +} + +// ToolStats counts executions grouped by tool under the same filters +// ListCommands takes. Unpaginated on purpose: the tally describes the whole +// filtered set, not the page currently on screen. +func (d *DB) ToolStats(expID *int64, q string) ([]ToolStat, error) { + where, args, _ := commandFilter(expID, q) + + rows, err := d.Query(` +SELECT COALESCE(NULLIF(u.tool,''),'-') AS tool, COUNT(*) AS total, + COUNT(*) FILTER (WHERE COALESCE(r.is_error,false)) AS errors +FROM activity u +LEFT JOIN activity r ON r.tool_use_id = u.tool_use_id AND r.kind = 'tool_result' +`+where+` +GROUP BY 1 +ORDER BY total DESC, tool ASC`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + out := []ToolStat{} + for rows.Next() { + var s ToolStat + if err := rows.Scan(&s.Tool, &s.Total, &s.Errors); err != nil { + return nil, err + } + out = append(out, s) + } + return out, rows.Err() +} + +// ListCommands returns tool executions (tool_use + paired tool_result) across all +// explorations, with optional filtering and pagination. Covers every tool, not +// just Bash; q matches the tool name or its input. +func (d *DB) ListCommands(expID *int64, q string, page, size int) ([]CommandRecord, int, error) { + if size <= 0 { + size = 50 + } + if page < 0 { + page = 0 + } + offset := page * size + + where, args, argN := commandFilter(expID, q) // count var total int diff --git a/db/config.go b/db/config.go index 40684d6..410994a 100644 --- a/db/config.go +++ b/db/config.go @@ -39,18 +39,23 @@ type LLMProfile struct { // usable when an agent/task binds it explicitly, it just never gets picked up // as a fallback target. PoolExclude bool `json:"pool_exclude"` + // Streaming selects the wire protocol: true (default) = streaming (SSE); + // false = real non-streaming (stream:false, single JSON response via + // Provider.Complete). Non-streaming sidesteps flaky gateway SSE at the cost of + // live in-run progress. Maps to agent.Config.Stream. + Streaming bool `json:"streaming"` } // profileCols is the read column list (hint variant, no api key) shared by the // list query; profileColsKey is the same with api_key for the single-row loads. -const profileCols = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key_hint,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude,COALESCE(thinking_type,'')` -const profileColsKey = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude,COALESCE(thinking_type,'')` +const profileCols = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key_hint,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude,COALESCE(thinking_type,''),COALESCE(streaming,true)` +const profileColsKey = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude,COALESCE(thinking_type,''),COALESCE(streaming,true)` // scanProfile reads one row in the profileCols / profileColsKey column order. The // 7th column lands in APIKeyHint or APIKey depending on which list the caller used. func scanProfile(sc interface{ Scan(...any) error }, into *string, p *LLMProfile) error { return sc.Scan(&p.ID, &p.Name, &p.Format, &p.BaseURL, &p.Proxy, &p.Model, into, - &p.RatePerSecond, &p.RatePerMinute, &p.ContextWindowK, &p.ReasoningEffort, &p.IsDefault, &p.Priority, &p.PoolExclude, &p.ThinkingType) + &p.RatePerSecond, &p.RatePerMinute, &p.ContextWindowK, &p.ReasoningEffort, &p.IsDefault, &p.Priority, &p.PoolExclude, &p.ThinkingType, &p.Streaming) } func (d *DB) ListProfiles() ([]*LLMProfile, error) { @@ -123,18 +128,18 @@ func (d *DB) SaveProfile(p *LLMProfile) (int64, error) { } if p.ID == 0 { var id int64 - err := d.QueryRow(`INSERT INTO llm_profiles(name,format,base_url,proxy,model,api_key,api_key_hint,rate_per_second,rate_per_minute,context_window_k,reasoning_effort,priority,pool_exclude,thinking_type) -VALUES ($1,$2,NULLIF($3,''),NULLIF($4,''),$5,NULLIF($6,''),NULLIF($7,''),$8,$9,$10,$11,$12,$13,$14) RETURNING id`, - p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType).Scan(&id) + err := d.QueryRow(`INSERT INTO llm_profiles(name,format,base_url,proxy,model,api_key,api_key_hint,rate_per_second,rate_per_minute,context_window_k,reasoning_effort,priority,pool_exclude,thinking_type,streaming) +VALUES ($1,$2,NULLIF($3,''),NULLIF($4,''),$5,NULLIF($6,''),NULLIF($7,''),$8,$9,$10,$11,$12,$13,$14,$15) RETURNING id`, + p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.Streaming).Scan(&id) return id, err } if p.APIKey == "" { - _, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,rate_per_second=$6,rate_per_minute=$7,context_window_k=$8,reasoning_effort=$9,priority=$10,pool_exclude=$11,thinking_type=$12 WHERE id=$13`, - p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.ID) + _, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,rate_per_second=$6,rate_per_minute=$7,context_window_k=$8,reasoning_effort=$9,priority=$10,pool_exclude=$11,thinking_type=$12,streaming=$13 WHERE id=$14`, + p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.Streaming, p.ID) return p.ID, err } - _, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,api_key=$6,api_key_hint=$7,rate_per_second=$8,rate_per_minute=$9,context_window_k=$10,reasoning_effort=$11,priority=$12,pool_exclude=$13,thinking_type=$14 WHERE id=$15`, - p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.ID) + _, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,api_key=$6,api_key_hint=$7,rate_per_second=$8,rate_per_minute=$9,context_window_k=$10,reasoning_effort=$11,priority=$12,pool_exclude=$13,thinking_type=$14,streaming=$15 WHERE id=$16`, + p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.Streaming, p.ID) return p.ID, err } diff --git a/db/exploration.go b/db/exploration.go index f62e280..eb0e677 100644 --- a/db/exploration.go +++ b/db/exploration.go @@ -672,6 +672,40 @@ ORDER BY id DESC LIMIT $4`, s.expID, kind, before, limit+1) return nodes, hasMore, nil } +// listByKindPageFiltered is ListByKindPage with an optional summary keyword +// filter (payload->>'summary' ILIKE %q%). Fetches one extra row so the caller can +// probe hasMore. Empty q = no filter. Newest-first by id. +func (s *ExplorationStore) listByKindPageFiltered(kind string, before int64, limit int, q string) ([]*Node, error) { + where := `exploration_id=$1 AND kind=$2 AND ($3 <= 0 OR id < $3)` + args := []any{s.expID, kind, before} + if q != "" { + args = append(args, "%"+q+"%") + where += fmt.Sprintf(` AND payload->>'summary' ILIKE $%d`, len(args)) + } + args = append(args, limit+1) + rows, err := s.db.Query(`SELECT `+nodeCols+` FROM exploration_nodes +WHERE `+where+` ORDER BY id DESC LIMIT $`+fmt.Sprintf("%d", len(args)), args...) + if err != nil { + return nil, err + } + defer rows.Close() + return scanNodes(rows) +} + +// countByKindFiltered counts nodes of a kind under the same summary filter, for a +// page's total. Empty q = count all of that kind. +func (s *ExplorationStore) countByKindFiltered(kind, q string) (int, error) { + where := `exploration_id=$1 AND kind=$2` + args := []any{s.expID, kind} + if q != "" { + args = append(args, "%"+q+"%") + where += fmt.Sprintf(` AND payload->>'summary' ILIKE $%d`, len(args)) + } + var n int + err := s.db.QueryRow(`SELECT COUNT(*) FROM exploration_nodes WHERE `+where, args...).Scan(&n) + return n, err +} + // GetNode returns one node of this exploration by id (nil, nil if not found). func (s *ExplorationStore) GetNode(id int64) (*Node, error) { n, err := scanNode(s.db.QueryRow(`SELECT `+nodeCols+` FROM exploration_nodes WHERE id=$1 AND exploration_id=$2`, id, s.expID)) diff --git a/db/exploration_sources.go b/db/exploration_sources.go index 3ccbb96..c12d5ba 100644 --- a/db/exploration_sources.go +++ b/db/exploration_sources.go @@ -108,6 +108,53 @@ func (s *ExplorationStore) ListByKindWithSources(kind string, limit int) ([]*Nod return out, nil } +// ListByKindPageWithSources is the paginated, keyword-filterable sibling of +// ListByKindWithSources: it returns one newest-first page (id < before, before<=0 +// = newest) spanning this exploration and its direct sources, plus hasMore and the +// filtered total across all of them. Node ids are globally unique, so merging each +// store's own page and re-sorting by id DESC yields the true global page; fetching +// limit+1 per store guarantees the merged top-`limit` is complete. +func (s *ExplorationStore) ListByKindPageWithSources(kind string, before int64, limit int, q string) (nodes []*Node, hasMore bool, total int, err error) { + if limit <= 0 { + limit = 20 + } + own, err := s.listByKindPageFiltered(kind, before, limit, q) + if err != nil { + return nil, false, 0, err + } + merged := own + total, err = s.countByKindFiltered(kind, q) + if err != nil { + return nil, false, 0, err + } + + sources, err := s.DirectSourceStores() + if err != nil { + return nil, false, 0, err + } + for _, source := range sources { + page, err := source.Store.listByKindPageFiltered(kind, before, limit, q) + if err != nil { + return nil, false, 0, err + } + for _, n := range page { + merged = append(merged, markInheritedNode(n, source.Task.TaskID)) + } + cnt, err := source.Store.countByKindFiltered(kind, q) + if err != nil { + return nil, false, 0, err + } + total += cnt + } + + sort.Slice(merged, func(i, j int) bool { return merged[i].ID > merged[j].ID }) + hasMore = len(merged) > limit + if hasMore { + merged = merged[:limit] + } + return merged, hasMore, total, nil +} + // GetNodeWithSources reads a node only when it belongs to this exploration or // one of its direct sources. Inherited nodes are tagged so tool callers can keep // them read-only and show their provenance. diff --git a/db/schema.sql b/db/schema.sql index de0b7cd..9be7064 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -310,6 +310,8 @@ CREATE TABLE IF NOT EXISTS llm_profiles ( -- pool_exclude true=不作为故障转移目标(仍可被 agent/任务显式绑定使用)。 priority INTEGER NOT NULL DEFAULT 0, pool_exclude BOOLEAN NOT NULL DEFAULT false, + -- streaming=true(默认)走流式 SSE；false 走真·非流式(stream:false，一次性 JSON)。 + streaming BOOLEAN NOT NULL DEFAULT true, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); @@ -320,6 +322,8 @@ CREATE TRIGGER trg_llm_upd BEFORE UPDATE ON llm_profiles -- 轮询顺位/排除标记；补旧库。默认 0 / false = 全部配置都参与轮询。 ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS priority INTEGER NOT NULL DEFAULT 0; ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS pool_exclude BOOLEAN NOT NULL DEFAULT false; +-- 流式开关；补旧库。默认 true = 保持既有的流式行为，旧配置无感升级。 +ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS streaming BOOLEAN NOT NULL DEFAULT true; -- 思考开关字段 thinking_type，从旧的单一 reasoning_effort 语义一次性拆分而来。 -- schema.sql 每次启动都执行，故迁移必须只跑一次：仅当该列尚不存在时才回填， diff --git a/go.mod b/go.mod index d56340a..b047c4b 100644 --- a/go.mod +++ b/go.mod @@ -3,9 +3,10 @@ module github.com/Autumn-27/artex go 1.26.3 require ( - github.com/Autumn-27/norma v0.2.9 + github.com/Autumn-27/norma v0.3.2 github.com/golang-jwt/jwt/v5 v5.3.1 github.com/jackc/pgx/v5 v5.10.0 + github.com/klauspost/compress v1.18.6 github.com/lqqyt2423/go-mitmproxy v1.9.2 github.com/miekg/dns v1.1.72 github.com/projectdiscovery/dnsx v1.2.3 @@ -29,7 +30,6 @@ require ( github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect - github.com/klauspost/compress v1.18.6 // indirect github.com/kr/pretty v0.3.1 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/microcosm-cc/bluemonday v1.0.27 // indirect diff --git a/go.sum b/go.sum index 835017c..845ee26 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -github.com/Autumn-27/norma v0.2.9 h1:VTt8Oe7XoiV85Dc+QOPu+zQ7IykQRm2aDTEbLEzbBe4= -github.com/Autumn-27/norma v0.2.9/go.mod h1:BudWrf7iHl96FgHqFr/HwN3mKYmGjJDt+xL2ZHXfFZw= +github.com/Autumn-27/norma v0.3.2 h1:3gW84SbSrz8luW8kbi2tv3O7CLdRMQvyylCHj1QUdNk= +github.com/Autumn-27/norma v0.3.2/go.mod h1:BudWrf7iHl96FgHqFr/HwN3mKYmGjJDt+xL2ZHXfFZw= github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro= github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 h1:DklsrG3dyBCFEj5IhUbnKptjxatkF07cF2ak3yi77so= diff --git a/llmpool/pool.go b/llmpool/pool.go index 30b3665..be0ddff 100644 --- a/llmpool/pool.go +++ b/llmpool/pool.go @@ -122,6 +122,44 @@ func (p *Pool) Stream(ctx context.Context, req llm.CompletionRequest) iter.Seq2[ } } +// Complete implements llm.Provider with failover for non-streaming calls. A +// non-streaming request is atomic — it never delivers partial output — so every +// failover-eligible failure lands in the safe window and the next member can be +// tried without risk of duplicated output. Mirrors Stream's health-tripping and +// chain-exhaustion behavior. +func (p *Pool) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) { + order := p.order(req) + var lastErr error + for i, m := range order { + msg, sr, usage, err := m.Prov.Complete(ctx, req) + if err == nil { + p.health.Pass(m.ID) + return msg, sr, usage, nil + } + if !shouldFailover(ctx, err) { + // Non-failover error (e.g. ctx cancel, deterministic 4xx): surface as-is + // without tripping health, matching Stream's non-failover path. + p.health.Pass(m.ID) + return llm.Message{}, "", llm.Usage{}, err + } + lastErr = err + hard := isHardFailure(err) + if p.health.Trip(m.ID, trimErr(err), hard) { + log.Printf("[llmpool] 配置 %q(%s) 已熔断：%s", m.Name, m.Model, trimErr(err)) + } + if i+1 < len(order) { + n := order[i+1] + log.Printf("[llmpool] LLM 故障转移：%q(%s) → %q(%s)，原因：%s", + m.Name, m.Model, n.Name, n.Model, trimErr(err)) + } + } + if lastErr == nil { + lastErr = ErrExhausted + } + log.Printf("[llmpool] 轮询链已耗尽(%d 个配置全部失败)，最后错误：%s", len(order), trimErr(lastErr)) + return llm.Message{}, "", llm.Usage{}, fmt.Errorf("%w：%v", ErrExhausted, lastErr) +} + // order picks the members to try, in order: skip those in a cooling-off window // and those whose context window can't hold this request, then rotate within each // equal-priority group so same-priority profiles share the load. Never returns an diff --git a/llmpool/pool_test.go b/llmpool/pool_test.go index b1e69c8..21b7568 100644 --- a/llmpool/pool_test.go +++ b/llmpool/pool_test.go @@ -55,6 +55,17 @@ func (f *fakeProv) Stream(ctx context.Context, req llm.CompletionRequest) iter.S } } +func (f *fakeProv) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) { + acc := llm.NewAccumulator() + for ev, err := range f.Stream(ctx, req) { + if err != nil { + return llm.Message{}, "", llm.Usage{}, err + } + acc.Add(ev) + } + return acc.Message(), acc.StopReason, acc.Usage, nil +} + // ok builds a provider that always succeeds with one text delta. func okProv(name, text string) *fakeProv { return &fakeProv{name: name, events: [][]llm.StreamEvent{{{Type: llm.SETextDelta, Text: text}}}} diff --git a/llmrec/llmrec.go b/llmrec/llmrec.go index 1bbd55c..71faf98 100644 --- a/llmrec/llmrec.go +++ b/llmrec/llmrec.go @@ -39,10 +39,10 @@ func TaskIDFrom(ctx context.Context) string { // Recorder wraps an llm.Provider and records every completion call. type Recorder struct { - inner llm.Provider - pg *db.DB - model string // model name (from config, not in CompletionRequest) - prof string // LLM profile name (from llm_profiles) + inner llm.Provider + pg *db.DB + model string // model name (from config, not in CompletionRequest) + prof string // LLM profile name (from llm_profiles) // thinkingType / reasoningEffort 是配置级思考参数(思考开关 / 思考强度)。它们在 // norma 的 buildBody() 里从 provider 配置注入真正的 HTTP body,不出现在 // CompletionRequest 上,故 Recorder 需在此单独带一份,序列化时写进录制。 @@ -179,6 +179,45 @@ func (r *Recorder) Stream(ctx context.Context, req llm.CompletionRequest) iter.S } } +// Complete implements the atomic completion path while preserving the same +// usage metering and optional body recording semantics as Stream. +func (r *Recorder) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) { + recordBodies := r.enabled == nil || r.enabled() + start := time.Now() + session := transcript.SessionIDFrom(ctx) + parsedID, worker := parseSession(session) + expID := db.ParseExpID(parsedID) + taskID := TaskIDFrom(ctx) + if taskID == "" { + taskID = parsedID + } + + reqBody := "" + if recordBodies { + reqBody = r.serializeRequest(req) + } + + msg, stopReason, usage, err := r.inner.Complete(ctx, req) + status := "ok" + if err != nil { + status = "error" + } + r.recordUsage(taskID, expID, worker, usage, int(time.Since(start).Milliseconds()), status) + if recordBodies { + var textBuf, thinkingBuf strings.Builder + for _, block := range msg.Content { + switch block.Type { + case llm.BlockText: + textBuf.WriteString(block.Text) + case llm.BlockThinking: + thinkingBuf.WriteString(block.Thinking) + } + } + r.record(req, session, taskID, worker, reqBody, start, textBuf.String(), thinkingBuf.String(), usage, stopReason, err) + } + return msg, stopReason, usage, err +} + // recordUsage appends one lightweight metering row to llm_usage (no bodies). Skips // zero-token calls with no model, which carry nothing worth metering. func (r *Recorder) recordUsage(taskID string, expID int64, worker string, usage llm.Usage, latencyMs int, status string) { diff --git a/llmrec/llmrec_test.go b/llmrec/llmrec_test.go index e998684..19c7d4b 100644 --- a/llmrec/llmrec_test.go +++ b/llmrec/llmrec_test.go @@ -2,9 +2,28 @@ package llmrec import ( "context" + "iter" "testing" + + "github.com/Autumn-27/norma/llm" ) +type completeProvider struct{} + +func (completeProvider) Stream(context.Context, llm.CompletionRequest) iter.Seq2[llm.StreamEvent, error] { + return func(func(llm.StreamEvent, error) bool) {} +} + +func (completeProvider) Complete(context.Context, llm.CompletionRequest) (llm.Message, string, llm.Usage, error) { + return llm.Message{ + Role: llm.RoleAssistant, + Content: []llm.ContentBlock{ + {Type: llm.BlockThinking, Thinking: "reasoning"}, + llm.TextBlock("answer"), + }, + }, "stop", llm.Usage{InputTokens: 7, OutputTokens: 3}, nil +} + func TestTaskIDContextUsesExplicitRegistryID(t *testing.T) { ctx := WithTaskID(context.Background(), " 42 ") if got := TaskIDFrom(ctx); got != "42" { @@ -27,3 +46,20 @@ func TestParseSessionFallbackIsExplorationScoped(t *testing.T) { t.Fatalf("unexpected non-session parse: (%q,%q)", taskID, worker) } } + +func TestCompleteForwardsAtomicResponse(t *testing.T) { + recorder := Wrap(completeProvider{}, nil, "model", "profile", "", "", func() bool { return false }) + msg, stopReason, usage, err := recorder.Complete(context.Background(), llm.CompletionRequest{}) + if err != nil { + t.Fatalf("Complete() error: %v", err) + } + if got := msg.Text(); got != "answer" { + t.Fatalf("Complete() text=%q want answer", got) + } + if stopReason != "stop" { + t.Fatalf("Complete() stop reason=%q want stop", stopReason) + } + if usage.InputTokens != 7 || usage.OutputTokens != 3 { + t.Fatalf("Complete() usage=%+v", usage) + } +} diff --git a/server/commands.go b/server/commands.go index 12904a7..aa05427 100644 --- a/server/commands.go +++ b/server/commands.go @@ -15,14 +15,7 @@ func (s *Server) pgListCommands(w http.ResponseWriter, r *http.Request) { size := atoiDefault(q.Get("size"), 50) keyword := q.Get("q") - var expID *int64 - if tv := q.Get("task"); tv != "" { - if n, err := strconv.ParseInt(tv, 10, 64); err == nil { - expID = &n - } - } - - records, total, err := s.m.PG().ListCommands(expID, keyword, page, size) + records, total, err := s.m.PG().ListCommands(commandTaskFilter(q.Get("task")), keyword, page, size) if err != nil { writeErr(w, 500, err.Error()) return @@ -30,6 +23,37 @@ func (s *Server) pgListCommands(w http.ResponseWriter, r *http.Request) { writeJSON(w, 200, map[string]any{"commands": records, "total": total}) } +// pgToolStats returns per-tool call counts for the tool-execution history, under +// the same task/keyword filters the list takes — the summary describes the whole +// filtered set, not the page on screen. +func (s *Server) pgToolStats(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + pg := s.m.PG() + if pg == nil { + writeJSON(w, 200, map[string]any{"stats": []db.ToolStat{}}) + return + } + stats, err := pg.ToolStats(commandTaskFilter(q.Get("task")), q.Get("q")) + if err != nil { + writeErr(w, 500, err.Error()) + return + } + writeJSON(w, 200, map[string]any{"stats": stats}) +} + +// commandTaskFilter parses the ?task= exploration id. Absent or unparsable → nil +// (no filter), matching the list endpoint's long-standing lenient behaviour. +func commandTaskFilter(v string) *int64 { + if v == "" { + return nil + } + n, err := strconv.ParseInt(v, 10, 64) + if err != nil { + return nil + } + return &n +} + // pgListLLMRecords returns paginated LLM call records. func (s *Server) pgListLLMRecords(w http.ResponseWriter, r *http.Request) { q := r.URL.Query() diff --git a/server/dto.go b/server/dto.go index cf7133d..c681bcf 100644 --- a/server/dto.go +++ b/server/dto.go @@ -580,6 +580,9 @@ type LLMProfileDTO struct { // pool_exclude=true 则不作为故障转移目标，但仍可被 agent/任务显式绑定。 Priority int `json:"priority"` PoolExclude bool `json:"pool_exclude"` + // 收发模式：true=流式(SSE) | false=非流式。没有 omitempty —— false 必须出现在 + // 响应里，否则前端读不到「非流式」，开关会回落成默认的流式。 + Streaming bool `json:"streaming"` } func llmProfileDTO(p *db.LLMProfile) LLMProfileDTO { @@ -599,6 +602,7 @@ func llmProfileDTO(p *db.LLMProfile) LLMProfileDTO { IsDefault: p.IsDefault, Priority: p.Priority, PoolExclude: p.PoolExclude, + Streaming: p.Streaming, } } diff --git a/server/goals.go b/server/goals.go index dbc497d..c8e6d98 100644 --- a/server/goals.go +++ b/server/goals.go @@ -383,7 +383,8 @@ func (s *Server) createGoals(ctx context.Context, t *Task, emit func(db.Activity } taskID, _ := strconv.ParseInt(t.ID, 10, 64) s.engine.BeginLLMCall(t.ID) - decomposed := agent.DecomposeGoalsWithProvider(ctx, s.agentsForTask(t).runtime, s.m.dir, t.Goal, t.Description, as, t.Store, taskID, emit) + goalRuntime := s.agentsForTask(t).runtime + decomposed := agent.DecomposeGoalsWithProvider(ctx, goalRuntime, s.m.dir, t.Goal, t.Description, as, t.Store, taskID, goalRuntime.nonStreaming(), emit) s.engine.EndLLMCall(t.ID) for _, g := range decomposed { if strings.TrimSpace(g.Text) != "" { diff --git a/server/orchestration.go b/server/orchestration.go index 9057ea7..a48d83b 100644 --- a/server/orchestration.go +++ b/server/orchestration.go @@ -450,9 +450,9 @@ func (s *Server) seedOrchestrationTools() { s.seedPlannerDefaultBindings() s.seedAutoReportFindingBinding() s.unbindGoalMetDefault() - s.reseedGoalsPrompt() // goals 提示词加入「抽操作约束」步 → 旧库追加一版新默认(一次性) + s.reseedGoalsPrompt() // goals 提示词加入「抽操作约束」步 → 旧库追加一版新默认(一次性) s.reseedMainAgentPrompt() // mainagent 提示词加入「目标达成后 add_intent 反问是否建目标」(一次性) - s.seedReporterAgent() // 预置「报告撰写」agent + 工具绑定 + finding 触发器(一次性) + s.seedReporterAgent() // 预置「报告撰写」agent + 工具绑定 + finding 触发器(一次性) // 注：pentest 的默认工具绑定无需迁移——BuiltinToolSeeds 在全新初始化时就把 // list_assets/insert_assets/report_finding/list_findings/list_companies 连同 // pentest 一起 seed 好了（项目尚无旧库，不做迁移）。 @@ -464,7 +464,7 @@ func (s *Server) seedOrchestrationTools() { // reaches an old DB otherwise. Preserves each tool's agent binding + enabled flag. // Bump the flag whenever these tools' schemas/descriptions change in code. func (s *Server) refreshBuiltinToolSchemas() { - const flag = "tool_schema_refresh_v6_insert_assets_related" + const flag = "tool_schema_refresh_v7_list_facts_paging" if v, _, _ := s.m.pg.GetSetting(flag); v == "true" { return } @@ -480,7 +480,9 @@ func (s *Server) refreshBuiltinToolSchemas() { // “结束空轮”的手段、刚开跑就误判整个任务完成。 // - insert_assets：新增 related 入参(标记资产是否与当前任务相关、决定是否入覆盖度)， // SeedTool 首插入only，旧库已 seed 的 schema 否则收不到这个新参数。 - refreshBuiltin := map[string]bool{"goal_met": true, "insert_assets": true} + // - list_facts：改为分页，新增 limit/before/q 入参；旧库已 seed 的空 schema 否则 + // 在工具管理页显示「无参数」，模型也拿不到这几个参数说明。 + refreshBuiltin := map[string]bool{"goal_met": true, "insert_assets": true, "list_facts": true} for _, sd := range agent.BuiltinToolSeeds() { if !refreshBuiltin[sd.Key] { continue diff --git a/server/server.go b/server/server.go index 942c8ca..4968091 100644 --- a/server/server.go +++ b/server/server.go @@ -311,6 +311,7 @@ func (s *Server) loadLLMConfig() (agent.Config, bool) { cfg.ContextWindowK = p.ContextWindowK cfg.ThinkingType = p.ThinkingType cfg.ReasoningEffort = p.ReasoningEffort + cfg.Stream = p.Streaming if cfg.APIKey == "" { return cfg, false } @@ -328,14 +329,15 @@ func (s *Server) saveLLMConfig(cfg agent.Config) error { } var id int64 // agent.Config carries no failover fields, so carry the stored ones forward — - // otherwise this legacy endpoint would silently reset the profile's priority - // and pool_exclude to zero values on every save. + // otherwise this legacy endpoint would silently reset the profile's priority, + // pool_exclude, and streaming to zero values on every save. var priority int var poolExclude bool + streaming := true // 旧库/新建默认流式 if profs, _ := s.m.pg.ListProfiles(); profs != nil { for _, p := range profs { if p.Name == "default" { - id, priority, poolExclude = p.ID, p.Priority, p.PoolExclude + id, priority, poolExclude, streaming = p.ID, p.Priority, p.PoolExclude, p.Streaming break } } @@ -344,7 +346,7 @@ func (s *Server) saveLLMConfig(cfg agent.Config) error { ID: id, Name: "default", Format: format, Model: cfg.Model, BaseURL: cfg.BaseURL, Proxy: cfg.Proxy, APIKey: cfg.APIKey, RatePerSecond: cfg.RatePerSecond, RatePerMinute: cfg.RatePerMinute, ContextWindowK: cfg.ContextWindowK, ThinkingType: cfg.ThinkingType, ReasoningEffort: cfg.ReasoningEffort, IsDefault: true, - Priority: priority, PoolExclude: poolExclude, + Priority: priority, PoolExclude: poolExclude, Streaming: streaming, }) if err != nil { return err @@ -394,6 +396,7 @@ func (s *Server) buildPlannerWorker(pinID *int64, gProv llm.Provider, gCfg agent wk.SetMemory(memory.NewStore(filepath.Join(s.m.dir, "memory"))) wk.SetWebSearch(s.webSearchFor("worker")) wk.SetConstraintInject(s.constraintInjectWorker) // 操作约束注入 worker(可配置,默认开;每轮读) + wk.SetNonStreaming(nonStreamingResolver(wCfg)) // 该 profile 选非流式时走 Provider.Complete pProv, pCfg := s.providerForAgent("planner", pinID, gProv, gCfg) pl := agent.NewPlanner(pProv, pCfg.Model, s.m.dir, tx, pCfg.CompactionWindow(), s.agentMaxTurns("planner")) pl.SetKillWork(s.engine.KillWork) // planner kill_work → terminate a running work @@ -401,9 +404,18 @@ func (s *Server) buildPlannerWorker(pinID *int64, gProv llm.Provider, gCfg agent pl.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) // WebFetch through the recording proxy pl.SetWebSearch(s.webSearchFor("planner")) pl.SetConstraintInject(s.constraintInjectPlanner) // 操作约束注入 planner(可配置,默认开;每轮读) + pl.SetNonStreaming(nonStreamingResolver(pCfg)) // 该 profile 选非流式时走 Provider.Complete return pl, wk } +// nonStreamingResolver returns a resolver capturing a profile's streaming choice. +// The agents read it per run; a profile change rebuilds the agents (applyLLM), +// so the captured value is always the one in effect for this build. +func nonStreamingResolver(cfg agent.Config) func() bool { + nonStreaming := !cfg.Stream + return func() bool { return nonStreaming } +} + // applyLLM (re)builds the planner/worker/main-agent from cfg and installs them on // the running engine as the GLOBAL active pair. Safe to call at runtime (UI configures LLM). func (s *Server) applyLLM(cfg agent.Config) error { @@ -447,12 +459,14 @@ func (s *Server) applyLLM(cfg agent.Config) error { s.mainAgent.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) // WebFetch through the recording proxy s.mainAgent.SetWebSearch(s.webSearchFor("mainagent")) s.mainAgent.SetSteerWork(s.engine.SteerWork) // steer_work：人对运行中 work 实时纠偏 + s.mainAgent.SetNonStreaming(nonStreamingResolver(mCfg)) // chat agent serves MANY custom agents by key → it holds the GLOBAL opts // (backend/key) and gates Enabled per-conversation-agent at Chat time. 对话始终用激活配置。 s.chatAgent = agent.NewChatAgent(prov, cfg.Model, s.m.dir, tx, win) // chat page runner s.chatAgent.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) s.chatAgent.SetWebSearch(s.m.WebSearchOpts()) s.chatAgent.SetGuard(s.chatGuard()) + s.chatAgent.SetNonStreaming(nonStreamingResolver(cfg)) s.llmProv = prov s.llmCfg = cfg s.llmOn = true @@ -478,6 +492,7 @@ func (s *Server) loadProfileConfig(id int64) (agent.Config, bool) { cfg.ContextWindowK = p.ContextWindowK cfg.ThinkingType = p.ThinkingType cfg.ReasoningEffort = p.ReasoningEffort + cfg.Stream = p.Streaming if cfg.APIKey == "" { return cfg, false } @@ -629,6 +644,7 @@ func (s *Server) chatAgentForProfile(id int64) *agent.ChatAgent { ca.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) ca.SetWebSearch(s.m.WebSearchOpts()) ca.SetGuard(s.chatGuard()) + ca.SetNonStreaming(nonStreamingResolver(cfg)) s.profMu.Lock() if ex := s.profChatAgents[id]; ex != nil { // lost the race → keep the winner ca = ex @@ -786,6 +802,7 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("GET /api/traffic/exchange", s.getTrafficExchange) mux.HandleFunc("GET /api/traffic/blob", s.getTrafficBlob) mux.HandleFunc("GET /api/commands", s.pgListCommands) + mux.HandleFunc("GET /api/commands/stats", s.pgToolStats) // 按工具聚合调用次数 mux.HandleFunc("GET /api/llm/records", s.pgListLLMRecords) mux.HandleFunc("DELETE /api/llm/records", s.pgDeleteLLMRecords) mux.HandleFunc("GET /api/llm/records/tasks", s.pgLLMTasks) @@ -1335,6 +1352,7 @@ func (s *Server) testLLM(w http.ResponseWriter, r *http.Request) { ThinkingType string `json:"thinking_type"` ReasoningEffort string `json:"reasoning_effort"` ProfileID *int64 `json:"profile_id"` // 测已存 profile 时传入：api_key 为空则用它存的 key + Streaming *bool `json:"streaming"` // 省略=流式，与保存 profile 时同一套默认 } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeErr(w, 400, err.Error()) @@ -1345,6 +1363,11 @@ func (s *Server) testLLM(w http.ResponseWriter, r *http.Request) { // reasoning_effort/thinking field fails the test too (no false "test ok, run 400"). cfg.ThinkingType = req.ThinkingType cfg.ReasoningEffort = req.ReasoningEffort + // 同理，收发模式也照该 profile 的选择来：只支持其中一种通道的端点必须在这里就 + // 暴露，而不是等会话里才发现"测试通过的配置根本跑不动"。 + if req.Streaming != nil { + cfg.Stream = *req.Streaming + } // API Key 解析优先级：表单输入 > 指定 profile 存的 key > 全局配置的 key。 // 已存 profile 的 key 不回传浏览器，所以测试已存配置时表单为空，需从 DB 取。 if cfg.APIKey == "" && req.ProfileID != nil { @@ -1361,12 +1384,27 @@ func (s *Server) testLLM(w http.ResponseWriter, r *http.Request) { writeJSON(w, 200, map[string]any{"ok": false, "error": "未提供 API Key"}) return } - lat, err := agent.TestConnection(r.Context(), cfg) + lat, reply, err := agent.TestConnection(r.Context(), cfg) if err != nil { writeJSON(w, 200, map[string]any{"ok": false, "error": err.Error()}) return } - writeJSON(w, 200, map[string]any{"ok": true, "latency_ms": lat.Milliseconds(), "model": cfg.Model}) + // 回传模型实际回复，让"测试通过"有据可查：看得见它确实说了话，而不只是 HTTP 200。 + writeJSON(w, 200, map[string]any{ + "ok": true, "latency_ms": lat.Milliseconds(), "model": cfg.Model, "reply": truncateReply(reply), + }) +} + +// truncateReply clips a connection-test reply for display. A model told to answer +// "OK" can still ramble (or think out loud); the UI only needs enough to show it +// really said something. Rune-based so multibyte text never splits mid-character. +func truncateReply(s string) string { + const max = 200 + r := []rune(s) + if len(r) <= max { + return s + } + return string(r[:max]) + "…" } type createTaskReq struct { diff --git a/server/server_mgmt.go b/server/server_mgmt.go index 3b2fad7..02baba2 100644 --- a/server/server_mgmt.go +++ b/server/server_mgmt.go @@ -1736,10 +1736,14 @@ func (s *Server) pgSaveProfile(w http.ResponseWriter, r *http.Request) { return } // APIKey has json:"-" on db.LLMProfile so it is never leaked to the UI on - // read; accept it here via a sibling field for create/update. + // read; accept it here via a sibling field for create/update. Streaming is a + // *bool shadowing the embedded json:"streaming": an absent field must default + // to streaming (true), which a plain bool zero value (false = non-streaming) + // would get wrong — legacy/partial clients that never send it stay streaming. var body struct { db.LLMProfile - APIKey string `json:"api_key"` + APIKey string `json:"api_key"` + Streaming *bool `json:"streaming"` } if err := decode(r, &body); err != nil { writeErr(w, 400, err.Error()) @@ -1747,6 +1751,7 @@ func (s *Server) pgSaveProfile(w http.ResponseWriter, r *http.Request) { } p := body.LLMProfile p.APIKey = body.APIKey + p.Streaming = body.Streaming == nil || *body.Streaming id, err := pg.SaveProfile(&p) if err != nil { writeErr(w, 500, err.Error()) diff --git a/server/task_llm.go b/server/task_llm.go index 01a94b7..5a22600 100644 --- a/server/task_llm.go +++ b/server/task_llm.go @@ -139,6 +139,36 @@ func (r *taskLLMRuntime) current() (*Task, int64, int64, llm.Provider, error) { return t, 0, pt.LLMChainRevision, prov, nil } +// nonStreaming reports whether the task's currently-active LLM source is set to +// non-streaming. It mirrors current()'s source precedence (agent binding → +// active chain profile → global) but reads only the streaming flag. Read-only +// and best-effort: any resolution error falls back to streaming (false), the +// safe default. If failover switches to a profile with a different streaming +// setting, the change takes effect on the next agent run (a fresh Session is +// built per run in captureRun). +func (r *taskLLMRuntime) nonStreaming() bool { + taskNum, err := parseTaskID(r.taskID) + if err != nil { + return false + } + pt, err := r.s.m.pg.GetTask(taskNum) + if err != nil || pt == nil { + return false + } + if _, cfg, ok := r.s.agentBindingProvider(r.agentKey); ok { + return !cfg.Stream + } + if len(pt.LLMProfileIDs) > 0 && pt.ActiveLLMProfileID != nil { + if _, cfg, ok := r.s.providerForProfile(*pt.ActiveLLMProfileID); ok { + return !cfg.Stream + } + } + if _, cfg, ok := r.s.globalProvider(); ok { + return !cfg.Stream + } + return false +} + func parseTaskID(id string) (int64, error) { n, err := strconv.ParseInt(id, 10, 64) if err != nil || n <= 0 { @@ -147,9 +177,11 @@ func parseTaskID(id string) (int64, error) { return n, nil } -func (r *taskLLMRuntime) Stream(ctx context.Context, req llm.CompletionRequest) iter.Seq2[llm.StreamEvent, error] { - ctx = llmrec.WithTaskID(ctx, r.taskID) - hooks := taskLLMStreamHooks{ +// streamHooks builds the failover/exhaustion callbacks shared by Stream and +// Complete: how to read the current profile selection, how to mark it quota +// exhausted, and how to emit a failover transition. +func (r *taskLLMRuntime) streamHooks() taskLLMStreamHooks { + return taskLLMStreamHooks{ current: func() (taskLLMSelection, error) { task, profileID, revision, provider, err := r.current() return taskLLMSelection{task: task, profileID: profileID, revision: revision, provider: provider}, err @@ -169,7 +201,70 @@ func (r *taskLLMRuntime) Stream(ctx context.Context, req llm.CompletionRequest) r.s.emitTaskLLMTransition(selection.task, transition, cause) }, } - return streamTaskLLM(ctx, r.taskID, req, hooks) +} + +func (r *taskLLMRuntime) Stream(ctx context.Context, req llm.CompletionRequest) iter.Seq2[llm.StreamEvent, error] { + ctx = llmrec.WithTaskID(ctx, r.taskID) + return streamTaskLLM(ctx, r.taskID, req, r.streamHooks()) +} + +// Complete is the non-streaming counterpart of Stream. A non-streaming call is +// atomic — it never delivers partial output — so every failure is safe to retry +// on the same provider or fail over to the next profile without risking +// duplicated model output or tool execution (the "committed" bookkeeping the +// streaming path needs is unnecessary here). +func (r *taskLLMRuntime) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) { + ctx = llmrec.WithTaskID(ctx, r.taskID) + return completeTaskLLM(ctx, r.taskID, req, r.streamHooks()) +} + +func completeTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest, hooks taskLLMStreamHooks) (llm.Message, string, llm.Usage, error) { + for { + selection, err := hooks.current() + if err != nil { + return llm.Message{}, "", llm.Usage{}, err + } + var ( + msg llm.Message + sr string + usage llm.Usage + callErr error + ) + // 同 provider 安全窗口重试:非流式调用要么整体成功、要么整体失败,没有 + // 中途已交付输出的问题,所以任何瞬时失败都可原样重试。 + for attempt := 0; ; attempt++ { + msg, sr, usage, callErr = selection.provider.Complete(ctx, req) + if callErr != nil && ctx.Err() == nil && + attempt < sameProviderStreamRetries && isRetryableStreamError(callErr) { + backoff := sameProviderRetryBackoff(attempt) + log.Printf("[task-llm] task %s 非流式调用失败,%v 后同 provider 重试 (%d/%d): %v", + taskID, backoff, attempt+1, sameProviderStreamRetries, callErr) + if sleepCtx(ctx, backoff) { + break // 退避期间 ctx 取消 → 停止重试 + } + continue + } + break + } + if callErr == nil { + return msg, sr, usage, nil + } + // profileID=0: 显式链已被清空;非额度错误:透传。二者都不改任务 failover 状态。 + if selection.profileID == 0 || !isQuotaExhaustedError(callErr) { + return llm.Message{}, "", llm.Usage{}, callErr + } + transition, markErr := hooks.exhaust(selection, callErr) + if markErr != nil { + return llm.Message{}, "", llm.Usage{}, fmt.Errorf("mark profile quota exhausted after %v: %w", callErr, markErr) + } + if transition.Advanced && !transition.Stale && hooks.transition != nil { + hooks.transition(selection, transition, callErr) + } + if !transition.Stale && transition.NextProfileID == nil { + return llm.Message{}, "", llm.Usage{}, &taskLLMError{taskID: taskID, chainExhausted: transition.ChainExhausted, cause: callErr} + } + // 没有任何输出交付给调用方,换下一个 profile 重放同一逻辑请求是安全的。 + } } func streamTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest, hooks taskLLMStreamHooks) iter.Seq2[llm.StreamEvent, error] { @@ -435,6 +530,7 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle { window := workerRuntime.CompactionWindow() wk := agent.NewWorker(workerRuntime, "task-router", s.m.dir, tx, window, s.agentMaxTurns("worker")) wk.SetCompactionWindowResolver(workerRuntime.CompactionWindow) + wk.SetNonStreaming(workerRuntime.nonStreaming) // 按任务当前激活 profile 的流式开关(每轮读) wk.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second) wk.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) wk.SetMemory(memory.NewStore(filepath.Join(s.m.dir, "memory"))) @@ -442,6 +538,7 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle { wk.SetConstraintInject(s.constraintInjectWorker) // 操作约束注入 worker(可配置,默认开) pl := agent.NewPlanner(plannerRuntime, "task-router", s.m.dir, tx, plannerRuntime.CompactionWindow(), s.agentMaxTurns("planner")) pl.SetCompactionWindowResolver(plannerRuntime.CompactionWindow) + pl.SetNonStreaming(plannerRuntime.nonStreaming) pl.SetKillWork(s.engine.KillWork) pl.SetSteerWork(s.engine.SteerWork) pl.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) @@ -449,6 +546,7 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle { pl.SetConstraintInject(s.constraintInjectPlanner) // 操作约束注入 planner(可配置,默认开) main := agent.NewMainAgent(mainRuntime, "task-router", s.m.dir, tx, mainRuntime.CompactionWindow(), s.agentMaxTurns("mainagent")) main.SetCompactionWindowResolver(mainRuntime.CompactionWindow) + main.SetNonStreaming(mainRuntime.nonStreaming) main.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) main.SetWebSearch(s.webSearchFor("mainagent")) main.SetSteerWork(s.engine.SteerWork) // steer_work：人对运行中 work 实时纠偏 diff --git a/server/task_llm_test.go b/server/task_llm_test.go index 48da7d3..5316b9b 100644 --- a/server/task_llm_test.go +++ b/server/task_llm_test.go @@ -37,6 +37,23 @@ func (p *scriptedLLMProvider) Stream(context.Context, llm.CompletionRequest) ite } } +func (p *scriptedLLMProvider) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) { + return accumulateStreamForTest(ctx, p.Stream, req) +} + +// accumulateStreamForTest drains a mock's Stream to satisfy the non-streaming +// Complete method. +func accumulateStreamForTest(ctx context.Context, stream func(context.Context, llm.CompletionRequest) iter.Seq2[llm.StreamEvent, error], req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) { + acc := llm.NewAccumulator() + for ev, err := range stream(ctx, req) { + if err != nil { + return llm.Message{}, "", llm.Usage{}, err + } + acc.Add(ev) + } + return acc.Message(), acc.StopReason, acc.Usage, nil +} + // withZeroRetryBackoff sets the same-provider retry backoff to zero for the // duration of a (serial) test and returns a restore func for defer. func withZeroRetryBackoff() func() { @@ -73,6 +90,10 @@ func (p *flakyThenOKProvider) Stream(context.Context, llm.CompletionRequest) ite } } +func (p *flakyThenOKProvider) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) { + return accumulateStreamForTest(ctx, p.Stream, req) +} + func collectTaskLLMStream(seq iter.Seq2[llm.StreamEvent, error]) ([]llm.StreamEvent, error) { var events []llm.StreamEvent var streamErr error diff --git a/web/src/app/(main)/function/commands/page.tsx b/web/src/app/(main)/function/commands/page.tsx index 7d58b48..6a5954e 100644 --- a/web/src/app/(main)/function/commands/page.tsx +++ b/web/src/app/(main)/function/commands/page.tsx @@ -2,17 +2,18 @@ import * as React from "react"; -import { ChevronLeftIcon, ChevronRightIcon, Loader2Icon, SearchIcon, TerminalIcon } from "lucide-react"; +import { BarChart3Icon, ChevronLeftIcon, ChevronRightIcon, Loader2Icon, SearchIcon, TerminalIcon } from "lucide-react"; import { Badge } from "@/components/ui/badge"; import { Button } from "@/components/ui/button"; import { Card } from "@/components/ui/card"; +import { Dialog, DialogContent, DialogDescription, DialogHeader, DialogTitle } from "@/components/ui/dialog"; import { Input } from "@/components/ui/input"; import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select"; import { Sheet, SheetContent, SheetDescription, SheetHeader, SheetTitle } from "@/components/ui/sheet"; import { Table, TableBody, TableCell, TableHead, TableHeader, TableRow } from "@/components/ui/table"; import { api } from "@/lib/api"; -import type { CommandRecord } from "@/lib/types"; +import type { CommandRecord, ToolStat } from "@/lib/types"; import { cn } from "@/lib/utils"; function fmtTime(ts: string) { @@ -59,6 +60,11 @@ export default function CommandsPage() { const [total, setTotal] = React.useState(0); const [loading, setLoading] = React.useState(false); + // 各工具调用次数。弹窗打开时才拉取（多一次聚合查询，不必每次翻页都付）。 + const [statsOpen, setStatsOpen] = React.useState(false); + const [stats, setStats] = React.useState([]); + const [statsLoading, setStatsLoading] = React.useState(false); + // Selected execution is rendered in a right-side detail sheet. const [selected, setSelected] = React.useState(null); @@ -96,6 +102,24 @@ export default function CommandsPage() { }; }, [page, size, queryQ, taskFilter]); + // 统计跟随筛选条件走，和表格描述的是同一批记录（但不分页）。 + React.useEffect(() => { + if (!statsOpen) return; + let alive = true; + setStatsLoading(true); + api + .commandStats({ task: taskFilter || undefined, q: queryQ || undefined }) + .then((r) => alive && setStats(r.stats ?? [])) + .catch(() => alive && setStats([])) + .finally(() => alive && setStatsLoading(false)); + return () => { + alive = false; + }; + }, [statsOpen, queryQ, taskFilter]); + + const statsTotal = stats.reduce((n, s) => n + s.total, 0); + const statsErrors = stats.reduce((n, s) => n + s.errors, 0); + const statsMax = stats.reduce((n, s) => Math.max(n, s.total), 0); const totalPages = Math.max(1, Math.ceil(total / size)); const rangeStart = total === 0 ? 0 : page * size + 1; const rangeEnd = page * size + commands.length; @@ -141,6 +165,11 @@ export default function CommandsPage() { + +
{rangeStart}–{rangeEnd} / {total} @@ -243,6 +272,66 @@ export default function CommandsPage() {
+ {/* 工具调用统计：与表格同一批记录（同筛选、不分页） */} + + + + 工具调用统计 + + {taskFilter || queryQ ? "当前筛选条件下的全部记录" : "全部工具执行记录"} + {stats.length > 0 && ( + <> + {" · "} + {stats.length} 个工具 ·{" "} + {statsTotal} 次调用 + {statsErrors > 0 && ( + <> + {" · 失败 "} + {statsErrors} + + )} + + )} + + + + {statsLoading && stats.length === 0 ? ( +
+ +
+ ) : stats.length === 0 ? ( +
暂无统计数据
+ ) : ( +
+ {stats.map((s) => ( +
+
+
+ {s.tool} + {s.errors > 0 && ( + 失败 {s.errors} + )} +
+
+
0 ? (s.total / statsMax) * 100 : 0}%` }} + /> +
+
+
+
{s.total}
+
+ {statsTotal > 0 ? ((s.total / statsTotal) * 100).toFixed(1) : "0.0"}% +
+
+
+ ))} +
+ )} + +
+ !open && setSelected(null)}> {selected && ( diff --git a/web/src/app/(main)/system/llm/page.tsx b/web/src/app/(main)/system/llm/page.tsx index b08c6ef..35c7806 100644 --- a/web/src/app/(main)/system/llm/page.tsx +++ b/web/src/app/(main)/system/llm/page.tsx @@ -309,6 +309,7 @@ function ProfileSheet({ const [effort, setEffort] = React.useState(NONE); const [priority, setPriority] = React.useState("0"); // 轮询顺位;越大越先 const [poolExclude, setPoolExclude] = React.useState(false); + const [streaming, setStreaming] = React.useState(true); // true=流式(默认);false=非流式 const [testing, setTesting] = React.useState(false); const [saving, setSaving] = React.useState(false); const [models, setModels] = React.useState([]); @@ -331,6 +332,7 @@ function ProfileSheet({ setEffort(fromStore(profile?.reasoning_effort)); setPriority(String(profile?.priority ?? 0)); setPoolExclude(profile?.pool_exclude ?? false); + setStreaming(profile?.streaming ?? true); setApiKey(""); setKeyHint(profile?.api_key_hint ?? ""); setModels([]); @@ -365,8 +367,22 @@ function ProfileSheet({ try { // 用配置实际会跑的思考参数来测，这样不支持该字段的模型在这里就失败， // 而不是等到跑任务时才炸。传 profile id：Key 输入框留空时用已存的 Key。 - const r = await api.testLLM(format, model, baseUrl, apiKey, proxy, toStore(thinkingType), toStore(effort), profileId); - if (r.ok) toast.success(`连接成功 · ${r.latency_ms ?? "?"}ms · ${r.model ?? model}`); + const r = await api.testLLM( + format, + model, + baseUrl, + apiKey, + proxy, + toStore(thinkingType), + toStore(effort), + profileId, + streaming, + ); + // 回复内容一并展示：看得见模型确实说了话，才算和会话里跑通是一回事。 + if (r.ok) + toast.success(`连接成功 · ${r.latency_ms ?? "?"}ms · ${r.model ?? model}`, { + description: r.reply ? `回复：${r.reply}` : undefined, + }); else toast.error(`连接失败：${r.error ?? "未知"}`); } catch (e) { toast.error(`测试出错：${(e as Error).message}`); @@ -398,6 +414,7 @@ function ProfileSheet({ reasoning_effort: toStore(effort), priority: Number(priority) || 0, pool_exclude: poolExclude, + streaming, }); if (isNew) toast.success(`已新建：${name.trim()}（在卡片上「设为激活」以启用）`); else toast.success(profile?.is_default ? "已保存，激活配置即时生效，无需重启" : "已保存"); @@ -525,7 +542,8 @@ function ProfileSheet({ onChange={(e) => setProxy(e.target.value)} />

- 仅 LLM 出站请求走此代理，支持 http/https/socks5，可带账号密码（如 socks5://user:pass@host:port，密码含特殊字符需 URL 编码）；留空表示不使用代理（直连）。 + 仅 LLM 出站请求走此代理，支持 http/https/socks5，可带账号密码（如 + socks5://user:pass@host:port，密码含特殊字符需 URL 编码）；留空表示不使用代理（直连）。

@@ -595,6 +613,17 @@ function ProfileSheet({ +
+
+ +

+ 开启（默认）走流式 SSE，有运行中实时进度与实时 token 计数。 关闭则走真·非流式（stream:false， + 一次性返回完整响应）——可绕开部分网关糟糕的 SSE 实现（空帧 / 思考字段丢帧）， + 代价是失去运行中的实时进度。 +

+
+ +
@@ -602,8 +631,8 @@ function ProfileSheet({

- 控制是否发送 thinking 字段。不发送=不带该字段（兼容 MiniMax 等不支持 - 的模型）；关闭=发 disabled；开启=发 enabled。与下面的强度互相独立。 + 控制是否发送 thinking 字段。不发送=不带该字段（兼容 MiniMax 等不支持 的模型）；关闭=发 + disabled；开启=发 enabled。与下面的强度互相独立。

diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index 9a9f3b7..ba197c9 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -74,6 +74,7 @@ import type { TokenTotal, TokenUsage, Tool, + ToolStat, TrafficDetail, TrafficHost, TrafficResp, @@ -677,8 +678,10 @@ export const api = { thinking_type = "", reasoning_effort = "", profile_id?: number, + streaming = true, // 用该配置真实的收发模式来测，别让"流式能通、非流式不通"漏到会话里 ) => - post<{ ok: boolean; error?: string; latency_ms?: number; model?: string }>("/llm/test", { + // reply = 模型实际回复(已截断);一个字都不回的配置后端直接判失败 + post<{ ok: boolean; error?: string; latency_ms?: number; model?: string; reply?: string }>("/llm/test", { provider, model, base_url, @@ -687,6 +690,7 @@ export const api = { thinking_type, reasoning_effort, profile_id, + streaming, }), llmProfiles: () => get<{ profiles: LLMProfile[] }>("/llm/profiles").then((r) => arr(r.profiles)), saveLLMProfile: (p: { @@ -704,6 +708,7 @@ export const api = { reasoning_effort?: string; // ""(不发送)|"low"|"medium"|"high"|"xhigh"|"max" priority?: number; // 轮询顺位，越大越先用 pool_exclude?: boolean; // true=不作为故障转移目标 + streaming?: boolean; // true(默认)=流式 | false=非流式 }) => post<{ id: number }>("/llm/profiles", p), deleteLLMProfile: (id: string) => del<{ deleted: number }>(`/llm/profiles/${id}`), activateLLMProfile: (id: string) => post<{ ok: boolean }>("/llm/profiles/active", { id: Number(id) }), @@ -972,6 +977,13 @@ export const api = { sp.set("size", String(params?.size ?? 50)); return get<{ commands: CommandRecord[]; total: number }>(`/commands?${sp}`); }, + // 各工具调用次数；沿用列表的 task/q 筛选，统计的是整个结果集而非当前页。 + commandStats: (params?: { task?: string; q?: string }) => { + const sp = new URLSearchParams(); + if (params?.task) sp.set("task", params.task); + if (params?.q) sp.set("q", params.q); + return get<{ stats: ToolStat[] }>(`/commands/stats?${sp}`); + }, // ---- LLM records ---- llmRecords: (params?: { model?: string; session?: string; task?: string; page?: number; size?: number }) => { diff --git a/web/src/lib/mock/handler.ts b/web/src/lib/mock/handler.ts index 6002dae..407fe4e 100644 --- a/web/src/lib/mock/handler.ts +++ b/web/src/lib/mock/handler.ts @@ -1493,6 +1493,17 @@ function route(m: string, path: string, seg: string[], q: URLSearchParams, b: Re // ── 工具执行历史 ── if (path === "/commands" && m === "GET") return { commands: D.commandRecords, total: D.commandRecords.length }; + if (path === "/commands/stats" && m === "GET") { + const tally = new Map(); + for (const c of D.commandRecords) { + const tool = c.tool || "-"; + const s = tally.get(tool) ?? { tool, total: 0, errors: 0 }; + s.total++; + if (c.is_error) s.errors++; + tally.set(tool, s); + } + return { stats: [...tally.values()].sort((a, b) => b.total - a.total || a.tool.localeCompare(b.tool)) }; + } // ── LLM ── if (path === "/llm/records" && m === "GET") return { records: mockLLMRecords, total: mockLLMRecords.length }; @@ -1509,7 +1520,8 @@ function route(m: string, path: string, seg: string[], q: URLSearchParams, b: Re } if (path === "/llm" && m === "GET") return D.llmConfig; if (path === "/llm" && m === "POST") return { ok: true }; - if (path === "/llm/test") return { ok: true, latency_ms: 128, model: String(b.model ?? "claude-opus-4-8") }; + if (path === "/llm/test") + return { ok: true, latency_ms: 128, model: String(b.model ?? "claude-opus-4-8"), reply: "OK" }; if (path === "/llm/profiles" && m === "GET") return { profiles: D.llmProfiles }; if (path === "/llm/profiles" && m === "POST") return { id: Number(b.id) || 3 }; if (path === "/llm/profiles/active") return { ok: true }; diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 633688e..5c672d1 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -869,6 +869,8 @@ export interface LLMProfile { priority?: number; // true = 不作为故障转移目标（仍可被 agent/任务显式绑定使用）。 pool_exclude?: boolean; + // true（默认）= 流式(SSE) | false = 真·非流式(stream:false，一次性返回)。 + streaming?: boolean; } // ---- LLM 轮询（故障转移）---- @@ -1133,6 +1135,13 @@ export interface CommandRecord { created_at: string; } +// 单个工具的调用统计（/commands/stats）；errors 为其中失败的次数。 +export interface ToolStat { + tool: string; + total: number; + errors: number; +} + // ---- LLM recording ---- export interface LLMRecordItem { id: number;