chore: sync upstream v0.3.6 changes

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()

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,
 	}
 }
 

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) != "" {

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

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 {

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())

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 实时纠偏

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