diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 16e33a1d91..bdea892327 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -323,6 +323,52 @@ jobs: if-no-files-found: ignore retention-days: 7 + - name: test (Windows Controller routing root budget) + if: env.RUN_STEPS == 'true' && runner.os == 'Windows' && github.event_name == 'pull_request' + timeout-minutes: 5 + shell: bash + run: | + set -o pipefail + go test -count=1 -timeout=3m -json -run '^Test(SemanticRootUsage|EffectRoutingRootBudget)$' ./internal/runtime/capability ./internal/assembly/boot | tee "$RUNNER_TEMP/controller-routing-budget.jsonl" + jq -se ' + [.[] | select(.Test == null and .Action == "pass") | .Package] as $packages | + [.[] | select(.Test != null and .Action == "pass") | .Test] as $tests | + all(.[]; .Action != "skip" and .Action != "fail") and + (["reasonix/internal/runtime/capability", "reasonix/internal/assembly/boot"] - $packages | length) == 0 and + ([ + "TestEffectRoutingRootBudget", + "TestEffectRoutingRootBudget/balanced", + "TestEffectRoutingRootBudget/cached_continuation", + "TestEffectRoutingRootBudget/cached_new_scope", + "TestEffectRoutingRootBudget/cached_new_task", + "TestEffectRoutingRootBudget/cached_reset", + "TestEffectRoutingRootBudget/context_zero", + "TestEffectRoutingRootBudget/dual", + "TestEffectRoutingRootBudget/headless", + "TestEffectRoutingRootBudget/interactive", + "TestEffectRoutingRootBudget/missing", + "TestEffectRoutingRootBudget/unconfigured", + "TestSemanticRootUsage", + "TestSemanticRootUsage/billed_error", + "TestSemanticRootUsage/cache", + "TestSemanticRootUsage/estimated", + "TestSemanticRootUsage/invalid_answer", + "TestSemanticRootUsage/missing", + "TestSemanticRootUsage/nil_sink", + "TestSemanticRootUsage/no_observer", + "TestSemanticRootUsage/reported" + ] - $tests | length) == 0 + ' "$RUNNER_TEMP/controller-routing-budget.jsonl" + + - name: upload Windows Controller routing root budget inventory + if: always() && env.RUN_STEPS == 'true' && runner.os == 'Windows' && github.event_name == 'pull_request' + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7 + with: + name: controller-routing-budget-windows-${{ github.run_attempt }} + path: ${{ runner.temp }}/controller-routing-budget.jsonl + if-no-files-found: ignore + retention-days: 7 + - name: test (Windows Coordinator root budget) if: env.RUN_STEPS == 'true' && runner.os == 'Windows' && github.event_name == 'pull_request' timeout-minutes: 5 diff --git a/docs/CONTROLLER_ROUTING_ROOT_BUDGET.md b/docs/CONTROLLER_ROUTING_ROOT_BUDGET.md new file mode 100644 index 0000000000..5a9cb258cf --- /dev/null +++ b/docs/CONTROLLER_ROUTING_ROOT_BUDGET.md @@ -0,0 +1,93 @@ +--- +owner: @SivanCola +backup: @esengine +status: active +reviewed: 2026-10-26 +--- + +# Controller routing root budget / Controller 路由根预算 + +## Behavior / 行为 + +Controller prepares the executor-owned token/wall root after turn admission, +checkpoint setup and blocking prompt hooks, before semantic capability routing. +Both headless Run and the interactive turn orchestrator use the same helper. +Coordinator and Agent inherit that existing scope instead of replacing it. + +Controller 在回合准入、检查点和可阻止提交的 hook 之后、语义能力路由之前准备 +executor 所有的 token/wall 根预算。headless Run 和交互编排走同一 helper； +Coordinator 和 Agent 继承该根，不再为路由后的执行重新创建根。 + +The semantic router reports its existing finalized usage to the accepted +auxiliary observer before its optional audit and public usage event. Reported +usage counts even when parsing fails or a billed attempt returns an error. +Cached decisions send no request and incur no second charge. + +语义路由把原有终态用量交给已有辅助 observer，再执行可选审计和公开事件。答案 +解析失败或已报告用量后出错仍计已知用量；缓存命中不发请求，也不再次记账。 + +## Lifetimes and authority / 生命周期与权限 + +| Boundary | Contract / 约定 | +| --- | --- | +| Root ownership | Existing Agent preparation owns the counter, clock and immutable limits / 沿用 Agent 根预算所有权 | +| Continuation | Same host delivery scope retains the root / 同一宿主 delivery scope 保留根 | +| New task or scope | Starts a new root, even with a cached semantic decision / 新任务或 scope 换根，路由缓存不改变此边界 | +| Reset | Existing ResetTaskBudget replaces the counter generation / 沿用显式 reset 的代际替换 | +| Usage identity | Existing Estimated flag and request count remain / 保留已有估算身份与请求数 | +| Cancellation | Existing turn, provider and Manager authorities remain / 不复制或扩大取消权限 | +| Permission | Routing suggestions do not grant execution authority / 路由建议不授予执行权限 | + +Zero limits remain unbounded. Positive estimated token counts follow the +existing conservative root rule; missing counts are unknown, not inferred from +request counts. Normal Agent streams retain their own accounting path, so the +semantic terminal does not charge a normal request twice. + +零额度仍不限。正数估算 token 沿用已有保守根预算规则；缺失计数保持未知，不能用 +请求数推断。普通 Agent 流仍自行记账，不通过语义路由出口重复计算普通请求。 + +Checks remain cooperative after a tool round with existing finalization. An +already exhausted root can still issue its first request and run a tool before +landing. This is not request admission, reservation, provider cancellation, +root money/currency aggregation, durable accounting or physical quiescence. + +仍在工具轮后协作检查并保留原有收尾。已超额的根仍可先发首个请求并执行工具，再 +进入软停止；不是请求准入、硬预留、供应商取消、根金额或币种汇总、持久记账或物理静默。 + +The wall clock now starts before semantic routing, after the earlier host gates. +It does not include boot, earlier queuing, input interception, reference +resolution or title generation. This unit does not cover every auxiliary +provider entry or unify each Agent's stricter local budget and telemetry. + +wall 起点移到语义路由之前、前述宿主 gate 之后；不包括 boot、更早排队、输入拦截、 +引用解析或标题生成。本单元未覆盖全部辅助供应商入口，也不统一各 Agent 更严格的 +本地额度和遥测。 + +## Evidence and compatibility / 证据与兼容 + +Actual Boot HTTP/SSE exercises headless, interactive and separate planner/model +paths, reading a real fixture through use_capability. Configured routing spend +reduces executor reads from two to one. Zero configuration, explicit context +zero, missing usage and balanced-profile controls retain their behavior. + +真实 Boot HTTP/SSE 覆盖 headless、交互和独立 planner 模型，通过 use_capability +读取实际夹具。配置额度时，路由用量使 executor 读取从两次变一次；零配置、显式 +context 零值、缺失用量和 balanced profile 对照保持原行为。 + +Repeated real Controller runs cover a cached decision across a new task, same +delivery scope, explicit reset and a new scope. These are constructed host-API +sequences, not frontend interaction, a production incident or process restart. +Router tests retain error fallback, optional sinks, usage copying and cache hits. + +重复执行真实 Controller，覆盖缓存路由下的新任务、同 scope 续跑、显式 reset 和 +新 scope。这是构造的宿主 API 序列，不是前端交互、线上事故或进程重启复现。 +路由 owner 测试保留错误回退、可选 sink、用量复制和缓存命中。 + +No API, config, wire, persisted schema, permission or default changes. Existing +model requests, routing decisions and stable prefixes remain; the HTTP fixture +checks prefix/schema equality within each run, not a live-provider cache rate. +The complete #14025 isolation, recovery and final-composition plan remains open. + +不改变 API、配置、传输、持久化 schema、权限或默认值。保留模型请求、路由决策和 +稳定前缀；HTTP 夹具只核对每轮前缀和 schema 相同，不测真实供应商缓存命中率。 +整个 #14025 隔离、恢复及最终组合验收仍未完成。 diff --git a/internal/assembly/boot/effect_routing_root_budget_test.go b/internal/assembly/boot/effect_routing_root_budget_test.go new file mode 100644 index 0000000000..add0336d5a --- /dev/null +++ b/internal/assembly/boot/effect_routing_root_budget_test.go @@ -0,0 +1,205 @@ +package boot + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + + "reasonix/internal/contract/event" + "reasonix/internal/runtime/agent" + "reasonix/internal/session/control" +) + +func TestEffectRoutingRootBudget(t *testing.T) { + for _, tc := range []struct { + name string + limit int + interactive, dual, missing, disable bool + balanced bool + repeat string + wantReads, wantHTTP int32 + }{ + {name: "headless", limit: 2000, wantReads: 1, wantHTTP: 2}, + {name: "interactive", limit: 2000, interactive: true, wantReads: 1, wantHTTP: 2}, + {name: "dual", limit: 3000, dual: true, wantReads: 1, wantHTTP: 2}, + {name: "unconfigured", wantReads: 3, wantHTTP: 4}, + {name: "context_zero", limit: 2000, disable: true, wantReads: 3, wantHTTP: 4}, + {name: "missing", limit: 2000, missing: true, wantReads: 2, wantHTTP: 3}, + {name: "balanced", limit: 2000, balanced: true, wantReads: 2, wantHTTP: 3}, + {name: "cached_new_task", limit: 2000, repeat: "new_task", wantReads: 1, wantHTTP: 2}, + {name: "cached_continuation", limit: 2000, repeat: "continuation", wantReads: 1, wantHTTP: 2}, + {name: "cached_reset", limit: 2000, repeat: "reset", wantReads: 1, wantHTTP: 2}, + {name: "cached_new_scope", limit: 2000, repeat: "new_scope", wantReads: 1, wantHTTP: 2}, + } { + t.Run(tc.name, func(t *testing.T) { + home, root := statedBootHome(t), t.TempDir() + writeFile(t, root, "fixture.txt", "routing budget fixture") + writeFile(t, home, "skills/aaa-routing-fixture/SKILL.md", "---\nname: aaa-routing-fixture\ndescription: A fixture capability\nauto-use: suggest\n---\nFixture body\n") + var routers, normals, plans, reads, routedUsage atomic.Int32 + var serial atomic.Int32 + var prefix, schemas []byte + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req struct { + Model string + Messages []struct { + Role string + Content json.RawMessage + } + Tools json.RawMessage + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Error(err) + w.WriteHeader(http.StatusBadRequest) + return + } + w.Header().Set("Content-Type", "text/event-stream") + switch req.Model { + case "router-model": + routers.Add(1) + if len(req.Messages) != 1 || req.Messages[0].Role != "user" || len(req.Tools) != 0 { + t.Error("router request contract changed") + } + if !bytes.Contains(req.Messages[0].Content, []byte("skill:aaa-routing-fixture")) { + t.Error("fixture candidate missing") + } + _, _ = io.WriteString(w, "data: {\"choices\":[{\"delta\":{\"content\":\"[\\\"skill:aaa-routing-fixture\\\"]\"},\"finish_reason\":\"stop\"}]}\n\n") + case "planner-model": + plans.Add(1) + _, _ = io.WriteString(w, "data: {\"choices\":[{\"delta\":{\"content\":\"Read fixture.txt and report its content.\"},\"finish_reason\":\"stop\"}]}\n\n") + default: + n := normals.Add(1) + if n == 1 { + prefix = bytes.Clone(req.Messages[0].Content) + schemas = bytes.Clone(req.Tools) + } else if !bytes.Equal(prefix, req.Messages[0].Content) || !bytes.Equal(schemas, req.Tools) { + t.Error("main stable prefix/schema changed") + } + if n <= 3 { + args, _ := json.Marshal(map[string]any{"action": "call", "capability_id": "tool:read_file", "arguments": map[string]string{"path": "fixture.txt"}}) + payload, _ := json.Marshal(map[string]any{"choices": []any{map[string]any{"delta": map[string]any{"tool_calls": []any{map[string]any{"index": 0, "id": fmt.Sprintf("read-%d", serial.Add(1)), "type": "function", "function": map[string]any{"name": "use_capability", "arguments": string(args)}}}}, "finish_reason": "tool_calls"}}}) + _, _ = fmt.Fprintf(w, "data: %s\n\n", payload) + } else { + _, _ = io.WriteString(w, "data: {\"choices\":[{\"delta\":{\"content\":\"Complete.\"},\"finish_reason\":\"stop\"}]}\n\n") + } + } + if req.Model != "router-model" || !tc.missing { + _, _ = io.WriteString(w, "data: {\"choices\":[],\"usage\":{\"prompt_tokens\":1000,\"completion_tokens\":100,\"total_tokens\":1100}}\n\n") + } + _, _ = io.WriteString(w, "data: [DONE]\n\n") + })) + t.Cleanup(srv.Close) + planner := "" + if tc.dual { + planner = "planner_model = \"planner\"\n" + } + writeFile(t, root, "reasonix.toml", fmt.Sprintf(`default_model = "executor" +[agent] +%stask_token_budget = %d +subagent_models = { capability-router = "router" } +[codegraph] +enabled = false +[[providers]] +name = "executor" +kind = "openai" +model = "executor-model" +base_url = %q +api_key_env = "ROUTING_ROOT_EFFECT_KEY" +[[providers]] +name = "router" +kind = "openai" +model = "router-model" +base_url = %q +api_key_env = "ROUTING_ROOT_EFFECT_KEY" +[[providers]] +name = "planner" +kind = "openai" +model = "planner-model" +base_url = %q +api_key_env = "ROUTING_ROOT_EFFECT_KEY" +`, planner, tc.limit, srv.URL, srv.URL, srv.URL)) + t.Setenv("ROUTING_ROOT_EFFECT_KEY", "test") + approveWorkspaceAt(t, home, root) + preset := AgentPresetDelivery + if tc.balanced { + preset = AgentPresetBalanced + } + ctrl, err := Build(t.Context(), Options{Home: home, WorkspaceRoot: root, AgentPreset: preset, Sink: event.FuncSink(func(e event.Event) { + if e.Kind == event.ToolResult && e.Tool.Name == "use_capability" && e.Tool.Err == "" && strings.Contains(e.Tool.Output, "routing budget fixture") { + reads.Add(1) + } + if e.Kind == event.Usage && e.UsageSource == event.UsageSourceCapabilityRouter { + routedUsage.Add(1) + if e.Usage == nil || e.Usage.RequestCount != 1 || e.Usage.Estimated != tc.missing { + t.Errorf("router usage=%+v", e.Usage) + } + } + })}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(ctrl.Close) + ctrl.SetToolApprovalMode(control.ToolApprovalYolo) + ctx := t.Context() + if tc.disable { + ctx = agent.WithTaskBudget(ctx, agent.TaskBudget{}) + } + if tc.repeat != "" && tc.repeat != "new_task" { + ctx = agent.WithDeliveryExecutionScope(ctx, agent.DeliveryExecutionScope{ID: "routing-scope", TaskText: "fixture task"}) + } + input := "hello" + if tc.dual { + input = control.PlannerRouteMarker + " " + input + } + if tc.interactive { + err = ctrl.RunTurn(ctx, input) + } else { + err = ctrl.Run(ctx, input) + } + if tc.limit > 0 && !tc.disable { + if p, ok := agent.InspectRunPause(err); !ok || p.Key != "token" { + t.Fatalf("pause=%+v err=%v", p, err) + } + } else if err != nil { + t.Fatal(err) + } + wantPlans := int32(0) + if tc.dual { + wantPlans = 1 + } + wantRouters := int32(1) + if tc.balanced { + wantRouters = 0 + } + if routers.Load() != wantRouters || normals.Load() != tc.wantHTTP || reads.Load() != tc.wantReads || routedUsage.Load() != wantRouters || plans.Load() != wantPlans { + t.Fatalf("router/normal/read/usage/plan=%d/%d/%d/%d/%d want%d/%d/%d/%d/%d", routers.Load(), normals.Load(), reads.Load(), routedUsage.Load(), plans.Load(), wantRouters, tc.wantHTTP, tc.wantReads, wantRouters, wantPlans) + } + if tc.repeat != "" { + normals.Store(0) + reads.Store(0) + if tc.repeat == "reset" { + ctrl.Executor().ResetTaskBudget() + } + if tc.repeat == "new_scope" { + ctx = agent.WithDeliveryExecutionScope(ctx, agent.DeliveryExecutionScope{ID: "next-routing-scope", TaskText: "next fixture task"}) + } + err = ctrl.Run(ctx, input) + if p, ok := agent.InspectRunPause(err); !ok || p.Key != "token" { + t.Fatalf("second pause=%+v err=%v", p, err) + } + want := int32(2) + if tc.repeat == "continuation" { + want = 1 + } + if reads.Load() != want || normals.Load() != want+1 || routers.Load() != 1 || routedUsage.Load() != 1 { + t.Fatalf("cached second read/http/router/usage=%d/%d/%d/%d want%d/%d/1/1", reads.Load(), normals.Load(), routers.Load(), routedUsage.Load(), want, want+1) + } + } + }) + } +} diff --git a/internal/runtime/capability/semantic.go b/internal/runtime/capability/semantic.go index 44baff7c90..818debdc95 100644 --- a/internal/runtime/capability/semantic.go +++ b/internal/runtime/capability/semantic.go @@ -172,6 +172,7 @@ func (r *SemanticRouter) callModel(ctx context.Context, input string, candidates if usage == nil { return } + provider.ObserveAuxiliaryUsage(ctx, usage) if (usage.PromptTokens > 0 || usage.CompletionTokens > 0) && r.Audit != nil { // Host-side audit only: cost is estimated from the rate card and never // enters the model request. Prefer CostQuote when wiring new audit sinks. diff --git a/internal/runtime/capability/semantic_usage_test.go b/internal/runtime/capability/semantic_usage_test.go new file mode 100644 index 0000000000..a6415b061e --- /dev/null +++ b/internal/runtime/capability/semantic_usage_test.go @@ -0,0 +1,71 @@ +package capability + +import ( + "errors" + "testing" + + "reasonix/internal/contract/provider" +) + +func TestSemanticRootUsage(t *testing.T) { + for _, mode := range []string{"reported", "estimated", "nil_sink", "billed_error", "invalid_answer", "missing", "cache", "no_observer"} { + t.Run(mode, func(t *testing.T) { + u := &provider.Usage{PromptTokens: 1000, CompletionTokens: 100} + u.Estimated = mode == "estimated" + chunks := []provider.Chunk{{Type: provider.ChunkText, Text: `["skill:fixture"]`}} + if mode == "invalid_answer" { + chunks[0].Text = "invalid JSON" + } + if mode != "missing" { + chunks = append(chunks, provider.Chunk{Type: provider.ChunkUsage, Usage: u}) + } + if mode == "billed_error" { + chunks = append(chunks, provider.Chunk{Type: provider.ChunkError, Err: errors.New("fixture failure")}) + } + sink := &captureSink{} + r := &SemanticRouter{Provider: &fakeStreamProvider{chunks: chunks}, Sink: sink, Model: "fixture"} + if mode == "nil_sink" { + r.Sink = nil + } + ctx := t.Context() + calls := 0 + if mode != "no_observer" { + ctx = provider.WithAuxiliaryUsageObserver(ctx, func(usage *provider.Usage) { + calls++ + if usage.PromptTokens != 1000 || usage.CompletionTokens != 100 { + t.Errorf("observed usage=%+v", usage) + } + usage.PromptTokens = 9999 + }) + } + catalog := Catalog{Entries: []Entry{{ID: "skill:fixture", Kind: KindSkill, Name: "fixture", Description: "fixture skill", Status: StatusReady}}} + decision := r.RouteSemantic(ctx, "task", catalog, RouteDecision{}) + if mode == "cache" { + r.RouteSemantic(ctx, "task", catalog, RouteDecision{}) + } + want := 1 + if mode == "missing" || mode == "no_observer" { + want = 0 + } + if calls != want { + t.Fatalf("observer calls=%d want%d", calls, want) + } + if mode != "invalid_answer" && mode != "billed_error" && len(decision.Candidates) != 1 { + t.Fatalf("route=%+v", decision) + } + if mode == "invalid_answer" || mode == "billed_error" { + if len(decision.Candidates) != 0 { + t.Fatal("failed semantic answer changed deterministic decision") + } + } + if u.PromptTokens != 1000 { + t.Fatal("observer changed provider record") + } + for _, e := range sink.events { + if e.Usage != nil && e.Usage.PromptTokens != 1000 { + t.Fatal("observer changed public usage event") + } + } + }) + } +} diff --git a/internal/session/control/run_ceiling.go b/internal/session/control/run_ceiling.go index c1f711207d..865aedefd2 100644 --- a/internal/session/control/run_ceiling.go +++ b/internal/session/control/run_ceiling.go @@ -7,6 +7,15 @@ import ( "reasonix/internal/runtime/agent" ) +// prepareTurnRootBudget runs after turn admission and blocking hooks, but before +// semantic routing. The executor owns the scope later inherited by its planner. +func (c *Controller) prepareTurnRootBudget(ctx context.Context) context.Context { + if c.executor == nil { + return ctx + } + return c.executor.PrepareRootTaskBudget(ctx) +} + // bindTurnScope binds a Goal turn's spend budget and its usage recorder, the // latter active until the FSM commits. Neither chat nor Goal gets a round // ceiling: rounds carry no information the spend axes lack. An explicit diff --git a/internal/session/control/turn_entry.go b/internal/session/control/turn_entry.go index 1fe6a95e4c..fb48259edd 100644 --- a/internal/session/control/turn_entry.go +++ b/internal/session/control/turn_entry.go @@ -262,6 +262,7 @@ func (c *Controller) runReady(ctx context.Context, input string) (err error) { ctx, marker = c.beginTurn(ctx, startMessages, true) ctx = c.announceAuthoredTurn(ctx, rawInput, startMessages) ctx = c.withPlannerTurnMetadata(ctx, rawInput, false) + ctx = c.prepareTurnRootBudget(ctx) err = c.runWithRunner(ctx, c.withCapabilityRoute(ctx, input, rawInput)) return err } diff --git a/internal/session/control/turn_orchestrator.go b/internal/session/control/turn_orchestrator.go index 9c111cce8a..0de397328a 100644 --- a/internal/session/control/turn_orchestrator.go +++ b/internal/session/control/turn_orchestrator.go @@ -259,6 +259,7 @@ func (o *turnOrchestrator) runOrchestratedTurn(ctx context.Context, turn orchest // Goal turns bind a scope+epoch recorder for update_goal and observational // usage. It stays active through FSM/evaluator work; error paths clear it. ctx = c.bindTurnScope(ctx, continuation) + ctx = c.prepareTurnRootBudget(ctx) modelInput := input if !turn.synthetic { modelInput = c.withCapabilityRoute(ctx, input, turn.raw) diff --git a/release-notes/studio/unreleased/14025-controller-routing-root-budget.md b/release-notes/studio/unreleased/14025-controller-routing-root-budget.md new file mode 100644 index 0000000000..a9fcd0f884 --- /dev/null +++ b/release-notes/studio/unreleased/14025-controller-routing-root-budget.md @@ -0,0 +1,2 @@ +- 修复：Controller 在语义路由前绑定根预算，路由用量计入规划与执行共享软额度，保留缓存、新任务和续跑边界。关联 #14025。 +- Fixed: Controller binds root budgets before semantic routing, counting routing usage in shared soft limits while preserving cached decisions and task lifetimes. Refs #14025.