test(reference): 贯通真实候选工具报告审核与采用链路

修复 manage_candidate 经 ToolRegistry 调用时未反序列化请求模型、以及错误构造 ToolResult 导致真实 Runner 无法执行的问题。

参考应用新增真实 AgentRunner walking skeleton：editor 通过 agent 创建 writer，writer 从自己的 ToolSet 获得 manage_candidate/update_task_progress/submit_task_report，框架执行候选质量验收，父级通过 review_task_result 采用精确候选 revision。

原双候选局部修订场景也改为经冻结的应用 ToolRegistry 创建和 fork 候选，不再直接调用 CandidateService 绕过执行门禁。

cyber_agent/tools/builtin/candidate.py

@@ -30,13 +30,17 @@ class CandidateManagementRequest(BaseModel):
     groups=["core"],
 )
 async def manage_candidate(
-    request: CandidateManagementRequest,
+    request: CandidateManagementRequest | dict[str, Any],
     context: dict | None = None,
 ) -> ToolResult | dict[str, Any]:
     if not context:
         return {"status": "failed", "error": "context is required"}
     if context.get("side_branch"):
         return {"status": "failed", "error": "Side branches cannot manage candidates"}
+    try:
+        request = CandidateManagementRequest.model_validate(request)
+    except ValueError as exc:
+        return {"status": "failed", "error": f"Invalid candidate request: {exc}"}
     service = context.get("candidate_service")
     trace_id = context.get("trace_id")
     if service is None or not trace_id:
