补齐 recursive 完整测试与模式隔离说明

cyber_agent/docs/tools.md

@@ -906,7 +906,8 @@ RunConfig(tools=["knowledge_search", "read_file"]) # 精确指定（优先于 to
 ```python
 @tool(description="创建子 Agent 执行任务", groups=["core"])
 async def agent(
-    task: Union[str, List[str]],
+    task: Optional[Union[str, List[str]]] = None,
+    task_brief: Optional[Union[TaskBrief, List[TaskBrief]]] = None,
     messages: Optional[Union[Messages, List[Messages]]] = None,
     continue_from: Optional[str] = None,
     agent_type: Optional[str] = None,
@@ -941,10 +942,10 @@ async def agent(
 
 | task 类型 | 模式 | 并行执行 | 工具权限 |
 |-----------|------|---------|---------|
-| `str`（单任务） | delegate | ❌ | 按 Trace 持久化的 `AGENT_MODE` 和深度授权 |
-| `List[str]`（多任务） | explore | ✅ | 与单任务使用同一模式策略 |
+| 单任务 | delegate | ❌ | 按 Trace 持久化的 `AGENT_MODE` 和深度授权 |
+| 多任务 | explore | Legacy 并行；Recursive 默认串行 | 与单任务使用同一模式策略 |
 
-**messages 参数**：
+**messages 参数**（Legacy 和 `remote_*` 保留）：
 - `None`：无预置消息
 - `Messages`（1D 列表）：所有 agent 共享
 - `List[Messages]`（2D 列表）：per-agent 独立消息
@@ -952,7 +953,10 @@ async def agent(
 运行时判断：`messages[0]` 是 dict → 1D 共享；是 list → 2D per-agent。
 
 **单任务（delegate）**：适合委托专门任务，完整工具权限；支持 `continue_from` 续跑已有 Sub-Trace。
-**多任务（explore）**：适合对比多个方案，并行执行，不支持 `continue_from`。
+**多任务（explore）**：适合对比多个方案，不支持 `continue_from`。
+Legacy 保留整批并行；Recursive 由 `RunConfig.child_execution_mode`
+选择 `sequential/parallel`，默认串行，显式并行时
+`max_parallel_children` 最大为 2。
 
 #### Legacy / Recursive 模式
 
@@ -984,14 +988,20 @@ Trace 会直接报错，必须迁移到 `AGENT_MODE`。
 
 新建的 `recursive` Trace 使用结构化的父子任务协议：
 
-- 本地 `agent()` 用 `task_brief` 下发 `objective`、
-  `completion_criteria`、可选 `context/constraints`；`remote_*` 仍使用 `task`。
+- 本地 `agent()` 用 `task_brief` 下发必填的 `objective`、
+  `reason`、`completion_criteria`、`expected_outputs`，以及可选的
+  `parent_findings/context/constraints`；父级硬约束会自动继承。本地 Recursive
+  不接受 `messages`，`remote_*` 仍使用原有 `task/messages`。
 - 非根 Sub-Agent 结束前必须调用 `submit_task_report`，提交带
   `Validation`、产出、证据和下一步建议的 `TaskReport`。
-- 子报告返回后，父 Goal 从 `waiting_children` 进入 `pending_review`；
+- 子报告返回后，框架先创建无工具的独立 Validator Trace，读取
+  持久化主路径和真实 Tool Result，生成权威 `ValidationResult`。然后父
+  Goal 才从 `waiting_children` 进入 `pending_review`；
   父 Agent 必须用 `review_task_result` 逐份生成 `TaskReview`。
 - 审核决策为 `ACCEPT_REFINE / REJECT_REFINE / REVISE_CHILD /
-  DESCEND_AGAIN / ASCEND / FAIL`。新孩子只能使用审核批准的
+  DESCEND_AGAIN / REPLAN_CURRENT / ASCEND / FAIL`。`REPLAN_CURRENT`
+  只能恢复正在审核孩子的直接父 Trace 和原 Goal，不能指定祖父或兄弟。
+  新孩子只能使用审核批准的
   `TaskBrief`；`REVISE_CHILD` 只能 `continue_from` 原直接孩子。
 - 已批准的 next action 执行完前不能提交父级报告或
   `ASCEND/FAIL`；批准新孩子前会同时校验现有孩子和已排队名额。
@@ -1000,6 +1010,32 @@ Trace 会直接报错，必须迁移到 `AGENT_MODE`。
 - 该协议权威状态保存在 `Trace.context.task_protocol`，可在续跑和服务
   重启后恢复；回溯到审核之前时会恢复原报告和 `pending_review`。
   Legacy 和已有 Recursive revision 1 Trace 不进入该协议。
+- 新建 Recursive 根任务必须在 `RunConfig.root_completion_criteria`
+  （REST 为 `root_completion_criteria`）显式给出完成标准。根 Agent 候选答案只有
+  通过根 Validator 才能 `completed`；首次未通过允许修正一次，第二次仍未通过则
+  根 Trace 为 `failed`。
+- 新 Recursive Trace 还会固化一份树级资源预算；旧实验 Trace 若缺少根
+  标准或预算快照，续跑时会明确要求重新创建，不做隐式升级。
+
+##### Recursive 树级资源预算
+
+```env
+AGENT_VALIDATOR_MODEL=                 # 留空则继承被验收 Agent 模型
+AGENT_RESOURCE_BUDGET_ENABLED=true
+AGENT_MAX_TOTAL_AGENTS=50
+AGENT_MAX_LLM_CALLS=150
+AGENT_MAX_TOTAL_TOKENS=1500000
+AGENT_MAX_TOTAL_COST_USD=15
+AGENT_MAX_DURATION_SECONDS=3600
+AGENT_RESERVED_FINAL_CALLS=1
+```
+
+预算只在新建 Recursive 根 Trace 时读取并固化；Legacy 不解析这些变量。
+`total_agents` 计根及本地业务子孙，Validator 不计 Agent 数但计模型调用、
+Token、成本和耗时。`remote_*` 第一版不计入本地树预算；工具内部模型只在
+返回 `tool_usage` 时才能事后计入。单个子 Agent 批次显式并行时最多同时运行
+两个孩子。Token/成本是响应后统计，因此已在途的调用可能带来小幅超限；
+本轮不提供跨 worker 的全局硬并发上限。
 
 #### 远端模式
 

tests/test_agent_mode.py

@@ -42,7 +42,10 @@ class AgentModePolicyTest(unittest.IsolatedAsyncioTestCase):
             os.environ["AGENT_MODE"] = mode
         return await self.runner._prepare_trace(
             [{"role": "user", "content": "test"}],
-            RunConfig(name="mode test"),
+            RunConfig(
+                name="mode test",
+                root_completion_criteria=["return a checked answer"],
+            ),
         )
 
     async def test_unset_defaults_to_legacy(self):
@@ -57,6 +60,20 @@ class AgentModePolicyTest(unittest.IsolatedAsyncioTestCase):
         os.environ["AGENT_MODE"] = "recursive"
         self.assertIs(AgentMode.RECURSIVE, policy_from_environment().mode)
 
+    async def test_legacy_ignores_invalid_recursive_budget_environment(self):
+        with patch.dict(
+            os.environ,
+            {
+                "AGENT_MODE": "legacy",
+                "AGENT_MAX_TOTAL_AGENTS": "not-a-number",
+                "AGENT_RESOURCE_BUDGET_ENABLED": "not-a-boolean",
+            },
+            clear=False,
+        ):
+            trace, _, _ = await self._prepare_new("legacy")
+        self.assertEqual("legacy", trace.context["agent_mode"])
+        self.assertNotIn("resource_budget", trace.context)
+
     async def test_invalid_mode_does_not_create_trace(self):
         with self.assertRaisesRegex(ValueError, "Invalid AGENT_MODE"):
             await self._prepare_new("unsupported")

tests/test_recursive_runtime_extensions.py

