test(m3b): 完成双候选局部回退参考应用和故障链路

新增独立 application_reference 业务目录，提供可替换的角色 Prompt、事实工具、ContextProvider、ArtifactResolver、CandidateRepository、占位符质量规则和 RunEventProjector；示例不实现第二套 Runner，也不修改框架协议。

端到端场景验证事实工具只调用一次，A 验收通过后保持不变，B 因占位文本仅从 output 层回退并 fork 为 B2；进程恢复后 B2 通过并被精确采用一次，事件重复对账无重复投影，采用后的越界 rewind 稳定拒绝。

补充外部采用成功但框架 finalize 前崩溃的故障注入，确认重启使用同一 operation ID 重放且业务写回恰好一次。

examples/application_reference/README.md

@@ -0,0 +1,5 @@
+# Application Reference
+
+这是 M2/M3 的可运行最小创作应用，不是新的 Runner。业务目录只提供角色、事实工具、Artifact/Candidate 存储、确定性质量规则和事件投影；递归协议、恢复、验证、回溯与执行门禁全部复用 `cyber_agent`。
+
+验收入口是仓库测试 `tests/test_application_reference.py`。场景固定产生 A/B 两个候选，B 因占位文本失败后只从 output 层 fork 为 B2；事实工具、A 的 Artifact 和 A 的 ValidationResult 不重跑。B2 通过后被精确采用一次，事件可幂等重放，采用后的越界回溯被框架拒绝。

examples/application_reference/__init__.py

@@ -0,0 +1,5 @@
+"""Runnable evidence-driven creative application reference."""
+
+from .application import ReferenceComponents, build_reference_components
+
+__all__ = ["ReferenceComponents", "build_reference_components"]

examples/application_reference/application.py

@@ -0,0 +1,168 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+from cyber_agent.application import (
+    AgentApplication,
+    AgentRole,
+    ApplicationServices,
+    ProviderRef,
+    QualityRuleSpec,
+    ToolSet,
+    ToolSpec,
+)
+from cyber_agent.tools import get_tool_registry
+
+from .providers import (
+    PlaceholderQualityProvider,
+    ReferenceArtifactStore,
+    ReferenceCandidateRepository,
+    ReferenceContextProvider,
+    ReferenceProjector,
+)
+from .tools import FactVerifier
+
+
+_ROLES = Path(__file__).with_name("roles")
+
+
+def _prompt(role_id: str) -> str:
+    return (_ROLES / f"{role_id}.prompt").read_text(encoding="utf-8").strip()
+
+
+@dataclass(frozen=True)
+class ReferenceComponents:
+    application: AgentApplication
+    services: ApplicationServices
+    artifacts: ReferenceArtifactStore
+    candidates: ReferenceCandidateRepository
+    context: ReferenceContextProvider
+    facts: FactVerifier
+    projector: ReferenceProjector
+
+
+def build_reference_components() -> ReferenceComponents:
+    artifacts = ReferenceArtifactStore()
+    candidates = ReferenceCandidateRepository(artifacts)
+    context = ReferenceContextProvider()
+    facts = FactVerifier(artifacts)
+    projector = ReferenceProjector()
+    registry = get_tool_registry().clone()
+
+    async def verify_fact(question: str) -> dict:
+        """Runtime tests use FactVerifier directly so hidden identity stays framework-owned."""
+        return {"question": question}
+
+    registry.register(verify_fact)
+    application = AgentApplication(
+        application_id="application_reference",
+        application_version="1",
+        root_role="editor",
+        roles=(
+            AgentRole(
+                role_id="editor",
+                model="reference-model",
+                system_prompt=_prompt("editor"),
+                tool_sets=("editor-tools",),
+                allowed_child_roles=("fact_checker", "writer"),
+            ),
+            AgentRole(
+                role_id="fact_checker",
+                model="reference-model",
+                system_prompt=_prompt("fact_checker"),
+                tool_sets=("fact-tools",),
+            ),
+            AgentRole(
+                role_id="writer",
+                model="reference-model",
+                system_prompt=_prompt("writer"),
+                tool_sets=("writer-tools",),
+            ),
+        ),
+        tool_sets=(
+            ToolSet(
+                tool_set_id="editor-tools",
+                tools=tuple(ToolSpec(
+                    tool_name=name,
+                    implementation_version="framework-1",
+                ) for name in (
+                    "agent",
+                    "update_task_progress",
+                    "review_task_result",
+                )),
+            ),
+            ToolSet(
+                tool_set_id="fact-tools",
+                tools=(
+                    ToolSpec(
+                        tool_name="verify_fact",
+                        implementation_version="1",
+                        side_effect_class="read",
+                    ),
+                    ToolSpec(
+                        tool_name="update_task_progress",
+                        implementation_version="framework-1",
+                    ),
+                    ToolSpec(
+                        tool_name="submit_task_report",
+                        implementation_version="framework-1",
+                    ),
+                ),
+            ),
+            ToolSet(
+                tool_set_id="writer-tools",
+                tools=tuple(ToolSpec(
+                    tool_name=name,
+                    implementation_version="framework-1",
+                    side_effect_class=("write" if name == "manage_candidate" else "none"),
+                ) for name in (
+                    "manage_candidate",
+                    "update_task_progress",
+                    "submit_task_report",
+                )),
+            ),
+        ),
+        artifact_resolver_ref=ProviderRef(
+            provider_id="reference-artifacts",
+            provider_version="1",
+        ),
+        context_provider_ref=ProviderRef(
+            provider_id="reference-context",
+            provider_version="1",
+        ),
+        candidate_repository_ref=ProviderRef(
+            provider_id="reference-candidates",
+            provider_version="1",
+        ),
+        quality_provider_ref=ProviderRef(
+            provider_id="reference-quality",
+            provider_version="1",
+        ),
+        event_projector_ref=ProviderRef(
+            provider_id="reference-projector",
+            provider_version="1",
+        ),
+        quality_rules=(QualityRuleSpec(
+            rule_id="no_placeholder",
+            version="1",
+            criterion="Candidate output contains no placeholder description.",
+        ),),
+    )
+    services = ApplicationServices(
+        tool_registry=registry,
+        context_provider=context,
+        artifact_resolver=artifacts,
+        candidate_repository=candidates,
+        quality_provider=PlaceholderQualityProvider(),
+        event_projector=projector,
+    )
+    return ReferenceComponents(
+        application=application,
+        services=services,
+        artifacts=artifacts,
+        candidates=candidates,
+        context=context,
+        facts=facts,
+        projector=projector,
+    )

