From f35d56a88dfaf6136c4a1c9e267b51449bc32753 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Sat, 19 Sep 2026 13:08:17 -0400 Subject: [PATCH 01/11] test: expose live run cost limit gap --- internal/run/cost_limit_test.go | 99 +++++++++++++++++++++++++++++++++ 1 file changed, 99 insertions(+) create mode 100644 internal/run/cost_limit_test.go diff --git a/internal/run/cost_limit_test.go b/internal/run/cost_limit_test.go new file mode 100644 index 0000000000..64120c007e --- /dev/null +++ b/internal/run/cost_limit_test.go @@ -0,0 +1,99 @@ +package run + +import ( + "context" + "sync/atomic" + "testing" + + "github.com/Agent-Field/codeaf/internal/plandb" +) + +func TestCostLimitEndsAndAbsorbsWorkingWorker(t *testing.T) { + store, err := plandb.Open(t.TempDir()+"/plan.json", "cost-limit", "root", "root", "run until the limit") + if err != nil { + t.Fatalf("open store: %v", err) + } + defer store.Close() + + ended := make(chan struct{}) + var launches atomic.Int32 + factory := func(task plandb.Task) Worker { + launches.Add(1) + return workerFunc(func(ctx context.Context, task plandb.Task) (Report, error) { + bankSpend(ctx, 1.25) + <-ctx.Done() + if _, err := store.AddMany([]plandb.TaskSpec{{ID: "after-limit", Title: "must not launch", ParentID: task.ID}}); err != nil { + t.Errorf("add post-limit task: %v", err) + } + close(ended) + return Report{USD: 1.25}, ctx.Err() + }) + } + supervisor := NewSupervisor(store, t.TempDir(), 2, Limits{CostUSD: 1}, factory) + if outcome := supervisor.Run(context.Background()); outcome != OutcomeLimit { + t.Fatalf("outcome = %q, want %q", outcome, OutcomeLimit) + } + select { + case <-ended: + default: + t.Fatal("Run returned before its in-flight worker ended and was absorbed") + } + if supervisor.spent != 1.25 { + t.Fatalf("run spend = %v, want 1.25 exactly once", supervisor.spent) + } + if supervisor.inFlight != 0 || len(supervisor.cancels) != 0 { + t.Fatalf("run retained workers: inFlight=%d cancels=%d", supervisor.inFlight, len(supervisor.cancels)) + } + if got := launches.Load(); got != 1 { + t.Fatalf("worker launches = %d, want only the worker cut at the limit", got) + } + if task := store.Task("after-limit"); task == nil || task.Status != plandb.StatusReady { + t.Fatalf("post-limit task = %#v, want ready and unlaunched", task) + } +} + +func TestCostLimitSumsWorkingWorkers(t *testing.T) { + store, err := plandb.Open(t.TempDir()+"/plan.json", "cost-limit-sum", "root", "root", "run until the limit") + if err != nil { + t.Fatalf("open store: %v", err) + } + defer store.Close() + if _, err := store.AddMany([]plandb.TaskSpec{{ID: "peer", Title: "peer", ParentID: store.RootID()}}); err != nil { + t.Fatalf("add peer: %v", err) + } + + bothStarted := make(chan struct{}) + var started atomic.Int32 + var ended atomic.Int32 + factory := func(task plandb.Task) Worker { + return workerFunc(func(ctx context.Context, task plandb.Task) (Report, error) { + if started.Add(1) == 2 { + close(bothStarted) + } + <-bothStarted + bankSpend(ctx, 0.6) + <-ctx.Done() + ended.Add(1) + return Report{USD: 0.6}, ctx.Err() + }) + } + supervisor := NewSupervisor(store, t.TempDir(), 2, Limits{CostUSD: 1}, factory) + if outcome := supervisor.Run(context.Background()); outcome != OutcomeLimit { + t.Fatalf("outcome = %q, want %q", outcome, OutcomeLimit) + } + if got := ended.Load(); got != 2 { + t.Fatalf("absorbed worker returns = %d, want 2", got) + } + if supervisor.spent != 1.2 { + t.Fatalf("run spend = %v, want 1.2 exactly once", supervisor.spent) + } + if supervisor.inFlight != 0 || len(supervisor.cancels) != 0 { + t.Fatalf("run retained workers: inFlight=%d cancels=%d", supervisor.inFlight, len(supervisor.cancels)) + } +} + +func TestSpendBankWithoutLimitDoesNotEndWorker(t *testing.T) { + ctx := WithSpendBank(context.Background(), func(float64) { t.Fatal("a missing limit must not install a supervisor callback") }) + bankSpend(context.Background(), 3) + _ = ctx +} From c4ff85810a69d7a1a5b83de337afc40783f28c66 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Sat, 19 Sep 2026 13:09:26 -0400 Subject: [PATCH 02/11] run: enforce cost limit while workers are active --- internal/run/bashworker.go | 1 + internal/run/cost_limit_test.go | 23 ++++++++-- internal/run/run.go | 75 +++++++++++++++++++++++---------- internal/run/worker.go | 25 +++++++++-- 4 files changed, 94 insertions(+), 30 deletions(-) diff --git a/internal/run/bashworker.go b/internal/run/bashworker.go index e59e2fe525..0726a6898a 100644 --- a/internal/run/bashworker.go +++ b/internal/run/bashworker.go @@ -214,6 +214,7 @@ func (w *BashWorker) Run(ctx context.Context, task plandb.Task) (Report, error) usd += event.Usage.CostUSD inTok += event.Usage.Input outTok += event.Usage.Output + bankSpend(runCtx, usd) case session.EventError: if turnErr == nil { turnErr = event.Err diff --git a/internal/run/cost_limit_test.go b/internal/run/cost_limit_test.go index 64120c007e..36cd43f587 100644 --- a/internal/run/cost_limit_test.go +++ b/internal/run/cost_limit_test.go @@ -92,8 +92,23 @@ func TestCostLimitSumsWorkingWorkers(t *testing.T) { } } -func TestSpendBankWithoutLimitDoesNotEndWorker(t *testing.T) { - ctx := WithSpendBank(context.Background(), func(float64) { t.Fatal("a missing limit must not install a supervisor callback") }) - bankSpend(context.Background(), 3) - _ = ctx +func TestSpendBankWithoutLimitLetsWorkerReturn(t *testing.T) { + store, err := plandb.Open(t.TempDir()+"/plan.json", "unlimited", "root", "root", "finish") + if err != nil { + t.Fatalf("open store: %v", err) + } + defer store.Close() + factory := func(task plandb.Task) Worker { + return workerFunc(func(ctx context.Context, task plandb.Task) (Report, error) { + bankSpend(ctx, 3) + return Report{Result: "done", USD: 3}, nil + }) + } + supervisor := NewSupervisor(store, t.TempDir(), 1, Limits{}, factory) + if outcome := supervisor.Run(context.Background()); outcome != OutcomeDone { + t.Fatalf("outcome = %q, want %q", outcome, OutcomeDone) + } + if supervisor.spent != 3 { + t.Fatalf("run spend = %v, want returned 3 exactly once", supervisor.spent) + } } diff --git a/internal/run/run.go b/internal/run/run.go index 84c233be44..40f3621c63 100644 --- a/internal/run/run.go +++ b/internal/run/run.go @@ -54,6 +54,12 @@ type workerReturn struct { err error } +type workerEvent struct { + ret *workerReturn + taskID string + banked float64 +} + // Supervisor is the launch loop over one plan store. It claims ready leaves // as the task's own agent — the name its finish command answers to — and // records the process that holds the claim beside it, so a run is owned per @@ -94,7 +100,7 @@ type Supervisor struct { // touched by no one else. cancels maps a running task to the context // cancel that ends its worker, so a pass can end one worker's context // without touching the rest. - finished chan workerReturn + finished chan workerEvent // workers counts the goroutines launch has started and not yet seen end. // It is the run's one promise to its caller — NO WORKER OUTLIVES ITS RUN — // and drain is its only reader: every road out of Run waits on it before @@ -103,6 +109,7 @@ type Supervisor struct { workers sync.WaitGroup inFlight int spent float64 + banked map[string]float64 nodes int steps int rootResult string @@ -147,7 +154,7 @@ func NewSupervisor(store *plandb.Store, workspace string, slots int, limits Limi limits: limits, factory: factory, after: time.After, - finished: make(chan workerReturn, slots+1), + finished: make(chan workerEvent, slots+1), cancels: make(map[string]context.CancelFunc), checkOf: make(map[string]string), } @@ -183,6 +190,7 @@ func (s *Supervisor) Run(ctx context.Context) Outcome { return OutcomeCannotRun } s.spent = 0 + s.banked = make(map[string]float64) s.nodes = 0 s.steps = 0 s.rootResult = "" @@ -216,9 +224,8 @@ func (s *Supervisor) Run(ctx context.Context) Outcome { return outcome } select { - case ret := <-s.finished: - s.inFlight-- - s.absorb(ret) + case event := <-s.finished: + s.consume(event) case <-timer.C: timer.Reset(passInterval) case <-elapsed: @@ -226,9 +233,8 @@ func (s *Supervisor) Run(ctx context.Context) Outcome { // spend counter marks, so the next pass launches nothing and answers // the existing OutcomeLimit. // - // THE DIFFERENCE IS THE WORK IN FLIGHT. A cost limit lets it finish, - // because what it will spend is already committed; a time limit - // cannot, because the time is gone. So every worker is ended here, and + // TIME AND COST BOTH END WORK IN FLIGHT. Once either ceiling is + // reached, every worker is ended here, and // ITS ENDING IS ABSORBED, the way the caller's wall below reads its // endings: a return that was dropped would leave its task claimed and // reading as running on a run that is over, and what the worker spent @@ -239,9 +245,7 @@ func (s *Supervisor) Run(ctx context.Context) Outcome { cancel() } for s.inFlight > 0 { - ret := <-s.finished - s.inFlight-- - s.absorb(ret) + s.consume(<-s.finished) } case <-ctx.Done(): // The caller's wall: workers still out there were handed this @@ -251,9 +255,7 @@ func (s *Supervisor) Run(ctx context.Context) Outcome { // the endings rather than dropping them, because the run it stops // is one a later pass picks up from the store. for s.inFlight > 0 { - ret := <-s.finished - s.inFlight-- - s.absorb(ret) + s.consume(<-s.finished) } s.drain() return OutcomeIncomplete @@ -349,6 +351,12 @@ func (s *Supervisor) launch(ctx context.Context, task plandb.Task, wake string) taskCtx = WithWakeClause(taskCtx, wake) } s.cancels[task.ID] = cancel + s.banked[task.ID] = 0 + if s.limits.CostUSD > 0 { + taskCtx = WithSpendBank(taskCtx, func(usd float64) { + s.finished <- workerEvent{taskID: task.ID, banked: usd} + }) + } s.inFlight++ s.nodes++ // THE GOROUTINE IS COUNTED BEFORE IT STARTS, so a drain that begins the @@ -362,7 +370,8 @@ func (s *Supervisor) launch(ctx context.Context, task plandb.Task, wake string) } else { report, err = worker.Run(WithStepsPerTask(taskCtx, s.limits.StepsPerTask), task) } - s.finished <- workerReturn{task: task, report: report, err: err} + ret := workerReturn{task: task, report: report, err: err} + s.finished <- workerEvent{ret: &ret} }() } @@ -391,6 +400,26 @@ func (s *Supervisor) drain() { s.workers.Wait() } +// consume serializes live spend and worker returns on the supervisor loop. +func (s *Supervisor) consume(event workerEvent) { + if event.ret != nil { + s.inFlight-- + s.absorb(*event.ret) + return + } + previous := s.banked[event.taskID] + if event.banked > previous { + s.spent += event.banked - previous + s.banked[event.taskID] = event.banked + } + if s.limits.CostUSD > 0 && s.spent >= s.limits.CostUSD && !s.limitHit { + s.limitHit = true + for _, cancel := range s.cancels { + cancel() + } + } +} + // absorb writes one worker's ending into the store and keeps the run's // counters true. A worker that came home well is done with its result; one // that came home with an error fails with that error as the reason. Either @@ -398,6 +427,14 @@ func (s *Supervisor) drain() { // that stays open would stall the whole run: nothing else can make it // terminal, and the run would wait on it forever. func (s *Supervisor) absorb(ret workerReturn) { + banked := s.banked[ret.task.ID] + if ret.report.USD > banked { + s.spent += ret.report.USD - banked + } + delete(s.banked, ret.task.ID) + if s.limits.CostUSD > 0 && s.spent >= s.limits.CostUSD { + s.limitHit = true + } // A RETURN FOR A TASK THE STORE ALREADY ENDED is written as nothing. The // ending the store carries — a cancellation that landed while the worker // ran, which the same write cleared the claim and cascaded down — is the @@ -412,11 +449,7 @@ func (s *Supervisor) absorb(ret workerReturn) { // task stays open until the run wakes it ([launchWaits]). Nothing is // written over the store's own park, and the spend the worker made is // still the run's. - s.spent += ret.report.USD s.steps += ret.report.Steps - if s.limits.CostUSD > 0 && s.spent >= s.limits.CostUSD { - s.limitHit = true - } s.completeTree() return } @@ -429,11 +462,7 @@ func (s *Supervisor) absorb(ret workerReturn) { // counts what the worker spent, and the root's own report is still the // run's result. endedByStore := ret.err == nil && s.store.Task(ret.task.ID).Status == plandb.StatusDone - s.spent += ret.report.USD s.steps += ret.report.Steps - if s.limits.CostUSD > 0 && s.spent >= s.limits.CostUSD { - s.limitHit = true - } if ret.task.ID == s.store.RootID() { // The harness owns the root, and the store refuses worker writes to it. // Its worker's report is kept for the completion; its error is what diff --git a/internal/run/worker.go b/internal/run/worker.go index f96bfde4be..039a5a0abb 100644 --- a/internal/run/worker.go +++ b/internal/run/worker.go @@ -49,9 +49,10 @@ type WorkerFactory func(task plandb.Task) Worker // Elapsed of zero (or less) sets no corresponding run limit, a StepsPerTask // of zero hands the worker no cap, and ReviewRound's false is the run every caller had before it. type Limits struct { - // CostUSD is what the whole run may spend, as the sum of every Report's - // USD. When the counter has reached it no new worker starts and the run - // ends on the limit word of the outcome ladder. + // CostUSD is what the whole run may spend. Banked calls count while a worker + // is still working and reconcile with its final Report. When the counter has + // reached it no new worker starts, work in flight ends, and the run ends on + // the limit word of the outcome ladder. CostUSD float64 // Elapsed is how long the whole run may remain active. When it passes, // workers already in flight are ended and drained, no new worker starts, @@ -84,6 +85,24 @@ type Limits struct { // collide with. type stepsPerTaskKey struct{} +// spendBankKey carries the run-owned observer of a worker banking cumulative +// spend. The cumulative figure is reconciled with the worker return, so the +// live limit and the final receipt share one account. +type spendBankKey struct{} + +// WithSpendBank returns a context that reports cumulative spend as a worker +// banks calls. Workers without this property remain valid and report at return. +func WithSpendBank(ctx context.Context, bank func(float64)) context.Context { + return context.WithValue(ctx, spendBankKey{}, bank) +} + +// bankSpend publishes the cumulative spend banked by this worker. +func bankSpend(ctx context.Context, usd float64) { + if bank, _ := ctx.Value(spendBankKey{}).(func(float64)); bank != nil { + bank(usd) + } +} + // wakeClauseKey is the type behind the context value that carries a woken // parent's resume clause, for the same reason as the step cap beside it: a // typed lookup no other package can collide with. From 756289201ffa5181cfcca31ebcee2a07e6966f51 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Sat, 19 Sep 2026 13:18:42 -0400 Subject: [PATCH 03/11] run: preserve cancelled worker accounting --- internal/run/run.go | 40 +++++++++++++++++++++++----------------- 1 file changed, 23 insertions(+), 17 deletions(-) diff --git a/internal/run/run.go b/internal/run/run.go index 40f3621c63..2f17f292bc 100644 --- a/internal/run/run.go +++ b/internal/run/run.go @@ -407,12 +407,17 @@ func (s *Supervisor) consume(event workerEvent) { s.absorb(*event.ret) return } + // With no ceiling, live reports are observational: returned workers keep + // the shipped accounting semantics, including dropping cancelled work. + if s.limits.CostUSD <= 0 { + return + } previous := s.banked[event.taskID] if event.banked > previous { s.spent += event.banked - previous s.banked[event.taskID] = event.banked } - if s.limits.CostUSD > 0 && s.spent >= s.limits.CostUSD && !s.limitHit { + if s.spent >= s.limits.CostUSD && !s.limitHit { s.limitHit = true for _, cancel := range s.cancels { cancel() @@ -427,14 +432,6 @@ func (s *Supervisor) consume(event workerEvent) { // that stays open would stall the whole run: nothing else can make it // terminal, and the run would wait on it forever. func (s *Supervisor) absorb(ret workerReturn) { - banked := s.banked[ret.task.ID] - if ret.report.USD > banked { - s.spent += ret.report.USD - banked - } - delete(s.banked, ret.task.ID) - if s.limits.CostUSD > 0 && s.spent >= s.limits.CostUSD { - s.limitHit = true - } // A RETURN FOR A TASK THE STORE ALREADY ENDED is written as nothing. The // ending the store carries — a cancellation that landed while the worker // ran, which the same write cleared the claim and cascaded down — is the @@ -443,17 +440,26 @@ func (s *Supervisor) absorb(ret workerReturn) { // hands back no account this run counts. The completion check below still // runs, because the cancellation may be the write that finished the tree. delete(s.cancels, ret.task.ID) - if ret.report.Waiting { - // A PARKED RETURN IS NOT AN ENDING. The worker called `plandb wait`; the - // store already released the claim and flagged the task waiting, and the - // task stays open until the run wakes it ([launchWaits]). Nothing is - // written over the store's own park, and the spend the worker made is - // still the run's. - s.steps += ret.report.Steps + if task := s.store.Task(ret.task.ID); task == nil || task.Status == plandb.StatusCancelled { + // Live spend was provisional. Preserve the existing rule that an + // externally cancelled task contributes no report to this run. + s.spent -= s.banked[ret.task.ID] + delete(s.banked, ret.task.ID) s.completeTree() return } - if task := s.store.Task(ret.task.ID); task == nil || task.Status == plandb.StatusCancelled { + banked := s.banked[ret.task.ID] + if ret.report.USD > banked { + s.spent += ret.report.USD - banked + } + delete(s.banked, ret.task.ID) + if s.limits.CostUSD > 0 && s.spent >= s.limits.CostUSD { + s.limitHit = true + } + if ret.report.Waiting { + // A parked return is not an ending. The store already released the + // claim, and the spend the worker made is still the run's. + s.steps += ret.report.Steps s.completeTree() return } From beba87b3a1ce9994b44ddb431b78fe9ba09d6b83 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Sat, 19 Sep 2026 13:33:00 -0400 Subject: [PATCH 04/11] docs: correct live dollar-limit behavior --- internal/manual/chat/starting-codeaf.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/manual/chat/starting-codeaf.md b/internal/manual/chat/starting-codeaf.md index 9c11c7b533..477f251553 100644 --- a/internal/manual/chat/starting-codeaf.md +++ b/internal/manual/chat/starting-codeaf.md @@ -142,8 +142,8 @@ comes home the ordinary way. The run's row ends `incomplete` and says conversation's spend ceiling. A task that was cut part way through reads `incomplete` too. The dollar limit on a run is the conversation's spend ceiling, not `--max-cost`. A run -that reaches it lets the work already going finish, because that money is already -committed; a time limit cannot wait, so it ends that work. +that reaches it ends the work still going and says `a limit you set stopped it`, just +as when its elapsed-time limit is up. ## Leaving it running on its own · leaving a headless run going with a budget · --once yolo · no screen · unattended · overnight · nobody watching From 8ce3d689ecb3202343aca91e1a1cb21f030ba342 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Sat, 19 Sep 2026 13:33:35 -0400 Subject: [PATCH 05/11] test: keep sub-limit worker spend exact --- internal/run/cost_limit_test.go | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/internal/run/cost_limit_test.go b/internal/run/cost_limit_test.go index 36cd43f587..732b9a8245 100644 --- a/internal/run/cost_limit_test.go +++ b/internal/run/cost_limit_test.go @@ -112,3 +112,26 @@ func TestSpendBankWithoutLimitLetsWorkerReturn(t *testing.T) { t.Fatalf("run spend = %v, want returned 3 exactly once", supervisor.spent) } } + +func TestCostLimitCountsReturnedWorkerBelowCeilingOnce(t *testing.T) { + store, err := plandb.Open(t.TempDir()+"/plan.json", "cost-limit-under", "root", "root", "finish below the limit") + if err != nil { + t.Fatalf("open store: %v", err) + } + defer store.Close() + + factory := func(task plandb.Task) Worker { + return workerFunc(func(ctx context.Context, task plandb.Task) (Report, error) { + bankSpend(ctx, 0.4) + bankSpend(ctx, 0.9) + return Report{Result: "done", USD: 0.9}, nil + }) + } + supervisor := NewSupervisor(store, t.TempDir(), 1, Limits{CostUSD: 1}, factory) + if outcome := supervisor.Run(context.Background()); outcome != OutcomeDone { + t.Fatalf("outcome = %q, want %q", outcome, OutcomeDone) + } + if supervisor.spent != 0.9 { + t.Fatalf("run spend = %v, want cumulative 0.9 exactly once", supervisor.spent) + } +} From ec209c461c1ebdc9ee6d5373c449733b0d7f1f49 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Sat, 19 Sep 2026 15:27:37 -0400 Subject: [PATCH 06/11] run: a worker never waits on the run to say what it spent The live dollar limit carried each worker's figure over the returns channel with a blocking send. Drain waits for workers and reads nothing, so a worker that banked a call while the run drained blocked for good and the run never answered. The figure is now written under a lock and the loop is told with a token that is dropped when it does not fit: the figure is cumulative, so a lost token loses nothing. Returns go back to their own channel, one per worker, as the channel's comment promises. Absorb keeps its shipped order (a parked return first, then a task the store cancelled), and the elapsed case's comment no longer says a cost limit lets work finish. Tests: a worker that reports its spend 64 times after its context ended, on the road where the tree finished while it was still out; and the real worker through a scripted seat, one cumulative figure per paid call, the last equal to its report. Co-Authored-By: Claude Fable 5.1 --- internal/manual/chat/starting-codeaf.md | 9 +- internal/run/bashworker_test.go | 44 ++++++ internal/run/cost_limit_test.go | 60 ++++++++ internal/run/run.go | 182 +++++++++++++++++------- 4 files changed, 238 insertions(+), 57 deletions(-) diff --git a/internal/manual/chat/starting-codeaf.md b/internal/manual/chat/starting-codeaf.md index 477f251553..ff460212f3 100644 --- a/internal/manual/chat/starting-codeaf.md +++ b/internal/manual/chat/starting-codeaf.md @@ -141,9 +141,12 @@ comes home the ordinary way. The run's row ends `incomplete` and says `a limit you set stopped it`, the same sentence a run says when it reaches the conversation's spend ceiling. A task that was cut part way through reads `incomplete` too. -The dollar limit on a run is the conversation's spend ceiling, not `--max-cost`. A run -that reaches it ends the work still going and says `a limit you set stopped it`, just -as when its elapsed-time limit is up. +The dollar limit on a run is the conversation's spend ceiling, not `--max-cost`. A run's +spend is counted as each model call is paid for, while its tasks are still working, so +one long task cannot carry the run far past the figure. A run that reaches it ends the +work still going and says `a limit you set stopped it`, just as when its elapsed-time +limit is up. The call that reached the limit is already paid for, so the run can end a +little over it. ## Leaving it running on its own · leaving a headless run going with a budget · --once yolo · no screen · unattended · overnight · nobody watching diff --git a/internal/run/bashworker_test.go b/internal/run/bashworker_test.go index 251899548d..8e748455fe 100644 --- a/internal/run/bashworker_test.go +++ b/internal/run/bashworker_test.go @@ -509,3 +509,47 @@ func TestBashWorkerEndsItsLoopAtTheStepCap(t *testing.T) { t.Fatalf("the ending reads %d steps with %q, want the cap's count and reason", end.Steps, end.Reason) } } + +// THE WORKER SAYS WHAT IT HAS SPENT AS EACH CALL IS PAID FOR, not only when it +// comes home. A run's dollar limit is read from these figures while the worker +// is still working, so they must arrive once per paid call, each one the whole +// of what the worker has spent so far, and the last one must be the figure the +// report carries: two accounts of one worker's money would drift. +func TestBashWorkerReportsItsSpendAsEachCallIsPaidFor(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "bash") + t.Setenv("CODEAF_PLANDB_BIN", stubCLI(t)) + store := runOpenStore(t) + seat := &seat{ever: func(context.Context, []ai.Message) (*ai.Response, error) { + reply := toolReply(`{"command":"true"}`) + cost := 0.25 + reply.Usage.Cost = &cost + return reply, nil + }} + worker := run.NewBashWorker(store, t.TempDir(), "test/model", seat) + var mu sync.Mutex + var figures []float64 + ctx := run.WithSpendBank(run.WithStepsPerTask(runContext(t), 3), func(usd float64) { + mu.Lock() + figures = append(figures, usd) + mu.Unlock() + }) + + report, _ := worker.Run(ctx, *store.Task(store.RootID())) + + mu.Lock() + defer mu.Unlock() + if len(figures) < 3 { + t.Fatalf("the worker reported its spend %d times over three paid calls: %v", len(figures), figures) + } + for i := 1; i < len(figures); i++ { + if figures[i] <= figures[i-1] { + t.Fatalf("spend figures are not the running whole: %v", figures) + } + } + if figures[0] != 0.25 { + t.Fatalf("the first figure = %v, want the first call's 0.25", figures[0]) + } + if last := figures[len(figures)-1]; last != report.USD { + t.Fatalf("the last figure reported = %v and the report carries %v, want one account", last, report.USD) + } +} diff --git a/internal/run/cost_limit_test.go b/internal/run/cost_limit_test.go index 732b9a8245..cdb40f38d7 100644 --- a/internal/run/cost_limit_test.go +++ b/internal/run/cost_limit_test.go @@ -4,6 +4,7 @@ import ( "context" "sync/atomic" "testing" + "time" "github.com/Agent-Field/codeaf/internal/plandb" ) @@ -135,3 +136,62 @@ func TestCostLimitCountsReturnedWorkerBelowCeilingOnce(t *testing.T) { t.Fatalf("run spend = %v, want cumulative 0.9 exactly once", supervisor.spent) } } + +// A WORKER NEVER WAITS ON THE RUN TO SAY WHAT IT SPENT. The road this pins is +// the one on which nobody is listening: the tree finished while one worker was +// still out, so the run drains, and drain waits for that worker and reads +// nothing. A worker that had to be heard before it could go on would hold the +// run open for good, which is an engine that never answers. The worker here +// reports its spend far more often than any buffer is deep, after its context +// has ended, and the run must still come home. +func TestAWorkerReportingSpendWhileTheRunDrainsDoesNotHoldTheRunOpen(t *testing.T) { + store, err := plandb.Open(t.TempDir()+"/plan.json", "cost-limit-drain", "root", "root", "finish while a worker is still out") + if err != nil { + t.Fatalf("open store: %v", err) + } + defer store.Close() + if _, err := store.AddMany([]plandb.TaskSpec{{ID: "peer", Title: "peer", ParentID: store.RootID()}}); err != nil { + t.Fatalf("add peer: %v", err) + } + + peerEnded := make(chan struct{}) + var reports atomic.Int32 + factory := func(task plandb.Task) Worker { + if task.ID == "peer" { + return workerFunc(func(ctx context.Context, task plandb.Task) (Report, error) { + if _, err := store.Done(task.ID, task.ID, "the peer wrote its own ending", nil, nil); err != nil { + t.Errorf("peer done: %v", err) + } + close(peerEnded) + <-ctx.Done() + for i := 1; i <= 64; i++ { + bankSpend(ctx, float64(i)/1000) + reports.Add(1) + } + return Report{USD: 0.064}, ctx.Err() + }) + } + return workerFunc(func(ctx context.Context, task plandb.Task) (Report, error) { + select { + case <-peerEnded: + case <-ctx.Done(): + return Report{}, ctx.Err() + } + return Report{Result: "the tree is finished"}, nil + }) + } + supervisor := NewSupervisor(store, t.TempDir(), 2, Limits{CostUSD: 100}, factory) + answered := make(chan Outcome, 1) + go func() { answered <- supervisor.Run(context.Background()) }() + select { + case outcome := <-answered: + if outcome != OutcomeDone { + t.Fatalf("outcome = %q, want %q", outcome, OutcomeDone) + } + case <-time.After(20 * time.Second): + t.Fatalf("the run never answered: a worker reporting its spend is holding the drain open (%d of 64 reports made)", reports.Load()) + } + if got := reports.Load(); got != 64 { + t.Fatalf("spend reports made = %d, want all 64", got) + } +} diff --git a/internal/run/run.go b/internal/run/run.go index 2f17f292bc..10934395f2 100644 --- a/internal/run/run.go +++ b/internal/run/run.go @@ -54,12 +54,6 @@ type workerReturn struct { err error } -type workerEvent struct { - ret *workerReturn - taskID string - banked float64 -} - // Supervisor is the launch loop over one plan store. It claims ready leaves // as the task's own agent — the name its finish command answers to — and // records the process that holds the claim beside it, so a run is owned per @@ -100,16 +94,34 @@ type Supervisor struct { // touched by no one else. cancels maps a running task to the context // cancel that ends its worker, so a pass can end one worker's context // without touching the rest. - finished chan workerEvent + finished chan workerReturn // workers counts the goroutines launch has started and not yet seen end. // It is the run's one promise to its caller — NO WORKER OUTLIVES ITS RUN — // and drain is its only reader: every road out of Run waits on it before // answering, so the store, the working copy and the process are the // caller's alone the moment the run is over. - workers sync.WaitGroup - inFlight int - spent float64 - banked map[string]float64 + workers sync.WaitGroup + inFlight int + spent float64 + // THE DOLLAR LIMIT IS READ WHILE THE WORK IS GOING, not only when a worker + // comes home. live is what each worker in flight says it has banked so far, + // a cumulative figure its own goroutine writes under liveMu; counted is how + // much of that figure the loop has already added to spent, and it belongs to + // the loop alone. Both are keyed by task, both are cumulative, and a + // worker's return reconciles them against its report, so a dollar is + // counted once whether the loop saw it on the way or only at the end. + // + // A WORKER NEVER WAITS ON THE RUN TO SAY WHAT IT SPENT. It writes its figure + // and drops a token into liveMoved if there is room; a token that does not + // fit is lost and nothing else is, because the figure is cumulative and the + // next read takes all of it. That matters most on the way out: drain waits + // for workers and reads nothing, so a worker that had to be heard before it + // could go on would hold the run open for good. + liveMu sync.Mutex + live map[string]float64 + counted map[string]float64 + liveMoved chan struct{} + nodes int steps int rootResult string @@ -154,7 +166,8 @@ func NewSupervisor(store *plandb.Store, workspace string, slots int, limits Limi limits: limits, factory: factory, after: time.After, - finished: make(chan workerEvent, slots+1), + finished: make(chan workerReturn, slots+1), + liveMoved: make(chan struct{}, 1), cancels: make(map[string]context.CancelFunc), checkOf: make(map[string]string), } @@ -190,7 +203,10 @@ func (s *Supervisor) Run(ctx context.Context) Outcome { return OutcomeCannotRun } s.spent = 0 - s.banked = make(map[string]float64) + s.liveMu.Lock() + s.live = make(map[string]float64) + s.liveMu.Unlock() + s.counted = make(map[string]float64) s.nodes = 0 s.steps = 0 s.rootResult = "" @@ -224,8 +240,11 @@ func (s *Supervisor) Run(ctx context.Context) Outcome { return outcome } select { - case event := <-s.finished: - s.consume(event) + case ret := <-s.finished: + s.inFlight-- + s.absorb(ret) + case <-s.liveMoved: + s.countLiveSpend() case <-timer.C: timer.Reset(passInterval) case <-elapsed: @@ -233,8 +252,9 @@ func (s *Supervisor) Run(ctx context.Context) Outcome { // spend counter marks, so the next pass launches nothing and answers // the existing OutcomeLimit. // - // TIME AND COST BOTH END WORK IN FLIGHT. Once either ceiling is - // reached, every worker is ended here, and + // TIME ENDS THE WORK IN FLIGHT, AS DOLLARS DO ([countLiveSpend]): the + // time is gone whatever a worker was in the middle of. So every worker + // is ended here, and // ITS ENDING IS ABSORBED, the way the caller's wall below reads its // endings: a return that was dropped would leave its task claimed and // reading as running on a run that is over, and what the worker spent @@ -245,7 +265,9 @@ func (s *Supervisor) Run(ctx context.Context) Outcome { cancel() } for s.inFlight > 0 { - s.consume(<-s.finished) + ret := <-s.finished + s.inFlight-- + s.absorb(ret) } case <-ctx.Done(): // The caller's wall: workers still out there were handed this @@ -255,7 +277,9 @@ func (s *Supervisor) Run(ctx context.Context) Outcome { // the endings rather than dropping them, because the run it stops // is one a later pass picks up from the store. for s.inFlight > 0 { - s.consume(<-s.finished) + ret := <-s.finished + s.inFlight-- + s.absorb(ret) } s.drain() return OutcomeIncomplete @@ -351,11 +375,16 @@ func (s *Supervisor) launch(ctx context.Context, task plandb.Task, wake string) taskCtx = WithWakeClause(taskCtx, wake) } s.cancels[task.ID] = cancel - s.banked[task.ID] = 0 + s.counted[task.ID] = 0 if s.limits.CostUSD > 0 { - taskCtx = WithSpendBank(taskCtx, func(usd float64) { - s.finished <- workerEvent{taskID: task.ID, banked: usd} - }) + // ONLY A RUN WITH A DOLLAR LIMIT LISTENS. A worker launched again for a + // wake starts its own count from nothing, so the figure a former run of + // the same task left behind is cleared before this one can write. + id := task.ID + s.liveMu.Lock() + s.live[id] = 0 + s.liveMu.Unlock() + taskCtx = WithSpendBank(taskCtx, func(usd float64) { s.bankLive(id, usd) }) } s.inFlight++ s.nodes++ @@ -370,8 +399,7 @@ func (s *Supervisor) launch(ctx context.Context, task plandb.Task, wake string) } else { report, err = worker.Run(WithStepsPerTask(taskCtx, s.limits.StepsPerTask), task) } - ret := workerReturn{task: task, report: report, err: err} - s.finished <- workerEvent{ret: &ret} + s.finished <- workerReturn{task: task, report: report, err: err} }() } @@ -400,23 +428,46 @@ func (s *Supervisor) drain() { s.workers.Wait() } -// consume serializes live spend and worker returns on the supervisor loop. -func (s *Supervisor) consume(event workerEvent) { - if event.ret != nil { - s.inFlight-- - s.absorb(*event.ret) - return +// bankLive is the one thing a worker's goroutine does to the run's account: it +// writes what the worker has banked so far and lets the loop know there is +// something to read. IT NEVER BLOCKS, for the reason the fields state. The +// figure only ever rises, so a late or repeated call cannot take a dollar back. +func (s *Supervisor) bankLive(id string, usd float64) { + s.liveMu.Lock() + if usd > s.live[id] { + s.live[id] = usd + } + s.liveMu.Unlock() + select { + case s.liveMoved <- struct{}{}: + default: } - // With no ceiling, live reports are observational: returned workers keep - // the shipped accounting semantics, including dropping cancelled work. +} + +// countLiveSpend adds to the run's spend whatever the workers in flight have +// banked since the loop last looked, and ends the work in flight the moment the +// sum reaches the person's dollar limit. +// +// A DOLLAR LIMIT MEANS THE RUN'S DOLLARS, NOT THE DOLLARS OF WORKERS THAT HAVE +// COME HOME. Read only at a return, one long worker could pass the limit many +// times over before anyone counted, and the limit would be a word about the +// run after it rather than a hold on this one. The ending is the time limit's: +// the limit state is marked, every worker's context is ended, the loop goes on +// reading their returns and absorbing each one, and the pass that finds nothing +// in flight answers the limit. Nothing is dropped, because drain is not the +// road: a dropped return would leave its task claimed on a run that is over. +func (s *Supervisor) countLiveSpend() { if s.limits.CostUSD <= 0 { return } - previous := s.banked[event.taskID] - if event.banked > previous { - s.spent += event.banked - previous - s.banked[event.taskID] = event.banked + s.liveMu.Lock() + for id, counted := range s.counted { + if now := s.live[id]; now > counted { + s.spent += now - counted + s.counted[id] = now + } } + s.liveMu.Unlock() if s.spent >= s.limits.CostUSD && !s.limitHit { s.limitHit = true for _, cancel := range s.cancels { @@ -425,6 +476,31 @@ func (s *Supervisor) consume(event workerEvent) { } } +// settleSpend reconciles a returned worker's report with what the loop counted +// while it worked, so its dollars are in the run's spend exactly once. The +// report is the receipt; a report smaller than what was already counted (a +// worker that lost its figure on the way out) takes nothing back, because the +// money was spent. A return can itself be what reaches the limit, as it always +// could. +func (s *Supervisor) settleSpend(ret workerReturn) { + if counted := s.counted[ret.task.ID]; ret.report.USD > counted { + s.spent += ret.report.USD - counted + } + s.forgetLive(ret.task.ID) + if s.limits.CostUSD > 0 && s.spent >= s.limits.CostUSD { + s.limitHit = true + } +} + +// forgetLive drops a returned worker's live figure and the loop's count of it, +// so the set the loop reads is the set of workers in flight. +func (s *Supervisor) forgetLive(id string) { + delete(s.counted, id) + s.liveMu.Lock() + delete(s.live, id) + s.liveMu.Unlock() +} + // absorb writes one worker's ending into the store and keeps the run's // counters true. A worker that came home well is done with its result; one // that came home with an error fails with that error as the reason. Either @@ -440,26 +516,23 @@ func (s *Supervisor) absorb(ret workerReturn) { // hands back no account this run counts. The completion check below still // runs, because the cancellation may be the write that finished the tree. delete(s.cancels, ret.task.ID) - if task := s.store.Task(ret.task.ID); task == nil || task.Status == plandb.StatusCancelled { - // Live spend was provisional. Preserve the existing rule that an - // externally cancelled task contributes no report to this run. - s.spent -= s.banked[ret.task.ID] - delete(s.banked, ret.task.ID) + if ret.report.Waiting { + // A PARKED RETURN IS NOT AN ENDING. The worker called `plandb wait`; the + // store already released the claim and flagged the task waiting, and the + // task stays open until the run wakes it ([launchWaits]). Nothing is + // written over the store's own park, and the spend the worker made is + // still the run's. + s.settleSpend(ret) + s.steps += ret.report.Steps s.completeTree() return } - banked := s.banked[ret.task.ID] - if ret.report.USD > banked { - s.spent += ret.report.USD - banked - } - delete(s.banked, ret.task.ID) - if s.limits.CostUSD > 0 && s.spent >= s.limits.CostUSD { - s.limitHit = true - } - if ret.report.Waiting { - // A parked return is not an ending. The store already released the - // claim, and the spend the worker made is still the run's. - s.steps += ret.report.Steps + if task := s.store.Task(ret.task.ID); task == nil || task.Status == plandb.StatusCancelled { + // WHAT WAS COUNTED ON THE WAY IS TAKEN BACK, so the rule above reads the + // same with a dollar limit as without one: a worker whose task the store + // cancelled under it hands back no account this run counts. + s.spent -= s.counted[ret.task.ID] + s.forgetLive(ret.task.ID) s.completeTree() return } @@ -468,6 +541,7 @@ func (s *Supervisor) absorb(ret workerReturn) { // counts what the worker spent, and the root's own report is still the // run's result. endedByStore := ret.err == nil && s.store.Task(ret.task.ID).Status == plandb.StatusDone + s.settleSpend(ret) s.steps += ret.report.Steps if ret.task.ID == s.store.RootID() { // The harness owns the root, and the store refuses worker writes to it. From e2d9d456c35d2748f9c43427d444507eb03fbd32 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Sat, 19 Sep 2026 15:28:15 -0400 Subject: [PATCH 07/11] docs: the #1268 change entry Co-Authored-By: Claude Fable 5.1 --- ...8-a-dollar-limit-holds-while-a-worker-works.md | 15 +++++++++++++++ 1 file changed, 15 insertions(+) create mode 100644 docs/changes/unreleased/1268-a-dollar-limit-holds-while-a-worker-works.md diff --git a/docs/changes/unreleased/1268-a-dollar-limit-holds-while-a-worker-works.md b/docs/changes/unreleased/1268-a-dollar-limit-holds-while-a-worker-works.md new file mode 100644 index 0000000000..9e346061e8 --- /dev/null +++ b/docs/changes/unreleased/1268-a-dollar-limit-holds-while-a-worker-works.md @@ -0,0 +1,15 @@ +--- +kind: fixed +title: a run's dollar limit now holds while a worker is still working +pr: 1268 +surface: [chat, engine] +invalidates: + - "A run's dollar limit was read only when a worker came home. A worker's spend reached the run's count in its report, so one long worker could pass the limit many times over before anything counted it, and the limit was a word about the run after it instead of a hold on this one." + - "A worker now says what it has spent as each model call is paid for, and the run adds it while the worker is still working. When the sum reaches the limit the run launches nothing new, ends the work in flight, absorbs every ending, and answers `a limit you set stopped it`. That is the time limit's road, so the sentence in the entry for the time limit that said a cost limit lets work in flight finish is no longer true." + - "A worker never waits on the run to report its spend. It writes a running figure and leaves a token the run may miss; the figure is the whole of what the worker has spent, so a missed token loses nothing, and a run that is draining is never held open by a worker that has something to say." + - "The call that reaches the limit is already paid for, so a run can end a little over its figure. A run with no dollar limit counts as it always did, at each return." +--- + +A task the store cancelled under its worker still hands back no spend the run counts, +with a limit or without one. What the run counted on the way for that worker is taken +back when it returns, so the rule reads the same as before. From 6fd6d206464be9507221215d7a93a95100e497a9 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Sat, 19 Sep 2026 15:31:43 -0400 Subject: [PATCH 08/11] run: the worker tells its run what it has spent as each call is paid for The spend figure was told to the run where the turn's usage arrives, which is the turn's ending. A worker's turn is its whole working life, so the run's live dollar limit never saw a figure before the worker came home. The test through the real worker showed one report for three paid calls. The agent's running total moves on every paid call; the worker reads it as each event comes by and tells the run whenever it has risen. Co-Authored-By: Claude Fable 5.1 --- internal/run/bashworker.go | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/internal/run/bashworker.go b/internal/run/bashworker.go index 0726a6898a..25049318f1 100644 --- a/internal/run/bashworker.go +++ b/internal/run/bashworker.go @@ -127,6 +127,13 @@ func (w *BashWorker) Run(ctx context.Context, task plandb.Task) (Report, error) // did call a tool resets the run to one — its own trailing words are the // first of the new run — and the fourth in a row fails the task. noAction := 0 + // banked is the last spend figure this worker told its run. THE RUN'S DOLLAR + // LIMIT IS READ WHILE THE WORKER WORKS, so the figure has to move when a + // call is paid for and not when a turn ends: a turn is the worker's whole + // working life, and its usage arrives with its ending. The agent's own + // running total is the one account that moves per paid call, so it is read + // as each event comes by and told to the run whenever it has risen. + banked := 0.0 for { events, err := agent.Submit(runCtx, brief) if err != nil { @@ -145,6 +152,10 @@ func (w *BashWorker) Run(ctx context.Context, task plandb.Task) (Report, error) ending storeEnding ) for event := range events { + if spent := agent.Usage().CostUSD; spent > banked { + banked = spent + bankSpend(runCtx, spent) + } switch event.Kind { case session.EventToolBegin: // A LIVE STEP IS TRUE ONLY WHILE ITS COMMAND RUNS. The begin event is @@ -214,7 +225,6 @@ func (w *BashWorker) Run(ctx context.Context, task plandb.Task) (Report, error) usd += event.Usage.CostUSD inTok += event.Usage.Input outTok += event.Usage.Output - bankSpend(runCtx, usd) case session.EventError: if turnErr == nil { turnErr = event.Err From b78243cb37233059f6b43619b5381f45603befa2 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Sat, 19 Sep 2026 15:35:16 -0400 Subject: [PATCH 09/11] test: the real worker's spend report is held to a fact, not a race The figures are readings of a running total, and a fast worker folds several calls into one reading, so their count is not the property. The seat now holds the second call until the first is reported: the first figure must be the first call's, told while the worker is still working. Co-Authored-By: Claude Fable 5.1 --- internal/run/bashworker_test.go | 45 +++++++++++++++++++++++---------- 1 file changed, 32 insertions(+), 13 deletions(-) diff --git a/internal/run/bashworker_test.go b/internal/run/bashworker_test.go index 8e748455fe..0a29b5937e 100644 --- a/internal/run/bashworker_test.go +++ b/internal/run/bashworker_test.go @@ -22,6 +22,7 @@ import ( "path/filepath" "strings" "sync" + "sync/atomic" "testing" "time" @@ -510,16 +511,33 @@ func TestBashWorkerEndsItsLoopAtTheStepCap(t *testing.T) { } } -// THE WORKER SAYS WHAT IT HAS SPENT AS EACH CALL IS PAID FOR, not only when it -// comes home. A run's dollar limit is read from these figures while the worker -// is still working, so they must arrive once per paid call, each one the whole -// of what the worker has spent so far, and the last one must be the figure the -// report carries: two accounts of one worker's money would drift. -func TestBashWorkerReportsItsSpendAsEachCallIsPaidFor(t *testing.T) { +// THE WORKER SAYS WHAT IT HAS SPENT WHILE IT IS STILL WORKING, not only when it +// comes home. A run's dollar limit is read from these figures, so the first +// paid call must be told to the run before the worker's turn is over, every +// figure must be the whole of what the worker has spent so far, and the last +// one must be the figure the report carries: two accounts of one worker's money +// would drift. +// +// THE SEAT HOLDS THE SECOND CALL UNTIL THE FIRST IS REPORTED, which is what makes +// "while it is still working" a fact and not a race: the figures are readings of +// a running total and a fast worker may fold several calls into one reading, so +// their count is not the property. A worker that reports only at its turn's +// end never frees the seat, and the first figure it then gives is not the first +// call's. +func TestBashWorkerReportsItsSpendWhileItIsStillWorking(t *testing.T) { t.Setenv("CODEAF_TASK_BELT", "bash") t.Setenv("CODEAF_PLANDB_BIN", stubCLI(t)) store := runOpenStore(t) - seat := &seat{ever: func(context.Context, []ai.Message) (*ai.Response, error) { + firstReported := make(chan struct{}) + var calls atomic.Int32 + seat := &seat{ever: func(ctx context.Context, _ []ai.Message) (*ai.Response, error) { + if calls.Add(1) > 1 { + select { + case <-firstReported: + case <-ctx.Done(): + case <-time.After(10 * time.Second): + } + } reply := toolReply(`{"command":"true"}`) cost := 0.25 reply.Usage.Cost = &cost @@ -531,24 +549,25 @@ func TestBashWorkerReportsItsSpendAsEachCallIsPaidFor(t *testing.T) { ctx := run.WithSpendBank(run.WithStepsPerTask(runContext(t), 3), func(usd float64) { mu.Lock() figures = append(figures, usd) + first := len(figures) == 1 mu.Unlock() + if first { + close(firstReported) + } }) report, _ := worker.Run(ctx, *store.Task(store.RootID())) mu.Lock() defer mu.Unlock() - if len(figures) < 3 { - t.Fatalf("the worker reported its spend %d times over three paid calls: %v", len(figures), figures) + if len(figures) == 0 || figures[0] != 0.25 { + t.Fatalf("spend figures = %v, want the first paid call's 0.25 told before the second call", figures) } for i := 1; i < len(figures); i++ { if figures[i] <= figures[i-1] { - t.Fatalf("spend figures are not the running whole: %v", figures) + t.Fatalf("spend figures are not a rising whole: %v", figures) } } - if figures[0] != 0.25 { - t.Fatalf("the first figure = %v, want the first call's 0.25", figures[0]) - } if last := figures[len(figures)-1]; last != report.USD { t.Fatalf("the last figure reported = %v and the report carries %v, want one account", last, report.USD) } From 0060b2bb349470a073ce6335cc795bf5bab6396e Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Sat, 19 Sep 2026 15:35:16 -0400 Subject: [PATCH 10/11] run: what a run counts as spent never goes down A worker whose task the store cancelled under it handed back no spend the run counted. Those dollars were paid, and a person's limit is about dollars, not about how the work came out. The cancelled worker's words and steps are still dropped; its dollars are settled like any other return's. The shipped test that pinned the old figure now expects the leaf's dollars and none of its steps. Co-Authored-By: Claude Fable 5.1 --- ...dollar-limit-holds-while-a-worker-works.md | 6 +- internal/run/cost_limit_test.go | 62 +++++++++++++++++++ internal/run/run.go | 15 +++-- internal/run/supervisor_test.go | 17 ++--- 4 files changed, 81 insertions(+), 19 deletions(-) diff --git a/docs/changes/unreleased/1268-a-dollar-limit-holds-while-a-worker-works.md b/docs/changes/unreleased/1268-a-dollar-limit-holds-while-a-worker-works.md index 9e346061e8..1adc5a6924 100644 --- a/docs/changes/unreleased/1268-a-dollar-limit-holds-while-a-worker-works.md +++ b/docs/changes/unreleased/1268-a-dollar-limit-holds-while-a-worker-works.md @@ -7,9 +7,9 @@ invalidates: - "A run's dollar limit was read only when a worker came home. A worker's spend reached the run's count in its report, so one long worker could pass the limit many times over before anything counted it, and the limit was a word about the run after it instead of a hold on this one." - "A worker now says what it has spent as each model call is paid for, and the run adds it while the worker is still working. When the sum reaches the limit the run launches nothing new, ends the work in flight, absorbs every ending, and answers `a limit you set stopped it`. That is the time limit's road, so the sentence in the entry for the time limit that said a cost limit lets work in flight finish is no longer true." - "A worker never waits on the run to report its spend. It writes a running figure and leaves a token the run may miss; the figure is the whole of what the worker has spent, so a missed token loses nothing, and a run that is draining is never held open by a worker that has something to say." + - "What a run counts as spent never goes down. A worker whose task the store cancelled under it handed back no spend the run counted, so those dollars were missing from the run's figure and from the limit. Its words and steps are still dropped; its dollars are now counted, whatever way the task ended." - "The call that reaches the limit is already paid for, so a run can end a little over its figure. A run with no dollar limit counts as it always did, at each return." --- -A task the store cancelled under its worker still hands back no spend the run counts, -with a limit or without one. What the run counted on the way for that worker is taken -back when it returns, so the rule reads the same as before. +`--max-cost` still does not reach a run: a run's dollar limit is the conversation's +spend ceiling. diff --git a/internal/run/cost_limit_test.go b/internal/run/cost_limit_test.go index cdb40f38d7..13bca49ae9 100644 --- a/internal/run/cost_limit_test.go +++ b/internal/run/cost_limit_test.go @@ -195,3 +195,65 @@ func TestAWorkerReportingSpendWhileTheRunDrainsDoesNotHoldTheRunOpen(t *testing. t.Fatalf("spend reports made = %d, want all 64", got) } } + +// WHAT A RUN COUNTS AS SPENT NEVER GOES DOWN, and it includes every paid call +// whatever way its task ended. A person's limit is about dollars, not about how +// the work came out: a task the store cancelled under its worker hands back no +// work the run counts, and its dollars were still paid. With a limit the run +// has already counted them on the way and must not take them back at the +// return; without one the return is where they are counted. +func TestSpendOfAWorkerWhoseTaskTheStoreCancelledStaysInTheRunsCount(t *testing.T) { + for _, test := range []struct { + name string + limit float64 + }{ + {name: "with a dollar limit", limit: 100}, + {name: "with no dollar limit", limit: 0}, + } { + t.Run(test.name, func(t *testing.T) { + store, err := plandb.Open(t.TempDir()+"/plan.json", "cancelled-spend", "root", "root", "cancel one task part way") + if err != nil { + t.Fatalf("open store: %v", err) + } + defer store.Close() + if _, err := store.AddMany([]plandb.TaskSpec{{ID: "peer", Title: "peer", ParentID: store.RootID()}}); err != nil { + t.Fatalf("add peer: %v", err) + } + peerPaid := make(chan struct{}) + var paidOnce atomic.Bool + factory := func(task plandb.Task) Worker { + if task.ID == "peer" { + return workerFunc(func(ctx context.Context, task plandb.Task) (Report, error) { + bankSpend(ctx, 0.30) + if paidOnce.CompareAndSwap(false, true) { + close(peerPaid) + } + <-ctx.Done() + return Report{Result: "late", Steps: 4, USD: 0.30}, ctx.Err() + }) + } + return workerFunc(func(ctx context.Context, task plandb.Task) (Report, error) { + select { + case <-peerPaid: + case <-ctx.Done(): + return Report{}, ctx.Err() + } + // Already cancelled on a wake of the root: the first cancellation stands. + _, _ = store.Cancel("peer", "stopped by hand") + return Report{Result: "the root's word"}, nil + }) + } + supervisor := NewSupervisor(store, t.TempDir(), 2, Limits{CostUSD: test.limit}, factory) + supervisor.Run(context.Background()) + if peer := store.Task("peer"); peer == nil || peer.Status != plandb.StatusCancelled { + t.Fatalf("peer = %#v, want cancelled by the store", peer) + } + if supervisor.spent != 0.30 { + t.Fatalf("run spend = %v, want the cancelled task's 0.30 counted exactly once", supervisor.spent) + } + if supervisor.steps != 0 { + t.Fatalf("run steps = %d, want none of the cancelled task's steps", supervisor.steps) + } + }) + } +} diff --git a/internal/run/run.go b/internal/run/run.go index 10934395f2..c765032a29 100644 --- a/internal/run/run.go +++ b/internal/run/run.go @@ -512,9 +512,12 @@ func (s *Supervisor) absorb(ret workerReturn) { // ending the store carries — a cancellation that landed while the worker // ran, which the same write cleared the claim and cascaded down — is the // one that stands, and no Done or Fail of this run may speak over it. The - // report is dropped whole, spend included: a worker stopped mid-flight - // hands back no account this run counts. The completion check below still - // runs, because the cancellation may be the write that finished the tree. + // report's words and its steps are dropped: a worker stopped mid-flight + // hands back no work this run counts. ITS DOLLARS ARE NOT DROPPED. What a + // run counts as spent never goes down and includes every paid call whatever + // way its task ended, because a person's limit is about dollars and not + // about how the work came out. The completion check below still runs, + // because the cancellation may be the write that finished the tree. delete(s.cancels, ret.task.ID) if ret.report.Waiting { // A PARKED RETURN IS NOT AN ENDING. The worker called `plandb wait`; the @@ -528,11 +531,7 @@ func (s *Supervisor) absorb(ret workerReturn) { return } if task := s.store.Task(ret.task.ID); task == nil || task.Status == plandb.StatusCancelled { - // WHAT WAS COUNTED ON THE WAY IS TAKEN BACK, so the rule above reads the - // same with a dollar limit as without one: a worker whose task the store - // cancelled under it hands back no account this run counts. - s.spent -= s.counted[ret.task.ID] - s.forgetLive(ret.task.ID) + s.settleSpend(ret) s.completeTree() return } diff --git a/internal/run/supervisor_test.go b/internal/run/supervisor_test.go index 43d3a0ccd0..895c8b9ec8 100644 --- a/internal/run/supervisor_test.go +++ b/internal/run/supervisor_test.go @@ -609,7 +609,7 @@ func TestStartRefusesADoorBuiltWithoutItsStoreOrItsFactory(t *testing.T) { } } -func TestStartCountsNothingFromAWorkerWhoseTaskWasCancelled(t *testing.T) { +func TestStartCountsOnlyTheDollarsOfAWorkerWhoseTaskWasCancelled(t *testing.T) { store := startOpenStore(t, "the run's own title") ctx := runContext(t) seat := newFakeSeat() @@ -617,8 +617,9 @@ func TestStartCountsNothingFromAWorkerWhoseTaskWasCancelled(t *testing.T) { // The leaf is cancelled mid-flight and comes home well afterwards, so its // report is a late one carrying a completion, steps and spend. The store // would refuse the write on its own law; what only the drop in absorb does - // is keep the run's counters out of it, and the counters are what Start - // answers with. + // is keep the work's counters out of it, and the counters are what Start + // answers with. THE DOLLARS ARE THE EXCEPTION: they were paid whatever way + // the task ended, so they are in the run's account. seat.actions["l1"] = func(ctx context.Context, task plandb.Task) (run.Report, error) { <-ctx.Done() return run.Report{Result: "late " + task.ID, Steps: 4, USD: 0.30}, nil @@ -637,14 +638,14 @@ func TestStartCountsNothingFromAWorkerWhoseTaskWasCancelled(t *testing.T) { if outcome != run.OutcomeIncomplete { t.Fatalf("outcome = %q, want %q", outcome, run.OutcomeIncomplete) } - // The root worker's own figures and the wake's are the whole account: three - // workers went out (the root, the cancelled leaf, and the root's wake), and - // the cancelled one's 0.30 and four steps counted for nothing. + // Three workers went out (the root, the cancelled leaf, and the root's + // wake). The cancelled one's four steps count for nothing, and its 0.30 is + // counted beside the root's 0.1 and the wake's 0.05. if summary.Nodes != 3 { t.Fatalf("summary nodes = %d, want the root, the cancelled leaf, and the root's wake", summary.Nodes) } - if !usdClose(summary.USD, 0.15) { - t.Fatalf("summary usd = %v, want the root's 0.1 and the wake's 0.05 alone", summary.USD) + if !usdClose(summary.USD, 0.45) { + t.Fatalf("summary usd = %v, want the root's 0.1, the wake's 0.05 and the cancelled leaf's 0.30", summary.USD) } if summary.Steps != 2 { t.Fatalf("summary steps = %d, want the root's 1 and the wake's 1 alone", summary.Steps) From ca32515e44f223362b1a8b711b6e55b1fc27f114 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Sat, 19 Sep 2026 15:37:55 -0400 Subject: [PATCH 11/11] run: the dollars of a worker the run outlived are still the run's The drain dropped the return of every worker it ended, spend included. The shipped cancelled-leaf test takes that road: the run ends before the leaf comes home, so counting a store-cancelled return in absorb alone left its dollars out. The drain now reads the returns its workers left behind, without waiting, and settles only their spend. It leaves the channel empty. Co-Authored-By: Claude Fable 5.1 --- ...dollar-limit-holds-while-a-worker-works.md | 2 +- internal/run/cost_limit_test.go | 8 ++++++++ internal/run/run.go | 19 ++++++++++++++++++- 3 files changed, 27 insertions(+), 2 deletions(-) diff --git a/docs/changes/unreleased/1268-a-dollar-limit-holds-while-a-worker-works.md b/docs/changes/unreleased/1268-a-dollar-limit-holds-while-a-worker-works.md index 1adc5a6924..3278d7d298 100644 --- a/docs/changes/unreleased/1268-a-dollar-limit-holds-while-a-worker-works.md +++ b/docs/changes/unreleased/1268-a-dollar-limit-holds-while-a-worker-works.md @@ -7,7 +7,7 @@ invalidates: - "A run's dollar limit was read only when a worker came home. A worker's spend reached the run's count in its report, so one long worker could pass the limit many times over before anything counted it, and the limit was a word about the run after it instead of a hold on this one." - "A worker now says what it has spent as each model call is paid for, and the run adds it while the worker is still working. When the sum reaches the limit the run launches nothing new, ends the work in flight, absorbs every ending, and answers `a limit you set stopped it`. That is the time limit's road, so the sentence in the entry for the time limit that said a cost limit lets work in flight finish is no longer true." - "A worker never waits on the run to report its spend. It writes a running figure and leaves a token the run may miss; the figure is the whole of what the worker has spent, so a missed token loses nothing, and a run that is draining is never held open by a worker that has something to say." - - "What a run counts as spent never goes down. A worker whose task the store cancelled under it handed back no spend the run counted, so those dollars were missing from the run's figure and from the limit. Its words and steps are still dropped; its dollars are now counted, whatever way the task ended." + - "What a run counts as spent never goes down. A worker whose task the store cancelled under it handed back no spend the run counted, so those dollars were missing from the run's figure and from the limit. Its words and steps are still dropped; its dollars are now counted, whatever way the task ended. The same holds for a worker the run outlived, ended when the tree finished under it: its return was dropped whole, and its dollars are now settled." - "The call that reaches the limit is already paid for, so a run can end a little over its figure. A run with no dollar limit counts as it always did, at each return." --- diff --git a/internal/run/cost_limit_test.go b/internal/run/cost_limit_test.go index 13bca49ae9..0f59ebf0ed 100644 --- a/internal/run/cost_limit_test.go +++ b/internal/run/cost_limit_test.go @@ -194,6 +194,14 @@ func TestAWorkerReportingSpendWhileTheRunDrainsDoesNotHoldTheRunOpen(t *testing. if got := reports.Load(); got != 64 { t.Fatalf("spend reports made = %d, want all 64", got) } + // AND THE WORKER THE RUN OUTLIVED WAS STILL PAID FOR. Its words are dropped + // with its return; its dollars are in the run's count, once. + if supervisor.spent != 0.064 { + t.Fatalf("run spend = %v, want the outlived worker's 0.064 counted exactly once", supervisor.spent) + } + if supervisor.inFlight != 0 || len(supervisor.finished) != 0 { + t.Fatalf("the run left work behind it: inFlight=%d, returns unread=%d", supervisor.inFlight, len(supervisor.finished)) + } } // WHAT A RUN COUNTS AS SPENT NEVER GOES DOWN, and it includes every paid call diff --git a/internal/run/run.go b/internal/run/run.go index c765032a29..fc58afe73c 100644 --- a/internal/run/run.go +++ b/internal/run/run.go @@ -417,15 +417,32 @@ func (s *Supervisor) launch(ctx context.Context, task plandb.Task, wake string) // a worker whose context a pass already ended — a cancelled task, the caller's // wall — is in the map no more and has only to be waited for. The endings are // dropped rather than absorbed: the run is over, the store already carries the -// ending that stands, and the return of a worker the run outlived has no +// ending that stands, and the words of a worker the run outlived have no // reader — the same reading absorb makes of a worker whose task the store // cancelled under it. +// +// THE DOLLARS ARE NOT DROPPED, for absorb's own reason: what a run counts as +// spent includes every paid call whatever way its task ended, and a worker the +// run outlived was paid for like any other. Every worker has returned by the +// time the wait is over and each left exactly one return in the channel, which +// is deep enough to hold them all, so they are read here without waiting and +// only their spend is settled. It also leaves the channel empty, so a return +// from this run can never be read as one of the next. func (s *Supervisor) drain() { for id, cancel := range s.cancels { cancel() delete(s.cancels, id) } s.workers.Wait() + for { + select { + case ret := <-s.finished: + s.inFlight-- + s.settleSpend(ret) + default: + return + } + } } // bankLive is the one thing a worker's goroutine does to the run's account: it