From 3b16255c038750fed05b05971b6df4ebd8064592 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Fri, 6 Nov 2026 09:13:24 -0500 Subject: [PATCH 1/4] Tasks: the older task road refuses duplicate work and holds a tree spend total With CODEAF_TASK_BELT=node the session-tree road bounded depth (3), children per node (20) and each node's own bill, but two nodes of one tree could hand out the same work, and the tree as a whole had no spend total. It now applies the default road's bounds from #2446: - A child whose title matches, after plandb's normalization (now exported as plandb.NormalizedTitle, one definition for both roads), any non-cancelled task in the same tree is refused with the same words, and a division naming the same work twice is refused whole. Pending proposals reserve their titles. A typed task starts a separate tree, so separate trees may reuse a title. - Every node, helper and check of one tree shares one spend tally held to the lower of crew.task_cap and the conversation's remaining limit (the same arithmetic as the default road). Reaching it stops the whole tree and the ending names the limit and how to raise it. A checkpoint keeps the tree's spent total. Two existing division tests used repeated titles as filler; their fixtures now name distinct work and keep every assertion. Co-Authored-By: Claude Opus 5.5 --- internal/manual/chat/tasks.md | 21 +- internal/plandb/treebound.go | 3 + internal/session/agent.go | 4 + internal/session/clientdoor.go | 3 + internal/session/plandb_plan.go | 2 +- internal/session/session.go | 2 + internal/session/task.go | 8 +- internal/session/task_audit.go | 1 + internal/session/task_bounds.go | 15 +- .../session/task_divide_family_check_test.go | 4 +- internal/session/task_divide_test.go | 14 +- internal/session/task_divide_wip.go | 24 +- internal/session/task_node_bounds_test.go | 438 ++++++++++++++++++ internal/session/task_node_spend.go | 230 +++++++++ internal/session/task_node_titles.go | 86 ++++ internal/session/task_quick.go | 2 +- internal/session/task_run.go | 40 +- internal/session/task_store.go | 13 +- internal/session/taskname.go | 2 +- 19 files changed, 877 insertions(+), 35 deletions(-) create mode 100644 internal/session/task_node_bounds_test.go create mode 100644 internal/session/task_node_spend.go create mode 100644 internal/session/task_node_titles.go diff --git a/internal/manual/chat/tasks.md b/internal/manual/chat/tasks.md index 59c0b97f87..7cc080b3f6 100644 --- a/internal/manual/chat/tasks.md +++ b/internal/manual/chat/tasks.md @@ -2322,15 +2322,16 @@ you set it — and under the day's limit above it. An adaptive run they start op $100.00 default. **An ordinary `/task`, or any task codeaf's own worker does, also has a money limit of its own: $5 unless you set another**, shown on the -Spending tab's `per task` row and set with `/crew cap task <$>`. On the run road this is +Spending tab's `per task` row and set with `/crew cap task <$>`. On both task roads this is one ceiling for the whole tree, including planning and checks. The smaller of that limit and the conversation's remaining allowance wins, even when no crew was routed. Continuing interrupted work keeps what the same tree already spent: only the unspent part of its per-task limit remains available. -A routed crew refuses a call that would cross its per-task limit. Reaching the run's +A routed crew refuses a call that would cross its per-task limit. Reaching the tree's ceiling ends workers in flight and starts no more. When the per-task limit binds, the report says `this task reached its $5 limit · raise it in /crew` (with your configured amount). Raise it with `/crew cap task <$>`, then rerun the remaining work with `/task`. +When the conversation's smaller limit binds, the report names that amount and `/settings`. Continuing a run is for interrupted work; a settled dollar-limit ending needs a rerun. **senior-dev, sec and review have their own dollar and time ceilings for each run**, separate from that `per task` row; see each program's page for its defaults and how to @@ -4307,6 +4308,22 @@ splits. `task.parallel` still applies to the whole session: pieces queue behind it exactly as top-level tasks do. +## Duplicate task titles on the node belt + +With `CODEAF_TASK_BELT=node`, `legacy` or `off`, a worker cannot add the same work +twice anywhere in its task tree. Letter case is ignored; runs of punctuation or +whitespace fold to one space: +`Read: the LAW!` and `read-the law` name the same work. Ancestors, siblings, cousins and +finished tasks all count; a cancelled task's title may be reused. A division naming +the same work twice is refused whole, before any part starts. Quick tasks use the same +check. Separate conversation tasks have separate trees; a person's typed task stays +exempt. + +The refusal says `no: this run already has a task named "Read: the LAW!" (7), so nothing +was added.` It tells the worker to do that work in its own hands, or name different +work for what is different. The shared tree money limit is described above under +*A task started from the composer carries a cap*. + ## How the bash belt run splits work instead of nesting node tasks With the switch unset, or `CODEAF_TASK_BELT=bash`, a task uses the run engine instead of the node tree. diff --git a/internal/plandb/treebound.go b/internal/plandb/treebound.go index 2dbc174c58..39f83c36c0 100644 --- a/internal/plandb/treebound.go +++ b/internal/plandb/treebound.go @@ -24,6 +24,9 @@ func normalizedTitle(title string) string { }), " ") } +// NormalizedTitle is the work identity shared by both task admission roads. +func NormalizedTitle(title string) string { return normalizedTitle(title) } + // GrowthRefusal answers why worker-created tasks cannot join this run, or // nothing when they fit. Internal review and person-approved writes bypass it. func (s *Store) GrowthRefusal(parentID string, titles []string) string { diff --git a/internal/session/agent.go b/internal/session/agent.go index 81d64c8f64..be4e9d7f0f 100644 --- a/internal/session/agent.go +++ b/internal/session/agent.go @@ -92,6 +92,10 @@ func New(config Config) (*Agent, error) { func (a *Agent) newChildAgent(config Config) (*Agent, error) { config.ClientTools = nil config.System = clientToolsWithoutSections(config.System) + // A helper built inside a task spends from that task's tree too. + if config.nodeTree == nil { + config.nodeTree = a.config.nodeTree + } if pair, ok := a.ConversationModelPair(); ok && config.ConversationModels == nil { pair.Main = config.Model if config.OneModel || pair.Fallback == pair.Main { diff --git a/internal/session/clientdoor.go b/internal/session/clientdoor.go index b6edc500d0..eb841ea58e 100644 --- a/internal/session/clientdoor.go +++ b/internal/session/clientdoor.go @@ -841,6 +841,9 @@ func (a *Agent) completeWithNamedModel(ctx context.Context, purpose callPurpose, ctx = provider.WithServedModel(ctx, a.servedModel) defer func() { a.endModelRequest(actual) }() } + if tree := a.nodeTreeForCall(ctx); tree != nil { + client = nodeTreeCompleter{tree: tree, model: called, next: client} + } response, err := client.CompleteWithMessages(toldItsWindow(ctx), materializeImageRefs(messages), append(options, ai.WithModel(wire))...) if response != nil && response.Model != "" { actual = response.Model diff --git a/internal/session/plandb_plan.go b/internal/session/plandb_plan.go index 11a7718430..9db73fa53b 100644 --- a/internal/session/plandb_plan.go +++ b/internal/session/plandb_plan.go @@ -513,7 +513,7 @@ func (g *TaskGraph) planPulse() { if parent == nil || parent.state == TaskQueued || parent.depth+1 > taskDepthLimit { continue } - if refused := g.claimChild(parent.nodeID); refused != "" { + if refused := g.claimChild(parent.nodeID, task.Title); refused != "" { continue } spec := taskSpec{ diff --git a/internal/session/session.go b/internal/session/session.go index 0f4f7aba69..a980b23a42 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -2543,6 +2543,8 @@ type Config struct { taskControl *TaskNode // The saved per-task cap applies across this node's worker and repair rounds. taskCapUSD float64 + // Workers, repairs and checks on the older road share their root's guard. + nodeTree *nodeTreeSpend // fixesDir is the project bucket the error→fix sidecar keeps its file in // (fixstore.go), and it is set only when this agent is a NODE. A node's diff --git a/internal/session/task.go b/internal/session/task.go index ffca6e31ad..45f3fe930c 100644 --- a/internal/session/task.go +++ b/internal/session/task.go @@ -827,7 +827,7 @@ func (a *Agent) stageTask(ctx context.Context, args json.RawMessage) bare.Staged // THE SLOT IS TAKEN BEFORE THE QUESTION and handed back by everything that // is not an admission, so a batch of proposals cannot walk through the fan // cap together ([TaskGraph.claimChild]). - if refusal := graph.claimChild(spec.parent); refusal != "" { + if refusal := graph.claimChild(spec.parent, spec.title); refusal != "" { return bare.Settled(refusal, true) } id := graph.reserve() @@ -843,7 +843,7 @@ func (a *Agent) stageTask(ctx context.Context, args json.RawMessage) bare.Staged if err != nil { // The session closed under the question. Nothing was asked and nothing // admitted, so the proposal simply never became one. - graph.releaseChild(spec.parent) + graph.releaseChild(spec.parent, spec.title) return bare.Settled(taskNeverAnswered, true) } return &stagedProposal{agent: a, graph: graph, id: id, spec: spec, stand: stand, elsewhere: elsewhere, wait: wait} @@ -871,7 +871,7 @@ type stagedProposal struct { // other ending of [stagedProposal.Commit], and exactly one of the two runs. func (p *stagedProposal) Withdraw() { p.wait.withdraw() - p.graph.releaseChild(p.spec.parent) + p.graph.releaseChild(p.spec.parent, p.spec.title) } // Commit reads what the wait came to and admits on it. It runs once the message @@ -885,7 +885,7 @@ func (p *stagedProposal) Commit(ctx context.Context) (string, bool, error) { admitted := false defer func() { if !admitted { - graph.releaseChild(spec.parent) + graph.releaseChild(spec.parent, spec.title) } }() answer, err := p.wait.answer() diff --git a/internal/session/task_audit.go b/internal/session/task_audit.go index ffd71de6cb..8a42705c67 100644 --- a/internal/session/task_audit.go +++ b/internal/session/task_audit.go @@ -3227,6 +3227,7 @@ func (a *Agent) newAuditAgent(dir string, node *TaskNode, door auditDoor, on str // usage row's task ([Config.checksNode]) — while taskID above stays 0, // because the auditor is not the node it reads. checksNode: node.id, + nodeTree: node.graph.spendTree(node), // The auditor is the node too, as far as anybody watching is concerned: // it runs on the node's clock, in the node's worktree, and a card whose // audit is parked on a provider's pacing is a card whose task is not diff --git a/internal/session/task_bounds.go b/internal/session/task_bounds.go index e271a37e85..fe94421182 100644 --- a/internal/session/task_bounds.go +++ b/internal/session/task_bounds.go @@ -33,15 +33,22 @@ func (a *Agent) taskRunCost(run *beltRun, cost float64) float64 { if run.store != nil { spent = storeSpent(run.store) } - left := runCostLeft(taskCap, spent) // Every dispatch passes here, including continuation. The ending names // the configured cap, while the engine receives only its unspent allowance. run.costCeiling = taskCap - run.conversationCostLimit = cost > 0 && cost <= left + allowance, conversation := taskTreeCost(taskCap, spent, cost) + run.conversationCostLimit = conversation + return allowance +} + +// taskTreeCost chooses the same remaining allowance for either task road. +func taskTreeCost(taskCap, spent, cost float64) (float64, bool) { + left := runCostLeft(taskCap, spent) + conversation := cost > 0 && cost <= left if cost <= 0 || left < cost { - return left + return left, conversation } - return cost + return cost, conversation } func (run *beltRun) taskLimitWords(summary RunSummary) string { diff --git a/internal/session/task_divide_family_check_test.go b/internal/session/task_divide_family_check_test.go index 14e5b5da51..60c7780751 100644 --- a/internal/session/task_divide_family_check_test.go +++ b/internal/session/task_divide_family_check_test.go @@ -706,7 +706,9 @@ func TestAThirdAskAfterARepairNeitherRefusesNorDoublesTheFamilysChecks(t *testin nest.divide(t, divideArgsFor(wideEvidence, familyParts("rank", "browse")...)) nest.divide(t, divideArgsFor(wideEvidence, familyParts("rank", "browse")...)) - third := nest.divide(t, divideArgsFor(wideEvidence, familyParts("rank", "browse")...)) + // The repaired ask already admitted the original work. Different work + // can reuse its family check without dispatching those tasks twice. + third := nest.divide(t, divideArgsFor(wideEvidence, familyParts("ranking", "browsing")...)) if !strings.HasPrefix(third, "split into 2 parts:") { t.Fatalf("the third ask was told %q, want the division admitted", third) } diff --git a/internal/session/task_divide_test.go b/internal/session/task_divide_test.go index 9f05566603..e8ac42b14c 100644 --- a/internal/session/task_divide_test.go +++ b/internal/session/task_divide_test.go @@ -1120,14 +1120,20 @@ func TestAStoppedPartsSpendStillReachesTheSessionLedger(t *testing.T) { func TestADivisionWiderThanTheCapIsRefusedWholeRatherThanHalfTaken(t *testing.T) { nest := newDivideNest(t, wideBrief, 0) - // Four parts already handed out leaves room for one, and a division of - // three is not a division of one. - for i := 0; i < taskFanLimit-1; i++ { - if answer := nest.divide(t, divideArgs(wideEvidence, 2)); i == 0 && !strings.Contains(answer, "split into") { + // Distinct parts leave room for two, so a division of three cannot fit. + for i := 0; i < (taskFanLimit-1)/2; i++ { + parts := []dividePart{ + {Title: fmt.Sprintf("Existing north %d", i), Summary: "s", Brief: "b", Acceptance: "a"}, + {Title: fmt.Sprintf("Existing south %d", i), Summary: "s", Brief: "b", Acceptance: "a"}, + } + if answer := nest.divide(t, divideArgsFor(wideEvidence, parts...)); i == 0 && !strings.Contains(answer, "split into") { t.Fatalf("the first division was refused: %q", answer) } } before := nest.graph.children(nest.parent.id) + if len(before) != taskFanLimit-2 { + t.Fatalf("the fixture admitted %d distinct parts, want %d", len(before), taskFanLimit-2) + } answer := nest.divide(t, divideArgs(wideEvidence, 3)) if !strings.Contains(answer, "as many as one task may") { t.Fatalf("the worker was told %q, want the cap said in the words it is said in everywhere else", answer) diff --git a/internal/session/task_divide_wip.go b/internal/session/task_divide_wip.go index 767a7a30b3..83610997e5 100644 --- a/internal/session/task_divide_wip.go +++ b/internal/session/task_divide_wip.go @@ -114,7 +114,11 @@ func (a *Agent) startTheParts(node *TaskNode, parts []dividePart, line *journalD // with one release rather than by a counter each ending has to remember. hands := divisionHands{graph: graph, parent: a.config.taskID} defer hands.release() - if refusal := hands.take(len(parts)); refusal != "" { + partTitles := make([]string, len(parts)) + for i, part := range parts { + partTitles[i] = part.Title + } + if refusal := hands.take(len(parts), partTitles...); refusal != "" { line.Decision = divisionRefusedCap return nil, nil, refusal } @@ -191,6 +195,7 @@ func (a *Agent) startTheParts(node *TaskNode, parts []dividePart, line *journalD // has already gone back ([TaskGraph.admit] releases it under the graph's // own lock) and this is the book agreeing with it. hands.held-- + hands.titles = hands.titles[1:] ids = append(ids, id) titles = append(titles, part.Title) } @@ -209,11 +214,19 @@ type divisionHands struct { graph *TaskGraph parent uint64 held int + titles []string } // take claims one hand per part, or answers the cap's own refusal having given // back whatever it managed to claim first. -func (h *divisionHands) take(parts int) string { +func (h *divisionHands) take(parts int, titles ...string) string { + if len(titles) > 0 { + if refusal := h.graph.claimChild(h.parent, titles...); refusal != "" { + return refusal + } + h.held, h.titles = parts, titles + return "" + } for i := 0; i < parts; i++ { if refusal := h.graph.claimChild(h.parent); refusal != "" { return refusal @@ -225,7 +238,12 @@ func (h *divisionHands) take(parts int) string { func (h *divisionHands) release() { for ; h.held > 0; h.held-- { - h.graph.releaseChild(h.parent) + if len(h.titles) > 0 { + h.graph.releaseChild(h.parent, h.titles[0]) + h.titles = h.titles[1:] + } else { + h.graph.releaseChild(h.parent) + } } } diff --git a/internal/session/task_node_bounds_test.go b/internal/session/task_node_bounds_test.go new file mode 100644 index 0000000000..df9d5c7784 --- /dev/null +++ b/internal/session/task_node_bounds_test.go @@ -0,0 +1,438 @@ +package session + +import ( + "context" + "fmt" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/Agent-Field/agentfield/sdk/go/ai" + "github.com/Agent-Field/codeaf/internal/config" + "github.com/Agent-Field/codeaf/internal/plandb" +) + +// C1, C6: proposals compare every relative, including finished work, using the shared normalization. +func TestNodeTreeRefusesDuplicateWorkAnywhere(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "node") + for _, relative := range []string{"ancestor", "sibling", "cousin", "finished"} { + t.Run(relative, func(t *testing.T) { + n := newNest(t, nil, nil) + n.handOut(t, "Read: the LAW!") + first := n.graph.children(n.parent.id)[0] + worker := n.node + if relative == "ancestor" { + worker = n.workerFor(t, first) + first = n.parent + } + if relative == "cousin" { + n.handOut(t, "Inspect the parser") + worker = n.workerFor(t, n.graph.children(n.parent.id)[1]) + firstWorker := n.workerFor(t, first) + n.handOutFrom(t, firstWorker, "Read the rules") + first = n.graph.children(first.id)[0] + } + if relative == "finished" { + n.graph.complete(first, TaskDone) + } + before := len(n.graph.nodes) + title := "read---THE law" + if relative == "ancestor" { + title = "THE---whole JOB" + } + if relative == "cousin" { + title = "read---THE rules" + } + answer, refused, err := worker.proposeTask(t.Context(), pieceArgs(title)) + for _, want := range []string{"no: this run already has a task named", fmt.Sprintf("(%d)", first.id), "nothing was added", "do it in your own hands", "name it for what is different"} { + if err != nil || !refused || !strings.Contains(answer, want) { + t.Fatalf("duplicate refusal missing %q: %q, %v, %v", want, answer, refused, err) + } + } + if len(n.graph.nodes) != before { + t.Fatal("duplicate work was admitted") + } + if answer := n.handOutFrom(t, worker, "Different work"); !strings.Contains(answer, "started") { + t.Fatalf("different work refused: %s", answer) + } + }) + } +} + +// C1, C3: cancelled titles are reusable, and two person-authored roots have independent title scopes. +func TestNodeTreeAllowsCancelledTitlesAndSeparateRoots(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "node") + n := newNest(t, nil, nil) + n.handOut(t, "Read the law") + kid := n.graph.children(n.parent.id)[0] + if _, err := n.graph.stop(kid.id); err != nil { + t.Fatal(err) + } + if answer := n.handOut(t, "read-the LAW"); !strings.Contains(answer, "started") { + t.Fatalf("cancelled title refused: %s", answer) + } + id, _, _, err := n.session.StartTask(t.Context(), n.parent.title(), false) + if err != nil { + t.Fatalf("a person's typed task was refused: %v", err) + } + other := n.workerFor(t, n.graph.node(id)) + if answer := n.handOutFrom(t, other, "Read the law"); !strings.Contains(answer, "started") { + t.Fatalf("a different tree inherited the first tree's titles: %s", answer) + } +} + +// C1, C2: division batches are refused before any part exists; quick tasks use the same door. +func TestNodeTreeRefusesAnEntireDuplicateDivisionAndQuickTask(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "node") + n := newNest(t, nil, nil) + line := &journalDivision{} + ids, _, refusal := n.node.startTheParts(n.parent, []dividePart{{Title: "Read the law"}, {Title: "read---the LAW"}}, line) + if len(ids) != 0 || len(n.graph.children(n.parent.id)) != 0 || !strings.Contains(refusal, "this batch names the same work") || !strings.Contains(refusal, "nothing was added") { + t.Fatalf("division partially admitted or refused for the wrong reason: %v, %s", ids, refusal) + } + n.handOut(t, "Read the law") + ids, _, refusal = n.node.startTheParts(n.parent, []dividePart{{Title: "Different part"}, {Title: "read the LAW"}}, &journalDivision{}) + if len(ids) != 0 || len(n.graph.children(n.parent.id)) != 1 || !strings.Contains(refusal, "already has a task named") { + t.Fatalf("division containing existing work partly started: %v, %s", ids, refusal) + } + id, _, quick := n.node.admitQuick(quickAsk{line: "read---the LAW"}) + if id != 0 || !strings.Contains(quick.said, "already has a task named") || len(n.graph.children(n.parent.id)) != 1 { + t.Fatalf("quick task bypassed duplicate refusal: %d, %+v", id, quick) + } +} + +// C1, C2: staged sibling proposals reserve their titles and withdrawal releases exactly its reservation. +func TestNodeTreeReservesPendingProposalTitles(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "node") + n := newNest(t, nil, nil) + first, ok := n.node.stageTask(t.Context(), pieceArgs("Read the law")).(*stagedProposal) + if !ok { + t.Fatal("first proposal did not stage") + } + other, ok := n.node.stageTask(t.Context(), pieceArgs("Inspect the parser")).(*stagedProposal) + if !ok { + t.Fatal("different proposal did not stage") + } + first.Withdraw() + if answer := n.handOut(t, "READ the law"); !strings.Contains(answer, "started") { + t.Fatalf("withdrawn title stayed reserved: %s", answer) + } + answer, refused, err := n.node.proposeTask(t.Context(), pieceArgs("inspect---the PARSER")) + if err != nil || !refused || !strings.Contains(answer, "this batch names the same work") { + t.Fatalf("another pending title lost its reservation: %q, %v, %v", answer, refused, err) + } + other.Withdraw() +} + +// C4: several nodes exhaust one ceiling, cut an in-flight call, fail queued work and prevent further admission. +func TestNodeTreeSpendStopsEveryNodeAndNamesTheBindingLimit(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "node") + for _, conversation := range []bool{false, true} { + t.Run(fmt.Sprint(conversation), func(t *testing.T) { + entered := make(chan struct{}) + completer := &scriptedCompleter{steps: []step{ + func(context.Context, []ai.Message) (*ai.Response, error) { return pricedResponse("first", .4), nil }, + func(ctx context.Context, _ []ai.Message) (*ai.Response, error) { + close(entered) + <-ctx.Done() + return nil, ctx.Err() + }, + func(context.Context, []ai.Message) (*ai.Response, error) { + cost := 1.1 + if conversation { + cost = .6 + } + return pricedResponse("last", cost), nil + }, + }} + a, _ := newTestAgent(t, completer, func(c *Config) { + c.ProfileDir = t.TempDir() + c.Interactive = true + if conversation { + c.Budget = Budget{USD: 1.25} + } + }) + if err := config.SetCrewTaskCap(a.config.ProfileDir, "1.5"); err != nil { + t.Fatal(err) + } + if conversation { + a.usage.CostUSD = .25 + } + g := a.graph() + g.report = nil + g.run = func(*TaskNode) {} + g.limit = 3 + rootID := g.reserve() + g.admit(rootID, taskSpec{title: "Whole assignment", named: true, depth: 1, model: "test/model"}) + root := g.node(rootID) + workers := make([]*Agent, 0, 3) + nodes := []*TaskNode{root} + for i := 0; i < 3; i++ { + title := fmt.Sprintf("Part %d", i) + if refusal := g.claimChild(rootID, title); refusal != "" { + t.Fatal(refusal) + } + id := g.reserve() + g.admit(id, taskSpec{title: title, named: true, parent: rootID, depth: 2, model: "test/model"}) + nodes = append(nodes, g.node(id)) + } + for _, node := range nodes[:3] { + worker, err := a.newTaskAgent(t.Context(), t.TempDir(), node, "") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = worker.Close() }) + workers = append(workers, worker) + } + if _, err := workers[0].completeWithModel(t.Context(), purposeTurn, nil, "test/model"); err != nil { + t.Fatal(err) + } + // Keep this node owned until its provider observes cancellation. + if !nodes[1].claimRun() { + t.Fatal("in-flight node was not running") + } + ended := make(chan error, 1) + go func() { + _, err := workers[1].completeWithModel(t.Context(), purposeTurn, nil, "test/model") + ended <- err + }() + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("scripted call never entered") + } + if _, err := workers[2].completeWithModel(t.Context(), purposeTask, nil, "test/model"); err != nil { + t.Fatal(err) + } + select { + case err := <-ended: + if err == nil { + t.Fatal("in-flight call was not cancelled") + } + case <-time.After(5 * time.Second): + t.Fatal("tree ceiling left a call in flight") + } + g.complete(nodes[1], TaskFailed) + want := config.CrewTaskCapAction(1.5) + if conversation { + want = "conversation" + } + for _, node := range nodes { + row := node.notice() + if row.State != TaskFailed || row.Ending != TaskEndingCostLimit || !strings.Contains(row.Report, want) || conversation && strings.Contains(row.Report, "raise it in /crew") { + t.Fatalf("tree limit ending is wrong: %+v", row) + } + if conversation && (!strings.Contains(row.Report, "$1.25") || !strings.Contains(row.Report, "/settings")) { + t.Fatalf("conversation limit lost its amount or raise action: %+v", row) + } + } + if refusal := g.claimChild(rootID, "Another part"); !strings.Contains(refusal, want) { + t.Fatalf("spent tree admitted another node: %s", refusal) + } + completer.mu.Lock() + calls := len(completer.seen) + completer.mu.Unlock() + if calls != 3 || !nodes[3].started.IsZero() { + t.Fatalf("queued work reached the provider: calls=%d, queued started=%v", calls, nodes[3].started) + } + }) + } +} + +// C5: one node below the ceiling runs through the real quick-task runner and finishes normally. +func TestNodeTreeUnderTheCeilingFinishesNormally(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "node") + a, _ := newTestAgent(t, &scriptedCompleter{steps: []step{func(context.Context, []ai.Message) (*ai.Response, error) { + return pricedResponse("finished", .25), nil + }}}, func(c *Config) { c.ProfileDir = t.TempDir() }) + g := a.graph() + g.report = nil + id, _, refusal := a.admitQuick(quickAsk{line: "Read the law"}) + if refusal.said != "" { + t.Fatal(refusal.said) + } + node := g.node(id) + waitDoneNode(t, node) + if row := node.notice(); row.State != TaskDone || row.Report != "finished" || row.CostUSD != .25 || row.Ending == TaskEndingCostLimit { + t.Fatalf("under-ceiling node changed: %+v", row) + } +} + +// C4: a restored root keeps exactly the shared spend, including child work already folded into its bill. +func TestNodeTreeCheckpointKeepsTheUnspentAllowance(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "node") + profile := t.TempDir() + if err := config.SetCrewTaskCap(profile, "1"); err != nil { + t.Fatal(err) + } + a, _ := newTestAgent(t, &scriptedCompleter{steps: []step{func(context.Context, []ai.Message) (*ai.Response, error) { + return pricedResponse("partly done", .75), nil + }}}, func(c *Config) { c.ProfileDir = profile }) + g := a.graph() + g.run, g.report = func(*TaskNode) {}, nil + checkpoint := filepath.Join(t.TempDir(), "tasks.json") + g.store = newTaskStore(checkpoint) + id := g.reserve() + g.admit(id, taskSpec{title: "Read the law", named: true, brief: "Read the law", acceptance: "The law is read", depth: 1, model: "test/model"}) + worker, err := a.newTaskAgent(t.Context(), t.TempDir(), g.node(id), "") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = worker.Close() }) + if _, err := worker.completeWithModel(t.Context(), purposeTurn, nil, "test/model"); err != nil { + t.Fatal(err) + } + document := readCheckpoint(t, checkpoint) + if saved := recordOf(t, document, id).TreeSpend; saved == nil || saved.Spent != .75 { + t.Fatalf("checkpoint lost the tree's live spend: %+v", saved) + } + // A parent's historic bill can already contain its finished child's bill. + document.Nodes[0].State, document.Nodes[0].CostUSD = TaskQueued, .75 + document.Nodes = append(document.Nodes, taskRecord{ID: id + 1, Parent: id, Depth: 2, Title: "Finished part", Brief: "Read the rules", Acceptance: "The rules are read", State: TaskDone, CostUSD: .25}) + document.Seq = id + 1 + b, _ := newTestAgent(t, &scriptedCompleter{steps: []step{func(context.Context, []ai.Message) (*ai.Response, error) { + return pricedResponse("the remaining work", .25), nil + }}}, func(c *Config) { c.ProfileDir = profile }) + restored := b.graph() + restored.run, restored.report = func(*TaskNode) {}, nil + restored.rehydrate(document, b.config.Workspace, TaskSettleAsk) + restored.runFrontier() + resumed, err := b.newTaskAgent(t.Context(), t.TempDir(), restored.node(id), "") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = resumed.Close() }) + if _, err := resumed.completeWithModel(t.Context(), purposeTurn, nil, "test/model"); err != nil { + t.Fatalf("the actual remaining allowance was lost: %v", err) + } + if row := restored.node(id).notice(); row.Ending != TaskEndingCostLimit || !strings.Contains(row.Report, config.CrewTaskCapAction(1)) { + t.Fatalf("resumed work received a fresh tree allowance: %+v", row) + } +} + +// C6: both admission roads make the same decision for Unicode, digits and punctuation. +func TestNodeTreeUsesThePlanStoresWorkIdentity(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "node") + for _, title := range []string{"READ---café 42", "read_café...42", "Read / CAFÉ / 42"} { + t.Run(title, func(t *testing.T) { + store, err := plandb.Open(filepath.Join(t.TempDir(), "plan.db"), "work", "root", "Read: Café 42!", "the work") + if err != nil { + t.Fatal(err) + } + defer store.Close() + if _, err := store.AddManyBounded([]plandb.TaskSpec{{Title: title}}); err == nil || !strings.Contains(err.Error(), "already has a task named") { + t.Fatalf("plan admission changed its work identity: %v", err) + } + n := newNest(t, nil, nil) + n.handOut(t, "Read: Café 42!") + if answer := n.handOut(t, title); !strings.Contains(answer, "already has a task named") { + t.Fatalf("node admission disagrees with plan admission: %s", answer) + } + }) + } +} + +// C3, C4: workers, repairs, checks and owner helpers share one tree's money; another tree remains live. +func TestNodeTreeCountsEveryCallAndKeepsOtherTreesIndependent(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "node") + completer := &scriptedCompleter{} + for i := 0; i < 7; i++ { + completer.steps = append(completer.steps, func(context.Context, []ai.Message) (*ai.Response, error) { + return pricedResponse("a paid answer", .125), nil + }) + } + a, _ := newTestAgent(t, completer, func(c *Config) { c.ProfileDir = t.TempDir() }) + if err := config.SetCrewTaskCap(a.config.ProfileDir, "0.75"); err != nil { + t.Fatal(err) + } + g := a.graph() + g.run, g.report = func(*TaskNode) {}, nil + rootID := g.reserve() + g.admit(rootID, taskSpec{title: "Read the law", named: true, depth: 1, model: "test/model"}) + otherID := g.reserve() + g.admit(otherID, taskSpec{title: "Read the law", named: true, depth: 1, model: "test/model"}) + childID := g.reserve() + g.admit(childID, taskSpec{title: "Read the rules", named: true, parent: rootID, depth: 2, model: "test/model"}) + root := g.node(rootID) + builds := []func() (*Agent, error){ + func() (*Agent, error) { return a.newTaskAgent(t.Context(), t.TempDir(), root, "") }, + func() (*Agent, error) { + return a.newTaskAgentOn(t.Context(), t.TempDir(), root, "-repair", "test/model", true) + }, + func() (*Agent, error) { + return a.newAuditAgent(t.TempDir(), root, plainDoor(auditReadCommands), "test/model") + }, + func() (*Agent, error) { return a.newTaskAgent(t.Context(), t.TempDir(), g.node(childID), "") }, + } + var rootWorker *Agent + for _, build := range builds { + worker, err := build() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = worker.Close() }) + if rootWorker == nil { + rootWorker = worker + } + if _, err := worker.completeWithModel(t.Context(), purposeTurn, nil, "test/model"); err != nil { + t.Fatal(err) + } + } + helper, err := rootWorker.newChildAgent(Config{Workspace: t.TempDir(), Model: "test/model", System: "SYSTEM", InTask: true}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = helper.Close() }) + if _, err := helper.completeWithModel(t.Context(), purposeTurn, nil, "test/model"); err != nil { + t.Fatal(err) + } + if _, err := a.completeWithModel(withNodeTree(t.Context(), root), purposeTask, nil, "test/model"); err != nil { + t.Fatal(err) + } + for _, id := range []uint64{rootID, childID} { + if row := g.node(id).notice(); row.State != TaskFailed || row.Ending != TaskEndingCostLimit { + t.Fatalf("a paid role escaped the tree tally: %+v", row) + } + } + other := g.node(otherID) + if other.stateNow() != TaskRunning { + t.Fatal("one tree's limit stopped a separate assignment") + } + worker, err := a.newTaskAgent(t.Context(), t.TempDir(), other, "") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = worker.Close() }) + if _, err := worker.completeWithModel(t.Context(), purposeTurn, nil, "test/model"); err != nil { + t.Fatalf("another tree lost its own allowance: %v", err) + } + other.finish("finished", nil, "", "") + g.complete(other, TaskDone) + if other.stateNow() != TaskDone { + t.Fatal("another tree could not finish") + } + if got := completer.requests(); got != 7 { + t.Fatalf("paid calls = %d, want 7", got) + } +} + +// C4: an already spent conversation allowance starts no node, even with an unpriced completer. +func TestNodeTreeWithNoConversationAllowanceNeverStarts(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "node") + completer := &scriptedCompleter{} + a, _ := newTestAgent(t, completer, func(c *Config) { + c.ProfileDir = t.TempDir() + c.Interactive = true + c.Budget = Budget{USD: 1} + }) + a.usage.CostUSD = 1 + g := a.graph() + g.run, g.report = func(*TaskNode) { t.Error("a node without any allowance started") }, nil + id := g.reserve() + g.admit(id, taskSpec{title: "Read the law", named: true, brief: "Read the law", acceptance: "The law is read", depth: 1}) + row := g.node(id).notice() + if row.State != TaskFailed || row.Ending != TaskEndingCostLimit || !row.StartedAt.IsZero() || !strings.Contains(row.Report, "conversation") || !strings.Contains(row.Report, "$1") || completer.requests() != 0 { + t.Fatalf("an exhausted conversation allowance reached a worker: %+v, calls=%d", row, completer.requests()) + } +} diff --git a/internal/session/task_node_spend.go b/internal/session/task_node_spend.go new file mode 100644 index 0000000000..ba4a2d01d4 --- /dev/null +++ b/internal/session/task_node_spend.go @@ -0,0 +1,230 @@ +package session + +import ( + "context" + "errors" + "fmt" + "math" + "strings" + + "github.com/Agent-Field/agentfield/sdk/go/ai" + "github.com/Agent-Field/codeaf/internal/config" + "github.com/Agent-Field/codeaf/internal/provider" +) + +// nodeTreeSpend is one older-road assignment's money and cancellation scope. +// The graph lock protects stopped; SpendGuard protects the shared tally. +type nodeTreeSpend struct { + graph *TaskGraph + root uint64 + guard *SpendGuard + ctx context.Context + halt context.CancelFunc + stopped bool +} + +type nodeTreeSpendRecord struct { + Spent float64 `json:"spent,omitempty"` +} + +type nodeTreeKey struct{} + +// Supervising helpers run on the owner, so their context carries the tree +// without giving the conversation the worker's persistent guard. +func withNodeTree(ctx context.Context, node *TaskNode) context.Context { + if tree := node.graph.spendTree(node); tree != nil { + return context.WithValue(ctx, nodeTreeKey{}, tree) + } + return ctx +} + +func (a *Agent) nodeTreeForCall(ctx context.Context) *nodeTreeSpend { + if tree, ok := ctx.Value(nodeTreeKey{}).(*nodeTreeSpend); ok { + return tree + } + return a.config.nodeTree +} + +func (g *TaskGraph) nodeSpendRecordLocked(node *TaskNode) *nodeTreeSpendRecord { + if tree := g.nodeTrees[node.id]; tree != nil { + return &nodeTreeSpendRecord{Spent: tree.guard.tally().Total()} + } + return node.savedTreeSpend +} + +func (g *TaskGraph) prepareNodeTrees() { + if bashBeltAsked() { + return + } + g.mu.Lock() + var roots []*TaskNode + for _, id := range g.order { + if node := g.nodes[id]; node != nil && node.parent == 0 && g.nodeTrees[id] == nil { + roots = append(roots, node) + } + } + g.mu.Unlock() + for _, node := range roots { + g.spendTree(node) + } +} + +func (g *TaskGraph) savedNodeTreeSpendLocked(root uint64) float64 { + if node := g.nodes[root]; node != nil && node.savedTreeSpend != nil { + return node.savedTreeSpend.Spent + } + // Older checkpoints have only node bills, which may include folded + // descendants. Counting those conservatively cannot grant fresh money. + spent := 0.0 + for _, node := range g.nodes { + if node != nil && g.nodeRootLocked(node.id) == root { + spent += node.cost + } + } + return spent +} + +func (g *TaskGraph) spendTree(node *TaskNode) *nodeTreeSpend { + if bashBeltAsked() || node == nil { + return nil + } + g.mu.Lock() + root, home := g.nodeRootLocked(node.id), g.home + tree := g.nodeTrees[root] + if tree != nil || home == nil { + g.mu.Unlock() + return tree + } + spent := g.savedNodeTreeSpendLocked(root) + g.mu.Unlock() + // Read settings and the conversation's books outside the graph lock. + taskCap, action := config.CrewTaskSpendCap(home.config.ProfileDir) + conversationCap := home.railCap(0) + allowance, conversation := taskTreeCost(taskCap, spent, runCostLeft(conversationCap, home.Usage().CostUSD)) + // The run engine's positive sentinel means no allowance, including for + // an unpriced model whose first call cannot be estimated by SpendGuard. + if allowance == math.SmallestNonzeroFloat64 { + allowance = 0 + } + if conversation { + action = fmt.Sprintf("the conversation reached its %s limit · raise the per conversation limit in /settings", config.CrewTaskMoney(conversationCap)) + } + ctx, halt := context.WithCancel(context.Background()) + candidate := &nodeTreeSpend{graph: g, root: root, ctx: ctx, halt: halt, + guard: &SpendGuard{Price: config.CrewCallPriceAt(home.config.ProfileDir), TaskCap: spent + allowance, TaskAction: action, Task: &SpendTask{spent: spent}}} + g.mu.Lock() + if g.nodeTrees == nil { + g.nodeTrees = make(map[uint64]*nodeTreeSpend) + } + if tree = g.nodeTrees[root]; tree == nil { + tree = candidate + g.nodeTrees[root] = tree + } + g.mu.Unlock() + if tree != candidate { + halt() + } + if tree.guard.tally().Total() >= tree.guard.TaskCap { + tree.stop() + } + return tree +} + +// nodeTreeStopLocked keeps the frontier and the landing on the same ending. +func (g *TaskGraph) nodeTreeStopLocked(node *TaskNode) string { + tree := g.nodeTrees[g.nodeRootLocked(node.id)] + if tree == nil || !tree.stopped { + return "" + } + if node.ending == "" { + node.ending = TaskEndingCostLimit + } + return tree.guard.TaskAction +} + +func (g *TaskGraph) nodeReadinessLocked(node *TaskNode) (bool, string) { + if limit := g.nodeTreeStopLocked(node); limit != "" { + return false, limit + } + return g.readinessLocked(node) +} + +func (g *TaskGraph) nodeTreeLandingLocked(node *TaskNode, state TaskState) TaskState { + if limit := g.nodeTreeStopLocked(node); limit != "" { + if !strings.Contains(node.report, limit) { + node.landLocked(withReport(node.landed, limit)) + } + return TaskFailed + } + return state +} + +func (tree *nodeTreeSpend) stop() { + g := tree.graph + g.mu.Lock() + if tree.stopped { + g.mu.Unlock() + return + } + tree.stopped = true + var ids []uint64 + for _, id := range g.order { + node := g.nodes[id] + if node == nil || node.state.settled() || g.nodeRootLocked(id) != tree.root { + continue + } + if node.ending == "" { + node.ending = TaskEndingCostLimit + } + ids = append(ids, id) + } + g.mu.Unlock() + // Mark the whole scope before any stop can turn the frontier again. + for _, id := range ids { + _, _ = g.stopFor(id, tree.guard.TaskAction) + } + tree.halt() +} + +func (tree *nodeTreeSpend) book(ctx context.Context, model string, response *ai.Response, held float64) { + tree.guard.after(ctx, model, response, held) + tree.graph.checkpoint() + if tree.guard.TaskCap > 0 && tree.guard.tally().Total() >= tree.guard.TaskCap { + tree.stop() + } +} + +// nodeTreeCompleter covers turns and helpers alike, including discarded paid +// attempts. The existing crew and seat guards keep their independent bounds. +type nodeTreeCompleter struct { + tree *nodeTreeSpend + model string + next Completer +} + +func (c nodeTreeCompleter) CompleteWithMessages(ctx context.Context, messages []ai.Message, options ...ai.Option) (*ai.Response, error) { + c.tree.graph.mu.Lock() + stopped := c.tree.stopped + c.tree.graph.mu.Unlock() + if stopped { + return nil, ErrSpendStopped{Action: c.tree.guard.TaskAction, TaskLimit: true} + } + held, err := c.tree.guard.before(ctx, c.model, messages, options) + if err != nil { + var stopped ErrSpendStopped + if errors.As(err, &stopped) && stopped.TaskLimit { + c.tree.stop() + } + return nil, err + } + callCtx, release := context.WithCancel(ctx) + defer release() + unhook := context.AfterFunc(c.tree.ctx, release) + defer unhook() + callCtx = provider.WithDiscardedUsage(callCtx, func(_ string, _ string, response *ai.Response) { + c.tree.book(ctx, c.model, response, 0) + }) + response, err := c.next.CompleteWithMessages(callCtx, messages, options...) + c.tree.book(ctx, c.model, response, held) + return response, err +} diff --git a/internal/session/task_node_titles.go b/internal/session/task_node_titles.go new file mode 100644 index 0000000000..beaa7d1198 --- /dev/null +++ b/internal/session/task_node_titles.go @@ -0,0 +1,86 @@ +package session + +import ( + "fmt" + + "github.com/Agent-Field/codeaf/internal/plandb" +) + +// nodeRootLocked follows ownership, not dependency edges: separate assignments +// in one conversation may name the same work. +func (g *TaskGraph) nodeRootLocked(id uint64) uint64 { + for steps := 0; steps <= len(g.nodes); steps++ { + node := g.nodes[id] + if node == nil || node.parent == 0 { + return id + } + id = node.parent + } + return id +} + +func (g *TaskGraph) nodeGrowthRefusalLocked(parent uint64, titles []string) string { + if bashBeltAsked() { + return "" + } + root := g.nodeRootLocked(parent) + if tree := g.nodeTrees[root]; tree != nil && tree.stopped { + return tree.guard.TaskAction + } + named := g.nodeTreeTitlesLocked(root) + batch := make(map[string]bool, len(titles)) + for _, title := range titles { + key := plandb.NormalizedTitle(title) + if node := named[key]; node != nil { + return fmt.Sprintf("no: this run already has a task named %q (%d), so nothing was added. If that work is yours, do it in your own hands; if it is different work, name it for what is different.", node.spec.title, node.id) + } + if batch[key] || g.pendingNodeTitleLocked(root, key) { + return fmt.Sprintf("no: this batch names the same work %q more than once, so nothing was added. Do that work in your own hands, or name each different part for what is different.", title) + } + batch[key] = true + } + return "" +} + +func (g *TaskGraph) nodeTreeTitlesLocked(root uint64) map[string]*TaskNode { + named := make(map[string]*TaskNode) + for _, id := range g.order { + node := g.nodes[id] + // The older road records a cancellation as a stopped node. A + // dollar-limit ending still reserves its unfinished assignment. + if node == nil || node.stopped && node.ending != TaskEndingCostLimit || g.nodeRootLocked(id) != root { + continue + } + named[plandb.NormalizedTitle(node.spec.title)] = node + } + return named +} + +func (g *TaskGraph) pendingNodeTitleLocked(root uint64, key string) bool { + for owner, pending := range g.childTitles { + if g.nodeRootLocked(owner) != root { + continue + } + for _, held := range pending { + if plandb.NormalizedTitle(held) == key { + return true + } + } + } + return false +} + +func (g *TaskGraph) releaseChildTitleLocked(parent uint64, titles []string) { + for _, title := range titles { + pending := g.childTitles[parent] + for i, held := range pending { + if plandb.NormalizedTitle(held) == plandb.NormalizedTitle(title) { + g.childTitles[parent] = append(pending[:i], pending[i+1:]...) + break + } + } + } + if len(g.childTitles[parent]) == 0 { + delete(g.childTitles, parent) + } +} diff --git a/internal/session/task_quick.go b/internal/session/task_quick.go index 77291c839b..d4a49562b2 100644 --- a/internal/session/task_quick.go +++ b/internal/session/task_quick.go @@ -501,7 +501,7 @@ func (a *Agent) admitQuick(ask quickAsk) (uint64, taskSpec, quickRefusal) { return 0, taskSpec{}, quickRefusal{said: refusal} } - if refused := graph.claimChild(spec.parent); refused != "" { + if refused := graph.claimChild(spec.parent, spec.title); refused != "" { return 0, taskSpec{}, quickRefusal{said: refused, fanFull: true} } id := graph.reserve() diff --git a/internal/session/task_run.go b/internal/session/task_run.go index 455dc272a9..433687fab0 100644 --- a/internal/session/task_run.go +++ b/internal/session/task_run.go @@ -353,6 +353,8 @@ type TaskNode struct { admit sync.Mutex spec taskSpec + // The root retains its shared spend across a checkpoint restore. + savedTreeSpend *nodeTreeSpendRecord // kind is what sort of node this is ([TaskKind]), settled at admission from // the spec and never touched again. It is a FIELD rather than a question put // to the spec because a node restored from a checkpoint has no spec worth @@ -1136,6 +1138,10 @@ type TaskGraph struct { // at rest — every claim is released by the admission it authorized or by the // proposal that came to nothing. claims map[uint64]int + // Pending titles stay reserved until their own proposal lands or withdraws. + childTitles map[uint64][]string + // Each older-road root owns one allowance shared by all its descendants. + nodeTrees map[uint64]*nodeTreeSpend // run executes one node to completion and calls [TaskGraph.complete] when // it lands. It is a field rather than a method so the graph can be exercised @@ -1459,7 +1465,7 @@ func (g *TaskGraph) admit(id uint64, spec taskSpec) TaskState { refused := g.quitting || g.deleted[spec.parent] if g.deleted[spec.parent] { g.deleted[id] = true - g.releaseChildLocked(spec.parent) + g.releaseChildLocked(spec.parent, spec.title) g.mu.Unlock() return TaskFailed } @@ -1472,8 +1478,9 @@ func (g *TaskGraph) admit(id uint64, spec taskSpec) TaskState { g.order = append(g.order, id) // The node counts itself from here, so the slot its proposal was holding // goes back (see [TaskGraph.claimChild]). - g.releaseChildLocked(spec.parent) + g.releaseChildLocked(spec.parent, spec.title) g.mu.Unlock() + g.spendTree(node) // ADMISSION IS A TRANSITION, and it is checkpointed before the frontier turns // rather than after: a process killed between these two lines still resumes @@ -1504,7 +1511,7 @@ func (g *TaskGraph) admit(id uint64, spec taskSpec) TaskState { // model fanning out emits its propose_task calls in one batch, which is exactly // the moment this cap is for. Every one of them would read the same count and // every one of them would pass. -func (g *TaskGraph) claimChild(parent uint64) string { +func (g *TaskGraph) claimChild(parent uint64, titles ...string) string { if parent == 0 { // The conversation's own work is not fanned out and is not capped: the // person is watching every proposal go by and can stop any of them. @@ -1512,19 +1519,29 @@ func (g *TaskGraph) claimChild(parent uint64) string { } g.mu.Lock() defer g.mu.Unlock() + if refusal := g.nodeGrowthRefusalLocked(parent, titles); refusal != "" { + return refusal + } + count := max(1, len(titles)) held := g.claims[parent] for _, node := range g.nodes { if node != nil && node.parent == parent { held++ } } - if held >= taskFanLimit { + if held+count > taskFanLimit { return fmt.Sprintf("no: you have already handed out %d pieces of this work, which is as many as one task may. Do the rest in your own hands, or finish these and report what is left undone.", taskFanLimit) } if g.claims == nil { g.claims = make(map[uint64]int, 1) } - g.claims[parent]++ + g.claims[parent] += count + if !bashBeltAsked() && len(titles) > 0 { + if g.childTitles == nil { + g.childTitles = make(map[uint64][]string) + } + g.childTitles[parent] = append(g.childTitles[parent], titles...) + } return "" } @@ -1542,17 +1559,18 @@ func (g *TaskGraph) claimChild(parent uint64) string { // never come and the run's ending cancels it first. The pass runs OUTSIDE the // lock: it takes the plan gate before the graph's mu, and nothing may hold // that mu while asking for the gate. -func (g *TaskGraph) releaseChild(parent uint64) { +func (g *TaskGraph) releaseChild(parent uint64, titles ...string) { g.mu.Lock() freed := g.claims[parent] > 0 - g.releaseChildLocked(parent) + g.releaseChildLocked(parent, titles...) g.mu.Unlock() if freed { g.planPulse() } } -func (g *TaskGraph) releaseChildLocked(parent uint64) { +func (g *TaskGraph) releaseChildLocked(parent uint64, titles ...string) { + g.releaseChildTitleLocked(parent, titles) if parent == 0 || g.claims == nil { return } @@ -1601,6 +1619,7 @@ func (g *TaskGraph) children(parent uint64) []*TaskNode { // call when nothing can move: the cost of a pass with nothing to do is one lock // and a walk of the order slice. func (g *TaskGraph) runFrontier() { + g.prepareNodeTrees() // The machine is READ before the lock and at most once a pass: a handful of // small file reads behind a one-second cache, and the graph's lock is held // by everything that announces a node — no reading of /proc belongs under @@ -1630,7 +1649,7 @@ func (g *TaskGraph) runFrontier() { if !queuedWithMemoryContext(node, orders) { continue } - ready, blocked := g.readinessLocked(node) + ready, blocked := g.nodeReadinessLocked(node) if blocked != "" { node.setStateLocked(TaskFailed) node.ended = time.Now() @@ -2168,6 +2187,7 @@ func shareReports(reports []string, pot int) []string { // state transition and the slot being handed back. func (g *TaskGraph) complete(node *TaskNode, state TaskState) { g.mu.Lock() + state = g.nodeTreeLandingLocked(node, state) node.setStateLocked(state) node.app.Version++ node.ended = time.Now() @@ -5250,6 +5270,7 @@ func (a *Agent) runTaskNode(node *TaskNode) { // whether or not this line has been reached yet. Time is checked only // between completed turns below. ctx, cancel := node.runContext() + ctx = withNodeTree(ctx, node) // AND THE NODE IS TAKEN UP HERE, OR NOT RUN AT ALL. Close and a person's stop // both land in the window between the admission and this line, and the whole @@ -7964,6 +7985,7 @@ func (a *Agent) newTaskAgentOn(ctx context.Context, dir string, node *TaskNode, quickItems: node.quickDoor(), taskControl: node, taskCapUSD: taskCapUSD, + nodeTree: node.graph.spendTree(node), // AND THE DESIGN THREAD'S ONE EXTRA HAND, wired here for roomThread's // reason: a belt is assembled once, when the agent is constructed // (agent.go), so a door handed over after this call would be a verb the diff --git a/internal/session/task_store.go b/internal/session/task_store.go index 2ca74dce61..595f198243 100644 --- a/internal/session/task_store.go +++ b/internal/session/task_store.go @@ -431,11 +431,12 @@ type taskRecord struct { // carried a bill, which resumes as zero — the same thing an unpriced model // leaves behind, and the same thing every surface here already draws as // nothing rather than as "$0.00". - CostUSD float64 `json:"costUsd,omitempty"` - Input int `json:"input,omitempty"` - Output int `json:"output,omitempty"` - CacheRead int `json:"cacheRead,omitempty"` - CacheWrite int `json:"cacheWrite,omitempty"` + CostUSD float64 `json:"costUsd,omitempty"` + TreeSpend *nodeTreeSpendRecord `json:"tree_spend,omitempty"` + Input int `json:"input,omitempty"` + Output int `json:"output,omitempty"` + CacheRead int `json:"cacheRead,omitempty"` + CacheWrite int `json:"cacheWrite,omitempty"` // Noted says this node's completion note has been handed to the steering // lane. It is what stops a resumed session re-announcing work the transcript @@ -1444,6 +1445,7 @@ func (n *TaskNode) recordLocked() taskRecord { StartedAt: n.started, EndedAt: n.ended, CostUSD: n.cost, + TreeSpend: n.graph.nodeSpendRecordLocked(n), Input: n.input, Output: n.output, CacheRead: n.cacheRead, @@ -2386,6 +2388,7 @@ func restoreNode(graph *TaskGraph, record taskRecord) *TaskNode { started: record.StartedAt, ended: record.EndedAt, cost: record.CostUSD, + savedTreeSpend: record.TreeSpend, input: record.Input, output: record.Output, cacheRead: record.CacheRead, diff --git a/internal/session/taskname.go b/internal/session/taskname.go index 7378c02220..052676c0af 100644 --- a/internal/session/taskname.go +++ b/internal/session/taskname.go @@ -186,7 +186,7 @@ func (g *TaskGraph) nameNode(node *TaskNode) { if subject == "" { return } - if name := home.taskName(context.Background(), subject); name != "" { + if name := home.taskName(withNodeTree(context.Background(), node), subject); name != "" { g.rename(node, name) } }() From 4a2e533452dfbc3244fe335666c11d15d55274e5 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Fri, 6 Nov 2026 09:13:48 -0500 Subject: [PATCH 2/4] docs: change entry for the older road's tree bounds (#2464) Co-Authored-By: Claude Opus 5.5 --- .../unreleased/2464-node-road-tree-bounds.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 docs/changes/unreleased/2464-node-road-tree-bounds.md diff --git a/docs/changes/unreleased/2464-node-road-tree-bounds.md b/docs/changes/unreleased/2464-node-road-tree-bounds.md new file mode 100644 index 0000000000..512537a212 --- /dev/null +++ b/docs/changes/unreleased/2464-node-road-tree-bounds.md @@ -0,0 +1,13 @@ +--- +kind: fixed +title: the older task road refuses duplicate work in a tree and holds the tree to one spend total +pr: 2464 +surface: [engine, chat, docs] +invalidates: + - "Was true: with CODEAF_TASK_BELT=node two tasks in one tree could hand out the same work, and only each node's own bill was limited. Is true now: duplicate titles are refused tree-wide and the whole tree shares one spend total." +--- + +The duplicate check uses the plan store's title normalization, so both task roads compare +work the same way. The tree total is the lower of the per-task limit (crew.task_cap) and +the conversation's remaining limit; reaching it stops the whole tree and names the limit. +A typed task starts its own tree. The default road is unchanged. From 7061e50343d4557e788c6d28099d5d8384476f15 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Fri, 6 Nov 2026 09:39:28 -0500 Subject: [PATCH 3/4] Tasks: the older road's tree total is checked where the node limit is Review round: the first version held the tree to a new shared guard with its own stop and cancellation, which read a limit stop as a person's stop, missed paid tool receipts, could re-grant a conversation-bound allowance after a restore, and waited on checkpoint writes before cancelling. The tree total now reuses the older road's existing per-node check instead: before each worker request, the summed bill of the tree's root and every descendant is compared with the tree's ceiling, which is fixed when the root is admitted (the default road's arithmetic) and kept in the checkpoint. A node over it ends on the cost-limit ending with the configured-limit words, or the conversation-limit words when that bound it. The shared guard and its hooks are gone. TestAPartsBriefHoldsItsBoundWhateverLengthTheFieldsArriveAt sent two parts with the same title, which a division now refuses; its second part's title differs and every field stays at full length. Co-Authored-By: Claude Opus 5.5 --- internal/manual/chat/tasks.md | 6 +- internal/session/agent.go | 4 - internal/session/clientdoor.go | 3 - internal/session/loop.go | 4 + internal/session/session.go | 2 - internal/session/task_app.go | 3 +- internal/session/task_audit.go | 1 - internal/session/task_divide_compose_test.go | 5 +- internal/session/task_node_bounds_test.go | 493 +++++++++++-------- internal/session/task_node_spend.go | 249 +++------- internal/session/task_node_titles.go | 3 - internal/session/task_run.go | 15 +- internal/session/task_store.go | 34 +- internal/session/taskname.go | 2 +- 14 files changed, 414 insertions(+), 410 deletions(-) diff --git a/internal/manual/chat/tasks.md b/internal/manual/chat/tasks.md index 7cc080b3f6..59d9458b58 100644 --- a/internal/manual/chat/tasks.md +++ b/internal/manual/chat/tasks.md @@ -2327,8 +2327,10 @@ one ceiling for the whole tree, including planning and checks. The smaller of th and the conversation's remaining allowance wins, even when no crew was routed. Continuing interrupted work keeps what the same tree already spent: only the unspent part of its per-task limit remains available. -A routed crew refuses a call that would cross its per-task limit. Reaching the tree's -ceiling ends workers in flight and starts no more. When the per-task limit binds, the +A routed crew refuses a call that would cross its per-task limit. Reaching the run's +ceiling ends workers in flight and starts no more. +On the node belt, the saved ceiling survives restarts; reaching it refuses each worker's next request, queued workers spend nothing, and in-flight calls finish without cancellation. +When the per-task limit binds, the report says `this task reached its $5 limit · raise it in /crew` (with your configured amount). Raise it with `/crew cap task <$>`, then rerun the remaining work with `/task`. When the conversation's smaller limit binds, the report names that amount and `/settings`. diff --git a/internal/session/agent.go b/internal/session/agent.go index be4e9d7f0f..81d64c8f64 100644 --- a/internal/session/agent.go +++ b/internal/session/agent.go @@ -92,10 +92,6 @@ func New(config Config) (*Agent, error) { func (a *Agent) newChildAgent(config Config) (*Agent, error) { config.ClientTools = nil config.System = clientToolsWithoutSections(config.System) - // A helper built inside a task spends from that task's tree too. - if config.nodeTree == nil { - config.nodeTree = a.config.nodeTree - } if pair, ok := a.ConversationModelPair(); ok && config.ConversationModels == nil { pair.Main = config.Model if config.OneModel || pair.Fallback == pair.Main { diff --git a/internal/session/clientdoor.go b/internal/session/clientdoor.go index eb841ea58e..b6edc500d0 100644 --- a/internal/session/clientdoor.go +++ b/internal/session/clientdoor.go @@ -841,9 +841,6 @@ func (a *Agent) completeWithNamedModel(ctx context.Context, purpose callPurpose, ctx = provider.WithServedModel(ctx, a.servedModel) defer func() { a.endModelRequest(actual) }() } - if tree := a.nodeTreeForCall(ctx); tree != nil { - client = nodeTreeCompleter{tree: tree, model: called, next: client} - } response, err := client.CompleteWithMessages(toldItsWindow(ctx), materializeImageRefs(messages), append(options, ai.WithModel(wire))...) if response != nil && response.Model != "" { actual = response.Model diff --git a/internal/session/loop.go b/internal/session/loop.go index 6c04151bf2..9c41b25384 100644 --- a/internal/session/loop.go +++ b/internal/session/loop.go @@ -1941,6 +1941,10 @@ func (a *Agent) completeWithRetryReasoning(ctx context.Context, hub *eventHub, m if err := node.appBoundary(ctx); err != nil { return nil, model, err } + if action := node.treeSpendLimit(); action != "" { + node.end(TaskEndingCostLimit) + return nil, model, ErrSpendStopped{Action: action} + } if a.config.taskCapUSD > 0 && node.spend() >= a.config.taskCapUSD { return nil, model, ErrSpendStopped{Action: config.CrewTaskCapAction(a.config.taskCapUSD)} } diff --git a/internal/session/session.go b/internal/session/session.go index a980b23a42..0f4f7aba69 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -2543,8 +2543,6 @@ type Config struct { taskControl *TaskNode // The saved per-task cap applies across this node's worker and repair rounds. taskCapUSD float64 - // Workers, repairs and checks on the older road share their root's guard. - nodeTree *nodeTreeSpend // fixesDir is the project bucket the error→fix sidecar keeps its file in // (fixstore.go), and it is set only when this agent is a NODE. A node's diff --git a/internal/session/task_app.go b/internal/session/task_app.go index f7fda3341e..36bacd05fa 100644 --- a/internal/session/task_app.go +++ b/internal/session/task_app.go @@ -196,6 +196,7 @@ func (g *TaskGraph) writeAppControl(change func() (func(), error)) error { store.mu.Lock() defer store.mu.Unlock() } + bills := g.nodeTreeBills() g.mu.Lock() defer g.mu.Unlock() rollback, err := change() @@ -203,7 +204,7 @@ func (g *TaskGraph) writeAppControl(change func() (func(), error)) error { return err } if store != nil { - if err := store.writeLocked(g.documentLocked()); err != nil { + if err := store.writeLocked(g.documentLocked(bills)); err != nil { if rollback != nil { rollback() } diff --git a/internal/session/task_audit.go b/internal/session/task_audit.go index 8a42705c67..ffd71de6cb 100644 --- a/internal/session/task_audit.go +++ b/internal/session/task_audit.go @@ -3227,7 +3227,6 @@ func (a *Agent) newAuditAgent(dir string, node *TaskNode, door auditDoor, on str // usage row's task ([Config.checksNode]) — while taskID above stays 0, // because the auditor is not the node it reads. checksNode: node.id, - nodeTree: node.graph.spendTree(node), // The auditor is the node too, as far as anybody watching is concerned: // it runs on the node's clock, in the node's worktree, and a card whose // audit is parked on a provider's pacing is a card whose task is not diff --git a/internal/session/task_divide_compose_test.go b/internal/session/task_divide_compose_test.go index 46c54c31b6..9bdf67968b 100644 --- a/internal/session/task_divide_compose_test.go +++ b/internal/session/task_divide_compose_test.go @@ -189,7 +189,10 @@ func TestAPartsBriefHoldsItsBoundWhateverLengthTheFieldsArriveAt(t *testing.T) { }, false}, {"every field at once", []dividePart{ {Title: huge, Summary: huge, Brief: huge}, - {Title: huge, Summary: huge, Brief: huge}, + // A division refuses two parts naming the same work, so the second + // part's title differs from its first word while every field stays + // at full length. + {Title: "the beta adapter " + huge, Summary: huge, Brief: huge}, }, false}, } { t.Run(test.name, func(t *testing.T) { diff --git a/internal/session/task_node_bounds_test.go b/internal/session/task_node_bounds_test.go index df9d5c7784..71e568e704 100644 --- a/internal/session/task_node_bounds_test.go +++ b/internal/session/task_node_bounds_test.go @@ -2,6 +2,7 @@ package session import ( "context" + "errors" "fmt" "path/filepath" "strings" @@ -125,29 +126,52 @@ func TestNodeTreeReservesPendingProposalTitles(t *testing.T) { other.Withdraw() } -// C4: several nodes exhaust one ceiling, cut an in-flight call, fail queued work and prevent further admission. +// C3, C4: real runners refuse each tree member's next request at the binding limit; other trees still finish. func TestNodeTreeSpendStopsEveryNodeAndNamesTheBindingLimit(t *testing.T) { t.Setenv("CODEAF_TASK_BELT", "node") + useTaskUsageHome(t) for _, conversation := range []bool{false, true} { t.Run(fmt.Sprint(conversation), func(t *testing.T) { - entered := make(chan struct{}) + rootEntered, childEntered := make(chan struct{}), make(chan struct{}) + rootHold, releaseRoot := context.WithCancel(t.Context()) + defer releaseRoot() + childHold, releaseChild := context.WithCancel(t.Context()) + defer releaseChild() completer := &scriptedCompleter{steps: []step{ - func(context.Context, []ai.Message) (*ai.Response, error) { return pricedResponse("first", .4), nil }, + func(context.Context, []ai.Message) (*ai.Response, error) { + return nodeTreeReadResponse("root-read", .5), nil + }, func(ctx context.Context, _ []ai.Message) (*ai.Response, error) { - close(entered) - <-ctx.Done() - return nil, ctx.Err() + close(rootEntered) + select { + case <-rootHold.Done(): + return nodeTreeReadResponse("root-read-again", 0), nil + case <-ctx.Done(): + return nil, ctx.Err() + } }, - func(context.Context, []ai.Message) (*ai.Response, error) { - cost := 1.1 - if conversation { - cost = .6 + func(ctx context.Context, _ []ai.Message) (*ai.Response, error) { + close(childEntered) + select { + case <-childHold.Done(): + cost := 1.0 + if conversation { + cost = .5 + } + return nodeTreeReadResponse("part-read", cost), nil + case <-ctx.Done(): + return nil, ctx.Err() } - return pricedResponse("last", cost), nil + }, + func(context.Context, []ai.Message) (*ai.Response, error) { + return pricedResponse("another tree finished", .125), nil }, }} a, _ := newTestAgent(t, completer, func(c *Config) { c.ProfileDir = t.TempDir() + writeFile(t, filepath.Join(c.Workspace, "rules.txt"), "Read the rules.\n") + c.TaskParallel = 2 + c.TaskAudit = false c.Interactive = true if conversation { c.Budget = Budget{USD: 1.25} @@ -161,153 +185,201 @@ func TestNodeTreeSpendStopsEveryNodeAndNamesTheBindingLimit(t *testing.T) { } g := a.graph() g.report = nil - g.run = func(*TaskNode) {} - g.limit = 3 - rootID := g.reserve() - g.admit(rootID, taskSpec{title: "Whole assignment", named: true, depth: 1, model: "test/model"}) - root := g.node(rootID) - workers := make([]*Agent, 0, 3) - nodes := []*TaskNode{root} - for i := 0; i < 3; i++ { - title := fmt.Sprintf("Part %d", i) - if refusal := g.claimChild(rootID, title); refusal != "" { - t.Fatal(refusal) - } - id := g.reserve() - g.admit(id, taskSpec{title: title, named: true, parent: rootID, depth: 2, model: "test/model"}) - nodes = append(nodes, g.node(id)) - } - for _, node := range nodes[:3] { - worker, err := a.newTaskAgent(t.Context(), t.TempDir(), node, "") - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = worker.Close() }) - workers = append(workers, worker) + g.store = newTaskStore(filepath.Join(t.TempDir(), "tasks.json")) + root := nodeTreeQuick(t, a, 0, "Whole assignment") + nodeTreeAwait(t, rootEntered) + part := nodeTreeQuick(t, a, root.id, "Read the rules") + nodeTreeAwait(t, childEntered) + queued := nodeTreeQuick(t, a, root.id, "Read the examples") + if queued.stateNow() != TaskQueued { + t.Fatal("third node did not queue behind the two live workers") } - if _, err := workers[0].completeWithModel(t.Context(), purposeTurn, nil, "test/model"); err != nil { - t.Fatal(err) + releaseChild() + waitDoneNode(t, part) + waitDoneNode(t, queued) + if queued.notice().StartedAt.IsZero() { + t.Fatal("queued node did not reach its first request boundary") } - // Keep this node owned until its provider observes cancellation. - if !nodes[1].claimRun() { - t.Fatal("in-flight node was not running") - } - ended := make(chan error, 1) - go func() { - _, err := workers[1].completeWithModel(t.Context(), purposeTurn, nil, "test/model") - ended <- err - }() + // Reaching the limit leaves the already admitted provider call alone. select { - case <-entered: - case <-time.After(5 * time.Second): - t.Fatal("scripted call never entered") + case <-root.done: + t.Fatal("the tree limit cancelled a request in flight") + default: } - if _, err := workers[2].completeWithModel(t.Context(), purposeTask, nil, "test/model"); err != nil { - t.Fatal(err) + if queued.spend() != 0 || completer.requests() != 3 { + t.Fatalf("queued node bought a request: spend=%v, calls=%d", queued.spend(), completer.requests()) } - select { - case err := <-ended: - if err == nil { - t.Fatal("in-flight call was not cancelled") - } - case <-time.After(5 * time.Second): - t.Fatal("tree ceiling left a call in flight") + otherID := g.reserve() + g.admit(otherID, taskSpec{title: "Whole assignment", named: true, brief: "Read the rules", acceptance: "The rules are read", depth: 1, model: "test/model"}) + other := g.node(otherID) + waitDoneNode(t, other) + if row := other.notice(); row.State != TaskDone || !strings.Contains(row.Report, "another tree finished") || row.CostUSD != .125 { + t.Fatalf("another tree inherited the exhausted allowance: %+v", row) } - g.complete(nodes[1], TaskFailed) + releaseRoot() + waitDoneNode(t, root) want := config.CrewTaskCapAction(1.5) if conversation { - want = "conversation" + want = "the conversation reached its $1.25 limit · raise the per conversation limit in /settings" } - for _, node := range nodes { - row := node.notice() - if row.State != TaskFailed || row.Ending != TaskEndingCostLimit || !strings.Contains(row.Report, want) || conversation && strings.Contains(row.Report, "raise it in /crew") { - t.Fatalf("tree limit ending is wrong: %+v", row) - } - if conversation && (!strings.Contains(row.Report, "$1.25") || !strings.Contains(row.Report, "/settings")) { - t.Fatalf("conversation limit lost its amount or raise action: %+v", row) + for _, node := range []*TaskNode{root, part, queued} { + nodeTreeCostEnding(t, node, want) + if conversation && strings.Contains(node.notice().Report, "/crew") { + t.Fatalf("conversation limit was described as the task cap: %+v", node.notice()) } } - if refusal := g.claimChild(rootID, "Another part"); !strings.Contains(refusal, want) { - t.Fatalf("spent tree admitted another node: %s", refusal) + if calls := completer.requests(); calls != 4 { + t.Fatalf("exhausted tree reached the provider again: calls=%d, want 4 including the other tree", calls) } - completer.mu.Lock() - calls := len(completer.seen) - completer.mu.Unlock() - if calls != 3 || !nodes[3].started.IsZero() { - t.Fatalf("queued work reached the provider: calls=%d, queued started=%v", calls, nodes[3].started) + g.checkpoint() + document := readCheckpoint(t, g.store.path) + recovered := newTaskGraph() + recovered.rehydrate(document, a.config.Workspace, TaskSettleAsk) + for _, node := range []*TaskNode{root, part, queued} { + record := recordOf(t, document, node.id) + if record.Ending != TaskEndingCostLimit || recovered.node(node.id).stoppedByPerson() { + t.Fatalf("cost ending became a person's stop after restore: %+v", record) + } } }) } } -// C5: one node below the ceiling runs through the real quick-task runner and finishes normally. +// C5: an ordinary node below the ceiling runs through the real runner and finishes normally. func TestNodeTreeUnderTheCeilingFinishesNormally(t *testing.T) { t.Setenv("CODEAF_TASK_BELT", "node") + useTaskUsageHome(t) a, _ := newTestAgent(t, &scriptedCompleter{steps: []step{func(context.Context, []ai.Message) (*ai.Response, error) { return pricedResponse("finished", .25), nil - }}}, func(c *Config) { c.ProfileDir = t.TempDir() }) + }}}, func(c *Config) { + c.ProfileDir = t.TempDir() + c.TaskAudit = false + }) g := a.graph() g.report = nil - id, _, refusal := a.admitQuick(quickAsk{line: "Read the law"}) - if refusal.said != "" { - t.Fatal(refusal.said) - } + id := g.reserve() + g.admit(id, taskSpec{title: "Read the law", named: true, brief: "Read the law", acceptance: "The law is read", depth: 1, model: "test/model"}) node := g.node(id) waitDoneNode(t, node) - if row := node.notice(); row.State != TaskDone || row.Report != "finished" || row.CostUSD != .25 || row.Ending == TaskEndingCostLimit { + if row := node.notice(); row.State != TaskDone || !strings.Contains(row.Report, "finished") || row.CostUSD != .25 || row.Ending == TaskEndingCostLimit { t.Fatalf("under-ceiling node changed: %+v", row) } } -// C4: a restored root keeps exactly the shared spend, including child work already folded into its bill. +// C4: recovery retains a conversation-bound ceiling and a paid live worker's bill before any owner fold. func TestNodeTreeCheckpointKeepsTheUnspentAllowance(t *testing.T) { t.Setenv("CODEAF_TASK_BELT", "node") + useTaskUsageHome(t) profile := t.TempDir() - if err := config.SetCrewTaskCap(profile, "1"); err != nil { + if err := config.SetCrewTaskCap(profile, "5"); err != nil { t.Fatal(err) } - a, _ := newTestAgent(t, &scriptedCompleter{steps: []step{func(context.Context, []ai.Message) (*ai.Response, error) { - return pricedResponse("partly done", .75), nil - }}}, func(c *Config) { c.ProfileDir = profile }) + entered := make(chan struct{}) + hold, release := context.WithCancel(t.Context()) + defer release() + completer := &scriptedCompleter{steps: []step{ + func(context.Context, []ai.Message) (*ai.Response, error) { + return nodeTreeReadResponse("before-recovery", .75), nil + }, + func(ctx context.Context, _ []ai.Message) (*ai.Response, error) { + close(entered) + select { + case <-hold.Done(): + return pricedResponse("finished after the snapshot", 0), nil + case <-ctx.Done(): + return nil, ctx.Err() + } + }, + }} + options := func(c *Config) { + c.ProfileDir = profile + writeFile(t, filepath.Join(c.Workspace, "rules.txt"), "Read the rules.\n") + c.Interactive = true + c.Budget = Budget{USD: 1.25} + c.TaskAudit = false + } + a, _ := newTestAgent(t, completer, options) + a.usage.CostUSD = .25 g := a.graph() - g.run, g.report = func(*TaskNode) {}, nil + g.report = nil checkpoint := filepath.Join(t.TempDir(), "tasks.json") g.store = newTaskStore(checkpoint) id := g.reserve() g.admit(id, taskSpec{title: "Read the law", named: true, brief: "Read the law", acceptance: "The law is read", depth: 1, model: "test/model"}) - worker, err := a.newTaskAgent(t.Context(), t.TempDir(), g.node(id), "") - if err != nil { - t.Fatal(err) + root := g.node(id) + nodeTreeAwait(t, entered) + g.checkpoint() + // Atomic control writes must retain the same unfolded bill as ordinary saves. + for _, action := range []string{"pause", "resume"} { + if err := a.AppTaskAction(AppTaskIntent{ID: fmt.Sprint(id), Action: action}); err != nil { + t.Fatal(err) + } } - t.Cleanup(func() { _ = worker.Close() }) - if _, err := worker.completeWithModel(t.Context(), purposeTurn, nil, "test/model"); err != nil { + document := readCheckpoint(t, checkpoint) + record := recordOf(t, document, id) + if record.TreeLimit == nil || record.TreeLimit.CeilingUSD != 1 || !record.TreeLimit.Conversation || record.TreeLimit.ConversationCapUSD != 1.25 || record.CostUSD != .75 { + t.Fatalf("checkpoint lost the admitted ceiling or live bill: %+v", record) + } + g.mu.Lock() + folded := root.cost + g.mu.Unlock() + if folded != 0 || a.Usage().CostUSD != .25 { + t.Fatalf("snapshot was taken after folding: node=%v, conversation=%v", folded, a.Usage().CostUSD) + } + // Changed settings and the owner's still-unfolded books cannot grant a new ceiling. + if err := config.SetCrewTaskCap(profile, "10"); err != nil { t.Fatal(err) } - document := readCheckpoint(t, checkpoint) - if saved := recordOf(t, document, id).TreeSpend; saved == nil || saved.Spent != .75 { - t.Fatalf("checkpoint lost the tree's live spend: %+v", saved) - } - // A parent's historic bill can already contain its finished child's bill. - document.Nodes[0].State, document.Nodes[0].CostUSD = TaskQueued, .75 - document.Nodes = append(document.Nodes, taskRecord{ID: id + 1, Parent: id, Depth: 2, Title: "Finished part", Brief: "Read the rules", Acceptance: "The rules are read", State: TaskDone, CostUSD: .25}) - document.Seq = id + 1 - b, _ := newTestAgent(t, &scriptedCompleter{steps: []step{func(context.Context, []ai.Message) (*ai.Response, error) { - return pricedResponse("the remaining work", .25), nil - }}}, func(c *Config) { c.ProfileDir = profile }) + resumedCalls := &scriptedCompleter{steps: []step{ + func(context.Context, []ai.Message) (*ai.Response, error) { + return nodeTreeReadResponse("after-recovery", .25), nil + }, + }} + b, _ := newTestAgent(t, resumedCalls, options) + b.usage.CostUSD = .25 restored := b.graph() - restored.run, restored.report = func(*TaskNode) {}, nil + restored.report = nil restored.rehydrate(document, b.config.Workspace, TaskSettleAsk) restored.runFrontier() - resumed, err := b.newTaskAgent(t.Context(), t.TempDir(), restored.node(id), "") - if err != nil { + resumed := restored.node(id) + waitDoneNode(t, resumed) + nodeTreeCostEnding(t, resumed, "the conversation reached its $1.25 limit · raise the per conversation limit in /settings") + if resumed.spend() != 1 || resumedCalls.requests() != 1 { + t.Fatalf("restored tree replenished its allowance: spend=%v, requests=%d", resumed.spend(), resumedCalls.requests()) + } + release() + waitDoneNode(t, root) +} + +// C4: a checkpoint without the additive ceiling fact retains the existing task-cap remainder. +func TestNodeTreeCheckpointWithoutLimitComputesItAtRestore(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "node") + useTaskUsageHome(t) + completer := &scriptedCompleter{steps: []step{ + func(context.Context, []ai.Message) (*ai.Response, error) { + return nodeTreeReadResponse("remaining-work", .25), nil + }, + }} + a, _ := newTestAgent(t, completer, func(c *Config) { + c.ProfileDir = t.TempDir() + writeFile(t, filepath.Join(c.Workspace, "rules.txt"), "Read the rules.\n") + c.TaskAudit = false + }) + if err := config.SetCrewTaskCap(a.config.ProfileDir, "1"); err != nil { t.Fatal(err) } - t.Cleanup(func() { _ = resumed.Close() }) - if _, err := resumed.completeWithModel(t.Context(), purposeTurn, nil, "test/model"); err != nil { - t.Fatalf("the actual remaining allowance was lost: %v", err) - } - if row := restored.node(id).notice(); row.Ending != TaskEndingCostLimit || !strings.Contains(row.Report, config.CrewTaskCapAction(1)) { - t.Fatalf("resumed work received a fresh tree allowance: %+v", row) + g := a.graph() + g.report = nil + g.rehydrate(taskDocument{Type: taskDocumentType, Version: taskFileVersion, Seq: 1, Nodes: []taskRecord{{ + ID: 1, Title: "Read the law", Brief: "Read the law", Acceptance: "The law is read", + State: TaskQueued, Depth: 1, Model: "test/model", CostUSD: .75, + }}}, a.config.Workspace, TaskSettleAsk) + g.runFrontier() + root := g.node(1) + waitDoneNode(t, root) + nodeTreeCostEnding(t, root, config.CrewTaskCapAction(1)) + if root.spend() != 1 || completer.requests() != 1 { + t.Fatalf("old checkpoint lost its remaining allowance: spend=%v, requests=%d", root.spend(), completer.requests()) } } @@ -333,106 +405,139 @@ func TestNodeTreeUsesThePlanStoresWorkIdentity(t *testing.T) { } } -// C3, C4: workers, repairs, checks and owner helpers share one tree's money; another tree remains live. -func TestNodeTreeCountsEveryCallAndKeepsOtherTreesIndependent(t *testing.T) { +// C4: a queued worker in an already exhausted conversation is refused at its first request without spending. +func TestNodeTreeWithNoConversationAllowanceSpendsNothing(t *testing.T) { t.Setenv("CODEAF_TASK_BELT", "node") + useTaskUsageHome(t) completer := &scriptedCompleter{} - for i := 0; i < 7; i++ { - completer.steps = append(completer.steps, func(context.Context, []ai.Message) (*ai.Response, error) { - return pricedResponse("a paid answer", .125), nil - }) + a, _ := newTestAgent(t, completer, func(c *Config) { + c.ProfileDir = t.TempDir() + c.Interactive = true + c.Budget = Budget{USD: 1} + }) + a.usage.CostUSD = 1 + g := a.graph() + g.report = nil + root := nodeTreeQuick(t, a, 0, "Read the law") + waitDoneNode(t, root) + nodeTreeCostEnding(t, root, "the conversation reached its $1 limit · raise the per conversation limit in /settings") + if root.spend() != 0 || completer.requests() != 0 { + t.Fatalf("an exhausted conversation reached the provider: spend=%v, calls=%d", root.spend(), completer.requests()) } - a, _ := newTestAgent(t, completer, func(c *Config) { c.ProfileDir = t.TempDir() }) - if err := config.SetCrewTaskCap(a.config.ProfileDir, "0.75"); err != nil { +} + +// C4: a reached ceiling refuses the next request even while checkpoint writes are blocked. +func TestNodeTreeLimitRefusesBeforeCheckpointWrites(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "node") + useTaskUsageHome(t) + entered := make(chan struct{}) + hold, release := context.WithCancel(t.Context()) + defer release() + completer := &scriptedCompleter{steps: []step{ + func(ctx context.Context, _ []ai.Message) (*ai.Response, error) { + close(entered) + select { + case <-hold.Done(): + return nodeTreeReadResponse("last-read", .5), nil + case <-ctx.Done(): + return nil, ctx.Err() + } + }, + }} + a, _ := newTestAgent(t, completer, func(c *Config) { + c.ProfileDir = t.TempDir() + writeFile(t, filepath.Join(c.Workspace, "rules.txt"), "Read the rules.\n") + c.Interactive = true + c.Budget = Budget{USD: .5} + }) + if err := config.SetCrewTaskCap(a.config.ProfileDir, "1.5"); err != nil { t.Fatal(err) } g := a.graph() - g.run, g.report = func(*TaskNode) {}, nil - rootID := g.reserve() - g.admit(rootID, taskSpec{title: "Read the law", named: true, depth: 1, model: "test/model"}) - otherID := g.reserve() - g.admit(otherID, taskSpec{title: "Read the law", named: true, depth: 1, model: "test/model"}) - childID := g.reserve() - g.admit(childID, taskSpec{title: "Read the rules", named: true, parent: rootID, depth: 2, model: "test/model"}) - root := g.node(rootID) - builds := []func() (*Agent, error){ - func() (*Agent, error) { return a.newTaskAgent(t.Context(), t.TempDir(), root, "") }, - func() (*Agent, error) { - return a.newTaskAgentOn(t.Context(), t.TempDir(), root, "-repair", "test/model", true) - }, - func() (*Agent, error) { - return a.newAuditAgent(t.TempDir(), root, plainDoor(auditReadCommands), "test/model") - }, - func() (*Agent, error) { return a.newTaskAgent(t.Context(), t.TempDir(), g.node(childID), "") }, - } - var rootWorker *Agent - for _, build := range builds { - worker, err := build() - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = worker.Close() }) - if rootWorker == nil { - rootWorker = worker - } - if _, err := worker.completeWithModel(t.Context(), purposeTurn, nil, "test/model"); err != nil { - t.Fatal(err) - } - } - helper, err := rootWorker.newChildAgent(Config{Workspace: t.TempDir(), Model: "test/model", System: "SYSTEM", InTask: true}) + g.report = nil + g.store = newTaskStore(filepath.Join(t.TempDir(), "tasks.json")) + root := nodeTreeQuick(t, a, 0, "Read the law") + nodeTreeAwait(t, entered) + events, leave, err := a.WatchTaskRoom(root.id) if err != nil { t.Fatal(err) } - t.Cleanup(func() { _ = helper.Close() }) - if _, err := helper.completeWithModel(t.Context(), purposeTurn, nil, "test/model"); err != nil { - t.Fatal(err) - } - if _, err := a.completeWithModel(withNodeTree(t.Context(), root), purposeTask, nil, "test/model"); err != nil { - t.Fatal(err) - } - for _, id := range []uint64{rootID, childID} { - if row := g.node(id).notice(); row.State != TaskFailed || row.Ending != TaskEndingCostLimit { - t.Fatalf("a paid role escaped the tree tally: %+v", row) + defer leave() + g.store.mu.Lock() + locked := true + defer func() { + if locked { + g.store.mu.Unlock() + } + }() + release() + action := "the conversation reached its $0.50 limit · raise the per conversation limit in /settings" + deadline := time.After(5 * time.Second) + refused := false + for !refused { + select { + case event, open := <-events: + if !open { + t.Fatal("worker ended before publishing its refusal") + } + if event.Kind == EventError { + var stopped ErrSpendStopped + if !errors.As(event.Err, &stopped) || stopped.Action != action { + t.Fatalf("wrong request refusal: %v", event.Err) + } + refused = true + } + case <-deadline: + t.Fatal("the spend boundary waited for checkpoint I/O") } } - other := g.node(otherID) - if other.stateNow() != TaskRunning { - t.Fatal("one tree's limit stopped a separate assignment") + if root.endingSet() != TaskEndingCostLimit || completer.requests() != 1 { + t.Fatalf("blocked checkpoint postponed the limit: ending=%q, calls=%d", root.endingSet(), completer.requests()) } - worker, err := a.newTaskAgent(t.Context(), t.TempDir(), other, "") - if err != nil { - t.Fatal(err) + g.store.mu.Unlock() + locked = false + waitDoneNode(t, root) + nodeTreeCostEnding(t, root, action) +} + +func nodeTreeReadResponse(id string, cost float64) *ai.Response { + response := toolResponse(id, "read", `{"path":"rules.txt"}`) + response.Usage.Cost = &cost + return response +} + +func nodeTreeAwait(t *testing.T, entered <-chan struct{}) { + t.Helper() + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("scripted worker did not reach its request") } - t.Cleanup(func() { _ = worker.Close() }) - if _, err := worker.completeWithModel(t.Context(), purposeTurn, nil, "test/model"); err != nil { - t.Fatalf("another tree lost its own allowance: %v", err) +} + +func nodeTreeQuick(t *testing.T, a *Agent, parent uint64, title string) *TaskNode { + t.Helper() + spec, refusal := a.newQuickSpec(quickAsk{line: title}) + if refusal != "" { + t.Fatal(refusal) } - other.finish("finished", nil, "", "") - g.complete(other, TaskDone) - if other.stateNow() != TaskDone { - t.Fatal("another tree could not finish") + g := a.graph() + spec.parent = parent + if parent != 0 { + spec.depth = g.node(parent).depth + 1 } - if got := completer.requests(); got != 7 { - t.Fatalf("paid calls = %d, want 7", got) + if refusal := g.claimChild(parent, title); refusal != "" { + t.Fatal(refusal) } + id := g.reserve() + g.admit(id, spec) + return g.node(id) } -// C4: an already spent conversation allowance starts no node, even with an unpriced completer. -func TestNodeTreeWithNoConversationAllowanceNeverStarts(t *testing.T) { - t.Setenv("CODEAF_TASK_BELT", "node") - completer := &scriptedCompleter{} - a, _ := newTestAgent(t, completer, func(c *Config) { - c.ProfileDir = t.TempDir() - c.Interactive = true - c.Budget = Budget{USD: 1} - }) - a.usage.CostUSD = 1 - g := a.graph() - g.run, g.report = func(*TaskNode) { t.Error("a node without any allowance started") }, nil - id := g.reserve() - g.admit(id, taskSpec{title: "Read the law", named: true, brief: "Read the law", acceptance: "The law is read", depth: 1}) - row := g.node(id).notice() - if row.State != TaskFailed || row.Ending != TaskEndingCostLimit || !row.StartedAt.IsZero() || !strings.Contains(row.Report, "conversation") || !strings.Contains(row.Report, "$1") || completer.requests() != 0 { - t.Fatalf("an exhausted conversation allowance reached a worker: %+v, calls=%d", row, completer.requests()) +func nodeTreeCostEnding(t *testing.T, node *TaskNode, action string) { + t.Helper() + row := node.notice() + if row.State != TaskFailed || row.Ending != TaskEndingCostLimit || row.Stopped || !strings.Contains(row.Report, action) { + t.Fatalf("wrong cost-limit ending, want %q: %+v", action, row) } } diff --git a/internal/session/task_node_spend.go b/internal/session/task_node_spend.go index ba4a2d01d4..df7cd2ed32 100644 --- a/internal/session/task_node_spend.go +++ b/internal/session/task_node_spend.go @@ -1,230 +1,129 @@ package session import ( - "context" - "errors" "fmt" "math" - "strings" - "github.com/Agent-Field/agentfield/sdk/go/ai" "github.com/Agent-Field/codeaf/internal/config" - "github.com/Agent-Field/codeaf/internal/provider" ) -// nodeTreeSpend is one older-road assignment's money and cancellation scope. -// The graph lock protects stopped; SpendGuard protects the shared tally. -type nodeTreeSpend struct { - graph *TaskGraph - root uint64 - guard *SpendGuard - ctx context.Context - halt context.CancelFunc - stopped bool +// nodeTreeLimit is fixed at root admission and survives changes to the +// conversation's books or settings while an interrupted tree is restored. +type nodeTreeLimit struct { + CeilingUSD float64 `json:"ceiling_usd"` + Conversation bool `json:"conversation,omitempty"` + ConversationCapUSD float64 `json:"conversation_cap_usd,omitempty"` } -type nodeTreeSpendRecord struct { - Spent float64 `json:"spent,omitempty"` -} - -type nodeTreeKey struct{} - -// Supervising helpers run on the owner, so their context carries the tree -// without giving the conversation the worker's persistent guard. -func withNodeTree(ctx context.Context, node *TaskNode) context.Context { - if tree := node.graph.spendTree(node); tree != nil { - return context.WithValue(ctx, nodeTreeKey{}, tree) +// nodeTreeBills snapshots live bills before a checkpoint takes the graph lock. +// Atomic direction and app writes need the same unfolded costs as ordinary saves. +func (g *TaskGraph) nodeTreeBills() map[uint64]float64 { + if bashBeltAsked() { + return nil } - return ctx -} - -func (a *Agent) nodeTreeForCall(ctx context.Context) *nodeTreeSpend { - if tree, ok := ctx.Value(nodeTreeKey{}).(*nodeTreeSpend); ok { - return tree + g.mu.Lock() + nodes := make([]*TaskNode, 0, len(g.nodes)) + for _, node := range g.nodes { + if node != nil { + nodes = append(nodes, node) + } } - return a.config.nodeTree -} - -func (g *TaskGraph) nodeSpendRecordLocked(node *TaskNode) *nodeTreeSpendRecord { - if tree := g.nodeTrees[node.id]; tree != nil { - return &nodeTreeSpendRecord{Spent: tree.guard.tally().Total()} + g.mu.Unlock() + bills := make(map[uint64]float64, len(nodes)) + for _, node := range nodes { + bills[node.id] = node.spend() } - return node.savedTreeSpend + return bills } -func (g *TaskGraph) prepareNodeTrees() { +func (g *TaskGraph) prepareNodeTreeLimits() { if bashBeltAsked() { return } g.mu.Lock() var roots []*TaskNode for _, id := range g.order { - if node := g.nodes[id]; node != nil && node.parent == 0 && g.nodeTrees[id] == nil { + if node := g.nodes[id]; node != nil && node.parent == 0 && node.treeLimit == nil { roots = append(roots, node) } } g.mu.Unlock() for _, node := range roots { - g.spendTree(node) - } -} - -func (g *TaskGraph) savedNodeTreeSpendLocked(root uint64) float64 { - if node := g.nodes[root]; node != nil && node.savedTreeSpend != nil { - return node.savedTreeSpend.Spent + g.setNodeTreeLimit(node) } - // Older checkpoints have only node bills, which may include folded - // descendants. Counting those conservatively cannot grant fresh money. - spent := 0.0 - for _, node := range g.nodes { - if node != nil && g.nodeRootLocked(node.id) == root { - spent += node.cost - } - } - return spent } -func (g *TaskGraph) spendTree(node *TaskNode) *nodeTreeSpend { - if bashBeltAsked() || node == nil { - return nil +func (g *TaskGraph) setNodeTreeLimit(node *TaskNode) { + if bashBeltAsked() || node == nil || node.parent != 0 { + return } g.mu.Lock() - root, home := g.nodeRootLocked(node.id), g.home - tree := g.nodeTrees[root] - if tree != nil || home == nil { - g.mu.Unlock() - return tree + home, saved, spent := g.home, node.treeLimit, 0.0 + if saved == nil { + for _, member := range g.nodes { + if member != nil && g.nodeRootLocked(member.id) == node.id { + spent += member.cost + } + } } - spent := g.savedNodeTreeSpendLocked(root) g.mu.Unlock() - // Read settings and the conversation's books outside the graph lock. - taskCap, action := config.CrewTaskSpendCap(home.config.ProfileDir) + if saved != nil || home == nil { + return + } + // Old checkpoints have no ceiling fact. Their existing bills consume the + // task cap, and only the conversation's current remainder is still offered. + taskCap, _ := config.CrewTaskSpendCap(home.config.ProfileDir) conversationCap := home.railCap(0) allowance, conversation := taskTreeCost(taskCap, spent, runCostLeft(conversationCap, home.Usage().CostUSD)) - // The run engine's positive sentinel means no allowance, including for - // an unpriced model whose first call cannot be estimated by SpendGuard. if allowance == math.SmallestNonzeroFloat64 { allowance = 0 } + limit := &nodeTreeLimit{CeilingUSD: taskCap, Conversation: conversation} if conversation { - action = fmt.Sprintf("the conversation reached its %s limit · raise the per conversation limit in /settings", config.CrewTaskMoney(conversationCap)) + limit.CeilingUSD = spent + allowance + limit.ConversationCapUSD = conversationCap } - ctx, halt := context.WithCancel(context.Background()) - candidate := &nodeTreeSpend{graph: g, root: root, ctx: ctx, halt: halt, - guard: &SpendGuard{Price: config.CrewCallPriceAt(home.config.ProfileDir), TaskCap: spent + allowance, TaskAction: action, Task: &SpendTask{spent: spent}}} g.mu.Lock() - if g.nodeTrees == nil { - g.nodeTrees = make(map[uint64]*nodeTreeSpend) - } - if tree = g.nodeTrees[root]; tree == nil { - tree = candidate - g.nodeTrees[root] = tree + if node.treeLimit == nil { + node.treeLimit = limit } g.mu.Unlock() - if tree != candidate { - halt() - } - if tree.guard.tally().Total() >= tree.guard.TaskCap { - tree.stop() - } - return tree } -// nodeTreeStopLocked keeps the frontier and the landing on the same ending. -func (g *TaskGraph) nodeTreeStopLocked(node *TaskNode) string { - tree := g.nodeTrees[g.nodeRootLocked(node.id)] - if tree == nil || !tree.stopped { +// treeSpendLimit reads the same bills as the per-node boundary. Collect under +// the graph lock, then release it before spend takes that lock for each node. +func (n *TaskNode) treeSpendLimit() string { + if bashBeltAsked() { return "" } - if node.ending == "" { - node.ending = TaskEndingCostLimit - } - return tree.guard.TaskAction -} - -func (g *TaskGraph) nodeReadinessLocked(node *TaskNode) (bool, string) { - if limit := g.nodeTreeStopLocked(node); limit != "" { - return false, limit - } - return g.readinessLocked(node) -} - -func (g *TaskGraph) nodeTreeLandingLocked(node *TaskNode, state TaskState) TaskState { - if limit := g.nodeTreeStopLocked(node); limit != "" { - if !strings.Contains(node.report, limit) { - node.landLocked(withReport(node.landed, limit)) - } - return TaskFailed - } - return state -} - -func (tree *nodeTreeSpend) stop() { - g := tree.graph + g := n.graph g.mu.Lock() - if tree.stopped { - g.mu.Unlock() - return - } - tree.stopped = true - var ids []uint64 - for _, id := range g.order { - node := g.nodes[id] - if node == nil || node.state.settled() || g.nodeRootLocked(id) != tree.root { - continue + root := g.nodeRootLocked(n.id) + var limit *nodeTreeLimit + if node := g.nodes[root]; node != nil { + limit = node.treeLimit + } + var nodes []*TaskNode + if limit != nil { + for _, node := range g.nodes { + if node != nil && g.nodeRootLocked(node.id) == root { + nodes = append(nodes, node) + } } - if node.ending == "" { - node.ending = TaskEndingCostLimit - } - ids = append(ids, id) } g.mu.Unlock() - // Mark the whole scope before any stop can turn the frontier again. - for _, id := range ids { - _, _ = g.stopFor(id, tree.guard.TaskAction) + if limit == nil { + return "" } - tree.halt() -} - -func (tree *nodeTreeSpend) book(ctx context.Context, model string, response *ai.Response, held float64) { - tree.guard.after(ctx, model, response, held) - tree.graph.checkpoint() - if tree.guard.TaskCap > 0 && tree.guard.tally().Total() >= tree.guard.TaskCap { - tree.stop() + spent := 0.0 + for _, node := range nodes { + spent += node.spend() } -} - -// nodeTreeCompleter covers turns and helpers alike, including discarded paid -// attempts. The existing crew and seat guards keep their independent bounds. -type nodeTreeCompleter struct { - tree *nodeTreeSpend - model string - next Completer -} - -func (c nodeTreeCompleter) CompleteWithMessages(ctx context.Context, messages []ai.Message, options ...ai.Option) (*ai.Response, error) { - c.tree.graph.mu.Lock() - stopped := c.tree.stopped - c.tree.graph.mu.Unlock() - if stopped { - return nil, ErrSpendStopped{Action: c.tree.guard.TaskAction, TaskLimit: true} - } - held, err := c.tree.guard.before(ctx, c.model, messages, options) - if err != nil { - var stopped ErrSpendStopped - if errors.As(err, &stopped) && stopped.TaskLimit { - c.tree.stop() - } - return nil, err - } - callCtx, release := context.WithCancel(ctx) - defer release() - unhook := context.AfterFunc(c.tree.ctx, release) - defer unhook() - callCtx = provider.WithDiscardedUsage(callCtx, func(_ string, _ string, response *ai.Response) { - c.tree.book(ctx, c.model, response, 0) - }) - response, err := c.next.CompleteWithMessages(callCtx, messages, options...) - c.tree.book(ctx, c.model, response, held) - return response, err + if spent < limit.CeilingUSD { + return "" + } + if limit.Conversation { + return fmt.Sprintf("the conversation reached its %s limit · raise the per conversation limit in /settings", config.CrewTaskMoney(limit.ConversationCapUSD)) + } + return config.CrewTaskCapAction(limit.CeilingUSD) } diff --git a/internal/session/task_node_titles.go b/internal/session/task_node_titles.go index beaa7d1198..9a88752dd6 100644 --- a/internal/session/task_node_titles.go +++ b/internal/session/task_node_titles.go @@ -24,9 +24,6 @@ func (g *TaskGraph) nodeGrowthRefusalLocked(parent uint64, titles []string) stri return "" } root := g.nodeRootLocked(parent) - if tree := g.nodeTrees[root]; tree != nil && tree.stopped { - return tree.guard.TaskAction - } named := g.nodeTreeTitlesLocked(root) batch := make(map[string]bool, len(titles)) for _, title := range titles { diff --git a/internal/session/task_run.go b/internal/session/task_run.go index 433687fab0..6482c0a0bd 100644 --- a/internal/session/task_run.go +++ b/internal/session/task_run.go @@ -353,8 +353,8 @@ type TaskNode struct { admit sync.Mutex spec taskSpec - // The root retains its shared spend across a checkpoint restore. - savedTreeSpend *nodeTreeSpendRecord + // Only the root holds the tree's fixed admission ceiling. + treeLimit *nodeTreeLimit // kind is what sort of node this is ([TaskKind]), settled at admission from // the spec and never touched again. It is a FIELD rather than a question put // to the spec because a node restored from a checkpoint has no spec worth @@ -1140,8 +1140,6 @@ type TaskGraph struct { claims map[uint64]int // Pending titles stay reserved until their own proposal lands or withdraws. childTitles map[uint64][]string - // Each older-road root owns one allowance shared by all its descendants. - nodeTrees map[uint64]*nodeTreeSpend // run executes one node to completion and calls [TaskGraph.complete] when // it lands. It is a field rather than a method so the graph can be exercised @@ -1480,7 +1478,7 @@ func (g *TaskGraph) admit(id uint64, spec taskSpec) TaskState { // goes back (see [TaskGraph.claimChild]). g.releaseChildLocked(spec.parent, spec.title) g.mu.Unlock() - g.spendTree(node) + g.setNodeTreeLimit(node) // ADMISSION IS A TRANSITION, and it is checkpointed before the frontier turns // rather than after: a process killed between these two lines still resumes @@ -1619,7 +1617,7 @@ func (g *TaskGraph) children(parent uint64) []*TaskNode { // call when nothing can move: the cost of a pass with nothing to do is one lock // and a walk of the order slice. func (g *TaskGraph) runFrontier() { - g.prepareNodeTrees() + g.prepareNodeTreeLimits() // The machine is READ before the lock and at most once a pass: a handful of // small file reads behind a one-second cache, and the graph's lock is held // by everything that announces a node — no reading of /proc belongs under @@ -1649,7 +1647,7 @@ func (g *TaskGraph) runFrontier() { if !queuedWithMemoryContext(node, orders) { continue } - ready, blocked := g.nodeReadinessLocked(node) + ready, blocked := g.readinessLocked(node) if blocked != "" { node.setStateLocked(TaskFailed) node.ended = time.Now() @@ -2187,7 +2185,6 @@ func shareReports(reports []string, pot int) []string { // state transition and the slot being handed back. func (g *TaskGraph) complete(node *TaskNode, state TaskState) { g.mu.Lock() - state = g.nodeTreeLandingLocked(node, state) node.setStateLocked(state) node.app.Version++ node.ended = time.Now() @@ -5270,7 +5267,6 @@ func (a *Agent) runTaskNode(node *TaskNode) { // whether or not this line has been reached yet. Time is checked only // between completed turns below. ctx, cancel := node.runContext() - ctx = withNodeTree(ctx, node) // AND THE NODE IS TAKEN UP HERE, OR NOT RUN AT ALL. Close and a person's stop // both land in the window between the admission and this line, and the whole @@ -7985,7 +7981,6 @@ func (a *Agent) newTaskAgentOn(ctx context.Context, dir string, node *TaskNode, quickItems: node.quickDoor(), taskControl: node, taskCapUSD: taskCapUSD, - nodeTree: node.graph.spendTree(node), // AND THE DESIGN THREAD'S ONE EXTRA HAND, wired here for roomThread's // reason: a belt is assembled once, when the agent is constructed // (agent.go), so a door handed over after this call would be a verb the diff --git a/internal/session/task_store.go b/internal/session/task_store.go index 595f198243..f78881bc80 100644 --- a/internal/session/task_store.go +++ b/internal/session/task_store.go @@ -431,12 +431,12 @@ type taskRecord struct { // carried a bill, which resumes as zero — the same thing an unpriced model // leaves behind, and the same thing every surface here already draws as // nothing rather than as "$0.00". - CostUSD float64 `json:"costUsd,omitempty"` - TreeSpend *nodeTreeSpendRecord `json:"tree_spend,omitempty"` - Input int `json:"input,omitempty"` - Output int `json:"output,omitempty"` - CacheRead int `json:"cacheRead,omitempty"` - CacheWrite int `json:"cacheWrite,omitempty"` + CostUSD float64 `json:"costUsd,omitempty"` + TreeLimit *nodeTreeLimit `json:"tree_limit,omitempty"` + Input int `json:"input,omitempty"` + Output int `json:"output,omitempty"` + CacheRead int `json:"cacheRead,omitempty"` + CacheWrite int `json:"cacheWrite,omitempty"` // Noted says this node's completion note has been handed to the steering // lane. It is what stops a resumed session re-announcing work the transcript @@ -1123,6 +1123,7 @@ func (g *TaskGraph) admitWritten(take func() directionHeard, drop func(direction } store.mu.Lock() defer store.mu.Unlock() + bills := g.nodeTreeBills() g.mu.Lock() defer g.mu.Unlock() heard := take() @@ -1131,7 +1132,7 @@ func (g *TaskGraph) admitWritten(take func() directionHeard, drop func(direction if !heard.fresh() { return heard, nil } - if err := store.writeLocked(g.documentLocked()); err != nil { + if err := store.writeLocked(g.documentLocked(bills)); err != nil { drop(heard) return heard, err } @@ -1157,31 +1158,38 @@ func (g *TaskGraph) dropWritten(drop func()) error { } store.mu.Lock() defer store.mu.Unlock() + bills := g.nodeTreeBills() g.mu.Lock() defer g.mu.Unlock() drop() - return store.writeLocked(g.documentLocked()) + return store.writeLocked(g.documentLocked(bills)) } // document is the graph as the file sees it, in admission order — the order that // makes the frontier deterministic, and the order a person reads the file in. func (g *TaskGraph) document() taskDocument { + bills := g.nodeTreeBills() g.mu.Lock() defer g.mu.Unlock() - return g.documentLocked() + return g.documentLocked(bills) } // documentLocked is the same reading with the graph's lock already held, for the // admission that publishes and writes under one hold of it // ([TaskGraph.admitWritten]). -func (g *TaskGraph) documentLocked() taskDocument { +func (g *TaskGraph) documentLocked(bills ...map[uint64]float64) taskDocument { document := taskDocument{Type: taskDocumentType, Version: taskFileVersion, Seq: g.seq} for _, id := range g.order { node := g.nodes[id] if node == nil || g.deleted[id] { continue } - document.Nodes = append(document.Nodes, node.recordLocked()) + record := node.recordLocked() + if len(bills) > 0 { + // A fold after the live snapshot may already have raised the frozen bill. + record.CostUSD = max(record.CostUSD, bills[0][id]) + } + document.Nodes = append(document.Nodes, record) } // AND THE ADAPTIVE RUNS' ROWS GO DOWN IN THE SAME DOCUMENT, written by the // same store on the same beats. There is no second writer of tasks.json: a @@ -1445,7 +1453,7 @@ func (n *TaskNode) recordLocked() taskRecord { StartedAt: n.started, EndedAt: n.ended, CostUSD: n.cost, - TreeSpend: n.graph.nodeSpendRecordLocked(n), + TreeLimit: n.treeLimit, Input: n.input, Output: n.output, CacheRead: n.cacheRead, @@ -2388,7 +2396,7 @@ func restoreNode(graph *TaskGraph, record taskRecord) *TaskNode { started: record.StartedAt, ended: record.EndedAt, cost: record.CostUSD, - savedTreeSpend: record.TreeSpend, + treeLimit: record.TreeLimit, input: record.Input, output: record.Output, cacheRead: record.CacheRead, diff --git a/internal/session/taskname.go b/internal/session/taskname.go index 052676c0af..7378c02220 100644 --- a/internal/session/taskname.go +++ b/internal/session/taskname.go @@ -186,7 +186,7 @@ func (g *TaskGraph) nameNode(node *TaskNode) { if subject == "" { return } - if name := home.taskName(withNodeTree(context.Background(), node), subject); name != "" { + if name := home.taskName(context.Background(), subject); name != "" { g.rename(node, name) } }() From f9b3471314951e928611cb445a39228bbe255f69 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Fri, 6 Nov 2026 09:55:02 -0500 Subject: [PATCH 4/4] test: older-road tree-bound fixtures deliver child reports and give plan specs an id The spend test's graph dropped every report, so the root never finished folding its parts; it now delivers child reports as production does. The work-identity test's plan write lacked an id, so plandb refused it before comparing titles. No production change. Co-Authored-By: Claude Opus 5.5 --- internal/session/task_node_bounds_test.go | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/internal/session/task_node_bounds_test.go b/internal/session/task_node_bounds_test.go index 71e568e704..340a7f2f0f 100644 --- a/internal/session/task_node_bounds_test.go +++ b/internal/session/task_node_bounds_test.go @@ -184,7 +184,13 @@ func TestNodeTreeSpendStopsEveryNodeAndNamesTheBindingLimit(t *testing.T) { a.usage.CostUSD = .25 } g := a.graph() - g.report = nil + // Deliver child reports so the root can finish folding its parts. + // Root reports stay quiet to keep conversation wakes off the script. + g.report = func(node *TaskNode) { + if node.parent != 0 { + a.reportTaskNode(node) + } + } g.store = newTaskStore(filepath.Join(t.TempDir(), "tasks.json")) root := nodeTreeQuick(t, a, 0, "Whole assignment") nodeTreeAwait(t, rootEntered) @@ -393,7 +399,7 @@ func TestNodeTreeUsesThePlanStoresWorkIdentity(t *testing.T) { t.Fatal(err) } defer store.Close() - if _, err := store.AddManyBounded([]plandb.TaskSpec{{Title: title}}); err == nil || !strings.Contains(err.Error(), "already has a task named") { + if _, err := store.AddManyBounded([]plandb.TaskSpec{{ID: "variant", Title: title}}); err == nil || !strings.Contains(err.Error(), "already has a task named") { t.Fatalf("plan admission changed its work identity: %v", err) } n := newNest(t, nil, nil)