examples/application_reference/providers/__init__.py

@@ -0,0 +1,13 @@
+from .artifacts import ReferenceArtifactStore
+from .candidates import ReferenceCandidateRepository
+from .context import ReferenceContextProvider
+from .projector import ReferenceProjector
+from .quality import PlaceholderQualityProvider
+
+__all__ = [
+    "PlaceholderQualityProvider",
+    "ReferenceArtifactStore",
+    "ReferenceCandidateRepository",
+    "ReferenceContextProvider",
+    "ReferenceProjector",
+]

examples/application_reference/providers/artifacts.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from cyber_agent.core.artifacts import (
+    ArtifactRef,
+    ValidationMaterial,
+    material_content_hash,
+)
+
+
+@dataclass(frozen=True)
+class _StoredArtifact:
+    ref: ArtifactRef
+    root_trace_id: str
+    uid: str | None
+    content: Any
+
+
+class ReferenceArtifactStore:
+    """Example-owned content storage exposed through the framework resolver port."""
+
+    def __init__(self) -> None:
+        self._items: dict[tuple[str, str], _StoredArtifact] = {}
+
+    def put(
+        self,
+        *,
+        artifact_id: str,
+        version: str,
+        kind: str,
+        root_trace_id: str,
+        uid: str | None,
+        content: Any,
+    ) -> ArtifactRef:
+        ref = ArtifactRef(
+            artifact_id=artifact_id,
+            version=version,
+            content_hash=material_content_hash(content),
+            kind=kind,
+            mime_type="application/json",
+        )
+        self._items[(artifact_id, version)] = _StoredArtifact(
+            ref=ref,
+            root_trace_id=root_trace_id,
+            uid=uid,
+            content=content,
+        )
+        return ref
+
+    async def resolve(self, ref, root_trace_id, uid):
+        item = self._items[(ref.artifact_id, ref.version)]
+        if item.ref != ref:
+            raise ValueError("ArtifactRef content hash does not match storage")
+        if item.root_trace_id != root_trace_id or item.uid != uid:
+            raise ValueError("Artifact belongs to another root or owner")
+        return ValidationMaterial(
+            **ref.model_dump(mode="json"),
+            root_trace_id=root_trace_id,
+            uid=uid,
+            content=item.content,
+        )