@@ -0,0 +1,907 @@
+import asyncio
+import json
+import os
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from cyber_agent.core.agent_mode import RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY
+from cyber_agent.core.resource_budget import (
+    RESOURCE_BUDGET_CONTEXT_KEY,
+    ResourceBudget,
+    ResourceBudgetController,
+)
+from cyber_agent.core.runner import AgentRunner, RunConfig
+from cyber_agent.core.task_protocol import ensure_task_protocol, new_task_protocol
+from cyber_agent.tools import get_tool_registry
+from cyber_agent.tools.builtin.knowledge import KnowledgeConfig
+from cyber_agent.tools.builtin.subagent import agent
+from cyber_agent.tools.registry import ToolRegistry
+from cyber_agent.trace.models import Trace
+from cyber_agent.trace.store import FileSystemTraceStore
+
+
+ROOT_CRITERIA = ["return a checked final answer"]
+CHILD_BRIEF = {
+    "objective": "check one concrete fact",
+    "reason": "the root needs independently checked evidence",
+    "completion_criteria": ["return one supported conclusion"],
+    "expected_outputs": ["one checked conclusion"],
+    "constraints": ["do not guess"],
+}
+
+
+def knowledge_disabled():
+    return KnowledgeConfig(
+        enable_extraction=False,
+        enable_completion_extraction=False,
+        enable_injection=False,
+    )
+
+
+def recursive_env(**overrides):
+    values = {
+        "AGENT_MODE": "recursive",
+        "AGENT_RESOURCE_BUDGET_ENABLED": "true",
+        "AGENT_MAX_TOTAL_AGENTS": "50",
+        "AGENT_MAX_LLM_CALLS": "150",
+        "AGENT_MAX_TOTAL_TOKENS": "1500000",
+        "AGENT_MAX_TOTAL_COST_USD": "15",
+        "AGENT_MAX_DURATION_SECONDS": "3600",
+        "AGENT_RESERVED_FINAL_CALLS": "1",
+    }
+    values.update(overrides)
+    return patch.dict(os.environ, values, clear=False)
+
+
+def tool_names(schemas):
+    return {
+        schema["function"]["name"]
+        for schema in schemas or []
+    }
+
+
+def is_validator_call(messages):
+    return any(
+        message.get("role") == "system"
+        and "independent validator" in str(message.get("content", ""))
+        for message in messages
+    )
+
+
+def validator_response(messages, outcome="passed"):
+    packet = json.loads(messages[-1]["content"])
+    scope = packet["validation_scope"]
+    passed = outcome == "passed"
+    return {
+        "content": json.dumps({
+            "outcome": outcome,
+            "scope": scope,
+            "reason": "the persisted record was checked",
+            "issues": [] if passed else ["the answer lacks required support"],
+            "retry_from": None if passed else "evidence",
+        }),
+        "tool_calls": [],
+        "finish_reason": "stop",
+        "prompt_tokens": 3,
+        "completion_tokens": 2,
+        "cost": 0.001,
+    }
+
+
+class RootValidationIntegrationTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.store = FileSystemTraceStore(self.temp_dir.name)
+
+    async def asyncTearDown(self):
+        self.temp_dir.cleanup()
+
+    def config(self, **overrides):
+        values = {
+            "tools": [],
+            "tool_groups": [],
+            "enable_research_flow": False,
+            "root_completion_criteria": ROOT_CRITERIA,
+            "knowledge": knowledge_disabled(),
+        }
+        values.update(overrides)
+        return RunConfig(**values)
+
+    async def test_root_completes_only_after_independent_validator_passes(self):
+        calls = []
+
+        async def llm_call(**kwargs):
+            calls.append(kwargs)
+            if is_validator_call(kwargs["messages"]):
+                return validator_response(kwargs["messages"])
+            return {
+                "content": "checked final answer",
+                "tool_calls": [],
+                "finish_reason": "stop",
+                "prompt_tokens": 5,
+                "completion_tokens": 4,
+                "cost": 0.002,
+            }
+
+        runner = AgentRunner(trace_store=self.store, llm_call=llm_call)
+        with recursive_env():
+            result = await runner.run_result(
+                messages=[{"role": "user", "content": "answer carefully"}],
+                config=self.config(),
+            )
+
+        self.assertEqual("completed", result["status"])
+        root = await self.store.get_trace(result["trace_id"])
+        state = ensure_task_protocol(root.context)
+        self.assertTrue(state["root_validation_passed"])
+        self.assertEqual(1, state["root_validation_attempts"])
+        validators = await self.store.list_traces(
+            parent_trace_id=root.trace_id,
+            created_by_tool="validator",
+            limit=10,
+        )
+        self.assertEqual(1, len(validators))
+        self.assertEqual(root.trace_id, validators[0].context["root_trace_id"])
+        self.assertEqual([], validators[0].tools)
+        usage = await ResourceBudgetController(self.store).get_usage(root.trace_id)
+        self.assertEqual(1, usage.total_agents)
+        self.assertEqual(2, usage.llm_calls)
+        self.assertEqual(14, usage.total_tokens)
+        self.assertEqual(2, len(calls))
+
+    async def test_root_gets_one_revision_then_fails_on_second_rejection(self):
+        ordinary_calls = 0
+        validation_calls = 0
+
+        async def llm_call(**kwargs):
+            nonlocal ordinary_calls, validation_calls
+            if is_validator_call(kwargs["messages"]):
+                validation_calls += 1
+                return validator_response(kwargs["messages"], outcome="failed")
+            ordinary_calls += 1
+            return {
+                "content": f"candidate-{ordinary_calls}",
+                "tool_calls": [],
+                "finish_reason": "stop",
+            }
+
+        runner = AgentRunner(trace_store=self.store, llm_call=llm_call)
+        with recursive_env():
+            result = await runner.run_result(
+                messages=[{"role": "user", "content": "answer carefully"}],
+                config=self.config(),
+            )
+
+        self.assertEqual("failed", result["status"])
+        self.assertEqual(2, ordinary_calls)
+        self.assertEqual(2, validation_calls)
+        root = await self.store.get_trace(result["trace_id"])
+        state = ensure_task_protocol(root.context)
+        self.assertEqual(2, state["root_validation_attempts"])
+        self.assertEqual(
+            ["failed", "failed"],
+            [item["outcome"] for item in state["root_validation_history"]],
+        )
+        self.assertFalse(state["root_validation_passed"])
+
+    async def test_root_can_complete_after_one_validator_requested_revision(self):
+        ordinary_calls = 0
+        validation_outcomes = iter(["failed", "passed"])
+        second_candidate_messages = None
+
+        async def llm_call(**kwargs):
+            nonlocal ordinary_calls, second_candidate_messages
+            if is_validator_call(kwargs["messages"]):
+                return validator_response(
+                    kwargs["messages"],
+                    outcome=next(validation_outcomes),
+                )
+            ordinary_calls += 1
+            if ordinary_calls == 2:
+                second_candidate_messages = kwargs["messages"]
+            return {
+                "content": f"candidate-{ordinary_calls}",
+                "tool_calls": [],
+                "finish_reason": "stop",
+            }
+
+        runner = AgentRunner(trace_store=self.store, llm_call=llm_call)
+        with recursive_env():
+            result = await runner.run_result(
+                messages=[{"role": "user", "content": "revise if needed"}],
+                config=self.config(),
+            )
+
+        self.assertEqual("completed", result["status"])
+        root = await self.store.get_trace(result["trace_id"])
+        state = ensure_task_protocol(root.context)
+        self.assertEqual(2, state["root_validation_attempts"])
+        self.assertEqual(
+            ["failed", "passed"],
+            [item["outcome"] for item in state["root_validation_history"]],
+        )
+        self.assertTrue(any(
+            message.get("role") == "user"
+            and "framework-owned ValidationResult" in str(message.get("content"))
+            for message in second_candidate_messages
+        ))
+        persisted_messages = await self.store.get_trace_messages(root.trace_id)
+        self.assertTrue(any(
+            message.role == "user"
+            and "framework-owned ValidationResult" in str(message.content)
+            for message in persisted_messages
+        ))
+
+        calls_after_completion = 0
+
+        async def must_not_call(**_kwargs):
+            nonlocal calls_after_completion
+            calls_after_completion += 1
+            raise AssertionError("a third validation attempt must not start")
+
+        resumed = AgentRunner(trace_store=self.store, llm_call=must_not_call)
+        with recursive_env():
+            with self.assertRaisesRegex(ValueError, "two independent"):
+                await resumed.run_result(
+                    [],
+                    self.config(trace_id=root.trace_id),
+                )
+        self.assertEqual(0, calls_after_completion)
+
+    async def test_root_validator_reads_current_tool_result_and_candidate(self):
+        registry = ToolRegistry()
+
+        async def evidence_tool():
+            return {"verified": "persisted-tool-result"}
+
+        registry.register(evidence_tool, groups=["test"])
+        ordinary_calls = 0
+        validator_packet = None
+
+        async def llm_call(**kwargs):
+            nonlocal ordinary_calls, validator_packet
+            if is_validator_call(kwargs["messages"]):
+                validator_packet = json.loads(kwargs["messages"][-1]["content"])
+                return validator_response(kwargs["messages"])
+            ordinary_calls += 1
+            if ordinary_calls == 1:
+                return {
+                    "content": "",
+                    "tool_calls": [{
+                        "id": "evidence-call",
+                        "type": "function",
+                        "function": {
+                            "name": "evidence_tool",
+                            "arguments": "{}",
+                        },
+                    }],
+                    "finish_reason": "tool_calls",
+                }
+            return {
+                "content": "candidate-from-real-tool-result",
+                "tool_calls": [],
+                "finish_reason": "stop",
+            }
+
+        runner = AgentRunner(
+            trace_store=self.store,
+            tool_registry=registry,
+            llm_call=llm_call,
+        )
+        with recursive_env():
+            result = await runner.run_result(
+                [{"role": "user", "content": "use the evidence tool"}],
+                self.config(tools=["evidence_tool"]),
+            )
+
+        self.assertEqual("completed", result["status"])
+        trajectory = validator_packet["trajectory"]
+        self.assertTrue(any(
+            item.get("role") == "tool"
+            and "persisted-tool-result" in str(item.get("content"))
+            for item in trajectory
+        ))
+        self.assertTrue(any(
+            item.get("role") == "assistant"
+            and "candidate-from-real-tool-result" in str(item.get("content"))
+            for item in trajectory
+        ))
+
+    async def test_new_recursive_root_requires_explicit_completion_criteria(self):
+        async def llm_call(**_kwargs):
+            raise AssertionError("LLM must not be called")
+
+        runner = AgentRunner(trace_store=self.store, llm_call=llm_call)
+        with recursive_env():
+            with self.assertRaisesRegex(ValueError, "root_completion_criteria"):
+                await runner.run_result(
+                    messages=[{"role": "user", "content": "missing criteria"}],
+                    config=self.config(root_completion_criteria=[]),
+                )
+        self.assertEqual([], await self.store.list_traces(limit=10))
+
+    async def test_old_recursive_revision_two_trace_is_not_silently_migrated(self):
+        trace_id = "old-recursive-experiment"
+        await self.store.create_trace(Trace(
+            trace_id=trace_id,
+            mode="agent",
+            context={
+                "agent_mode": "recursive",
+                "agent_mode_revision": 2,
+                "agent_depth": 0,
+                "root_trace_id": trace_id,
+                "task_protocol": {},
+            },
+        ))
+        runner = AgentRunner(trace_store=self.store, llm_call=lambda **_: None)
+        with self.assertRaisesRegex(ValueError, "create a new trace"):
+            await runner.run_result([], RunConfig(trace_id=trace_id))
+
+    async def test_validator_trace_cannot_be_continued_as_an_agent(self):
+        validator_id = "root@validator-fixture"
+        await self.store.create_trace(Trace(
+            trace_id=validator_id,
+            mode="agent",
+            agent_type="validator",
+            tools=[],
+            status="completed",
+            context={
+                "created_by_tool": "validator",
+                "agent_mode": "recursive",
+                "agent_mode_revision": 2,
+                "root_trace_id": "root",
+            },
+        ))
+        calls = 0
+
+        async def llm_call(**_kwargs):
+            nonlocal calls
+            calls += 1
+            return {"content": "must not run", "tool_calls": []}
+
+        runner = AgentRunner(trace_store=self.store, llm_call=llm_call)
+        with self.assertRaisesRegex(ValueError, "cannot be continued"):
+            await runner.run_result(
+                [{"role": "user", "content": "gain tools"}],
+                RunConfig(trace_id=validator_id),
+            )
+        self.assertEqual(0, calls)
+        validator = await self.store.get_trace(validator_id)
+        self.assertEqual("completed", validator.status)
+        self.assertEqual([], validator.tools)
+
+
+class RecursiveBudgetIntegrationTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.store = FileSystemTraceStore(self.temp_dir.name)
+
+    async def asyncTearDown(self):
+        self.temp_dir.cleanup()
+
+    async def test_response_over_token_budget_never_executes_its_tool_call(self):
+        dangerous_calls = 0
+        registry = ToolRegistry()
+
+        async def dangerous():
+            nonlocal dangerous_calls
+            dangerous_calls += 1
+            return "must not run"
+
+        registry.register(dangerous, groups=["test"])
+
+        async def llm_call(**_kwargs):
+            return {
+                "content": "over-budget response",
+                "tool_calls": [{
+                    "id": "dangerous-call",
+                    "type": "function",
+                    "function": {"name": "dangerous", "arguments": "{}"},
+                }],
+                "finish_reason": "tool_calls",
+                "prompt_tokens": 8,
+                "completion_tokens": 8,
+            }
+
+        runner = AgentRunner(
+            trace_store=self.store,
+            tool_registry=registry,
+            llm_call=llm_call,
+        )
+        config = RunConfig(
+            tools=["dangerous"],
+            tool_groups=[],
+            enable_research_flow=False,
+            root_completion_criteria=ROOT_CRITERIA,
+            knowledge=knowledge_disabled(),
+        )
+        with recursive_env(AGENT_MAX_TOTAL_TOKENS="10"):
+            result = await runner.run_result(
+                [{"role": "user", "content": "do not overspend"}],
+                config,
+            )
+
+        self.assertEqual("failed", result["status"])
+        self.assertEqual(0, dangerous_calls)
+        root = await self.store.get_trace(result["trace_id"])
+        self.assertEqual("budget_exhausted:tokens", root.context["termination_reason"])
+        usage = await ResourceBudgetController(self.store).get_usage(root.trace_id)
+        self.assertEqual(16, usage.total_tokens)
+        self.assertEqual("budget_exhausted:tokens", usage.exhausted_reason)
+        assistant = [
+            message
+            for message in await self.store.get_trace_messages(root.trace_id)
+            if message.role == "assistant"
+        ]
+        self.assertIsNone(assistant[-1].content["tool_calls"])
+
+    async def test_agent_budget_denial_blocks_child_but_preserves_root_validator(self):
+        ordinary_calls = 0
+
+        async def llm_call(**kwargs):
+            nonlocal ordinary_calls
+            if is_validator_call(kwargs["messages"]):
+                return validator_response(kwargs["messages"])
+            ordinary_calls += 1
+            if ordinary_calls == 1:
+                return {
+                    "content": "",
+                    "tool_calls": [{
+                        "id": "delegate",
+                        "type": "function",
+                        "function": {
+                            "name": "agent",
+                            "arguments": json.dumps({"task_brief": CHILD_BRIEF}),
+                        },
+                    }],
+                    "finish_reason": "tool_calls",
+                }
+            return {
+                "content": "root can still finish without a child",
+                "tool_calls": [],
+                "finish_reason": "stop",
+            }
+
+        runner = AgentRunner(
+            trace_store=self.store,
+            tool_registry=get_tool_registry(),
+            llm_call=llm_call,
+        )
+        config = RunConfig(
+            tools=["agent"],
+            tool_groups=[],
+            enable_research_flow=False,
+            root_completion_criteria=ROOT_CRITERIA,
+            knowledge=knowledge_disabled(),
+        )
+        with recursive_env(AGENT_MAX_TOTAL_AGENTS="1"):
+            result = await runner.run_result(
+                [{"role": "user", "content": "try one delegated check"}],
+                config,
+            )
+
+        self.assertEqual("completed", result["status"])
+        root = await self.store.get_trace(result["trace_id"])
+        children = await self.store.list_traces(
+            parent_trace_id=root.trace_id,
+            created_by_tool="agent",
+            limit=10,
+        )
+        self.assertEqual([], children)
+        usage = await ResourceBudgetController(self.store).get_usage(root.trace_id)
+        self.assertEqual(1, usage.total_agents)
+        self.assertEqual("budget_exhausted:agents", usage.exhausted_reason)
+        self.assertTrue(ensure_task_protocol(root.context)["root_validation_passed"])
+
+    async def test_persisted_child_is_not_refunded_when_goal_write_fails(self):
+        root_id = "partial-child-create-root"
+        budget = ResourceBudget()
+        await self.store.create_trace(Trace(
+            trace_id=root_id,
+            mode="agent",
+            uid="user-1",
+            model="fake",
+            context={
+                "agent_mode": "recursive",
+                "agent_mode_revision": 2,
+                "agent_depth": 0,
+                "root_trace_id": root_id,
+                "root_completion_criteria": ROOT_CRITERIA,
+                RESOURCE_BUDGET_CONTEXT_KEY: budget.to_dict(),
+                "task_protocol": new_task_protocol(),
+            },
+        ))
+        await ResourceBudgetController(self.store).initialize(root_id, budget)
+        original_update_goal_tree = self.store.update_goal_tree
+
+        async def fail_child_goal_write(trace_id, goal_tree):
+            if trace_id != root_id:
+                raise RuntimeError("simulated child GoalTree write failure")
+            return await original_update_goal_tree(trace_id, goal_tree)
+
+        self.store.update_goal_tree = fail_child_goal_write
+        runner = AgentRunner(
+            trace_store=self.store,
+            tool_registry=get_tool_registry(),
+            llm_call=lambda **_: None,
+        )
+
+        with self.assertRaisesRegex(RuntimeError, "GoalTree write failure"):
+            await agent(
+                task_brief=CHILD_BRIEF,
+                context={
+                    "store": self.store,
+                    "trace_id": root_id,
+                    "runner": runner,
+                    RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY: ["agent"],
+                },
+            )
+
+        children = await self.store.list_traces(
+            parent_trace_id=root_id,
+            created_by_tool="agent",
+            limit=10,
+        )
+        self.assertEqual(1, len(children))
+        self.assertEqual("failed", children[0].status)
+        self.assertIn("Child initialization failed", children[0].error_message)
+        usage = await ResourceBudgetController(self.store).get_usage(root_id)
+        self.assertEqual(2, usage.total_agents)
+
+    async def test_parent_goal_projection_failure_closes_precreated_child(self):
+        root_id = "parent-goal-projection-root"
+        budget = ResourceBudget()
+        await self.store.create_trace(Trace(
+            trace_id=root_id,
+            mode="agent",
+            uid="user-1",
+            model="fake",
+            context={
+                "agent_mode": "recursive",
+                "agent_mode_revision": 2,
+                "agent_depth": 0,
+                "root_trace_id": root_id,
+                "root_completion_criteria": ROOT_CRITERIA,
+                RESOURCE_BUDGET_CONTEXT_KEY: budget.to_dict(),
+                "task_protocol": new_task_protocol(),
+            },
+        ))
+        await ResourceBudgetController(self.store).initialize(root_id, budget)
+        runner = AgentRunner(
+            trace_store=self.store,
+            tool_registry=get_tool_registry(),
+            llm_call=lambda **_: None,
+        )
+
+        async def fail_parent_projection(*_args, **_kwargs):
+            raise RuntimeError("simulated parent Goal projection failure")
+
+        with patch(
+            "cyber_agent.tools.builtin.subagent._update_goal_start",
+            new=fail_parent_projection,
+        ):
+            with self.assertRaisesRegex(RuntimeError, "projection failure"):
+                await agent(
+                    task_brief=CHILD_BRIEF,
+                    context={
+                        "store": self.store,
+                        "trace_id": root_id,
+                        "runner": runner,
+                        RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY: ["agent"],
+                    },
+                )
+
+        children = await self.store.list_traces(
+            parent_trace_id=root_id,
+            created_by_tool="agent",
+            limit=10,
+        )
+        self.assertEqual(1, len(children))
+        self.assertEqual("failed", children[0].status)
+        usage = await ResourceBudgetController(self.store).get_usage(root_id)
+        self.assertEqual(2, usage.total_agents)
+
+    async def test_approved_action_write_failure_closes_precreated_child(self):
+        root_id = "approved-action-write-root"
+        budget = ResourceBudget()
+        context = {
+            "agent_mode": "recursive",
+            "agent_mode_revision": 2,
+            "agent_depth": 0,
+            "root_trace_id": root_id,
+            "root_completion_criteria": ROOT_CRITERIA,
+            RESOURCE_BUDGET_CONTEXT_KEY: budget.to_dict(),
+            "task_protocol": new_task_protocol(),
+        }
+        state = ensure_task_protocol(context)
+        state["next_actions"] = [{
+            "decision": "ACCEPT_REFINE",
+            "child_trace_id": "reviewed-child",
+            "goal_id": None,
+            "task_brief": CHILD_BRIEF,
+            "created_at_sequence": 1,
+        }]
+        await self.store.create_trace(Trace(
+            trace_id=root_id,
+            mode="agent",
+            uid="user-1",
+            model="fake",
+            context=context,
+        ))
+        await ResourceBudgetController(self.store).initialize(root_id, budget)
+        runner = AgentRunner(
+            trace_store=self.store,
+            tool_registry=get_tool_registry(),
+            llm_call=lambda **_: None,
+        )
+        original_update_trace = self.store.update_trace
+        rejected_parent_write = False
+
+        async def fail_approved_action_write(trace_id, **updates):
+            nonlocal rejected_parent_write
+            if trace_id == root_id and "context" in updates and not rejected_parent_write:
+                rejected_parent_write = True
+                raise RuntimeError("simulated approved action write failure")
+            return await original_update_trace(trace_id, **updates)
+
+        self.store.update_trace = fail_approved_action_write
+        with self.assertRaisesRegex(RuntimeError, "approved action write failure"):
+            await agent(
+                task_brief=CHILD_BRIEF,
+                context={
+                    "store": self.store,
+                    "trace_id": root_id,
+                    "runner": runner,
+                    RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY: ["agent"],
+                },
+            )
+
+        children = await self.store.list_traces(
+            parent_trace_id=root_id,
+            created_by_tool="agent",
+            limit=10,
+        )
+        self.assertEqual(1, len(children))
+        self.assertEqual("failed", children[0].status)
+        usage = await ResourceBudgetController(self.store).get_usage(root_id)
+        self.assertEqual(2, usage.total_agents)
+
+
+class RecursiveLifecycleIntegrationTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.store = FileSystemTraceStore(self.temp_dir.name)
+
+    async def asyncTearDown(self):
+        self.temp_dir.cleanup()
+
+    async def test_child_report_is_validated_before_parent_review(self):
+        child_submitted = False
+        root_delegated = False
+        root_reviewed = False
+
+        async def llm_call(**kwargs):
+            nonlocal child_submitted, root_delegated, root_reviewed
+            messages = kwargs["messages"]
+            if is_validator_call(messages):
+                return validator_response(messages)
+            names = tool_names(kwargs.get("tools"))
+            user_text = "\n".join(
+                str(message.get("content", ""))
+                for message in messages
+                if message.get("role") == "user"
+            )
+            if "submit_task_report" in names and not child_submitted:
+                child_submitted = True
+                return {
+                    "content": "",
+                    "tool_calls": [{
+                        "id": "submit-report",
+                        "type": "function",
+                        "function": {
+                            "name": "submit_task_report",
+                            "arguments": json.dumps({
+                                "task_report": {
+                                    "summary": "one fact was checked",
+                                    "outcome": "satisfied",
+                                    "validation": {
+                                        "hard_passed": True,
+                                        "open_issues": [],
+                                    },
+                                    "next_step_suggestion": {
+                                        "direction": "NONE",
+                                        "reason": "the delegated check is complete",
+                                    },
+                                    "outputs": [{"kind": "answer", "value": "checked"}],
+                                    "evidence": [{"kind": "tool", "value": "persisted"}],
+                                    "remaining_issues": [],
+                                }
+                            }),
+                        },
+                    }],
+                    "finish_reason": "tool_calls",
+                }
+            if "# Task Brief" in user_text:
+                return {
+                    "content": "structured child report submitted",
+                    "tool_calls": [],
+                    "finish_reason": "stop",
+                }
+            if names == {"review_task_result"}:
+                roots = [
+                    trace
+                    for trace in await self.store.list_traces(limit=20)
+                    if trace.parent_trace_id is None
+                ]
+                root = roots[0]
+                pending = ensure_task_protocol(root.context)["pending_reviews"]
+                child_id = next(iter(pending))
+                root_reviewed = True
+                return {
+                    "content": "",
+                    "tool_calls": [{
+                        "id": "review-report",
+                        "type": "function",
+                        "function": {
+                            "name": "review_task_result",
+                            "arguments": json.dumps({
+                                "child_trace_id": child_id,
+                                "decision": "ASCEND",
+                                "reason": "the validated child satisfies this goal",
+                            }),
+                        },
+                    }],
+                    "finish_reason": "tool_calls",
+                }
+            if "agent" in names and not root_delegated:
+                root_delegated = True
+                return {
+                    "content": "",
+                    "tool_calls": [{
+                        "id": "delegate-child",
+                        "type": "function",
+                        "function": {
+                            "name": "agent",
+                            "arguments": json.dumps({"task_brief": CHILD_BRIEF}),
+                        },
+                    }],
+                    "finish_reason": "tool_calls",
+                }
+            return {
+                "content": "final answer based on the validated child",
+                "tool_calls": [],
+                "finish_reason": "stop",
+            }
+
+        runner = AgentRunner(
+            trace_store=self.store,
+            tool_registry=get_tool_registry(),
+            llm_call=llm_call,
+        )
+        config = RunConfig(
+            tools=["agent", "submit_task_report", "review_task_result"],
+            tool_groups=[],
+            enable_research_flow=False,
+            root_completion_criteria=ROOT_CRITERIA,
+            knowledge=knowledge_disabled(),
+        )
+        with recursive_env():
+            result = await runner.run_result(
+                [{"role": "user", "content": "perform one delegated check"}],
+                config,
+            )
+
+        self.assertEqual("completed", result["status"])
+        self.assertTrue(root_reviewed)
+        root = await self.store.get_trace(result["trace_id"])
+        business_children = await self.store.list_traces(
+            parent_trace_id=root.trace_id,
+            created_by_tool="agent",
+            limit=10,
+        )
+        self.assertEqual(1, len(business_children))
+        child = business_children[0]
+        child_validators = await self.store.list_traces(
+            parent_trace_id=child.trace_id,
+            created_by_tool="validator",
+            limit=10,
+        )
+        root_validators = await self.store.list_traces(
+            parent_trace_id=root.trace_id,
+            created_by_tool="validator",
+            limit=10,
+        )
+        self.assertEqual(1, len(child_validators))
+        self.assertEqual(1, len(root_validators))
+        self.assertEqual(
+            child.trace_id,
+            child_validators[0].context["evaluated_trace_id"],
+        )
+        child_state = ensure_task_protocol(child.context)
+        self.assertEqual(
+            child_validators[0].trace_id,
+            child_state["task_report_validation"]["validation_result"][
+                "validator_trace_id"
+            ],
+        )
+        root_state = ensure_task_protocol(root.context)
+        self.assertEqual({}, root_state["pending_reviews"])
+        self.assertEqual(child.trace_id, root_state["reviews"][0]["child_trace_id"])
+        usage = await ResourceBudgetController(self.store).get_usage(root.trace_id)
+        self.assertEqual(2, usage.total_agents)
+        self.assertEqual(7, usage.llm_calls)
+
+    async def test_parent_stop_reaches_validator_after_child_runner_finished(self):
+        root_id = "validator-cancel-root"
+        child_id = "validator-cancel-child"
+        budget = ResourceBudget()
+        await self.store.create_trace(Trace(
+            trace_id=root_id,
+            mode="agent",
+            model="fake-model",
+            context={
+                "agent_mode": "recursive",
+                "agent_mode_revision": 2,
+                "agent_depth": 0,
+                "root_trace_id": root_id,
+                "root_completion_criteria": ROOT_CRITERIA,
+                RESOURCE_BUDGET_CONTEXT_KEY: budget.to_dict(),
+                "task_protocol": new_task_protocol(),
+            },
+        ))
+        await self.store.create_trace(Trace(
+            trace_id=child_id,
+            mode="agent",
+            model="fake-model",
+            parent_trace_id=root_id,
+            context={
+                "agent_mode": "recursive",
+                "agent_mode_revision": 2,
+                "agent_depth": 1,
+                "root_trace_id": root_id,
+                "created_by_tool": "agent",
+                "task_protocol": new_task_protocol(CHILD_BRIEF),
+            },
+        ))
+        await ResourceBudgetController(self.store).initialize(
+            root_id,
+            budget,
+            initial_agents=2,
+        )
+        started = asyncio.Event()
+        release = asyncio.Event()
+
+        async def llm_call(**kwargs):
+            started.set()
+            await release.wait()
+            return validator_response(kwargs["messages"])
+
+        runner = AgentRunner(trace_store=self.store, llm_call=llm_call)
+        root_event = asyncio.Event()
+        runner._cancel_events[root_id] = root_event
+        runner._recursive_active_traces[root_id] = root_event
+        validation_task = asyncio.create_task(runner.validate_recursive_trace(
+            child_id,
+            scope="task",
+            task_brief=CHILD_BRIEF,
+            task_report={"summary": "candidate"},
+        ))
+        await asyncio.wait_for(started.wait(), timeout=1)
+
+        self.assertIn(child_id, runner._active_children[root_id])
+        self.assertTrue(await runner.stop(root_id))
+        release.set()
+        validation_run = await asyncio.wait_for(validation_task, timeout=1)
+
+        self.assertEqual("error", validation_run.result.outcome)
+        validator = await self.store.get_trace(validation_run.trace_id)
+        self.assertEqual("failed", validator.status)
+        self.assertEqual(child_id, validator.parent_trace_id)
+        self.assertEqual(root_id, validator.context["root_trace_id"])
+        self.assertNotIn(child_id, runner._recursive_active_traces)
+        self.assertNotIn(validation_run.trace_id, runner._recursive_active_traces)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_recursive_subtree_cancellation.py

