补齐子报告聚合与审核测试

tests/test_task_protocol.py

@@ -36,7 +36,15 @@ 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
+from cyber_agent.core.validation import (
+    ScopeValidationResult,
+    ValidationCheck,
+    ValidationPolicy,
+    ValidationResult,
+    ValidationRun,
+    ValidatorSettings,
+    persist_validation_policy,
+)
 
 
 BRIEF = {
@@ -77,6 +85,7 @@ def report_payload(*, outcome="satisfied"):
 
 
 def validation_payload(child_trace_id, *, outcome="passed"):
+    plan_hash = "a" * 64
     if outcome == "passed":
         issues = []
         retry_from = None
@@ -86,15 +95,55 @@ def validation_payload(child_trace_id, *, outcome="passed"):
     else:
         issues = ["validator could not evaluate the task"]
         retry_from = None
-    return ValidationResult(
+    scope_result = ScopeValidationResult(
         validator_trace_id=f"{child_trace_id}@validator-fixture",
-        evaluated_trace_id=child_trace_id,
-        outcome=outcome,
         scope="task",
+        outcome=outcome,
+        checks=[ValidationCheck(
+            check_id="task.fixture",
+            status=("unknown" if outcome == "error" else outcome),
+            issue=None if outcome == "passed" else issues[0],
+        )],
         reason="fixture validation",
+        retry_from=retry_from,
+        plan_hash=plan_hash,
+    )
+    return ValidationResult(
+        evaluated_trace_id=child_trace_id,
+        outcome=outcome,
+        scope_results=[scope_result],
         issues=issues,
         retry_from=retry_from,
-    ).model_dump()
+        plan_hash=plan_hash,
+    ).model_dump(mode="json")
+
+
+def passed_validator_response(kwargs):
+    """按当前 ValidationPlan 返回完整的通过结果。"""
+    messages = kwargs.get("messages") or []
+    packet = json.loads(messages[1]["content"])
+    scope = packet["validation_scope"]
+    checks = [
+        {
+            "check_id": item["check_id"],
+            "status": "passed",
+            "evidence_refs": [],
+            "issue": None,
+        }
+        for item in packet["validation_plan"]["checks"]
+        if item["scope"] == scope
+    ]
+    return {
+        "content": json.dumps({
+            "outcome": "passed",
+            "scope": scope,
+            "checks": checks,
+            "reason": "all planned checks passed",
+            "retry_from": None,
+        }),
+        "tool_calls": [],
+        "finish_reason": "stop",
+    }
 
 
 async def create_recursive_root(store, trace_id, *, state=None, uid="user-1"):
@@ -108,6 +157,11 @@ async def create_recursive_root(store, trace_id, *, state=None, uid="user-1"):
         "task_protocol": state or new_task_protocol(),
     }
     persist_root_task_anchor(context, ROOT_ANCHOR)
+    persist_validation_policy(
+        context,
+        ValidationPolicy(),
+        ValidatorSettings(),
+    )
     await store.create_trace(Trace(
         trace_id=trace_id,
         mode="agent",
@@ -165,8 +219,19 @@ class ReportingRunner:
         deterministic_failure=None,
         **_kwargs,
     ):
-        self.validation_count += 1
         evaluated = await self.store.get_trace(evaluated_trace_id)
+        child_state = ensure_task_protocol(evaluated.context)
+        cached = child_state.get("task_report_validation")
+        if isinstance(cached, dict) and cached.get("aggregate_result"):
+            result = ValidationResult.model_validate(cached["aggregate_result"])
+            return ValidationRun(
+                result=result,
+                trace_ids=[
+                    item.validator_trace_id for item in result.scope_results
+                ],
+                cached=True,
+            )
+        self.validation_count += 1
         validator_trace_id = (
             f"{evaluated_trace_id}@validator-fixture-{self.validation_count}"
         )
@@ -180,14 +245,21 @@ class ReportingRunner:
             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,
+        result = ValidationResult.model_validate(
+            validation_payload(evaluated_trace_id, outcome=outcome)
+        )
+        child_state["task_report_validation"] = {
+            "validation_plan": None,
+            "plan_hash": result.plan_hash,
+            "scope_results": [
+                item.model_dump(mode="json") for item in result.scope_results
+            ],
+            "aggregate_result": result.model_dump(mode="json"),
+            "validated_at_sequence": evaluated.head_sequence,
+        }
+        await self.store.update_trace(
+            evaluated_trace_id,
+            context=evaluated.context,
         )
         await self.store.create_trace(Trace(
             trace_id=validator_trace_id,
@@ -203,7 +275,7 @@ class ReportingRunner:
                 "agent_depth": evaluated.context["agent_depth"],
             }),
         ))
