feat(candidate): 增加候选引用根级账本仓储端口和候选服务

新增不可变 CandidateRef、CandidatePointer、CandidateRepository 与 CandidateVersionRequest；候选正文始终由业务 Repository 保存，框架只持有 ArtifactRef、归属、合同和血缘。

根 Trace 的 candidate_ledger.json 使用现有原子写和根级锁保存 append-only 候选、生命周期与幂等操作。CandidateService 统一处理 create、fork、merge，校验 application/root/owner/uid、Task 合同、未来 revision、循环血缘和 Repository 返回的 ArtifactRef。

manage_candidate 作为 lifecycle-exclusive 工具接入现有 Runner，不形成第二套执行器。外部保存成功但最终账本发布失败时保留 pending intent，恢复后以同一 operation ID 安全重放。

同时抽出共享 TaskContractRef，避免 TaskProgress 与 Candidate 复制合同模型；测试覆盖并发幂等、跨 root/Task 越权、错误 Artifact、merge 血缘和账本损坏。

cyber_agent/application/__init__.py

@@ -1,5 +1,7 @@
 """Experimental application framework API (no compatibility promise before M6)."""
 
+from importlib import import_module
+
 from cyber_agent.application.models import (
     AgentApplication,
     AgentRole,
@@ -11,18 +13,34 @@ from cyber_agent.application.models import (
     ToolSet,
     ToolSpec,
 )
-from cyber_agent.application.ports import (
-    ContextMaterial,
-    ContextRequest,
-    ContextResourceDescriptor,
-    TaskContextProvider,
-)
-from cyber_agent.application.runtime import (
-    ApplicationBinding,
-    ApplicationRegistry,
-    ApplicationRuntime,
-    ApplicationServices,
-)
+_LAZY_EXPORTS = {
+    "ApplicationBinding": ("cyber_agent.application.runtime", "ApplicationBinding"),
+    "ApplicationRegistry": ("cyber_agent.application.runtime", "ApplicationRegistry"),
+    "ApplicationRuntime": ("cyber_agent.application.runtime", "ApplicationRuntime"),
+    "ApplicationServices": ("cyber_agent.application.runtime", "ApplicationServices"),
+    "CandidateAdoptionReceipt": ("cyber_agent.application.candidate", "CandidateAdoptionReceipt"),
+    "CandidateAdoptionRequest": ("cyber_agent.application.candidate", "CandidateAdoptionRequest"),
+    "CandidateLedger": ("cyber_agent.application.candidate", "CandidateLedger"),
+    "CandidatePointer": ("cyber_agent.application.candidate", "CandidatePointer"),
+    "CandidateRef": ("cyber_agent.application.candidate", "CandidateRef"),
+    "CandidateRepository": ("cyber_agent.application.candidate", "CandidateRepository"),
+    "CandidateVersionRequest": ("cyber_agent.application.candidate", "CandidateVersionRequest"),
+    "ContextMaterial": ("cyber_agent.application.ports", "ContextMaterial"),
+    "ContextRequest": ("cyber_agent.application.ports", "ContextRequest"),
+    "ContextResourceDescriptor": ("cyber_agent.application.ports", "ContextResourceDescriptor"),
+    "TaskContextProvider": ("cyber_agent.application.ports", "TaskContextProvider"),
+}
+
+
+def __getattr__(name):
+    target = _LAZY_EXPORTS.get(name)
+    if target is None:
+        raise AttributeError(name)
+    module_name, attribute = target
+    value = getattr(import_module(module_name), attribute)
+    globals()[name] = value
+    return value
+
 
 __all__ = [
     "AgentApplication",
@@ -32,6 +50,13 @@ __all__ = [
     "ApplicationRegistry",
     "ApplicationRuntime",
     "ApplicationServices",
+    "CandidateAdoptionReceipt",
+    "CandidateAdoptionRequest",
+    "CandidateLedger",
+    "CandidatePointer",
+    "CandidateRef",
+    "CandidateRepository",
+    "CandidateVersionRequest",
     "ContextMaterial",
     "ContextRequest",
     "ContextResourceDescriptor",

cyber_agent/application/candidate.py

@@ -0,0 +1,173 @@
+"""Application-owned candidate content behind framework-owned immutable references."""
+
+from __future__ import annotations
+
+from typing import Any, Literal, Protocol, runtime_checkable
+
+from pydantic import Field, model_validator
+
+from cyber_agent.application.models import ApplicationModel, ApplicationRef, canonical_json
+from cyber_agent.core.artifacts import ArtifactRef
+from cyber_agent.core.task_contract import TaskContractRef
+
+
+MAX_CANDIDATES_PER_ROOT = 256
+MAX_CANDIDATE_OPERATIONS_PER_ROOT = 2_048
+
+
+class CandidatePointer(ApplicationModel):
+    candidate_id: str = Field(min_length=1, max_length=200)
+    revision: int = Field(ge=1)
+
+
+class CandidateRef(CandidatePointer):
+    schema_version: Literal[1] = 1
+    application_ref: ApplicationRef
+    root_trace_id: str = Field(min_length=1)
+    owner_trace_id: str = Field(min_length=1)
+    task_contract_ref: TaskContractRef
+    artifact_ref: ArtifactRef
+    parent_refs: tuple[CandidatePointer, ...] = ()
+    created_at_sequence: int = Field(ge=0)
+
+    @model_validator(mode="after")
+    def validate_parent_refs(self) -> "CandidateRef":
+        identities = [
+            (item.candidate_id, item.revision)
+            for item in self.parent_refs
+        ]
+        if len(identities) != len(set(identities)):
+            raise ValueError("CandidateRef parent_refs must be unique")
+        if (self.candidate_id, self.revision) in set(identities):
+            raise ValueError("CandidateRef cannot reference itself")
+        return self
+
+
+class CandidateVersionRequest(ApplicationModel):
+    operation_id: str = Field(min_length=1, max_length=300)
+    application_ref: ApplicationRef
+    root_trace_id: str = Field(min_length=1)
+    owner_trace_id: str = Field(min_length=1)
+    uid: str | None = None
+    candidate_id: str = Field(min_length=1, max_length=200)
+    revision: int = Field(ge=1)
+    task_contract_ref: TaskContractRef
+    parent_refs: tuple[CandidatePointer, ...] = ()
+    content: Any
+
+    @model_validator(mode="after")
+    def validate_content(self) -> "CandidateVersionRequest":
+        canonical_json(self.content)
+        return self
+
+
+class CandidateAdoptionRequest(ApplicationModel):
+    operation_id: str = Field(min_length=1, max_length=300)
+    candidate_ref: CandidateRef
+
+
+class CandidateAdoptionReceipt(ApplicationModel):
+    operation_id: str = Field(min_length=1, max_length=300)
+    external_id: str = Field(min_length=1, max_length=500)
+    committed: bool = True
+
+
+@runtime_checkable
+class CandidateRepository(Protocol):
+    async def put_version(
+        self,
+        request: CandidateVersionRequest,
+    ) -> ArtifactRef: ...
+
+    async def load_version(self, candidate_ref: CandidateRef) -> Any: ...
+
+    async def merge(
+        self,
+        request: CandidateVersionRequest,
+    ) -> ArtifactRef: ...
+
+    async def commit_adoption(
+        self,
+        request: CandidateAdoptionRequest,
+    ) -> CandidateAdoptionReceipt: ...
+
+
+class CandidateLifecycleRecord(ApplicationModel):
+    operation_id: str = Field(min_length=1, max_length=300)
+    candidate: CandidatePointer
+    state: Literal[
+        "proposed",
+        "adoption_pending",
+        "adopted",
+        "discarded",
+        "adoption_failed",
+    ]
+    effective_at_sequence: int = Field(ge=0)
+    reason: str | None = Field(default=None, max_length=2_000)
+    receipt: CandidateAdoptionReceipt | None = None
+
+
+class CandidateOperationRecord(ApplicationModel):
+    operation_id: str = Field(min_length=1, max_length=300)
+    operation: Literal["create", "fork", "merge", "adopt", "discard"]
+    input_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
+    status: Literal["pending", "completed", "failed"] = "pending"
+    candidate: CandidatePointer | None = None
+    error: str | None = Field(default=None, max_length=2_000)
+
+
+class CandidateLedger(ApplicationModel):
+    schema_version: Literal[1] = 1
+    candidates: tuple[CandidateRef, ...] = ()
+    lifecycle: tuple[CandidateLifecycleRecord, ...] = ()
+    operations: tuple[CandidateOperationRecord, ...] = ()
+    validations: tuple[dict[str, Any], ...] = ()
+
+    @model_validator(mode="after")
+    def validate_ledger(self) -> "CandidateLedger":
+        if len(self.candidates) > MAX_CANDIDATES_PER_ROOT:
+            raise ValueError("Candidate ledger candidate limit exceeded")
+        if len(self.operations) > MAX_CANDIDATE_OPERATIONS_PER_ROOT:
+            raise ValueError("Candidate ledger operation limit exceeded")
+        identities = [
+            (item.candidate_id, item.revision)
+            for item in self.candidates
+        ]
+        if len(identities) != len(set(identities)):
+            raise ValueError("Candidate ledger contains duplicate revisions")
+        operation_ids = [item.operation_id for item in self.operations]
+        if len(operation_ids) != len(set(operation_ids)):
+            raise ValueError("Candidate ledger contains duplicate operation IDs")
+        known = set(identities)
+        for candidate in self.candidates:
+            for parent in candidate.parent_refs:
+                parent_key = (parent.candidate_id, parent.revision)
+                if parent_key not in known:
+                    raise ValueError("Candidate ledger contains an unknown parent")
+                if (
+                    parent.candidate_id == candidate.candidate_id
+                    and parent.revision >= candidate.revision
+                ):
+                    raise ValueError("Candidate lineage references a future revision")
+        return self
+
+    def candidate(self, pointer: CandidatePointer) -> CandidateRef:
+        for item in self.candidates:
+            if (
+                item.candidate_id == pointer.candidate_id
+                and item.revision == pointer.revision
+            ):
+                return item
+        raise ValueError(
+            f"Candidate revision not found: {pointer.candidate_id}@{pointer.revision}"
+        )
+
+    def current_state(self, pointer: CandidatePointer) -> str | None:
+        state = None
+        for item in self.lifecycle:
+            if (
+                item.candidate.candidate_id == pointer.candidate_id
+                and item.candidate.revision == pointer.revision
+            ):
+                state = item.state
+        return state

cyber_agent/application/candidate_service.py

@@ -0,0 +1,337 @@
+"""Candidate command service with one root-level consistency boundary."""
+
+from __future__ import annotations
+
+import asyncio
+from hashlib import sha256
+from typing import Any
+from weakref import WeakValueDictionary
+
+from cyber_agent.application.candidate import (
+    CandidateLedger,
+    CandidateLifecycleRecord,
+    CandidateOperationRecord,
+    CandidatePointer,
+    CandidateRef,
+    CandidateRepository,
+    CandidateVersionRequest,
+)
+from cyber_agent.application.models import canonical_json
+from cyber_agent.core.agent_mode import require_mutable_trace_policy
+from cyber_agent.core.artifacts import ArtifactRef, ArtifactResolver, resolve_artifact_refs
+from cyber_agent.core.task_protocol import ensure_task_protocol, task_contract_ref
+
+
+class CandidateStateError(ValueError):
+    pass
+
+
+class CandidateService:
+    """Validate, persist, and recover candidate version operations."""
+
+    _locks: "WeakValueDictionary[str, asyncio.Lock]" = WeakValueDictionary()
+
+    def __init__(
+        self,
+        *,
+        store: Any,
+        application_binding: Any,
+        repository: CandidateRepository,
+        artifact_resolver: ArtifactResolver,
+    ) -> None:
+        self.store = store
+        self.binding = application_binding
+        self.repository = repository
+        self.artifact_resolver = artifact_resolver
+
+    @classmethod
+    def _lock_for(cls, root_trace_id: str) -> asyncio.Lock:
+        lock = cls._locks.get(root_trace_id)
+        if lock is None:
+            lock = asyncio.Lock()
+            cls._locks[root_trace_id] = lock
+        return lock
+
+    async def manage(
+        self,
+        trace_id: str,
+        *,
+        operation: str,
+        content: Any,
+        parent_refs: list[CandidatePointer],
+        effective_at_sequence: int,
+    ) -> CandidateRef:
+        trace, root = await self._require_owner_trace(trace_id)
+        root_trace_id = root.trace_id
+        operation_id = f"candidate:{trace_id}:{effective_at_sequence}"
+        input_payload = {
+            "operation": operation,
+            "content": content,
+            "parent_refs": [item.model_dump(mode="json") for item in parent_refs],
+        }
+        input_hash = sha256(
+            canonical_json(input_payload).encode("utf-8")
+        ).hexdigest()
+        async with self._lock_for(root_trace_id):
+            ledger = await self._load_ledger(root_trace_id)
+            existing_operation = next(
+                (
+                    item
+                    for item in ledger.operations
+                    if item.operation_id == operation_id
+                ),
+                None,
+            )
+            if existing_operation is not None:
+                if existing_operation.input_hash != input_hash:
+                    raise CandidateStateError(
+                        "Candidate operation payload changed during recovery"
+                    )
+                if existing_operation.status == "completed":
+                    if existing_operation.candidate is None:
+                        raise CandidateStateError(
+                            "Completed candidate operation has no result"
+                        )
+                    return ledger.candidate(existing_operation.candidate)
+                if existing_operation.status == "failed":
+                    raise CandidateStateError(
+                        existing_operation.error or "Candidate operation failed"
+                    )
+            else:
+                try:
+                    pointer = self._allocate_pointer(
+                        operation,
+                        trace_id,
+                        effective_at_sequence,
+                        parent_refs,
+                        ledger,
+                    )
+                except ValueError as exc:
+                    raise CandidateStateError(str(exc)) from exc
+                operation_record = CandidateOperationRecord(
+                    operation_id=operation_id,
+                    operation=operation,
+                    input_hash=input_hash,
+                    candidate=pointer,
+                )
+                ledger = ledger.model_copy(update={
+                    "operations": (*ledger.operations, operation_record),
+                })
+                await self.store.replace_candidate_ledger(
+                    root_trace_id,
+                    ledger.model_dump(mode="json"),
+                )
+                existing_operation = operation_record
+
+            assert existing_operation.candidate is not None
+            pointer = existing_operation.candidate
+            try:
+                parents = [ledger.candidate(item) for item in parent_refs]
+                self._validate_parents(trace, root, parents)
+                state = ensure_task_protocol(trace.context)
+                request = CandidateVersionRequest(
+                    operation_id=operation_id,
+                    application_ref=self.binding.application_ref,
+                    root_trace_id=root_trace_id,
+                    owner_trace_id=trace_id,
+                    uid=trace.uid,
+                    candidate_id=pointer.candidate_id,
+                    revision=pointer.revision,
+                    task_contract_ref=task_contract_ref(
+                        state,
+                        root_task_anchor_hash=trace.context.get(
+                            "root_task_anchor_hash"
+                        ),
+                    ),
+                    parent_refs=tuple(parent_refs),
+                    content=content,
+                )
+                artifact_ref = ArtifactRef.model_validate(
+                    await self.repository.merge(request)
+                    if operation == "merge"
+                    else await self.repository.put_version(request)
+                )
+                materials, issues = await resolve_artifact_refs(
+                    [artifact_ref],
+                    resolver=self.artifact_resolver,
+                    root_trace_id=root_trace_id,
+                    uid=trace.uid,
+                )
+                if issues or len(materials) != 1:
+                    reason = issues[0].reason if issues else "Artifact was not resolved"
+                    raise CandidateStateError(
+                        f"CandidateRepository returned an invalid ArtifactRef: {reason}"
+                    )
+                candidate_ref = CandidateRef(
+                    **pointer.model_dump(mode="json"),
+                    application_ref=self.binding.application_ref,
+                    root_trace_id=root_trace_id,
+                    owner_trace_id=trace_id,
+                    task_contract_ref=request.task_contract_ref,
+                    artifact_ref=artifact_ref,
+                    parent_refs=tuple(parent_refs),
+                    created_at_sequence=effective_at_sequence,
+                )
+            except Exception as exc:
+                failed = self._fail_operation(ledger, operation_id, str(exc))
+                await self.store.replace_candidate_ledger(
+                    root_trace_id,
+                    failed.model_dump(mode="json"),
+                )
+                raise
+            completed = self._complete_operation(
+                ledger,
+                operation_id,
+                candidate_ref,
+            )
+            # If this atomic publish fails after the Repository committed, the
+            # pending operation remains recoverable. Re-execution uses the same
+            # operation_id and the Repository's idempotency contract.
+            await self.store.replace_candidate_ledger(
+                root_trace_id,
+                completed.model_dump(mode="json"),
+            )
+            return candidate_ref
+
+    async def validate_report_refs(
+        self,
+        trace_id: str,
+        refs: list[CandidateRef],
+    ) -> None:
+        trace, root = await self._require_owner_trace(trace_id)
+        async with self._lock_for(root.trace_id):
+            ledger = await self._load_ledger(root.trace_id)
+            for ref in refs:
+                persisted = ledger.candidate(ref)
+                if persisted != ref:
+                    raise CandidateStateError(
+                        "TaskReport CandidateRef does not match the root ledger"
+                    )
+                self._validate_candidate_owner(persisted, trace, root)
+                if ledger.current_state(ref) != "proposed":
+                    raise CandidateStateError(
+                        "TaskReport may only reference proposed candidates"
+                    )
+
+    async def _require_owner_trace(self, trace_id: str):
+        trace = await self.store.get_trace(trace_id)
+        if trace is None:
+            raise CandidateStateError(f"Trace not found: {trace_id}")
+        policy = require_mutable_trace_policy(trace.context)
+        if not policy.requires_task_progress:
+            raise CandidateStateError(
+                "Candidates require Recursive revision 3"
+            )
+        application_ref = trace.context.get("application_ref")
+        if application_ref != self.binding.application_ref.model_dump(mode="json"):
+            raise CandidateStateError("Candidate Trace application binding mismatch")
+        root_trace_id = trace.context.get("root_trace_id") or trace.trace_id
+        root = trace if trace.trace_id == root_trace_id else await self.store.get_trace(
+            root_trace_id
+        )
+        if root is None or root.uid != trace.uid:
+            raise CandidateStateError("Candidate root or owner mismatch")
+        if root.context.get("application_ref") != application_ref:
+            raise CandidateStateError("Candidate root application mismatch")
+        state = ensure_task_protocol(trace.context)
+        if state.get("task_report") is not None:
+            raise CandidateStateError("Candidates cannot change after TaskReport submission")
+        if state.get("pending_reviews") or state.get("next_actions"):
+            raise CandidateStateError("Resolve protocol lifecycle work before candidates")
+        return trace, root
+
+    async def _load_ledger(self, root_trace_id: str) -> CandidateLedger:
+        raw = await self.store.get_candidate_ledger(root_trace_id)
+        return CandidateLedger.model_validate(raw or {})
+
+    def _allocate_pointer(
+        self,
+        operation: str,
+        trace_id: str,
+        sequence: int,
+        parents: list[CandidatePointer],
+        ledger: CandidateLedger,
+    ) -> CandidatePointer:
+        if operation == "create":
+            if parents:
+                raise CandidateStateError("create does not accept parent_refs")
+            digest = sha256(f"{trace_id}:{sequence}".encode("utf-8")).hexdigest()
+            return CandidatePointer(candidate_id=f"cand_{digest[:24]}", revision=1)
+        if operation == "fork":
+            if len(parents) != 1:
+                raise CandidateStateError("fork requires exactly one parent")
+            parent = ledger.candidate(parents[0])
+            revisions = [
+                item.revision
+                for item in ledger.candidates
+                if item.candidate_id == parent.candidate_id
+            ]
+            return CandidatePointer(
+                candidate_id=parent.candidate_id,
+                revision=max(revisions) + 1,
+            )
+        if operation == "merge":
+            if len(parents) < 2:
+                raise CandidateStateError("merge requires at least two parents")
+            digest = sha256(f"{trace_id}:{sequence}:merge".encode("utf-8")).hexdigest()
+            return CandidatePointer(candidate_id=f"cand_{digest[:24]}", revision=1)
+        raise CandidateStateError(f"Unsupported candidate operation: {operation}")
+
+    def _validate_parents(self, trace, root, parents: list[CandidateRef]) -> None:
+        for parent in parents:
+            self._validate_candidate_owner(parent, trace, root)
+
+    def _validate_candidate_owner(self, candidate, trace, root) -> None:
+        if (
+            candidate.application_ref != self.binding.application_ref
+            or candidate.root_trace_id != root.trace_id
+            or candidate.owner_trace_id != trace.trace_id
+        ):
+            raise CandidateStateError(
+                "Candidate belongs to another application, root, or Task"
+            )
+
+    @staticmethod
+    def _complete_operation(
+        ledger: CandidateLedger,
+        operation_id: str,
+        candidate: CandidateRef,
+    ) -> CandidateLedger:
+        operations = tuple(
+            item.model_copy(update={"status": "completed", "error": None})
+            if item.operation_id == operation_id
+            else item
+            for item in ledger.operations
+        )
+        lifecycle = (*ledger.lifecycle, CandidateLifecycleRecord(
+            operation_id=operation_id,
+            candidate=CandidatePointer(
+                candidate_id=candidate.candidate_id,
+                revision=candidate.revision,
+            ),
+            state="proposed",
+            effective_at_sequence=candidate.created_at_sequence,
+        ))
+        return CandidateLedger(
+            candidates=(*ledger.candidates, candidate),
+            lifecycle=lifecycle,
+            operations=operations,
+            validations=ledger.validations,
+        )
+
+    @staticmethod
+    def _fail_operation(
+        ledger: CandidateLedger,
+        operation_id: str,
+        error: str,
+    ) -> CandidateLedger:
+        operations = tuple(
+            item.model_copy(update={
+                "status": "failed",
+                "error": error[:2_000],
+            })
+            if item.operation_id == operation_id
+            else item
+            for item in ledger.operations
+        )
+        return ledger.model_copy(update={"operations": operations})

cyber_agent/application/models.py

@@ -193,6 +193,10 @@ class AgentApplication(ApplicationModel):
                     f"{sorted(missing_sets)}"
                 )
         canonical_json(self.validation_policy)
+        if self.candidate_repository_ref and not self.artifact_resolver_ref:
+            raise ValueError(
+                "candidate_repository_ref requires artifact_resolver_ref"
+            )
         return self
 
     def role(self, role_id: str) -> AgentRole:

cyber_agent/application/runtime.py

@@ -233,6 +233,21 @@ class ApplicationRuntime:
     def build_runner(self, binding: ApplicationBinding):
         from cyber_agent.core.runner import AgentRunner
 
+        candidate_service = None
+        if binding.services.candidate_repository is not None:
+            from cyber_agent.application.candidate_service import CandidateService
+
+            if binding.services.artifact_resolver is None:
+                raise ValueError(
+                    "CandidateRepository requires an ArtifactResolver"
+                )
+            candidate_service = CandidateService(
+                store=self.trace_store,
+                application_binding=binding,
+                repository=binding.services.candidate_repository,
+                artifact_resolver=binding.services.artifact_resolver,
+            )
+
         return AgentRunner(
             trace_store=self.trace_store,
             tool_registry=binding.tool_registry,
@@ -242,6 +257,7 @@ class ApplicationRuntime:
             artifact_resolver=binding.services.artifact_resolver,
             application_binding=binding,
             context_provider=binding.services.context_provider,
+            candidate_service=candidate_service,
         )
 
     def new_run(

cyber_agent/core/runner.py

@@ -242,7 +242,7 @@ class RunConfig:
     enable_research_flow: bool = True  # 是否启用自动研究流程（知识检索→经验检索→调研→计划）
     # --- 知识管理配置 ---
     knowledge: KnowledgeConfig = field(default_factory=KnowledgeConfig)
-    # --- Memory 配置（见 cyber_agent/docs/memory.md） ---
+    # --- Memory 配置（见 cyber_agent/docs/framework/runtime/memory.md） ---
     # None = 默认 Agent（无长期记忆）；赋值 MemoryConfig 使该 Agent 成为 memory-bearing Agent
     memory: Optional["MemoryConfig"] = None
 
@@ -337,6 +337,7 @@ class AgentRunner:
         artifact_resolver: Optional[ArtifactResolver] = None,
         application_binding: Any = None,
         context_provider: Any = None,
+        candidate_service: Any = None,
     ):
         """
         初始化 AgentRunner
@@ -369,6 +370,7 @@ class AgentRunner:
         self.artifact_resolver = artifact_resolver
         self.application_binding = application_binding
         self.context_provider = context_provider
+        self.candidate_service = candidate_service
         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] = {}
@@ -2812,6 +2814,7 @@ class AgentRunner:
                 "submit_task_report",
                 "review_task_result",
                 "update_task_progress",
+                "manage_candidate",
             }
             lifecycle_call_count = sum(
                 1 for tc in (tool_calls or [])
@@ -5104,6 +5107,7 @@ class AgentRunner:
             "review_task_result",
             "update_task_progress",
             "read_context_ref",
+            "manage_candidate",
         }
 
         if not policy.requires_task_protocol or in_side_branch:
@@ -5134,6 +5138,8 @@ class AgentRunner:
             return tool_names & {"agent", "read_context_ref"}
 
         tool_names.discard("review_task_result")
+        if self.candidate_service is None or state.get("task_report") is not None:
+            tool_names.discard("manage_candidate")
         if not policy.requires_task_progress or state.get("task_report") is not None:
             tool_names.discard("update_task_progress")
         if not trace.parent_trace_id or state.get("task_report") is not None:
@@ -5227,6 +5233,7 @@ class AgentRunner:
         framework_context = {
             "store": self.trace_store,
             "task_protocol_service": self.task_protocol_service,
+            "candidate_service": self.candidate_service,
             "trace_id": trace_id,
             "goal_id": goal_id,
             "runner": self,
@@ -5330,7 +5337,7 @@ class AgentRunner:
         if config.max_iterations and config.max_iterations > 0:
             system_prompt += f"\n\n## Execution Constraint\n这是一项有严格步数限制的任务。你最多可以用 {config.max_iterations} 轮交互来解决问题。\n请务必【边查边写、随时存档】！每当你收集或得出一个有价值的独立结果（如收集到一个独立 Case），请立刻调用工具写入或追加到结果文件中，绝对不要等到所有任务都做完再最后一次性输出。这样即使触达步数上限被强制打断，你已经收集的成果也能安全保留！"
         # Memory 注入（memory-bearing Agent）——在 system prompt 末尾追加
-        # 初版选择 system prompt 追加（见 cyber_agent/docs/memory.md 待定问题 1）。
+        # 初版选择 system prompt 追加（见 cyber_agent/docs/framework/runtime/memory.md 待定问题 1）。
         # 好处：run 启动一次性注入、所有后续轮次都能看到、与 skills 注入方式一致。
         # 代价：若记忆文件很大会持续占 prompt tokens —— 待观察后决定是否切换方案。
         if config.memory:

cyber_agent/core/task_contract.py

@@ -0,0 +1,37 @@
+"""Shared immutable Task contract reference used by progress and candidates."""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+
+class TaskContractRef(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+        frozen=True,
+        str_strip_whitespace=True,
+    )
+
+    kind: Literal["root_task_anchor", "task_brief"]
+    root_task_anchor_hash: str | None = Field(
+        default=None,
+        pattern=r"^[0-9a-f]{64}$",
+    )
+    task_brief_version: int | None = Field(default=None, ge=1)
+    task_brief_hash: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$")
+
+    @model_validator(mode="after")
+    def validate_kind_fields(self) -> "TaskContractRef":
+        if self.kind == "root_task_anchor":
+            if not self.root_task_anchor_hash:
+                raise ValueError("root_task_anchor requires root_task_anchor_hash")
+            if self.task_brief_version is not None or self.task_brief_hash is not None:
+                raise ValueError("root_task_anchor does not accept TaskBrief fields")
+        else:
+            if self.task_brief_version is None or not self.task_brief_hash:
+                raise ValueError("task_brief requires version and hash")
+            if self.root_task_anchor_hash is not None:
+                raise ValueError("task_brief does not accept root_task_anchor_hash")
+        return self

cyber_agent/core/task_protocol.py

@@ -15,7 +15,9 @@ from typing import Any, Literal
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 from cyber_agent.core.artifacts import ArtifactRef
+from cyber_agent.core.task_contract import TaskContractRef
 from cyber_agent.core.validation import RetryFrom
+from cyber_agent.application.candidate import CandidateRef
 
 
 TaskOutcome = Literal["satisfied", "partial", "failed", "protocol_error"]
@@ -173,6 +175,7 @@ class TaskReportSubmission(StrictProtocolModel):
     outputs: list[dict[str, Any]] = Field(default_factory=list)
     evidence: list[dict[str, Any]] = Field(default_factory=list)
     artifact_refs: list[ArtifactRef] = Field(default_factory=list, max_length=20)
+    candidate_refs: list[CandidateRef] = Field(default_factory=list, max_length=20)
     source_urls: list[str] = Field(default_factory=list, max_length=20)
     remaining_issues: list[str] = Field(default_factory=list)
 
@@ -251,32 +254,6 @@ class TaskReview(StrictProtocolModel):
         return self
 
 
-class TaskContractRef(StrictProtocolModel):
-    """TaskProgress 绑定的不可变根目标或 TaskBrief 合同。"""
-
-    kind: Literal["root_task_anchor", "task_brief"]
-    root_task_anchor_hash: str | None = Field(
-        default=None,
-        pattern=r"^[0-9a-f]{64}$",
-    )
-    task_brief_version: int | None = Field(default=None, ge=1)
-    task_brief_hash: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$")
-
-    @model_validator(mode="after")
-    def validate_kind_fields(self) -> "TaskContractRef":
-        if self.kind == "root_task_anchor":
-            if not self.root_task_anchor_hash:
-                raise ValueError("root_task_anchor requires root_task_anchor_hash")
-            if self.task_brief_version is not None or self.task_brief_hash is not None:
-                raise ValueError("root_task_anchor does not accept TaskBrief fields")
-        else:
-            if self.task_brief_version is None or not self.task_brief_hash:
-                raise ValueError("task_brief requires version and hash")
-            if self.root_task_anchor_hash is not None:
-                raise ValueError("task_brief does not accept root_task_anchor_hash")
-        return self
-
-
 class ProgressRefs(StrictProtocolModel):
     """任务推进条目可引用的现有上下文与产物句柄。"""
 

cyber_agent/tools/builtin/__init__.py

@@ -23,13 +23,14 @@ from cyber_agent.tools.builtin.task_protocol import (
 )
 # sandbox 工具已废弃（2026-04）；search.py / crawler.py 已重构为 content/ 工具族（2026-04）
 from cyber_agent.tools.builtin.knowledge import(knowledge_search,knowledge_save,knowledge_save_pending,knowledge_list,knowledge_update,knowledge_batch_update,knowledge_slim)
-# Memory / Dream（见 cyber_agent/docs/memory.md）
+# Memory / Dream（见 cyber_agent/docs/framework/runtime/memory.md）
 from cyber_agent.tools.builtin.memory import dream
 # 知识上传/查询已统一到 agent 工具：
 #   agent(agent_type="remote_librarian", task=...)         # 查询
 #   agent(agent_type="remote_librarian_ingest", task=...)  # 上传（异步）
 #   agent(agent_type="remote_research", task=...)          # 深度调研
 from cyber_agent.tools.builtin.context import get_current_context, read_context_ref
+from cyber_agent.tools.builtin.candidate import manage_candidate
 from cyber_agent.tools.builtin.toolhub import toolhub_health, toolhub_search, toolhub_call
 from cyber_agent.tools.builtin.resource import resource_list_tools, resource_get_tool
 from cyber_agent.tools.builtin.content import (
@@ -68,6 +69,7 @@ __all__ = [
     "submit_task_report",
     "review_task_result",
     "update_task_progress",
+    "manage_candidate",
     # 内容工具族（重构自 search.py + crawler.py）
     "content_platforms",
     "content_search",

cyber_agent/tools/builtin/candidate.py

@@ -0,0 +1,64 @@
+"""Lifecycle-exclusive candidate version command for application runs."""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from cyber_agent.application.candidate import CandidatePointer
+from cyber_agent.application.candidate_service import CandidateStateError
+from cyber_agent.tools import tool
+from cyber_agent.tools.models import ToolResult
+
+
+class CandidateManagementRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    operation: Literal["create", "fork", "merge"]
+    content: Any
+    parent_refs: list[CandidatePointer] = Field(default_factory=list, max_length=20)
+
+
+@tool(
+    description=(
+        "Create, fork, or merge one application candidate version. The framework "
+        "owns identity and lineage; the application repository owns content."
+    ),
+    hidden_params=["context"],
+    groups=["core"],
+)
+async def manage_candidate(
+    request: CandidateManagementRequest,
+    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"}
+    service = context.get("candidate_service")
+    trace_id = context.get("trace_id")
+    if service is None or not trace_id:
+        return {
+            "status": "failed",
+            "error": "CandidateService is unavailable for this application",
+        }
+    try:
+        candidate = await service.manage(
+            trace_id,
+            operation=request.operation,
+            content=request.content,
+            parent_refs=request.parent_refs,
+            effective_at_sequence=int(context.get("sequence", 0) or 0),
+        )
+    except (CandidateStateError, ValueError) as exc:
+        return {"status": "failed", "error": str(exc)}
+    payload = candidate.model_dump(mode="json")
+    return ToolResult(
+        text=json.dumps({
+            "status": "completed",
+            "candidate_ref": payload,
+        }, ensure_ascii=False),
+        artifact_refs=[candidate.artifact_ref.model_dump(mode="json")],
+    )

cyber_agent/trace/protocols.py

@@ -81,6 +81,21 @@ class TraceStore(Protocol):
         ResourceBudgetController 在单进程根树锁内调用，存储实现需避免留下半写入快照。"""
         ...
 
+    async def get_candidate_ledger(
+        self,
+        root_trace_id: str,
+    ) -> Optional[Dict[str, Any]]:
+        """Read the root Trace's framework-owned candidate ledger."""
+        ...
+
+    async def replace_candidate_ledger(
+        self,
+        root_trace_id: str,
+        ledger: Dict[str, Any],
+    ) -> None:
+        """Atomically replace the root Trace's candidate ledger."""
+        ...
+
     # ===== GoalTree 操作 =====
 
     async def get_goal_tree(self, trace_id: str) -> Optional[GoalTree]:

cyber_agent/trace/store.py

@@ -217,6 +217,9 @@ class FileSystemTraceStore:
         """获取根 Trace 的 resource_usage.json 文件路径。"""
         return self._get_trace_dir(root_trace_id) / "resource_usage.json"
 
+    def _get_candidate_ledger_file(self, root_trace_id: str) -> Path:
+        return self._get_trace_dir(root_trace_id) / "candidate_ledger.json"
+
     def _get_tool_approvals_file(self, trace_id: str) -> Path:
         return self._get_trace_dir(trace_id) / "tool_approvals.json"
 
@@ -391,6 +394,38 @@ class FileSystemTraceStore:
                 usage,
             )
 
+    # ===== Application candidate ledger =====
+
+    async def get_candidate_ledger(
+        self,
+        root_trace_id: str,
+    ) -> Optional[Dict[str, Any]]:
+        path = self._get_candidate_ledger_file(root_trace_id)
+        if not path.exists():
+            return None
+        try:
+            raw = json.loads(path.read_text(encoding="utf-8"))
+            if not isinstance(raw, dict):
+                raise TypeError("candidate_ledger.json must contain an object")
+            return raw
+        except Exception as exc:
+            raise TraceStoreCorruptionError(
+                f"corrupt candidate ledger: {path}"
+            ) from exc
+
+    async def replace_candidate_ledger(
+        self,
+        root_trace_id: str,
+        ledger: Dict[str, Any],
+    ) -> None:
+        async with self._lock_for(root_trace_id):
+            if await self.get_trace(root_trace_id) is None:
+                raise ValueError(f"Root Trace not found: {root_trace_id}")
+            self._atomic_write_json(
+                self._get_candidate_ledger_file(root_trace_id),
+                ledger,
+            )
+
     # ===== Tool approval state =====
 
     async def get_tool_approval_batch(self, trace_id: str):

tests/test_candidate_service.py

@@ -0,0 +1,346 @@
+import asyncio
+import tempfile
+import unittest
+
+from pydantic import ValidationError
+
+from cyber_agent.application import (
+    AgentApplication,
+    AgentRole,
+    ApplicationRegistry,
+    ApplicationServices,
+    ProviderRef,
+)
+from cyber_agent.application.candidate import (
+    CandidateLedger,
+    CandidatePointer,
+)
+from cyber_agent.application.candidate_service import (
+    CandidateService,
+    CandidateStateError,
+)
+from cyber_agent.core.artifacts import (
+    ArtifactRef,
+    ValidationMaterial,
+    material_content_hash,
+)
+from cyber_agent.core.task_protocol import TaskBrief, new_task_protocol
+from cyber_agent.tools.registry import ToolRegistry
+from cyber_agent.trace.models import Trace
+from cyber_agent.trace.store import (
+    FileSystemTraceStore,
+    TraceStoreCorruptionError,
+)
+
+
+class MemoryCandidateRepository:
+    def __init__(self):
+        self.contents = {}
+        self.put_calls = 0
+        self.merge_calls = 0
+        self.bad_artifact = False
+
+    async def put_version(self, request):
+        self.put_calls += 1
+        return self._save(request)
+
+    async def merge(self, request):
+        self.merge_calls += 1
+        return self._save(request)
+
+    def _save(self, request):
+        key = request.operation_id
+        self.contents.setdefault(key, request)
+        persisted = self.contents[key]
+        digest = material_content_hash(persisted.content)
+        if self.bad_artifact:
+            digest = "0" * 64
+        return ArtifactRef(
+            artifact_id=(
+                f"candidate:{persisted.candidate_id}:{persisted.revision}"
+            ),
+            version=str(persisted.revision),
+            content_hash=digest,
+            kind="candidate.output",
+            mime_type="application/json",
+        )
+
+    async def resolve(self, ref, root_trace_id, uid):
+        request = next(
+            item
+            for item in self.contents.values()
+            if (
+                ref.artifact_id
+                == f"candidate:{item.candidate_id}:{item.revision}"
+            )
+        )
+        return ValidationMaterial(
+            **ref.model_dump(),
+            root_trace_id=root_trace_id,
+            uid=uid,
+            content=request.content,
+        )
+
+    async def load_version(self, candidate_ref):
+        return next(
+            item.content
+            for item in self.contents.values()
+            if item.candidate_id == candidate_ref.candidate_id
+            and item.revision == candidate_ref.revision
+        )
+
+    async def commit_adoption(self, request):
+        raise NotImplementedError
+
+
+def application():
+    return AgentApplication(
+        application_id="candidate.test",
+        application_version="1",
+        root_role="writer",
+        roles=(AgentRole(
+            role_id="writer",
+            model="fake",
+            system_prompt="write",
+        ),),
+        artifact_resolver_ref=ProviderRef(
+            provider_id="artifacts",
+            provider_version="1",
+        ),
+        candidate_repository_ref=ProviderRef(
+            provider_id="candidates",
+            provider_version="1",
+        ),
+    )
+
+
+class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.store = FileSystemTraceStore(self.temp.name)
+        self.repository = MemoryCandidateRepository()
+        self.binding = ApplicationRegistry().register(
+            application(),
+            ApplicationServices(
+                tool_registry=ToolRegistry(),
+                artifact_resolver=self.repository,
+                candidate_repository=self.repository,
+            ),
+        )
+        self.service = CandidateService(
+            store=self.store,
+            application_binding=self.binding,
+            repository=self.repository,
+            artifact_resolver=self.repository,
+        )
+        await self._create_root("root-a")
+
+    async def asyncTearDown(self):
+        self.temp.cleanup()
+
+    def _context(self, root_trace_id, *, task_brief=None):
+        return {
+            "agent_mode": "recursive",
+            "agent_mode_revision": 3,
+            "root_trace_id": root_trace_id,
+            "root_task_anchor_hash": "a" * 64,
+            "application_ref": self.binding.application_ref.model_dump(mode="json"),
+            "application_role_id": "writer",
+            "application_role_hash": self.binding.role("writer").role_hash,
+            "task_protocol": new_task_protocol(task_brief),
+        }
+
+    async def _create_root(self, trace_id, *, uid="user-1"):
+        await self.store.create_trace(Trace(
+            trace_id=trace_id,
+            mode="agent",
+            agent_type="writer",
+            uid=uid,
+            context=self._context(trace_id),
+        ))
+
+    async def _create_child(self, trace_id, root_trace_id="root-a"):
+        await self.store.create_trace(Trace(
+            trace_id=trace_id,
+            mode="agent",
+            agent_type="writer",
+            uid="user-1",
+            parent_trace_id=root_trace_id,
+            context=self._context(
+                root_trace_id,
+                task_brief=TaskBrief(
+                    objective="write candidate",
+                    reason="the root needs an option",
+                    completion_criteria=["candidate is complete"],
+                    expected_outputs=["one candidate"],
+                ),
+            ),
+        ))
+
+    async def test_create_fork_merge_and_idempotent_recovery(self):
+        first, duplicate = await asyncio.gather(
+            self.service.manage(
+                "root-a",
+                operation="create",
+                content={"text": "A"},
+                parent_refs=[],
+                effective_at_sequence=1,
+            ),
+            self.service.manage(
+                "root-a",
+                operation="create",
+                content={"text": "A"},
+                parent_refs=[],
+                effective_at_sequence=1,
+            ),
+        )
+        self.assertEqual(first, duplicate)
+        self.assertEqual(1, self.repository.put_calls)
+
+        second = await self.service.manage(
+            "root-a",
+            operation="create",
+            content={"text": "B"},
+            parent_refs=[],
+            effective_at_sequence=2,
+        )
+        fork = await self.service.manage(
+            "root-a",
+            operation="fork",
+            content={"text": "A2"},
+            parent_refs=[CandidatePointer(
+                candidate_id=first.candidate_id,
+                revision=first.revision,
+            )],
+            effective_at_sequence=3,
+        )
+        self.assertEqual(first.candidate_id, fork.candidate_id)
+        self.assertEqual(2, fork.revision)
+        merged = await self.service.manage(
+            "root-a",
+            operation="merge",
+            content={"text": "AB"},
+            parent_refs=[
+                CandidatePointer(
+                    candidate_id=fork.candidate_id,
+                    revision=fork.revision,
+                ),
+                CandidatePointer(
+                    candidate_id=second.candidate_id,
+                    revision=second.revision,
+                ),
+            ],
+            effective_at_sequence=4,
+        )
+        self.assertNotIn(
+            merged.candidate_id,
+            {first.candidate_id, second.candidate_id},
+        )
+        self.assertEqual(2, len(merged.parent_refs))
+        ledger = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        self.assertEqual(4, len(ledger.candidates))
+        self.assertTrue(all(
+            ledger.current_state(item) == "proposed"
+            for item in ledger.candidates
+        ))
+
+    async def test_cross_root_and_cross_task_parent_access_is_rejected(self):
+        first = await self.service.manage(
+            "root-a",
+            operation="create",
+            content={"text": "A"},
+            parent_refs=[],
+            effective_at_sequence=1,
+        )
+        await self._create_root("root-b")
+        with self.assertRaisesRegex(CandidateStateError, "not found"):
+            await self.service.manage(
+                "root-b",
+                operation="fork",
+                content={"text": "stolen"},
+                parent_refs=[CandidatePointer(
+                    candidate_id=first.candidate_id,
+                    revision=first.revision,
+                )],
+                effective_at_sequence=2,
+            )
+
+        await self._create_child("child-a")
+        with self.assertRaisesRegex(CandidateStateError, "another application, root, or Task"):
+            await self.service.manage(
+                "child-a",
+                operation="fork",
+                content={"text": "stolen"},
+                parent_refs=[CandidatePointer(
+                    candidate_id=first.candidate_id,
+                    revision=first.revision,
+                )],
+                effective_at_sequence=3,
+            )
+
+    async def test_bad_artifact_fails_without_registering_candidate(self):
+        self.repository.bad_artifact = True
+        with self.assertRaisesRegex(CandidateStateError, "invalid ArtifactRef"):
+            await self.service.manage(
+                "root-a",
+                operation="create",
+                content={"text": "bad"},
+                parent_refs=[],
+                effective_at_sequence=1,
+            )
+        ledger = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        self.assertEqual(0, len(ledger.candidates))
+        self.assertEqual("failed", ledger.operations[0].status)
+
+    async def test_report_validation_rejects_tamper_and_wrong_owner(self):
+        candidate = await self.service.manage(
+            "root-a",
+            operation="create",
+            content={"text": "A"},
+            parent_refs=[],
+            effective_at_sequence=1,
+        )
+        await self.service.validate_report_refs("root-a", [candidate])
+        tampered = candidate.model_copy(update={
+            "artifact_ref": candidate.artifact_ref.model_copy(update={
+                "version": "tampered",
+            }),
+        })
+        with self.assertRaisesRegex(CandidateStateError, "does not match"):
+            await self.service.validate_report_refs("root-a", [tampered])
+        await self._create_child("child-a")
+        with self.assertRaisesRegex(CandidateStateError, "another application, root, or Task"):
+            await self.service.validate_report_refs("child-a", [candidate])
+
+    async def test_ledger_rejects_future_lineage_and_store_reports_corruption(self):
+        candidate = await self.service.manage(
+            "root-a",
+            operation="create",
+            content={"text": "A"},
+            parent_refs=[],
+            effective_at_sequence=1,
+        )
+        raw = (await self.store.get_candidate_ledger("root-a"))
+        future = candidate.model_copy(update={
+            "revision": 2,
+            "parent_refs": (CandidatePointer(
+                candidate_id=candidate.candidate_id,
+                revision=3,
+            ),),
+        })
+        raw["candidates"].append(future.model_dump(mode="json"))
+        with self.assertRaisesRegex(ValidationError, "unknown parent"):
+            CandidateLedger.model_validate(raw)
+
+        ledger_path = self.store._get_candidate_ledger_file("root-a")
+        ledger_path.write_text("{broken", encoding="utf-8")
+        with self.assertRaisesRegex(TraceStoreCorruptionError, "candidate ledger"):
+            await self.store.get_candidate_ledger("root-a")
+
+
+if __name__ == "__main__":
+    unittest.main()