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.
diff --git a/internal/manual/chat/tasks.md b/internal/manual/chat/tasks.md
index 59c0b97f87..59d9458b58 100644
--- a/internal/manual/chat/tasks.md
+++ b/internal/manual/chat/tasks.md
@@ -2322,15 +2322,18 @@ 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
-ceiling ends workers in flight and starts no more. When the per-task limit binds, the
+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`.
 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 +4310,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/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/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/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_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_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_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_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..340a7f2f0f
--- /dev/null
+++ b/internal/session/task_node_bounds_test.go
@@ -0,0 +1,549 @@
+package session
+
+import (
+	"context"
+	"errors"
+	"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()
+}
+
+// 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) {
+			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 nodeTreeReadResponse("root-read", .5), nil
+				},
+				func(ctx context.Context, _ []ai.Message) (*ai.Response, error) {
+					close(rootEntered)
+					select {
+					case <-rootHold.Done():
+						return nodeTreeReadResponse("root-read-again", 0), nil
+					case <-ctx.Done():
+						return nil, ctx.Err()
+					}
+				},
+				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()
+					}
+				},
+				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}
+				}
+			})
+			if err := config.SetCrewTaskCap(a.config.ProfileDir, "1.5"); err != nil {
+				t.Fatal(err)
+			}
+			if conversation {
+				a.usage.CostUSD = .25
+			}
+			g := a.graph()
+			// 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)
+			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")
+			}
+			releaseChild()
+			waitDoneNode(t, part)
+			waitDoneNode(t, queued)
+			if queued.notice().StartedAt.IsZero() {
+				t.Fatal("queued node did not reach its first request boundary")
+			}
+			// Reaching the limit leaves the already admitted provider call alone.
+			select {
+			case <-root.done:
+				t.Fatal("the tree limit cancelled a request in flight")
+			default:
+			}
+			if queued.spend() != 0 || completer.requests() != 3 {
+				t.Fatalf("queued node bought a request: spend=%v, calls=%d", queued.spend(), completer.requests())
+			}
+			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)
+			}
+			releaseRoot()
+			waitDoneNode(t, root)
+			want := config.CrewTaskCapAction(1.5)
+			if conversation {
+				want = "the conversation reached its $1.25 limit · raise the per conversation limit in /settings"
+			}
+			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 calls := completer.requests(); calls != 4 {
+				t.Fatalf("exhausted tree reached the provider again: calls=%d, want 4 including the other tree", calls)
+			}
+			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: 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()
+		c.TaskAudit = false
+	})
+	g := a.graph()
+	g.report = 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, model: "test/model"})
+	node := g.node(id)
+	waitDoneNode(t, node)
+	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: 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, "5"); err != nil {
+		t.Fatal(err)
+	}
+	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.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"})
+	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)
+		}
+	}
+	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)
+	}
+	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.report = nil
+	restored.rehydrate(document, b.config.Workspace, TaskSettleAsk)
+	restored.runFrontier()
+	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)
+	}
+	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())
+	}
+}
+
+// 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{{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)
+			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)
+			}
+		})
+	}
+}
+
+// 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{}
+	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())
+	}
+}
+
+// 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.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)
+	}
+	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")
+		}
+	}
+	if root.endingSet() != TaskEndingCostLimit || completer.requests() != 1 {
+		t.Fatalf("blocked checkpoint postponed the limit: ending=%q, calls=%d", root.endingSet(), completer.requests())
+	}
+	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")
+	}
+}
+
+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)
+	}
+	g := a.graph()
+	spec.parent = parent
+	if parent != 0 {
+		spec.depth = g.node(parent).depth + 1
+	}
+	if refusal := g.claimChild(parent, title); refusal != "" {
+		t.Fatal(refusal)
+	}
+	id := g.reserve()
+	g.admit(id, spec)
+	return g.node(id)
+}
+
+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
new file mode 100644
index 0000000000..df7cd2ed32
--- /dev/null
+++ b/internal/session/task_node_spend.go
@@ -0,0 +1,129 @@
+package session
+
+import (
+	"fmt"
+	"math"
+
+	"github.com/Agent-Field/codeaf/internal/config"
+)
+
+// 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"`
+}
+
+// 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
+	}
+	g.mu.Lock()
+	nodes := make([]*TaskNode, 0, len(g.nodes))
+	for _, node := range g.nodes {
+		if node != nil {
+			nodes = append(nodes, node)
+		}
+	}
+	g.mu.Unlock()
+	bills := make(map[uint64]float64, len(nodes))
+	for _, node := range nodes {
+		bills[node.id] = node.spend()
+	}
+	return bills
+}
+
+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 && node.treeLimit == nil {
+			roots = append(roots, node)
+		}
+	}
+	g.mu.Unlock()
+	for _, node := range roots {
+		g.setNodeTreeLimit(node)
+	}
+}
+
+func (g *TaskGraph) setNodeTreeLimit(node *TaskNode) {
+	if bashBeltAsked() || node == nil || node.parent != 0 {
+		return
+	}
+	g.mu.Lock()
+	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
+			}
+		}
+	}
+	g.mu.Unlock()
+	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))
+	if allowance == math.SmallestNonzeroFloat64 {
+		allowance = 0
+	}
+	limit := &nodeTreeLimit{CeilingUSD: taskCap, Conversation: conversation}
+	if conversation {
+		limit.CeilingUSD = spent + allowance
+		limit.ConversationCapUSD = conversationCap
+	}
+	g.mu.Lock()
+	if node.treeLimit == nil {
+		node.treeLimit = limit
+	}
+	g.mu.Unlock()
+}
+
+// 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 ""
+	}
+	g := n.graph
+	g.mu.Lock()
+	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)
+			}
+		}
+	}
+	g.mu.Unlock()
+	if limit == nil {
+		return ""
+	}
+	spent := 0.0
+	for _, node := range nodes {
+		spent += node.spend()
+	}
+	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
new file mode 100644
index 0000000000..9a88752dd6
--- /dev/null
+++ b/internal/session/task_node_titles.go
@@ -0,0 +1,83 @@
+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)
+	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..6482c0a0bd 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
+	// 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
@@ -1136,6 +1138,8 @@ 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
 
 	// 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 +1463,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 +1476,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.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
@@ -1504,7 +1509,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 +1517,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 +1557,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 +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.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
diff --git a/internal/session/task_store.go b/internal/session/task_store.go
index 2ca74dce61..f78881bc80 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"`
+	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
@@ -1122,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()
@@ -1130,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
 	}
@@ -1156,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
@@ -1444,6 +1453,7 @@ func (n *TaskNode) recordLocked() taskRecord {
 		StartedAt:      n.started,
 		EndedAt:        n.ended,
 		CostUSD:        n.cost,
+		TreeLimit:      n.treeLimit,
 		Input:          n.input,
 		Output:         n.output,
 		CacheRead:      n.cacheRead,
@@ -2386,6 +2396,7 @@ func restoreNode(graph *TaskGraph, record taskRecord) *TaskNode {
 		started:        record.StartedAt,
 		ended:          record.EndedAt,
 		cost:           record.CostUSD,
+		treeLimit:      record.TreeLimit,
 		input:          record.Input,
 		output:         record.Output,
 		cacheRead:      record.CacheRead,