@@ -56,7 +60,8 @@ async def manage_candidate(
         return {"status": "failed", "error": str(exc)}
     payload = candidate.model_dump(mode="json")
     return ToolResult(
-        text=json.dumps({
+        title="",
+        output=json.dumps({
             "status": "completed",
             "candidate_ref": payload,
         }, ensure_ascii=False),

tests/test_application_reference.py

@@ -9,13 +9,18 @@ from cyber_agent.application import (
     ApplicationRuntime,
     CandidateReviewAction,
 )
-from cyber_agent.application.candidate import CandidateLedger, CandidatePointer
+from cyber_agent.application.candidate import (
+    CandidateLedger,
+    CandidatePointer,
+    CandidateRef,
+)
 from cyber_agent.core.artifacts import ArtifactRef
 from cyber_agent.core.run_snapshot import (
     RunConfigSnapshotV2,
     persist_run_config_snapshot,
 )
 from cyber_agent.core.runner import RunConfig
+from cyber_agent.tools.builtin.knowledge import KnowledgeConfig
 from cyber_agent.core.task_protocol import (
     Finding,
     Hypothesis,
@@ -28,10 +33,222 @@ from cyber_agent.core.task_protocol import (
 )
 from cyber_agent.trace.models import Message, Trace
 from cyber_agent.trace.store import FileSystemTraceStore
+from cyber_agent.tools.builtin.subagent import _record_pending_task_reports
 from examples.application_reference import build_reference_components
 
 
+def _tool_call(call_id, name, arguments):
+    return {
+        "content": "",
+        "tool_calls": [{
+            "id": call_id,
+            "type": "function",
+            "function": {
+                "name": name,
+                "arguments": json.dumps(arguments),
+            },
+        }],
+        "finish_reason": "tool_calls",
+    }
+
+
+def _has_tool_result(messages, name):
+    return any(
+        item.get("role") == "tool" and item.get("name") == name
+        for item in messages
+    )
+
+
+def _tool_names(schemas):
+    return {item["function"]["name"] for item in schemas or []}
+
+
 class ApplicationReferenceEndToEndTest(unittest.IsolatedAsyncioTestCase):
+    async def test_real_agent_candidate_report_review_and_adoption_chain(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            store = FileSystemTraceStore(temp_dir)
+            components = build_reference_components()
+            registry = ApplicationRegistry()
+            registry.register(components.application, components.services)
+            delegated = False
+            reviewed = False
+
+            async def llm_call(**kwargs):
+                nonlocal delegated, reviewed
+                messages = kwargs["messages"]
+                if any(
+                    "recursive_validation_protocol" in str(item.get("content", ""))
+                    for item in messages
+                    if item.get("role") == "system"
+                ):
+                    packet = next(
+                        json.loads(item["content"])
+                        for item in messages
+                        if item.get("role") == "user"
+                        and "recursive_validation_protocol"
+                        in str(item.get("content", ""))
+                    )
+                    scope = packet["validation_scope"]
+                    return {
+                        "content": json.dumps({
+                            "scope": scope,
+                            "outcome": "passed",
+                            "checks": [{
+                                "check_id": item["check_id"],
+                                "status": "passed",
+                                "evidence_refs": [],
+                                "issue": None,
+                            } for item in packet["validation_plan"]["checks"]
+                            if item["scope"] == scope],
+                            "reason": "reference output is complete",
+                            "retry_from": None,
+                        }),
+                        "tool_calls": [],
+                        "finish_reason": "stop",
+                    }
+                system = "\n".join(
+                    str(item.get("content", ""))
+                    for item in messages
+                    if item.get("role") == "system"
+                )
+                names = _tool_names(kwargs.get("tools"))
+                if "Create replaceable candidate versions" in system:
+                    if not _has_tool_result(messages, "manage_candidate"):
+                        self.assertIn("manage_candidate", names)
+                        self.assertIn("submit_task_report", names)
+                        return _tool_call("candidate-create", "manage_candidate", {
+                            "request": {
+                                "operation": "create",
+                                "content": {"text": "A finished candidate."},
+                                "parent_refs": [],
+                            },
+                        })
+                    candidate_message = next(
+                        item for item in reversed(messages)
+                        if item.get("role") == "tool"
+                        and item.get("name") == "manage_candidate"
+                    )
+                    candidate_ref = json.loads(
+                        candidate_message["content"]
+                    )["candidate_ref"]
+                    if not _has_tool_result(messages, "update_task_progress"):
+                        return _tool_call("writer-progress", "update_task_progress", {
+                            "expected_revision": 1,
+                            "progress": {
+                                "phase": "ready_to_submit",
+                                "questions": [],
+                                "blockers": [],
+                                "findings": [],
+                                "hypotheses": [],
+                                "work_items": [],
+                                "decision_rationale": "The exact candidate is registered.",
+                            },
+                        })
+                    if not _has_tool_result(messages, "submit_task_report"):
+                        return _tool_call("writer-report", "submit_task_report", {
+                            "task_report": {
+                                "summary": "One complete candidate is ready.",
+                                "outcome": "satisfied",
+                                "validation": {
+                                    "hard_passed": True,
+                                    "open_issues": [],
+                                },
+                                "next_step_suggestion": {
+                                    "direction": "NONE",
+                                    "reason": "Parent may adopt the exact revision.",
+                                },
+                                "outputs": [],
+                                "evidence": [],
+                                "remaining_issues": [],
+                                "candidate_refs": [candidate_ref],
+                            },
+                        })
+                    return {"content": "writer submitted", "tool_calls": []}
+                if names == {"review_task_result"}:
+                    root = next(
+                        item for item in await store.list_traces(limit=20)
+                        if item.parent_trace_id is None
+                    )
+                    child_id, pending = next(iter(
+                        root.context["task_protocol"]["pending_reviews"].items()
+                    ))
+                    candidate_ref = pending["task_report"]["candidate_refs"][0]
+                    reviewed = True
+                    return _tool_call("adopt-review", "review_task_result", {
+                        "child_trace_id": child_id,
+                        "decision": "ASCEND",
+                        "reason": "Adopt the validated exact candidate revision.",
+                        "candidate_actions": [{
+                            "action": "adopt",
+                            "candidate_ref": candidate_ref,
+                            "reason": "The candidate passed independent validation.",
+                        }],
+                    })
+                if not delegated:
+                    delegated = True
+                    return _tool_call("delegate-writer", "agent", {
+                        "agent_type": "writer",
+                        "task_brief": {
+                            "objective": "Produce one finished candidate",
+                            "reason": "The editor needs an adoptable output",
+                            "completion_criteria": ["No placeholder remains"],
+                            "expected_outputs": ["One candidate revision"],
+                            "validation_scopes": ["output"],
+                        },
+                    })
+                if reviewed and not _has_tool_result(
+                    messages,
+                    "update_task_progress",
+                ):
+                    return _tool_call("root-progress", "update_task_progress", {
+                        "expected_revision": 1,
+                        "progress": {
+                            "phase": "ready_to_submit",
+                            "questions": [],
+                            "blockers": [],
+                            "findings": [],
+                            "hypotheses": [],
+                            "work_items": [],
+                            "decision_rationale": "The validated candidate was adopted.",
+                        },
+                    })
+                return {"content": "Final adopted content", "tool_calls": []}
+
+            runtime = ApplicationRuntime(
+                registry=registry,
+                trace_store=store,
+                llm_call=llm_call,
+            )
+            runner, config = runtime.new_run(
+                "application_reference",
+                "1",
+                uid="reference-user",
+                root_task_anchor={
+                    "objective": "Create one complete output",
+                    "completion_criteria": ["Adopt one validated candidate"],
+                    "constraints": ["No placeholders"],
+                },
+            )
+            config.knowledge = KnowledgeConfig(
+                enable_extraction=False,
+                enable_completion_extraction=False,
+                enable_injection=False,
+            )
+            with patch.dict("os.environ", {"AGENT_MODE": "recursive"}, clear=False):
+                result = await runner.run_result(
+                    [{"role": "user", "content": "Create the output"}],
+                    config,
+                )
+            self.assertEqual("completed", result["status"])
+            self.assertTrue(reviewed)
+            self.assertEqual(1, len(components.candidates.versions))
+            self.assertEqual(1, components.candidates.adoption_calls)
+            root = await store.get_trace(result["trace_id"])
+            ledger = CandidateLedger.model_validate(
+                await store.get_candidate_ledger(root.trace_id)
+            )
+            self.assertEqual("adopted", ledger.current_state(ledger.candidates[0]))
+
     async def test_local_candidate_retry_preserves_fact_and_peer_validation(self):
         with tempfile.TemporaryDirectory() as temp_dir:
             store = FileSystemTraceStore(temp_dir)
@@ -102,6 +319,26 @@ class ApplicationReferenceEndToEndTest(unittest.IsolatedAsyncioTestCase):
             ))
             await store.update_trace(root.trace_id, head_sequence=2)
 
+            async def execute_tool(trace_id, tool_name, arguments, sequence):
+                result = await binding.tool_registry.execute(
+                    tool_name,
+                    arguments,
+                    uid=root.uid,
+                    context={
+                        "store": store,
+                        "trace_id": trace_id,
+                        "sequence": sequence,
+                        "task_protocol_service": runner.task_protocol_service,
+                        "candidate_service": runner.candidate_service,
+                        "event_service": runner.event_service,
+                    },
+                    allowed_tool_names={tool_name},
+                )
+                if isinstance(result, str):
+                    return json.loads(result)
+                text = result.get("text") if isinstance(result, dict) else None
+                return json.loads(text) if isinstance(text, str) else result
+
             async def create_child(trace_id, role_id, brief):
                 context = deepcopy(root.context)
                 context.pop("run_config_snapshot", None)
@@ -214,20 +451,28 @@ class ApplicationReferenceEndToEndTest(unittest.IsolatedAsyncioTestCase):
                     validation_scopes=["output"],
                 ),
             )
-            candidate_a = await runner.candidate_service.manage(
+            candidate_a = CandidateRef.model_validate((await execute_tool(
                 writer.trace_id,
-                operation="create",
-                content={"text": "A finished explanation of reusable contracts."},
-                parent_refs=[],
-                effective_at_sequence=4,
-            )
-            candidate_b = await runner.candidate_service.manage(
+                "manage_candidate",
+                {"request": {
+                    "operation": "create",
+                    "content": {
+                        "text": "A finished explanation of reusable contracts."
+                    },
+                    "parent_refs": [],
+                }},
+                4,
+            ))["candidate_ref"])
+            candidate_b = CandidateRef.model_validate((await execute_tool(
                 writer.trace_id,
-                operation="create",
-                content={"text": "B says {{placeholder}}."},
-                parent_refs=[],
-                effective_at_sequence=5,
-            )
+                "manage_candidate",
+                {"request": {
+                    "operation": "create",
+                    "content": {"text": "B says {{placeholder}}."},
+                    "parent_refs": [],
+                }},
+                5,
+            ))["candidate_ref"])
             await runner.task_protocol_service.update_progress(
                 writer.trace_id,
                 expected_revision=1,
@@ -304,16 +549,25 @@ class ApplicationReferenceEndToEndTest(unittest.IsolatedAsyncioTestCase):
                 )],
                 effective_at_sequence=7,
             )
-            candidate_b2 = await restored_runner.candidate_service.manage(
+            runner = restored_runner
+            binding = restarted_registry.resolve("application_reference", "1")
+            candidate_b2 = CandidateRef.model_validate((await execute_tool(
                 writer.trace_id,
-                operation="fork",
-                content={"text": "B2 is a finished explanation of reusable contracts."},
-                parent_refs=[CandidatePointer(
-                    candidate_id=candidate_b.candidate_id,
-                    revision=candidate_b.revision,
-                )],
-                effective_at_sequence=8,
-            )
+                "manage_candidate",
+                {"request": {
+                    "operation": "fork",
+                    "content": {
+                        "text": (
+                            "B2 is a finished explanation of reusable contracts."
+                        )
+                    },
+                    "parent_refs": [CandidatePointer(
+                        candidate_id=candidate_b.candidate_id,
+                        revision=candidate_b.revision,
+                    ).model_dump(mode="json")],
+                }},
+                8,
+            ))["candidate_ref"])
             validation_b2 = await restored_runner.validate_recursive_trace(
                 writer.trace_id,
                 candidate_ref=candidate_b2,