增加 recursive 独立验收与根完成门禁

cyber_agent/core/runner.py

@@ -26,6 +26,7 @@ from typing import AsyncIterator, Optional, Dict, Any, List, Callable, Literal,
 
 from cyber_agent.trace.models import Trace, Message
 from cyber_agent.trace.protocols import TraceStore
+from cyber_agent.trace.trace_id import generate_sub_trace_id
 from cyber_agent.trace.goal_models import GoalTree
 from cyber_agent.trace.compaction import (
     CompressionConfig,
@@ -54,6 +55,19 @@ from cyber_agent.core.agent_mode import (
 from cyber_agent.core.task_protocol import (
     ensure_task_protocol,
     protocol_error_report,
+    rebuild_pending_replans,
+)
+from cyber_agent.core.resource_budget import (
+    RESOURCE_BUDGET_CONTEXT_KEY,
+    ResourceBudget,
+    ResourceBudgetController,
+    ResourceBudgetExceeded,
+    ResourceBudgetStateError,
+)
+from cyber_agent.core.validation import (
+    LLMValidator,
+    ValidationRun,
+    ValidationScope,
 )
 from cyber_agent.core.prompts import (
     DEFAULT_SYSTEM_PREFIX,
@@ -143,6 +157,7 @@ class RunConfig:
     parallel_tool_execution: bool = False      # 是否启用并发 Tool Call 执行（慎用，需确保无资源冲突）
     child_execution_mode: Literal["sequential", "parallel"] = "sequential"
     max_parallel_children: int = 2
+    root_completion_criteria: List[str] = field(default_factory=list)
 
     # --- Trace 控制 ---
     trace_id: Optional[str] = None             # None = 新建
@@ -249,6 +264,9 @@ class AgentRunner:
         # 当前 run 的 MemoryConfig（由 run() 根据 RunConfig.memory 设置）
         # dream 工具从 context.runner 读取此字段，判断是否 memory-bearing
         self._current_memory_config: Optional[MemoryConfig] = None
+        self.resource_budget = (
+            ResourceBudgetController(trace_store) if trace_store else None
+        )
 
     # ===== 核心公开方法 =====
 
@@ -256,6 +274,178 @@ class AgentRunner:
         """获取指定 trace 的 context 使用情况"""
         return self._context_usage.get(trace_id)
 
+    async def _resource_budget_for_trace(
+        self,
+        trace_id: str,
+    ) -> tuple[str, ResourceBudget] | None:
+        """Resolve the immutable Recursive root budget for one local Trace."""
+        if not self.trace_store:
+            return None
+        trace = await self.trace_store.get_trace(trace_id)
+        if not trace or not policy_from_context(trace.context).requires_task_protocol:
+            return None
+        root_trace_id = trace.context.get("root_trace_id") or trace.trace_id
+        root = trace if root_trace_id == trace.trace_id else await self.trace_store.get_trace(root_trace_id)
+        if not root:
+            raise ResourceBudgetStateError(
+                f"Recursive root Trace not found: {root_trace_id}"
+            )
+        snapshot = root.context.get(RESOURCE_BUDGET_CONTEXT_KEY)
+        if snapshot is None:
+            raise ResourceBudgetStateError(
+                "Recursive tree has no persisted resource budget; create a new trace"
+            )
+        return root_trace_id, ResourceBudget.from_dict(snapshot)
+
+    async def call_recursive_llm(
+        self,
+        trace_id: str,
+        *,
+        purpose: Literal["ordinary", "root_validator"] = "ordinary",
+        call: Optional[Callable] = None,
+        fail_on_post_response_exhaustion: bool = False,
+        **kwargs: Any,
+    ) -> Dict[str, Any]:
+        """Call an LLM and atomically account for Recursive tree usage."""
+        llm = call or self.llm_call
+        if not llm:
+            raise ValueError("llm_call function not provided")
+        resolved = await self._resource_budget_for_trace(trace_id)
+        if resolved is None:
+            return await llm(**kwargs)
+        root_trace_id, budget = resolved
+        if not self.resource_budget:
+            raise ResourceBudgetStateError("ResourceBudgetController is unavailable")
+        await self.resource_budget.reserve_llm_call(
+            root_trace_id,
+            budget,
+            purpose=purpose,
+        )
+        result = await llm(**kwargs)
+        try:
+            await self.resource_budget.record_llm_usage(
+                root_trace_id,
+                budget,
+                prompt_tokens=int(result.get("prompt_tokens", 0) or 0),
+                completion_tokens=int(result.get("completion_tokens", 0) or 0),
+                cost_usd=float(result.get("cost", 0) or 0),
+            )
+        except ResourceBudgetExceeded as exc:
+            if fail_on_post_response_exhaustion:
+                raise
+            result = dict(result)
+            result["_resource_budget_exceeded"] = exc.dimension
+        return result
+
+    async def record_recursive_tool_usage(
+        self,
+        trace_id: str,
+        tool_usage: Dict[str, Any],
+    ) -> None:
+        """Account for a tool-owned model call when the tool reports usage."""
+        resolved = await self._resource_budget_for_trace(trace_id)
+        if resolved is None:
+            return
+        root_trace_id, budget = resolved
+        if not self.resource_budget:
+            raise ResourceBudgetStateError("ResourceBudgetController is unavailable")
+        await self.resource_budget.record_external_llm_usage(
+            root_trace_id,
+            budget,
+            prompt_tokens=int(tool_usage.get("prompt_tokens", 0) or 0),
+            completion_tokens=int(tool_usage.get("completion_tokens", 0) or 0),
+            cost_usd=float(tool_usage.get("cost", 0) or 0),
+        )
+
+    async def validate_recursive_trace(
+        self,
+        evaluated_trace_id: str,
+        *,
+        scope: ValidationScope,
+        task_brief: Optional[Dict[str, Any]] = None,
+        task_report: Optional[Dict[str, Any]] = None,
+        completion_criteria: Optional[List[str]] = None,
+        expected_outputs: Optional[List[str]] = None,
+        candidate_output: Optional[str] = None,
+        deterministic_failure: Optional[Dict[str, Any]] = None,
+        root_validator: bool = False,
+    ) -> ValidationRun:
+        """Run one framework-owned, tool-free validation Trace."""
+        if not self.trace_store or not self.llm_call:
+            raise RuntimeError("Validator requires trace_store and llm_call")
+        evaluated = await self.trace_store.get_trace(evaluated_trace_id)
+        if not evaluated:
+            raise ValueError(f"Trace not found: {evaluated_trace_id}")
+        lineage_event = None
+        if (
+            evaluated.parent_trace_id
+            and evaluated_trace_id not in self._recursive_active_traces
+        ):
+            lineage_event = self.register_recursive_child(
+                evaluated.parent_trace_id,
+                evaluated_trace_id,
+            )
+        validator_trace_id = generate_sub_trace_id(
+            evaluated_trace_id,
+            "validator",
+        )
+        event = self.register_recursive_child(
+            evaluated_trace_id,
+            validator_trace_id,
+        )
+
+        async def validator_llm_call(**kwargs: Any) -> Dict[str, Any]:
+            if self.is_cancel_requested(validator_trace_id):
+                raise RuntimeError("Validator execution was stopped")
+            result = await self.call_recursive_llm(
+                evaluated_trace_id,
+                purpose="root_validator" if root_validator else "ordinary",
+                **kwargs,
+            )
+            if self.is_cancel_requested(validator_trace_id):
+                raise RuntimeError("Validator execution was stopped")
+            dimension = result.get("_resource_budget_exceeded")
+            if dimension:
+                raise RuntimeError(
+                    f"Validator exceeded tree resource budget: {dimension}"
+                )
+            return result
+
+        validator = LLMValidator(
+            llm_call=validator_llm_call,
+            trace_store=self.trace_store,
+        )
+        try:
+            if deterministic_failure:
+                return await validator.record_non_success(
+                    evaluated_trace=evaluated,
+                    scope=scope,
+                    outcome=deterministic_failure.get("outcome", "error"),
+                    reason=deterministic_failure.get("reason", "Task did not complete"),
+                    issues=deterministic_failure.get("issues"),
+                    retry_from=deterministic_failure.get("retry_from"),
+                    validator_trace_id=validator_trace_id,
+                )
+            trajectory = await self.trace_store.get_main_path_messages(
+                evaluated_trace_id,
+                evaluated.head_sequence or evaluated.last_sequence,
+            )
+            return await validator.validate(
+                evaluated_trace=evaluated,
+                trajectory=trajectory,
+                scope=scope,
+                task_brief=task_brief,
+                task_report=task_report,
+                completion_criteria=completion_criteria,
+                expected_outputs=expected_outputs,
+                candidate_output=candidate_output,
+                validator_trace_id=validator_trace_id,
+            )
+        finally:
+            self.release_recursive_trace(validator_trace_id, event)
+            if lineage_event is not None:
+                self.release_recursive_trace(evaluated_trace_id, lineage_event)
+
     async def dream(
         self,
         memory_config: MemoryConfig,
@@ -370,17 +560,28 @@ class AgentRunner:
 
         except Exception as e:
             self.log.error(f"Agent run failed: {e}")
-            tid = config.trace_id or (trace.trace_id if trace else None)
+            # Preparation rejections (for example, attempting to resume an
+            # immutable Validator Trace) must not rewrite the existing record.
+            tid = trace.trace_id if trace else None
             if self.trace_store and tid:
                 # 读取当前 last_sequence 作为 head_sequence，确保续跑时能加载完整历史
                 current = await self.trace_store.get_trace(tid)
                 head_seq = current.last_sequence if current else None
+                updates: Dict[str, Any] = {
+                    "status": "failed",
+                    "head_sequence": head_seq,
+                    "error_message": str(e),
+                    "completed_at": datetime.now(),
+                }
+                if isinstance(e, ResourceBudgetExceeded):
+                    current_context = dict(current.context if current else {})
+                    current_context["termination_reason"] = (
+                        f"budget_exhausted:{e.dimension}"
+                    )
+                    updates["context"] = current_context
                 await self.trace_store.update_trace(
                     tid,
-                    status="failed",
-                    head_sequence=head_seq,
-                    error_message=str(e),
-                    completed_at=datetime.now()
+                    **updates,
                 )
                 trace_obj = await self.trace_store.get_trace(tid)
                 if trace_obj:
@@ -647,10 +848,28 @@ class AgentRunner:
                 config.child_execution_mode,
                 config.max_parallel_children,
             )
+            criteria = [
+                item.strip()
+                for item in config.root_completion_criteria
+                if isinstance(item, str) and item.strip()
+            ]
+            if not criteria:
+                raise ValueError(
+                    "New Recursive root traces require root_completion_criteria"
+                )
+            if len(criteria) != len(config.root_completion_criteria):
+                raise ValueError(
+                    "root_completion_criteria must contain non-empty strings"
+                )
+            budget = ResourceBudget.from_environment()
         trace_id = str(uuid.uuid4())
 
         # 生成任务名称
-        task_name = config.name or await self._generate_task_name(messages)
+        task_name = config.name or (
+            self._fallback_task_name(messages)
+            if policy.mode is AgentMode.RECURSIVE
+            else await self._generate_task_name(messages)
+        )
 
         # 准备工具 Schema
         tool_schemas = self._get_tool_schemas(config.tools, config.tool_groups, config.exclude_tools)
@@ -659,6 +878,8 @@ class AgentRunner:
         trace_context.setdefault("agent_depth", 0)
         trace_context.setdefault("root_trace_id", trace_id)
         if policy.requires_task_protocol:
+            trace_context["root_completion_criteria"] = criteria
+            trace_context[RESOURCE_BUDGET_CONTEXT_KEY] = budget.to_dict()
             ensure_task_protocol(trace_context)
 
         trace_obj = Trace(
@@ -681,6 +902,13 @@ class AgentRunner:
         if self.trace_store:
             await self.trace_store.create_trace(trace_obj)
             await self.trace_store.update_goal_tree(trace_id, goal_tree)
+            assert self.resource_budget is not None
+            if policy.mode is AgentMode.RECURSIVE:
+                await self.resource_budget.initialize(
+                    trace_id,
+                    budget,
+                    initial_agents=1,
+                )
 
         return trace_obj, goal_tree, 1
 
@@ -695,6 +923,13 @@ class AgentRunner:
         trace_obj = await self.trace_store.get_trace(config.trace_id)
         if not trace_obj:
             raise ValueError(f"Trace not found: {config.trace_id}")
+        if (
+            trace_obj.agent_type == "validator"
+            or trace_obj.context.get("created_by_tool") == "validator"
+        ):
+            raise ValueError(
+                "Validator traces cannot be continued or rewound"
+            )
 
         # Historical traces predate AGENT_MODE.  They resume in safe Legacy mode
         # and the choice is persisted immediately; environment changes never
@@ -713,6 +948,37 @@ class AgentRunner:
                     config.child_execution_mode,
                     config.max_parallel_children,
                 )
+                if policy.requires_task_protocol:
+                    root_trace_id = trace_obj.context.get("root_trace_id")
+                    if root_trace_id == trace_obj.trace_id and not trace_obj.context.get(
+                        "root_completion_criteria"
+                    ):
+                        raise ValueError(
+                            "This experimental Recursive trace predates required root "
+                            "completion criteria; create a new trace"
+                        )
+                    root = (
+                        trace_obj
+                        if root_trace_id == trace_obj.trace_id
+                        else await self.trace_store.get_trace(root_trace_id)
+                    )
+                    if not root or RESOURCE_BUDGET_CONTEXT_KEY not in root.context:
+                        raise ValueError(
+                            "This experimental Recursive trace predates tree resource "
+                            "budgets; create a new trace"
+                        )
+                    ResourceBudget.from_dict(
+                        root.context[RESOURCE_BUDGET_CONTEXT_KEY]
+                    )
+                    if root_trace_id == trace_obj.trace_id:
+                        state = ensure_task_protocol(trace_obj.context)
+                        if state["root_validation_attempts"] >= 2:
+                            raise ValueError(
+                                "Root task already used its two independent "
+                                "validation attempts; create a new trace"
+                            )
+                    assert self.resource_budget is not None
+                    await self.resource_budget.get_usage(root.trace_id)
 
         goal_tree = await self.trace_store.get_goal_tree(config.trace_id)
         if goal_tree is None:
@@ -982,7 +1248,11 @@ class AgentRunner:
 
         # Level 2 压缩：检查 Level 1 后是否仍超阈值
         # 注意：Level 1 压缩后需要重新优化图片并计算 token
-        optimized_history_after = await self._optimize_images(history, config.model)
+        optimized_history_after = await self._optimize_images(
+            history,
+            config.model,
+            trace_id=trace_id,
+        )
         token_count_after = estimate_tokens(optimized_history_after)
         needs_level2 = token_count_after > max_tokens
 
@@ -1105,11 +1375,14 @@ class AgentRunner:
         )
 
         # 单次 LLM 调用（无工具）
-        result = await self.llm_call(
+        result = await self.call_recursive_llm(
+            trace_id,
+            purpose="ordinary",
             messages=compress_messages,
             model=config.model,
             tools=[],  # 不提供工具
             temperature=config.temperature,
+            fail_on_post_response_exhaustion=True,
             **config.extra_llm_params,
         )
 
@@ -1426,7 +1699,11 @@ class AgentRunner:
             llm_messages = [{k: v for k, v in msg.items() if not k.startswith("_")} for msg in history]
 
             # 优化已处理的图片（分级处理：保留/压缩/描述）
-            llm_messages = await self._optimize_images(llm_messages, config.model)
+            llm_messages = await self._optimize_images(
+                llm_messages,
+                config.model,
+                trace_id=trace_id,
+            )
 
             # 对历史消息应用 Prompt Caching
             llm_messages = self._add_cache_control(
@@ -1436,7 +1713,9 @@ class AgentRunner:
             )
 
             # 调用 LLM（等待完成后再检查 cancel 信号，不中断正在进行的调用）
-            result = await self.llm_call(
+            result = await self.call_recursive_llm(
+                trace_id,
+                purpose="ordinary",
                 messages=llm_messages,
                 model=config.model,
                 tools=tool_schemas,
@@ -1462,6 +1741,10 @@ class AgentRunner:
             step_cost = result.get("cost", 0)
             cache_creation_tokens = result.get("cache_creation_tokens")
             cache_read_tokens = result.get("cache_read_tokens")
+            budget_exceeded_dimension = result.get("_resource_budget_exceeded")
+            if budget_exceeded_dimension:
+                tool_calls = None
+                finish_reason = "budget_exhausted"
 
             lifecycle_tools = {"agent", "submit_task_report", "review_task_result"}
             lifecycle_call_count = sum(
@@ -1491,7 +1774,11 @@ class AgentRunner:
             )
 
             # 周期性自动注入上下文（仅主路径）
-            if not side_branch_ctx and iteration % CONTEXT_INJECTION_INTERVAL == 0:
+            if (
+                not side_branch_ctx
+                and not budget_exceeded_dimension
+                and iteration % CONTEXT_INJECTION_INTERVAL == 0
+            ):
                 # 检查是否已经调用了 get_current_context
                 if tool_calls:
                     has_context_call = any(
@@ -1523,6 +1810,7 @@ class AgentRunner:
             # Skill 指定注入（仅主路径，首轮 iteration==0 时执行）
             if (
                 not side_branch_ctx
+                and not budget_exceeded_dimension
                 and inject_skills
                 and iteration == 0
                 and not lifecycle_call_count
@@ -1654,6 +1942,19 @@ class AgentRunner:
             head_seq = sequence
             sequence += 1
 
+            if budget_exceeded_dimension:
+                completion_status = "failed"
+                trace.context["termination_reason"] = (
+                    f"budget_exhausted:{budget_exceeded_dimension}"
+                )
+                if self.trace_store:
+                    await self.trace_store.update_trace(
+                        trace_id,
+                        context=trace.context,
+                        error_message=trace.context["termination_reason"],
+                    )
+                break
+
             # 检查侧分支是否应该退出
             if side_branch_ctx:
                 # 计算侧分支已执行的轮次
@@ -1971,6 +2272,10 @@ class AgentRunner:
                             await self.trace_store.add_message(tool_msg)
                             if tool_usage:
                                 await self.trace_store.record_model_usage(trace_id=trace_id, sequence=sequence, role="tool", tool_name=tool_name, model=tool_usage.get("model"), prompt_tokens=tool_usage.get("prompt_tokens", 0), completion_tokens=tool_usage.get("completion_tokens", 0), cache_read_tokens=tool_usage.get("cache_read_tokens", 0))
+                                await self.record_recursive_tool_usage(
+                                    trace_id,
+                                    tool_usage,
+                                )
                             if tool_images:
                                 import base64 as b64mod
                                 for img in tool_images:
@@ -2144,6 +2449,10 @@ class AgentRunner:
                                     completion_tokens=tool_usage.get("completion_tokens", 0),
                                     cache_read_tokens=tool_usage.get("cache_read_tokens", 0),
                                 )
+                                await self.record_recursive_tool_usage(
+                                    trace_id,
+                                    tool_usage,
+                                )
                             # 截图单独存为同名 PNG 文件
                             if tool_images:
                                 import base64 as b64mod
@@ -2295,6 +2604,91 @@ class AgentRunner:
                 self.log.info("任务完成，进入完成后反思侧分支")
                 continue
 
+            if (
+                not side_branch_ctx
+                and self.trace_store
+                and policy_from_context(trace.context).requires_task_protocol
+                and not trace.parent_trace_id
+            ):
+                state = ensure_task_protocol(trace.context)
+                if (
+                    state["root_validation_attempts"] >= 2
+                ):
+                    raise ValueError(
+                        "Root task already used its two independent validation "
+                        "attempts; create a new trace"
+                    )
+                # The Validator must read the exact persisted main path that
+                # produced this candidate, including real Tool Results.
+                await self.trace_store.update_trace(
+                    trace_id,
+                    head_sequence=head_seq,
+                )
+                trace.head_sequence = head_seq
+                validation_run = await self.validate_recursive_trace(
+                    trace_id,
+                    scope="root",
+                    completion_criteria=trace.context["root_completion_criteria"],
+                    candidate_output=response_content,
+                    root_validator=True,
+                )
+                validation_record = {
+                    **validation_run.result.model_dump(),
+                    "evaluated_at_sequence": head_seq,
+                }
+                state["root_validation_history"].append(validation_record)
+                state["root_validation_attempts"] += 1
+                state["root_validation_passed"] = (
+                    validation_run.result.outcome == "passed"
+                )
+                await self.trace_store.update_trace(
+                    trace_id,
+                    context=trace.context,
+                )
+                if not state["root_validation_passed"]:
+                    if state["root_validation_attempts"] < 2:
+                        validation_feedback = {
+                            "role": "user",
+                            "content": (
+                                "Root validation did not pass. Revise the current "
+                                "answer using this framework-owned ValidationResult:\n"
+                                + json.dumps(
+                                    validation_run.result.model_dump(),
+                                    ensure_ascii=False,
+                                )
+                            ),
+                        }
+                        history.append(validation_feedback)
+                        feedback_msg = Message.from_llm_dict(
+                            validation_feedback,
+                            trace_id=trace_id,
+                            sequence=sequence,
+                            goal_id=(goal_tree.current_id if goal_tree else None),
+                            parent_sequence=head_seq,
+                        )
+                        await self.trace_store.add_message(feedback_msg)
+                        head_seq = sequence
+                        sequence += 1
+                        await self.trace_store.update_trace(
+                            trace_id,
+                            head_sequence=head_seq,
+                        )
+                        trace.head_sequence = head_seq
+                        if goal_tree and goal_tree.current_id:
+                            goal = goal_tree.find(goal_tree.current_id)
+                            if goal:
+                                goal.status = "in_progress"
+                                await self.trace_store.update_goal_tree(
+                                    trace_id,
+                                    goal_tree,
+                                )
+                        continue
+                    completion_status = "failed"
+                    await self.trace_store.update_trace(
+                        trace_id,
+                        error_message="Root task did not pass independent validation",
+                    )
+
             break
 
         if (
@@ -2331,6 +2725,11 @@ class AgentRunner:
                     completion_status = "failed"
                 if state["next_actions"]:
                     completion_status = "failed"
+                if (
+                    not trace.parent_trace_id
+                    and not state.get("root_validation_passed")
+                ):
+                    completion_status = "failed"
 
         # 清理 trace 相关的跟踪数据
         self._context_warned.pop(trace_id, None)
@@ -2462,6 +2861,7 @@ class AgentRunner:
                     state["report_history"].append(state["task_report"])
                 state["task_report"] = None
                 state["task_report_submitted_at_sequence"] = None
+                state["task_report_validation"] = None
             state["pending_reviews"] = {
                 child_id: entry
                 for child_id, entry in state["pending_reviews"].items()
@@ -2488,6 +2888,7 @@ class AgentRunner:
                 action for action in state["next_actions"]
                 if action.get("created_at_sequence", 0) <= cutoff
             ]
+            state["pending_replans"] = rebuild_pending_replans(state)
             state["protocol_correction_attempts"] = 0
             await self.trace_store.update_trace(trace_id, context=trace.context)
 
@@ -2844,7 +3245,13 @@ class AgentRunner:
 
     # ===== 辅助方法 =====
 
-    async def _optimize_images(self, messages: List[Dict], model: str) -> List[Dict]:
+    async def _optimize_images(
+        self,
+        messages: List[Dict],
+        model: str,
+        *,
+        trace_id: Optional[str] = None,
+    ) -> List[Dict]:
         """
         分级优化已处理的图片，节省 token
 
@@ -3069,7 +3476,11 @@ class AgentRunner:
                             stats["described"] += 1
                             stats["cache_hit"] += 1
                         else:
-                            description = await self._generate_image_description(image_url, model)
+                            description = await self._generate_image_description(
+                                image_url,
+                                model,
+                                trace_id=trace_id,
+                            )
                             url_info = f" (URL: {image_url[:100]}...)" if not image_url.startswith("data:") else ""
                             desc_block = {
                                 "type": "text",
@@ -3177,7 +3588,13 @@ class AgentRunner:
             self.log.warning(f"[Image Process] 处理图片尺寸失败: {e}")
             return None
 
-    async def _generate_image_description(self, image_url: str, current_model: str) -> str:
+    async def _generate_image_description(
+        self,
+        image_url: str,
+        current_model: str,
+        *,
+        trace_id: Optional[str] = None,
+    ) -> str:
         """
         使用小模型生成图片的文本描述
 
@@ -3211,16 +3628,28 @@ class AgentRunner:
             ]
 
             # 调用 LLM
-            result = await self.llm_call(
-                messages=messages,
-                model=description_model,
-                tools=None,
-                temperature=0.3,
+            call_kwargs = {
+                "messages": messages,
+                "model": description_model,
+                "tools": None,
+                "temperature": 0.3,
+            }
+            result = (
+                await self.call_recursive_llm(
+                    trace_id,
+                    purpose="ordinary",
+                    fail_on_post_response_exhaustion=True,
+                    **call_kwargs,
+                )
+                if trace_id
+                else await self.llm_call(**call_kwargs)
             )
 
             description = result.get("content", "").strip()
             return description if description else "图片内容"
 
+        except (ResourceBudgetExceeded, ResourceBudgetStateError):
+            raise
         except Exception as e:
             self.log.warning(f"[Image Description] 生成描述失败: {e}")
             return "图片内容"
@@ -3429,9 +3858,13 @@ class AgentRunner:
         protocol_tools = {"submit_task_report", "review_task_result"}
 
         if not policy.requires_task_protocol or in_side_branch:
-            return tool_names - protocol_tools
+            available = tool_names - protocol_tools
+            if policy.requires_task_protocol:
+                available.discard("evaluate")
+            return available
 
         state = ensure_task_protocol(trace.context)
+        tool_names.discard("evaluate")
         if state["pending_reviews"]:
             return tool_names & {"review_task_result"}
         if state["next_actions"]:
@@ -3471,6 +3904,7 @@ class AgentRunner:
             properties = parameters.get("properties", {})
             required = parameters.setdefault("required", [])
             if policy.requires_task_protocol:
+                properties.pop("messages", None)
                 if "task" in properties:
                     properties["task"]["description"] = (
                         "Only for agent_type=remote_*; local Recursive calls use task_brief."
@@ -3607,9 +4041,9 @@ class AgentRunner:
 
         return system_prompt
 
-    async def _generate_task_name(self, messages: List[Dict]) -> str:
-        """生成任务名称：优先使用 utility_llm，fallback 到文本截取"""
-        # 提取 messages 中的文本内容
+    @staticmethod
+    def _task_text(messages: List[Dict]) -> str:
+        """Extract the same plain task text used by Legacy title generation."""
         text_parts = []
         for msg in messages:
             content = msg.get("content", "")
@@ -3619,10 +4053,20 @@ class AgentRunner:
                 for part in content:
                     if isinstance(part, dict) and part.get("type") == "text":
                         text_parts.append(part.get("text", ""))
-        raw_text = " ".join(text_parts).strip()
+        return " ".join(text_parts).strip()
 
+    @classmethod
+    def _fallback_task_name(cls, messages: List[Dict]) -> str:
+        """Build a deterministic title without spending an untracked LLM call."""
+        raw_text = cls._task_text(messages)
         if not raw_text:
             return TASK_NAME_FALLBACK
+        return raw_text[:50] + ("..." if len(raw_text) > 50 else "")
+
+    async def _generate_task_name(self, messages: List[Dict]) -> str:
+        """生成任务名称：优先使用 utility_llm，fallback 到文本截取"""
+        fallback = self._fallback_task_name(messages)
+        raw_text = self._task_text(messages)
 
         # 尝试使用 utility_llm 生成标题
         if self.utility_llm_call:
@@ -3641,7 +4085,7 @@ class AgentRunner:
                 pass
 
         # Fallback: 截取前 50 字符
-        return raw_text[:50] + ("..." if len(raw_text) > 50 else "")
+        return fallback
 
     def _format_skills(self, skills: List[Skill]) -> str:
         if not skills:

cyber_agent/tools/builtin/subagent.py

@@ -36,6 +36,16 @@ from cyber_agent.core.task_protocol import (
     protocol_error_report,
     stopped_task_report,
 )
+from cyber_agent.core.context_policy import (
+    ContextPolicyError,
+    normalize_task_brief,
+    task_briefs_match,
+)
+from cyber_agent.core.resource_budget import (
+    ResourceBudgetExceeded,
+    ResourceBudgetStateError,
+)
+from cyber_agent.core.validation import ValidationResult
 from cyber_agent.trace.models import Trace, Messages
 from cyber_agent.trace.trace_id import generate_sub_trace_id
 from cyber_agent.trace.goal_models import GoalTree
@@ -269,12 +279,14 @@ async def _load_or_create_task_report(
     )
     state["task_report"] = report.model_dump()
     state["task_report_submitted_at_sequence"] = generated_at_sequence
+    state["task_report_validation"] = None
     await store.update_trace(child_trace_id, context=child.context)
     return report
 
 
 async def _record_pending_task_reports(
     store,
+    runner,
     parent_trace: Trace,
     goal_id: str | None,
     child_results: List[tuple[str, Dict[str, Any]]],
@@ -291,9 +303,53 @@ async def _record_pending_task_reports(
             result.get("status", "unknown"),
             received_at_sequence,
         )
+        child_trace = await store.get_trace(child_trace_id)
+        if not child_trace:
+            raise ValueError(f"Child Trace not found for validation: {child_trace_id}")
+        child_state = ensure_task_protocol(child_trace.context)
+        task_brief = child_state.get("task_brief")
+        report_payload = report.model_dump()
+        cached = child_state.get("task_report_validation")
+        validation_result = None
+        if isinstance(cached, dict) and cached.get("task_report") == report_payload:
+            try:
+                validation_result = ValidationResult.model_validate(
+                    cached.get("validation_result")
+                )
+                if validation_result.evaluated_trace_id != child_trace_id:
+                    validation_result = None
+            except ValidationError:
+                validation_result = None
+        if validation_result is None:
+            if report.outcome in {"satisfied", "partial"}:
+                validation_run = await runner.validate_recursive_trace(
+                    child_trace_id,
+                    scope="task",
+                    task_brief=task_brief,
+                    task_report=report_payload,
+                )
+            else:
+                is_protocol_error = report.outcome == "protocol_error"
+                validation_run = await runner.validate_recursive_trace(
+                    child_trace_id,
+                    scope="task",
+                    deterministic_failure={
+                        "outcome": "error" if is_protocol_error else "failed",
+                        "reason": report.summary,
+                        "issues": report.remaining_issues or [report.summary],
+                        "retry_from": None if is_protocol_error else "task_definition",
+                    },
+                )
+            validation_result = validation_run.result
+            child_state["task_report_validation"] = {
+                "task_report": report_payload,
+                "validation_result": validation_result.model_dump(),
+            }
+            await store.update_trace(child_trace_id, context=child_trace.context)
         state["pending_reviews"][child_trace_id] = pending_review_entry(
             goal_id=goal_id,
             report=report,
+            validation_result=validation_result.model_dump(),
             received_at_sequence=received_at_sequence,
         )
         reports.append(report)
@@ -723,6 +779,7 @@ async def _run_agents(
     approved_action = None
     if policy.requires_task_protocol:
         protocol_state = ensure_task_protocol(parent_trace.context)
+        parent_task_brief = protocol_state.get("task_brief")
         if goal_id:
             parent_tree = await store.get_goal_tree(trace_id)
             parent_goal = parent_tree.find(goal_id) if parent_tree else None
@@ -748,7 +805,18 @@ async def _run_agents(
                         "status": "failed",
                         "error": "REVISE_CHILD requires continue_from for the reviewed child",
                     }
-                if expected_brief and task_briefs[0].model_dump() != expected_brief:
+                try:
+                    matches_approved = not expected_brief or task_briefs_match(
+                        task_briefs[0],
+                        expected_brief,
+                        parent_task_brief=parent_task_brief,
+                    )
+                except ContextPolicyError as exc:
+                    return {
+                        "status": "failed",
+                        "error": f"Approved Recursive TaskBrief is invalid; recreate the trace: {exc}",
+                    }
+                if not matches_approved:
                     return {
                         "status": "failed",
                         "error": "task_brief does not match the approved revision",
@@ -759,7 +827,18 @@ async def _run_agents(
                         "status": "failed",
                         "error": f"{decision} requires exactly one new approved child",
                     }
-                if not expected_brief or task_briefs[0].model_dump() != expected_brief:
+                try:
+                    matches_approved = bool(expected_brief) and task_briefs_match(
+                        task_briefs[0],
+                        expected_brief,
+                        parent_task_brief=parent_task_brief,
+                    )
+                except ContextPolicyError as exc:
+                    return {
+                        "status": "failed",
+                        "error": f"Approved Recursive TaskBrief is invalid; recreate the trace: {exc}",
+                    }
+                if not matches_approved:
                     return {
                         "status": "failed",
                         "error": "task_brief does not match the approved next task",
@@ -776,6 +855,28 @@ async def _run_agents(
     goal_started = False
     child_records: List[Dict[str, Any]] = []
     all_sub_trace_ids: List[Dict[str, str]] = []
+    created_trace_ids: list[str] = []
+
+    async def fail_created_children(error: Exception) -> None:
+        """Close Recursive children that never reached the execution phase."""
+        for created_trace_id in created_trace_ids:
+            created = await store.get_trace(created_trace_id)
+            if created and created.status == "running":
+                await store.update_trace(
+                    created_trace_id,
+                    status="failed",
+                    error_message=f"Child initialization failed: {error}",
+                    completed_at=datetime.now(),
+                )
+
+    async def run_initialization(operation):
+        """Apply one pre-execution write with consistent child cleanup."""
+        try:
+            return await operation
+        except Exception as exc:
+            if policy.requires_task_protocol:
+                await fail_created_children(exc)
+            raise
 
     if single and continue_from:
         existing = await store.get_trace(continue_from)
@@ -897,6 +998,9 @@ async def _run_agents(
                     created_at=datetime.now(),
                 )
                 await store.create_trace(sub_trace)
+                # A successfully persisted Trace has consumed its Agent budget,
+                # even if a later GoalTree write fails.
+                created_trace_ids.append(stid)
                 await store.update_goal_tree(stid, GoalTree(mission=task_label))
                 all_sub_trace_ids.append({"trace_id": stid, "mission": task_label})
                 child_records.append({
@@ -933,14 +1037,48 @@ async def _run_agents(
                             f"requested={requested_children}, max={child_limit}"
                         ),
                     }
-                await create_child_traces()
-
-        await _update_goal_start(
-            store, trace_id, goal_id,
-            "delegate" if single else "explore",
-            all_sub_trace_ids,
-            accumulate_sub_trace_ids=policy.accumulate_sub_trace_ids,
-            status=("waiting_children" if policy.requires_task_protocol else "in_progress"),
+                resolved_budget = None
+                if policy.requires_task_protocol:
+                    resolved_budget = await runner._resource_budget_for_trace(trace_id)
+                reserved_agents = 0
+                if resolved_budget is not None:
+                    root_trace_id, budget = resolved_budget
+                    if not runner.resource_budget:
+                        return {
+                            "status": "failed",
+                            "error": "ResourceBudgetController is unavailable",
+                        }
+                    try:
+                        await runner.resource_budget.reserve_agents(
+                            root_trace_id,
+                            budget,
+                            requested_children,
+                        )
+                        reserved_agents = requested_children
+                    except (ResourceBudgetExceeded, ResourceBudgetStateError) as exc:
+                        return {"status": "failed", "error": str(exc)}
+                try:
+                    await create_child_traces()
+                except Exception as exc:
+                    created_count = len(created_trace_ids)
+                    uncreated = max(0, reserved_agents - created_count)
+                    if uncreated:
+                        await runner.resource_budget.release_agents(
+                            root_trace_id,
+                            budget,
+                            uncreated,
+                        )
+                    await fail_created_children(exc)
+                    raise
+
+        await run_initialization(
+            _update_goal_start(
+                store, trace_id, goal_id,
+                "delegate" if single else "explore",
+                all_sub_trace_ids,
+                accumulate_sub_trace_ids=policy.accumulate_sub_trace_ids,
+                status=("waiting_children" if policy.requires_task_protocol else "in_progress"),
+            )
         )
         if policy.requires_task_protocol:
             _project_goal_status(context, goal_id, "waiting_children")
@@ -949,7 +1087,9 @@ async def _run_agents(
     if approved_action and protocol_state is not None:
         protocol_state["next_actions"].pop(0)
         protocol_state["protocol_correction_attempts"] = 0
-        await store.update_trace(trace_id, context=parent_trace.context)
+        await run_initialization(
+            store.update_trace(trace_id, context=parent_trace.context)
+        )
 
     # 创建延迟执行规格。Trace 已按批次预创建，但不会提前创建 coroutine，
     # 因而排队孩子被停止时不会产生未 await coroutine 或模型调用。
@@ -961,19 +1101,23 @@ async def _run_agents(
         child_depth = child_record["depth"]
         task_label = task_briefs[i].objective if task_briefs else task_item
         if child_record["is_new"]:
-            await broadcast_sub_trace_started(
-                trace_id, cur_stid, goal_id or "",
-                child_record["agent_type"], task_label,
+            await run_initialization(
+                broadcast_sub_trace_started(
+                    trace_id, cur_stid, goal_id or "",
+                    child_record["agent_type"], task_label,
+                )
             )
 
         # 注册为活跃协作者
         collab_name = task_label[:30] if single and not continued else (
             f"delegate-{cur_stid[:8]}" if single else f"explore-{i+1}"
         )
-        await _update_collaborator(
-            store, trace_id,
-            name=collab_name, sub_trace_id=cur_stid,
-            status="running", summary=task_label[:80],
+        await run_initialization(
+            _update_collaborator(
+                store, trace_id,
+                name=collab_name, sub_trace_id=cur_stid,
+                status="running", summary=task_label[:80],
+            )
         )
 
         # 构建消息
@@ -1015,12 +1159,14 @@ async def _run_agents(
 
     # continue_from 不进入新建临界区，但仍需要恢复 Goal 的运行状态。
     if not goal_started:
-        await _update_goal_start(
-            store, trace_id, goal_id,
-            "delegate" if single else "explore",
-            all_sub_trace_ids,
-            accumulate_sub_trace_ids=policy.accumulate_sub_trace_ids,
-            status=("waiting_children" if policy.requires_task_protocol else "in_progress"),
+        await run_initialization(
+            _update_goal_start(
+                store, trace_id, goal_id,
+                "delegate" if single else "explore",
+                all_sub_trace_ids,
+                accumulate_sub_trace_ids=policy.accumulate_sub_trace_ids,
+                status=("waiting_children" if policy.requires_task_protocol else "in_progress"),
+            )
         )
         if policy.requires_task_protocol:
             _project_goal_status(context, goal_id, "waiting_children")
@@ -1058,12 +1204,16 @@ async def _run_agents(
             if policy.requires_task_protocol:
                 reports = await _record_pending_task_reports(
                     store,
+                    runner,
                     parent_trace,
                     goal_id,
                     [(stid, result)],
                     context.get("sequence", 0),
                 )
                 formatted["task_report"] = reports[0].model_dump()
+                formatted["validation_result"] = ensure_task_protocol(
+                    parent_trace.context
+                )["pending_reviews"][stid]["validation_result"]
                 _project_goal_status(context, goal_id, "pending_review")
             else:
                 await _update_goal_complete(
@@ -1092,12 +1242,16 @@ async def _run_agents(
             if policy.requires_task_protocol:
                 reports = await _record_pending_task_reports(
                     store,
+                    runner,
                     parent_trace,
                     goal_id,
                     [(stid, failed_result)],
                     context.get("sequence", 0),
                 )
                 failed_result["task_report"] = reports[0].model_dump()
+                failed_result["validation_result"] = ensure_task_protocol(
+                    parent_trace.context
+                )["pending_reviews"][stid]["validation_result"]
                 _project_goal_status(context, goal_id, "pending_review")
             else:
                 await _update_goal_complete(
@@ -1170,6 +1324,7 @@ async def _run_agents(
         if policy.requires_task_protocol:
             reports = await _record_pending_task_reports(
                 store,
+                runner,
                 parent_trace,
                 goal_id,
                 [
@@ -1179,6 +1334,11 @@ async def _run_agents(
                 context.get("sequence", 0),
             )
             formatted["task_reports"] = [report.model_dump() for report in reports]
+            pending = ensure_task_protocol(parent_trace.context)["pending_reviews"]
+            formatted["validation_results"] = [
+                pending[entry["trace_id"]]["validation_result"]
+                for entry in all_sub_trace_ids
+            ]
             _project_goal_status(context, goal_id, "pending_review")
         else:
             await _update_goal_complete(
@@ -1347,6 +1507,11 @@ async def agent(
         policy = policy_from_context(parent_trace.context)
     except ValueError as exc:
         return {"status": "failed", "error": str(exc)}
+    if policy.requires_task_protocol and messages is not None:
+        return {
+            "status": "failed",
+            "error": "Recursive revision 2 does not accept messages; pass bounded context in task_brief",
+        }
 
     parsed_task_briefs: Optional[List[TaskBrief]] = None
     if policy.requires_task_protocol:
@@ -1362,8 +1527,16 @@ async def agent(
                 return {"status": "failed", "error": f"Invalid TaskBrief JSON: {exc}"}
         raw_briefs = task_brief if isinstance(task_brief, list) else [task_brief]
         try:
-            parsed_task_briefs = [TaskBrief.model_validate(item) for item in raw_briefs]
-        except ValidationError as exc:
+            parent_state = ensure_task_protocol(parent_trace.context)
+            parent_task_brief = parent_state.get("task_brief")
+            parsed_task_briefs = [
+                normalize_task_brief(
+                    item,
+                    parent_task_brief=parent_task_brief,
+                )
+                for item in raw_briefs
+            ]
+        except (ContextPolicyError, ValidationError) as exc:
             return {"status": "failed", "error": f"Invalid TaskBrief: {exc}"}
         tasks = [format_task_brief(brief) for brief in parsed_task_briefs]
         single = len(tasks) == 1
@@ -1470,6 +1643,11 @@ async def evaluate(
         policy = policy_from_context(parent_trace.context)
     except ValueError as exc:
         return {"status": "failed", "error": str(exc)}
+    if policy.requires_task_protocol:
+        return {
+            "status": "failed",
+            "error": "evaluate is unavailable in Recursive mode; validation is framework-managed",
+        }
 
     # 处理 continue_from 或创建新 Sub-Trace
     if continue_from:

cyber_agent/trace/run_api.py

@@ -68,6 +68,10 @@ class CreateRequest(BaseModel):
     name: Optional[str] = Field(None, description="任务名称（None = 自动生成）")
     uid: Optional[str] = Field(None)
     project_name: Optional[str] = Field(None, description="示例项目名称，若提供则动态加载其执行环境")
+    root_completion_criteria: List[str] = Field(
+        default_factory=list,
+        description="Recursive 根任务必须显式提供的完成标准；Legacy 忽略",
+    )
 
 
 class TraceRunRequest(BaseModel):
@@ -264,6 +268,7 @@ async def create_and_run(req: CreateRequest):
                     enable_research_flow=default_config.enable_research_flow,
                     child_execution_mode=default_config.child_execution_mode,
                     max_parallel_children=default_config.max_parallel_children,
+                    root_completion_criteria=req.root_completion_criteria,
                     context={"project_name": req.project_name}
                 )
         except ImportError as e:
@@ -285,6 +290,7 @@ async def create_and_run(req: CreateRequest):
             tools=req.tools,
             name=req.name,
             uid=req.uid,
+            root_completion_criteria=req.root_completion_criteria,
             context={"project_name": req.project_name} if req.project_name else {}
         )