@@ -17,7 +17,9 @@ from cyber_agent.trace.store import FileSystemTraceStore
 def brief(index):
     return {
         "objective": f"child-{index}",
+        "reason": "exercise subtree cancellation",
         "completion_criteria": [f"finish child-{index}"],
+        "expected_outputs": [f"result for child-{index}"],
     }
 
 
@@ -140,6 +142,7 @@ class RecursiveSubtreeCancellationTest(unittest.IsolatedAsyncioTestCase):
                     tools=["dangerous_cancel_test"],
                     tool_groups=[],
                     enable_research_flow=False,
+                    root_completion_criteria=["stopping prevents late tool execution"],
                 ),
             ))
             await asyncio.wait_for(llm_started.wait(), timeout=1)
@@ -203,6 +206,7 @@ class RecursiveSubtreeCancellationTest(unittest.IsolatedAsyncioTestCase):
                     child_execution_mode="parallel",
                     max_parallel_children=2,
                     enable_research_flow=False,
+                    root_completion_criteria=["stopping propagates through the child tree"],
                 ),
             ))
             await asyncio.wait_for(two_children_started.wait(), timeout=2)

tests/test_recursive_tool_capabilities.py

@@ -7,6 +7,7 @@ from unittest.mock import patch
 from cyber_agent.core.agent_mode import RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY
 from cyber_agent.core.runner import AgentRunner, RunConfig
 from cyber_agent.core.task_protocol import ensure_task_protocol
