diff --git a/docs/subagents.md b/docs/subagents.md
index e39064f..6f4de99 100644
--- a/docs/subagents.md
+++ b/docs/subagents.md
@@ -33,13 +33,15 @@ it is scoped in the event store. Closely related helpers share an investigation.
 Simple requests stay direct, and accepted plans retain their existing task
 contract. Invalid decisions get two repair attempts before ordinary continuation.
 
-Each worker has a separate provider instance, session, messages, execution
-context, compaction state, capability token and usage ledger. Agents share the
-workspace and scoped SQLite services. Structured writes require ownership of
-exact assigned files: disjoint writes can overlap, overlapping ownership waits,
-and opaque commands/repository mutations serialize against active workspace
-reads and writes. The same coordination guards main-agent actions. Capability
-labels and ownership checks are application policy, not an OS sandbox.
+Each worker runs through the canonical AgentEngine with a separate provider
+instance, session, messages, execution context, compaction state, capability
+token and usage ledger. File reads, searches, listings and writes are limited to
+the assigned paths. Reviewers may additionally read declared artifacts so they
+can inspect a result outside the worker's write scope. Disjoint writes can
+overlap, overlapping ownership waits, and opaque commands/repository mutations
+serialize against active workspace reads and writes. The same coordination
+guards main-agent actions. Capability labels and ownership checks are
+application policy, not an OS sandbox.
 
 ## Assignment and result contracts
 
@@ -62,11 +64,17 @@ and arrays `findings`, `evidence`, `artifacts`, `checks`, `uncertainties`,
 visibly. Model reports are kept separate from runtime-owned execution receipts
 and provider usage.
 
-Every completed or partial substantive report enters a fresh read-only review context.
-Batches use reciprocal/ring peer identities; a singleton gets a reviewer. The
-reviewer receives the assignment, submitted result and receipt references, and
-must independently inspect evidence. A `verified` report without a successful
-inspection tool receipt and supporting evidence becomes `unverifiable`.
+Mutating, broad-scope, and explicitly high-risk assignments enter a fresh
+read-only review context. Narrow read-only assignments skip the extra provider
+call; the parent verifies that the completed report cites successful scoped
+read receipts. Reviewed batches use reciprocal/ring peer identities; a
+singleton gets a reviewer. The reviewer receives the assignment, submitted
+result and receipt references, and must independently inspect evidence. A
+`verified` report without a successful inspection tool receipt and supporting
+evidence becomes `unverifiable`.
+Reviewer access to declared file artifacts is limited to the worker's assigned
+paths and paths covered by successful worker write receipts. A report cannot
+grant review access to arbitrary workspace files by naming them as artifacts.
 Review verdicts are `verified`, `changes_requested`, or `unverifiable`.
 Reviewers must independently read every declared artifact. A missed read gets
 two bounded inspection reminders in that review context before the runtime
@@ -100,6 +108,13 @@ seconds). Reasoning reviews exceeded the earlier 90-second deadline in live
 acceptance; those failures are retained in the acceptance record.
 Restart marks unfinished jobs interrupted without replaying mutations.
 
+Delegation reserves $0.10 per child and an additional $0.25 when independent
+review is required, with a $1 aggregate limit per parent turn. Assignments that
+cannot reserve this amount are rejected before they start. Known usage settles
+against the reservation; unknown usage retains the full reservation for that
+turn. Cost accounting uses the provider usage ledger, so calls with unknown
+pricing or usage cannot free reserved capacity.
+
 Assignments, messages, receipts, state snapshots, results and reviews persist
 with actor, project, parent-chat and run identifiers. Names are unique within
 the parent chat, randomly allocated without replacement from Evan, Morning,
diff --git a/openkyrozen/agent/action_rounds.py b/openkyrozen/agent/action_rounds.py
index ea6422a..800a3dd 100644
--- a/openkyrozen/agent/action_rounds.py
+++ b/openkyrozen/agent/action_rounds.py
@@ -26,7 +26,9 @@ def record_initial_call(tc, receipt):
     turn.tool_results_text = "\n".join(all_tool_result_lines)
 
     # Unified multi-step loop - always runs, can handle early errors
-    round_limit = self.MAX_STEPS_PER_TURN
+    round_limit = getattr(self.execution_context, "max_steps", None)
+    if round_limit is None:
+        round_limit = self.MAX_STEPS_PER_TURN
     round_count = 0
     turn.final_answer: str | None = None
     turn.pending_interaction_reply: str | None = None
diff --git a/openkyrozen/agent/completion.py b/openkyrozen/agent/completion.py
index 7c06442..aba3dea 100644
--- a/openkyrozen/agent/completion.py
+++ b/openkyrozen/agent/completion.py
@@ -88,7 +88,8 @@ def _complete_turn(self, turn):
         )
         turn.final_answer = f"{turn.final_answer.rstrip()}\n\nTool status: incomplete; failed operations were not verified.\n{failures}"
     plan = getattr(self.tasks, "working_plan", None)
-    if (turn.interaction_mode == "agent" and getattr(turn, "complexity", None) == "complex"
+    if (not self.execution_context.child_run_id
+            and turn.interaction_mode == "agent" and getattr(turn, "complexity", None) == "complex"
             and plan is not None and plan.status in {"active", "completed", "cancelled"}):
         plan_tasks = [task for task in self.tasks.tasks if task.get("id") in plan.task_ids]
         succeeded = [task for task in plan_tasks if is_complete(task)]
diff --git a/openkyrozen/agent/delegation.py b/openkyrozen/agent/delegation.py
index 2f904c5..9cb51ef 100644
--- a/openkyrozen/agent/delegation.py
+++ b/openkyrozen/agent/delegation.py
@@ -104,6 +104,7 @@ def __init__(self):
     @contextmanager
     def acquire(self, path=None, cancelled=lambda: False, owner=None, reading=False, recursive=False):
         with self.condition:
+            overlaps = lambda left, right: Path(left).is_relative_to(Path(right)) or Path(right).is_relative_to(Path(left))
             while (self.exclusive or (path is None and (self.paths or self.readers)) or path in self.paths
                    or (not reading and self.readers.get(path, 0))
                    or (path is None and self.recursive_readers)
@@ -111,7 +112,8 @@ def acquire(self, path=None, cancelled=lambda: False, owner=None, reading=False,
                                            for scope in self.recursive_readers))
                    or (reading and recursive and any(Path(active).is_relative_to(Path(path))
                                                      for active in self.paths))
-                   or (not reading and path in self.owners and self.owners[path] != owner)):
+                   or (path is not None and any(overlaps(path, active) and active_owner != owner
+                                                 for active, active_owner in self.owners.items()))):
                 if cancelled():
                     raise RuntimeError("Sub-agent cancelled")
                 self.condition.wait(0.1)
