mirror of
https://github.com/esengine/DeepSeek-Reasonix.git
synced 2026-11-07 10:45:59 -10:00
Merge pull request #14593 from SivanCola/codex/coordinator-root-budget-14025
fix(coordinator): share root task budgets / 规划与执行共享根任务预算
This commit is contained in:
10 files changed
+497
-9
No files matched your search
@@ -323,6 +323,39 @@ jobs:
|
||||
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
|
||||
shell: bash
|
||||
run: |
|
||||
set -o pipefail
|
||||
go test -count=1 -timeout=3m -json -run '^Test(PreparedRootBudget|CoordinatorRootBudget|EffectCoordinatorRoot)' ./internal/runtime/agent ./internal/runtime/coordinator ./internal/assembly/boot | tee "$RUNNER_TEMP/coordinator-root-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/runtime/coordinator", "reasonix/internal/assembly/boot"] - $packages | length) == 0 and
|
||||
(["TestPreparedRootBudgetLifetime", "TestPreparedRootBudgetLifetime/continuation",
|
||||
"TestPreparedRootBudgetLifetime/new_scope", "TestPreparedRootBudgetLifetime/new_task",
|
||||
"TestPreparedRootBudgetLifetime/reset", "TestPreparedRootBudgetLifetime/new_session",
|
||||
"TestPreparedRootBudgetProjectionAndPreservedTurn", "TestCoordinatorRootBudgetRoutes",
|
||||
"TestCoordinatorRootBudgetRoutes/raw", "TestCoordinatorRootBudgetRoutes/tool_planner",
|
||||
"TestCoordinatorRootBudgetRoutes/failure_fallback", "TestCoordinatorRootBudgetRoutes/plan_only_continuation",
|
||||
"TestCoordinatorRootBudgetRoutes/approval_declined_continuation", "TestCoordinatorRootBudgetRoutes/approval_allowed",
|
||||
"TestEffectCoordinatorRootTokenBudget", "TestEffectCoordinatorRootTokenBudget/shared",
|
||||
"TestEffectCoordinatorRootTokenBudget/unconfigured", "TestEffectCoordinatorRootTokenBudget/context_disables",
|
||||
"TestEffectCoordinatorRootTokenBudget/missing_planner", "TestEffectCoordinatorRootTokenBudget/executor_only"] - $tests | length) == 0
|
||||
' "$RUNNER_TEMP/coordinator-root-budget.jsonl"
|
||||
|
||||
- name: upload Windows Coordinator 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: coordinator-root-budget-windows-${{ github.run_attempt }}
|
||||
path: ${{ runner.temp }}/coordinator-root-budget.jsonl
|
||||
if-no-files-found: ignore
|
||||
retention-days: 7
|
||||
|
||||
- name: test (Windows root wall budget)
|
||||
if: env.RUN_STEPS == 'true' && runner.os == 'Windows' && github.event_name == 'pull_request'
|
||||
timeout-minutes: 5
|
||||
|
||||
@@ -0,0 +1,106 @@
|
||||
---
|
||||
owner: @SivanCola
|
||||
backup: @esengine
|
||||
status: active
|
||||
reviewed: 2026-10-26
|
||||
---
|
||||
|
||||
# Coordinator root budget / Coordinator 根预算
|
||||
|
||||
The Coordinator prepares the executor's root token/wall scope before planning.
|
||||
Its tool-enabled planner inherits that scope through Agent.Run; the legacy raw
|
||||
planner reports one finalized usage record through the existing auxiliary
|
||||
observer. Executor work then contributes to the same atomic token counter.
|
||||
|
||||
Coordinator 在规划前准备 executor 的根 token 和时间作用域。工具型 planner 经
|
||||
Agent.Run 继承它,旧无工具流式 planner 经已有 observer 记录一次终态用量;随后
|
||||
executor 使用同一个原子 token 计数器。
|
||||
|
||||
Each Agent's stricter local gate still takes precedence. Resetting the shared
|
||||
root does not erase another Agent's local telemetry or loosen its own limits.
|
||||
|
||||
每个 Agent 更严格的本地上限仍优先。重置共享根不会清除其他 Agent 的本地遥测,
|
||||
也不会放宽其已有本地限制。
|
||||
|
||||
Previously, the two Agent instances had independent roots, allowing an extra
|
||||
executor tool round under a configured token limit. Local telemetry remains
|
||||
per Agent. Local monetary limits are still separate; no floating-point sum or
|
||||
new monetary ledger is introduced.
|
||||
|
||||
此前两个 Agent 的根计数独立,显式 token 限额下 executor 会多运行一轮工具。
|
||||
本地遥测仍按 Agent 分开,金额上限也仍独立;不新增浮点汇总或金额账本。
|
||||
|
||||
## Lifetime and authority / 寿命与权限
|
||||
|
||||
| Transition | Root behavior / 根行为 |
|
||||
| --- | --- |
|
||||
| New ordinary task | Fresh counter and wall start / 新计数和时间起点 |
|
||||
| Same host delivery scope | Keep accumulated tokens and original clock / 保留同一宿主任务累计值和时钟 |
|
||||
| Explicit preserved continuation | Keep the old counter / 保留已有计数 |
|
||||
| Plan-only or denied approval | Preserve planning spend without executing tools / 保留规划支出,不执行工具 |
|
||||
| ResetTaskBudget | Next preparation creates a fresh counter / 下次准备新计数 |
|
||||
| Replacement conversation | Fresh root even if the host scope ID is reused / 换会话对象后即使任务 ID 相同也换根 |
|
||||
| Old projected job | Continue charging its captured old counter / 只计入捕获的旧计数 |
|
||||
| Inherited root | Keep parent authority; child limits cannot widen it / 继承父级根,子级不能放宽 |
|
||||
|
||||
The session object and host scope identify preparation, independently of the
|
||||
executor's evidence ledger. A no-executor turn persists conversation but never
|
||||
calls Agent.Run; it must not lose its planning spend merely because that Run
|
||||
has not opened an evidence ledger. No user text is parsed to infer continuation.
|
||||
|
||||
准备状态按会话对象和宿主任务作用域识别,不依赖 executor 证据账本。无执行回合
|
||||
会保存会话但不调用 Agent.Run,不能因它尚未打开账本而遗失规划支出;不从用户
|
||||
文字猜测是否继续。
|
||||
|
||||
Only the preparing executor adopts its own prepared root after opening its
|
||||
turn. A companion Agent inherits accounting without claiming ownership of the
|
||||
executor's preparation. Root limits are immutable per turn; overrides create
|
||||
a new limit snapshot over the same counter, leaving an old job's tuple intact.
|
||||
|
||||
只有准备它的 executor 在开启回合后采纳该根;其他 Agent 仅继承计账。每回合的
|
||||
根限制不可变,覆盖配置时用同一计数器生成新限制快照,不改旧任务捕获的组合。
|
||||
|
||||
PrepareRootTaskBudget is an internal Go host seam, serialized with Run,
|
||||
SetSession and ResetTaskBudget. It carries context values, not cancellation or
|
||||
permission changes. Existing Manager-owned jobs keep their own cancellation
|
||||
authority and the accepted budget-only projection.
|
||||
|
||||
PrepareRootTaskBudget 是内部 Go 宿主接口,与 Run、SetSession 和 ResetTaskBudget
|
||||
串行调用。它不增加取消或权限能力;Manager 作业保留自己的取消权限和已有预算投射。
|
||||
|
||||
## Evidence and limits / 证据与限制
|
||||
|
||||
Real Boot HTTP/SSE verifies configured, unconfigured, context-disabled,
|
||||
missing-planner-usage and executor-only paths. With2,000 tokens and a1,100-token
|
||||
plan, executor work now reads once and finalizes, rather than reading twice.
|
||||
Request counts and source-specific usage events remain separate.
|
||||
|
||||
真实 Boot HTTP/SSE 覆盖显式限制、默认关闭、上下文关闭、planner 缺失用量及仅执行
|
||||
路径。2,000 token 限制和 1,100 token 规划支出下,executor 从两次实际文件读取变为
|
||||
一次后收尾;请求数和按来源区分的用量事件保留。
|
||||
|
||||
Coordinator tests exercise raw/tool planners, failure fallback, plan-only,
|
||||
declined approval followed by continuation, and approved execution. Agent tests
|
||||
cover reset, conversation replacement, scope replacement, late old-scope
|
||||
accounting, immutable overrides and independent job cancellation.
|
||||
|
||||
Coordinator 回归覆盖两类 planner、失败回退、仅规划、拒绝审批后继续及批准执行。
|
||||
Agent 回归覆盖重置、换会话、换任务、晚到旧作用域记账、不可变覆盖和独立作业取消。
|
||||
|
||||
These remain cooperative checks after a tool round plus existing finalization.
|
||||
An in-flight first request/tool, concurrent children and approval waiting can
|
||||
exceed a limit. The wall clock starts at Coordinator preparation, not boot or
|
||||
earlier frontend queuing. There is no request reservation, timer or kill.
|
||||
|
||||
仍只在工具轮后合作式检查并保留已有收尾。在途首个请求或工具、并发子任务和审批
|
||||
等待均可越界。根时钟始于 Coordinator 准备,不包含更早启动或前端排队;无预留、
|
||||
定时器或强制终止。
|
||||
|
||||
Missing usage remains unknown. Routing/title work before Coordinator.Run,
|
||||
cross-currency money, durable accounting and physical process completion are
|
||||
not covered. Prefixes, request schemas, permissions, defaults and plan approval
|
||||
policy do not change. The complete #14025 plan remains open.
|
||||
|
||||
缺失用量仍未知,不覆盖 Coordinator.Run 前的路由和标题、跨币种金额、持久计账或
|
||||
物理进程完成。不改变前缀、请求 schema、权限、默认值或规划审批策略;#14025 完整
|
||||
方案仍未完成。
|
||||
@@ -0,0 +1,121 @@
|
||||
package boot
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
"reasonix/internal/contract/event"
|
||||
"reasonix/internal/runtime/agent"
|
||||
"reasonix/internal/session/control"
|
||||
)
|
||||
|
||||
func TestEffectCoordinatorRootTokenBudget(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
limit int
|
||||
missing, disable, skip bool
|
||||
wantReads, wantHTTP int32
|
||||
}{
|
||||
{name: "shared", limit: 2000, wantReads: 1, wantHTTP: 2},
|
||||
{name: "unconfigured", wantReads: 3, wantHTTP: 4},
|
||||
{name: "context_disables", limit: 2000, disable: true, wantReads: 3, wantHTTP: 4},
|
||||
{name: "missing_planner", limit: 2000, missing: true, wantReads: 2, wantHTTP: 3},
|
||||
{name: "executor_only", limit: 2000, skip: true, wantReads: 2, wantHTTP: 3},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
home, root := statedBootHome(t), t.TempDir()
|
||||
writeFile(t, root, "fixture.txt", "shared coordinator fixture")
|
||||
var plans, executions, reads, usage atomic.Int32
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
var req struct {
|
||||
Model string `json:"model"`
|
||||
}
|
||||
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")
|
||||
planner := req.Model == "planner-model"
|
||||
if planner {
|
||||
plans.Add(1)
|
||||
_, _ = io.WriteString(w, "data: {\"choices\":[{\"delta\":{\"content\":\"Read fixture.txt and report its content.\"},\"finish_reason\":\"stop\"}]}\n\n")
|
||||
} else if n := executions.Add(1); n <= 3 {
|
||||
_, _ = 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\":\"Complete.\"},\"finish_reason\":\"stop\"}]}\n\n")
|
||||
}
|
||||
if !planner || !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 = "executor"
|
||||
[agent]
|
||||
planner_model = "planner"
|
||||
task_token_budget = %d
|
||||
[codegraph]
|
||||
enabled = false
|
||||
[[providers]]
|
||||
name = "executor"
|
||||
kind = "openai"
|
||||
model = "executor-model"
|
||||
base_url = %q
|
||||
api_key_env = "COORDINATOR_ROOT_EFFECT_KEY"
|
||||
[[providers]]
|
||||
name = "planner"
|
||||
kind = "openai"
|
||||
model = "planner-model"
|
||||
base_url = %q
|
||||
api_key_env = "COORDINATOR_ROOT_EFFECT_KEY"
|
||||
`, tc.limit, srv.URL, srv.URL))
|
||||
t.Setenv("COORDINATOR_ROOT_EFFECT_KEY", "test")
|
||||
approveWorkspaceAt(t, home, root)
|
||||
ctrl, err := Build(t.Context(), Options{Home: home, WorkspaceRoot: root, Sink: event.FuncSink(func(e event.Event) {
|
||||
if e.Kind == event.ToolResult && e.Tool.Name == "read_file" && e.Tool.Err == "" {
|
||||
reads.Add(1)
|
||||
}
|
||||
if e.Kind == event.Usage && e.Usage != nil {
|
||||
usage.Add(1)
|
||||
if e.Usage.RequestCount != 1 {
|
||||
t.Errorf("requestcount=%+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{})
|
||||
}
|
||||
input := "read fixture.txt"
|
||||
if !tc.skip {
|
||||
input = control.PlannerRouteMarker + " " + input
|
||||
}
|
||||
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(1)
|
||||
if tc.skip {
|
||||
wantPlans = 0
|
||||
}
|
||||
if plans.Load() != wantPlans || executions.Load() != tc.wantHTTP || reads.Load() != tc.wantReads || usage.Load() != wantPlans+tc.wantHTTP {
|
||||
t.Fatalf("planner/executor/read/usage=%d/%d/%d/%d want%d/%d/%d/%d", plans.Load(), executions.Load(), reads.Load(), usage.Load(), wantPlans, tc.wantHTTP, tc.wantReads, wantPlans+tc.wantHTTP)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"reasonix/internal/contract/provider"
|
||||
"reasonix/internal/state/sessionstore"
|
||||
)
|
||||
|
||||
type rootBudgetOrigin struct {
|
||||
session *sessionstore.Session
|
||||
scopeID string
|
||||
}
|
||||
|
||||
type preparedRootBudget struct {
|
||||
owner *Agent
|
||||
origin rootBudgetOrigin
|
||||
}
|
||||
|
||||
type preparedRootBudgetKey struct{}
|
||||
|
||||
// PrepareRootTaskBudget binds task accounting before a host runs companion roles.
|
||||
// Calls must be serialized with Run, SetSession and ResetTaskBudget. Cancellation
|
||||
// and local telemetry remain owned by each role; inherited roots stay inherited.
|
||||
func (a *Agent) PrepareRootTaskBudget(ctx context.Context) context.Context {
|
||||
if _, inherited := ctx.Value(rootTokenBudgetKey{}).(*rootTokenBudget); inherited {
|
||||
return ctx
|
||||
}
|
||||
identity, scoped := DeliveryExecutionScopeFromContext(ctx)
|
||||
origin := rootBudgetOrigin{session: a.Session(), scopeID: identity.ID}
|
||||
previous := a.task.budget.rootTokens
|
||||
keep := previous != nil && a.task.budget.rootOrigin.session == origin.session &&
|
||||
(a.pending.preserveEvidence || scoped && a.task.budget.rootOrigin.scopeID == identity.ID)
|
||||
if !keep {
|
||||
previous = &rootTokenBudget{used: new(atomic.Uint64)}
|
||||
}
|
||||
scope := rootBudgetSnapshot(previous, a.taskBudgetLimit(ctx), time.Now())
|
||||
a.task.budget.rootTokens, a.task.budget.rootOrigin = scope, origin
|
||||
ctx = context.WithValue(ctx, rootTokenBudgetKey{}, scope)
|
||||
ctx = context.WithValue(ctx, preparedRootBudgetKey{}, preparedRootBudget{owner: a, origin: origin})
|
||||
return provider.WithAuxiliaryUsageObserver(ctx, scope.observe)
|
||||
}
|
||||
|
||||
func rootBudgetSnapshot(previous *rootTokenBudget, limit TaskBudget, started time.Time) *rootTokenBudget {
|
||||
wall := previous.wall
|
||||
if wall.started.IsZero() {
|
||||
wall.started = started
|
||||
}
|
||||
wall.limit = limit.Wall
|
||||
return &rootTokenBudget{used: previous.used, limit: uint64(max(0, limit.Tokens)), wall: wall}
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"reasonix/internal/contract/event"
|
||||
"reasonix/internal/contract/provider"
|
||||
"reasonix/internal/contract/tool"
|
||||
"reasonix/internal/state/sessionstore"
|
||||
)
|
||||
|
||||
func TestPreparedRootBudgetLifetime(t *testing.T) {
|
||||
for _, mode := range []string{"continuation", "new_scope", "new_task", "reset", "new_session"} {
|
||||
t.Run(mode, func(t *testing.T) {
|
||||
a := New(&spendingProvider{}, tool.NewRegistry(), sessionstore.NewSession("root"), Options{TaskBudget: TaskBudget{Tokens: 5000, Wall: time.Hour}}, event.Discard)
|
||||
ctx := WithDeliveryExecutionScope(t.Context(), DeliveryExecutionScope{ID: "first"})
|
||||
first := a.PrepareRootTaskBudget(ctx)
|
||||
provider.ObserveAuxiliaryUsage(first, &provider.Usage{PromptTokens: 1000, CompletionTokens: 100})
|
||||
old := a.task.budget.rootTokens
|
||||
projected := InheritRootTokenBudget(context.Background(), first)
|
||||
if err := a.Run(first, "read-only answer"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if a.task.budget.rootTokens != old || old.used.Load() != 2200 {
|
||||
t.Fatalf("planning scope lost at Run boundary: %d", old.used.Load())
|
||||
}
|
||||
switch mode {
|
||||
case "new_scope":
|
||||
ctx = WithDeliveryExecutionScope(t.Context(), DeliveryExecutionScope{ID: "second"})
|
||||
case "new_task":
|
||||
ctx = t.Context()
|
||||
case "reset":
|
||||
a.ResetTaskBudget()
|
||||
case "new_session":
|
||||
a.SetSession(sessionstore.NewSession("replacement"))
|
||||
}
|
||||
next := a.PrepareRootTaskBudget(WithTaskBudget(ctx, TaskBudget{Tokens: 7000, Wall: 2 * time.Hour}))
|
||||
current := a.task.budget.rootTokens
|
||||
if current.limit != 7000 || current.wall.limit != 2*time.Hour || old.limit != 5000 || old.wall.limit != time.Hour {
|
||||
t.Fatal("turn override mutated an old scope")
|
||||
}
|
||||
if mode == "continuation" {
|
||||
if current.used != old.used || !current.wall.started.Equal(old.wall.started) {
|
||||
t.Fatal("continuation lost counters or clock")
|
||||
}
|
||||
} else if current.used == old.used || current.used.Load() != 0 {
|
||||
t.Fatal("replacement inherited old spend")
|
||||
}
|
||||
provider.ObserveAuxiliaryUsage(projected, &provider.Usage{PromptTokens: 500})
|
||||
want := uint64(0)
|
||||
if mode == "continuation" {
|
||||
want = 2700
|
||||
}
|
||||
if current.used.Load() != want {
|
||||
t.Fatalf("late old scope charged wrong generation: %d want%d", current.used.Load(), want)
|
||||
}
|
||||
if inherited := a.PrepareRootTaskBudget(next); inherited != next {
|
||||
t.Fatal("inherited context was rebound")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPreparedRootBudgetProjectionAndPreservedTurn(t *testing.T) {
|
||||
a := New(&spendingProvider{}, tool.NewRegistry(), sessionstore.NewSession("root"), Options{}, event.Discard)
|
||||
parent, cancel := context.WithCancel(t.Context())
|
||||
defer cancel()
|
||||
bound := a.PrepareRootTaskBudget(WithTaskBudget(parent, TaskBudget{Tokens: 5000}))
|
||||
provider.ObserveAuxiliaryUsage(bound, &provider.Usage{PromptTokens: 1000})
|
||||
old := a.task.budget.rootTokens
|
||||
a.pending.preserveEvidence = true
|
||||
continued := a.PrepareRootTaskBudget(WithTaskBudget(parent, TaskBudget{Tokens: 6000}))
|
||||
if a.task.budget.rootTokens.used != old.used {
|
||||
t.Fatal("host-preserved continuation reset spend")
|
||||
}
|
||||
job, stop := context.WithCancel(t.Context())
|
||||
defer stop()
|
||||
projected := InheritRootTokenBudget(job, continued)
|
||||
cancel()
|
||||
if projected.Err() != nil {
|
||||
t.Fatal("budget projection copied parent cancellation")
|
||||
}
|
||||
provider.ObserveAuxiliaryUsage(projected, &provider.Usage{CompletionTokens: 100})
|
||||
if old.used.Load() != 1100 {
|
||||
t.Fatal("projected job lost root scope")
|
||||
}
|
||||
stop()
|
||||
if projected.Err() != context.Canceled {
|
||||
t.Fatal("projection lost job cancellation")
|
||||
}
|
||||
}
|
||||
@@ -41,18 +41,16 @@ func (a *Agent) bindRootTokenBudget(ctx context.Context, state *turnRuntime) con
|
||||
previous = &rootTokenBudget{used: new(atomic.Uint64)}
|
||||
previous.observe(&provider.Usage{PromptTokens: a.task.budget.promptTokens, CompletionTokens: a.task.budget.outputTokens})
|
||||
}
|
||||
limit := a.taskBudgetLimit(ctx)
|
||||
wall := previous.wall
|
||||
if wall.started.IsZero() {
|
||||
wall.started = a.task.budget.started
|
||||
if wall.started.IsZero() {
|
||||
wall.started = state.budget.started
|
||||
}
|
||||
started := a.task.budget.started
|
||||
if started.IsZero() {
|
||||
started = state.budget.started
|
||||
}
|
||||
wall.limit = limit.Wall
|
||||
scope = &rootTokenBudget{used: previous.used, limit: uint64(max(0, limit.Tokens)), wall: wall}
|
||||
scope = rootBudgetSnapshot(previous, a.taskBudgetLimit(ctx), started)
|
||||
a.task.budget.rootTokens = scope
|
||||
a.task.budget.rootOrigin = rootBudgetOrigin{session: a.Session(), scopeID: a.task.scopeID}
|
||||
ctx = context.WithValue(ctx, rootTokenBudgetKey{}, scope)
|
||||
} else if prepared, ok := ctx.Value(preparedRootBudgetKey{}).(preparedRootBudget); ok && prepared.owner == a {
|
||||
a.task.budget.rootTokens, a.task.budget.rootOrigin = scope, prepared.origin
|
||||
}
|
||||
state.budget.rootTokens = scope
|
||||
return provider.WithAuxiliaryUsageObserver(ctx, scope.observe)
|
||||
|
||||
@@ -52,6 +52,7 @@ type runBudget struct {
|
||||
// starts a new task.
|
||||
limit TaskBudget
|
||||
rootTokens *rootTokenBudget
|
||||
rootOrigin rootBudgetOrigin
|
||||
}
|
||||
|
||||
// observe folds one round's provider usage into the turn's running total.
|
||||
|
||||
@@ -321,6 +321,7 @@ func (c *Coordinator) SetPlannerPlanApprover(g agent.PlannerPlanApprover) {
|
||||
|
||||
// Run plans with the planner model, then hands the plan to the executor.
|
||||
func (c *Coordinator) Run(ctx context.Context, input string) error {
|
||||
ctx = c.executor.PrepareRootTaskBudget(ctx)
|
||||
// A host that announced this turn owns its boundary; opening a second one
|
||||
// here would make planner work and executor work two turns to every sink
|
||||
// that resets on a start.
|
||||
@@ -540,6 +541,7 @@ func (c *Coordinator) planFromStream(ctx context.Context, input string) (string,
|
||||
streamCompleted := false
|
||||
defer func() {
|
||||
accounted := provider.UsageWithRequestAttemptCount(ctx, usage)
|
||||
provider.ObserveAuxiliaryUsage(ctx, accounted)
|
||||
if accounted != nil || streamCompleted {
|
||||
c.sink.Emit(event.Event{Kind: event.Usage, ModelRef: c.plannerModelRef, Usage: accounted, Pricing: c.plannerPricing, Source: event.UsageSourcePlanner, UsageSource: event.UsageSourcePlanner})
|
||||
}
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
package coordinator
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"reasonix/internal/contract/event"
|
||||
"reasonix/internal/contract/provider"
|
||||
"reasonix/internal/contract/tool"
|
||||
"reasonix/internal/runtime/agent"
|
||||
"reasonix/internal/state/sessionstore"
|
||||
)
|
||||
|
||||
type coordinatorBudgetRead struct{ calls int }
|
||||
|
||||
func (*coordinatorBudgetRead) Name() string { return "read_fixture" }
|
||||
func (*coordinatorBudgetRead) Description() string { return "Read a fixture" }
|
||||
func (*coordinatorBudgetRead) Schema() json.RawMessage {
|
||||
return json.RawMessage(`{"type":"object"}`)
|
||||
}
|
||||
func (*coordinatorBudgetRead) ReadOnly() bool { return true }
|
||||
func (r *coordinatorBudgetRead) Execute(context.Context, json.RawMessage) (string, error) {
|
||||
r.calls++
|
||||
return "fixture", nil
|
||||
}
|
||||
|
||||
func coordinatorBudgetRounds() [][]provider.Chunk {
|
||||
var rounds [][]provider.Chunk
|
||||
for i := range 3 {
|
||||
rounds = append(rounds, []provider.Chunk{{Type: provider.ChunkToolCall, ToolCall: &provider.ToolCall{ID: fmt.Sprint(i), Name: "read_fixture", Arguments: "{}"}}, {Type: provider.ChunkUsage, Usage: &provider.Usage{PromptTokens: 1000, CompletionTokens: 100, TotalTokens: 1100}}, {Type: provider.ChunkDone}})
|
||||
}
|
||||
return append(rounds, []provider.Chunk{{Type: provider.ChunkText, Text: "Complete."}, {Type: provider.ChunkUsage, Usage: &provider.Usage{PromptTokens: 1000, CompletionTokens: 100, TotalTokens: 1100}}, {Type: provider.ChunkDone}})
|
||||
}
|
||||
|
||||
func TestCoordinatorRootBudgetRoutes(t *testing.T) {
|
||||
for _, mode := range []string{"raw", "tool_planner", "failure_fallback", "plan_only_continuation", "approval_declined_continuation", "approval_allowed"} {
|
||||
t.Run(mode, func(t *testing.T) {
|
||||
planner := &mockProvider{name: "planner", chunks: []provider.Chunk{{Type: provider.ChunkText, Text: "Read the fixture."}, {Type: provider.ChunkUsage, Usage: &provider.Usage{PromptTokens: 1000, CompletionTokens: 100, TotalTokens: 1100}}, {Type: provider.ChunkDone}}}
|
||||
if mode == "failure_fallback" {
|
||||
planner.chunks[2] = provider.Chunk{Type: provider.ChunkError, Err: errors.New("fixture failure")}
|
||||
}
|
||||
exec := &mockProvider{name: "executor", streams: coordinatorBudgetRounds()}
|
||||
reads := &coordinatorBudgetRead{}
|
||||
registry := tool.NewRegistry()
|
||||
registry.Add(reads)
|
||||
executor := agent.New(exec, registry, sessionstore.NewSession("executor"), agent.Options{TaskBudget: agent.TaskBudget{Tokens: 2000}}, event.Discard)
|
||||
var plannerTools *tool.Registry
|
||||
if mode == "tool_planner" {
|
||||
plannerTools = tool.NewRegistry()
|
||||
}
|
||||
route := agent.PlannerRoutePlanAndExecute
|
||||
if mode == "plan_only_continuation" {
|
||||
route = agent.PlannerRoutePlanOnly
|
||||
}
|
||||
if mode == "approval_declined_continuation" || mode == "approval_allowed" {
|
||||
route = agent.PlannerRoutePlanForApproval
|
||||
}
|
||||
coord := NewCoordinatorWithPlannerPolicy(planner, sessionstore.NewSession("planner"), nil, plannerTools, agent.Options{}, executor, 0, event.Discard, func(context.Context, string) agent.PlannerDecision {
|
||||
return agent.PlannerDecision{Route: route, Depth: agent.PlannerDepthFull}
|
||||
})
|
||||
coord.SetPlannerPlanApprover(&coordinatorApprovalGate{allow: mode == "approval_allowed"})
|
||||
ctx := agent.WithDeliveryExecutionScope(t.Context(), agent.DeliveryExecutionScope{ID: "shared-root", TaskText: "read the fixture"})
|
||||
err := coord.Run(ctx, "read the fixture")
|
||||
if mode == "plan_only_continuation" || mode == "approval_declined_continuation" {
|
||||
if err != nil || reads.calls != 0 {
|
||||
t.Fatalf("unapproved execution calls=%d err=%v", reads.calls, err)
|
||||
}
|
||||
route = agent.PlannerRouteExecutorOnly
|
||||
err = coord.Run(ctx, "continue")
|
||||
}
|
||||
if p, ok := agent.InspectRunPause(err); !ok || p.Key != "token" || reads.calls != 1 || len(exec.requests) != 2 {
|
||||
t.Fatalf("read=%d HTTP=%d pause=%+v err=%v", reads.calls, len(exec.requests), p, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
- 修复:规划与执行共用显式配置的根 token 和时间软预算;保留仅规划、审批、继续、换任务及后台作业边界。关联 #14025。
|
||||
- Fixed: planning and execution share configured root token/wall soft budgets, preserving plan-only, approval, continuation, task replacement and background-job boundaries. Refs #14025.
|
||||
Reference in new issue
Block a user