+from cyber_agent.tools.builtin.knowledge import KnowledgeConfig
 from cyber_agent.tools.builtin.subagent import agent
 from cyber_agent.tools.registry import ToolRegistry
 from cyber_agent.trace.goal_models import GoalTree
@@ -238,6 +239,18 @@ class RecursiveRuntimeDispatchTest(unittest.IsolatedAsyncioTestCase):
                 async def fake_llm(**kwargs):
                     nonlocal llm_calls
                     llm_calls += 1
+                    if kwargs.get("tools") == []:
+                        return {
+                            "content": json.dumps({
+                                "outcome": "passed",
+                                "scope": "root",
+                                "reason": "unauthorized tools were not executed",
+                                "issues": [],
+                                "retry_from": None,
+                            }),
+                            "tool_calls": [],
+                            "finish_reason": "stop",
+                        }
                     if llm_calls == 1:
                         return {
                             "content": "",
@@ -263,6 +276,12 @@ class RecursiveRuntimeDispatchTest(unittest.IsolatedAsyncioTestCase):
                             tool_groups=[],
                             parallel_tool_execution=parallel,
                             enable_research_flow=False,
+                            root_completion_criteria=["forged tools are never executed"],
+                            knowledge=KnowledgeConfig(
+                                enable_extraction=False,
+                                enable_completion_extraction=False,
+                                enable_injection=False,
+                            ),
                         ),
                     )
                 self.assertEqual("completed", result["status"])

