diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 61d6a68ff4..d326e583a9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -391,6 +391,38 @@ jobs: if-no-files-found: ignore retention-days: 7 + + - name: test (Windows root auxiliary 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(RootAuxiliary|EffectRootAuxiliary)' ./internal/runtime/agent ./internal/assembly/boot | tee "$RUNNER_TEMP/root-auxiliary-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/agent", "reasonix/internal/assembly/boot"] - $packages | length) == 0 and + (["TestRootAuxiliaryAutomaticCompaction", "TestRootAuxiliaryTokenBudget", + "TestRootAuxiliaryTokenBudget/summary", "TestRootAuxiliaryTokenBudget/summary_failed", + "TestRootAuxiliaryTokenBudget/summary_missing", "TestRootAuxiliaryTokenBudget/summary_disabled", + "TestRootAuxiliaryTokenBudget/diagnosis", "TestRootAuxiliaryTokenBudget/diagnosis_failed", + "TestRootAuxiliaryTokenBudget/diagnosis_missing", "TestRootAuxiliaryTokenBudget/diagnosis_disabled", + "TestRootAuxiliaryManualCompactionOutsideRun", "TestRootAuxiliaryBudgetProjectionKeepsOldTask", + "TestEffectRootAuxiliarySummaryBudget", "TestEffectRootAuxiliarySummaryBudget/configured", + "TestEffectRootAuxiliarySummaryBudget/disabled", "TestEffectRootAuxiliarySummaryBudget/missing_usage"] - $tests | length) == 0 + ' "$RUNNER_TEMP/root-auxiliary-budget.jsonl" + + - name: upload Windows root auxiliary budget inventory + if: always() && env.RUN_STEPS == 'true' && runner.os == 'Windows' && github.event_name == 'pull_request' + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7 + with: + name: root-auxiliary-budget-windows-${{ github.run_attempt }} + path: ${{ runner.temp }}/root-auxiliary-budget.jsonl + if-no-files-found: ignore + retention-days: 7 + - name: test (Windows estimated usage quotes) if: env.RUN_STEPS == 'true' && runner.os == 'Windows' && github.event_name == 'pull_request' timeout-minutes: 5 diff --git a/docs/ROOT_AUXILIARY_TOKEN_BUDGET.md b/docs/ROOT_AUXILIARY_TOKEN_BUDGET.md new file mode 100644 index 0000000000..0c2b819f15 --- /dev/null +++ b/docs/ROOT_AUXILIARY_TOKEN_BUDGET.md @@ -0,0 +1,86 @@ +--- +owner: @SivanCola +backup: @esengine +status: active +reviewed: 2026-10-26 +--- + +# Auxiliary usage in root token budgets / 根 token 预算中的辅助用量 + +Summary and failure-diagnosis requests made inside an active Agent root context +now charge that context's existing token counter. Their usage events and +compaction receipts were already recorded; the root counter omitted them. + +在活跃 Agent 根 context 内执行的摘要与失败诊断请求，现在计入该 context 既有 +token counter。此前用量事件与压缩收据已有记录，但根计数漏掉了这些调用。 + +## Accounting boundary / 计账边界 + +Each auxiliary terminal folds its usage once, beside the existing usage-event +owner. Positive prompt and completion counts use the same saturating counter as +normal rounds, including positive estimates under that existing rule. Missing +counts remain unknown; request counts do not imply token counts. + +每个辅助调用终态在既有用量事件 owner 处仅计一次。正 prompt/completion 计数使用 +与正常轮次相同的饱和 counter，包括该既有规则下的正估算值。缺失计数仍未知， +请求次数不能推导为 token 数。 + +Usage received before a summary or diagnosis fails still counts, even when its +answer is discarded. Provider retries retain their existing aggregate-usage +semantics. No extra charge is generated from a compaction receipt or ledger +replay. Normal-round local telemetry remains separate and unchanged. + +摘要或诊断失败前已收到的用量仍计入，即使答案被舍弃。供应商重试保留既有聚合 +用量语义，不通过压缩收据或账本重放再次扣计。正常轮次的局部遥测保持独立且不变。 + +The scope comes from the request context, never from the Agent's current task +field. A budget-only projected job context retains its original scope after a +new task starts. Manual maintenance without a root context still reports its +usage but does not create or charge an unrelated root task. + +scope 来自请求 context，不读取 Agent 当前 task 字段。仅投射预算的 job context +在新任务开始后仍保留原 scope。没有根 context 的手动维护仍上报用量，但不会 +创建或扣计无关的根任务。 + +## Cooperative limits and compatibility / 协作式阈值与兼容 + +The explicit token gate remains cooperative after a tool round, with the +existing finalization. An auxiliary request can cross the threshold before a +normal request or tool runs; this change does not reserve or reject in advance. +Zero limits still leave execution unbounded. + +显式 token 门禁仍在工具轮后协作式检查，并保留既有收尾。辅助请求可能在正常 +请求或工具执行前已经越线；本改动不做提前预留或拒绝。零阈值仍不限执行。 + +No public API, configuration, persisted format, wire, permission or default is +added. Compaction timing, request construction, schemas and prefix reuse are +unchanged. Cost completeness and currency ledgers retain their existing rules. + +没有新增公共 API、配置、持久化格式、wire、权限或默认值。压缩时机、请求构造、 +schema 和前缀复用保持不变。费用完整性与币种账本沿用既有规则。 + +## Evidence boundary / 验证边界 + +Agent regressions exercise automatic pressure compaction, manual maintenance +and protected-failure diagnosis inside a real root Run. They cover billed +failure, missing usage, disabled limits and original-scope projection after a +new task. The late-scope fixture is constructed, not a production job incident. + +Agent 回归通过真实根 Run 覆盖自动压力压缩、手动维护与受保护失败诊断，并检查 +有用量的失败、缺失用量、禁用阈值及新任务后的原 scope 投射。延后 scope 夹具是 +构造控制，不是生产 job 事故复现。 + +Real boot HTTP/SSE fixtures start with existing public transcript history, +trigger automatic compaction and verify actual file reads and usage provenance. +Existing cache-prefix regressions remain required; their mock results do not +measure a live provider's cache hit rate. + +真实 boot HTTP/SSE 夹具通过既有公共 transcript 历史触发自动压缩，验证实际文件 +读取和用量来源。既有缓存前缀回归仍须通过；mock 结果不是供应商命中率实测。 + +This is not complete auxiliary-provider admission, a shared Coordinator root, +currency aggregation, hard reservation, durable accounting or physical process +cancellation. See [root token budgets](ROOT_TOKEN_BUDGET_SECURITY.md). + +本改动不等于完整辅助 provider 准入、共享 Coordinator 根预算、币种汇总、硬预留、 +持久计账或物理进程取消。作者分析不等于独立 review 或原生平台执行。 diff --git a/internal/assembly/boot/effect_root_auxiliary_budget_test.go b/internal/assembly/boot/effect_root_auxiliary_budget_test.go new file mode 100644 index 0000000000..3d4dd43e21 --- /dev/null +++ b/internal/assembly/boot/effect_root_auxiliary_budget_test.go @@ -0,0 +1,116 @@ +package boot + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + + "reasonix/internal/contract/event" + "reasonix/internal/contract/provider" + "reasonix/internal/runtime/agent" + "reasonix/internal/session/control" + "reasonix/internal/state/sessionstore" +) + +func TestEffectRootAuxiliarySummaryBudget(t *testing.T) { + for _, tc := range []struct { + name string + limit int + missing bool + wantReads int32 + }{ + {name: "configured", limit: 1500, wantReads: 1}, + {name: "disabled", wantReads: 4}, + {name: "missing_usage", limit: 1500, missing: true, wantReads: 2}, + } { + t.Run(tc.name, func(t *testing.T) { + home, root := statedBootHome(t), t.TempDir() + writeFile(t, root, "fixture.txt", "actual root auxiliary read") + var normal, summaries, reads, summaryEvents atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req struct { + Tools json.RawMessage `json:"tools"` + } + 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") + summary := len(req.Tools) == 0 || string(req.Tools) == "null" + if summary { + summaries.Add(1) + _, _ = io.WriteString(w, "data: {\"choices\":[{\"delta\":{\"content\":\"## Standing facts\\n- preserve public APIs and read fixture.txt\"},\"finish_reason\":\"stop\"}]}\n\n") + } else if n := normal.Add(1); n <= 4 { + _, _ = fmt.Fprintf(w, "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"read-%d\",\"type\":\"function\",\"function\":{\"name\":\"read_file\",\"arguments\":\"{\\\"path\\\":\\\"fixture.txt\\\"}\"}}]},\"finish_reason\":\"tool_calls\"}]}\n\n", n) + } else { + _, _ = io.WriteString(w, "data: {\"choices\":[{\"delta\":{\"content\":\"Finished.\"},\"finish_reason\":\"stop\"}]}\n\n") + } + if !summary || !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) + writeFile(t, root, "reasonix.toml", fmt.Sprintf(`default_model = "test" +[agent] +system_prompt = "BASE" +compact_ratio = 0.5 +recent_keep = 2 +task_token_budget = %d +[codegraph] +enabled = false +[[providers]] +name = "test" +kind = "openai" +model = "test" +context_window = 96000 +max_output_tokens = 1024 +base_url = %q +api_key_env = "ROOT_AUXILIARY_KEY" +`, tc.limit, srv.URL)) + t.Setenv("ROOT_AUXILIARY_KEY", "test") + approveWorkspaceAt(t, home, root) + sink := event.FuncSink(func(e event.Event) { + if e.Kind == event.ToolResult && e.Tool.Name == "read_file" && e.Tool.Err == "" && strings.Contains(e.Tool.Output, "actual root auxiliary read") { + reads.Add(1) + } + if e.Kind == event.Usage && e.UsageSource == event.UsageSourceCompaction && e.Usage != nil { + if e.Usage.Estimated != tc.missing || e.Usage.RequestCount != 1 { + t.Errorf("summary usage provenance=%+v, missing=%v", e.Usage, tc.missing) + } + summaryEvents.Add(1) + } + }) + ctrl, err := Build(t.Context(), Options{Home: home, WorkspaceRoot: root, Sink: sink}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(ctrl.Close) + ctrl.SetSessionPath(sessionstore.NewSessionPath(ctrl.SessionDir(), ctrl.Label())) + ctrl.SetToolApprovalMode(control.ToolApprovalYolo) + // Shape existing public transcript history, rather than assigning a + // private clock or root counter to manufacture the budget crossing. + for range 40 { + ctrl.Executor().Session().Add(provider.Message{Role: provider.RoleAssistant, Content: strings.Repeat("history word ", 1200)}) + ctrl.Executor().Session().Add(provider.Message{Role: provider.RoleUser, Content: "continue"}) + } + err = ctrl.Run(t.Context(), "read fixture.txt") + if tc.limit == 0 { + if err != nil { + t.Fatal(err) + } + } else if pause, ok := agent.InspectRunPause(err); !ok || pause.Key != "token" { + t.Fatalf("pause=%+v err=%v", pause, err) + } + if summaries.Load() != 1 || summaryEvents.Load() != 1 || reads.Load() != tc.wantReads || normal.Load() != tc.wantReads+1 { + t.Fatalf("summary HTTP/events=%d/%d normal HTTP/actual reads=%d/%d, want1/1/%d/%d", summaries.Load(), summaryEvents.Load(), normal.Load(), reads.Load(), tc.wantReads+1, tc.wantReads) + } + }) + } +} diff --git a/internal/runtime/agent/compact.go b/internal/runtime/agent/compact.go index da069160e8..58baf5ca0b 100644 --- a/internal/runtime/agent/compact.go +++ b/internal/runtime/agent/compact.go @@ -639,6 +639,7 @@ func (a *contextWindow) summarizeAtMode(ctx context.Context, region []provider.M }() defer func() { usage = provider.UsageWithRequestAttemptCount(ctx, usage) + observeRootTokenUsage(ctx, usage) if usage != nil && (usage.TotalTokens > 0 || usage.RequestCount > 0) { // Recorded beside the emit, not at the call site that keeps the // answer: a repair whose digest is discarded was still charged. diff --git a/internal/runtime/agent/failure_diagnosis.go b/internal/runtime/agent/failure_diagnosis.go index 2dd141d091..51ea5cfe52 100644 --- a/internal/runtime/agent/failure_diagnosis.go +++ b/internal/runtime/agent/failure_diagnosis.go @@ -95,6 +95,7 @@ func (a *Agent) selectDiagnosticLines(ctx context.Context, content string) ([]in var usage *provider.Usage defer func() { usage = provider.UsageWithRequestAttemptCount(ctx, usage) + observeRootTokenUsage(ctx, usage) if usage != nil && (usage.TotalTokens > 0 || usage.RequestCount > 0) { a.svc.sink.Emit(event.Event{Kind: event.Usage, ModelRef: a.modelRef, Usage: usage, Pricing: a.svc.pricing, UsageSource: event.UsageSourceCompaction}) diff --git a/internal/runtime/agent/root_auxiliary_budget_test.go b/internal/runtime/agent/root_auxiliary_budget_test.go new file mode 100644 index 0000000000..ab95981060 --- /dev/null +++ b/internal/runtime/agent/root_auxiliary_budget_test.go @@ -0,0 +1,191 @@ +package agent + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + + "reasonix/internal/contract/event" + "reasonix/internal/contract/provider" + "reasonix/internal/contract/tool" + "reasonix/internal/state/sessionstore" +) + +type rootMaintenanceTool struct { + readProbe + child *Agent + diagnosis bool + calls int +} + +type rootAutomaticMaintenanceProvider struct { + normal spendingProvider + summary scriptedSummarizer +} + +func (p *rootAutomaticMaintenanceProvider) Name() string { return "root-automatic-maintenance" } + +func (p *rootAutomaticMaintenanceProvider) Stream(ctx context.Context, req provider.Request) (<-chan provider.Chunk, error) { + if len(req.Tools) == 0 { + return p.summary.Stream(ctx, req) + } + return p.normal.Stream(ctx, req) +} + +func TestRootAuxiliaryAutomaticCompaction(t *testing.T) { + p := &rootAutomaticMaintenanceProvider{normal: spendingProvider{max: 4}, summary: scriptedSummarizer{replies: []scriptedReply{{text: "## Standing facts\n- never change the public API", usage: billed(1000, 100)}}}} + reg := tool.NewRegistry() + reg.Add(readProbe{}) + a := New(p, reg, foldableSessionOverForce(30), Options{ContextWindow: 5000, CompactRatio: 0.5, RecentKeep: 2, ArchiveDir: t.TempDir(), TaskBudget: TaskBudget{Tokens: 1500}}, event.Discard) + err := a.Run(t.Context(), "read a few files") + if pause, ok := InspectRunPause(err); !ok || pause.Key != "token" { + t.Fatalf("pause=%+v err=%v, want token landing", pause, err) + } + if p.summary.calls != 1 || p.normal.rounds.Load() != 2 { + t.Fatalf("summary requests=%d normal requests=%d, want 1/2", p.summary.calls, p.normal.rounds.Load()) + } + if got := a.task.budget.rootTokens.used.Load(); got != 3300 { + t.Fatalf("root tokens=%d, want summary and two normal reported usages=3300", got) + } +} + +func (p *rootMaintenanceTool) Execute(ctx context.Context, _ json.RawMessage) (string, error) { + p.calls++ + if p.diagnosis { + code := 1 + region := []provider.Message{{Role: provider.RoleTool, Content: strings.Join(padded("failure line ", 40), "\n"), ToolExecution: &provider.ToolExecution{State: tool.ShellStateFailed, ExitCode: &code}}} + p.child.annotateFailureDiagnostics(ctx, region, nil) + } else { + _, _ = p.child.CompactNow(ctx, CompactRequest{IgnoreEconomics: true}) + } + return "maintenance attempted", nil +} + +func TestRootAuxiliaryTokenBudget(t *testing.T) { + for _, tc := range []struct { + name string + diagnosis bool + fail bool + missing bool + disabled bool + wantReads int + }{ + {name: "summary", wantReads: 1}, + {name: "summary_failed", fail: true, wantReads: 1}, + {name: "summary_missing", missing: true, wantReads: 2}, + {name: "summary_disabled", disabled: true, wantReads: 4}, + {name: "diagnosis", diagnosis: true, wantReads: 1}, + {name: "diagnosis_failed", diagnosis: true, fail: true, wantReads: 1}, + {name: "diagnosis_missing", diagnosis: true, missing: true, wantReads: 2}, + {name: "diagnosis_disabled", diagnosis: true, disabled: true, wantReads: 4}, + } { + t.Run(tc.name, func(t *testing.T) { + u := billed(1000, 100) + if tc.missing { + u = nil + } + var failure error + if tc.fail { + failure = errors.New("billed stream failed") + } + replies := make([]scriptedReply, 4) + for i := range replies { + replies[i] = scriptedReply{text: "## Standing facts\n- keep the public API", usage: u, fail: failure} + } + if tc.diagnosis { + for i := range replies { + replies[i].text = "1-3" + } + } + sink := &recordSink{} + child := New(&scriptedSummarizer{replies: replies}, tool.NewRegistry(), foldableSessionOverForce(30), Options{ContextWindow: 200000, RecentKeep: 2}, sink) + probe := &rootMaintenanceTool{child: child, diagnosis: tc.diagnosis} + reg := tool.NewRegistry() + reg.Add(probe) + prov := &spendingProvider{max: 4} + root := New(prov, reg, sessionstore.NewSession("root"), Options{TaskBudget: TaskBudget{Tokens: 1500}}, sink) + ctx := t.Context() + if tc.disabled { + ctx = WithTaskBudget(ctx, TaskBudget{}) + } + err := root.Run(ctx, "run maintenance") + if tc.disabled { + if err != nil { + t.Fatal(err) + } + } else if pause, ok := InspectRunPause(err); !ok || pause.Key != "token" { + t.Fatalf("pause=%+v err=%v, want token landing", pause, err) + } + if probe.calls != tc.wantReads || prov.rounds.Load() != int32(tc.wantReads+1) { + t.Fatalf("tool calls=%d provider rounds=%d, want %d/%d", probe.calls, prov.rounds.Load(), tc.wantReads, tc.wantReads+1) + } + _, input, output := ledgerTotal(sink) + want := uint64(prov.rounds.Load())*1100 + uint64(input+output) + if root.task.budget.promptTokens+root.task.budget.outputTokens != int(prov.rounds.Load())*1100 { + t.Fatal("auxiliary usage changed local normal-round telemetry") + } + if got := root.task.budget.rootTokens.used.Load(); got != want { + t.Fatalf("root tokens=%d, want normal plus auxiliary emitted usage=%d", got, want) + } + if !tc.missing && input+output == 0 { + t.Fatal("fixture did not bill auxiliary usage") + } + }) + } +} + +func TestRootAuxiliaryManualCompactionOutsideRun(t *testing.T) { + sink := &recordSink{} + a := New(&scriptedSummarizer{replies: []scriptedReply{{text: "## Standing facts\n- keep public APIs", usage: billed(1000, 100)}}}, tool.NewRegistry(), foldableSessionOverForce(30), Options{ContextWindow: 200000, RecentKeep: 2}, sink) + if _, err := a.CompactNow(t.Context(), CompactRequest{IgnoreEconomics: true}); err != nil { + t.Fatal(err) + } + if a.task.budget.rootTokens != nil { + t.Fatal("manual maintenance created an unrelated root scope") + } + if calls, input, output := ledgerTotal(sink); calls != 1 || input+output != 1100 { + t.Fatalf("maintenance bill=%d/%d/%d", calls, input, output) + } +} + +type rootMaintenanceCapture struct { + readProbe + projected context.Context +} + +func (p *rootMaintenanceCapture) Execute(ctx context.Context, _ json.RawMessage) (string, error) { + p.projected = InheritRootTokenBudget(context.Background(), ctx) + return "captured budget scope", nil +} + +func TestRootAuxiliaryBudgetProjectionKeepsOldTask(t *testing.T) { + reg := tool.NewRegistry() + capture := &rootMaintenanceCapture{} + reg.Add(capture) + root := New(&spendingProvider{max: 1}, reg, sessionstore.NewSession("root"), Options{}, event.Discard) + ctx, cancel := context.WithCancel(t.Context()) + t.Cleanup(cancel) + if err := root.Run(ctx, "capture root scope"); err != nil { + t.Fatal(err) + } + old := root.task.budget.rootTokens + before := old.used.Load() + cancel() + if err := root.Run(t.Context(), "new task"); err != nil { + t.Fatal(err) + } + current := root.task.budget.rootTokens + if current == old { + t.Fatal("new task kept old scope") + } + currentBefore := current.used.Load() + child := New(&scriptedSummarizer{replies: []scriptedReply{{text: "## Standing facts\n- keep public APIs", usage: billed(1000, 100)}}}, tool.NewRegistry(), foldableSessionOverForce(30), Options{ContextWindow: 200000, RecentKeep: 2}, event.Discard) + if _, err := child.CompactNow(capture.projected, CompactRequest{IgnoreEconomics: true}); err != nil { + t.Fatal(err) + } + if old.used.Load() != before+1100 || current.used.Load() != currentBefore { + t.Fatalf("late maintenance charged wrong scope: old=%d/%d new=%d/%d", old.used.Load(), before, current.used.Load(), currentBefore) + } +} diff --git a/internal/runtime/agent/root_token_budget.go b/internal/runtime/agent/root_token_budget.go index 2f85f3dc14..ae5baa3541 100644 --- a/internal/runtime/agent/root_token_budget.go +++ b/internal/runtime/agent/root_token_budget.go @@ -25,6 +25,13 @@ func InheritRootTokenBudget(ctx, parent context.Context) context.Context { return ctx } +// Auxiliary requests share the active context's scope. Using that scope rather +// than Agent task state also keeps late usage attached to its original task. +func observeRootTokenUsage(ctx context.Context, usage *provider.Usage) { + scope, _ := ctx.Value(rootTokenBudgetKey{}).(*rootTokenBudget) + scope.observe(usage) +} + func (a *Agent) bindRootTokenBudget(ctx context.Context, state *turnRuntime) context.Context { scope, inherited := ctx.Value(rootTokenBudgetKey{}).(*rootTokenBudget) if !inherited { diff --git a/release-notes/studio/unreleased/14025-root-auxiliary-token-usage.md b/release-notes/studio/unreleased/14025-root-auxiliary-token-usage.md new file mode 100644 index 0000000000..e7fb4afc30 --- /dev/null +++ b/release-notes/studio/unreleased/14025-root-auxiliary-token-usage.md @@ -0,0 +1,2 @@ +- 修复：活跃 Agent 根任务中的摘要与失败诊断用量现在计入既有 token 软预算，保留缺失用量、零阈值、局部遥测和无根 context 手动维护的行为。关联 #14025。 +- Fixed: active-root summary and failure-diagnosis usage now counts toward its token soft budget. Missing usage, zero limits and local telemetry retain their behavior. Refs #14025.