-        return ValidationRun(result=result, trace_id=validator_trace_id)
+        return ValidationRun(result=result, trace_ids=[validator_trace_id])
 
     async def run_result(self, messages, config, on_event=None):
         self.configs.append(config)
@@ -305,6 +377,7 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
                 "parent_findings": [],
                 "context": {},
                 "context_refs": [],
+                "validation_scopes": [],
             },
             brief.model_dump(exclude_none=True),
         )
@@ -1093,6 +1166,14 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
             "task_protocol": new_task_protocol(BRIEF),
         })
         child_context["task_protocol"]["task_report"] = failed_report.model_dump()
+        failed_validation = validation_payload(child_id, outcome="failed")
+        child_context["task_protocol"]["task_report_validation"] = {
+            "validation_plan": None,
+            "plan_hash": failed_validation["plan_hash"],
+            "scope_results": failed_validation["scope_results"],
+            "aggregate_result": failed_validation,
+            "validated_at_sequence": 2,
+        }
         await self.store.create_trace(Trace(
             trace_id=child_id,
             mode="agent",
@@ -1107,7 +1188,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"),
+            "validation_result": failed_validation,
             "received_at_sequence": 2,
         }
         await self.store.update_trace(self.root_id, context=root.context)
@@ -1188,18 +1269,9 @@ 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",
-                }
+            messages = kwargs.get("messages") or []
+            if messages and "recursive_validation_protocol" in messages[0]["content"]:
+                return passed_validator_response(kwargs)
             if calls == 1:
                 return {
                     "content": "",
@@ -1250,18 +1322,9 @@ 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 lifecycle behavior is valid",
-                        "issues": [],
-                        "retry_from": None,
-                    }),
-                    "tool_calls": [],
-                    "finish_reason": "stop",
-                }
+            messages = kwargs.get("messages") or []
+            if messages and "recursive_validation_protocol" in messages[0]["content"]:
+                return passed_validator_response(kwargs)
             if calls == 1:
                 return {
                     "content": "",
@@ -1442,6 +1505,50 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
         rewound_tree = await self.store.get_goal_tree(self.trace_id)
         self.assertEqual("pending_review", rewound_tree.find("1").status)
 
+    async def test_rewind_rebuilds_root_validation_history_and_cache(self):
+        root_id = "root-validation-rewind"
+        state = new_task_protocol()
+        passed = validation_payload(root_id, outcome="passed")
+        failed = validation_payload(root_id, outcome="failed")
+        state["root_validation_history"] = [
+            {**passed, "validated_at_sequence": 2},
+            {**failed, "validated_at_sequence": 5},
+        ]
+        state["root_validation_attempts"] = 2
+        state["root_validation_passed"] = False
+        state["task_report_validation"] = {
+            "plan_hash": failed["plan_hash"],
+            "aggregate_result": failed,
+            "scope_results": failed["scope_results"],
+            "validated_at_sequence": 5,
+        }
+        await create_recursive_root(self.store, root_id, state=state)
+        tree = GoalTree(
+            mission="root",
+            goals=[Goal(id="1", description="root", status="in_progress")],
+            current_id="1",
+        )
+        await self.store.update_goal_tree(root_id, tree)
+        for sequence in (1, 2, 3):
+            await self.store.add_message(Message.create(
+                trace_id=root_id,
+                role="user",
+                sequence=sequence,
+                content=f"message-{sequence}",
+            ))
+
+        await AgentRunner(trace_store=self.store)._rewind(root_id, 3, tree)
+
+        rewound = await self.store.get_trace(root_id)
+        state = ensure_task_protocol(rewound.context)
+        self.assertEqual(1, state["root_validation_attempts"])
+        self.assertTrue(state["root_validation_passed"])
+        self.assertEqual(
+            ["passed"],
+            [item["outcome"] for item in state["root_validation_history"]],
+        )
+        self.assertIsNone(state["task_report_validation"])
+
     async def test_root_cannot_finish_with_unconsumed_next_action(self):
         root_id = "next-action-gate-root"
         state = new_task_protocol()