tests/test_recursive_validation_core.py

@@ -0,0 +1,344 @@
+import json
+import tempfile
+import unittest
+
+from pydantic import ValidationError
+
+from cyber_agent.core.validation import (
+    LLMValidator,
+    ValidationResult,
+    build_validation_packet,
+    parse_validation_result,
+    validation_error,
+)
+from cyber_agent.trace.models import Message, Trace
+from cyber_agent.trace.store import FileSystemTraceStore
+
+
+class FakeLLM:
+    def __init__(self, response=None, error=None):
+        self.response = response
+        self.error = error
+        self.calls = []
+
+    async def __call__(self, **kwargs):
+        self.calls.append(kwargs)
+        if self.error:
+            raise self.error
+        return self.response
+
+
+def valid_response(scope="task"):
+    return {
+        "content": json.dumps({
+            "outcome": "passed",
+            "scope": scope,
+            "reason": "The persisted evidence satisfies every criterion.",
+            "issues": [],
+            "retry_from": None,
+        }),
+        "tool_calls": None,
+        "prompt_tokens": 120,
+        "completion_tokens": 30,
+        "cost": 0.012,
+        "finish_reason": "stop",
+    }
+
+
+class ValidationModelTest(unittest.TestCase):
+    def test_result_enforces_outcome_invariants(self):
+        passed = ValidationResult(
+            validator_trace_id="validator",
+            evaluated_trace_id="child",
+            outcome="passed",
+            scope="task",
+            reason="all criteria passed",
+            issues=[],
+            retry_from=None,
+        )
+        self.assertEqual("passed", passed.outcome)
+
+        invalid = [
+            {"outcome": "passed", "issues": ["unexpected"], "retry_from": None},
+            {"outcome": "failed", "issues": [], "retry_from": "evidence"},
+            {"outcome": "failed", "issues": ["gap"], "retry_from": None},
+            {"outcome": "error", "issues": ["broken"], "retry_from": "output"},
+        ]
+        for fields in invalid:
+            with self.subTest(fields=fields), self.assertRaises(ValidationError):
+                ValidationResult(
+                    validator_trace_id="validator",
+                    evaluated_trace_id="child",
+                    scope="task",
+                    reason="invalid combination",
+                    **fields,
+                )
+
+    def test_parser_injects_ids_and_rejects_model_owned_ids(self):
+        parsed = parse_validation_result(
+            valid_response()["content"],
+            validator_trace_id="validator",
+            evaluated_trace_id="child",
+            expected_scope="task",
+        )
+        self.assertEqual("validator", parsed.validator_trace_id)
+        self.assertEqual("child", parsed.evaluated_trace_id)
+
+        forged = json.loads(valid_response()["content"])
+        forged["evaluated_trace_id"] = "forged"
+        with self.assertRaises(ValidationError):
+            parse_validation_result(
+                json.dumps(forged),
+                validator_trace_id="validator",
+                evaluated_trace_id="child",
+                expected_scope="task",
+            )
+        with self.assertRaisesRegex(ValueError, "expected 'task'"):
+            parse_validation_result(
+                valid_response(scope="root")["content"],
+                validator_trace_id="validator",
+                evaluated_trace_id="child",
+                expected_scope="task",
+            )
+
+    def test_error_result_is_deterministic_and_fail_closed(self):
+        result = validation_error(
+            validator_trace_id="validator",
+            evaluated_trace_id="child",
+            scope="root",
+            reason=" invalid JSON ",
+        )
+        self.assertEqual("error", result.outcome)
+        self.assertEqual(["invalid JSON"], result.issues)
+        self.assertIsNone(result.retry_from)
+
+    def test_packet_keeps_contract_and_newest_main_path_within_limit(self):
+        trajectory = [
+            {
+                "sequence": index,
+                "role": "tool",
+                "name": "read_file",
+                "content": f"marker-{index}-" + ("x" * 300),
+                "reasoning_content": "must-not-leak",
+            }
+            for index in range(1, 8)
+        ]
+        trajectory.insert(6, {
+            "sequence": 99,
+            "role": "assistant",
+            "content": "side-branch-secret",
+            "branch_type": "compression",
+        })
+        packet = build_validation_packet(
+            validation_scope="task",
+            task_brief={
+                "objective": "check result",
+                "completion_criteria": ["criterion-kept"],
+                "expected_outputs": ["output-kept"],
+            },
+            task_report={"summary": "report-kept"},
+            trajectory=trajectory,
+            max_chars=1_000,
+        )
+        self.assertLessEqual(len(packet), 1_000)
+        self.assertIn("criterion-kept", packet)
+        self.assertIn("output-kept", packet)
+        self.assertIn("report-kept", packet)
+        self.assertIn("marker-7", packet)
+        self.assertNotIn("marker-1", packet)
+        self.assertNotIn("reasoning_content", packet)
+        self.assertNotIn("side-branch-secret", packet)
+
+    def test_packet_rejects_contract_that_alone_exceeds_limit(self):
+        with self.assertRaisesRegex(ValueError, "contract exceeds"):
+            build_validation_packet(
+                validation_scope="task",
+                task_brief={"objective": "x" * 1_000},
+                trajectory=[],
+                max_chars=100,
+            )
+
+    def test_packet_removes_reasoning_from_persisted_assistant_message(self):
+        message = Message.create(
+            trace_id="evaluated",
+            role="assistant",
+            sequence=1,
+            content={
+                "text": "observable answer",
+                "reasoning_content": "hidden chain of thought",
+            },
+        )
+        packet = build_validation_packet(
+            validation_scope="task",
+            trajectory=[message],
+        )
+        self.assertIn("observable answer", packet)
+        self.assertNotIn("hidden chain of thought", packet)
+        self.assertNotIn("reasoning_content", packet)
+
+
+class LLMValidatorTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.store = FileSystemTraceStore(self.temp_dir.name)
+        self.evaluated = Trace(
+            trace_id="root@delegate-child",
+            mode="agent",
+            task="child task",
+            uid="user-1",
+            model="fake-model",
+            current_goal_id="1.1",
+            context={
+                "agent_mode": "recursive",
+                "agent_mode_revision": 2,
+                "root_trace_id": "root",
+                "agent_depth": 2,
+            },
+        )
+        await self.store.create_trace(self.evaluated)
+        self.trajectory = [
+            Message.create(
+                trace_id=self.evaluated.trace_id,
+                role="user",
+                sequence=1,
+                content="produce checked evidence",
+            ),
+            Message.create(
+                trace_id=self.evaluated.trace_id,
+                role="tool",
+                sequence=2,
+                parent_sequence=1,
+                content={"tool_name": "read_file", "result": "actual evidence"},
+            ),
+        ]
+
+    async def asyncTearDown(self):
+        self.temp_dir.cleanup()
+
+    async def test_single_tool_free_call_persists_validator_trace_and_usage(self):
+        llm = FakeLLM(valid_response())
+        validator = LLMValidator(llm_call=llm, trace_store=self.store)
+        run = await validator.validate(
+            evaluated_trace=self.evaluated,
+            trajectory=self.trajectory,
+            scope="task",
+            task_brief={
+                "objective": "check result",
+                "completion_criteria": ["must cite evidence"],
+                "expected_outputs": ["one conclusion"],
+            },
+            task_report={"summary": "done", "evidence": [{"value": "actual"}]},
+            validator_trace_id="validator-1",
+        )
+
+        self.assertEqual(1, len(llm.calls))
+        call = llm.calls[0]
+        self.assertEqual([], call["tools"])
+        self.assertEqual(0, call["temperature"])
+        self.assertEqual("fake-model", call["model"])
+        self.assertEqual(2, len(call["messages"]))
+        self.assertEqual("passed", run.result.outcome)
+        self.assertEqual(120, run.prompt_tokens)
+        self.assertEqual(30, run.completion_tokens)
+        self.assertEqual(0.012, run.cost)
+
+        trace = await self.store.get_trace("validator-1")
+        self.assertEqual("completed", trace.status)
+        self.assertEqual("validator", trace.agent_type)
+        self.assertEqual(self.evaluated.trace_id, trace.parent_trace_id)
+        self.assertEqual("validator", trace.context["created_by_tool"])
+        self.assertEqual("root", trace.context["root_trace_id"])
+        self.assertEqual(2, trace.context["agent_depth"])
+        self.assertEqual([], trace.tools)
+        messages = await self.store.get_main_path_messages("validator-1", 3)
+        self.assertEqual(["system", "user", "assistant"], [m.role for m in messages])
+        self.assertIn("actual evidence", messages[1].content)
+
+    async def test_invalid_json_returns_error_without_correction_call(self):
+        response = valid_response()
+        response["content"] = "```json\n{}\n```"
+        llm = FakeLLM(response)
+        validator = LLMValidator(llm_call=llm, trace_store=self.store)
+        run = await validator.validate(
+            evaluated_trace=self.evaluated,
+            trajectory=self.trajectory,
+            scope="task",
+            validator_trace_id="validator-invalid",
+        )
+        self.assertEqual(1, len(llm.calls))
+        self.assertEqual("error", run.result.outcome)
+        trace = await self.store.get_trace(run.trace_id)
+        self.assertEqual("failed", trace.status)
+        self.assertIn("Validator failed", trace.error_message)
+
+    async def test_tool_call_is_rejected_without_dispatch(self):
+        response = valid_response()
+        response["tool_calls"] = [{
+            "id": "forged",
+            "type": "function",
+            "function": {"name": "agent", "arguments": "{}"},
+        }]
+        llm = FakeLLM(response)
+        validator = LLMValidator(llm_call=llm, trace_store=self.store)
+        run = await validator.validate(
+            evaluated_trace=self.evaluated,
+            trajectory=self.trajectory,
+            scope="task",
+        )
+        self.assertEqual("error", run.result.outcome)
+        self.assertIn("attempted to call tools", run.result.reason)
+        self.assertEqual(1, len(llm.calls))
+
+    async def test_model_exception_is_recorded_once(self):
+        llm = FakeLLM(error=RuntimeError("provider unavailable"))
+        validator = LLMValidator(llm_call=llm, trace_store=self.store)
+        run = await validator.validate(
+            evaluated_trace=self.evaluated,
+            trajectory=self.trajectory,
+            scope="root",
+            completion_criteria=["root done"],
+            candidate_output="candidate",
+        )
+        self.assertEqual(1, len(llm.calls))
+        self.assertEqual("error", run.result.outcome)
+        self.assertIn("provider unavailable", run.result.reason)
+
+    async def test_non_success_path_creates_trace_without_calling_llm(self):
+        llm = FakeLLM(valid_response())
+        validator = LLMValidator(llm_call=llm, trace_store=self.store)
+        run = await validator.record_non_success(
+            evaluated_trace=self.evaluated,
+            scope="task",
+            outcome="failed",
+            reason="child stopped",
+            issues=["execution stopped before evidence was produced"],
+            retry_from="evidence",
+            validator_trace_id="validator-stopped",
+        )
+        self.assertEqual([], llm.calls)
+        self.assertEqual("failed", run.result.outcome)
+        trace = await self.store.get_trace(run.trace_id)
+        self.assertEqual("failed", trace.status)
+        self.assertEqual(0, trace.total_tokens)
+        self.assertEqual(1, trace.total_messages)
+
+    async def test_oversized_fixed_input_fails_before_llm(self):
+        llm = FakeLLM(valid_response())
+        validator = LLMValidator(
+            llm_call=llm,
+            trace_store=self.store,
+            max_input_chars=2_000,
+        )
+        run = await validator.validate(
+            evaluated_trace=self.evaluated,
+            trajectory=[],
+            scope="task",
+            task_brief={"objective": "x" * 3_000},
+        )
+        self.assertEqual([], llm.calls)
+        self.assertEqual("error", run.result.outcome)
+        self.assertIn("input could not be built", run.result.reason)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_task_protocol.py

