接入多 Scope Validator 运行与缓存

cyber_agent/core/runner.py

@@ -76,10 +76,31 @@ from cyber_agent.core.resource_budget import (
     ResourceBudgetExceeded,
     ResourceBudgetStateError,
 )
+from cyber_agent.core.artifacts import (
+    ArtifactRef,
+    ArtifactResolver,
+    MaterialIssue,
+    extract_artifact_refs,
+    material_chars,
+    resolve_artifact_refs,
+)
 from cyber_agent.core.validation import (
     LLMValidator,
+    ScopeValidationResult,
+    ValidationPolicy,
+    ValidationResult,
     ValidationRun,
     ValidationScope,
+    ValidatorSettings,
+    persist_validation_policy,
+    require_validation_policy,
+)
+from cyber_agent.core.validator_web import (
+    PageFetcher,
+    SerperWebSearchProvider,
+    ValidatorToolLimits,
+    ValidatorToolSession,
+    ValidatorWebSearchProvider,
 )
 from cyber_agent.core.prompts import (
     DEFAULT_SYSTEM_PREFIX,
@@ -234,6 +255,10 @@ class AgentRunner:
         goal_tree: Optional[GoalTree] = None,
         debug: bool = False,
         logger_name: Optional[str] = None,
+        validation_policy: Optional[ValidationPolicy] = None,
+        validator_search_provider: Optional[ValidatorWebSearchProvider] = None,
+        validator_page_fetcher: Optional[PageFetcher] = None,
+        artifact_resolver: Optional[ArtifactResolver] = None,
     ):
         """
         初始化 AgentRunner
@@ -247,6 +272,10 @@ class AgentRunner:
             goal_tree: 初始 GoalTree（可选）
             debug: 保留参数（已废弃）
             logger_name: 自定义日志名称（如 "agents.knowledge_manager"），默认用模块名
+            validation_policy: Recursive 根 Trace 固化的可信验收策略
+            validator_search_provider: Validator 私有网页搜索适配器
+            validator_page_fetcher: Validator 受控页面读取适配器
+            artifact_resolver: Validator 只读产物解析器
         """
         self.trace_store = trace_store
         self.tools = tool_registry or get_tool_registry()
@@ -256,6 +285,10 @@ class AgentRunner:
         self.goal_tree = goal_tree
         self.debug = debug
         self.log = logging.getLogger(logger_name) if logger_name else logger
+        self.validation_policy = validation_policy or ValidationPolicy()
+        self.validator_search_provider = validator_search_provider
+        self.validator_page_fetcher = validator_page_fetcher
+        self.artifact_resolver = artifact_resolver
         self.stdin_check: Optional[Callable] = None  # 由外部设置，用于子 agent 执行期间检查 stdin
         self._cancel_events: Dict[str, asyncio.Event] = {}  # trace_id → cancel event
         self._recursive_active_traces: Dict[str, asyncio.Event] = {}
@@ -378,11 +411,34 @@ class AgentRunner:
             cost_usd=float(tool_usage.get("cost", 0) or 0),
         )
 
+    async def record_recursive_validation_usage(
+        self,
+        trace_id: str,
+        *,
+        tool_calls: int = 0,
+        material_chars_count: int = 0,
+        provider_cost_usd: float = 0.0,
+    ) -> None:
+        """把 Validator网页工具和真实材料字符计入同一棵树的预算。"""
+        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_validation_usage(
+            root_trace_id,
+            budget,
+            tool_calls=tool_calls,
+            material_chars=material_chars_count,
+            provider_cost_usd=provider_cost_usd,
+        )
+
     async def validate_recursive_trace(
         self,
         evaluated_trace_id: str,
         *,
-        scope: ValidationScope,
+        scope: Optional[ValidationScope] = None,
         task_brief: Optional[Dict[str, Any]] = None,
         task_report: Optional[Dict[str, Any]] = None,
         completion_criteria: Optional[List[str]] = None,
@@ -391,15 +447,163 @@ class AgentRunner:
         deterministic_failure: Optional[Dict[str, Any]] = None,
         root_validator: bool = False,
     ) -> ValidationRun:
-        """运行一次由框架控制、不带工具的独立验收 Trace。
-
-        子 Agent 结果汇合或根 Agent 候选答案完成时调用，返回权威 ``ValidationResult``。
-        """
+        """编译Plan、解析材料、顺序运行Scope并持久化聚合缓存。"""
         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}")
+        root_trace_id = evaluated.context.get("root_trace_id") or evaluated.trace_id
+        root = (
+            evaluated
+            if root_trace_id == evaluated.trace_id
+            else await self.trace_store.get_trace(root_trace_id)
+        )
+        if not root:
+            raise ContextPolicyError(f"Recursive root Trace not found: {root_trace_id}")
+        root_anchor = require_matching_root_task_anchor(
+            root.context,
+            evaluated.context,
+        )
+        policy, settings = require_validation_policy(root.context)
+        state = ensure_task_protocol(evaluated.context)
+        authoritative_brief = state.get("task_brief")
+        if authoritative_brief is not None:
+            task_brief = authoritative_brief
+        task_brief_version = int(state.get("task_brief_version", 0) or 0)
+        trajectory = await self.trace_store.get_main_path_messages(
+            evaluated_trace_id,
+            evaluated.head_sequence or evaluated.last_sequence,
+        )
+
+        refs: list[ArtifactRef] = []
+        source_urls: list[str] = []
+        material_issues: list[MaterialIssue] = []
+        try:
+            if isinstance(task_report, dict):
+                refs.extend(extract_artifact_refs(task_report))
+                source_urls = list(task_report.get("source_urls") or [])
+            if root_validator:
+                for message in trajectory:
+                    if message.role == "tool" and isinstance(message.content, dict):
+                        refs.extend(extract_artifact_refs(message.content))
+        except Exception as exc:
+            material_issues.append(MaterialIssue(
+                artifact_id="artifact_metadata",
+                outcome="error",
+                reason=f"Invalid artifact metadata: {exc}",
+            ))
+        unique_refs = {
+            (item.artifact_id, item.version, item.content_hash): item for item in refs
+        }
+        materials, resolved_issues = await resolve_artifact_refs(
+            list(unique_refs.values()),
+            resolver=self.artifact_resolver,
+            root_trace_id=root_trace_id,
+            uid=evaluated.uid,
+        )
+        material_issues.extend(resolved_issues)
+        total_material_chars = sum(material_chars(item) for item in materials)
+        if deterministic_failure:
+            material_issues.append(MaterialIssue(
+                artifact_id="execution",
+                outcome=deterministic_failure.get("outcome", "error"),
+                reason=deterministic_failure.get("reason", "Task did not complete"),
+            ))
+
+        default_model = settings.validator_model or evaluated.model or ""
+        root_model = settings.root_validator_model or default_model
+        model_by_scope = {
+            item: (root_model if item == "root" else default_model)
+            for item in ("evidence", "hypothesis", "output", "task", "root")
+        }
+        if scope == "root" or root_validator:
+            root_validator = True
+        elif scope and task_brief is None:
+            task_brief = {
+                "completion_criteria": completion_criteria or [],
+                "expected_outputs": expected_outputs or [],
+                "validation_scopes": [] if scope == "task" else [scope],
+            }
+        plan = policy.compile_plan(
+            task_brief=task_brief,
+            task_brief_version=task_brief_version,
+            root_task_anchor=root_anchor,
+            task_report=task_report,
+            candidate_output=candidate_output,
+            evaluated_head_sequence=evaluated.head_sequence or evaluated.last_sequence,
+            materials=materials,
+            material_issues=material_issues,
+            model_by_scope=model_by_scope,
+            root=root_validator,
+        )
+
+        cached = state.get("task_report_validation")
+        validation_cache: dict[str, Any]
+        resume_scope_results: list[ScopeValidationResult] = []
+        if isinstance(cached, dict) and cached.get("plan_hash") == plan.plan_hash:
+            validation_cache = cached
+            try:
+                aggregate_raw = cached.get("aggregate_result")
+                if aggregate_raw:
+                    aggregate = ValidationResult.model_validate(aggregate_raw)
+                    if (
+                        aggregate.evaluated_trace_id == evaluated_trace_id
+                        and aggregate.plan_hash == plan.plan_hash
+                    ):
+                        return ValidationRun(
+                            result=aggregate,
+                            trace_ids=[
+                                item.validator_trace_id
+                                for item in aggregate.scope_results
+                            ],
+                            cached=True,
+                        )
+                resume_scope_results = [
+                    ScopeValidationResult.model_validate(item)
+                    for item in cached.get("scope_results", [])
+                ]
+            except Exception:
+                resume_scope_results = []
+        else:
+            validation_cache = {
+                "validation_plan": plan.model_dump(mode="json"),
+                "plan_hash": plan.plan_hash,
+                "scope_results": [],
+                "aggregate_result": None,
+                "validated_at_sequence": plan.evaluated_head_sequence,
+                "material_usage_recorded": total_material_chars == 0,
+            }
+            state["task_report_validation"] = validation_cache
+            await self.trace_store.update_trace(
+                evaluated_trace_id,
+                context=evaluated.context,
+            )
+
+        if (
+            total_material_chars
+            and not validation_cache.get("material_usage_recorded", False)
+        ):
+            await self.record_recursive_validation_usage(
+                evaluated_trace_id,
+                material_chars_count=total_material_chars,
+            )
+            fresh = await self.trace_store.get_trace(evaluated_trace_id)
+            if not fresh:
+                raise ValueError(f"Trace not found: {evaluated_trace_id}")
+            fresh_state = ensure_task_protocol(fresh.context)
+            fresh_cache = fresh_state.get("task_report_validation")
+            if (
+                not isinstance(fresh_cache, dict)
+                or fresh_cache.get("plan_hash") != plan.plan_hash
+            ):
+                raise ValueError("Validation cache changed while recording materials")
+            fresh_cache["material_usage_recorded"] = True
+            await self.trace_store.update_trace(
+                evaluated_trace_id,
+                context=fresh.context,
+            )
+
         lineage_event = None
         if (
             evaluated.parent_trace_id
@@ -409,25 +613,13 @@ class AgentRunner:
                 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(
@@ -435,53 +627,115 @@ class AgentRunner:
                 )
             return result
 
+        provider: ValidatorWebSearchProvider | None
+        if settings.search_provider == "disabled":
+            provider = None
+        else:
+            provider = self.validator_search_provider or SerperWebSearchProvider()
+
+        def tool_session_factory(
+            validation_scope: ValidationScope,
+            allowed_urls: set[str],
+            validator_trace_id: str,
+        ) -> ValidatorToolSession | None:
+            limits_by_scope = {
+                "evidence": ValidatorToolLimits(5, 10, 15),
+                "hypothesis": ValidatorToolLimits(2, 4, 6),
+                "root": ValidatorToolLimits(2, 5, 7),
+            }
+            limits = limits_by_scope.get(validation_scope)
+            if limits is None:
+                return None
+
+            async def record_usage(
+                tool_calls: int,
+                chars: int,
+                provider_cost_usd: float,
+            ) -> None:
+                await self.record_recursive_validation_usage(
+                    evaluated_trace_id,
+                    tool_calls=tool_calls,
+                    material_chars_count=chars,
+                    provider_cost_usd=provider_cost_usd,
+                )
+
+            session_kwargs: Dict[str, Any] = {}
+            if self.validator_page_fetcher is not None:
+                session_kwargs["page_fetcher"] = self.validator_page_fetcher
+            return ValidatorToolSession(
+                provider=provider,
+                allowed_urls=allowed_urls,
+                limits=limits,
+                usage_recorder=record_usage,
+                **session_kwargs,
+            )
+
         validator = LLMValidator(
             llm_call=validator_llm_call,
             trace_store=self.trace_store,
+            policy=policy,
+            tool_session_factory=tool_session_factory,
+            cancel_check=self.is_cancel_requested,
+            trace_register=self.register_recursive_child,
+            trace_release=self.release_recursive_trace,
         )
         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,
-            )
-            root_trace_id = evaluated.context.get("root_trace_id") or evaluated.trace_id
-            root = (
-                evaluated
-                if root_trace_id == evaluated.trace_id
-                else await self.trace_store.get_trace(root_trace_id)
-            )
-            if not root:
-                raise ContextPolicyError(
-                    f"Recursive root Trace not found: {root_trace_id}"
+            async def persist_scope(result: ScopeValidationResult) -> None:
+                fresh = await self.trace_store.get_trace(evaluated_trace_id)
+                if not fresh:
+                    raise ValueError(f"Trace not found: {evaluated_trace_id}")
+                fresh_state = ensure_task_protocol(fresh.context)
+                cache = fresh_state.get("task_report_validation")
+                if not isinstance(cache, dict) or cache.get("plan_hash") != plan.plan_hash:
+                    raise ValueError("Validation cache changed while scopes were running")
+                by_scope = {
+                    item.get("scope"): item
+                    for item in cache.get("scope_results", [])
+                    if isinstance(item, dict)
+                }
+                by_scope[result.scope] = result.model_dump(mode="json")
+                cache["scope_results"] = [
+                    by_scope[item]
+                    for item in plan.effective_scopes
+                    if item in by_scope
+                ]
+                await self.trace_store.update_trace(
+                    evaluated_trace_id,
+                    context=fresh.context,
                 )
-            root_anchor = require_matching_root_task_anchor(
-                root.context,
-                evaluated.context,
-            )
-            return await validator.validate(
+
+            run = await validator.validate_plan(
                 evaluated_trace=evaluated,
                 trajectory=trajectory,
-                scope=scope,
+                plan=plan,
                 root_task_anchor=root_anchor,
                 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,
+                materials=materials,
+                material_issues=material_issues,
+                model_by_scope=model_by_scope,
+                source_urls=source_urls,
+                resume_scope_results=resume_scope_results,
+                on_scope_result=persist_scope,
+            )
+            fresh = await self.trace_store.get_trace(evaluated_trace_id)
+            if not fresh:
+                raise ValueError(f"Trace not found: {evaluated_trace_id}")
+            fresh_state = ensure_task_protocol(fresh.context)
+            cache = fresh_state.get("task_report_validation")
+            if not isinstance(cache, dict) or cache.get("plan_hash") != plan.plan_hash:
+                raise ValueError("Validation cache changed before aggregation")
+            cache["aggregate_result"] = run.result.model_dump(mode="json")
+            cache["scope_results"] = [
+                item.model_dump(mode="json") for item in run.result.scope_results
+            ]
+            await self.trace_store.update_trace(
+                evaluated_trace_id,
+                context=fresh.context,
             )
+            return run
         finally:
-            self.release_recursive_trace(validator_trace_id, event)
             if lineage_event is not None:
                 self.release_recursive_trace(evaluated_trace_id, lineage_event)
 
@@ -909,6 +1163,7 @@ class AgentRunner:
             except ContextPolicyError as exc:
                 raise ValueError(str(exc)) from exc
             budget = ResourceBudget.from_environment()
+            validator_settings = ValidatorSettings.from_environment()
         trace_id = str(uuid.uuid4())
 
         # 生成任务名称
@@ -926,6 +1181,11 @@ class AgentRunner:
         trace_context.setdefault("root_trace_id", trace_id)
         if policy.requires_task_protocol:
             persist_root_task_anchor(trace_context, root_task_anchor)
+            persist_validation_policy(
+                trace_context,
+                self.validation_policy,
+                validator_settings,
+            )
             trace_context[RESOURCE_BUDGET_CONTEXT_KEY] = budget.to_dict()
             state = ensure_task_protocol(trace_context)
             replace_context_access(
@@ -1020,8 +1280,11 @@ class AgentRunner:
                             root.context,
                             trace_obj.context,
                         )
+                        require_validation_policy(root.context)
                     except ContextPolicyError as exc:
                         raise ValueError(str(exc)) from exc
+                    except ValueError as exc:
+                        raise ValueError(str(exc)) from exc
                     if RESOURCE_BUDGET_CONTEXT_KEY not in root.context:
                         raise ValueError(
                             "This experimental Recursive trace predates tree resource "
@@ -2360,6 +2623,7 @@ class AgentRunner:
                         tool_text = tool_result.get("text", str(tool_result))
                         tool_images = tool_result.get("images", [])
                         tool_usage = tool_result.get("tool_usage")
+                        artifact_refs = tool_result.get("artifact_refs", [])
                         if tool_images:
                             tool_result_text = tool_text
                             tool_content_for_llm = [{"type": "text", "text": tool_text}]
@@ -2372,7 +2636,7 @@ class AgentRunner:
                         else:
                             tool_result_text = tool_text
                             tool_content_for_llm = tool_text
-                        tool_msg = Message.create(trace_id=trace_id, role="tool", sequence=sequence, goal_id=current_goal_id, parent_sequence=head_seq, tool_call_id=tc["id"], branch_type=side_branch_ctx.type if side_branch_ctx else None, branch_id=side_branch_ctx.branch_id if side_branch_ctx else None, content={"tool_name": tool_name, "result": tool_content_for_llm})
+                        tool_msg = Message.create(trace_id=trace_id, role="tool", sequence=sequence, goal_id=current_goal_id, parent_sequence=head_seq, tool_call_id=tc["id"], branch_type=side_branch_ctx.type if side_branch_ctx else None, branch_id=side_branch_ctx.branch_id if side_branch_ctx else None, content={"tool_name": tool_name, "result": tool_content_for_llm, "artifact_refs": artifact_refs})
                         if self.trace_store:
                             await self.trace_store.add_message(tool_msg)
                             if tool_usage:
@@ -2499,6 +2763,7 @@ class AgentRunner:
                         tool_text = tool_result.get("text", str(tool_result))
                         tool_images = tool_result.get("images", [])
                         tool_usage = tool_result.get("tool_usage")  # 新增：提取tool_usage
+                        artifact_refs = tool_result.get("artifact_refs", [])
     
                         # 处理多模态消息
                         if tool_images:
@@ -2537,7 +2802,7 @@ class AgentRunner:
                             branch_type=side_branch_ctx.type if side_branch_ctx else None,
                             branch_id=side_branch_ctx.branch_id if side_branch_ctx else None,
                             # 存储完整内容：有图片时保留 list（含 image_url），纯文本时存字符串
-                            content={"tool_name": tool_name, "result": tool_content_for_llm},
+                            content={"tool_name": tool_name, "result": tool_content_for_llm, "artifact_refs": artifact_refs},
                         )
     
                         if self.trace_store:
@@ -2739,9 +3004,16 @@ class AgentRunner:
                     candidate_output=response_content,
                     root_validator=True,
                 )
+                refreshed_trace = await self.trace_store.get_trace(trace_id)
+                if not refreshed_trace:
+                    raise ValueError(f"Trace not found after root validation: {trace_id}")
+                trace = refreshed_trace
+                state = ensure_task_protocol(trace.context)
+                validation_cache = state.get("task_report_validation") or {}
                 validation_record = {
                     **validation_run.result.model_dump(),
-                    "evaluated_at_sequence": head_seq,
+                    "validation_plan": validation_cache.get("validation_plan"),
+                    "validated_at_sequence": head_seq,
                 }
                 state["root_validation_history"].append(validation_record)
                 state["root_validation_attempts"] += 1
@@ -2969,6 +3241,13 @@ class AgentRunner:
                 state["task_report"] = None
                 state["task_report_submitted_at_sequence"] = None
                 state["task_report_validation"] = None
+            validation_cache = state.get("task_report_validation")
+            if (
+                isinstance(validation_cache, dict)
+                and int(validation_cache.get("validated_at_sequence", 0) or 0)
+                > cutoff
+            ):
+                state["task_report_validation"] = None
             state["pending_reviews"] = {
                 child_id: entry
                 for child_id, entry in state["pending_reviews"].items()
@@ -3008,6 +3287,22 @@ class AgentRunner:
                     0,
                 )
             state["pending_replans"] = rebuild_pending_replans(state)
+            root_history = [
+                item
+                for item in state.get("root_validation_history", [])
+                if int(
+                    item.get(
+                        "validated_at_sequence",
+                        item.get("evaluated_at_sequence", 0),
+                    )
+                    or 0
+                ) <= cutoff
+            ]
+            state["root_validation_history"] = root_history
+            state["root_validation_attempts"] = len(root_history)
+            state["root_validation_passed"] = bool(
+                root_history and root_history[-1].get("outcome") == "passed"
+            )
             state["protocol_correction_attempts"] = 0
             prune_context_access(trace.context, cutoff)
             await self.trace_store.update_trace(trace_id, context=trace.context)