fix(search): charge native search usage to the active root budget

Problem: an explicitly bounded Agent root omitted known native search usage,
allowing an extra normal tool round after a reported search answer.

Root cause: the search terminal finalized its independent request count and
reported usage without calling the accepted optional root observer.

Fix: observe the finalized record once before optional reporting. Preserve
query isolation, request counts, callback copies, unknown usage, native
completion, paid-backend consent and zero-limit behavior.

Verification: all 239 root packages pass through disjoint commands; four
complete owner race suites and real boot OpenAI/Anthropic HTTP regressions
pass. The compiled preceding owner fails seven events while nine controls
pass. Static gates, six cross-compiles and the literal 16-event Windows
contract pass locally; own native execution remains a hosted gate.
This commit is contained in:
SivanCola committed 2026-10-26 22:38:38 +08:00
1 parent e07e0cb335
commit 2fa7ad62b1
6 files changed
+420 -1

No files matched your search

+31
View File
@@ -421,6 +421,37 @@ jobs:
retention-days: 7
- name: test (Windows root websearch 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(RootWebSearch|EffectRootWebSearch)' ./internal/tools/websearch ./internal/assembly/boot | tee "$RUNNER_TEMP/root-websearch-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/tools/websearch", "reasonix/internal/assembly/boot"] - $packages | length) == 0 and
(["TestRootWebSearchUsageTerminal", "TestRootWebSearchUsageTerminal/reported",
"TestRootWebSearchUsageTerminal/no_reporter", "TestRootWebSearchUsageTerminal/missing",
"TestRootWebSearchUsageTerminal/no_observer", "TestRootWebSearchUsageTerminal/input_refused",
"TestRootWebSearchUsageTerminal/no_factory", "TestRootWebSearchUsageTerminal/factory_error",
"TestRootWebSearchUsageTerminal/billed_failure", "TestRootWebSearchUsageTerminal/cancel_after_usage",
"TestRootWebSearchUsageTerminal/cancel_before_start", "TestRootWebSearchUsageTerminal/stream_error",
"TestEffectRootWebSearchTokenBudget", "TestEffectRootWebSearchTokenBudget/reported",
"TestEffectRootWebSearchTokenBudget/disabled", "TestEffectRootWebSearchTokenBudget/missing"] - $tests | length) == 0
' "$RUNNER_TEMP/root-websearch-budget.jsonl"
- name: upload Windows root websearch 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-websearch-budget-windows-${{ github.run_attempt }}
path: ${{ runner.temp }}/root-websearch-budget.jsonl
if-no-files-found: ignore
retention-days: 7
- name: test (Windows root triage budget)
if: env.RUN_STEPS == 'true' && runner.os == 'Windows' && github.event_name == 'pull_request'
timeout-minutes: 5
+85
View File
@@ -0,0 +1,85 @@
---
owner: @SivanCola
backup: @esengine
status: active
reviewed: 2026-10-26
---
# Native search root tokens / 原生搜索根 token 计账
Native provider-backed `web_search` now reports finalized usage to the optional
active-root observer. It uses the accepted provider contract; Agent root
binding owns the scope and counter, while the search terminal owns one charge.
原生 provider-backed `web_search` 现在将终态用量上报给可选活跃根 observer。
它复用已接受的 provider 契约;Agent 根绑定拥有 scope/counter,搜索终态仅扣计一次。
## Accounting owner / 计账所有权
Observation follows the existing independent HTTP request counter and precedes
the optional usage reporter. A missing reporter does not hide known token
usage. The observer receives a copy, so it cannot alter the reporter's record.
观察发生在既有独立 HTTP 请求计数之后、可选 usage reporter 之前。没有 reporter
不隐藏已知 token;observer 收到副本,不能改写 reporter 的记录。
Known usage received before a stream error or cancellation still counts once.
Input refusal, pre-cancellation, unavailable factories and missing token usage
do not invent charges. Request-only records preserve exact HTTP counts and
their estimated identity; request counts never imply tokens.
流错误或取消前收到的已知用量仍仅计一次。输入拒绝、预先取消、factory 不可用
和缺失 token 用量不推造扣计。仅请求记录保留准确 HTTP 次数与 estimated 身份,
不能把请求次数当作 token。
Positive prompt/completion counts follow the existing conservative root rule,
including positive estimates. Normal-round local telemetry, model/source/
pricing emission and reporter behavior remain unchanged. Outside an observer
context, the tool keeps its existing behavior.
正 prompt/completion 沿用既有保守根规则,包括正估算值。正常轮次局部遥测、
model/source/pricing 上报和 reporter 行为保持不变。没有 observer 的 context
继续沿用工具原行为。
## Limits and search authority / 阈值与搜索权限
Explicit token limits may land earlier after a search, at the existing
after-tool check with finalization. Zero stays unbounded. This is not request
reservation, a monetary ledger, provider cancellation or process termination.
显式 token 阈值可能在搜索后更早触发既有工具轮后检查与收尾;零值仍不限。
本机制不是请求预留、金额账本、供应商取消或进程终止。
The ordered-backend `Chain` uses request-priced backends, not this native token
terminal. Its paid consent, fallback and request billing remain unchanged.
Search model selection, credentials, redirects, native completion, isolated
query history and external-content provenance are unchanged.
有序 backend `Chain` 使用按请求计费的后端,不经过本原生 token 终态;其付费
同意、回退与请求计费保持不变。搜索模型选择、凭据、重定向、原生完成边界、
独立查询历史和外部来源标签均保持原样。
## Evidence boundary / 证据边界
Real boot HTTP/SSE drives an OpenAI-compatible worker through the actual
capability gateway into an Anthropic-native search response. Tests inspect
the real summary, citation source, usage identity and independent request
counts, with disabled-limit and missing-usage controls.
真实 boot HTTP/SSE 将 OpenAI-compatible worker 经实际 capability gateway 接到
Anthropic 原生搜索响应。回归检查实际摘要、引用来源、用量身份与独立请求次数,
并保留禁用阈值和缺失用量对照。
The fixture simulates the provider protocol; it does not query the public web.
Serialized main prefix/schema equality is a cache guard, not a live-provider
hit-rate measurement. No API, config, persisted or provider-wire field changes.
夹具模拟供应商协议,不查询公共互联网。主序列化前缀/schema 相等是缓存守卫,
不是供应商命中率实测;不改变 API、配置、持久化或 provider-wire 字段。
This partial consumer does not complete all auxiliary admission, shared
Coordinator budgets, root currency aggregation, durable accounting or the full
isolation/recovery plan. See [bounded root usage](BOUNDED_ROOT_TOKEN_BUDGET.md).
本 consumer 不完成全部辅助准入、共享 Coordinator 预算、根币种汇总、持久计账
或完整隔离恢复方案。作者分析、独立 review 与原生平台执行分别记录。
@@ -0,0 +1,141 @@
package boot
import (
"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"
"reasonix/internal/state/sessionstore"
)
func TestEffectRootWebSearchTokenBudget(t *testing.T) {
for _, tc := range []struct {
name string
limit int
missing bool
wantSearch int32
}{
{name: "reported", limit: 1500, wantSearch: 1},
{name: "disabled", wantSearch: 4},
{name: "missing", limit: 1500, missing: true, wantSearch: 2},
} {
t.Run(tc.name, func(t *testing.T) {
home, root := statedBootHome(t), t.TempDir()
var normal, searches, receipts, results, normalReceipts atomic.Int32
var prefix atomic.Pointer[string]
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var req struct {
Model string `json:"model"`
Messages []json.RawMessage `json:"messages"`
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")
if req.Model == "chosen" {
searches.Add(1)
if len(req.Messages) != 1 || !strings.Contains(string(req.Tools), "web_search_20250305") || strings.Contains(string(req.Tools), "bash") {
t.Errorf("native search request messages/tools=%d/%s", len(req.Messages), req.Tools)
}
if !strings.Contains(string(req.Messages[0]), "neutral fixture query") {
t.Error("native search did not receive its query")
}
if tc.missing {
_, _ = io.WriteString(w, "data: {\"type\":\"message_start\",\"message\":{\"id\":\"search\"}}\n\n")
} else {
_, _ = io.WriteString(w, "data: {\"type\":\"message_start\",\"message\":{\"id\":\"search\",\"usage\":{\"input_tokens\":1000}}}\n\n")
}
_, _ = io.WriteString(w, "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"web_search_tool_result\",\"tool_use_id\":\"s\",\"content\":[{\"title\":\"Fixture\",\"url\":\"https://example.com/root-search\"}]}}\n\n"+
"data: {\"type\":\"content_block_delta\",\"index\":1,\"delta\":{\"type\":\"text_delta\",\"text\":\"Native root search summary\"}}\n\n")
if !tc.missing {
_, _ = io.WriteString(w, "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"},\"usage\":{\"output_tokens\":100}}\n\n")
}
_, _ = io.WriteString(w, "data: {\"type\":\"message_stop\"}\n\n")
return
}
shape := string(req.Messages[0]) + string(req.Tools)
if old := prefix.Swap(&shape); old != nil && *old != shape {
t.Error("main prefix/schema changed during the run")
}
if n := normal.Add(1); n <= 4 {
_, _ = fmt.Fprintf(w, "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"search-%d\",\"type\":\"function\",\"function\":{\"name\":\"use_capability\",\"arguments\":\"{\\\"action\\\":\\\"call\\\",\\\"capability_id\\\":\\\"tool:web_search\\\",\\\"arguments\\\":{\\\"query\\\":\\\"neutral fixture query\\\"}}\"}}]},\"finish_reason\":\"tool_calls\"}]}\n\n", n)
} else {
_, _ = io.WriteString(w, "data: {\"choices\":[{\"delta\":{\"content\":\"Finished.\"},\"finish_reason\":\"stop\"}]}\n\n")
}
_, _ = io.WriteString(w, "data: {\"choices\":[],\"usage\":{\"prompt_tokens\":1000,\"completion_tokens\":100,\"total_tokens\":1100}}\n\n"+
"data: [DONE]\n\n")
}))
t.Cleanup(srv.Close)
writeFile(t, root, "reasonix.toml", fmt.Sprintf(`default_model = "worker"
[agent]
system_prompt = "BASE"
web_search_model = "search/chosen"
task_token_budget = %d
[codegraph]
enabled = false
[[providers]]
name = "worker"
kind = "openai"
model = "worker"
context_window = 200000
base_url = %q
api_key_env = "ROOT_SEARCH_KEY"
[[providers]]
name = "search"
kind = "anthropic"
models = ["chosen"]
web_search = true
base_url = %q
api_key_env = "ROOT_SEARCH_KEY"
`, tc.limit, srv.URL, srv.URL))
t.Setenv("ROOT_SEARCH_KEY", "test")
approveWorkspaceAt(t, home, root)
sink := event.FuncSink(func(e event.Event) {
if e.Kind == event.Usage && e.UsageSource == "web-search" {
receipts.Add(1)
if e.ModelRef != "search/chosen" || e.Usage.RequestCount != 1 || e.Usage.Estimated != tc.missing {
t.Errorf("search provenance model=%s usage=%+v", e.ModelRef, e.Usage)
}
}
if e.Kind == event.Usage && e.UsageSource != "web-search" {
normalReceipts.Add(1)
if e.Usage.RequestCount != 1 {
t.Errorf("search attempts leaked into main usage=%+v", e.Usage)
}
}
if e.Kind == event.ToolResult && e.Tool.Name == "use_capability" && e.Tool.Err == "" && strings.Contains(e.Tool.Output, "Native root search summary") && strings.Contains(e.Tool.Output, "https://example.com/root-search") {
results.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)
err = ctrl.Run(t.Context(), "search the native fixture")
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 searches.Load() != tc.wantSearch || receipts.Load() != tc.wantSearch || results.Load() != tc.wantSearch || normal.Load() != tc.wantSearch+1 || normalReceipts.Load() != normal.Load() {
t.Fatalf("search/receipts/results/normal/normalReceipts=%d/%d/%d/%d/%d want%d/%d/%d/%d/%d", searches.Load(), receipts.Load(), results.Load(), normal.Load(), normalReceipts.Load(), tc.wantSearch, tc.wantSearch, tc.wantSearch, tc.wantSearch+1, tc.wantSearch+1)
}
})
}
}
+158
View File
@@ -0,0 +1,158 @@
package websearch
import (
"context"
"errors"
"testing"
"reasonix/internal/contract/provider"
)
type observedSearchProvider struct {
chunks []provider.Chunk
after func()
err error
}
func (*observedSearchProvider) Name() string { return "observed-search" }
func (p *observedSearchProvider) Stream(ctx context.Context, _ provider.Request) (<-chan provider.Chunk, error) {
if p.err != nil {
return nil, p.err
}
ch := make(chan provider.Chunk)
go func() {
defer close(ch)
for _, chunk := range p.chunks {
select {
case ch <- chunk:
case <-ctx.Done():
return
}
}
if p.after != nil {
p.after()
}
}()
return ch, nil
}
func TestRootWebSearchUsageTerminal(t *testing.T) {
billedFailure := errors.New("billed native search failed")
for _, tc := range []struct {
name string
missing bool
noReport bool
noObserver bool
invalid bool
noFactory bool
factoryErr bool
cancel bool
preCancel bool
streamErr bool
fail bool
want int
}{
{name: "reported", want: 1},
{name: "no_reporter", noReport: true, want: 1},
{name: "missing", missing: true},
{name: "no_observer", noObserver: true},
{name: "input_refused", invalid: true},
{name: "no_factory", noFactory: true},
{name: "factory_error", factoryErr: true},
{name: "billed_failure", fail: true, want: 1},
{name: "cancel_after_usage", cancel: true, want: 1},
{name: "cancel_before_start", preCancel: true},
{name: "stream_error", streamErr: true},
} {
t.Run(tc.name, func(t *testing.T) {
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
usage := &provider.Usage{PromptTokens: 1000, CompletionTokens: 100, TotalTokens: 1100, RequestCount: 3}
chunks := []provider.Chunk{}
if !tc.missing {
chunks = append(chunks, provider.Chunk{Type: provider.ChunkUsage, Usage: usage})
}
chunks = append(chunks, provider.Chunk{Type: provider.ChunkProviderTool, ToolCall: &provider.ToolCall{Name: "web_search"}, Text: "https://example.com/fixture"})
if tc.fail {
chunks = append(chunks, provider.Chunk{Type: provider.ChunkError, Err: billedFailure})
} else if !tc.cancel {
chunks = append(chunks, provider.Chunk{Type: provider.ChunkText, Text: "fixture answer"}, provider.Chunk{Type: provider.ChunkDone})
}
p := &observedSearchProvider{chunks: chunks}
if tc.streamErr {
p.err = billedFailure
}
if tc.cancel {
p.after = cancel
}
calls, reports, factory := 0, 0, 0
if !tc.noObserver {
ctx = provider.WithAuxiliaryUsageObserver(ctx, func(u *provider.Usage) {
calls++
if u.PromptTokens != 1000 || u.RequestCount != 3 {
t.Errorf("observer usage=%+v", u)
}
u.PromptTokens = -1
})
}
search := &Tool{Factory: func() (provider.Provider, error) {
factory++
if tc.factoryErr {
return nil, ErrUnavailable
}
return p, nil
}, ReportUsage: func(u *provider.Usage) {
reports++
if u.PromptTokens != 1000 || u.RequestCount != 3 {
t.Errorf("observer changed reported usage=%+v", u)
}
}}
if tc.noFactory {
search.Factory = nil
}
if tc.noReport {
search.ReportUsage = nil
}
args := []byte(`{"query":"fixture"}`)
if tc.invalid {
args = []byte(`{"query":" "}`)
}
if tc.preCancel {
cancel()
}
out, err := search.Execute(ctx, args)
switch {
case tc.fail || tc.streamErr:
if !errors.Is(err, billedFailure) {
t.Fatal(err)
}
case tc.cancel || tc.preCancel:
if !errors.Is(err, context.Canceled) {
t.Fatal(err)
}
if tc.preCancel && factory != 0 {
t.Fatal("canceled request reached the factory")
}
case tc.invalid:
if !errors.Is(err, ErrQuery) || factory != 0 {
t.Fatalf("refusal=%v factory=%d", err, factory)
}
case tc.noFactory || tc.factoryErr:
if !errors.Is(err, ErrUnavailable) {
t.Fatal(err)
}
default:
if err != nil || out == "" {
t.Fatalf("result=%q err=%v", out, err)
}
}
wantReport := 1
if tc.missing || tc.noReport || tc.invalid || tc.noFactory || tc.factoryErr || tc.streamErr || tc.preCancel {
wantReport = 0
}
if calls != tc.want || reports != wantReport || usage.PromptTokens != 1000 {
t.Fatalf("observe/report/original=%d/%d/%d want%d/%d/1000", calls, reports, usage.PromptTokens, tc.want, wantReport)
}
})
}
}
+3 -1
View File
@@ -72,7 +72,9 @@ func (t *Tool) Execute(ctx context.Context, args json.RawMessage) (string, error
}
var usage *provider.Usage
defer func() {
if u := provider.UsageWithRequestAttemptCount(ctx, usage); u != nil && t.ReportUsage != nil {
u := provider.UsageWithRequestAttemptCount(ctx, usage)
provider.ObserveAuxiliaryUsage(ctx, u)
if u != nil && t.ReportUsage != nil {
t.ReportUsage(u)
}
}()
@@ -0,0 +1,2 @@
- 修复:活跃根任务的 token 软预算现在计入原生 web_search 已报告用量,保留独立请求计数、缺失用量、来源标签和付费同意边界。关联 #14025。
- Fixed: active root token soft budgets now include native web_search usage, preserving independent request counts, missing usage, provenance and paid-search consent. Refs #14025.