diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1ee7a9a99a..e2cd602a08 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -556,7 +556,7 @@ jobs: shell: bash run: | set -o pipefail - go test -count=1 -timeout=3m -json -run '^Test(RootToken|EffectRootToken)' ./internal/runtime/agent ./internal/assembly/boot | tee "$RUNNER_TEMP/root-token-budget.jsonl" + go test -count=1 -timeout=3m -json -run '^Test(RootToken|EffectRootToken|BudgetCountsKnownReportedTokens|ReportedToken|EffectReportedTotalTokenBudget)' ./internal/runtime/agent ./internal/assembly/boot | tee "$RUNNER_TEMP/root-token-budget.jsonl" jq -se ' [.[] | select(.Test == null and .Action == "pass") | .Package] as $packages | [.[] | select(.Test != null and .Action == "pass") | .Test] as $tests | @@ -573,6 +573,38 @@ jobs: "TestEffectRootTokenBudgetReachesDelegatedJobs/foreground_capability", "TestEffectRootTokenBudgetReachesDelegatedJobs/background", "TestEffectRootTokenBudgetReachesDelegatedJobs/background_capability", "TestEffectRootTokenBudgetReachesDelegatedJobs/background_fleet"] - $tests | length) == 0 ' "$RUNNER_TEMP/root-token-budget.jsonl" + jq -se ' + . as $events | + {"reasonix/internal/runtime/agent": [ + "TestBudgetCountsKnownReportedTokens", + "TestBudgetCountsKnownReportedTokens/split", + "TestBudgetCountsKnownReportedTokens/total_only", + "TestBudgetCountsKnownReportedTokens/partial_split", + "TestBudgetCountsKnownReportedTokens/lower_total", + "TestBudgetCountsKnownReportedTokens/missing_total", + "TestBudgetCountsKnownReportedTokens/negative_split", + "TestBudgetCountsKnownReportedTokens/zero", + "TestBudgetCountsKnownReportedTokens/negative", + "TestReportedTokensEnforceLocalCapOnInheritedRoot", + "TestReportedTokensFollowOriginalAuxiliaryRoot", + "TestReportedTokenCountersSaturate"], + "reasonix/internal/assembly/boot": [ + "TestEffectReportedTotalTokenBudget", + "TestEffectReportedTotalTokenBudget/split", + "TestEffectReportedTotalTokenBudget/total_only", + "TestEffectReportedTotalTokenBudget/partial_split", + "TestEffectReportedTotalTokenBudget/lower_total", + "TestEffectReportedTotalTokenBudget/missing_total", + "TestEffectReportedTotalTokenBudget/disabled"]} | + to_entries | all(.[]; . as $package | + ([ $events[] | select(.Package == $package.key and .Test == null and + (.Action == "pass" or .Action == "fail" or .Action == "skip")) ] | + length == 1 and .[0].Action == "pass") and + all(.value[]; . as $test | + [ $events[] | select(.Package == $package.key and .Test == $test and + (.Action == "pass" or .Action == "fail" or .Action == "skip")) ] | + length == 1 and .[0].Action == "pass")) + ' "$RUNNER_TEMP/root-token-budget.jsonl" - name: upload Windows root token budget inventory if: always() && env.RUN_STEPS == 'true' && runner.os == 'Windows' && github.event_name == 'pull_request' diff --git a/docs/REPORTED_TOTAL_TOKEN_BUDGET.md b/docs/REPORTED_TOTAL_TOKEN_BUDGET.md new file mode 100644 index 0000000000..8c3e05c29e --- /dev/null +++ b/docs/REPORTED_TOTAL_TOKEN_BUDGET.md @@ -0,0 +1,72 @@ +--- +owner: @SivanCola +backup: @esengine +status: active +reviewed: 2026-10-28 +--- + +# Reported totals and token budgets / 已报告总量与 token 预算 + +## Contract / 合同 + +A completion may report a positive total without an input/output breakdown. +The Agent's local and shared-root token budgets count the larger of that total +and the sum of nonnegative input/output counts. These overlapping measures are +never added together. Negative values contribute zero. + +一次完成可能报告正的总 token 数而没有输入/输出拆分。Agent 本地与共享根 token 预算 +统一取该总数和非负输入/输出之和中的较大值,不把两种重叠计数相加。负值按零计入。 + +Accumulation saturates instead of wrapping. A configured local role cap still +applies when the role inherits a larger root cap. Root usage belongs to the +original context: a late auxiliary request after a task reset charges its old +root, and the new root starts empty. + +累计计数饱和而不回绕。角色继承更大的根额度时,已配置的本地额度仍有效。根用量属于 +原上下文:任务重置后迟到的辅助请求计入旧根,新根从零开始。 + +No input/output split is invented. Provider usage, its existing normalization, +pricing, currencies, persisted data and wire formats are unchanged. The optional +RunBudget shadow sample continues to expose reported component counters; it does +not gain a total field. Budget pause details report the counted total. + +不补造输入/输出拆分。provider usage、已有规范化、计价、币种、持久化数据和 wire 格式 +不变。可选 RunBudget shadow 样本仍暴露已报告的分项计数,不新增 total 字段;预算暂停 +说明报告用于门禁的总数。 + +## Effect and limits / 效果与限制 + +A real OpenAI HTTP/SSE adapter and Boot fixture issue actual read_file calls. +With a 1500-token budget and 1100 reported tokens per completion, total-only and +complete-split records both pause after two reads. The existing finalization +request still runs. This is a synthetic wire fixture, not a live vendor incident. + +真实 OpenAI HTTP/SSE 解析器与 Boot 夹具执行实际 read_file。每次完成报告 1100 tokens、 +预算 1500 时,仅总数和完整拆分两种记录都在两次读取后暂停,原有收尾请求仍执行。 +这是合成 wire 夹具,不是线上供应商事故。 + +The budget remains an after-tool soft gate. In-flight requests and finalization +may exceed it; this adds no hard reservation, monetary root budget or durable +billing. Missing usage remains unknown. Zero configured limits remain unlimited. +Older binaries retain their previous handling of total-only records. + +Counters keep their existing in-process lifetime. This does not add +restart-persistent budget enforcement. + +预算仍是工具执行后的软门禁。进行中的请求和收尾可以越界;本次不新增硬预留、共享根 +金额预算或持久化计费。缺失 usage 仍为未知,配置零额度仍不限额。旧二进制保留原有的 +仅总数记录处理方式。 + +计数器保留原有的进程内生命周期,本次不新增跨重启的预算执行能力。 + +Provider request construction, tool schemas and cache-stable prefixes are +unchanged. A run may intentionally pause sooner when previously uncounted known +tokens reach its configured limit. + +Windows/Linux compilation does not establish +native runtime acceptance; the Windows root-token inventory requires all new +named regressions on the PR's own head. + +provider 请求构造、工具 schema 与稳定缓存前缀不变。过去漏计的已知 tokens 达到用户 +额度时,运行会按预期更早暂停。Windows/Linux 编译不代表原生运行验收;Windows 根 +token 清单要求在 PR 自己的 head 上通过全部新增命名回归。 diff --git a/internal/assembly/boot/effect_reported_total_budget_test.go b/internal/assembly/boot/effect_reported_total_budget_test.go new file mode 100644 index 0000000000..a5abfdbe23 --- /dev/null +++ b/internal/assembly/boot/effect_reported_total_budget_test.go @@ -0,0 +1,104 @@ +package boot + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + + "reasonix/internal/contract/event" + "reasonix/internal/contract/provider" + "reasonix/internal/model/openai" + "reasonix/internal/runtime/agent" + "reasonix/internal/session/control" +) + +func TestEffectReportedTotalTokenBudget(t *testing.T) { + for _, tc := range []struct { + name string + prompt, completion, total, budget int + wantReads int32 + }{ + {"split", 1000, 100, 1100, 1500, 2}, + {"total_only", 0, 0, 1100, 1500, 2}, + {"partial_split", 100, 0, 1100, 1500, 2}, + {"lower_total", 1000, 100, 10, 1500, 2}, + {"missing_total", 1000, 100, 0, 1500, 2}, + {"disabled", 0, 0, 1100, 0, 3}, + } { + t.Run(tc.name, func(t *testing.T) { + var requests, reads, seenUsage atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var request struct { + Tools []json.RawMessage `json:"tools"` + } + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + t.Error(err) + return + } + n := requests.Add(1) + delta := map[string]any{"content": "finished"} + finish := "stop" + if len(request.Tools) > 0 && n <= 3 { + delta = map[string]any{"tool_calls": []any{map[string]any{"index": 0, "id": fmt.Sprintf("read-%d", n), "type": "function", "function": map[string]any{"name": "read_file", "arguments": `{"path":"fixture.txt"}`}}}} + finish = "tool_calls" + } + u := map[string]any{"total_tokens": tc.total} + if tc.prompt != 0 || tc.completion != 0 { + u["prompt_tokens"] = tc.prompt + u["completion_tokens"] = tc.completion + } + payload, _ := json.Marshal(map[string]any{"choices": []any{map[string]any{"index": 0, "delta": delta, "finish_reason": finish}}, "usage": u}) + w.Header().Set("Content-Type", "text/event-stream") + _, _ = fmt.Fprintf(w, "data: %s\n\ndata: [DONE]\n\n", payload) + })) + defer srv.Close() + p, err := openai.New(provider.Config{Name: "total-wire", BaseURL: srv.URL, Model: "fixture", APIKey: "fixture", Extra: map[string]any{"protocol": "none"}}) + if err != nil { + t.Fatal(err) + } + kind := uniqueKind("total-usage-wire") + provider.Register(kind, func(provider.Config) (provider.Provider, error) { return p, nil }) + home, root := statedBootHome(t), t.TempDir() + writeFile(t, root, "fixture.txt", "total-wire-fixture") + writeFile(t, root, "reasonix.toml", fmt.Sprintf("default_model = \"fixture\"\n[agent]\ntask_token_budget = %d\n[codegraph]\nenabled = false\n[[providers]]\nname = \"fixture\"\nkind = %q\nmodel = \"fixture\"\n", tc.budget, kind)) + 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, "total-wire-fixture") { + reads.Add(1) + } + if e.Kind == event.Usage && e.Usage != nil { + total := max(tc.total, tc.prompt+tc.completion) + if e.Usage.TotalTokens != total || e.Usage.PromptTokens != tc.prompt || e.Usage.CompletionTokens != tc.completion { + t.Errorf("budget changed normalized wire usage: %+v", e.Usage) + } + seenUsage.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.SetToolApprovalMode(control.ToolApprovalYolo) + err = ctrl.Run(t.Context(), "read the fixture repeatedly") + t.Logf("reads=%d requests=%d observed_total_usage=%d run_error=%v", reads.Load(), requests.Load(), seenUsage.Load(), err) + if seenUsage.Load() < 2 { + t.Fatal("wire usage did not reach runtime") + } + if reads.Load() != tc.wantReads { + t.Fatalf("reads=%d, want %d", reads.Load(), tc.wantReads) + } + if tc.budget == 0 { + if err != nil { + t.Fatal(err) + } + } else if pause, ok := agent.InspectRunPause(err); !ok || pause.Key != "token" || !pause.HostOwned { + t.Fatalf("missing host-owned token pause: %v", err) + } + }) + } +} diff --git a/internal/runtime/agent/budget_tokens.go b/internal/runtime/agent/budget_tokens.go new file mode 100644 index 0000000000..c9156d0bed --- /dev/null +++ b/internal/runtime/agent/budget_tokens.go @@ -0,0 +1,20 @@ +package agent + +import "reasonix/internal/contract/provider" + +// A reported total and its input/output split overlap. Either can establish a +// larger known count, but adding them would charge the same tokens twice. +func budgetUsageTokens(usage *provider.Usage) uint64 { + if usage == nil { + return 0 + } + split := uint64(max(0, usage.PromptTokens)) + uint64(max(0, usage.CompletionTokens)) + return max(split, uint64(max(0, usage.TotalTokens))) +} + +func addBudgetTokens(used, delta uint64) uint64 { + if next := used + delta; next >= used { + return next + } + return ^uint64(0) +} diff --git a/internal/runtime/agent/reported_tokens_test.go b/internal/runtime/agent/reported_tokens_test.go new file mode 100644 index 0000000000..f1b36868e9 --- /dev/null +++ b/internal/runtime/agent/reported_tokens_test.go @@ -0,0 +1,120 @@ +package agent + +import ( + "context" + "sync/atomic" + "testing" + + "reasonix/internal/contract/event" + "reasonix/internal/contract/provider" + "reasonix/internal/contract/tool" + "reasonix/internal/state/sessionstore" +) + +func TestBudgetCountsKnownReportedTokens(t *testing.T) { + for _, tc := range []struct { + name string + usage provider.Usage + want uint64 + }{ + {"split", provider.Usage{PromptTokens: 1000, CompletionTokens: 100, TotalTokens: 1100}, 1100}, + {"total_only", provider.Usage{TotalTokens: 1100}, 1100}, + {"partial_split", provider.Usage{PromptTokens: 100, TotalTokens: 1100}, 1100}, + {"lower_total", provider.Usage{PromptTokens: 1000, CompletionTokens: 100, TotalTokens: 10}, 1100}, + {"missing_total", provider.Usage{PromptTokens: 1000, CompletionTokens: 100}, 1100}, + {"negative_split", provider.Usage{PromptTokens: -100, CompletionTokens: 100, TotalTokens: 1100}, 1100}, + {"zero", provider.Usage{}, 0}, + {"negative", provider.Usage{PromptTokens: -1, CompletionTokens: -1, TotalTokens: -1}, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + root := &rootTokenBudget{used: new(atomic.Uint64)} + before := tc.usage + var local runBudget + for range 2 { + root.observe(&tc.usage) + local.observe(&tc.usage, nil) + } + if got := root.used.Load(); got != 2*tc.want { + t.Fatalf("root tokens=%d, want %d", got, 2*tc.want) + } + axis, _ := local.exceeded(TaskBudget{Tokens: 1500}) + if (axis == "token") != (tc.want > 0) { + t.Fatalf("local axis=%q for usage %+v", axis, tc.usage) + } + if tc.usage != before { + t.Fatal("budget accounting rewrote provider usage") + } + if local.totals().PromptTokens != 2*before.PromptTokens || local.totals().OutputTokens != 2*before.CompletionTokens { + t.Fatal("budget accounting invented a token split") + } + }) + } +} + +type reportedTotalProvider struct{ spendingProvider } + +func (p *reportedTotalProvider) Stream(ctx context.Context, req provider.Request) (<-chan provider.Chunk, error) { + input, err := p.spendingProvider.Stream(ctx, req) + if err != nil { + return nil, err + } + out := make(chan provider.Chunk, 4) + for c := range input { + if c.Usage != nil { + c.Usage = &provider.Usage{TotalTokens: c.Usage.TotalTokens, RequestCount: 1} + } + out <- c + } + close(out) + return out, nil +} + +func TestReportedTokensEnforceLocalCapOnInheritedRoot(t *testing.T) { + parent := New(&mockProvider{}, tool.NewRegistry(), sessionstore.NewSession("parent"), Options{TaskBudget: TaskBudget{Tokens: 100000}}, event.Discard) + ctx := InheritRootTokenBudget(t.Context(), parent.PrepareRootTaskBudget(t.Context())) + reg := tool.NewRegistry() + reg.Add(readProbe{}) + p := &reportedTotalProvider{spendingProvider: spendingProvider{max: 3}} + child := New(p, reg, sessionstore.NewSession("child"), Options{TaskBudget: TaskBudget{Tokens: 1500}}, event.Discard) + err := child.Run(ctx, "read fixtures") + pause, ok := InspectRunPause(err) + if !ok || pause.Key != "token" || !pause.HostOwned { + t.Fatalf("local cap did not pause: %v", err) + } + if p.rounds.Load() != 3 { + t.Fatalf("requests=%d, want two tool rounds and finalization", p.rounds.Load()) + } +} + +func TestReportedTokensFollowOriginalAuxiliaryRoot(t *testing.T) { + a := New(&mockProvider{}, tool.NewRegistry(), sessionstore.NewSession("root"), Options{TaskBudget: TaskBudget{Tokens: 1500}}, event.Discard) + first := a.PrepareRootTaskBudget(t.Context()) + old := first.Value(rootTokenBudgetKey{}).(*rootTokenBudget) + provider.ObserveAuxiliaryUsage(first, &provider.Usage{TotalTokens: 1100}) + a.ResetTaskBudget() + second := a.PrepareRootTaskBudget(t.Context()) + current := second.Value(rootTokenBudgetKey{}).(*rootTokenBudget) + late := InheritRootTokenBudget(context.Background(), first) + provider.ObserveAuxiliaryUsage(late, &provider.Usage{TotalTokens: 1100}) + if old.used.Load() != 2200 || current.used.Load() != 0 { + t.Fatalf("old=%d new=%d", old.used.Load(), current.used.Load()) + } +} + +func TestReportedTokenCountersSaturate(t *testing.T) { + maximum := int(^uint(0) >> 1) + u := &provider.Usage{PromptTokens: maximum, CompletionTokens: maximum, TotalTokens: maximum} + root := &rootTokenBudget{used: new(atomic.Uint64)} + root.used.Store(^uint64(0) - 5) + var local runBudget + for range 3 { + root.observe(u) + local.observe(u, nil) + } + if root.used.Load() != ^uint64(0) { + t.Fatalf("root did not saturate: %d", root.used.Load()) + } + if axis, _ := local.exceeded(TaskBudget{Tokens: maximum}); axis != "token" { + t.Fatal("overflow reopened local token budget") + } +} diff --git a/internal/runtime/agent/root_token_budget.go b/internal/runtime/agent/root_token_budget.go index e6e7169939..838826da3e 100644 --- a/internal/runtime/agent/root_token_budget.go +++ b/internal/runtime/agent/root_token_budget.go @@ -39,7 +39,7 @@ func (a *Agent) bindRootTokenBudget(ctx context.Context, state *turnRuntime) con previous := a.task.budget.rootTokens if previous == nil { previous = &rootTokenBudget{used: new(atomic.Uint64)} - previous.observe(&provider.Usage{PromptTokens: a.task.budget.promptTokens, CompletionTokens: a.task.budget.outputTokens}) + previous.used.Store(a.task.budget.tokens) } started := a.task.budget.started if started.IsZero() { @@ -60,13 +60,10 @@ func (b *rootTokenBudget) observe(usage *provider.Usage) { if b == nil || usage == nil { return } - delta := uint64(max(0, usage.PromptTokens)) + uint64(max(0, usage.CompletionTokens)) + delta := budgetUsageTokens(usage) for { old := b.used.Load() - next := old + delta - if next < old { - next = ^uint64(0) - } + next := addBudgetTokens(old, delta) if b.used.CompareAndSwap(old, next) { return } diff --git a/internal/runtime/agent/run_budget.go b/internal/runtime/agent/run_budget.go index e21edf2674..236208f7a3 100644 --- a/internal/runtime/agent/run_budget.go +++ b/internal/runtime/agent/run_budget.go @@ -46,6 +46,7 @@ type runBudget struct { requests int promptTokens int outputTokens int + tokens uint64 cost runCost // limit is configuration, not accumulation: it survives the reset that // starts a new task. @@ -59,6 +60,7 @@ type runBudget struct { // reads cheaper than the turn actually was. func (b *runBudget) observe(usage *provider.Usage, quote *pricing.CostQuote) { b.rounds++ + b.tokens = addBudgetTokens(b.tokens, budgetUsageTokens(usage)) b.cost.observe(usage, quote) if usage == nil { return @@ -93,7 +95,7 @@ func (b *runBudget) totals() event.RunBudgetTotals { // usage or prices; unknown spend cannot erase a proven crossing. func (b *runBudget) exceeded(limit TaskBudget) (axis, detail string) { if limit.Tokens > 0 { - if used := b.promptTokens + b.outputTokens; used >= limit.Tokens { + if used := b.tokens; used >= uint64(limit.Tokens) { return "token", fmt.Sprintf("task used %d tokens, reaching the %d budget", used, limit.Tokens) } } diff --git a/release-notes/studio/unreleased/14025-reported-total-token-budget.md b/release-notes/studio/unreleased/14025-reported-total-token-budget.md new file mode 100644 index 0000000000..788f001a95 --- /dev/null +++ b/release-notes/studio/unreleased/14025-reported-total-token-budget.md @@ -0,0 +1,9 @@ +## 修复 + +- 本地与共享根 token 预算现在计入仅总数或不完整拆分的已知用量,不重复累计重叠字段;保留零额度不限额和原任务归属。 (#14025) + + + +## Fixed + +- Count known total-only or partially split usage in local and shared-root token budgets without double counting; preserve unlimited defaults and original-task attribution. (#14025)