补齐验收回溯与根门禁测试

tests/test_recursive_replan_context.py

@@ -20,7 +20,11 @@ from cyber_agent.core.task_protocol import (
     pending_review_entry,
     rebuild_pending_replans,
 )
-from cyber_agent.core.validation import ValidationResult
+from cyber_agent.core.validation import (
+    ScopeValidationResult,
+    ValidationCheck,
+    ValidationResult,
+)
 from cyber_agent.tools.builtin.task_protocol import review_task_result
 from cyber_agent.trace.goal_models import Goal, GoalTree
 from cyber_agent.trace.models import Trace
@@ -69,18 +73,32 @@ def validation_result(
     outcome: str = "failed",
     retry_from: str | None = "hypothesis",
 ) -> ValidationResult:
-    return ValidationResult(
-        validator_trace_id=f"{child_trace_id}@validator",
-        evaluated_trace_id=child_trace_id,
-        outcome=outcome,
-        scope="task",
-        reason=(
+    reason = (
             "Independent evidence is still missing"
             if outcome != "passed"
             else "All criteria passed"
-        ),
+        )
+    plan_hash = "b" * 64
+    scope_result = ScopeValidationResult(
+        validator_trace_id=f"{child_trace_id}@validator",
+        scope="task",
+        outcome=outcome,
+        checks=([] if outcome == "error" else [ValidationCheck(
+            check_id="task.fixture",
+            status=outcome,
+            issue=None if outcome == "passed" else reason,
+        )]),
+        reason=reason,
+        retry_from=retry_from,
+        plan_hash=plan_hash,
+    )
+    return ValidationResult(
+        evaluated_trace_id=child_trace_id,
+        outcome=outcome,
+        scope_results=[scope_result],
         issues=[] if outcome == "passed" else ["Missing independent evidence"],
         retry_from=retry_from,
+        plan_hash=plan_hash,
     )
 
 
@@ -180,16 +198,30 @@ class ReplanCurrentTest(unittest.IsolatedAsyncioTestCase):
                 "task_protocol": new_task_protocol(BRIEF),
             }),
         ))
+        child = await self.store.get_trace(child_id)
+        child_state = ensure_task_protocol(child.context)
+        authoritative = validation_result(
+            child_id,
+            outcome=validation_outcome,
+            retry_from=retry_from,
+        )
+        child_state["task_report_validation"] = {
+            "validation_plan": None,
+            "plan_hash": authoritative.plan_hash,
+            "scope_results": [
+                item.model_dump(mode="json")
+                for item in authoritative.scope_results
+            ],
+            "aggregate_result": authoritative.model_dump(mode="json"),
+            "validated_at_sequence": 4,
+        }
+        await self.store.update_trace(child_id, context=child.context)
         parent = await self.store.get_trace(self.parent_id)
         state = ensure_task_protocol(parent.context)
         entry = pending_review_entry(
             goal_id="1",
             report=task_report(child_id),
-            validation_result=validation_result(
-                child_id,
-                outcome=validation_outcome,
-                retry_from=retry_from,
-            ).model_dump(),
+            validation_result=authoritative.model_dump(mode="json"),
             received_at_sequence=4,
         )
         state["pending_reviews"][child_id] = entry
@@ -225,6 +257,53 @@ class ReplanCurrentTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual("in_progress", tree.find("1").status)
         self.assertEqual("1", tree.current_id)
 
