Share explicit wall budgets through Agent delegation

Problem: a host wall budget could land the root Agent while delegated Agents
and Manager-owned background jobs kept independent start times or no wall bound.

Root cause: the opaque inherited root scope carried only token accounting; the
existing task time gate read each Agent's own clock after its tool rounds.

Fix: attach an immutable root task start/limit pair to that existing scope.
Preserve the start across continuations and snapshot new root overrides without
rewriting old job bounds. New tasks/reset use a new clock. The existing task and
fleet projection helper also carries this sidecar while keeping Manager-owned
cancellation/deadlines. Use existing time soft landing and finalization; keep
defaults unlimited, with no hard deadline, new timer, schema or permission.

Verification: frozen6730Go/fourmodules; required2+rest237 cover239packages,
23754test/subtestpasses/215existingskips. Complete Agent/delegation/jobs race
2652passes/10existing skips; actual boot background task/capability/fleet4race
passes; accepted14523sixGo selectedoverlay54racepasses; prefixguards4passes.
Exact preceding owner compiles and fails9events/2controls across2packages.
Windows17-event step passes locally and rejects skip/missingterminal controls;
six Windows/Linuxamd64 test binaries compile, not native execution.
Rootvet/full lint/makecheck/format/diff pass; baseline byte-identical2663.

Refs #14025
Refs #14133
This commit is contained in:
SivanCola committed 2026-10-26 15:15:08 +08:00
1 parent ebab571cdc
commit 1e17cad975
9 files changed
+455 -10

No files matched your search

