Merge pull request #14595 from SivanCola/codex/controller-root-routing-budget-14025

fix(control): include semantic routing in root budgets / 语义路由计入根预算

.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

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 隔离、恢复及最终组合验收仍未完成。

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)
+				}
+			}
+		})
+	}
+}

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.

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")
+				}
+			}
+		})
+	}
+}

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

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
 }

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)

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.