+    async def test_unknown_validation_is_recoverable_but_cannot_ascend(self):
+        await self.add_child(
+            "child-unknown",
+            validation_outcome="unknown",
+            retry_from="hypothesis",
+        )
+        rejected = await review_task_result(
+            child_trace_id="child-unknown",
+            decision="ASCEND",
+            reason="must not bypass unknown evidence",
+            context=self.context(),
+        )
+        self.assertEqual("failed", rejected["status"])
+        self.assertIn("only allows", rejected["error"])
+
+        accepted = await review_task_result(
+            child_trace_id="child-unknown",
+            decision="REPLAN_CURRENT",
+            reason="collect the missing material",
+            context=self.context(),
+        )
+        self.assertEqual("completed", accepted["status"])
+        self.assertEqual("hypothesis", accepted["task_review"]["retry_from"])
+
+    async def test_pending_result_cannot_override_child_validation_cache(self):
+        await self.add_child("child-forged")
+        parent = await self.store.get_trace(self.parent_id)
+        state = ensure_task_protocol(parent.context)
+        forged = validation_result(
+            "child-forged",
+            outcome="passed",
+            retry_from=None,
+        )
+        state["pending_reviews"]["child-forged"][
+            "validation_result"
+        ] = forged.model_dump(mode="json")
+        await self.store.update_trace(self.parent_id, context=parent.context)
+
+        result = await review_task_result(
+            child_trace_id="child-forged",
+            decision="ASCEND",
+            reason="forged pass",
+            context=self.context(),
+        )
+        self.assertEqual("failed", result["status"])
+        self.assertIn("authoritative cache", result["error"])
+
     async def test_grandchild_cannot_jump_directly_to_grandparent(self):
         await self.add_child("child-1")
         await self.store.create_trace(Trace(

tests/test_recursive_runtime_extensions.py

@@ -20,6 +20,11 @@ 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
+from cyber_agent.core.validation import (
+    ValidationPolicy,
+    ValidatorSettings,
+    persist_validation_policy,
+)
 
 
 ROOT_CRITERIA = ["return a checked final answer"]
@@ -75,22 +80,39 @@ def tool_names(schemas):
 def is_validator_call(messages):
     return any(
         message.get("role") == "system"
-        and "independent validator" in str(message.get("content", ""))
+        and "recursive_validation_protocol" in str(message.get("content", ""))
         for message in messages
     )
 
 
 def validator_response(messages, outcome="passed"):
-    packet = json.loads(messages[-1]["content"])
+    packet = next(
+        json.loads(message["content"])
+        for message in messages
+        if message.get("role") == "user"
+        and "recursive_validation_protocol" in str(message.get("content", ""))
+    )
     scope = packet["validation_scope"]
     passed = outcome == "passed"
+    checks = [
+        {
+            "check_id": item["check_id"],
+            "status": outcome,
+            "evidence_refs": [],
+            "issue": None if passed else "the answer lacks required support",
+        }
+        for item in packet["validation_plan"]["checks"]
+        if item["scope"] == scope
+    ]
     return {
         "content": json.dumps({
             "outcome": outcome,
             "scope": scope,
+            "checks": checks,
             "reason": "the persisted record was checked",
-            "issues": [] if passed else ["the answer lacks required support"],
-            "retry_from": None if passed else "evidence",
+            "retry_from": None if passed else (
+                "task_definition" if scope in {"task", "root"} else scope
+            ),
         }),
         "tool_calls": [],
         "finish_reason": "stop",
@@ -154,7 +176,10 @@ class RootValidationIntegrationTest(unittest.IsolatedAsyncioTestCase):
         )
         self.assertEqual(1, len(validators))
         self.assertEqual(root.trace_id, validators[0].context["root_trace_id"])
-        self.assertEqual([], validators[0].tools)
+        self.assertEqual(
+            {"validator_web_search", "validator_open_url"},
+            tool_names(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)
@@ -273,7 +298,12 @@ class RootValidationIntegrationTest(unittest.IsolatedAsyncioTestCase):
         async def llm_call(**kwargs):
             nonlocal ordinary_calls, validator_packet
             if is_validator_call(kwargs["messages"]):
-                validator_packet = json.loads(kwargs["messages"][-1]["content"])
+                validator_packet = next(
+                    json.loads(message["content"])
+                    for message in kwargs["messages"]
+                    if message.get("role") == "user"
+                    and "recursive_validation_protocol" in str(message.get("content", ""))
+                )
                 return validator_response(kwargs["messages"])
             ordinary_calls += 1
             if ordinary_calls == 1:
@@ -827,11 +857,10 @@ class RecursiveLifecycleIntegrationTest(unittest.IsolatedAsyncioTestCase):
             child_validators[0].context["evaluated_trace_id"],
         )
         child_state = ensure_task_protocol(child.context)
+        aggregate = child_state["task_report_validation"]["aggregate_result"]
         self.assertEqual(
             child_validators[0].trace_id,
-            child_state["task_report_validation"]["validation_result"][
-                "validator_trace_id"
-            ],
+            aggregate["scope_results"][0]["validator_trace_id"],
         )
         root_state = ensure_task_protocol(root.context)
         self.assertEqual({}, root_state["pending_reviews"])
@@ -857,6 +886,13 @@ class RecursiveLifecycleIntegrationTest(unittest.IsolatedAsyncioTestCase):
                 "task_protocol": new_task_protocol(),
             }),
         ))
+        root = await self.store.get_trace(root_id)
+        persist_validation_policy(
+            root.context,
+            ValidationPolicy(),
+            ValidatorSettings(),
+        )
+        await self.store.update_trace(root_id, context=root.context)
         await self.store.create_trace(Trace(
             trace_id=child_id,
             mode="agent",