@@ -6,6 +6,11 @@ from unittest.mock import AsyncMock, patch
 from pydantic import ValidationError
 
 from cyber_agent.core.agent_mode import RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY
+from cyber_agent.core.resource_budget import (
+    RESOURCE_BUDGET_CONTEXT_KEY,
+    ResourceBudget,
+    ResourceBudgetController,
+)
 from cyber_agent.core.runner import AgentRunner, RunConfig
 from cyber_agent.core.task_protocol import (
     TaskBrief,
@@ -16,7 +21,7 @@ from cyber_agent.core.task_protocol import (
 )
 from cyber_agent.tools import get_tool_registry
 from cyber_agent.tools.builtin.knowledge import KnowledgeConfig
-from cyber_agent.tools.builtin.subagent import agent
+from cyber_agent.tools.builtin.subagent import _record_pending_task_reports, agent
 from cyber_agent.tools.builtin.task_protocol import (
     review_task_result,
     submit_task_report,
@@ -25,11 +30,14 @@ from cyber_agent.trace.goal_models import Goal, GoalTree
 from cyber_agent.trace.goal_tool import goal_tool
 from cyber_agent.trace.models import Message, Trace
 from cyber_agent.trace.store import FileSystemTraceStore
+from cyber_agent.core.validation import ValidationResult, ValidationRun
 
 
 BRIEF = {
     "objective": "research one concrete question",
+    "reason": "the parent needs one independently checked result",
     "completion_criteria": ["provide one checked conclusion"],
+    "expected_outputs": ["one checked conclusion"],
     "constraints": ["do not guess"],
 }
 
@@ -52,6 +60,54 @@ def report_payload(*, outcome="satisfied"):
     }
 
 
+def validation_payload(child_trace_id, *, outcome="passed"):
+    if outcome == "passed":
+        issues = []
+        retry_from = None
+    elif outcome == "failed":
+        issues = ["task needs revision"]
+        retry_from = "task_definition"
+    else:
+        issues = ["validator could not evaluate the task"]
+        retry_from = None
+    return ValidationResult(
+        validator_trace_id=f"{child_trace_id}@validator-fixture",
+        evaluated_trace_id=child_trace_id,
+        outcome=outcome,
+        scope="task",
+        reason="fixture validation",
+        issues=issues,
+        retry_from=retry_from,
+    ).model_dump()
+
+
+async def create_recursive_root(store, trace_id, *, state=None, uid="user-1"):
+    budget = ResourceBudget()
+    context = {
+        "agent_mode": "recursive",
+        "agent_mode_revision": 2,
+        "agent_depth": 0,
+        "root_trace_id": trace_id,
+        "root_completion_criteria": ["finish the root task"],
+        RESOURCE_BUDGET_CONTEXT_KEY: budget.to_dict(),
+        "task_protocol": state or new_task_protocol(),
+    }
+    await store.create_trace(Trace(
+        trace_id=trace_id,
+        mode="agent",
+        task="root",
+        uid=uid,
+        model="fake-model",
+        context=context,
+    ))
+    await ResourceBudgetController(store).initialize(
+        trace_id,
+        budget,
+        initial_agents=1,
+    )
+    return await store.get_trace(trace_id)
+
+
 class FakeToolRegistry:
     def get_tool_names(self):
         return [
@@ -73,6 +129,64 @@ class ReportingRunner:
         self.stdin_check = None
         self._cancel_events = {}
         self.configs = []
+        self.resource_budget = ResourceBudgetController(store)
+        self.validation_count = 0
+
+    async def _resource_budget_for_trace(self, trace_id):
+        trace = await self.store.get_trace(trace_id)
+        root_id = trace.context["root_trace_id"]
+        root = trace if root_id == trace_id else await self.store.get_trace(root_id)
+        return root_id, ResourceBudget.from_dict(
+            root.context[RESOURCE_BUDGET_CONTEXT_KEY]
+        )
+
+    async def validate_recursive_trace(
+        self,
+        evaluated_trace_id,
+        *,
+        scope,
+        deterministic_failure=None,
+        **_kwargs,
+    ):
+        self.validation_count += 1
+        evaluated = await self.store.get_trace(evaluated_trace_id)
+        validator_trace_id = (
+            f"{evaluated_trace_id}@validator-fixture-{self.validation_count}"
+        )
+        if deterministic_failure:
+            outcome = deterministic_failure.get("outcome", "error")
+            issues = deterministic_failure.get("issues") or [
+                deterministic_failure.get("reason", "validation failed")
+            ]
+            retry_from = deterministic_failure.get("retry_from")
+        else:
+            outcome = "passed"
+            issues = []
+            retry_from = None
+        result = ValidationResult(
+            validator_trace_id=validator_trace_id,
+            evaluated_trace_id=evaluated_trace_id,
+            outcome=outcome,
+            scope=scope,
+            reason="fixture validation",
+            issues=issues,
+            retry_from=retry_from,
+        )
+        await self.store.create_trace(Trace(
+            trace_id=validator_trace_id,
+            mode="agent",
+            task=f"validate {evaluated_trace_id}",
+            parent_trace_id=evaluated_trace_id,
+            uid=evaluated.uid,
+            status="completed",
+            tools=[],
+            context={
+                "created_by_tool": "validator",
+                "root_trace_id": evaluated.context["root_trace_id"],
+                "agent_depth": evaluated.context["agent_depth"],
+            },
+        ))
+        return ValidationRun(result=result, trace_id=validator_trace_id)
 
     async def run_result(self, messages, config, on_event=None):
         self.configs.append(config)
@@ -137,21 +251,7 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
         self.store = FileSystemTraceStore(self.temp_dir.name)
         self.runner = ReportingRunner(self.store)
         self.root_id = "recursive-root"
-        context = {
-            "agent_mode": "recursive",
-            "agent_mode_revision": 2,
-            "agent_depth": 0,
-            "root_trace_id": self.root_id,
-            "task_protocol": new_task_protocol(),
-        }
-        await self.store.create_trace(Trace(
-            trace_id=self.root_id,
-            mode="agent",
-            task="root",
-            uid="user-1",
-            model="fake-model",
-            context=context,
-        ))
+        await create_recursive_root(self.store, self.root_id)
         self.tree = GoalTree(
             mission="root",
             goals=[Goal(id="1", description="delegate", status="in_progress")],
@@ -182,7 +282,10 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
 
     async def test_models_validate_and_round_trip(self):
         brief = TaskBrief.model_validate(BRIEF)
-        self.assertEqual(BRIEF, brief.model_dump(exclude_none=True))
+        self.assertEqual(
+            {**BRIEF, "parent_findings": [], "context": {}},
+            brief.model_dump(exclude_none=True),
+        )
         with self.assertRaises(ValidationError):
             TaskBrief(objective="", completion_criteria=[])
         with self.assertRaises(ValidationError):
@@ -233,7 +336,44 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual("failed", blocked["status"])
         self.assertIn("Review pending", blocked["error"])
 
-    async def test_qwen_json_string_containers_are_normalized(self):
+    async def test_child_inherits_brief_constraints_not_parent_runtime_context(self):
+        parent = await self.store.get_trace(self.root_id)
+        parent.context["parent_private_runtime"] = {"secret": "must-not-leak"}
+        await self.store.update_trace(self.root_id, context=parent.context)
+
+        result = await agent(task_brief=BRIEF, context=self.context())
+
+        child = await self.store.get_trace(result["sub_trace_id"])
+        self.assertNotIn("parent_private_runtime", child.context)
+        child_brief = ensure_task_protocol(child.context)["task_brief"]
+        self.assertEqual(["do not guess"], child_brief["constraints"])
+        child_config_context = self.runner.configs[-1].context
+        self.assertNotIn("parent_private_runtime", child_config_context)
+
+    async def test_same_task_report_version_reuses_its_validator_result(self):
+        result = await agent(task_brief=BRIEF, context=self.context())
+        child_id = result["sub_trace_id"]
+        self.assertEqual(1, self.runner.validation_count)
+        parent = await self.store.get_trace(self.root_id)
+
+        await _record_pending_task_reports(
+            self.store,
+            self.runner,
+            parent,
+            "1",
+            [(child_id, {"status": "completed", "summary": "same report"})],
+            10,
+        )
+
+        self.assertEqual(1, self.runner.validation_count)
+        validators = await self.store.list_traces(
+            parent_trace_id=child_id,
+            created_by_tool="validator",
+            limit=10,
+        )
+        self.assertEqual(1, len(validators))
+
+    async def test_recursive_messages_are_rejected(self):
         result = await agent(
             task_brief=json.dumps(BRIEF),
             messages=json.dumps([{
@@ -242,12 +382,8 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
             }]),
             context=self.context(),
         )
-        self.assertEqual("completed", result["status"])
-        child = await self.store.get_trace(result["sub_trace_id"])
-        self.assertEqual(
-            TaskBrief.model_validate(BRIEF).model_dump(),
-            ensure_task_protocol(child.context)["task_brief"],
-        )
+        self.assertEqual("failed", result["status"])
+        self.assertIn("does not accept messages", result["error"])
 
     async def test_messages_reject_mixed_dimensions(self):
         mixed_messages = [
@@ -286,7 +422,9 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
         first = await agent(task_brief=BRIEF, context=self.context())
         next_brief = {
             "objective": "check the independent gap",
+            "reason": "the first report identified a separate gap",
             "completion_criteria": ["close the gap"],
+            "expected_outputs": ["a checked gap analysis"],
         }
         parent = await self.store.get_trace(self.root_id)
         ensure_task_protocol(parent.context)["protocol_correction_attempts"] = 2
@@ -327,7 +465,9 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
             BRIEF,
             {
                 "objective": "research second question",
+                "reason": "the batch needs a second independent result",
                 "completion_criteria": ["provide second conclusion"],
+                "expected_outputs": ["a second conclusion"],
             },
         ]
         result = await agent(task_brief=briefs, context=self.context())
@@ -378,7 +518,9 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
                 BRIEF,
                 {
                     "objective": "second batch child",
+                    "reason": "exercise queued batch behavior",
                     "completion_criteria": ["finish the second child"],
+                    "expected_outputs": ["the second child result"],
                 },
             ],
             context=self.context(),
@@ -386,7 +528,9 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
         child_ids = [item["trace_id"] for item in result["sub_trace_ids"]]
         next_brief = {
             "objective": "approved follow-up",
+            "reason": "the parent approved one follow-up",
             "completion_criteria": ["finish the approved follow-up"],
+            "expected_outputs": ["the approved follow-up result"],
         }
         accepted = await review_task_result(
             child_trace_id=child_ids[0],
@@ -505,7 +649,9 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
         wrong = await agent(
             task_brief={
                 "objective": "silently replace the reviewed task",
+                "reason": "attempt a mismatched revision",
                 "completion_criteria": ["this must be rejected"],
+                "expected_outputs": ["a rejected replacement"],
             },
             continue_from=child_id,
             context=self.context(),
@@ -532,7 +678,9 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
         briefs = [
             {
                 "objective": f"batch child {index}",
+                "reason": "fill the local child quota",
                 "completion_criteria": [f"finish child {index}"],
+                "expected_outputs": [f"result for child {index}"],
             }
             for index in range(5)
         ]
@@ -540,7 +688,9 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
         child_ids = [item["trace_id"] for item in result["sub_trace_ids"]]
         next_brief = {
             "objective": "one allowed reserved child",
+            "reason": "use the final available child slot",
             "completion_criteria": ["use the final available slot"],
+            "expected_outputs": ["one final child result"],
         }
         first = await review_task_result(
             child_trace_id=child_ids[0],
@@ -557,7 +707,9 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
             reason="A seventh child must not be approved",
             approved_next_task={
                 "objective": "overflow child",
+                "reason": "exercise quota rejection",
                 "completion_criteria": ["must never be scheduled"],
+                "expected_outputs": ["no output because creation is rejected"],
             },
             context=self.context(),
         )
@@ -708,7 +860,9 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
         first = await agent(task_brief=BRIEF, context=child_tool_context)
         next_brief = {
             "objective": "check a second grandchild task",
+            "reason": "the child approved another direct task",
             "completion_criteria": ["provide a second checked conclusion"],
+            "expected_outputs": ["a second checked conclusion"],
         }
         reviewed = await review_task_result(
             child_trace_id=first["sub_trace_id"],
@@ -775,6 +929,7 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
             mode="agent",
             task="failed",
             parent_trace_id=self.root_id,
+            parent_goal_id="1",
             uid="user-1",
             context=child_context,
         ))
@@ -783,6 +938,7 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
         state["pending_reviews"][child_id] = {
             "goal_id": "1",
             "task_report": failed_report.model_dump(),
+            "validation_result": validation_payload(child_id, outcome="failed"),
             "received_at_sequence": 2,
         }
         await self.store.update_trace(self.root_id, context=root.context)
@@ -802,6 +958,10 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
         self.temp_dir = tempfile.TemporaryDirectory()
         self.store = FileSystemTraceStore(self.temp_dir.name)
         self.trace_id = "gate-child"
+        await create_recursive_root(self.store, "root")
+        controller = ResourceBudgetController(self.store)
+        budget = ResourceBudget()
+        await controller.reserve_agents("root", budget, 1)
         await self.store.create_trace(Trace(
             trace_id=self.trace_id,
             mode="agent",
@@ -859,6 +1019,18 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
         async def llm_call(**kwargs):
             nonlocal calls
             calls += 1
+            if kwargs.get("tools") == []:
+                return {
+                    "content": json.dumps({
+                        "outcome": "passed",
+                        "scope": "root",
+                        "reason": "the corrected behavior meets the criterion",
+                        "issues": [],
+                        "retry_from": None,
+                    }),
+                    "tool_calls": [],
+                    "finish_reason": "stop",
+                }
             if calls == 1:
                 return {
                     "content": "",
@@ -902,25 +1074,25 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
 
     async def test_lifecycle_tool_batch_is_rejected_before_execution(self):
         root_id = "batch-gate-root"
-        await self.store.create_trace(Trace(
-            trace_id=root_id,
-            mode="agent",
-            task="root",
-            uid="user-1",
-            context={
-                "agent_mode": "recursive",
-                "agent_mode_revision": 2,
-                "agent_depth": 0,
-                "root_trace_id": root_id,
-                "task_protocol": new_task_protocol(),
-            },
-        ))
+        await create_recursive_root(self.store, root_id)
         await self.store.update_goal_tree(root_id, GoalTree(mission="root"))
         calls = 0
 
         async def llm_call(**kwargs):
             nonlocal calls
             calls += 1
+            if kwargs.get("tools") == []:
+                return {
+                    "content": json.dumps({
+                        "outcome": "passed",
+                        "scope": "root",
+                        "reason": "the corrected lifecycle behavior is valid",
+                        "issues": [],
+                        "retry_from": None,
+                    }),
+                    "tool_calls": [],
+                    "finish_reason": "stop",
+                }
             if calls == 1:
                 return {
                     "content": "",
@@ -966,7 +1138,11 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
             ),
         )
         self.assertEqual("completed", result["status"])
-        children = await self.store.list_traces(parent_trace_id=root_id, limit=10)
+        children = await self.store.list_traces(
+            parent_trace_id=root_id,
+            created_by_tool="agent",
+            limit=10,
+        )
         self.assertEqual([], children)
         tool_messages = [
             message
@@ -992,6 +1168,7 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
             "grandchild": {
                 "goal_id": "1",
                 "task_report": report,
+                "validation_result": validation_payload("grandchild"),
                 "received_at_sequence": 6,
             }
         }
@@ -1037,6 +1214,9 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
             "preserved-grandchild": {
                 "goal_id": "1",
                 "task_report": report,
+                "validation_result": validation_payload(
+                    "preserved-grandchild"
+                ),
                 "received_at_sequence": 2,
             }
         }
@@ -1075,19 +1255,7 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
             "task_brief": BRIEF,
             "created_at_sequence": 1,
         }]
-        await self.store.create_trace(Trace(
-            trace_id=root_id,
-            mode="agent",
-            task="root",
-            uid="user-1",
-            context={
-                "agent_mode": "recursive",
-                "agent_mode_revision": 2,
-                "agent_depth": 0,
-                "root_trace_id": root_id,
-                "task_protocol": state,
-            },
-        ))
+        await create_recursive_root(self.store, root_id, state=state)
         await self.store.update_goal_tree(root_id, GoalTree(mission="root"))
         calls = 0
 
@@ -1155,6 +1323,22 @@ class RuntimeProtocolSchemaTest(unittest.TestCase):
         child_schemas = self.runner._get_runtime_tool_schemas(RunConfig(), child)
         self.assertIn("submit_task_report", self.names(child_schemas))
         self.assertNotIn("review_task_result", self.names(child_schemas))
+        self.assertNotIn("evaluate", self.names(child_schemas))
+        recursive_agent = next(
+            item for item in child_schemas if item["function"]["name"] == "agent"
+        )
+        recursive_properties = recursive_agent["function"]["parameters"]["properties"]
+        self.assertNotIn("messages", recursive_properties)
+        brief_object = recursive_properties["task_brief"]["oneOf"][0]
+        self.assertEqual(
+            {
+                "objective",
+                "reason",
+                "completion_criteria",
+                "expected_outputs",
+            },
+            set(brief_object["required"]),
+        )
 
         state = ensure_task_protocol(child.context)
         state["pending_reviews"]["grandchild"] = {"task_report": {}}
@@ -1182,6 +1366,7 @@ class RuntimeProtocolSchemaTest(unittest.TestCase):
                 "REJECT_REFINE",
                 "REVISE_CHILD",
                 "DESCEND_AGAIN",
+                "REPLAN_CURRENT",
                 "ASCEND",
                 "FAIL",
             ],