+32
View File
@@ -294,6 +294,38 @@ jobs:
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
shell: bash
run: |
set -o pipefail
go test -count=1 -timeout=3m -json -run '^Test(RootWall|EffectRootWall)' ./internal/runtime/agent ./internal/assembly/boot | tee "$RUNNER_TEMP/root-wall-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
(["TestRootWallBudgetReachesDelegatedAgents", "TestRootWallBudgetReachesDelegatedAgents/foreground",
"TestRootWallBudgetReachesDelegatedAgents/nested", "TestRootWallBudgetReachesDelegatedAgents/siblings",
"TestRootWallBudgetReachesDelegatedAgents/context", "TestRootWallBudgetReachesDelegatedAgents/unconfigured",
"TestRootWallBudgetReachesDelegatedAgents/explicit_zero", "TestRootWallBudgetContinuationAndReset",
"TestRootWallBudgetContinuationAndReset/continuation", "TestRootWallBudgetContinuationAndReset/new_task",
"TestRootWallBudgetContinuationAndReset/reset", "TestRootWallBudgetContinuationAndReset/override",
"TestRootWallBudgetProjectionKeepsJobDeadline", "TestEffectRootWallBudgetReachesManagedJobs",
"TestEffectRootWallBudgetReachesManagedJobs/task", "TestEffectRootWallBudgetReachesManagedJobs/capability",
"TestEffectRootWallBudgetReachesManagedJobs/fleet"] - $tests | length) == 0
' "$RUNNER_TEMP/root-wall-budget.jsonl"
- name: upload Windows root wall 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-wall-budget-windows-${{ github.run_attempt }}
path: ${{ runner.temp }}/root-wall-budget.jsonl
if-no-files-found: ignore
retention-days: 7
- name: test (Windows root token budget)
if: env.RUN_STEPS == 'true' && runner.os == 'Windows' && github.event_name == 'pull_request'
timeout-minutes: 5
+11 -7
View File
@@ -27,13 +27,17 @@ root task, root session replacement or explicit root reset starts another counte
不能通过局部 override 放宽继承的根阈值。继续同一根任务会保留计数器;新根任务、
替换根会话或显式重置根预算则使用另一个计数器。
Task and fleet background jobs project only the token scope through
`agent.InheritRootTokenBudget`. Job Manager cancellation and deadlines remain
authoritative. A parent turn ending does not erase the handed-off scope. An old
job still charges its old counter after the parent starts a different task.
Task and fleet background jobs project the opaque root scope through
`agent.InheritRootTokenBudget`. It carries token bounds and explicit
[wall bounds](ROOT_WALL_BUDGET_SECURITY.md), while Job Manager cancellation and
deadlines remain authoritative.
task 和 fleet 后台 job 通过 `agent.InheritRootTokenBudget` 仅传递 token scope,
取消与截止时间仍由 Job Manager 掌握。父 turn 结束不会抹掉已交出的 scope。父任务
A parent turn ending does not erase the handed-off scope. An old job still
charges its old counter after the parent starts a different task.
task 和 fleet 后台 job 通过 `agent.InheritRootTokenBudget` 传递不透明根 scope,
包括 token 与显式[耗时门禁](ROOT_WALL_BUDGET_SECURITY.md),取消与截止时间仍由
Job Manager 掌握。父 turn 结束不会抹掉已交出的 scope。父任务
换成另一项工作后,旧 job 仍记入旧计数器,不会扣减新任务的预算。
## Authority and limits / 权限与边界
@@ -58,7 +62,7 @@ request by this gate. A missing usage report is unknown, not proof of free work.
Auxiliary provider calls outside `observeRunBudget`, remote processes, durable
restart accounting and a shared planner/executor Coordinator root are not
covered. Existing cost and wall-time gates and per-Agent budget telemetry retain
covered. Existing local cost/time gates and per-Agent budget telemetry retain
their meanings. No currency amounts are mixed; unpriced models still count tokens.
`observeRunBudget` 之外的辅助模型调用、远程进程、持久化重启记账,以及跨 planner
+90
View File
@@ -0,0 +1,90 @@
---
owner: @SivanCola
backup: @esengine
status: active
reviewed: 2026-10-26
---
# Agent-root wall budget / Agent 根任务耗时预算
An explicit host `agent.TaskBudget.Wall` now follows the Agent-root delegation
scope. The existing local time gate remains; a child also checks elapsed time
from the root task's start after its tool round. Defaults remain unlimited.
host 显式传入的 `agent.TaskBudget.Wall` 现在跟随 Agent 根委派 scope。既有局部
时间门禁保留;子任务在工具轮结束后,也会检查从根任务起点累计的耗时。默认仍不限。
## Clock and authority / 时钟与权限
The opaque scope carries an immutable start/limit pair alongside the existing
token counter. The start comes from the root's existing task/turn clock.
Derived contexts share the pair; child overrides cannot relax an inherited
root limit. Local narrower limits still apply first.
不透明 scope 在既有 token counter 旁携带不可变的起点与阈值,起点来自根任务
既有 task/turn 时钟。派生 context 共享它;子任务 override 不能放宽继承的根阈值,
更严格的局部门禁仍优先适用。
The existing `InheritRootTokenBudget` API name is retained. Task and fleet job
producers already copy the opaque scope, so they carry the wall bound too.
They retain Job Manager cancellation and deadlines; a parent turn ending or
being canceled does not cancel those contexts.
保留既有 `InheritRootTokenBudget` API 名称。task 与 fleet 的 job producer 已复制
不透明 scope,因此也携带耗时门禁。取消与截止时间仍由 Job Manager 掌握;父 turn
结束或被取消不会因此取消 job context。
## Lifetime / 寿命
A root continuation preserves the task start. New tasks or explicit budget
reset start a new clock. Old jobs retain their old immutable pair. A new root
turn snapshots its current explicit limit; it does not rewrite limits already
handed to older jobs.
继续同一根任务会保留 task 起点;新任务或显式重置预算启用新时钟。旧 job 保留
原来的不可变起点与阈值。新根 turn 捕获当前显式阈值,不会改写此前已交给旧 job
的阈值。
Elapsed time includes queueing and gaps between continuations while the same
ledger remains active. It is not CPU time. The scope is process-local and is
not a durable deadline across restart. An unknown start cannot prove expiry.
同一账本存续时,耗时包含排队及 continuation 间隔,不是 CPU 时间。scope 仅在
进程内有效,不是跨重启的持久截止时间。未知起点不能证明已经超时。
## Landing and compatibility / 收尾与兼容
Expiry uses the existing host-owned time pause and finalization round.
It does not cancel requests, install a timer, reject queued work before launch
or terminate processes. Each child can execute an in-flight round and its
existing finalization before landing.
超时沿用既有 host-owned time pause 与收尾轮,不取消请求、不安装 timer、不在
启动前拒绝排队任务,也不终止进程。各子任务仍可能完成进行中的轮次及既有收尾轮。
No new configuration field, public API, persisted shape, wire, tool schema or
permission is added. This uses existing host Options/context budgets; it does
not add a user-facing wall-budget setting or change provider-visible prefixes.
Local cost telemetry and token accounting remain unchanged.
没有新增配置字段、公共 API、持久化形状、wire、工具 schema 或权限。使用既有
host Options/context 预算,不新增面向用户的耗时设置,也不改写供应商可见前缀。
局部费用遥测及 token 计账保持不变。
## Evidence boundary / 验证边界
Agent regressions exercise foreground/nested/concurrent calls and disabled
limits. Clock lifetime tests select the existing host continuation flag.
Real boot runs task, capability task and fleet background jobs after the parent
has returned and been canceled, with actual fixture reads.
Agent 回归覆盖前台、嵌套、并发调用及禁用阈值。时钟寿命测试选择既有 host
continuation 标记。真实 boot 让后台 task、capability task、fleet 在父 turn
返回并取消后继续执行,并验证真实夹具读取。
This does not establish physical process quiescence, root currency aggregation,
auxiliary provider admission, a shared Coordinator root, durable recovery or the
complete concurrency/isolation plan. Author analysis is not independent review.
该机制不证明物理进程静止、根币种汇总、辅助 provider 准入、共享 Coordinator 根预算、
持久恢复或完整并发隔离方案。作者分析不等于独立审查。
@@ -0,0 +1,105 @@
package boot
import (
"context"
"strings"
"sync/atomic"
"testing"
"time"
"reasonix/internal/contract/event"
"reasonix/internal/contract/provider"
"reasonix/internal/runtime/agent"
"reasonix/internal/session/control"
)
func TestEffectRootWallBudgetReachesManagedJobs(t *testing.T) {
for _, mode := range []string{"task", "capability", "fleet"} {
t.Run(mode, func(t *testing.T) {
home, root := statedBootHome(t), t.TempDir()
unblock := make(chan struct{})
children := []*rootTokenEffectProvider{{wait: unblock}, {wait: unblock}}
count := 1
parent := &rootTokenEffectProvider{first: func() *provider.ToolCall {
args := map[string]any{"prompt": "read the fixture", "model": "child-a", "profile": "explore", "tools": []string{"read_file"}, "max_steps": 8, "run_in_background": true}
if mode == "fleet" {
count = 2
tasks := []map[string]any{}
for _, model := range []string{"child-a", "child-b"} {
tasks = append(tasks, map[string]any{"prompt": "read the fixture", "model": model, "read_only": true, "tools": []string{"read_file"}, "max_steps": 8})
}
return browserCall("delegate", "use_capability", map[string]any{"action": "call", "capability_id": "tool:fleet", "arguments": map[string]any{"tasks": tasks, "run_in_background": true}})
}
if mode == "capability" {
return browserCall("delegate", "use_capability", map[string]any{"action": "call", "capability_id": "tool:task", "arguments": args})
}
return browserCall("delegate", "task", args)
}}
kind := uniqueKind("root-wall")
provider.Register(kind, func(cfg provider.Config) (provider.Provider, error) {
switch cfg.Model {
case "child-a":
return children[0], nil
case "child-b":
return children[1], nil
default:
return parent, nil
}
})
writeFile(t, root, "fixture.txt", "fixture content")
writeFile(t, root, "reasonix.toml", `default_model = "parent"
[codegraph]
enabled = false
[[providers]]
name = "parent"
kind = "`+kind+`"
model = "parent"
[[providers]]
name = "child-a"
kind = "`+kind+`"
model = "child-a"
[[providers]]
name = "child-b"
kind = "`+kind+`"
model = "child-b"
`)
approveWorkspaceAt(t, home, root)
var reads atomic.Int32
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, "fixture content") {
reads.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)
runCtx, cancel := context.WithCancel(agent.WithTaskBudget(t.Context(), agent.TaskBudget{Wall: time.Nanosecond}))
defer cancel()
runErr := ctrl.Run(runCtx, "delegate a fixture read")
cancel()
close(unblock)
pause, ok := agent.InspectRunPause(runErr)
if !ok || pause.Key != "time" || !pause.HostOwned {
t.Errorf("root ended with %v (%+v), want host-owned wall landing", runErr, pause)
}
done := make(chan struct{})
ctrl.JobManager().AfterUnfinished(func() { close(done) })
select {
case <-done:
case <-time.After(10 * time.Second):
t.Fatal("managed children did not finish")
}
for i, child := range children[:count] {
if child.calls.Load() != 2 {
t.Errorf("child%d requests = %d, want one read and one existing finalization round", i, child.calls.Load())
}
}
if reads.Load() != int32(count) || parent.calls.Load() != 2 {
t.Errorf("completed reads=%d parent requests=%d, want %d/2 after parent cancellation", reads.Load(), parent.calls.Load(), count)
}
})
}
}
+16 -3
View File
@@ -11,12 +11,13 @@ import (
type rootTokenBudget struct {
used *atomic.Uint64
limit uint64
wall rootWallBudget
}
type rootTokenBudgetKey struct{}
// InheritRootTokenBudget carries only the Agent-root token scope into a job.
// The job keeps its own cancellation and deadline authority.
// InheritRootTokenBudget carries the opaque Agent-root budget scope into a job.
// Token and wall bounds share its lifetime; job cancellation remains separate.
func InheritRootTokenBudget(ctx, parent context.Context) context.Context {
if scope, ok := parent.Value(rootTokenBudgetKey{}).(*rootTokenBudget); ok {
return context.WithValue(ctx, rootTokenBudgetKey{}, scope)
@@ -32,7 +33,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})
}
scope = &rootTokenBudget{used: previous.used, limit: uint64(max(0, a.taskBudgetLimit(ctx).Tokens))}
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
}
}
wall.limit = limit.Wall
scope = &rootTokenBudget{used: previous.used, limit: uint64(max(0, limit.Tokens)), wall: wall}
a.task.budget.rootTokens = scope
ctx = context.WithValue(ctx, rootTokenBudgetKey{}, scope)
}
@@ -64,5 +74,8 @@ func (a *Agent) exceededTaskBudget(ctx context.Context, state *turnRuntime) (str
if root := state.budget.rootTokens; root != nil && root.limit > 0 && root.used.Load() >= root.limit {
return "token", fmt.Sprintf("root task reached its %d token budget across delegated runs", root.limit)
}
if root := state.budget.rootTokens; root != nil {
return root.wall.exceeded()
}
return "", ""
}
@@ -0,0 +1,20 @@
package agent
import (
"fmt"
"time"
)
type rootWallBudget struct {
started time.Time
limit time.Duration
}
func (b rootWallBudget) exceeded() (string, string) {
if b.limit > 0 && !b.started.IsZero() {
if elapsed := time.Since(b.started); elapsed >= b.limit {
return "time", fmt.Sprintf("root task ran %s, past the %s budget across delegated runs", elapsed.Round(time.Second), b.limit)
}
}
return "", ""
}
@@ -0,0 +1,98 @@
package agent
import (
"context"
"encoding/json"
"testing"
"time"
"reasonix/internal/contract/event"
"reasonix/internal/contract/tool"
"reasonix/internal/state/sessionstore"
)
type wallBudgetDelayTool struct {
readProbe
next tool.Tool
}
func (t *wallBudgetDelayTool) Execute(ctx context.Context, args json.RawMessage) (string, error) {
timer := time.NewTimer(time.Millisecond)
defer timer.Stop()
select {
case <-timer.C:
return t.next.Execute(ctx, args)
case <-ctx.Done():
return "", ctx.Err()
}
}
func TestRootWallBudgetReachesDelegatedAgents(t *testing.T) {
for _, mode := range []string{"foreground", "nested", "siblings", "context", "unconfigured", "explicit_zero"} {
t.Run(mode, func(t *testing.T) {
leafProvider := &spendingProvider{max: 10}
leafTools := tool.NewRegistry()
leafTools.Add(readProbe{})
leaf := New(leafProvider, leafTools, sessionstore.NewSession("leaf"), Options{}, event.Discard)
call := &tokenBudgetChildTool{child: leaf, budget: &TaskBudget{}}
var delegate tool.Tool = call
var siblings *tokenBudgetSiblings
switch mode {
case "nested":
middleTools := tool.NewRegistry()
middleTools.Add(call)
middle := New(&spendingProvider{max: 1}, middleTools, sessionstore.NewSession("middle"), Options{}, event.Discard)
delegate = &tokenBudgetChildTool{child: middle}
case "siblings":
otherTools := tool.NewRegistry()
otherTools.Add(readProbe{})
other := New(&spendingProvider{max: 10}, otherTools, sessionstore.NewSession("sibling"), Options{}, event.Discard)
siblings = &tokenBudgetSiblings{children: []*Agent{leaf, other}}
delegate = siblings
}
rootTools := tool.NewRegistry()
rootTools.Add(&wallBudgetDelayTool{next: delegate})
limit := TaskBudget{Wall: time.Nanosecond}
if mode == "unconfigured" || mode == "context" {
limit = TaskBudget{}
}
root := New(&spendingProvider{max: 1}, rootTools, sessionstore.NewSession("root"), Options{TaskBudget: limit}, event.Discard)
ctx := t.Context()
switch mode {
case "context":
ctx = WithTaskBudget(ctx, TaskBudget{Wall: time.Nanosecond})
case "explicit_zero":
ctx = WithTaskBudget(ctx, TaskBudget{})
}
err := root.Run(ctx, "delegate reads")
bounded := mode != "unconfigured" && mode != "explicit_zero"
want := int32(11)
if bounded {
want = 2
}
if leafProvider.rounds.Load() != want {
t.Fatalf("leaf requests = %d, want %d including its existing finalization round", leafProvider.rounds.Load(), want)
}
if !bounded {
if err != nil || call.err != nil {
t.Fatalf("disabled budget stopped root=%v child=%v", err, call.err)
}
return
}
pause, ok := InspectRunPause(err)
if !ok || pause.Kind != "task_budget" || pause.Key != "time" || !pause.HostOwned {
t.Fatalf("root pause = %+v (%v), want host-owned time budget", pause, err)
}
if siblings != nil {
for i, childErr := range siblings.errors {
pause, ok := InspectRunPause(childErr)
if !ok || pause.Key != "time" || siblings.children[i].svc.prov.(*spendingProvider).rounds.Load() != 2 {
t.Fatalf("sibling%d did not land on root time: %v", i, childErr)
}
}
} else if pause, ok := InspectRunPause(call.err); !ok || pause.Key != "time" {
t.Fatalf("leaf pause = %+v (%v), want root time", pause, call.err)
}
})
}
}
@@ -0,0 +1,81 @@
package agent
import (
"context"
"testing"
"time"
"reasonix/internal/contract/event"
"reasonix/internal/contract/tool"
"reasonix/internal/state/sessionstore"
)
func TestRootWallBudgetContinuationAndReset(t *testing.T) {
for _, mode := range []string{"continuation", "new_task", "reset", "override"} {
t.Run(mode, func(t *testing.T) {
root := New(&spendingProvider{}, tool.NewRegistry(), sessionstore.NewSession("root"), Options{TaskBudget: TaskBudget{Wall: time.Hour}}, event.Discard)
if err := root.Run(t.Context(), "first work"); err != nil {
t.Fatal(err)
}
old := root.task.budget.rootTokens
started := old.wall.started
if started.IsZero() || old.wall.limit != time.Hour {
t.Fatalf("root wall = %+v, want a real task start and explicit limit", old.wall)
}
time.Sleep(time.Millisecond)
if mode != "new_task" {
root.pending.preserveEvidence = true
}
if mode == "reset" {
root.ResetTaskBudget()
}
ctx := t.Context()
if mode == "override" {
ctx = WithTaskBudget(ctx, TaskBudget{Wall: 2 * time.Hour})
}
if err := root.Run(ctx, "second work"); err != nil {
t.Fatal(err)
}
same := root.task.budget.rootTokens.wall.started.Equal(started)
if same != (mode == "continuation" || mode == "override") {
t.Fatalf("same root start = %v for %s", same, mode)
}
if mode == "override" && root.task.budget.rootTokens.wall.limit != 2*time.Hour {
t.Fatal("new root turn did not snapshot its explicit override")
}
if !old.wall.started.Equal(started) || old.wall.limit != time.Hour {
t.Fatal("a new run mutated the wall budget already handed to an old job")
}
})
}
}
func TestRootWallBudgetProjectionKeepsJobDeadline(t *testing.T) {
root := &rootTokenBudget{wall: rootWallBudget{started: time.Now().Add(-time.Hour), limit: time.Minute}}
parent, cancelParent := context.WithTimeout(context.WithValue(t.Context(), rootTokenBudgetKey{}, root), 2*time.Hour)
defer cancelParent()
job, cancelJob := context.WithTimeout(t.Context(), time.Hour)
defer cancelJob()
projected := InheritRootTokenBudget(job, parent)
want, _ := job.Deadline()
got, ok := projected.Deadline()
if !ok || !got.Equal(want) || projected.Value(rootTokenBudgetKey{}) != root {
t.Fatal("budget projection changed job deadline or lost the immutable root scope")
}
cancelParent()
if projected.Err() != nil {
t.Fatal("parent cancellation canceled the Manager-owned job")
}
if axis, _ := root.wall.exceeded(); axis != "time" {
t.Fatalf("axis = %q, want inherited elapsed limit without context cancellation", axis)
}
cancelJob()
if projected.Err() != context.Canceled {
t.Fatal("job cancellation no longer controls its context")
}
for _, wall := range []rootWallBudget{{limit: time.Second}, {started: time.Now().Add(-time.Hour)}} {
if axis, _ := wall.exceeded(); axis != "" {
t.Fatalf("axis = %q, an unknown start or disabled limit cannot certify a crossing", axis)
}
}
}
@@ -0,0 +1,2 @@
- 修复:host 显式耗时预算跟随 Agent 委派树及后台 task/fleet,共享任务起点并保留 Job Manager 取消权和既有软收尾;默认仍不限。关联 #14025。
- Fixed: explicit host wall budgets follow Agent delegation and background jobs, retaining Manager cancellation and soft landing. Defaults stay unlimited. Refs #14025.