examples/application_reference/providers/candidates.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from cyber_agent.application.candidate import CandidateAdoptionReceipt
+
+
+class ReferenceCandidateRepository:
+    def __init__(self, artifacts) -> None:
+        self.artifacts = artifacts
+        self.versions = {}
+        self.adoptions = {}
+        self.adoption_calls = 0
+
+    async def put_version(self, request):
+        self.versions.setdefault(request.operation_id, request)
+        saved = self.versions[request.operation_id]
+        return self.artifacts.put(
+            artifact_id=f"candidate:{saved.candidate_id}:{saved.revision}",
+            version=str(saved.revision),
+            kind="candidate.output",
+            root_trace_id=saved.root_trace_id,
+            uid=saved.uid,
+            content=saved.content,
+        )
+
+    async def merge(self, request):
+        return await self.put_version(request)
+
+    async def load_version(self, candidate_ref):
+        return next(
+            item.content for item in self.versions.values()
+            if item.candidate_id == candidate_ref.candidate_id
+            and item.revision == candidate_ref.revision
+        )
+
+    async def commit_adoption(self, request):
+        self.adoption_calls += 1
+        self.adoptions.setdefault(request.operation_id, request.candidate_ref)
+        return CandidateAdoptionReceipt(
+            operation_id=request.operation_id,
+            external_id=f"published:{request.candidate_ref.candidate_id}",
+        )

examples/application_reference/providers/context.py

@@ -0,0 +1,30 @@
+from cyber_agent.application import ContextMaterial, ContextResourceDescriptor
+from cyber_agent.application.models import stable_hash
+
+
+class ReferenceContextProvider:
+    def __init__(self) -> None:
+        self.content = {
+            "source": "reference-fixture",
+            "topic": "agent framework responsibilities",
+        }
+
+    async def list_context(self, request):
+        return [ContextResourceDescriptor(
+            application_ref=request.application_ref,
+            root_trace_id=request.root_trace_id,
+            trace_id=request.trace_id,
+            uid=request.uid,
+            role_id=request.role_id,
+            source_id="reference-brief",
+            source_version="1",
+            content_hash=stable_hash(self.content),
+            kind="creative_brief",
+            summary="Reference creative brief",
+            inheritance="inheritable",
+            estimated_chars=len(str(self.content)),
+        )]
+
+    async def resolve_context(self, request, descriptor):
+        del request
+        return ContextMaterial(descriptor=descriptor, content=self.content)

examples/application_reference/providers/projector.py

@@ -0,0 +1,16 @@
+class ReferenceProjector:
+    """Tiny transactional projection used to prove cursor/idempotency semantics."""
+
+    def __init__(self) -> None:
+        self.cursor = 0
+        self.rows = {}
+
+    async def load_cursor(self, application_ref, root_trace_id):
+        del application_ref, root_trace_id
+        return self.cursor
+
+    async def project(self, event, expected_cursor):
+        if expected_cursor != self.cursor:
+            raise ValueError("stale projection cursor")
+        self.rows.setdefault(event.event_key, event.payload)
+        self.cursor = event.event_id

examples/application_reference/providers/quality.py

@@ -0,0 +1,16 @@
+from cyber_agent.application.quality import QualityCheckOutcome
+
+
+class PlaceholderQualityProvider:
+    async def check(self, request):
+        text = str(request.material.content.get("text", ""))
+        failed = "{{placeholder}}" in text
+        return [
+            QualityCheckOutcome(
+                rule_id=rule.rule_id,
+                rule_version=rule.version,
+                status="failed" if failed else "passed",
+                issue="candidate still contains placeholder text" if failed else None,
+            )
+            for rule in request.rules
+        ]

examples/application_reference/roles/editor.prompt

@@ -0,0 +1 @@
+Coordinate evidence-driven content. Treat ValidationResult as the acceptance authority and review exact candidate revisions.

examples/application_reference/roles/fact_checker.prompt