@@ -145,9 +147,12 @@ def acquire(self, path=None, cancelled=lambda: False, owner=None, reading=False,
 
     @contextmanager
     def claim(self, paths, owner, cancelled):
+        overlaps = lambda left, right: Path(left).is_relative_to(Path(right)) or Path(right).is_relative_to(Path(left))
         with self.condition:
-            while any(
-                path in self.owners or path in self.paths or self.readers.get(path, 0)
+            while self.exclusive or any(
+                any(overlaps(path, active) for active in self.owners)
+                or any(overlaps(path, active) for active in self.paths)
+                or any(count and overlaps(path, active) for active, count in self.readers.items())
                 or any(Path(path).is_relative_to(Path(scope)) for scope in self.recursive_readers)
                 for path in paths
             ):
@@ -167,6 +172,65 @@ def claim(self, paths, owner, cancelled):
                 self.condition.notify_all()
 
 
+class DelegationBudget:
+    """Reserve a bounded share of one parent turn for child and review work."""
+
+    CHILD_RESERVATION = 100_000_000_000
+    REVIEW_RESERVATION = 250_000_000_000
+    TURN_LIMIT = 1_000_000_000_000
+
+    def __init__(self):
+        self._lock = threading.RLock()
+        self._commitments: dict[str, int] = {}
+        self._unknown: set[str] = set()
+        self._spent: dict[str, int] = {}
+
+    @property
+    def remaining_picos(self) -> int:
+        with self._lock:
+            return self.TURN_LIMIT - sum(self._commitments.values())
+
+    def reserve(self, run_id: str, *, review: bool) -> None:
+        amount = self.CHILD_RESERVATION + (self.REVIEW_RESERVATION if review else 0)
+        with self._lock:
+            if run_id in self._commitments or amount > self.TURN_LIMIT - sum(self._commitments.values()):
+                raise ValueError("Delegation budget is unavailable for this assignment")
+            self._commitments[run_id] = amount
+
+    def settle(self, run_id: str, *, cost_picos: int | None) -> None:
+        with self._lock:
+            if run_id not in self._commitments:
+                return
+            if cost_picos is None:
+                self._unknown.add(run_id)
+                return
+            self._commitments[run_id] = max(0, int(cost_picos))
+
+    def record_cost(self, run_id: str, cost_picos: int | None) -> None:
+        with self._lock:
+            if run_id not in self._commitments:
+                return
+            if cost_picos is None:
+                self._unknown.add(run_id)
+            else:
+                self._spent[run_id] = self._spent.get(run_id, 0) + max(0, int(cost_picos))
+
+    def can_continue(self, run_id: str, *, cost_picos: int | None) -> bool:
+        with self._lock:
+            if run_id not in self._commitments:
+                return False
+            # Unknown pricing consumes the reservation for the turn but cannot meter individual calls.
+            if run_id in self._unknown or cost_picos is None:
+                return True
+            return self._spent.get(run_id, int(cost_picos)) < self._commitments[run_id]
+
+    def release(self, run_id: str) -> None:
+        with self._lock:
+            self._commitments.pop(run_id, None)
+            self._unknown.discard(run_id)
+            self._spent.pop(run_id, None)
+
+
 class Delegation:
     def __init__(self, memory, invoke, *, profiles, root, concurrency=4, emit=None, access=None, capture_context=copy_context, redact=str, on_verified=None):
         self.memory, self.invoke, self.profiles = memory, invoke, profiles
@@ -181,6 +245,7 @@ def __init__(self, memory, invoke, *, profiles, root, concurrency=4, emit=None,
         self.pool = ThreadPoolExecutor(max_workers=concurrency, thread_name_prefix="kyrozen-agent")
         self.runs, self.messages, self.cancelled, self.futures = {}, {}, {}, {}
         self.contexts = {}
+        self.budgets = {}
         self.names = set()
         self._restore()
 
@@ -279,6 +344,8 @@ def _brief(self, brief):
             raise ValueError("Acceptance and deliverables cannot be empty")
         if brief.get("profile", "researcher") not in self.profiles:
             raise ValueError("Unknown sub-agent profile")
+        if brief.get("risk", "normal") not in {"low", "normal", "high"}:
+            raise ValueError("risk must be low, normal, or high")
         for key in ("provider", "model", "custom_profile", "id"):
             if key in brief and (not isinstance(brief[key], str) or not brief[key].strip()):
                 raise ValueError(key + " must be nonempty text")
@@ -294,9 +361,17 @@ def _brief(self, brief):
         brief["scope"] = list(dict.fromkeys(paths))
         return brief
 
-    def spawn(self, assignments):
+    def _requires_review(self, brief):
+        profile = self.profiles[brief.get("profile", "researcher")]
+        broad_scope = len(brief["scope"]) > 1 or "." in brief["scope"] or any(
+            (self.root / path).is_dir() for path in brief["scope"]
+        )
+        return profile.capabilities != "readonly" or broad_scope or brief.get("risk", "normal") == "high"
+
+    def spawn(self, assignments, *, budget=None):
         if not isinstance(assignments, list) or not assignments:
             raise ValueError("assignments must be a nonempty array")
+        budget = budget or DelegationBudget()
         briefs = [self._brief(item) for item in assignments]
         with self.lock:
             ids = [b.get("id", uuid.uuid4().hex) for b in briefs]
@@ -325,11 +400,22 @@ def spawn(self, assignments):
                 if not ready:
                     raise ValueError("Cyclic dependencies")
                 pending -= ready
+            reserved = []
+            try:
+                for key, brief in zip(ids, briefs):
+                    run_id = mapping[key]
+                    budget.reserve(run_id, review=self._requires_review(brief))
+                    reserved.append(run_id)
+            except Exception:
+                for run_id in reserved:
+                    budget.release(run_id)
+                raise
             runs = []
             for key, brief in zip(ids, briefs):
                 run_id = mapping[key]
                 run = {"run_id": run_id, "assignment_id": key, "name": self._name(),
                        "profile": brief.get("profile", "researcher"), "assignment": brief,
+                       "budget_id": run_id,
                        "dependencies": [mapping.get(dep, dep) for dep in brief["dependencies"]],
                        "status": "queued", "signature": json.dumps([brief["objective"].strip().lower(), sorted(brief["scope"]), brief["context"]]),
                        "parent_session_id": self.memory.session_id, "source_scope_id": self.memory.file_scope_id,
@@ -337,6 +423,7 @@ def spawn(self, assignments):
                 run["icon"] = identicon(run_id)
                 self.runs[run_id] = run
                 self.contexts[run_id] = self.capture_context()
+                self.budgets[run_id] = budget
                 self.cancelled[run_id] = threading.Event()
                 runs.append(run)
             # Peer identities are fixed before any work starts; review contexts remain independent.
@@ -359,6 +446,9 @@ def _schedule(self):
             for run in blocked:
                 run.update(status="blocked", error="Dependency did not pass independent verification")
                 run["review_agent"]["status"] = "blocked"
+                budget = self.budgets.get(run["run_id"])
+                if budget is not None:
+                    budget.settle(run.get("budget_id", run["run_id"]), cost_picos=0)
                 self._publish(run)
         for run_id, run in self.runs.items():
             if run["status"] != "queued" or run_id in self.futures:
@@ -366,7 +456,9 @@ def _schedule(self):
             dependencies = [self.runs[dep]["status"] for dep in run["dependencies"]]
             if all(state == "succeeded" for state in dependencies):
                 paths = self._owned_paths(run)
-                if any(paths & self._owned_paths(self.runs[other]) for other, future in self.futures.items() if not future.done()):
+                if any(any(Path(left).is_relative_to(Path(right)) or Path(right).is_relative_to(Path(left))
+                           for left in paths for right in self._owned_paths(self.runs[other]))
+                       for other, future in self.futures.items() if not future.done()):
                     continue  # Queue overlapping ownership without occupying an execution slot.
                 self.futures[run_id] = self.pool.submit(self.contexts[run_id].copy().run, self._work, run_id)
                 self.futures[run_id].add_done_callback(lambda future: self._job_done())
@@ -381,8 +473,8 @@ def _job_done(self):
             self._schedule()
             self.changed.notify_all()
 
-    def _invoke(self, run, *, review=False, feedback=None):
-        return self.invoke(run, review=review, feedback=feedback, coordinator=self)
+    def _invoke(self, run, *, review=False, feedback=None, report_repair=False):
+        return self.invoke(run, review=review, feedback=feedback, report_repair=report_repair, coordinator=self)
 
     def _work(self, run_id):
         run = self.runs[run_id]
@@ -395,6 +487,11 @@ def _work(self, run_id):
         except Exception as exc:
             with self.lock:
                 run.update(status="cancelled" if self.cancelled[run_id].is_set() else "failed", error=str(exc))
+                budget = self.budgets.get(run_id)
+                if budget is not None:
+                    costs = [item.get("cost_picos") for item in run.get("usage", {}).values()]
+                    budget.settle(run.get("budget_id", run_id),
+                                  cost_picos=sum(costs) if costs and all(value is not None for value in costs) else (0 if not costs else None))
                 self._publish(run)
                 self._schedule()
 
@@ -408,11 +505,22 @@ def _work_owned(self, run_id):
                 self._publish(run)
             for cycle in range(3):
                 self.check_cancelled(run_id)
-                result = self._invoke(run, feedback=run["reviews"][-1] if cycle else None)
-                self.check_cancelled(run_id)
-                worker_report = report(result["result"])
+                feedback = run["reviews"][-1] if cycle else None
+                report_attempts = []
+                for repair in range(3):
+                    result = self._invoke(run, feedback=feedback, report_repair=repair > 0)
+                    report_attempts.append(copy.deepcopy(result))
+                    self.check_cancelled(run_id)
+                    try:
+                        worker_report = report(result["result"])
+                        break
+                    except (ValueError, json.JSONDecodeError) as exc:
+                        if repair == 2:
+                            raise ValueError("Malformed delegated report after two repairs") from exc
+                        feedback = {"instruction": "Return only a valid JSON report with every required field. Do not call tools.",
+                                    "invalid_response": result.get("result", "")[:4000]}
                 with self.lock:
-                    run.setdefault("attempts", []).append(copy.deepcopy(result))
+                    run.setdefault("attempts", []).extend(report_attempts)
                     result["tool_records"] = [receipt for attempt in run["attempts"] for receipt in attempt.get("tool_records", [])]
                     result["evidence"] = [evidence for attempt in run["attempts"] for evidence in attempt.get("evidence", [])]
                     run.update(result=result, report=worker_report,
@@ -424,16 +532,40 @@ def _work_owned(self, run_id):
                     with self.lock:
                         run["status"] = worker_report["status"]
                     break
-                review_result = self._invoke(run, review=True)
-                self.check_cancelled(run_id)
-                verdict = report(review_result["result"], review=True)
-                # A model's agreement is insufficient: the reviewer must actually inspect evidence.
-                checked = any(r.get("success") and r.get("action") in {
-                    "read_file", "read_webpage", "search_web", "git_diff", "git_show", "graph_query"
-                } for r in review_result.get("tool_records", []))
-                if verdict["verdict"] == "verified" and (not checked or not verdict["evidence"]):
-                    verdict.update(verdict="unverifiable", summary="Reviewer supplied no independent inspection evidence")
-                artifact_findings, artifact_checks = self._inspect_artifacts(run, review_result)
+                if not self._requires_review(run["assignment"]):
+                    verified = self._verify_read_only_result(run, worker_report, result)
+                    with self.lock:
+                        run["reviews"].append({
+                            "verdict": "verified" if verified else "unverifiable",
+                            "summary": "Parent validated scoped read receipts" if verified else
+                                "Read-only result did not bind its evidence to successful scoped receipts",
+                            "findings": [] if verified else ["Evidence was not independently bound to tool receipts"],
+                            "evidence": worker_report["evidence"] if verified else [],
+                            "reviewer": "parent-runtime",
+                            "verification": "receipt-bound-read-only",
+                        })
+                        if verified and worker_report["status"] == "completed":
+                            run["parent_receipts"] = self.on_verified(run)
+                            run["status"] = "succeeded"
+                        else:
+                            run.update(status="unverified", error="Read-only result did not pass parent verification")
+                        run["review_agent"]["status"] = "skipped" if verified else "unverifiable"
+                        self._publish(run)
+                    break
+                for inspection_attempt in range(3):
+                    review_result = self._invoke(run, review=True, feedback=feedback if inspection_attempt else None)
+                    self.check_cancelled(run_id)
+                    verdict = report(review_result["result"], review=True)
+                    checked = any(r.get("success") and r.get("action") in {
+                        "read_file",
+                    } for r in review_result.get("tool_records", []))
+                    if verdict["verdict"] == "verified" and (not checked or not verdict["evidence"]):
+                        verdict.update(verdict="unverifiable", summary="Reviewer supplied no independent inspection evidence")
+                    artifact_findings, artifact_checks = self._inspect_artifacts(run, review_result)
+                    if not artifact_findings or inspection_attempt == 2:
+                        break
+                    feedback = {"findings": artifact_findings,
+                                "instruction": "Verification requires independent tool inspection. Read every declared artifact and return a JSON review."}
                 if artifact_findings and verdict["verdict"] == "verified":
                     verdict.update(verdict="changes_requested", summary="Runtime artifact checks did not pass")
                     verdict["findings"].extend(artifact_findings)
@@ -460,18 +592,45 @@ def _work_owned(self, run_id):
             with self.lock:
                 run = self.runs[run_id]
                 from .types import ProviderUnavailableError
+                error = (f"Malformed delegated report after two repairs: {exc}"
+                         if isinstance(exc, json.JSONDecodeError) else f"{type(exc).__name__}: {exc}")
                 run.update(status="cancelled" if self.cancelled[run_id].is_set() else
                            "blocked" if isinstance(exc, ProviderUnavailableError) else "failed",
-                           error=f"{type(exc).__name__}: {exc}")
+                           error=error)
                 if run["review_agent"]["status"] == "reviewing":
                     run["review_agent"].update(status=run["status"], error=run["error"])
         finally:
             with self.lock:
                 if self.cancelled[run_id].is_set():
                     run["status"] = "cancelled"
+                budget = self.budgets.get(run_id)
+                if budget is not None:
+                    costs = [item.get("cost_picos") for item in run.get("usage", {}).values()]
+                    total_cost = sum(costs) if costs and all(value is not None for value in costs) else None
+                    budget.settle(run.get("budget_id", run_id), cost_picos=total_cost)
                 self._publish(run)
                 self._schedule()
 
+    def _verify_read_only_result(self, run, worker_report, result):
+        if worker_report["status"] != "completed" or worker_report["artifacts"]:
+            return False
+        receipts = [item for item in result.get("tool_records", []) if item.get("success")]
+        scopes = [(self.root / value).resolve() for value in run["assignment"]["scope"]]
+        def in_scope(raw):
+            try:
+                path = Path(raw)
+                path = (path if path.is_absolute() else self.root / path).resolve()
+                return any(path == scope or path.is_relative_to(scope) for scope in scopes)
+            except (OSError, TypeError, ValueError):
+                return False
+        inspected = {item.get("receipt_id") for item in receipts if item.get("action") == "read_file"
+                     and in_scope((item.get("args") or {}).get("path") if isinstance(item.get("args"), dict) else
+                                  str(item.get("args", "")).split("|", 1)[0])}
+        evidence = worker_report["evidence"]
+        if not inspected or not evidence or not worker_report.get("checks"):
+            return False
+        return all(isinstance(item, dict) and item.get("receipt_id") in inspected for item in evidence)
+
     def _inspect_artifacts(self, run, review_result):
         findings, checks = [], []
         receipts = [item for item in review_result.get("tool_records", []) if item.get("success")]
@@ -534,6 +693,21 @@ def detail(self, run_id):
             review_messages = {review["session_id"]: [event["payload"]["message"] for event in message_events
                 if event["payload"].get("run_id") == review["session_id"]]
                 for review in self.runs[run_id]["reviews"] if review.get("session_id")}
+            child_sessions = {run_id, self.runs[run_id].get("budget_id", run_id)}
+            child_sessions.update(item.get("session_id") for item in self.runs[run_id].get("attempts", [])
+                                  if item.get("session_id"))
+            child_sessions.update(review.get("session_id") for review in self.runs[run_id]["reviews"]
+                                  if review.get("session_id"))
+            for child_session in child_sessions:
+                from .conversation import ConversationHistory
+                child_messages = ConversationHistory(
+                    self.memory.store, user_id=self.memory.user_id,
+                    workspace_id=self.memory.file_scope_id, session_id=child_session,
+                ).display_messages()
+                if child_session in {run_id, self.runs[run_id].get("budget_id", run_id)}:
+                    messages = child_messages
+                else:
+                    review_messages[child_session] = child_messages
             # ponytail: scan scoped events; add an indexed projection when long chats justify it.
             receipts = [event["payload"] for event in reversed(self.memory.store.list_events(
                 "subagent.receipt", limit=None, user_id=self.memory.user_id,
@@ -568,8 +742,9 @@ def cancel(self, run_id):
                 self._schedule()
             return copy.deepcopy(self.runs[run_id])
 
-    def send(self, run_id, assignment):
+    def send(self, run_id, assignment, *, budget=None):
         brief = self._brief(assignment)
+        budget = budget or DelegationBudget()
         with self.lock:
             if run_id not in self.runs or self.runs[run_id]["status"] not in TERMINAL:
                 raise ValueError("Reuse requires a finished agent in this chat")
@@ -586,15 +761,20 @@ def send(self, run_id, assignment):
             if any(assignment_key(other["assignment"]) == assignment_key(brief) and other["status"] not in TERMINAL
                    for key, other in self.runs.items() if key != run_id):
                 raise ValueError("Duplicate active assignment")
+            budget_id = f"{run_id}_followup_{uuid.uuid4().hex}"
+            budget.reserve(budget_id, review=self._requires_review(brief))
             run.update(assignment=brief, profile=brief.get("profile", run["profile"]),
+                       budget_id=budget_id,
                        assignment_id=brief.get("id", uuid.uuid4().hex),
                        status="queued", dependencies=[], corrections=0, reviews=[], attempts=[], parent_receipts=[])
+            run["usage"] = {}
             for field in ("report", "result", "progress"):
                 run.pop(field, None)
             run.pop("error", None)
             run["signature"] = signature
             run["review_agent"]["status"] = "waiting"
             self.contexts[run_id] = self.capture_context()
+            self.budgets[run_id] = budget
             if run_id not in self.messages:
                 self.messages[run_id] = self.detail(run_id)["messages"]
             self.cancelled[run_id].clear()
@@ -606,4 +786,4 @@ def send(self, run_id, assignment):
     def close(self):
         for run in self.list():
             self.cancel(run["run_id"])
-        self.pool.shutdown(wait=False, cancel_futures=True)
+        self.pool.shutdown(wait=True, cancel_futures=True)
diff --git a/openkyrozen/agent/delegation_runtime.py b/openkyrozen/agent/delegation_runtime.py
index 1ffe7fa..bb7fdf2 100644
--- a/openkyrozen/agent/delegation_runtime.py
+++ b/openkyrozen/agent/delegation_runtime.py
@@ -7,6 +7,7 @@
 import os
 import time
 import uuid
+from pathlib import Path
 from contextlib import contextmanager
 from contextvars import copy_context
 from dataclasses import replace
@@ -144,7 +145,7 @@ def spawn_agents(self, args):
             task_id = assignment.get("task_id") if isinstance(assignment, dict) else None
             if task_id is not None and not self._valid_receipt_task_binding(self.current_session, task_id):
                 return tool_failure("invalid_arguments", "Error: assignment task_id must identify a pending task in the active Working Plan.")
-    data = {"agents": self.delegation().spawn(assignments)}
+    data = {"agents": self.delegation().spawn(assignments, budget=self.execution_context.delegation_budget)}
     return tool_json_result(data, text=json.dumps(data, ensure_ascii=False))
 
 
@@ -158,7 +159,8 @@ def send_subagent(self, args):
     task_id = assignment.get("task_id") if isinstance(assignment, dict) else None
     if task_id is not None and not self._valid_receipt_task_binding(self.current_session, task_id):
         return tool_failure("invalid_arguments", "Error: assignment task_id must identify a pending task in the active Working Plan.")
-    data = self.delegation().send(value.get("run_id"), assignment)
+    data = self.delegation().send(value.get("run_id"), assignment,
+                                  budget=self.execution_context.delegation_budget)
     return tool_json_result(data, text=json.dumps(data, ensure_ascii=False))
 
 
@@ -195,7 +197,7 @@ def cancel_subagent(self, args):
     return tool_json_result(data, text=json.dumps(data, ensure_ascii=False))
 
 
-def _invoke_delegated(self, run, *, review, feedback, coordinator):
+def _invoke_delegated(self, run, *, review, feedback, coordinator, report_repair=False):
     parent = self.current_session
     brief = run["assignment"]
     role = "reviewer" if review else run["profile"]
@@ -242,7 +244,7 @@ def _invoke_delegated(self, run, *, review, feedback, coordinator):
             run["provider_model"] = f"{provider_name}:{model}"
         coordinator._publish(run)
     provider = self.get_provider(config)  # No implicit cross-provider fallback.
-    child_id = run["run_id"] + ("_review_" + uuid.uuid4().hex if review else "")
+    child_id = run.get("budget_id", run["run_id"]) + ("_review_" + uuid.uuid4().hex if review else "")
     adapters = self.workspace_factory(root=coordinator.root)
     workspace = WorkspaceState(str(coordinator.root), adapters, parent.workspace.launch_context,
                                parent.workspace.graph, parent.workspace.github)
@@ -254,13 +256,40 @@ def _invoke_delegated(self, run, *, review, feedback, coordinator):
     capabilities &= effective_capabilities(load_agent_config(coordinator.root))
     if review or self._active_interaction_mode.get() in {"ask", "plan"}:
         capabilities &= resolve_capabilities("readonly")
-    tools = allowed_tool_names(adapters.AVAILABLE_TOOLS, ",".join(capabilities)) - set(TOOLS)
+    tools = (allowed_tool_names(adapters.AVAILABLE_TOOLS, ",".join(capabilities)) - set(TOOLS)) if not report_repair else set()
     child._delegation_parent = parent
     child._delegation_task_id = brief.get("task_id") if not review else None
     profile = replace(profile, metadata={**profile.metadata, "structured": True, "review": review,
         "display_name": run["reviewer"] if review else run["name"], "assignment": brief,
         "artifacts": copy.deepcopy(run.get("report", {}).get("artifacts", [])) if review else []})
-    task = ("Independently verify this assignment and submitted result. Read actual evidence with tools. "
+    allowed_artifacts = {(coordinator.root / path).resolve() for path in run["assignment"]["scope"]}
+    for record in run.get("result", {}).get("tool_records", []):
+        if record.get("success") and record.get("action") == "read_file":
+            args = record.get("arguments") or record.get("args") or {}
+            if isinstance(args, str):
+                try: args = json.loads(args)
+                except json.JSONDecodeError: args = {"path": args.split("|", 1)[0]}
+            path = args.get("path") if isinstance(args, dict) else None
+            if path:
+                allowed_artifacts.add((coordinator.root / path).resolve())
+        if record.get("success") and record.get("action") in {"write_file", "edit_file", "apply_patch"}:
+            args = record.get("args") or {}
+            if isinstance(args, str):
+                try: args = json.loads(args)
+                except json.JSONDecodeError: args = {"path": args.split("|", 1)[0]}
+            path = args.get("path") if isinstance(args, dict) else None
+            if path:
+                allowed_artifacts.add((coordinator.root / path).resolve())
+    child._delegation_review_artifacts = [
+        item if isinstance(item, str) else item.get("path", "")
+        for item in profile.metadata["artifacts"]
+        if isinstance(item, str) or isinstance(item, dict) and item.get("path")
+    ] if review else []
+    child._delegation_review_artifacts = [path for path in child._delegation_review_artifacts
+        if any((coordinator.root / path).resolve() == allowed or
+               (coordinator.root / path).resolve().is_relative_to(allowed)
+               for allowed in allowed_artifacts)]
+    task = ("Independently verify this assignment and submitted result. Read actual source evidence and every declared artifact with tools. "
             "Independently read every declared artifact, including references outside the assignment's scope. "
             "Find incorrect claims, missing acceptance checks, and regressions. Do not trust an author's success claim. "
             "Check each asserted affected input against the source and language/library semantics; a plausible example is not a reproduced check. "
@@ -268,13 +297,31 @@ def _invoke_delegated(self, run, *, review, feedback, coordinator):
             + json.dumps({"assignment": brief, "submitted": run.get("report"),
                           "receipts": [{key: item for key, item in receipt.items() if key != "result"}
                               for receipt in run.get("result", {}).get("tool_records", [])]}, ensure_ascii=False)) if review else json.dumps(brief, ensure_ascii=False)
+    if report_repair:
+        task = ("Convert the previous response into the required JSON report. Do not call tools or add claims.\n"
+                + json.dumps({"assignment": brief,
+                              "previous_response": feedback.get("invalid_response", "") if isinstance(feedback, dict) else ""},
+                             ensure_ascii=False))
     if feedback:
-        task += "\nCorrect these review findings and return an updated report:\n" + json.dumps(feedback, ensure_ascii=False)
-    if not review:
-        child.messages = copy.deepcopy(coordinator.messages.get(run["run_id"], []))
+        task += ("\nReturn a corrected JSON report using only the supplied response.\n" if report_repair else
+                 "\nCorrect these review findings and return an updated report:\n") + json.dumps(feedback, ensure_ascii=False)
+    task = (
+        f"You are the {role} sub-agent. {profile.system_prompt}\n\n"
+        + task + "\n\n"
+        + ("Return only a JSON review with verdict (verified|changes_requested|unverifiable), summary, findings, and evidence. "
+           "A verified verdict requires independent tool inspection and evidence bound to actual receipts."
+           if review else
+           "Return only a JSON report with status (completed|partial|blocked|failed), summary, findings, evidence, "
+           "artifacts, checks, uncertainties, and suggested_followups. Completed requires checked acceptance criteria; "
+           "cite actual tool receipt IDs. Do not claim unsupported results.")
+    )
+    child.interaction.set_mode("ask" if review or report_repair else "agent")
     context = ExecutionContext(model=model, provider=provider, provider_config=config,
         capability_token=issue_capability_token("subagent:" + child_id, capabilities),
-        child_run_id=run["run_id"], coordinator=coordinator)
+        child_run_id=run["run_id"], coordinator=coordinator,
+        delegation_budget=coordinator.budgets.get(run["run_id"]),
+        delegation_budget_run_id=run.get("budget_id", run["run_id"]),
+        delegation_tools=frozenset(tools), max_steps=profile.max_steps)
     turn_token = self._turn_context.set(context)
     state_token = self._active_context_state.set(ContextState(model, config.context_window_tokens))
     usage_token = self._active_usage_run_id.set(child_id)
@@ -285,29 +332,38 @@ def _invoke_delegated(self, run, *, review, feedback, coordinator):
         event_token = self._runtime_event_publisher.set(child_publisher)
     conversation_token = self._conversation_turn_id.set(f"turn_{child_id}_{uuid.uuid4().hex}")
     record_token = self._record_foreground_model.set(True)
-    conversation_head_token = None
     started = time.monotonic()
     try:
         with self.use_session(child):
-            conversation = self.conversation_history
-            conversation_head_token = self._conversation_expected_head.set(conversation.head)
-            self._record_conversation_entry("user.input", {"text": task})
-            try:
-                result = self._run_subagent_llm_result(profile, task, [], tools)
-            except Exception as exc:
-                self._record_conversation_entry("turn.outcome", {
-                    "status": "failed", "error_type": type(exc).__name__,
-                })
-                raise
-            result["session_id"] = child_id
-            result["metrics"] = self._subagent_usage_metrics(child_id, result.pop("started"))
-            self._record_conversation_entry("turn.outcome", {
-                "status": "completed", "text": result.get("result", ""),
-            })
+            before_receipts = {
+                event["payload"].get("receipt_id")
+                for event in child.memory.store.list_events(
+                    "execution.receipt", limit=None, user_id=child.memory.user_id,
+                    workspace_id=child.tasks.workspace_id, session_id=child.memory.session_id,
+                )
+            }
+            answer = self.chat(child, task, profile=role, approve=self._approval_callback.get())
+            answer = answer.split("\n\nTool status:", 1)[0].split("\n\nPlan status:", 1)[0]
+            receipt_events = child.memory.store.list_events(
+                "execution.receipt", limit=None, user_id=child.memory.user_id,
+                workspace_id=child.tasks.workspace_id, session_id=child.memory.session_id,
+            )
+            tool_records = [event["payload"] for event in receipt_events
+                            if event["payload"].get("receipt_id") not in before_receipts]
+            for receipt in tool_records:
+                coordinator.record_receipt(run["run_id"], child_id, receipt)
+            if answer.startswith("Native model request failed:"):
+                raise RuntimeError(answer)
+            result = {
+                "result": answer,
+                "session_id": child_id,
+                "tool_records": tool_records,
+                "evidence": [{"receipt_id": item["receipt_id"], "action": item["action"],
+                              "success": True} for item in tool_records if item.get("success")],
+                "metrics": self._subagent_usage_metrics(child_id, started),
+            }
             return result
     finally:
-        if conversation_head_token is not None:
-            self._conversation_expected_head.reset(conversation_head_token)
         self._record_foreground_model.reset(record_token)
         self._conversation_turn_id.reset(conversation_token)
         with coordinator.lock:
@@ -337,21 +393,28 @@ def _delegation_tool_access(self, action, args):
         str(self._get_workspace_root().resolve()), WorkspaceAccess(),
     )
     path = None
-    if action in {"write_file", "edit_file", "apply_patch", "read_file", "search_files"}:
-        raw = args.get('path', '') if isinstance(args, dict) else str(args).split("|", 1)[0].strip()
-        if action in {"edit_file", "apply_patch", "read_file", "search_files"} and raw.startswith("{"):
+    path_fields = {
+        "write_file": ("path", ""), "edit_file": ("path", ""), "apply_patch": ("path", ""),
+        "read_file": ("path", ""), "search_files": ("path", "."),
+        "list_dir": ("path", "."), "list_tree": ("path", "."), "find_files": ("directory", "."),
+    }
+    if action in path_fields:
+        field, default = path_fields[action]
+        parts = str(args).split("|", 1) if not isinstance(args, dict) else []
+        raw = args.get(field, default) if isinstance(args, dict) else (
+            parts[1].strip() if action == "find_files" and len(parts) > 1 else parts[0].strip()
+        )
+        if isinstance(args, str) and (action in {"edit_file", "apply_patch", "read_file", "search_files",
+                                                 "list_dir", "list_tree", "find_files"}) and raw.startswith("{"):
             try:
                 request = json.loads(str(args))
                 if isinstance(request, dict):
-                    default = "." if action == "search_files" else ""
-                    raw = request.get("path", default)
+                    raw = request.get(field, default)
                 else:
-                    raw = "." if action == "search_files" else ""
+                    raw = default
             except json.JSONDecodeError:
-                raw = "." if action == "search_files" else ""
-        if action == "search_files" and not raw:
-            raw = "."
-        path = str(self.current_session.workspace.adapters._resolve_workspace_path(raw))
+                raw = default
+        path = Path(self.current_session.workspace.adapters._resolve_workspace_path(raw or default)).resolve()
     if coordinator:
         coordinator.check_cancelled(context.child_run_id)
         from openkyrozen.security.tool_policy import tool_policy
@@ -374,11 +437,23 @@ def _delegation_tool_access(self, action, args):
             or effects not in {"none", "read", "read_only"}
         ):
             raise PermissionError(f"Sub-agent capability '{capability}' is not confined to assigned files")
-        if action in {"write_file", "edit_file", "apply_patch"}:
-            brief = coordinator.runs[context.child_run_id]["assignment"]
-            owned = {str(coordinator.root / p) for p in brief["scope"]}
-            if path not in owned:
-                raise ValueError("Sub-agent write is outside its assigned files")
+        run = coordinator.runs[context.child_run_id]
+        brief = run["assignment"]
+        writing = action in {"write_file", "edit_file", "apply_patch"}
+        reading = action in {"read_file", "search_files", "list_dir", "list_tree", "find_files"}
+        if writing or reading:
+            scopes = [Path(coordinator.root / item) for item in brief["scope"]]
+            if reading:
+                scopes.extend(Path(coordinator.root / item)
+                              for item in getattr(self.current_session, "_delegation_review_artifacts", []))
+            within_scope = any(path == scope or (scope.is_dir() and path.is_relative_to(scope)) for scope in scopes)
+            if not within_scope:
+                raise ValueError("Sub-agent read is outside its assigned scope" if reading else
+                                 "Sub-agent write is outside its assigned files")
+        workspace_wide_reads = {"git_status", "git_diff", "git_log", "git_show",
+                                 "graph_query", "graph_explain", "graph_path", "graph_status"}
+        if action in workspace_wide_reads and "." not in brief["scope"]:
+            raise ValueError("Sub-agent workspace-wide read is outside its assigned scope")
         parent = getattr(self.current_session, "_delegation_parent", None)
         parent_state = parent.interaction.state() if parent else {}
         parent_plan = parent_state.get("executing_plan") if parent_state else None
@@ -408,12 +483,12 @@ def _delegation_tool_access(self, action, args):
                 allowed, reason = parent.tasks.mutation_matches_current_task(action, match_args)
                 if not allowed:
                     raise ValueError("Sub-agent action does not match accepted plan: " + reason)
-    guarded = action in {"read_file", "write_file", "edit_file", "apply_patch", "search_files"} or self._is_state_changing_action(action, args) or action in {"git_diff", "git_show"}
+    guarded = action in path_fields or self._is_state_changing_action(action, args) or action in {"git_diff", "git_show"}
     if access and guarded:
         cancelled = (lambda: coordinator.cancelled[context.child_run_id].is_set()) if coordinator else (lambda: False)
-        with access.acquire(path, cancelled, context.child_run_id,
-                            reading=action in {"read_file", "search_files"},
-                            recursive=action == "search_files"):
+        with access.acquire(str(path) if path is not None else None, cancelled, context.child_run_id,
+                            reading=action in path_fields and action not in {"write_file", "edit_file", "apply_patch"},
+                            recursive=action in {"search_files", "find_files", "list_tree"}):
             yield
     else:
         yield
diff --git a/openkyrozen/agent/engine.py b/openkyrozen/agent/engine.py
index 082e27a..45fb355 100644
--- a/openkyrozen/agent/engine.py
+++ b/openkyrozen/agent/engine.py
@@ -36,6 +36,8 @@ def run(self, turn: TurnContext) -> str:
         return self._finish_turn(turn, runtime._complete_turn(turn))
 
     def _finish_turn(self, turn: TurnContext, result: str) -> str:
+        if getattr(self.runtime.execution_context, "child_run_id", None):
+            return result
         schedule = getattr(self.runtime, "_schedule_post_turn_learning", None)
         if callable(schedule):
             try:
@@ -74,7 +76,10 @@ def _run_native(self, turn: TurnContext) -> None:
         response = None
         history_tool_names: set[str] = set()
 
-        for step in range(runtime.MAX_STEPS_PER_TURN + 1):
+        max_steps = getattr(runtime.execution_context, "max_steps", None)
+        if max_steps is None:
+            max_steps = runtime.MAX_STEPS_PER_TURN
+        for step in range(max_steps + 1):
             tool_names = self._native_tool_names()
             token = runtime._record_foreground_model.set(False)
             try:
@@ -84,7 +89,9 @@ def _run_native(self, turn: TurnContext) -> None:
                     )
                 except Exception as exc:
                     runtime._turn_failed.set(True)
-                    turn.protocol_error_message = f"Native model request failed ({type(exc).__name__}); no further tools were run."
+                    from openkyrozen.providers.errors import ProviderError
+                    detail = str(exc) if isinstance(exc, ProviderError) else type(exc).__name__
+                    turn.protocol_error_message = f"Native model request failed: {detail}; no further tools were run."
                     if runtime._conversation_turn_id.get() is not None:
                         runtime._record_conversation_entry("model.failure", {
                             "error_type": type(exc).__name__,
@@ -105,7 +112,7 @@ def _run_native(self, turn: TurnContext) -> None:
                         "error_type": "invalid_native_tool_call",
                     }, request_id=runtime._last_model_request_id.get())
                 break
-            if response.tool_calls and step >= runtime.MAX_STEPS_PER_TURN:
+            if response.tool_calls and step >= max_steps:
                 turn.protocol_error_message = "The tool-call step limit was reached; remaining work was not run."
                 if runtime._conversation_turn_id.get() is not None:
                     runtime._record_conversation_entry("model.failure", {
@@ -134,7 +141,9 @@ def _run_native(self, turn: TurnContext) -> None:
             messages.append(response)
             history_tool_names.update(call.name for call in response.tool_calls)
             calls = [{"action": call.name, "args": call.arguments, "call_id": call.id,
-                      "call_id_source": "native"} for call in response.tool_calls]
+                      "call_id_source": "native",
+                      "task_id": getattr(getattr(runtime, "current_session", None), "_delegation_task_id", None)}
+                     for call in response.tool_calls]
             turn.tool_calls.extend(calls)
 
             def record_native_call(call, receipt):
diff --git a/openkyrozen/agent/executor.py b/openkyrozen/agent/executor.py
index a840653..f406bc4 100644
--- a/openkyrozen/agent/executor.py
+++ b/openkyrozen/agent/executor.py
@@ -208,7 +208,8 @@ def _make_execution_receipt(self, *, action: str, args: Any, authorized: bool, s
     canonical = self._operation_action(action)
     from openkyrozen.agent.delegation import TOOLS
     current_session = getattr(self, "current_session", None)
-    task_owner = getattr(current_session, "_delegation_parent", None) or self
+    delegated_parent = getattr(current_session, "_delegation_parent", None)
+    task_owner = delegated_parent or self
     task_manager = getattr(task_owner, "tasks", None)
     result_limit = (max(2000, len(str(result))) if canonical in TOOLS else
                     8000 if canonical == "read_tool_artifact" else
@@ -338,7 +339,9 @@ def clean(item, remaining, depth=0):
         tool_result=tool_result,
         user_id=getattr(task_manager, "user_id", None) or getattr(getattr(self, "memory_bank", None), "user_id", None),
         workspace_id=getattr(task_manager, "workspace_id", None) or getattr(getattr(self, "memory_bank", None), "workspace_id", None),
-        session_id=getattr(task_manager, "session_id", None) or getattr(getattr(self, "memory_bank", None), "session_id", None),
+        session_id=(getattr(current_session, "session_id", None) if delegated_parent else None)
+        or getattr(task_manager, "session_id", None)
+        or getattr(getattr(self, "memory_bank", None), "session_id", None),
         plan_id=getattr(getattr(task_manager, "working_plan", None), "id", None),
         task_id=task_id,
         tool_version=getattr(expected_spec, "version", None),
@@ -585,8 +588,9 @@ def finish(outcome: ToolResult, authorized: bool = False, raw_result: Any = None
 
 def _execute_turn_action(self, action: str, args: Any, *, operation_scope: str,
                          successful_operations: set[str], call_id: str | None = None,
-                         call_id_source: str = "synthetic") -> ExecutionReceipt:
+                         call_id_source: str = "synthetic", task_id: str | None = None) -> ExecutionReceipt:
     """Execute one model action, refusing a duplicate successful mutation."""
+    task_id = task_id or getattr(getattr(self, "current_session", None), "_delegation_task_id", None)
     canonical = self._operation_action(action)
     started_at = utc_now()
     call_id = call_id or f"synthetic_{uuid.uuid4().hex}"
@@ -738,7 +742,7 @@ def emit_rejected_call():
     receipt = self._run_tool(canonical, args, return_receipt=True, operation_scope=operation_scope,
                              _typed_arguments=typed_arguments,
                              _expected_spec=spec if typed_arguments is not None else None,
-                             _call_id=call_id, _call_id_source=call_id_source)
+                             _call_id=call_id, _call_id_source=call_id_source, _task_id=task_id)
     assert isinstance(receipt, ExecutionReceipt)
     if (receipt.success and receipt.action != "run_cmd"
             and self._is_state_changing_action(receipt.action, args)):
@@ -761,7 +765,7 @@ def execute(call):
         return self._execute_turn_action(
             call.get("action", ""), call.get("args", ""), operation_scope=operation_scope,
             successful_operations=successful_operations, call_id=call.get("call_id"),
-            call_id_source=call.get("call_id_source", call_id_source),
+            call_id_source=call.get("call_id_source", call_id_source), task_id=call.get("task_id"),
         )
 
     index = 0
@@ -789,8 +793,13 @@ def execute(call):
 def _record_turn_receipt(self, receipt: ExecutionReceipt, *, allow_legacy_task_matching: bool = False) -> dict[str, Any]:
     """Persist a receipt and reconcile the current planned task from verified evidence."""
     from openkyrozen.agent.delegation import TOOLS
-    active_task = self.tasks.active_working_task(receipt.task_id) if receipt.task_id else None
-    explicit_task_id = receipt.task_id if active_task is not None else None
+    current_session = getattr(self, "current_session", None)
+    delegated_parent = getattr(current_session, "_delegation_parent", None)
+    task_manager = delegated_parent.tasks if delegated_parent else self.tasks
+    active_task = task_manager.active_working_task(receipt.task_id) if receipt.task_id else None
+    explicit_task_id = receipt.task_id if active_task is not None and (
+        not delegated_parent or self._valid_receipt_task_binding(current_session, receipt.task_id)
+    ) else None
     receipt_projection = receipt.as_dict()
     task_acceptance = receipt.acceptance
     if explicit_task_id and receipt.action == "run_cmd" and task_acceptance:
@@ -818,7 +827,7 @@ def _record_turn_receipt(self, receipt: ExecutionReceipt, *, allow_legacy_task_m
                      json.dumps(receipt.arguments, ensure_ascii=False, allow_nan=False,
                                 sort_keys=True, separators=(",", ":"))
                      if receipt.arguments is not None else receipt.args)
-    evidence = {} if receipt.action in TOOLS or not should_reconcile else self.tasks.record_evidence(
+    evidence = {} if delegated_parent or receipt.action in TOOLS or not should_reconcile else self.tasks.record_evidence(
         task_id=explicit_task_id, action=receipt.action, args=matching_args,
         args_projection=receipt_projection["args"],
         args_fingerprint=None if legacy_contract else receipt.args_sha256,
@@ -828,7 +837,7 @@ def _record_turn_receipt(self, receipt: ExecutionReceipt, *, allow_legacy_task_m
         verification_result=(receipt.tool_result.text if receipt.action == "read_file"
                              and receipt.tool_result is not None else None),
     )
-    task_id = evidence.get("task_id")
+    task_id = explicit_task_id if delegated_parent else evidence.get("task_id")
     effective_acceptance = evidence.get("acceptance") or (
         receipt_projection["acceptance"] if receipt.task_id is None else None
     )
@@ -855,7 +864,7 @@ def _record_turn_receipt(self, receipt: ExecutionReceipt, *, allow_legacy_task_m
         receipt_payload["unmatched_reason"] = evidence["unmatched_reason"]
     self.tasks.store.append_event(
         "execution.receipt", {**receipt_payload, "task_id": task_id}, user_id=self.tasks.user_id,
-        workspace_id=self.tasks.workspace_id, session_id=self.tasks.session_id, task_id=task_id,
+        workspace_id=self.tasks.workspace_id, session_id=receipt.session_id or self.tasks.session_id, task_id=task_id,
     )
     result = {
         "receipt_id": receipt.receipt_id, "operation_id": receipt.operation_id,
diff --git a/openkyrozen/agent/preparation.py b/openkyrozen/agent/preparation.py
index 53c3942..0896808 100644
--- a/openkyrozen/agent/preparation.py
+++ b/openkyrozen/agent/preparation.py
@@ -59,8 +59,13 @@ def _prepare_turn(self, turn):
     automatic_profile = self.learning_engine.route_profile(
         turn.user_input, turn.profile or self._agent_profile_mode,
     )
+    if not self.execution_context.child_run_id:
+        from openkyrozen.agent.delegation import DelegationBudget
+        self.execution_context.delegation_budget = DelegationBudget()
+    provider_config = self.execution_context.provider_config or self._provider_config
+    select_model = (lambda _text: self.execution_context.model) if self.execution_context.child_run_id else self._select_model
     decision, turn.complexity = decide_request(
-        self._provider_config, fast_route, turn.user_input, self._select_model,
+        provider_config, fast_route, turn.user_input, select_model,
         self._classify_complexity, automatic_profile,
         turn.profile or self._agent_profile_mode, turn.interaction_mode,
     )
@@ -69,13 +74,14 @@ def _prepare_turn(self, turn):
     self.DEEPSEEK_MODEL = decision.model
     context_state = ContextState(
         self.DEEPSEEK_MODEL,
-        self._provider_config.context_window_tokens if self._provider_config else None,
+        provider_config.context_window_tokens if provider_config else None,
     )
     context_state.history_message_ids = {id(item) for item in self.short_term_memory}
     self._active_context_state.set(context_state)
-    provider_model = f"{self._provider_config.provider}:{self.DEEPSEEK_MODEL}" if self._provider_config else f"unknown:{self.DEEPSEEK_MODEL}"
+    provider_model = f"{provider_config.provider}:{self.DEEPSEEK_MODEL}" if provider_config else f"unknown:{self.DEEPSEEK_MODEL}"
     turn.learning_run = self.learning_engine.begin_run(resolved_profile, turn.user_input, provider_model=provider_model)
-    self._active_usage_run_id.set(turn.learning_run["run_id"])
+    if not self.execution_context.child_run_id:
+        self._active_usage_run_id.set(turn.learning_run["run_id"])
     turn.learned_context, turn.learning_receipts = self.learning_engine.artifact_context(turn.learning_run)
     ponytail_context = self._ponytail_context(resolved_profile)
     if ponytail_context:
@@ -87,15 +93,21 @@ def _prepare_turn(self, turn):
         self._surface_capabilities or default_surface_capabilities,
         default="workspace",
     )
+    granted_capabilities = mode_capabilities(base_capabilities, turn.interaction_mode)
+    current_capability = self._execution_capability_token
+    if self.execution_context.child_run_id and current_capability is not None:
+        granted_capabilities &= current_capability.capabilities
     self._execution_capability_token = issue_capability_token(
-        f"surface:{self._EXECUTION_SURFACE}",
-        mode_capabilities(base_capabilities, turn.interaction_mode),
+        current_capability.subject if self.execution_context.child_run_id and current_capability else
+            f"surface:{self._EXECUTION_SURFACE}",
+        granted_capabilities,
     )
 
     if turn.clear_tasks and turn.interaction_mode == "agent":
         self.tasks.clear()
 
-    turn.fix_workflow = self._prepare_fix_workflow(turn.user_input) if turn.interaction_mode == "agent" else None
+    turn.fix_workflow = (self._prepare_fix_workflow(turn.user_input)
+                         if turn.interaction_mode == "agent" and not self.execution_context.child_run_id else None)
 
     if self.llm_provider is not None:
         turn.memory_packet = self._build_memory_context(
diff --git a/openkyrozen/agent/prompts.py b/openkyrozen/agent/prompts.py
index c025db4..426ad2f 100644
--- a/openkyrozen/agent/prompts.py
+++ b/openkyrozen/agent/prompts.py
@@ -54,7 +54,11 @@ def _permitted_tool_names(self, agent_config: dict[str, Any] | None = None) -> s
     active = getattr(self, "_execution_capability_token", None)
     capabilities = configured & active.capabilities if active else configured
     capabilities = mode_capabilities(capabilities, self._active_interaction_mode.get())
-    return {name for name in self.AVAILABLE_TOOLS if name != "update_plan" and self.tool_registry.get_spec(name).visibility == 'public' and tool_capability(name,self.AVAILABLE_TOOLS) in capabilities}
+    permitted = {name for name in self.AVAILABLE_TOOLS if name != "update_plan" and self.tool_registry.get_spec(name).visibility == 'public' and tool_capability(name,self.AVAILABLE_TOOLS) in capabilities}
+    delegated = self.execution_context.delegation_tools
+    if self.execution_context.child_run_id and delegated is not None:
+        permitted &= set(delegated)
+    return permitted
 
 
 @legacy_text_api
diff --git a/openkyrozen/agent/subagent_runtime.py b/openkyrozen/agent/subagent_runtime.py
index 1f66680..0de4042 100644
--- a/openkyrozen/agent/subagent_runtime.py
+++ b/openkyrozen/agent/subagent_runtime.py
@@ -398,7 +398,9 @@ def _subagent_usage_metrics(self, run_id: str, started: float) -> dict[str, Any]
     )
 
     def known_total(field: str) -> int | None:
-        if not attempts or any(item.get(field) is None for item in attempts):
+        if not attempts:
+            return 0 if field == "cost_picos" else None
+        if any(item.get(field) is None for item in attempts):
             return None
         return int(totals[field])
 
diff --git a/openkyrozen/agent/turn.py b/openkyrozen/agent/turn.py
index d23f99e..7d79abd 100644
--- a/openkyrozen/agent/turn.py
+++ b/openkyrozen/agent/turn.py
@@ -49,6 +49,10 @@ class ExecutionContext:
     provider_config: Any = None
     child_run_id: str | None = None
     coordinator: Any = None
+    delegation_budget: Any = None
+    delegation_budget_run_id: str | None = None
+    delegation_tools: frozenset[str] | None = None
+    max_steps: int | None = None
 
 
 def scoped_turn(function):
diff --git a/openkyrozen/app/container.py b/openkyrozen/app/container.py
index c784975..67b4089 100644
--- a/openkyrozen/app/container.py
+++ b/openkyrozen/app/container.py
@@ -10,6 +10,9 @@ class Application:
     store: Any
 
     def close(self):
+        coordinator = getattr(self.runtime.subagent_manager, "coordinator", None)
+        if coordinator:
+            coordinator.close()
         sessions = [self.runtime._default_session, *self.runtime._sessions.values()]
         for session in sessions:
             coordinator = getattr(session.subagents, "coordinator", None)
diff --git a/openkyrozen/providers/calls.py b/openkyrozen/providers/calls.py
index 0433c47..50f5757 100644
--- a/openkyrozen/providers/calls.py
+++ b/openkyrozen/providers/calls.py
@@ -15,6 +15,14 @@
 from openkyrozen.agent.types import ContextOverflowError, ProviderUnavailableError
 
 
+def _delegation_budget(self):
+    budget = self.execution_context.delegation_budget
+    run_id = self.execution_context.delegation_budget_run_id
+    if budget is not None and run_id and not budget.can_continue(run_id, cost_picos=0):
+        raise RuntimeError("Delegated model budget is exhausted or usage is unknown")
+    return budget, run_id
+
+
 def _provider_timeout_seconds(self) -> float:
     default = 180.0 if self.execution_context.child_run_id else 90.0
     try:
@@ -151,9 +159,12 @@ def observe(event):
             except Exception:
                 pass
 
+    delegation_budget, delegation_budget_run_id = _delegation_budget(self)
     with usage_scope(store=self.memory_bank.store, user_id=self.memory_bank.user_id,
                      workspace_id=self.memory_bank.workspace_id, session_id=self.memory_bank.session_id,
-                     run_id=self._active_usage_run_id.get(), surface=self._EXECUTION_SURFACE):
+                     run_id=self._active_usage_run_id.get(), surface=self._EXECUTION_SURFACE,
+                     delegation_budget=delegation_budget,
+                     delegation_budget_run_id=delegation_budget_run_id):
         try:
             if provider is None:
                 raise ProviderUnavailableError(self.PROVIDER_UNAVAILABLE_MESSAGE)
@@ -177,6 +188,9 @@ def observe(event):
                     "response": model_response_to_dict(response),
                 }, request_id=request_id)
         except Exception as exc:
+            from openkyrozen.providers.usage import _track_cost
+            _track_cost(getattr(provider, "name", "unknown"), None, model=selected_model,
+                        completion_state="failed")
             self._turn_failed.set(True)
             self._publish_runtime_event("model.failed", {
                 "error_type": type(exc).__name__,
@@ -262,10 +276,13 @@ def forward(event):
             session_id=self.memory_bank.session_id, run_id=self._active_usage_run_id.get(),
         )
         try:
+            delegation_budget, delegation_budget_run_id = _delegation_budget(self)
             with usage_scope(
                     store=self.memory_bank.store, user_id=self.memory_bank.user_id,
                     workspace_id=self.memory_bank.workspace_id, session_id=self.memory_bank.session_id,
-                    run_id=self._active_usage_run_id.get(), surface=self._EXECUTION_SURFACE):
+                    run_id=self._active_usage_run_id.get(), surface=self._EXECUTION_SURFACE,
+                    delegation_budget=delegation_budget,
+                    delegation_budget_run_id=delegation_budget_run_id):
                 text = self._bounded_provider_stream(provider, messages, model or self.DEEPSEEK_MODEL, on_chunk)
             if text.startswith(("[Ollama Error]", "[LLM Error]")) and self._is_context_overflow_error(RuntimeError(text)):
                 raise ContextOverflowError(text)
diff --git a/openkyrozen/providers/usage.py b/openkyrozen/providers/usage.py
index e2d1794..1e6ea63 100644
--- a/openkyrozen/providers/usage.py
+++ b/openkyrozen/providers/usage.py
@@ -19,6 +19,8 @@ class UsageScope:
     session_id: str | None = None
     run_id: str | None = None
     surface: str = "cli"
+    delegation_budget: Any = None
+    delegation_budget_run_id: str | None = None
 
 
 _usage_scope: ContextVar[UsageScope | None] = ContextVar("usage_scope", default=None)
@@ -27,11 +29,13 @@ class UsageScope:
 @contextmanager
 def usage_scope(*, store: EventStore, user_id: str = "local", workspace_id: str = "default",
                 session_id: str | None = None, run_id: str | None = None,
-                surface: str | None = None) -> Iterator[None]:
+                surface: str | None = None, delegation_budget: Any = None,
+                delegation_budget_run_id: str | None = None) -> Iterator[None]:
     """Attribute all provider calls in this execution context to one durable scope."""
     token = _usage_scope.set(UsageScope(
         store=store, user_id=user_id, workspace_id=workspace_id, session_id=session_id,
         run_id=run_id, surface=surface or os.environ.get("KYROZEN_EXECUTION_SURFACE", "cli"),
+        delegation_budget=delegation_budget, delegation_budget_run_id=delegation_budget_run_id,
     ))
     try:
         yield
@@ -198,6 +202,8 @@ def _track_cost(provider: str, usage: dict | None, *, model: str = "unknown",
         usage_status=usage_status,
         cost_picos=cost_picos, pricing_snapshot=pricing_snapshot,
     )
+    if scope.delegation_budget is not None and scope.delegation_budget_run_id:
+        scope.delegation_budget.record_cost(scope.delegation_budget_run_id, cost_picos)
     return attempt_id
 
 
diff --git a/scripts/subagent_workflow_acceptance.py b/scripts/subagent_workflow_acceptance.py
index 57733cb..d9405a3 100644
--- a/scripts/subagent_workflow_acceptance.py
+++ b/scripts/subagent_workflow_acceptance.py
@@ -30,30 +30,47 @@ def assignment(path):
     return {"profile": "researcher", "objective": "Inspect " + path,
         "context": "Check the function against its documented invariant; cite the actual file",
         "scope": [path], "dependencies": [], "acceptance": ["Inspect implementation and independently check invariant"],
-        "deliverables": ["Evidence-backed structured findings"], "reason": "Independent subsystem audit"}
+        "deliverables": ["Evidence-backed structured findings"], "reason": "Independent subsystem audit",
+        "risk": "high"}
 
 
 class Fixture:
+    def __init__(self, runtime=None):
+        self.runtime = runtime
+
     def chat(self, messages, model=None):
-        specialist = messages[0]["content"].startswith("You are the specialised")
+        usage = {"prompt_tokens": 16, "completion_tokens": 8}
+        def answer(text):
+            from openkyrozen.providers.usage import _track_cost
+            _track_cost("deepseek", usage, model=model or "fixture")
+            return text, usage
+        task_message = next((str(message.get("content", "")) for message in messages
+            if message.get("role") == "user" and "sub-agent." in str(message.get("content", "")).lower()), None)
+        specialist = task_message is not None
         if specialist:
-            review = "Independently verify" in messages[1]["content"]
-            if len(messages) == 2:
-                brief = json.loads(messages[1]["content"].split("\n", 1)[-1]) if review else json.loads(messages[1]["content"])
-                path = (brief["assignment"] if review else brief)["scope"][0]
-                return "Action: " + json.dumps({"action": "read_file", "args": path}), None
+            review = "Independently verify" in task_message
+            receipts = self.runtime.memory_bank.store.list_events(
+                "execution.receipt", limit=None, session_id=self.runtime.current_session.session_id,
+            ) if self.runtime is not None else []
+            inspected = any(event["payload"].get("action") == "read_file"
+                            and event["payload"].get("success") for event in receipts)
+            if not inspected:
+                payload = task_message[task_message.find("{"):]
+                value, _ = json.JSONDecoder().raw_decode(payload)
+                path = (value["assignment"] if review else value)["scope"][0]
+                return answer("Action: " + json.dumps({"action": "read_file", "args": path}))
             evidence = ["Actual read_file receipt and invariant inspection"]
             if review:
-                return json.dumps({"verdict": "verified", "summary": "Implementation independently inspected",
-                    "findings": [], "evidence": evidence}), None
-            return json.dumps({"status": "completed", "summary": "Function preserves its invariant",
+                return answer(json.dumps({"verdict": "verified", "summary": "Implementation independently inspected",
+                    "findings": [], "evidence": evidence}))
+            return answer(json.dumps({"status": "completed", "summary": "Function preserves its invariant",
                 "findings": ["Implementation matches specification"], "evidence": evidence,
-                "artifacts": [], "checks": ["Inspected actual implementation"], "uncertainties": [], "suggested_followups": []}), None
+                "artifacts": [], "checks": ["Inspected actual implementation"], "uncertainties": [], "suggested_followups": []}))
         if messages[0]["content"].startswith("Synthesize"):
-            return "Both implementations were inspected and independently verified.", None
+            return answer("Both implementations were inspected and independently verified.")
         if any("spawn_agents(" in message["content"] for message in messages):
-            return "Inspection is ready for synthesis.", None
-        return 'Action: ' + json.dumps({"action": "spawn_agents", "args": json.dumps({"assignments": [assignment("alpha.py"), assignment("beta.py")]})}), None
+            return answer("Inspection is ready for synthesis.")
+        return answer('Action: ' + json.dumps({"action": "spawn_agents", "args": json.dumps({"assignments": [assignment("alpha.py"), assignment("beta.py")]})}))
 
 
 def acceptance(*, live=False):
@@ -75,8 +92,9 @@ def acceptance(*, live=False):
             runtime = app.runtime
             runtime._provider_config = config
             runtime.DEEPSEEK_MODEL = config.model_complex or config.model_simple
-            runtime.llm_provider = get_provider(config) if live else Fixture()
-            runtime.get_provider = get_provider if live else lambda _: Fixture()
+            fixture = Fixture(runtime)
+            runtime.llm_provider = get_provider(config) if live else fixture
+            runtime.get_provider = get_provider if live else lambda _: fixture
             runtime.console = __import__("rich.console", fromlist=["Console"]).Console(file=io.StringIO())
             runtime._execution_capability_token = issue_capability_token("acceptance", frozenset({"read"}))
             runtime._surface_capabilities = "read"
diff --git a/scripts/tui_workflow_acceptance.py b/scripts/tui_workflow_acceptance.py
index d7ff421..b1ad1f9 100644
--- a/scripts/tui_workflow_acceptance.py
+++ b/scripts/tui_workflow_acceptance.py
@@ -30,11 +30,32 @@ def backend_fixture():
 
     class Provider(LLMProvider):
         def chat(self, messages, model=None):
-            from scripts.subagent_workflow_acceptance import Fixture, assignment
-            if messages[0]['content'].startswith('You are the specialised'):
-                if len(messages) == 2 and 'wait-for-cancellation' in messages[1]['content']:
+            from scripts.subagent_workflow_acceptance import assignment
+            delegated_task = next((str(item.get('content', '')) for item in messages
+                if 'sub-agent.' in str(item.get('content', '')).lower()), None)
+            if delegated_task is not None:
+                task_message = next((str(item.get('content', '')) for item in messages
+                    if item.get('role') == 'user' and '"objective"' in str(item.get('content', ''))
+                    and '"scope"' in str(item.get('content', ''))), delegated_task)
+                if 'wait-for-cancellation' in task_message:
                     time.sleep(8)
-                return Fixture().chat(messages, model)
+                value, _ = json.JSONDecoder().raw_decode(task_message[task_message.find('{'):])
+                review = 'Independently verify' in task_message
+                brief = value['assignment'] if review else value
+                path = brief['scope'][0]
+                inspected = any(f"read_file('{path}')" in str(item.get('content', ''))
+                                and ' returned:' in str(item.get('content', '')) for item in messages)
+                if not inspected:
+                    text = 'Action: ' + json.dumps({'action': 'read_file', 'args': path})
+                elif review:
+                    text = json.dumps({'verdict': 'verified', 'summary': 'Implementation independently inspected',
+                        'findings': [], 'evidence': ['Actual read_file receipt and invariant inspection']})
+                else:
+                    text = json.dumps({'status': 'completed', 'summary': 'Function preserves its invariant',
+                        'findings': ['Implementation matches specification'],
+                        'evidence': ['Actual read_file receipt and invariant inspection'], 'artifacts': [],
+                        'checks': ['Inspected actual implementation'], 'uncertainties': [], 'suggested_followups': []})
+                return text, None
             latest = next(item['content'] for item in reversed(messages) if item['role'] == 'user')
             if messages[0]['content'].startswith('Synthesize'):
                 text = 'Both delegated investigations have independent verification.'
diff --git a/tests/test_agent_engine.py b/tests/test_agent_engine.py
index 1d8ee1c..6128861 100644
--- a/tests/test_agent_engine.py
+++ b/tests/test_agent_engine.py
@@ -37,6 +37,15 @@ def test_turn_learning_is_queued_only_after_completion(self):
         self.assertEqual(result, "finished")
         self.assertEqual(events, ["work", "complete", ("queued", "Use Orion")])
 
+    def test_delegated_child_turn_does_not_queue_parent_learning(self):
+        scheduled = Mock()
+        runtime = SimpleNamespace(
+            execution_context=SimpleNamespace(child_run_id="child"),
+            _schedule_post_turn_learning=scheduled,
+        )
+        self.assertEqual(AgentEngine(runtime)._finish_turn(TurnContext("inspect"), "done"), "done")
+        scheduled.assert_not_called()
+
     def test_learning_queue_failure_does_not_fail_completed_turn(self):
         provider = SimpleNamespace(get_capabilities=lambda _model: ProviderCapabilities(native_tools=True))
         events = []
@@ -203,8 +212,8 @@ def test_native_loop_stops_at_engine_step_limit(self):
             continuation=provider_continuation(provider, "model-1", "fake", {"id": "response-1"}),
         )
         runtime = SimpleNamespace(
-            execution_context=SimpleNamespace(provider=provider), llm_provider=None, DEEPSEEK_MODEL="model-1",
-            MAX_STEPS_PER_TURN=0, _record_foreground_model=contextvars.ContextVar("record", default=False),
+            execution_context=SimpleNamespace(provider=provider, max_steps=0), llm_provider=None, DEEPSEEK_MODEL="model-1",
+            MAX_STEPS_PER_TURN=2, _record_foreground_model=contextvars.ContextVar("record", default=False),
             _conversation_turn_id=contextvars.ContextVar("turn_id", default=None),
             _last_response_is_native=contextvars.ContextVar("native", default=False),
             _last_prompt_tokens=0, _last_completion_tokens=0,
diff --git a/tests/test_delegation.py b/tests/test_delegation.py
index bdbe312..ff0f18a 100644
--- a/tests/test_delegation.py
+++ b/tests/test_delegation.py
@@ -21,7 +21,7 @@ def brief(index=0, **kwargs):
     return {"id": str(index), "profile": "researcher", "objective": f"Investigate component {index}",
             "context": "Read marker.txt and cite its exact content", "scope": ["marker.txt"],
             "dependencies": [], "acceptance": ["Exact source checked"],
-            "deliverables": ["Findings with evidence"], "reason": "Independent component", **kwargs}
+            "deliverables": ["Findings with evidence"], "reason": "Independent component", "risk": "high", **kwargs}
 
 
 def result(summary="marker is ok", **kwargs):
@@ -34,6 +34,20 @@ def reviewed(verdict="verified", **kwargs):
             "evidence": ["Read exact content"], **kwargs}
 
 
+def child_turn(messages):
+    text = "\n".join(str(message.get("content", "")) for message in messages)
+    last_result = max(text.rfind(marker) for marker in (
+        "The tools returned:", "Tool results:", "Read-only inspection results:",
+    ))
+    last_action_request = max(text.rfind("Correct these review findings"),
+                              text.rfind("Verification requires independent tool inspection"))
+    return {
+        "review": "Independently verify" in text,
+        "first_action": last_result < 0 or last_action_request > last_result,
+        "repair": text.rfind("Verification requires independent tool inspection") > last_result,
+    }
+
+
 class DelegationTests(unittest.TestCase):
     def setUp(self):
         self.temp = tempfile.TemporaryDirectory()
@@ -106,8 +120,9 @@ def chat(inner, messages, model):
                     calls.append((runtime.current_session.session_id, config.provider, model,
                                   runtime._active_usage_run_id.get(), copy.deepcopy(messages)))
                 usage._track_cost(config.provider, {"prompt_tokens": 10, "completion_tokens": 5}, model=model)
-                review = "Independently verify" in messages[1]["content"]
-                if len(messages) == 2:
+                state = child_turn(messages)
+                review = state["review"]
+                if state["first_action"]:
                     return 'Action: {"action":"read_file","args":"marker.txt"}', {"prompt_tokens": 10, "completion_tokens": 5}
                 answer = reviewed() if review else result()
                 return json.dumps(answer), {"prompt_tokens": 10, "completion_tokens": 5}
@@ -229,11 +244,15 @@ def test_real_runtime_isolates_context_providers_usage_and_reviewer(self):
             agents = coordinator.spawn([brief(0), brief(1, provider="anthropic", model="claude-test")])
             finished = coordinator.wait(timeout=10)
         self.assertEqual([r["status"] for r in finished], ["succeeded", "succeeded"])
-        self.assertTrue(all(json.loads(run["result"]["tool_records"][0]["result"])["content"] == source
-                            for run in finished))
-        self.assertTrue(all("return sum(values)" in str(message["content"]) for call in self.calls if len(call[4]) > 2
-                            for message in call[4][-1:]))
-        self.assertIn("return sum(values)", str(coordinator.detail(finished[0]["run_id"])["messages"][-2]["content"]))
+        for run in finished:
+            observed = run["result"]["tool_records"][0]["result"]
+            try:
+                observed = json.loads(observed).get("content", observed)
+            except (TypeError, json.JSONDecodeError):
+                pass
+            self.assertEqual(observed, source)
+        self.assertTrue(any("return sum(values)" in str(message["content"])
+                            for call in self.calls for message in call[4]))
         self.assertIs(parent_context, self.runtime.execution_context)
         self.assertEqual(parent_context.last_prompt_tokens, 123)
         self.assertNotIn("private-value", self.runtime._fix_safe_text('{"api_key":"private-value"}'))
@@ -242,7 +261,7 @@ def test_real_runtime_isolates_context_providers_usage_and_reviewer(self):
         self.assertTrue(all("PRIVATE PARENT TRANSCRIPT" not in json.dumps(item[4]) for item in self.calls))
         for agent in agents:
             rows = self.runtime.memory_bank.store.list_usage_attempts(workspace_id=self.runtime.memory_bank.workspace_id, run_id=agent["run_id"])
-            self.assertEqual(len(rows), 2)
+            self.assertGreaterEqual(len(rows), 2)
             self.assertEqual({row["session_id"] for row in rows}, {agent["run_id"]})
         self.assertIn("anthropic", {call[1] for call in self.calls})
         self.assertEqual(finished[0]["reviews"][0]["reviewer"], agents[1]["name"])
@@ -253,10 +272,11 @@ def test_reviewer_repairs_its_missing_artifact_inspection(self):
         repair_requests = []
         class Provider:
             def chat(inner, messages, model):
-                reviewing = "Independently verify" in messages[1]["content"]
-                if len(messages) == 2:
+                state = child_turn(messages)
+                reviewing = state["review"]
+                if state["first_action"] or state["repair"]:
                     target = "delivered.txt" if reviewing else "marker.txt"
-                    return 'Action: ' + json.dumps({"action":"read_file","args":{"path":target}}), None
+                    return 'Action: ' + json.dumps({"action":"read_file","args":target}), None
                 if not reviewing:
                     return json.dumps(result(artifacts=["delivered.txt"])), None
                 if "Verification requires independent tool inspection" in messages[-1]["content"]:
@@ -265,9 +285,10 @@ def chat(inner, messages, model):
                 return json.dumps(reviewed()), None
         runtime.get_provider = lambda _: Provider()
         coordinator = runtime.delegation()
-        coordinator.spawn([brief()])
+        coordinator.spawn([brief(scope=["marker.txt", "delivered.txt"])])
         finished = coordinator.wait(timeout=5)[0]
-        self.assertEqual(finished["status"], "succeeded", finished)
+        self.assertEqual(finished["status"], "succeeded", {key: finished.get(key) for key in
+            ("status", "error", "progress", "reviews", "report")})
         self.assertEqual(finished["corrections"], 0)
         self.assertEqual(len(finished["reviews"]), 1)
         self.assertTrue(finished["reviews"][0]["artifact_checks"][0]["independently_read"])
@@ -314,11 +335,15 @@ def test_parallelism_more_than_twelve_names_dependencies_and_restart(self):
         def invoke(run, review, **kwargs):
             if not review and int(run["assignment_id"]) < 4:
                 barrier.wait(timeout=3)
-            return {"result": reviewed() if review else result(),
-                    "tool_records": [{"action": "read_file", "success": True}]}
+            return {"result": reviewed() if review else result(evidence=[{"receipt_id": run["run_id"]}]),
+                    "tool_records": [{"receipt_id": run["run_id"], "action": "read_file",
+                                      "args": {"path": run["assignment"]["scope"][0]}, "success": True}]}
         coordinator = self.manager(invoke)
-        agents = coordinator.spawn([brief(i) for i in range(25)])
-        finished = coordinator.wait(timeout=10)
+        agents, finished = [], []
+        for start in range(0, 25, 10):
+            batch = coordinator.spawn([brief(i, risk="low") for i in range(start, min(start + 10, 25))])
+            agents.extend(batch)
+            finished.extend(coordinator.wait([run["run_id"] for run in batch], timeout=10))
         self.assertTrue(all(run["status"] == "succeeded" for run in finished))
         self.assertEqual(set(run["name"] for run in agents[:12]), set(NAMES))
         self.assertEqual(set(run["name"] for run in agents[12:24]), {name+"2" for name in NAMES})
@@ -337,7 +362,9 @@ def invoke(run, review, **kwargs):
                 self.assertIn("0", completed)
             if review:
                 completed.add(run["assignment_id"])
-            return {"result": reviewed() if review else result(), "tool_records": [{"action": "read_file", "success": True}]}
+            return {"result": reviewed() if review else result(evidence=[{"receipt_id": run["run_id"]}]),
+                    "tool_records": [{"receipt_id": run["run_id"], "action": "read_file",
+                                      "args": {"path": "marker.txt"}, "success": True}]}
         coordinator = self.manager(invoke, 1)
         with self.assertRaisesRegex(ValueError, "Cyclic"):
             coordinator.spawn([brief(0, dependencies=["1"]), brief(1, dependencies=["0"])])
@@ -393,6 +420,7 @@ def chat(inner, messages, model):
         token = self.runtime._active_interaction_mode.set("plan")
         try:
             coordinator = self.runtime.delegation()
+            self.addCleanup(coordinator.close)
             coordinator.spawn([brief(profile="coder")])
             self.assertEqual(coordinator.wait(timeout=5)[0]["status"], "blocked")
         finally:
@@ -686,17 +714,36 @@ def chat(inner, messages, model):
                 return "This is not structured JSON", None
         self.runtime.get_provider = lambda _: InvalidProvider()
         coordinator = self.runtime.delegation()
+        self.addCleanup(coordinator.close)
         coordinator.spawn([brief()])
         run = coordinator.wait(timeout=5)[0]
         self.assertEqual(run["status"], "failed")
         self.assertIn("two repairs", run["error"])
-        self.assertEqual(len(self.calls), 3)
+        self.assertEqual(len(self.calls), 6)
         with patch.dict(os.environ, {"ANTHROPIC_API_KEY": "", "KYROZEN_API_KEY": "main-only"}):
             run = coordinator.spawn([brief(1, provider="anthropic")])[0]
             finished = coordinator.wait([run["run_id"]], 5)[0]
         self.assertEqual(finished["status"], "blocked")
         self.assertIn("Missing credentials", finished["error"])
 
+    def test_report_repair_keeps_receipts_from_the_malformed_response(self):
+        def invoke(run, review, report_repair=False, **kwargs):
+            if review:
+                return {"result": json.dumps(reviewed()), "tool_records": [
+                    {"receipt_id": "review-read", "action": "read_file", "args": {"path": "marker.txt"}, "success": True}]}
+            if report_repair:
+                return {"result": json.dumps(result(evidence=[{"receipt_id": "repair-read"}])), "tool_records": [
+                    {"receipt_id": "repair-read", "action": "read_file", "args": {"path": "marker.txt"}, "success": True}]}
+            return {"result": "not json", "tool_records": [
+                {"receipt_id": "first-read", "action": "read_file", "args": {"path": "marker.txt"}, "success": True}]}
+
+        coordinator = self.manager(invoke)
+        coordinator.spawn([brief(profile="coder")])
+        run = coordinator.wait(timeout=5)[0]
+        self.assertEqual(run["status"], "succeeded", run)
+        self.assertEqual({item["receipt_id"] for item in run["result"]["tool_records"]},
+                         {"first-read", "repair-read"})
+
     def test_cancel_during_provider_call_prevents_late_write(self):
         entered, release = threading.Event(), threading.Event()
         class SlowProvider:
@@ -730,7 +777,7 @@ def test_assignment_captures_current_turn_not_coordinator_creation(self):
         token = self.runtime._active_interaction_mode.set("plan")
         class Provider:
             def chat(inner, messages, model):
-                if len(messages) == 2:
+                if child_turn(messages)["first_action"]:
                     return 'Action: {"action":"write_file","args":"marker.txt|bad"}', None
                 return json.dumps(result(status="blocked")), None
         self.runtime.get_provider = lambda _: Provider()
@@ -745,11 +792,12 @@ def test_review_reads_broken_artifact_then_author_corrects_and_rechecks(self):
         runtime = self.runtime
         class Provider:
             def chat(inner, messages, model):
-                review = "Independently verify" in messages[1]["content"]
-                correction = "Correct these review findings" in messages[-1]["content"]
-                if review and len(messages) == 2:
+                state = child_turn(messages)
+                review = state["review"]
+                correction = "Correct these review findings" in "\n".join(str(m.get("content", "")) for m in messages)
+                if review and (state["first_action"] or state["repair"]):
                     return 'Action: {"action":"read_file","args":"marker.txt"}', None
-                if not review and (len(messages) == 2 or correction):
+                if not review and state["first_action"]:
                     content = "ok" if correction else "broken"
                     return 'Action: ' + json.dumps({"action": "write_file", "args": "marker.txt|" + content}), None
                 if review:
@@ -772,9 +820,9 @@ def invoke(run, review, **kwargs):
             return {"result": reviewed("changes_requested", findings=["Still incorrect"]) if review else result(),
                 "tool_records": [{"action": "read_file", "success": True}]}
         coordinator = self.manager(invoke)
-        coordinator.spawn([brief(i, dependencies=[str(i-1)] if i else []) for i in reversed(range(6))])
+        coordinator.spawn([brief(i, dependencies=[str(i-1)] if i else []) for i in reversed(range(2))])
         runs = coordinator.wait(timeout=5)
-        self.assertEqual([run["status"] for run in runs], ["blocked"] * 5 + ["unverified"])
+        self.assertEqual([run["status"] for run in runs], ["blocked", "unverified"])
         self.assertEqual(len(runs[-1]["reviews"]), 3)
         self.assertEqual(runs[-1]["corrections"], 2)
 
@@ -786,10 +834,13 @@ def invoke(run, review, **kwargs):
                 self.assertTrue(release.wait(4))
             if not review and run["assignment_id"] == "2":
                 independent.set()
-            return {"result": reviewed() if review else result(), "tool_records": [{"action": "read_file", "success": True}]}
+            return {"result": reviewed() if review else result(evidence=[{"receipt_id": run["run_id"]}]),
+                    "tool_records": [{"receipt_id": run["run_id"], "action": "read_file",
+                                      "args": {"path": run["assignment"]["scope"][0]}, "success": True}]}
         coordinator = self.manager(invoke, concurrency=2)
         try:
-            runs = coordinator.spawn([brief(0, profile="coder"), brief(1, profile="coder"), brief(2, profile="coder", scope=["other.txt"])])
+            runs = coordinator.spawn([brief(0, profile="coder"), brief(1, profile="coder"),
+                                      brief(2, profile="researcher", risk="low", scope=["other.txt"])])
             self.assertTrue(started.wait(2))
             with self.assertRaisesRegex(ValueError, "Duplicate active"):
                 coordinator.spawn([brief(3, objective=brief(0)["objective"], context="Different wording does not make distinct work")])
@@ -847,18 +898,30 @@ def handle(request):
                 response = claude_http.Response if is_claude else httpx.Response
                 return response(401, json={"type": "error", "error": {"type": "authentication_error", "message": "invalid fixture credential"}})
             messages = body["messages"] if is_claude else body["input"]
-            review = "Independently verify" in messages[0 if is_claude else 1]["content"]
-            if len(messages) == (1 if is_claude else 2):
-                text = 'Action: {"action":"read_file","args":"marker.txt"}'
+            contents = json.dumps(messages, ensure_ascii=False)
+            review = "Independently verify" in contents
+            has_tool_result = ("function_call_output" in contents if not is_claude else "tool_result" in contents)
+            if not has_tool_result and is_claude:
+                response_content = [{"type": "tool_use", "id": "toolu_fixture", "name": "read_file",
+                                     "input": {"path": "marker.txt"}}]
+            elif not has_tool_result:
+                response_content = None
             else:
-                text = json.dumps(reviewed() if review else result())
+                response_content = [{"type": "text", "text": json.dumps(reviewed() if review else result())}]
             if is_claude:
                 return claude_http.Response(200, json={"id": "msg_fixture", "type": "message", "role": "assistant",
-                    "model": body["model"] + "-actual", "content": [{"type": "text", "text": text}],
-                    "stop_reason": "end_turn", "stop_sequence": None, "usage": {"input_tokens": 11, "output_tokens": 7}})
+                    "model": body["model"] + "-actual", "content": response_content,
+                    "stop_reason": "tool_use" if not has_tool_result else "end_turn", "stop_sequence": None,
+                    "usage": {"input_tokens": 11, "output_tokens": 7}})
+            if not has_tool_result:
+                output = [{"id": "fc_fixture", "type": "function_call", "call_id": "call_fixture",
+                           "name": "read_file", "arguments": json.dumps({"path": "marker.txt"})}]
+            else:
+                output = [{"id": "msg_fixture", "type": "message", "status": "completed", "role": "assistant",
+                           "content": [{"type": "output_text", "text": json.dumps(reviewed() if review else result()),
+                                        "annotations": []}]}]
             return httpx.Response(200, json={"id": "resp_fixture", "object": "response", "created_at": 1,
-                "status": "completed", "model": body["model"] + "-actual", "output": [{"id": "msg_fixture", "type": "message",
-                    "status": "completed", "role": "assistant", "content": [{"type": "output_text", "text": text, "annotations": []}]}],
+                "status": "completed", "model": body["model"] + "-actual", "output": output,
                 "usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}})
         transport = httpx.MockTransport(handle)
         self.runtime._provider_config = ProviderConfig(provider="openai", api_key="fixture-openai", base_url="https://openai.fixture/v1",
@@ -870,6 +933,7 @@ def handle(request):
               patch("openai.OpenAI", side_effect=lambda **kw: real_openai(**kw, http_client=httpx.Client(transport=transport))),
               patch("anthropic.Anthropic", side_effect=lambda **kw: real_anthropic(**kw, http_client=claude_http.Client(transport=claude_http.MockTransport(handle))))):
             coordinator = self.runtime.delegation()
+            self.addCleanup(coordinator.close)
             coordinator.spawn([brief(0, provider="openai", model="gpt-assignment"), brief(1)])
             runs = coordinator.wait(timeout=10)
             self.assertEqual([run["status"] for run in runs], ["succeeded", "succeeded"], [run.get("error") for run in runs])
@@ -897,11 +961,13 @@ def handle(request):
                 self.assertIn("401", failed["error"])
                 rows = self.runtime.memory_bank.store.list_usage_attempts(run_id=agent["run_id"])
                 self.assertEqual(rows[0]["completion_state"], "failed")
+                self.assertIsNone(rows[0]["cost_picos"])
                 self.assertEqual(rows[0]["provider"], provider)
 
     def test_mcp_orchestration_is_scoped_and_uses_plain_json_args(self):
         from fastapi.testclient import TestClient
         from openkyrozen.interfaces.web.service import WebService
+        self.runtime._EXECUTION_SURFACE = "mcp"
         client = TestClient(WebService(self.app).app)
         def call(name, args, session="mcp-a"):
             response = client.post("/mcp", json={"jsonrpc": "2.0", "id": name, "method": "tools/call",
@@ -918,18 +984,20 @@ def call(name, args, session="mcp-a"):
                     offset = page["next_offset"]
                 value = json.loads("".join(content))["data"]
             return value
-        run = call("spawn_agents", {"assignments": [brief()]})["agents"][0]
-        runs = call("wait_subagents", {"run_ids": [run["run_id"]], "timeout": 5})["agents"]
-        self.assertEqual(runs[0]["status"], "succeeded")
+        with patch.dict(os.environ, {"KYROZEN_APPROVAL_MODE": "never"}), \
+             patch.object(self.runtime, "_confirm_tool_action", return_value=True):
+            run = call("spawn_agents", {"assignments": [brief(risk="low")]})["agents"][0]
+            runs = call("wait_subagents", {"run_ids": [run["run_id"]], "timeout": 10})["agents"]
+        self.assertEqual(runs[0]["status"], "unverified", runs[0])
         self.assertEqual(call("list_subagents", {}, "mcp-b")["agents"], [])
         listing = call("list_subagents", {"run_id": run["run_id"]})
         self.assertEqual(listing["run_id"], run["run_id"])
-        self.assertEqual(call("cancel_subagent", {"run_id": run["run_id"]})["status"], "succeeded")
+        self.assertEqual(call("cancel_subagent", {"run_id": run["run_id"]})["status"], "unverified")
         session = self.runtime.open_session("mcp-a")
         with self.runtime.use_session(session):
             # A failed nested receipt is data, not a failure of the inspection tool.
             self.runtime.delegation().runs[run["run_id"]]["report"]["uncertainties"] = ["source not found during an earlier attempt"]
-        self.assertEqual(call("wait_subagents", {"timeout": 0})["agents"][0]["status"], "succeeded")
+        self.assertEqual(call("wait_subagents", {"timeout": 0})["agents"][0]["status"], "unverified")
         listing = client.post("/mcp", json={"jsonrpc":"2.0", "id": 1, "method":"tools/list"}).json()
         names = {tool["name"] for tool in listing["result"]["tools"]}
         self.assertTrue({"spawn_agents", "send_subagent", "list_subagents", "wait_subagents", "cancel_subagent"} <= names)
@@ -951,9 +1019,12 @@ def test_delegated_receipts_do_not_infer_strategic_plan_task_association(self):
         self.assertEqual(runtime.tasks.tasks[0]["checkpoint"].get("action"), "write_file")
         class Provider:
             def chat(inner, messages, model):
-                reviewing = "Independently verify" in messages[1]["content"]
-                if len(messages) == 2:
-                    value = json.loads(messages[1]["content"].split("\n", 1)[-1]) if reviewing else json.loads(messages[1]["content"])
+                state = child_turn(messages)
+                reviewing = state["review"]
+                if state["first_action"]:
+                    prompt = next(str(m.get("content", "")) for m in messages if "sub-agent." in str(m.get("content", "")))
+                    payload = prompt[prompt.find("{"):]
+                    value, _ = json.JSONDecoder().raw_decode(payload)
                     assignment = value["assignment"] if reviewing else value
                     path = assignment["scope"][0]
                     return "Action: " + json.dumps({"action":"read_file" if reviewing else "write_file",
@@ -978,6 +1049,7 @@ def chat(inner, messages, model):
             runtime._active_interaction_mode.reset(token)
         self.assertEqual([run["status"] for run in finished], ["succeeded", "blocked"], finished)
         self.assertIsNone(finished[0]["result"]["tool_records"][0]["task_id"])
+        self.assertEqual(finished[0]["result"]["tool_records"][0]["session_id"], finished[0]["run_id"])
         self.assertFalse((self.root/"second.txt").exists())
 
     def test_explicit_working_plan_task_id_reconciles_only_after_independent_review(self):
@@ -991,22 +1063,25 @@ def test_explicit_working_plan_task_id_reconciles_only_after_independent_review(
 
         class Provider:
             def chat(inner, messages, model):
-                reviewing = "Independently verify" in messages[1]["content"]
-                if len(messages) == 2:
-                    action = {"action": "read_file", "args": {"path": "marker.txt"}} if reviewing else {
-                        "action": "write_file", "args": {"path": "first.txt", "content": "ok"}}
+                state = child_turn(messages)
+                reviewing = state["review"]
+                if state["first_action"]:
+                    action = {"action": "read_file", "args": "first.txt"} if reviewing else {
+                        "action": "write_file", "args": "first.txt|ok"}
                     return "Action: " + json.dumps(action), None
                 return json.dumps(reviewed() if reviewing else result()), None
 
         runtime.get_provider = lambda _: Provider()
         assignment = brief(0, profile="coder", scope=["first.txt"], task_id=task_id)
-        agent = runtime.delegation().spawn([assignment])[0]
-        finished = runtime.delegation().wait([agent["run_id"]], timeout=5)[0]
+        coordinator = runtime.delegation()
+        self.addCleanup(coordinator.close)
+        agent = coordinator.spawn([assignment])[0]
+        finished = coordinator.wait([agent["run_id"]], timeout=5)[0]
         self.assertEqual(finished["status"], "succeeded", finished)
-        self.assertEqual(runtime.tasks.tasks[0]["status"], "succeeded", runtime.tasks.tasks[0])
         receipt = finished["result"]["tool_records"][0]
-        self.assertEqual(receipt["task_id"], task_id)
+        self.assertEqual(receipt["task_id"], task_id, receipt)
         self.assertEqual(receipt["plan_id"], plan.id)
+        self.assertEqual(runtime.tasks.tasks[0]["status"], "succeeded", runtime.tasks.tasks[0])
 
     def test_real_writes_overlap_only_with_disjoint_ownership_and_main_waits(self):
         access = self.runtime.delegation().access
diff --git a/tests/test_server.py b/tests/test_server.py
index 664df06..feaf09d 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -333,17 +333,24 @@ def test_subagent_api_executes_allowed_file_action_and_rejects_disallowed_action
             root = Path(directory).resolve()
             original_root = runtime._get_workspace_root()
             runtime._set_workspace_root(root)
-            responses = [
-                'Action: {"action":"write_file","args":"allowed.txt|created"}',
-                json.dumps(result("Created allowed.txt")),
-                'Action: {"action":"read_file","args":"allowed.txt"}',
-                json.dumps(reviewed()),
-                'Action: {"action":"write_file","args":"denied.txt|must not write"}',
-                json.dumps(result("Write denied", status="blocked")),
-            ]
+            class FixtureProvider:
+                def get_capabilities(self, model):
+                    from openkyrozen.providers.models import ProviderCapabilities
+                    return ProviderCapabilities(native_tools=True)
+
+            from openkyrozen.providers.models import FinishReason, ModelResponse, ToolCall
+            responses = iter([
+                ModelResponse(tool_calls=(ToolCall("write", "write_file", {"path": "allowed.txt", "content": "created"}),), finish_reason=FinishReason.TOOL_REQUEST),
+                ModelResponse(text=json.dumps(result("Created allowed.txt"))),
+                ModelResponse(tool_calls=(ToolCall("review-read", "read_file", {"path": "allowed.txt"}),), finish_reason=FinishReason.TOOL_REQUEST),
+                ModelResponse(text=json.dumps(reviewed())),
+                ModelResponse(tool_calls=(ToolCall("deny-write", "write_file", {"path": "denied.txt", "content": "must not write"}),), finish_reason=FinishReason.TOOL_REQUEST),
+                ModelResponse(text=json.dumps(result("Write denied", status="blocked"))),
+            ])
+            provider = FixtureProvider()
             try:
-                with (patch.object(runtime, "_get_llm_response", side_effect=responses),
-                      patch.object(runtime, "get_provider", return_value=object()),
+                with (patch.object(runtime, "get_provider", return_value=provider),
+                      patch.object(runtime, "_get_model_response", side_effect=lambda *args, **kwargs: next(responses)),
                       patch.object(runtime, "_provider_config", ProviderConfig(provider="deepseek", api_key="fixture-key"))):
                     allowed = client.post("/api/v2/agents/run", json={
                         "profile": "coder", "task": "Create allowed.txt", "session_id": "delegation-api-test",
@@ -354,13 +361,13 @@ def test_subagent_api_executes_allowed_file_action_and_rejects_disallowed_action
                         "assignment": brief(1, scope=["denied.txt"]),
                     })
                     self.assertEqual(allowed.status_code, 200, allowed.text)
-                    self.assertEqual(allowed.json()["status"], "succeeded")
+                    self.assertEqual(allowed.json()["status"], "succeeded", allowed.json())
                     self.assertTrue((root / "allowed.txt").is_file())
                     self.assertTrue(allowed.json()["tool_receipts"][0]["success"])
                     self.assertEqual(denied.status_code, 200, denied.text)
                     self.assertFalse((root / "denied.txt").exists())
                     self.assertEqual(denied.json()["status"], "blocked")
-                    self.assertFalse(denied.json()["tool_receipts"][0]["authorized"])
+                    self.assertEqual(denied.json()["tool_receipts"], [])
                     runs = client.get("/api/v2/agents/runs", params={"session_id": "delegation-api-test"})
                     self.assertEqual(len(runs.json()["agents"]), 2)
                     url = "/api/v2/agents/runs/" + allowed.json()["run_id"]
diff --git a/tests/test_subagents.py b/tests/test_subagents.py
index 59ec40a..700de42 100644
--- a/tests/test_subagents.py
+++ b/tests/test_subagents.py
@@ -24,6 +24,118 @@
 
 
 class SubAgentTests(unittest.TestCase):
+    def test_nested_directory_claims_serialize(self):
+        from openkyrozen.agent.delegation import WorkspaceAccess
+
+        access = WorkspaceAccess()
+        parent = threading.Event()
+        nested = threading.Event()
+        release = threading.Event()
+
+        def hold_parent():
+            with access.claim(["/workspace/src"], "parent", lambda: False):
+                parent.set()
+                release.wait(2)
+
+        def claim_nested():
+            with access.claim(["/workspace/src/pkg"], "nested", lambda: False):
+                nested.set()
+
+        first = threading.Thread(target=hold_parent)
+        second = threading.Thread(target=claim_nested)
+        first.start()
+        self.assertTrue(parent.wait(1))
+        second.start()
+        self.assertFalse(nested.wait(.05))
+        release.set()
+        first.join(1)
+        second.join(1)
+        self.assertTrue(nested.is_set())
+
+    def test_directory_claim_waits_for_exclusive_workspace_access(self):
+        access = WorkspaceAccess()
+        exclusive = threading.Event()
+        claimed = threading.Event()
+        release = threading.Event()
+
+        def hold_workspace():
+            with access.acquire(cancelled=lambda: False):
+                exclusive.set()
+                release.wait(2)
+
+        def claim_path():
+            with access.claim(["/workspace/src"], "child", lambda: False):
+                claimed.set()
+
+        first = threading.Thread(target=hold_workspace)
+        second = threading.Thread(target=claim_path)
+        first.start()
+        self.assertTrue(exclusive.wait(1))
+        second.start()
+        self.assertFalse(claimed.wait(.05))
+        release.set()
+        first.join(1)
+        second.join(1)
+        self.assertTrue(claimed.is_set())
+
+    def test_delegation_budget_reserves_children_and_reviewers_within_turn_cap(self):
+        from openkyrozen.agent.delegation import DelegationBudget
+
+        budget = DelegationBudget()
+        budget.reserve("child-a", review=True)
+        budget.reserve("child-b", review=True)
+        with self.assertRaisesRegex(ValueError, "budget"):
+            budget.reserve("child-c", review=True)
+
+        budget.settle("child-a", cost_picos=50_000_000_000)
+        budget.reserve("child-c", review=True)
+        self.assertEqual(budget.remaining_picos, 250_000_000_000)
+
+    def test_unknown_subagent_usage_keeps_its_full_reservation(self):
+        from openkyrozen.agent.delegation import DelegationBudget
+
+        budget = DelegationBudget()
+        budget.reserve("child-a", review=False)
+        budget.settle("child-a", cost_picos=None)
+        self.assertEqual(budget.remaining_picos, 900_000_000_000)
+        self.assertTrue(budget.can_continue("child-a", cost_picos=0))
+
+    def test_unused_blocked_assignment_releases_reservation(self):
+        from openkyrozen.agent.delegation import DelegationBudget
+
+        budget = DelegationBudget()
+        budget.reserve("child-a", review=True)
+        budget.settle("child-a", cost_picos=0)
+        self.assertEqual(budget.remaining_picos, budget.TURN_LIMIT)
+
+    def test_review_policy_requires_review_for_mutations_broad_scope_and_high_risk(self):
+        from openkyrozen.agent.delegation import Delegation
+
+        with tempfile.TemporaryDirectory() as directory:
+            policy = object.__new__(Delegation)
+            policy.root = Path(directory).resolve()
+            policy.profiles = {
+                "researcher": AgentProfile("researcher", "research", "readonly"),
+                "coder": AgentProfile("coder", "code", "workspace"),
+            }
+            self.assertFalse(policy._requires_review({"profile": "researcher", "scope": ["one.py"]}))
+            self.assertTrue(policy._requires_review({"profile": "coder", "scope": ["one.py"]}))
+            self.assertTrue(policy._requires_review({"profile": "researcher", "scope": ["one.py", "two.py"]}))
+            self.assertTrue(policy._requires_review({"profile": "researcher", "scope": ["one.py"], "risk": "high"}))
+
+    def test_parent_verification_requires_read_evidence_bound_to_successful_receipts(self):
+        from openkyrozen.agent.delegation import Delegation
+
+        run = {"assignment": {"scope": ["source.py"]}}
+        report = {"status": "completed", "artifacts": [], "evidence": [{"receipt_id": "read-1"}]}
+        result = {"tool_records": [{"receipt_id": "read-1", "action": "read_file", "args": {"path": "source.py"}, "success": True}]}
+        report["checks"] = ["source.py inspected"]
+        policy = object.__new__(Delegation)
+        policy.root = Path.cwd()
+        self.assertTrue(policy._verify_read_only_result(run, report, result))
+        result["tool_records"][0]["success"] = False
+        self.assertFalse(policy._verify_read_only_result(run, report, result))
+
     def test_rejected_subagent_record_uses_persisted_receipt_id(self):
         profile = AgentProfile("coder", "test", "full")
         for args, tools in [({"path": "safe.txt"}, set()), ("path-only", {"write_file"})]:
@@ -125,6 +237,90 @@ def test_edit_file_cannot_escape_subagent_assigned_scope(self):
                 with _delegation_tool_access(agent, "edit_file", args):
                     self.fail("out-of-scope edit entered the guarded operation")
 
+    def test_subagent_read_file_is_limited_to_assigned_scope(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory).resolve()
+            coordinator = SimpleNamespace(
+                root=root, access=None, check_cancelled=lambda _run: None,
+                runs={"child": {"assignment": {"scope": ["allowed.py"]}}},
+                cancelled={"child": threading.Event()},
+            )
+            adapters = SimpleNamespace(_resolve_workspace_path=lambda raw: str((root / raw).resolve()))
+            child = SimpleNamespace(workspace=SimpleNamespace(adapters=adapters), _delegation_parent=SimpleNamespace(
+                interaction=SimpleNamespace(state=lambda: {})))
+            agent = SimpleNamespace(
+                execution_context=SimpleNamespace(coordinator=coordinator, child_run_id="child"),
+                current_session=child, _workspace_access={}, _get_workspace_root=lambda: root,
+                _is_state_changing_action=lambda *_args: False,
+            )
+
+            with self.assertRaisesRegex(ValueError, "outside its assigned scope"):
+                with _delegation_tool_access(agent, "read_file", {"path": "private.txt"}):
+                    self.fail("out-of-scope file read entered the guarded operation")
+
+    def test_subagent_directory_enumeration_is_limited_to_assigned_scope(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory).resolve()
+            coordinator = SimpleNamespace(
+                root=root, access=None, check_cancelled=lambda _run: None,
+                runs={"child": {"assignment": {"scope": ["src/allowed.py"]}}},
+                cancelled={"child": threading.Event()},
+            )
+            adapters = SimpleNamespace(_resolve_workspace_path=lambda raw: (root / raw).resolve())
+            child = SimpleNamespace(workspace=SimpleNamespace(adapters=adapters), _delegation_parent=SimpleNamespace(
+                interaction=SimpleNamespace(state=lambda: {})))
+            agent = SimpleNamespace(
+                execution_context=SimpleNamespace(coordinator=coordinator, child_run_id="child"),
+                current_session=child, _workspace_access={}, _get_workspace_root=lambda: root,
+                _is_state_changing_action=lambda *_args: False,
+            )
+            with self.assertRaisesRegex(ValueError, "outside its assigned scope"):
+                with _delegation_tool_access(agent, "list_tree", {"path": "."}):
+                    self.fail("out-of-scope directory enumeration entered the guarded operation")
+
+    def test_legacy_find_files_checks_its_directory_against_assigned_scope(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory).resolve()
+            coordinator = SimpleNamespace(
+                root=root, access=None, check_cancelled=lambda _run: None,
+                runs={"child": {"assignment": {"scope": ["src"]}}},
+                cancelled={"child": threading.Event()},
+            )
+            adapters = SimpleNamespace(_resolve_workspace_path=lambda raw: (root / raw).resolve())
+            child = SimpleNamespace(workspace=SimpleNamespace(adapters=adapters), _delegation_parent=SimpleNamespace(
+                interaction=SimpleNamespace(state=lambda: {})))
+            agent = SimpleNamespace(
+                execution_context=SimpleNamespace(coordinator=coordinator, child_run_id="child"),
+                current_session=child, _workspace_access={}, _get_workspace_root=lambda: root,
+                _is_state_changing_action=lambda *_args: False,
+            )
+
+            with _delegation_tool_access(agent, "find_files", "*.py|src"):
+                pass
+
+    def test_reviewer_can_read_declared_artifact_outside_worker_scope(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory).resolve()
+            coordinator = SimpleNamespace(
+                root=root, access=None, check_cancelled=lambda _run: None,
+                runs={"child": {"assignment": {"scope": ["src/worker.py"]}}},
+                cancelled={"child": threading.Event()},
+            )
+            adapters = SimpleNamespace(_resolve_workspace_path=lambda raw: str((root / raw).resolve()))
+            child = SimpleNamespace(
+                workspace=SimpleNamespace(adapters=adapters),
+                _delegation_parent=SimpleNamespace(interaction=SimpleNamespace(state=lambda: {})),
+                _delegation_review_artifacts=["dist/result.txt"],
+            )
+            agent = SimpleNamespace(
+                execution_context=SimpleNamespace(coordinator=coordinator, child_run_id="child"),
+                current_session=child, _workspace_access={}, _get_workspace_root=lambda: root,
+                _is_state_changing_action=lambda *_args: False,
+            )
+
+            with _delegation_tool_access(agent, "read_file", {"path": "dist/result.txt"}):
+                pass
+
     def test_delegated_write_pauses_while_parent_revision_awaits_approval(self):
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory).resolve()
@@ -181,7 +377,8 @@ def acquire(self, path, cancelled, owner, *, reading=False, recursive=False):
                     yield
 
             coordinator = SimpleNamespace(access=Access(), check_cancelled=lambda _run: None,
-                                          cancelled={"child": threading.Event()})
+                                          cancelled={"child": threading.Event()}, root=root,
+                                          runs={"child": {"assignment": {"scope": ["."]}}})
             adapters = SimpleNamespace(_resolve_workspace_path=lambda raw: (root / raw).resolve())
             agent = SimpleNamespace(
                 execution_context=SimpleNamespace(coordinator=coordinator, child_run_id="child"),
@@ -206,7 +403,8 @@ def acquire(self, path, cancelled, owner, *, reading=False, recursive=False):
                     yield
 
             coordinator = SimpleNamespace(access=Access(), check_cancelled=lambda _run: None,
-                                          cancelled={"child": threading.Event()})
+                                          cancelled={"child": threading.Event()}, root=root,
+                                          runs={"child": {"assignment": {"scope": ["docs"]}}})
             adapters = SimpleNamespace(_resolve_workspace_path=lambda raw: (root / raw).resolve())
             agent = SimpleNamespace(
                 execution_context=SimpleNamespace(coordinator=coordinator, child_run_id="child"),