diff --git a/server/chat_mentions_test.go b/server/chat_mentions_test.go index 7b3b87d..616a00e 100644 --- a/server/chat_mentions_test.go +++ b/server/chat_mentions_test.go @@ -150,7 +150,9 @@ func TestChatMentionWorkerReceivesServerDetails(t *testing.T) { return llm.Message{Role: llm.RoleAssistant, Content: []llm.ContentBlock{llm.TextBlock("已读取引用")}}, "end_turn", llm.Usage{}, nil }}, "test", m.dir, nil, 10000, 1) worker.SetNonStreaming(func() bool { return true }) - s.engine.UseLLM(nil, worker) + s.engine.SetAuthoritativeAgentResolver(func(*Task) (*agent.Planner, *agent.Worker) { + return nil, worker + }) send := func(message string) *httptest.ResponseRecorder { body, _ := json.Marshal(map[string]string{"message": message, "request_id": "worker-mention-test"}) r := httptest.NewRequest("POST", "/", strings.NewReader(string(body))) diff --git a/server/engine.go b/server/engine.go index e79e018..13aca38 100644 --- a/server/engine.go +++ b/server/engine.go @@ -96,9 +96,6 @@ type Engine struct { bc *Broadcaster // live activity pub/sub (SSE) - mu sync.RWMutex - planner *agent.Planner - worker *agent.Worker started sync.Map // taskID -> bool, so Run is idempotent per task lastAct sync.Map // taskID -> int64 unix, last planner/worker activity (heartbeat) llmCalls sync.Map // taskID -> *int64, actual planner/worker/main-agent LLM calls @@ -146,11 +143,14 @@ type Engine struct { inflight sync.Map // taskID -> *int64, 在跑的 planner.Plan + worker.Execute 计数(用于 drain) coordStarted sync.Map // taskID -> bool, deadline 协调器是否已启动(Run/reload 去重) - // resolve, if set, returns a task's dedicated planner/worker. In legacy mode - // nil,nil falls back to the global pair; an authoritative resolver uses nil,nil - // to mean this task is deliberately unavailable (for example an exhausted chain). + // resolve returns a task's dedicated planner/worker (wired by the server as the + // authoritative task-router). nil,nil means this task is deliberately unavailable + // (for example an exhausted failover chain) — there is no global-pair fallback. resolve func(t *Task) (*agent.Planner, *agent.Worker) resolveAuthoritative bool + // readiness reports whether a global LLM provider is configured — the signal behind + // Ready()/the llm_configured indicator. Wired once at startup; nil → not ready. + readiness func() bool } type taskRuntime struct { @@ -696,18 +696,9 @@ func (e *Engine) appendActivity(t *Task, r db.Activity) (int64, error) { return 0, err } -// UseLLM installs (or replaces) the LLM planner + work agent at runtime. -func (e *Engine) UseLLM(p *agent.Planner, w *agent.Worker) { - e.mu.Lock() - e.planner, e.worker = p, w - e.mu.Unlock() -} - -func (e *Engine) snapshot() (*agent.Planner, *agent.Worker) { - e.mu.RLock() - defer e.mu.RUnlock() - return e.planner, e.worker -} +// SetReadiness wires the global "an LLM provider is configured" predicate (read by +// Ready() / the llm_configured indicator). Called once at startup. +func (e *Engine) SetReadiness(fn func() bool) { e.readiness = fn } // SetAgentResolver installs a per-task planner/worker resolver (wired by the server). // Called once at startup before any task loop runs, so no lock is needed on reads. @@ -724,9 +715,9 @@ func (e *Engine) SetAuthoritativeAgentResolver(fn func(t *Task) (*agent.Planner, e.resolveAuthoritative = true } -// snapshotFor returns the planner/worker a task should run on: the task's dedicated -// pair when it pins a specific LLM profile (via the resolver), else the global active -// pair. The resolver returning nil means "no override" (fall back to global). +// snapshotFor returns the planner/worker a task should run on, from the task-router +// resolver. nil,nil means the task is deliberately unavailable (e.g. an exhausted +// failover chain); there is no global-pair fallback. func (e *Engine) snapshotFor(t *Task) (*agent.Planner, *agent.Worker) { if e.resolve != nil { p, w := e.resolve(t) @@ -734,13 +725,13 @@ func (e *Engine) snapshotFor(t *Task) (*agent.Planner, *agent.Worker) { return p, w } } - return e.snapshot() + return nil, nil } -// Ready reports whether an LLM provider is configured. +// Ready reports whether a global LLM provider is configured (via the readiness +// predicate wired at startup). func (e *Engine) Ready() bool { - p, w := e.snapshot() - return p != nil && w != nil + return e.readiness != nil && e.readiness() } // ReadyFor reports whether a specific task can resolve a planner/worker pair. @@ -751,14 +742,6 @@ func (e *Engine) ReadyFor(t *Task) bool { return p != nil && w != nil } -// Mode reports "llm" when ready, else "idle". -func (e *Engine) Mode() string { - if e.Ready() { - return "llm" - } - return "idle" -} - // Run starts the planner loop + N worker loops for a task. The loops always run // but no-op until an LLM is configured (so a task created while idle picks up // automatically once LLM is set from the UI). diff --git a/server/server.go b/server/server.go index a9e9ff2..c52eba5 100644 --- a/server/server.go +++ b/server/server.go @@ -164,6 +164,13 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi b := s.agentsForTask(t) return b.pl, b.wk }) + // Global readiness (Ready()/llm_configured): a global active LLM provider is + // installed. Task-level runnability is separate (ReadyFor → the resolver above). + s.engine.SetReadiness(func() bool { + s.cfgMu.Lock() + defer s.cfgMu.Unlock() + return s.llmOn + }) // Wire DB-stored prompt templates into the agents (新版方案 §3.3 / §5a). With no // override row, agents keep their built-in defaults — behavior is unchanged. if m.pg != nil { @@ -398,44 +405,6 @@ func (s *Server) webSearchFor(key string) agent.WebSearchOpts { return o } -// buildPlannerWorker builds a planner+worker pair. Each agent resolves its OWN LLM by -// precedence agent-binding → pin → the passed global fallback (gProv/gCfg), so planner -// and worker can run on different models (e.g. a stronger planner, a cheaper worker). -// pinID is the task's pinned profile (nil on the global active path). With no agent -// binding and no pin, both fall back to gProv/gCfg — identical to the previous single- -// provider behavior. Called by applyLLM (global active pair). -func (s *Server) buildPlannerWorker(pinID *int64, gProv llm.Provider, gCfg agent.Config) (*agent.Planner, *agent.Worker) { - tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts")) // raw LLM conversation logs - // traffic host tools flow through ToolAugment for every agent and are filtered by - // the tools-table binding (default = worker), so worker behavior is unchanged. - wProv, wCfg := s.providerForAgent("worker", pinID, gProv, gCfg) - wk := agent.NewWorker(wProv, wCfg.Model, s.m.dir, tx, wCfg.CompactionWindow(), s.agentMaxTurns("worker")) - wk.SetFindingRecorder(s.evidenceStore()) - wk.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second) - wk.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) - wk.SetWebSearch(s.webSearchFor("worker")) - wk.SetConstraintInject(s.constraintInjectWorker) // 操作约束注入 worker(可配置,默认开;每轮读) - wk.SetNonStreaming(nonStreamingResolver(wCfg)) // 该 profile 选非流式时走 Provider.Complete - wk.SetMaxTokens(maxTokensResolver(wCfg)) // 单次回复输出上限(0 = 不发) - wk.SetNoaEnabled(s.m.NoaCompactionEnabled) // 实验功能:noa 上下文压缩(每 run 读) - pProv, pCfg := s.providerForAgent("planner", pinID, gProv, gCfg) - pl := agent.NewPlanner(pProv, pCfg.Model, s.m.dir, tx, pCfg.CompactionWindow(), s.agentMaxTurns("planner")) - pl.SetFindingRecorder(s.evidenceStore()) - pl.SetKillWork(s.engine.KillWork) // planner kill_work → terminate a running work - pl.SetSteerWork(s.engine.SteerWork) // planner steer_work → inject mid-run course-correction - 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 - pl.SetMaxTokens(maxTokensResolver(pCfg)) // 单次回复输出上限(0 = 不发) - pl.SetNoaEnabled(s.m.NoaCompactionEnabled) // 实验功能:noa 上下文压缩(每 run 读) - // cold-digest §7: background cold-node compaction, on the planner's provider/ - // model (§4 uses the running agent's model). Uses Complete (non-streaming) for - // the one-shot body summarization. - pl.SetCompactor(agent.NewCompactor(pProv, pCfg.Model)) - 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. @@ -483,8 +452,9 @@ func (s *Server) applyLLM(cfg agent.Config) error { if act, err := s.m.pg.ActiveProfile(); err == nil && act != nil { prov = s.poolForActive(act.ID, prov, cfg) } - pl, wk := s.buildPlannerWorker(nil, prov, cfg) - s.engine.UseLLM(pl, wk) + // No global planner/worker pair: every task runs on its own task-routed pair + // (agentsForTask), resolved through the engine's authoritative resolver. Global + // readiness (Ready()/llm_configured) is reported from s.llmOn, set below. tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts")) win := cfg.CompactionWindow() @@ -622,21 +592,6 @@ func (s *Server) providerForProfile(id int64) (llm.Provider, agent.Config, bool) return prov, cfg, true } -// providerForAgent returns the provider+cfg one agent should run on: its bound/pinned -// profile if that resolves, else the passed global fallback (gProv/gCfg). -// A bound/pinned profile is EXCLUSIVE by default: it does not fail over, so an -// agent deliberately put on a specific model never silently runs on another one. -// Turning on llm_pool_bind_fallback relaxes that — poolForBinding then appends the -// failover chain behind it. -func (s *Server) providerForAgent(agentKey string, pinID *int64, gProv llm.Provider, gCfg agent.Config) (llm.Provider, agent.Config) { - if eff := s.effectiveProfileForAgent(agentKey, pinID); eff != nil { - if prov, cfg, ok := s.providerForProfile(*eff); ok { - return s.poolForBinding(*eff, prov, cfg), cfg - } - } - return gProv, gCfg -} - // chatAgentForProfile returns a ChatAgent built from a specific LLM profile, cached // per profile id. Returns nil if the profile is missing or has no API key. func (s *Server) chatAgentForProfile(id int64) *agent.ChatAgent { diff --git a/server/task_llm.go b/server/task_llm.go index 0856f75..2cae9ce 100644 --- a/server/task_llm.go +++ b/server/task_llm.go @@ -583,9 +583,8 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle { pl.SetWebSearch(s.webSearchFor("planner")) pl.SetConstraintInject(s.constraintInjectPlanner) // 操作约束注入 planner(可配置,默认开) // cold-digest §7: 冷节点后台压缩。引擎经权威解析器实际驱动的就是这套 per-task planner - // (agentsForTask),所以 Compactor 必须接在这里;buildPlannerWorker 那套全局 pair 被解析器 - // 旁路、从不跑规划轮,接在那里等于不生效。走任务路由的 planner provider(§4:与 agent 同模型), - // 压缩用 Complete 一次性生成 body。 + // (agentsForTask),Compactor 必须接在这里。走任务路由的 planner provider(§4:与 agent + // 同模型,随任务 LLM 链解析),压缩用 Complete 一次性生成 body。 pl.SetCompactor(agent.NewCompactor(plannerRuntime, "task-router")) main := agent.NewMainAgent(mainRuntime, "task-router", s.m.dir, tx, mainRuntime.CompactionWindow(), s.agentMaxTurns("mainagent")) main.SetFindingRecorder(s.evidenceStore()) diff --git a/server/task_llm_test.go b/server/task_llm_test.go index 5316b9b..4f33bea 100644 --- a/server/task_llm_test.go +++ b/server/task_llm_test.go @@ -601,9 +601,11 @@ func TestTaskRuntimeAvailableUnboundRoleFollowsChainThenGlobal(t *testing.T) { func TestAuthoritativeTaskResolverDoesNotFallBackWhenChainUnavailable(t *testing.T) { t.Parallel() e := &Engine{} - e.UseLLM(&agent.Planner{}, &agent.Worker{}) + e.SetAuthoritativeAgentResolver(func(*Task) (*agent.Planner, *agent.Worker) { + return &agent.Planner{}, &agent.Worker{} + }) if !e.ReadyFor(&Task{ID: "7"}) { - t.Fatal("global provider should initially make task ready") + t.Fatal("an available authoritative chain should make the task ready") } e.SetAuthoritativeAgentResolver(func(*Task) (*agent.Planner, *agent.Worker) { return nil, nil