@@ -0,0 +1 @@
+Verify one bounded fact, preserve its source as an ArtifactRef, and report unresolved uncertainty explicitly.

examples/application_reference/roles/writer.prompt

@@ -0,0 +1 @@
+Create replaceable candidate versions from authorized evidence. Never adopt content and never hide placeholder text.

examples/application_reference/tools/__init__.py

@@ -0,0 +1,3 @@
+from .verify_fact import FactVerifier
+
+__all__ = ["FactVerifier"]

examples/application_reference/tools/verify_fact.py

@@ -0,0 +1,25 @@
+class FactVerifier:
+    def __init__(self, artifacts) -> None:
+        self.artifacts = artifacts
+        self.calls = 0
+
+    async def verify(
+        self,
+        *,
+        question: str,
+        root_trace_id: str,
+        uid: str | None,
+    ):
+        self.calls += 1
+        return self.artifacts.put(
+            artifact_id="fact:framework-definition",
+            version="1",
+            kind="evidence.fact",
+            root_trace_id=root_trace_id,
+            uid=uid,
+            content={
+                "question": question,
+                "answer": "An agent framework supplies reusable execution contracts.",
+                "source": "reference-fixture",
+            },
+        )

tests/test_application_reference.py

@@ -0,0 +1,312 @@
+from copy import deepcopy
+import json
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from cyber_agent.application import (
+    ApplicationRegistry,
+    ApplicationRuntime,
+    CandidateReviewAction,
+)
+from cyber_agent.application.candidate import CandidateLedger, CandidatePointer
+from cyber_agent.core.task_protocol import (
+    Finding,
+    Hypothesis,
+    Question,
+    TaskBrief,
+    TaskProgress,
+    WorkItem,
+    initialize_task_progress,
+    new_task_protocol,
+)
+from cyber_agent.trace.models import Message, Trace
+from cyber_agent.trace.store import FileSystemTraceStore
+from examples.application_reference import build_reference_components
+
+
+class ApplicationReferenceEndToEndTest(unittest.IsolatedAsyncioTestCase):
+    async def test_local_candidate_retry_preserves_fact_and_peer_validation(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            store = FileSystemTraceStore(temp_dir)
+            components = build_reference_components()
+            registry = ApplicationRegistry()
+            binding = registry.register(
+                components.application,
+                components.services,
+            )
+            llm_calls = 0
+
+            async def llm_call(**kwargs):
+                nonlocal llm_calls
+                llm_calls += 1
+                packet = json.loads(kwargs["messages"][-1]["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"]
+                        ],
+                        "reason": "reference candidate is complete",
+                        "retry_from": None,
+                    }),
+                    "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 an evidence-grounded explanation",
+                    "completion_criteria": ["Use verified facts and finished copy"],
+                    "constraints": ["Do not publish placeholder text"],
+                },
+            )
+            with patch.dict("os.environ", {"AGENT_MODE": "recursive"}, clear=False):
+                root, _goal, _sequence = await runner._prepare_new_trace(
+                    [{"role": "user", "content": "write the explanation"}],
+                    config,
+                )
+            await store.add_message(Message.create(
+                trace_id=root.trace_id,
+                role="user",
+                sequence=1,
+                content="write the explanation",
+            ))
+            await store.add_message(Message.create(
+                trace_id=root.trace_id,
+                role="assistant",
+                sequence=2,
+                parent_sequence=1,
+                content="drafting",
+            ))
+            await store.update_trace(root.trace_id, head_sequence=2)
+
+            async def create_child(trace_id, role_id, brief):
+                context = deepcopy(root.context)
+                context.pop("run_config_snapshot", None)
+                context["application_role_id"] = role_id
+                context["application_role_hash"] = binding.role(role_id).role_hash
+                context["agent_depth"] = 1
+                state = new_task_protocol(brief)
+                initialize_task_progress(state, effective_at_sequence=1)
+                context["task_protocol"] = state
+                child = Trace(
+                    trace_id=trace_id,
+                    mode="agent",
+                    agent_type=role_id,
+                    uid=root.uid,
+                    model=binding.role(role_id).role.model,
+                    parent_trace_id=root.trace_id,
+                    context=context,
+                )
+                await store.create_trace(child)
+                return child
+
+            fact_child = await create_child(
+                "reference-fact",
+                "fact_checker",
+                TaskBrief(
+                    objective="Verify what an agent framework supplies",
+                    reason="The content needs one traceable fact",
+                    completion_criteria=["Return one sourced finding"],
+                    expected_outputs=["fact artifact"],
+                    validation_scopes=["evidence"],
+                ),
+            )
+            fact_ref = await components.facts.verify(
+                question="What does an agent framework supply?",
+                root_trace_id=root.trace_id,
+                uid=root.uid,
+            )
+            await runner.task_protocol_service.update_progress(
+                fact_child.trace_id,
+                expected_revision=1,
+                progress=TaskProgress(
+                    phase="ready_to_submit",
+                    questions=[Question(
+                        item_id="fact-question",
+                        text="What does the framework supply?",
+                        state="answered",
+                        answer="Reusable execution contracts",
+                        artifact_refs=[fact_ref],
+                    )],
+                    findings=[Finding(
+                        item_id="fact-finding",
+                        statement="The framework supplies reusable execution contracts.",
+                        basis="reference fixture",
+                        artifact_refs=[fact_ref],
+                    )],
+                    hypotheses=[Hypothesis(
+                        item_id="fact-hypothesis",
+                        statement="The explanation can distinguish framework from business code.",
+                        state="supported",
+                        rationale="The verified definition supports the distinction.",
+                        artifact_refs=[fact_ref],
+                    )],
+                    work_items=[WorkItem(
+                        item_id="fact-work",
+                        description="verify the definition",
+                        state="done",
+                        result_summary="one fact artifact recorded",
+                        artifact_refs=[fact_ref],
+                    )],
+                    decision_rationale="The question is answered by one stable fixture.",
+                ),
+                effective_at_sequence=2,
+            )
+
+            writer = await create_child(
+                "reference-writer",
+                "writer",
+                TaskBrief(
+                    objective="Produce two candidate explanations",
+                    reason="The editor needs alternatives",
+                    completion_criteria=["Both candidates use the verified fact"],
+                    expected_outputs=["candidate A", "candidate B"],
+                    validation_scopes=["output"],
+                ),
+            )
+            candidate_a = await runner.candidate_service.manage(
+                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(
+                writer.trace_id,
+                operation="create",
+                content={"text": "B says {{placeholder}}."},
+                parent_refs=[],
+                effective_at_sequence=5,
+            )
+            await runner.task_protocol_service.update_progress(
+                writer.trace_id,
+                expected_revision=1,
+                progress=TaskProgress(
+                    phase="ready_to_submit",
+                    findings=[Finding(
+                        item_id="writer-finding",
+                        statement="Both candidates derive from the verified definition.",
+                        basis="fact artifact",
+                        artifact_refs=[fact_ref],
+                    )],
+                    work_items=[WorkItem(
+                        item_id="writer-work",
+                        description="draft two alternatives",
+                        state="done",
+                        result_summary="A and B registered",
+                    )],
+                    decision_rationale="Both candidate revisions are ready for validation.",
+                ),
+                effective_at_sequence=6,
+            )
+
+            validation_a = await runner.validate_recursive_trace(
+                writer.trace_id,
+                candidate_ref=candidate_a,
+            )
+            validation_b = await runner.validate_recursive_trace(
+                writer.trace_id,
+                candidate_ref=candidate_b,
+            )
+            self.assertEqual("passed", validation_a.result.outcome)
+            self.assertEqual("failed", validation_b.result.outcome)
+            self.assertEqual(1, components.facts.calls)
+            self.assertEqual(1, llm_calls)
+            frozen_a = validation_a.result.model_dump(mode="json")
+
+            restored_runner, _restored_config = await runtime.restore(root.trace_id)
+            await restored_runner.candidate_service.apply_review_actions(
+                root.trace_id,
+                writer.trace_id,
+                report_refs=[candidate_a, candidate_b],
+                actions=[CandidateReviewAction(
+                    action="revise",
+                    candidate_ref=candidate_b,
+                    reason="remove the placeholder without repeating fact research",
+                )],
+                effective_at_sequence=7,
+            )
+            candidate_b2 = await restored_runner.candidate_service.manage(
+                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,
+            )
+            validation_b2 = await restored_runner.validate_recursive_trace(
+                writer.trace_id,
+                candidate_ref=candidate_b2,
+            )
+            self.assertEqual("passed", validation_b2.result.outcome)
+            self.assertEqual(2, llm_calls)
+            self.assertEqual(1, components.facts.calls)
+            self.assertEqual(
+                frozen_a,
+                (await restored_runner.validate_recursive_trace(
+                    writer.trace_id,
+                    candidate_ref=candidate_a,
+                )).result.model_dump(mode="json"),
+            )
+            self.assertEqual(2, llm_calls)
+
+            messages = await store.get_trace_messages(root.trace_id)
+            cutoff = min(item.sequence for item in messages)
+            await restored_runner._rewind(root.trace_id, cutoff, None)
+            await restored_runner.candidate_service.apply_review_actions(
+                root.trace_id,
+                writer.trace_id,
+                report_refs=[candidate_a, candidate_b2],
+                actions=[CandidateReviewAction(
+                    action="adopt",
+                    candidate_ref=candidate_b2,
+                    reason="publish the corrected exact revision",
+                )],
+                effective_at_sequence=20,
+            )
+            self.assertEqual(1, components.candidates.adoption_calls)
+            self.assertEqual(1, len(components.candidates.adoptions))
+
+            before_replay = dict(components.projector.rows)
+            await runtime.reconcile_events(root.trace_id)
+            await runtime.reconcile_events(root.trace_id)
+            self.assertEqual(before_replay, components.projector.rows)
+            adopted_keys = [
+                key for key in components.projector.rows
+                if key.endswith(":adopted")
+            ]
+            self.assertEqual(1, len(adopted_keys))
+            with self.assertRaisesRegex(ValueError, "committed candidate adoption"):
+                await restored_runner._rewind(root.trace_id, cutoff, None)
+
+            ledger = CandidateLedger.model_validate(
+                await store.get_candidate_ledger(root.trace_id)
+            )
+            self.assertEqual(3, len(ledger.candidates))
+            self.assertEqual(3, len(ledger.validations))
+            self.assertEqual(
+                [(candidate_b.candidate_id, candidate_b.revision)],
+                [
+                    (item.candidate_id, item.revision)
+                    for item in candidate_b2.parent_refs
+                ],
+            )

tests/test_candidate_service.py

@@ -1,6 +1,7 @@
 import asyncio
 import tempfile
 import unittest
+from unittest.mock import patch
 
 from pydantic import ValidationError
 
@@ -477,6 +478,62 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual("discarded", records[0].state)
         self.assertIn("retry_from=output", records[0].reason)
 
+    async def test_adoption_replays_same_operation_after_finalize_crash(self):
+        await self._create_child("child-a")
+        candidate = await self.service.manage(
+            "child-a",
+            operation="create",
+            content={"text": "A"},
+            parent_refs=[],
+            effective_at_sequence=4,
+        )
+        await self._record_validation("child-a", candidate)
+        action = CandidateReviewAction(
+            action="adopt",
+            candidate_ref=candidate,
+            reason="publish once",
+        )
+        original_replace = self.store.replace_candidate_ledger
+        calls = 0
+
+        async def fail_finalize(root_trace_id, ledger):
+            nonlocal calls
+            calls += 1
+            if calls == 2:
+                raise OSError("crash before framework finalize")
+            await original_replace(root_trace_id, ledger)
+
+        with patch.object(
+            self.store,
+            "replace_candidate_ledger",
+            side_effect=fail_finalize,
+        ):
+            with self.assertRaisesRegex(OSError, "finalize"):
+                await self.service.apply_review_actions(
+                    "root-a",
+                    "child-a",
+                    report_refs=[candidate],
+                    actions=[action],
+                    effective_at_sequence=9,
+                )
+        pending = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        self.assertEqual("adoption_pending", pending.current_state(candidate))
+        await self.service.apply_review_actions(
+            "root-a",
+            "child-a",
+            report_refs=[candidate],
+            actions=[action],
+            effective_at_sequence=9,
+        )
+        self.assertEqual(2, self.repository.adoption_calls)
+        self.assertEqual(1, len(self.repository.adoptions))
+        completed = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        self.assertEqual("adopted", completed.current_state(candidate))
+
 
 if __name__ == "__main__":
     unittest.main()