DocForge2/tests/test_changesets.py

from __future__ import annotations

import hashlib
import json
import multiprocessing
import os
import shutil
import stat
import tempfile
import unittest
from pathlib import Path
from typing import Any
from unittest import mock

import docforge.application as application_module
from docforge.application import CanonicalApplicationService, GenericCanonicalApplier
from docforge.changesets import ChangesetStore
from docforge.errors import DocForgeError
from docforge.project import Project

ROOT = Path(__file__).resolve().parents[1]
FIXTURES = ROOT / "tests" / "fixtures"


def _append_in_process(
    root: str,
    node_id: str,
    expected_changeset_hash: str,
    expected_content_hash: str,
    start: Any,
    result: Any,
) -> None:
    """Attempt one append in an independent process and report its stable outcome."""

    start.wait()
    try:
        ChangesetStore(Project.open(root), "alpha-editor").propose_update(
            changeset_id="simultaneous",
            expected_changeset_hash=expected_changeset_hash,
            node_id=node_id,
            expected_content_hash=expected_content_hash,
            metadata={"summary": f"Concurrent proposal for {node_id}."},
            content=None,
            relationship_changes=[],
            rationale="Prove process-safe optimistic serialization.",
        )
    except DocForgeError as error:
        result.send(error.code)
    else:
        result.send("ok")
    finally:
        result.close()


class DocForgeChangesetTests(unittest.TestCase):
    def copy_fixture(self, destination: Path) -> Path:
        root = destination / "alpha"
        shutil.copytree(FIXTURES / "alpha", root)
        return root

    def canonical_digest(self, root: Path) -> str:
        paths = [
            root / ".docforge" / "project.toml",
            root / "POLICY.md",
            *(root / "docs" / "content").glob("*"),
        ]
        digest = hashlib.sha256()
        for path in sorted((path for path in paths if path.is_file()), key=lambda item: str(item)):
            digest.update(path.relative_to(root).as_posix().encode())
            digest.update(path.read_bytes())
        return digest.hexdigest()

    @staticmethod
    def node_hash(project: Project, node_id: str) -> str:
        return next(node.content_hash for node in project.load().nodes if node.node_id == node_id)

    @staticmethod
    def new_metadata(*, family: str = "guide") -> dict[str, object]:
        return {
            "title": "New proposal node",
            "family": family,
            "authority": "proposal",
            "status": "active",
            "tags": ["proposal", "test"],
            "summary": "Exercises an isolated structured creation proposal.",
        }

    def test_all_four_operations_validate_and_never_change_canonical_sources(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            parent = Path(directory)

            create_root = self.copy_fixture(parent / "create")
            create_before = self.canonical_digest(create_root)
            create_store = ChangesetStore(Project.open(create_root), "alpha-editor")
            created = create_store.create("create-node")
            proposed = create_store.propose_create(
                changeset_id="create-node",
                expected_changeset_hash=created["changeset_hash"],
                node_id="guide.new-node",
                target_source="docs/content/new-node.md",
                metadata=self.new_metadata(),
                content="This content exists only inside the proposal.",
                relationship_changes=[
                    {
                        "action": "add",
                        "source_id": "guide.new-node",
                        "relation": "depends_on",
                        "target_id": "guide.foundation",
                    }
                ],
                rationale="Add one independently reviewable guide node.",
            )
            self.assertEqual(1, proposed["operation_count"])
            self.assertEqual(create_before, self.canonical_digest(create_root))
            self.assertFalse((create_root / "docs/content/new-node.md").exists())
            create_diff = create_store.diff("create-node")
            self.assertEqual("create", create_diff["changes"][0]["operation"])
            self.assertIn("proposal", create_diff["changes"][0]["content_diff"])
            self.assertEqual(create_diff, create_store.diff("create-node"))

            update_root = self.copy_fixture(parent / "update")
            update_before = self.canonical_digest(update_root)
            update_project = Project.open(update_root)
            update_store = ChangesetStore(update_project, "alpha-editor")
            update_changeset = update_store.create("update-node")
            updated = update_store.propose_update(
                changeset_id="update-node",
                expected_changeset_hash=update_changeset["changeset_hash"],
                node_id="guide.workflow",
                expected_content_hash=self.node_hash(update_project, "guide.workflow"),
                metadata={"summary": "A proposed replacement workflow summary."},
                content="Proposed workflow content that is not canonical yet.",
                relationship_changes=[],
                rationale="Replace the workflow explanation without applying it.",
            )
            self.assertEqual(3, updated["projected_node_count"])
            update_diff = update_store.diff("update-node")["changes"][0]
            self.assertEqual("update", update_diff["operation"])
            self.assertIn("summary", update_diff["metadata"])
            self.assertEqual(update_before, self.canonical_digest(update_root))

            move_root = self.copy_fixture(parent / "move")
            move_before = self.canonical_digest(move_root)
            move_project = Project.open(move_root)
            move_store = ChangesetStore(move_project, "alpha-editor")
            move_changeset = move_store.create("move-node")
            move_store.propose_move(
                changeset_id="move-node",
                expected_changeset_hash=move_changeset["changeset_hash"],
                node_id="guide.foundation",
                expected_content_hash=self.node_hash(move_project, "guide.foundation"),
                target_source="docs/content/foundation-moved.md",
                rationale="Move the node while preserving its stable identity.",
            )
            move_diff = move_store.diff("move-node")["changes"][0]
            self.assertEqual(
                {
                    "before": "docs/content/foundation.md",
                    "after": "docs/content/foundation-moved.md",
                },
                move_diff["source"],
            )
            self.assertEqual(move_before, self.canonical_digest(move_root))

            delete_root = self.copy_fixture(parent / "delete")
            delete_before = self.canonical_digest(delete_root)
            delete_project = Project.open(delete_root)
            delete_store = ChangesetStore(delete_project, "alpha-editor")
            delete_changeset = delete_store.create("delete-node")
            deleted = delete_store.propose_delete(
                changeset_id="delete-node",
                expected_changeset_hash=delete_changeset["changeset_hash"],
                node_id="proof.validation",
                expected_content_hash=self.node_hash(delete_project, "proof.validation"),
                relationship_changes=[
                    {
                        "action": "remove",
                        "source_id": "proof.validation",
                        "relation": "proves",
                        "target_id": "guide.workflow",
                    }
                ],
                rationale="Remove the proof node and its owned relationship in one proposal.",
            )
            self.assertEqual(2, deleted["projected_node_count"])
            self.assertEqual("delete", delete_store.diff("delete-node")["changes"][0]["operation"])
            self.assertEqual(delete_before, self.canonical_digest(delete_root))

    def test_hash_bound_application_writes_projection_and_refreshes_derived_state(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            project = Project.open(root)
            store = ChangesetStore(project, "alpha-editor")
            hashes = {node.node_id: node.content_hash for node in project.load().nodes}
            created = store.create("apply-all")
            updated = store.propose_update(
                changeset_id="apply-all",
                expected_changeset_hash=created["changeset_hash"],
                node_id="guide.workflow",
                expected_content_hash=hashes["guide.workflow"],
                metadata={"summary": "Applied through the canonical application service."},
                content="The applied workflow is now canonical.",
                relationship_changes=[],
                rationale="Exercise a canonical update.",
            )
            added = store.propose_create(
                changeset_id="apply-all",
                expected_changeset_hash=updated["changeset_hash"],
                node_id="guide.applied",
                target_source="docs/content/applied.md",
                metadata=self.new_metadata(),
                content="This node was created by an approved changeset.",
                relationship_changes=[],
                rationale="Exercise a canonical creation.",
            )
            moved = store.propose_move(
                changeset_id="apply-all",
                expected_changeset_hash=added["changeset_hash"],
                node_id="guide.foundation",
                expected_content_hash=hashes["guide.foundation"],
                target_source="docs/content/foundation-moved.md",
                rationale="Exercise a canonical move.",
            )
            final = store.propose_delete(
                changeset_id="apply-all",
                expected_changeset_hash=moved["changeset_hash"],
                node_id="proof.validation",
                expected_content_hash=hashes["proof.validation"],
                relationship_changes=[
                    {
                        "action": "remove",
                        "source_id": "proof.validation",
                        "relation": "proves",
                        "target_id": "guide.workflow",
                    }
                ],
                rationale="Exercise a canonical deletion.",
            )
            service = CanonicalApplicationService(
                project,
                applier_id="alpha-editor",
                applier=GenericCanonicalApplier(project),
            )
            result = service.apply("apply-all", str(final["changeset_hash"]))

            snapshot = project.load()
            node_ids = {node.node_id for node in snapshot.nodes}
            workflow = next(node for node in snapshot.nodes if node.node_id == "guide.workflow")
            self.assertTrue(result["applied"])
            self.assertEqual("applied", result["lifecycle"]["status"])
            self.assertEqual("ok", result["derived_refresh"]["status"])
            self.assertEqual(
                [
                    "docs/content/applied.md",
                    "docs/content/foundation-moved.md",
                    "docs/content/foundation.md",
                    "docs/content/proof.toml",
                    "docs/content/workflow.md",
                ],
                result["applied_sources"],
            )
            self.assertEqual({"guide.applied", "guide.foundation", "guide.workflow"}, node_ids)
            self.assertEqual(
                "Applied through the canonical application service.",
                workflow.summary,
            )
            self.assertFalse((root / "docs/content/foundation.md").exists())
            self.assertFalse((root / "docs/content/proof.toml").exists())
            self.assertTrue((root / "docs/content/foundation-moved.md").is_file())
            self.assertTrue((root / ".docforge/cache/index.sqlite3").is_file())
            self.assertTrue((root / ".docforge/rendered/manual.html").is_file())

            with self.assertRaisesRegex(DocForgeError, "cannot be modified") as closed:
                service.apply("apply-all", str(final["changeset_hash"]))
            self.assertEqual("changeset_closed", closed.exception.code)

    def test_canonical_update_exchange_preserves_a_raced_external_edit(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            project = Project.open(root)
            store = ChangesetStore(project, "alpha-editor")
            proposal = store.register(
                "update-race",
                [
                    {
                        "operation": "update",
                        "node_id": "guide.workflow",
                        "metadata": {"summary": "Approved summary."},
                        "rationale": "Exercise the atomic update boundary.",
                    }
                ],
            )
            proposal_path = root / ".docforge/changesets/update-race.json"
            proposal_bytes = proposal_path.read_bytes()
            target = root / "docs/content/workflow.md"
            exchange = application_module.rename_exchange_between_at
            raced = False

            def race(
                first_directory_fd: int,
                first: str,
                second_directory_fd: int,
                second: str,
            ) -> None:
                nonlocal raced
                if second == target.name and not raced:
                    raced = True
                    target.write_bytes(target.read_bytes() + b"\nExternal edit at exchange.\n")
                exchange(first_directory_fd, first, second_directory_fd, second)

            with (
                mock.patch(
                    "docforge.application.rename_exchange_between_at",
                    side_effect=race,
                ),
                self.assertRaises(DocForgeError) as captured,
            ):
                store.apply(
                    changeset_id="update-race",
                    expected_changeset_hash=str(proposal["changeset_hash"]),
                    applier_id="alpha-editor",
                    application=GenericCanonicalApplier(project).apply,
                )

            self.assertEqual("base_conflict", captured.exception.code)
            self.assertIn("External edit at exchange.", target.read_text(encoding="utf-8"))
            self.assertEqual(proposal_bytes, proposal_path.read_bytes())
            self.assertFalse((root / ".docforge/changesets/.state/update-race.json").exists())
            self.assertFalse(tuple(target.parent.glob(".docforge-apply-*")))

    def test_explicit_applier_accepts_a_configured_contributor_changeset(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            descriptor = root / ".docforge/project.toml"
            descriptor.write_text(
                descriptor.read_text(encoding="utf-8")
                + """

[[changesets.writers]]
id = "contributor"
families = ["guide"]
operations = ["update"]
""",
                encoding="utf-8",
            )
            project = Project.open(root)
            proposal = ChangesetStore(project, "contributor").register(
                "contributor-update",
                [
                    {
                        "operation": "update",
                        "node_id": "guide.workflow",
                        "metadata": {"summary": "Accepted from a configured contributor."},
                        "rationale": "Prove explicit cross-identity acceptance.",
                    }
                ],
            )
            default_service = CanonicalApplicationService(
                project,
                applier_id="alpha-editor",
                applier=GenericCanonicalApplier(project),
            )

            with self.assertRaises(DocForgeError) as denied:
                default_service.apply(
                    "contributor-update",
                    str(proposal["changeset_hash"]),
                )

            self.assertEqual("changeset_owner_conflict", denied.exception.code)
            service = CanonicalApplicationService(
                project,
                applier_id="alpha-editor",
                applier=GenericCanonicalApplier(project),
                accepted_proposal_writers=("alpha-editor", "contributor"),
            )
            result = service.apply(
                "contributor-update",
                str(proposal["changeset_hash"]),
            )

            self.assertTrue(result["applied"])
            self.assertEqual("contributor", result["proposal_creator"])
            self.assertEqual("alpha-editor", result["applied_by"])
            self.assertEqual(
                ["alpha-editor", "contributor"],
                service.access()["accepted_proposal_writers"],
            )
            workflow = next(
                node for node in project.load().nodes if node.node_id == "guide.workflow"
            )
            self.assertEqual(
                "Accepted from a configured contributor.",
                workflow.summary,
            )

    def test_application_rejects_unknown_or_duplicate_accepted_writers(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            project = Project.open(root)

            with self.assertRaises(DocForgeError) as unknown:
                CanonicalApplicationService(
                    project,
                    applier_id="alpha-editor",
                    applier=GenericCanonicalApplier(project),
                    accepted_proposal_writers=("missing-writer",),
                )
            with self.assertRaises(DocForgeError) as duplicate:
                CanonicalApplicationService(
                    project,
                    applier_id="alpha-editor",
                    applier=GenericCanonicalApplier(project),
                    accepted_proposal_writers=("alpha-editor", "alpha-editor"),
                )
            with self.assertRaises(DocForgeError) as disabled:
                CanonicalApplicationService(
                    project,
                    applier_id=None,
                    applier=None,
                    accepted_proposal_writers=("alpha-editor",),
                )

            self.assertEqual("invalid_application_policy", unknown.exception.code)
            self.assertEqual("invalid_application_policy", duplicate.exception.code)
            self.assertEqual("invalid_application_policy", disabled.exception.code)

    def test_canonical_create_and_delete_races_preserve_foreign_targets(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            parent = Path(directory)

            create_root = self.copy_fixture(parent / "create")
            create_project = Project.open(create_root)
            create_store = ChangesetStore(create_project, "alpha-editor")
            create = create_store.register(
                "create-race",
                [
                    {
                        "operation": "create",
                        "node_id": "guide.raced",
                        "target_source": "docs/content/raced.md",
                        "metadata": self.new_metadata(),
                        "content": "Approved new content.",
                        "rationale": "Exercise no-replace creation.",
                    }
                ],
            )
            create_target = create_root / "docs/content/raced.md"
            real_link = application_module.os.link
            appeared = False

            def race_create(
                source: str,
                target: str,
                *,
                src_dir_fd: int,
                dst_dir_fd: int,
                follow_symlinks: bool,
            ) -> None:
                nonlocal appeared
                if target == create_target.name and not appeared:
                    appeared = True
                    create_target.write_bytes(b"foreign create target\n")
                real_link(
                    source,
                    target,
                    src_dir_fd=src_dir_fd,
                    dst_dir_fd=dst_dir_fd,
                    follow_symlinks=follow_symlinks,
                )

            with (
                mock.patch("docforge.application.os.link", side_effect=race_create),
                self.assertRaises(DocForgeError) as create_error,
            ):
                create_store.apply(
                    changeset_id="create-race",
                    expected_changeset_hash=str(create["changeset_hash"]),
                    applier_id="alpha-editor",
                    application=GenericCanonicalApplier(create_project).apply,
                )
            self.assertEqual("base_conflict", create_error.exception.code)
            self.assertEqual(b"foreign create target\n", create_target.read_bytes())
            self.assertFalse(tuple(create_target.parent.glob(".docforge-apply-*")))

            delete_root = self.copy_fixture(parent / "delete")
            delete_project = Project.open(delete_root)
            delete_store = ChangesetStore(delete_project, "alpha-editor")
            delete = delete_store.register(
                "delete-race",
                [
                    {
                        "operation": "delete",
                        "node_id": "proof.validation",
                        "relationship_changes": [
                            {
                                "action": "remove",
                                "source_id": "proof.validation",
                                "relation": "proves",
                                "target_id": "guide.workflow",
                            }
                        ],
                        "rationale": "Exercise atomic deletion.",
                    }
                ],
            )
            delete_target = delete_root / "docs/content/proof.toml"
            exchange = application_module.rename_exchange_between_at
            deleted_race = False

            def race_delete(
                first_directory_fd: int,
                first: str,
                second_directory_fd: int,
                second: str,
            ) -> None:
                nonlocal deleted_race
                if second == delete_target.name and not deleted_race:
                    deleted_race = True
                    delete_target.write_bytes(
                        delete_target.read_bytes() + b"\n# foreign delete edit\n"
                    )
                exchange(first_directory_fd, first, second_directory_fd, second)

            with (
                mock.patch(
                    "docforge.application.rename_exchange_between_at",
                    side_effect=race_delete,
                ),
                self.assertRaises(DocForgeError) as delete_error,
            ):
                delete_store.apply(
                    changeset_id="delete-race",
                    expected_changeset_hash=str(delete["changeset_hash"]),
                    applier_id="alpha-editor",
                    application=GenericCanonicalApplier(delete_project).apply,
                )
            self.assertEqual("base_conflict", delete_error.exception.code)
            self.assertIn("# foreign delete edit", delete_target.read_text(encoding="utf-8"))
            self.assertFalse(tuple(delete_target.parent.glob(".docforge-apply-*")))

    def test_delete_detach_race_restores_foreign_target_and_retains_original(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            project = Project.open(root)
            store = ChangesetStore(project, "alpha-editor")
            proposal = store.register(
                "delete-detach-race",
                [
                    {
                        "operation": "delete",
                        "node_id": "proof.validation",
                        "relationship_changes": [
                            {
                                "action": "remove",
                                "source_id": "proof.validation",
                                "relation": "proves",
                                "target_id": "guide.workflow",
                            }
                        ],
                        "rationale": "Race the final no-replace canonical detachment.",
                    }
                ],
            )
            target = root / "docs/content/proof.toml"
            original = target.read_bytes()
            move = application_module.rename_noreplace_between_at
            raced = False

            def race_detach(
                source_directory_fd: int,
                source: str,
                target_directory_fd: int,
                destination: str,
            ) -> bool:
                nonlocal raced
                if source == target.name and destination.startswith(".detached-") and not raced:
                    raced = True
                    replacement = target.with_name(".foreign-delete")
                    replacement.write_bytes(b"foreign replacement at delete detach\n")
                    os.replace(replacement, target)
                return move(
                    source_directory_fd,
                    source,
                    target_directory_fd,
                    destination,
                )

            with (
                mock.patch(
                    "docforge.application.rename_noreplace_between_at",
                    side_effect=race_detach,
                ),
                self.assertRaises(DocForgeError) as captured,
            ):
                store.apply(
                    changeset_id="delete-detach-race",
                    expected_changeset_hash=str(proposal["changeset_hash"]),
                    applier_id="alpha-editor",
                    application=GenericCanonicalApplier(project).apply,
                )

            self.assertTrue(raced)
            self.assertEqual("application_recovery_required", captured.exception.code)
            self.assertEqual(b"foreign replacement at delete detach\n", target.read_bytes())
            conflicts = captured.exception.details["conflicts"]
            retained = root / conflicts[0]["retained"]
            self.assertEqual(original, retained.read_bytes())
            self.assertFalse(
                (root / ".docforge/changesets/.state/delete-detach-race.json").exists()
            )

    def test_create_rollback_detach_race_never_unlinks_foreign_target(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            project = Project.open(root)
            store = ChangesetStore(project, "alpha-editor")
            proposal = store.register(
                "create-rollback-detach-race",
                [
                    {
                        "operation": "create",
                        "node_id": "guide.created",
                        "target_source": "docs/content/a-created.md",
                        "metadata": self.new_metadata(),
                        "content": "Approved content that publishes first.",
                        "rationale": "Exercise create rollback detachment.",
                    },
                    {
                        "operation": "update",
                        "node_id": "guide.workflow",
                        "metadata": {"summary": "Synthetic failing second publication."},
                        "rationale": "Trigger rollback after create publication.",
                    },
                ],
            )
            target = root / "docs/content/a-created.md"
            publish = GenericCanonicalApplier._publish
            move = application_module.rename_noreplace_between_at
            publish_calls = 0
            raced = False

            def fail_second(
                applier: GenericCanonicalApplier,
                publication: Any,
            ) -> None:
                nonlocal publish_calls
                publish_calls += 1
                if publish_calls == 1:
                    publish(applier, publication)
                    return
                raise DocForgeError("application_failure", "Synthetic second publication failure")

            def race_rollback_detach(
                source_directory_fd: int,
                source: str,
                target_directory_fd: int,
                destination: str,
            ) -> bool:
                nonlocal raced
                if source == target.name and destination.startswith(".detached-") and not raced:
                    raced = True
                    replacement = target.with_name(".foreign-create-rollback")
                    replacement.write_bytes(b"foreign replacement during create rollback\n")
                    os.replace(replacement, target)
                return move(
                    source_directory_fd,
                    source,
                    target_directory_fd,
                    destination,
                )

            with (
                mock.patch.object(
                    GenericCanonicalApplier,
                    "_publish",
                    autospec=True,
                    side_effect=fail_second,
                ),
                mock.patch(
                    "docforge.application.rename_noreplace_between_at",
                    side_effect=race_rollback_detach,
                ),
                self.assertRaises(DocForgeError) as captured,
            ):
                store.apply(
                    changeset_id="create-rollback-detach-race",
                    expected_changeset_hash=str(proposal["changeset_hash"]),
                    applier_id="alpha-editor",
                    application=GenericCanonicalApplier(project).apply,
                )

            self.assertTrue(raced)
            self.assertEqual("application_recovery_required", captured.exception.code)
            self.assertEqual(
                b"foreign replacement during create rollback\n",
                target.read_bytes(),
            )
            self.assertFalse(
                (root / ".docforge/changesets/.state/create-rollback-detach-race.json").exists()
            )

    def test_rollback_never_clobbers_a_foreign_edit_and_retains_original_bytes(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            project = Project.open(root)
            store = ChangesetStore(project, "alpha-editor")
            proposal = store.register(
                "rollback-race",
                [
                    {
                        "operation": "update",
                        "node_id": "guide.foundation",
                        "metadata": {"summary": "First approved update."},
                        "rationale": "Publish before the synthetic failure.",
                    },
                    {
                        "operation": "update",
                        "node_id": "guide.workflow",
                        "metadata": {"summary": "Second approved update."},
                        "rationale": "Trigger rollback after the first publication.",
                    },
                ],
            )
            first_target = root / "docs/content/foundation.md"
            first_before = first_target.read_bytes()
            publish = GenericCanonicalApplier._publish
            calls = 0

            def fail_after_foreign_edit(
                applier: GenericCanonicalApplier,
                publication: Any,
            ) -> None:
                nonlocal calls
                calls += 1
                if calls == 1:
                    publish(applier, publication)
                    first_target.write_bytes(
                        first_target.read_bytes() + b"\nForeign edit after publication.\n"
                    )
                    return
                raise DocForgeError("application_failure", "Synthetic second-target failure")

            with (
                mock.patch.object(
                    GenericCanonicalApplier,
                    "_publish",
                    autospec=True,
                    side_effect=fail_after_foreign_edit,
                ),
                self.assertRaises(DocForgeError) as captured,
            ):
                store.apply(
                    changeset_id="rollback-race",
                    expected_changeset_hash=str(proposal["changeset_hash"]),
                    applier_id="alpha-editor",
                    application=GenericCanonicalApplier(project).apply,
                )

            self.assertEqual("application_recovery_required", captured.exception.code)
            conflicts = captured.exception.details["conflicts"]
            self.assertEqual("target_or_backup_changed", conflicts[0]["reason"])
            self.assertIn(
                "Foreign edit after publication.",
                first_target.read_text(encoding="utf-8"),
            )
            retained = root / conflicts[0]["retained"]
            self.assertTrue(retained.is_file())
            self.assertEqual(first_before, retained.read_bytes())
            self.assertFalse((root / ".docforge/changesets/.state/rollback-race.json").exists())

    def test_canonical_update_preserves_existing_file_mode(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            target = root / "docs/content/workflow.md"
            target.chmod(0o6750)
            before = target.stat()
            project = Project.open(root)
            store = ChangesetStore(project, "alpha-editor")
            proposal = store.register(
                "mode",
                [
                    {
                        "operation": "update",
                        "node_id": "guide.workflow",
                        "metadata": {"summary": "Mode-preserving update."},
                        "rationale": "Preserve canonical file permissions.",
                    }
                ],
            )

            result = store.apply(
                changeset_id="mode",
                expected_changeset_hash=str(proposal["changeset_hash"]),
                applier_id="alpha-editor",
                application=GenericCanonicalApplier(project).apply,
            )

            self.assertTrue(result["applied"])
            after = target.stat()
            self.assertEqual(0o6750, stat.S_IMODE(after.st_mode))
            self.assertEqual(before.st_uid, after.st_uid)
            self.assertEqual(before.st_gid, after.st_gid)
            self.assertEqual([], result["retained_recovery_files"])
            self.assertEqual("clean", result["application_recovery"]["status"])
            self.assertFalse(tuple((root / ".docforge/application").glob("transaction-*")))

    def test_nested_creation_fsyncs_each_new_directory_and_parent_entry(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            project = Project.open(root)
            store = ChangesetStore(project, "alpha-editor")
            proposal = store.register(
                "nested-durable",
                [
                    {
                        "operation": "create",
                        "node_id": "guide.nested",
                        "target_source": "docs/content/nested/deeper/guide.md",
                        "metadata": self.new_metadata(),
                        "content": "Nested canonical content.",
                        "rationale": "Prove durable nested-directory creation.",
                    }
                ],
            )
            real_fsync = os.fsync
            fsynced_directories: set[Path] = set()

            def record_fsync(descriptor: int) -> None:
                try:
                    path = Path(os.readlink(f"/proc/self/fd/{descriptor}"))
                    if path.is_dir():
                        fsynced_directories.add(path)
                except OSError:
                    pass
                real_fsync(descriptor)

            with mock.patch(
                "docforge.application.os.fsync",
                side_effect=record_fsync,
            ):
                result = store.apply(
                    changeset_id="nested-durable",
                    expected_changeset_hash=str(proposal["changeset_hash"]),
                    applier_id="alpha-editor",
                    application=GenericCanonicalApplier(project).apply,
                )

            self.assertTrue(result["applied"])
            for path in (
                root / "docs/content",
                root / "docs/content/nested",
                root / "docs/content/nested/deeper",
            ):
                self.assertIn(path, fsynced_directories)

    def test_post_commit_cleanup_failure_closes_proposal_with_recovery_record(
        self,
    ) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            project = Project.open(root)
            store = ChangesetStore(project, "alpha-editor")
            proposal = store.register(
                "cleanup-recovery",
                [
                    {
                        "operation": "update",
                        "node_id": "guide.workflow",
                        "metadata": {"summary": "Committed despite private cleanup failure."},
                        "rationale": "Persist actionable post-commit recovery evidence.",
                    }
                ],
            )
            real_unlink = os.unlink
            failed = False

            def fail_private_cleanup(
                path: str | bytes,
                *,
                dir_fd: int | None = None,
            ) -> None:
                nonlocal failed
                if (
                    isinstance(path, str)
                    and path.startswith("staged-")
                    and dir_fd is not None
                    and not failed
                ):
                    failed = True
                    raise PermissionError("synthetic private cleanup failure")
                real_unlink(path, dir_fd=dir_fd)

            with mock.patch(
                "docforge.application.os.unlink",
                side_effect=fail_private_cleanup,
            ):
                result = store.apply(
                    changeset_id="cleanup-recovery",
                    expected_changeset_hash=str(proposal["changeset_hash"]),
                    applier_id="alpha-editor",
                    application=GenericCanonicalApplier(project).apply,
                )

            self.assertTrue(failed)
            self.assertTrue(result["applied"])
            self.assertEqual("applied", result["lifecycle"]["status"])
            self.assertEqual(
                "cleanup_required",
                result["application_recovery"]["status"],
            )
            self.assertEqual(
                "cleanup_required",
                result["lifecycle"]["application_recovery"]["status"],
            )
            retained = result["application_recovery"]["retained"]
            self.assertTrue(retained)
            lifecycle_path = root / ".docforge/changesets/.state/cleanup-recovery.json"
            lifecycle = json.loads(lifecycle_path.read_text(encoding="utf-8"))
            self.assertEqual(
                "cleanup_required",
                lifecycle["application_recovery"]["status"],
            )
            with self.assertRaises(DocForgeError) as closed:
                store.apply(
                    changeset_id="cleanup-recovery",
                    expected_changeset_hash=str(proposal["changeset_hash"]),
                    applier_id="alpha-editor",
                    application=GenericCanonicalApplier(project).apply,
                )
            self.assertEqual("changeset_closed", closed.exception.code)

    def test_changeset_rollback_fsyncs_the_parent_directory(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            target = root / "proposal.json"
            real_fsync = os.fsync
            fsynced_modes: list[int] = []

            def record_fsync(descriptor: int) -> None:
                fsynced_modes.append(os.fstat(descriptor).st_mode)
                real_fsync(descriptor)

            for previous in (b"previous proposal\n", None):
                with self.subTest(previous=previous):
                    target.write_bytes(b"replacement proposal\n")
                    fsynced_modes.clear()

                    with mock.patch(
                        "docforge.changesets.os.fsync",
                        side_effect=record_fsync,
                    ):
                        ChangesetStore._restore(target, previous, root)

                    self.assertTrue(any(stat.S_ISDIR(mode) for mode in fsynced_modes))
                    if previous is None:
                        self.assertFalse(target.exists())
                    else:
                        self.assertEqual(previous, target.read_bytes())

    def test_abandoned_proposal_releases_overlap_and_stale_work_remains_active(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            project = Project.open(root)
            store = ChangesetStore(project, "alpha-editor")
            first = store.register(
                "first",
                [
                    {
                        "operation": "update",
                        "node_id": "guide.foundation",
                        "metadata": {"summary": "Abandoned proposal."},
                        "rationale": "Reserve then release this node.",
                    }
                ],
            )
            store.abandon(
                "first",
                str(first["changeset_hash"]),
                "The proposal is no longer wanted.",
            )
            second = store.register(
                "second",
                [
                    {
                        "operation": "update",
                        "node_id": "guide.foundation",
                        "metadata": {"summary": "Replacement proposal."},
                        "rationale": "Verify terminal proposals release conflicts.",
                    }
                ],
            )
            workflow = root / "docs/content/workflow.md"
            workflow.write_text(
                workflow.read_text(encoding="utf-8") + "\nUnrelated current fact.\n",
                encoding="utf-8",
            )

            active = store.list_changesets(include_history=False)
            stale = store.list_changesets(include_history=False, status="stale")

            self.assertEqual(0, active["count"])
            self.assertEqual(["second"], [item["changeset_id"] for item in stale["changesets"]])
            self.assertEqual("stale", stale["changesets"][0]["lifecycle"]["status"])
            self.assertEqual("ready", second["lifecycle"])

    def test_lifecycle_receipts_obey_the_changeset_size_limit(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            store = ChangesetStore(Project.open(root), "alpha-editor")
            created = store.create("bounded-lifecycle")

            with self.assertRaises(DocForgeError) as oversized:
                store.abandon(
                    "bounded-lifecycle",
                    str(created["changeset_hash"]),
                    "x" * 100_001,
                )

            self.assertEqual("changeset_too_large", oversized.exception.code)
            self.assertFalse((root / ".docforge/changesets/.state/bounded-lifecycle.json").exists())

    def test_applied_receipt_survives_a_derived_refresh_failure(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            project = Project.open(root)
            registered = ChangesetStore(project, "alpha-editor").register(
                "degraded-refresh",
                [
                    {
                        "operation": "update",
                        "node_id": "guide.workflow",
                        "metadata": {"summary": "Canonical even if refresh fails."},
                        "rationale": "Separate canonical success from disposable refresh.",
                    }
                ],
            )
            service = CanonicalApplicationService(
                project,
                applier_id="alpha-editor",
                applier=GenericCanonicalApplier(project),
            )
            with mock.patch.object(
                service.index,
                "build",
                side_effect=DocForgeError("index_failure", "Synthetic derived failure"),
            ):
                result = service.apply(
                    "degraded-refresh",
                    str(registered["changeset_hash"]),
                )

            self.assertTrue(result["applied"])
            self.assertEqual("applied", result["lifecycle"]["status"])
            self.assertEqual("degraded", result["derived_refresh"]["status"])
            self.assertEqual("index", result["derived_refresh"]["errors"][0]["component"])
            with self.assertRaisesRegex(DocForgeError, "cannot be modified"):
                service.apply(
                    "degraded-refresh",
                    str(registered["changeset_hash"]),
                )

    def test_relationship_only_update_is_hash_bound_and_does_not_rewrite_node(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            project = Project.open(root)
            store = ChangesetStore(project, "alpha-editor")
            workflow = next(
                node for node in project.load().nodes if node.node_id == "guide.workflow"
            )
            created = store.create("relationship-only")
            proposed = store.propose_relationship_update(
                changeset_id="relationship-only",
                expected_changeset_hash=str(created["changeset_hash"]),
                node_id="guide.workflow",
                expected_content_hash=workflow.content_hash,
                relationship_changes=[
                    {
                        "action": "remove",
                        "source_id": "guide.workflow",
                        "relation": "depends_on",
                        "target_id": "guide.foundation",
                    }
                ],
                rationale="Queue one relationship correction without changing node content.",
            )

            change = store.diff("relationship-only")["changes"][0]
            self.assertEqual("update", change["operation"])
            self.assertEqual("", change["content_diff"])
            self.assertEqual({}, change["metadata"])
            self.assertEqual(1, proposed["projected_edge_count"])
            with self.assertRaisesRegex(DocForgeError, "at least one"):
                store.propose_relationship_update(
                    changeset_id="relationship-only",
                    expected_changeset_hash=str(proposed["changeset_hash"]),
                    node_id="guide.foundation",
                    expected_content_hash=self.node_hash(project, "guide.foundation"),
                    relationship_changes=[],
                    rationale="Reject an empty relationship operation.",
                )

    def test_optimistic_and_cross_changeset_conflicts_preserve_both_proposals(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            project = Project.open(root)
            store = ChangesetStore(project, "alpha-editor")
            first = store.create("first")
            second = store.create("second")
            foundation_hash = self.node_hash(project, "guide.foundation")
            first_result = store.propose_update(
                changeset_id="first",
                expected_changeset_hash=first["changeset_hash"],
                node_id="guide.foundation",
                expected_content_hash=foundation_hash,
                metadata={"summary": "First proposal owns this node."},
                content=None,
                relationship_changes=[],
                rationale="Claim the foundation node in the first proposal.",
            )

            with self.assertRaisesRegex(DocForgeError, "overlaps another proposal") as conflict:
                store.propose_update(
                    changeset_id="second",
                    expected_changeset_hash=second["changeset_hash"],
                    node_id="guide.foundation",
                    expected_content_hash=foundation_hash,
                    metadata={"summary": "Second proposal conflicts with the first."},
                    content=None,
                    relationship_changes=[],
                    rationale="Prove exact overlap detection.",
                )
            self.assertEqual("proposal_conflict", conflict.exception.code)
            self.assertEqual("first", conflict.exception.details["conflicts"][0]["changeset_id"])
            self.assertEqual(0, store.validate("second")["operation_count"])

            with self.assertRaisesRegex(DocForgeError, "changed after") as stale_hash:
                store.propose_create(
                    changeset_id="first",
                    expected_changeset_hash=first["changeset_hash"],
                    node_id="guide.another",
                    target_source="docs/content/another.md",
                    metadata=self.new_metadata(),
                    content="Another proposal operation.",
                    relationship_changes=[],
                    rationale="Use an obsolete changeset hash.",
                )
            self.assertEqual("changeset_conflict", stale_hash.exception.code)
            self.assertEqual(
                first_result["changeset_hash"], store.validate("first")["changeset_hash"]
            )

    def test_simultaneous_process_appends_serialize_without_losing_an_operation(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            project = Project.open(root)
            initial = ChangesetStore(project, "alpha-editor").create("simultaneous")
            hashes = {
                node.node_id: node.content_hash
                for node in project.load().nodes
                if node.node_id in {"guide.foundation", "guide.workflow"}
            }
            context = multiprocessing.get_context("spawn")
            start = context.Event()
            receivers: list[Any] = []
            processes: list[multiprocessing.Process] = []
            for node_id in ("guide.foundation", "guide.workflow"):
                receiver, sender = context.Pipe(duplex=False)
                process = context.Process(
                    target=_append_in_process,
                    args=(
                        str(root),
                        node_id,
                        initial["changeset_hash"],
                        hashes[node_id],
                        start,
                        sender,
                    ),
                )
                process.start()
                sender.close()
                receivers.append(receiver)
                processes.append(process)
            start.set()
            results = sorted(receiver.recv() for receiver in receivers)
            for receiver in receivers:
                receiver.close()
            for process in processes:
                process.join(timeout=10)
                self.assertEqual(0, process.exitcode)

            self.assertEqual(["changeset_conflict", "ok"], results)
            validated = ChangesetStore(project).validate("simultaneous")
            self.assertEqual(1, validated["operation_count"])

    def test_configured_changeset_count_operation_and_size_limits_fail_atomically(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            parent = Path(directory)

            count_root = self.copy_fixture(parent / "count")
            descriptor = count_root / ".docforge/project.toml"
            descriptor.write_text(
                descriptor.read_text(encoding="utf-8")
                .replace("max_changesets = 20", "max_changesets = 1")
                .replace("max_changeset_operations = 20", "max_changeset_operations = 1"),
                encoding="utf-8",
            )
            count_project = Project.open(count_root)
            count_store = ChangesetStore(count_project, "alpha-editor")
            limited = count_store.create("limited")
            with self.assertRaisesRegex(DocForgeError, "changeset limit") as count_error:
                count_store.create("excess")
            self.assertEqual("changeset_limit", count_error.exception.code)
            self.assertFalse((count_root / ".docforge/changesets/excess.json").exists())

            first = count_store.propose_update(
                changeset_id="limited",
                expected_changeset_hash=limited["changeset_hash"],
                node_id="guide.foundation",
                expected_content_hash=self.node_hash(count_project, "guide.foundation"),
                metadata={"summary": "The only permitted operation."},
                content=None,
                relationship_changes=[],
                rationale="Reach the configured operation limit.",
            )
            limited_path = count_root / ".docforge/changesets/limited.json"
            limited_bytes = limited_path.read_bytes()
            with self.assertRaisesRegex(DocForgeError, "operation limit") as operation_error:
                count_store.propose_update(
                    changeset_id="limited",
                    expected_changeset_hash=first["changeset_hash"],
                    node_id="guide.workflow",
                    expected_content_hash=self.node_hash(count_project, "guide.workflow"),
                    metadata={"summary": "This operation exceeds the limit."},
                    content=None,
                    relationship_changes=[],
                    rationale="Prove the configured operation cap.",
                )
            self.assertEqual("changeset_operation_limit", operation_error.exception.code)
            self.assertEqual(limited_bytes, limited_path.read_bytes())

            size_root = self.copy_fixture(parent / "size")
            size_descriptor = size_root / ".docforge/project.toml"
            size_descriptor.write_text(
                size_descriptor.read_text(encoding="utf-8").replace(
                    "max_changeset_bytes = 100000", "max_changeset_bytes = 64"
                ),
                encoding="utf-8",
            )
            size_store = ChangesetStore(Project.open(size_root), "alpha-editor")
            with self.assertRaisesRegex(DocForgeError, "size limit") as size_error:
                size_store.create("too-large")
            self.assertEqual("changeset_too_large", size_error.exception.code)
            self.assertFalse((size_root / ".docforge/changesets/too-large.json").exists())

    def test_stale_base_bad_content_hash_and_failed_delete_are_atomic(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            parent = Path(directory)

            stale_root = self.copy_fixture(parent / "stale")
            stale_project = Project.open(stale_root)
            stale_store = ChangesetStore(stale_project, "alpha-editor")
            stale = stale_store.create("stale")
            stale_path = stale_root / ".docforge/changesets/stale.json"
            stale_bytes = stale_path.read_bytes()
            workflow = stale_root / "docs/content/workflow.md"
            workflow.write_text(workflow.read_text() + "\nCanonical change.\n", encoding="utf-8")
            with self.assertRaisesRegex(DocForgeError, "Canonical project changed") as base:
                stale_store.propose_move(
                    changeset_id="stale",
                    expected_changeset_hash=stale["changeset_hash"],
                    node_id="guide.foundation",
                    expected_content_hash=self.node_hash(
                        Project.open(stale_root), "guide.foundation"
                    ),
                    target_source="docs/content/foundation-moved.md",
                    rationale="This proposal has an obsolete base.",
                )
            self.assertEqual("base_conflict", base.exception.code)
            self.assertEqual(stale_bytes, stale_path.read_bytes())
            self.assertEqual("stale", stale_store.inspect("stale")["base_state"])
            self.assertEqual("stale", stale_store.list_changesets()["changesets"][0]["base_state"])

            delete_root = self.copy_fixture(parent / "delete")
            delete_project = Project.open(delete_root)
            delete_store = ChangesetStore(delete_project, "alpha-editor")
            delete = delete_store.create("delete")
            delete_path = delete_root / ".docforge/changesets/delete.json"
            delete_bytes = delete_path.read_bytes()
            with self.assertRaisesRegex(DocForgeError, "every incident relationship") as edges:
                delete_store.propose_delete(
                    changeset_id="delete",
                    expected_changeset_hash=delete["changeset_hash"],
                    node_id="proof.validation",
                    expected_content_hash=self.node_hash(delete_project, "proof.validation"),
                    relationship_changes=[],
                    rationale="An incomplete delete must not be stored.",
                )
            self.assertEqual("unresolved_relationships", edges.exception.code)
            self.assertEqual(delete_bytes, delete_path.read_bytes())

            with self.assertRaisesRegex(DocForgeError, "changed after"):
                delete_store.propose_update(
                    changeset_id="delete",
                    expected_changeset_hash=delete["changeset_hash"],
                    node_id="guide.foundation",
                    expected_content_hash="0" * 64,
                    metadata={"summary": "Wrong base hash."},
                    content=None,
                    relationship_changes=[],
                    rationale="Reject a mismatched node hash.",
                )
            self.assertEqual(delete_bytes, delete_path.read_bytes())

    def test_writer_family_permissions_ownership_and_paths_fail_closed(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            descriptor = root / ".docforge/project.toml"
            descriptor.write_text(
                descriptor.read_text().replace(
                    'families = ["guide", "proof"]', 'families = ["guide"]'
                )
                + """

[[changesets.writers]]
id = "other-editor"
families = ["guide"]
operations = ["create", "update", "move", "delete"]
""",
                encoding="utf-8",
            )
            project = Project.open(root)
            store = ChangesetStore(project, "alpha-editor")
            changeset = store.create("permissions")
            path = root / ".docforge/changesets/permissions.json"
            before = path.read_bytes()

            with self.assertRaisesRegex(DocForgeError, "families") as family:
                store.propose_create(
                    changeset_id="permissions",
                    expected_changeset_hash=changeset["changeset_hash"],
                    node_id="proof.new",
                    target_source="docs/content/proof-new.md",
                    metadata=self.new_metadata(family="proof"),
                    content="A forbidden family proposal.",
                    relationship_changes=[],
                    rationale="Prove family permission enforcement.",
                )
            self.assertEqual("family_forbidden", family.exception.code)
            self.assertEqual(before, path.read_bytes())

            other = ChangesetStore(project, "other-editor")
            with self.assertRaisesRegex(DocForgeError, "does not own") as owner:
                other.propose_create(
                    changeset_id="permissions",
                    expected_changeset_hash=changeset["changeset_hash"],
                    node_id="guide.other",
                    target_source="docs/content/other.md",
                    metadata=self.new_metadata(),
                    content="Another writer cannot append here.",
                    relationship_changes=[],
                    rationale="Prove immutable changeset ownership.",
                )
            self.assertEqual("changeset_owner_conflict", owner.exception.code)

            with self.assertRaisesRegex(DocForgeError, "canonical content root"):
                store.propose_create(
                    changeset_id="permissions",
                    expected_changeset_hash=changeset["changeset_hash"],
                    node_id="guide.escape",
                    target_source="outside.md",
                    metadata=self.new_metadata(),
                    content="This path is outside the content root.",
                    relationship_changes=[],
                    rationale="Prove target confinement.",
                )

            outside = Path(directory) / "outside.json"
            outside.write_text(json.dumps({"secret": True}), encoding="utf-8")
            symlink = root / ".docforge/changesets/symlink.json"
            symlink.symlink_to(outside)
            with self.assertRaisesRegex(DocForgeError, "symbolic links"):
                store.validate("symlink")
            self.assertEqual({"secret": True}, json.loads(outside.read_text()))

    def test_changeset_configuration_overlap_and_invalid_permissions_are_rejected(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            parent = Path(directory)
            overlap_root = self.copy_fixture(parent / "overlap")
            overlap = overlap_root / ".docforge/project.toml"
            overlap.write_text(
                overlap.read_text().replace(
                    'root = ".docforge/changesets"', 'root = "docs/content/changesets"'
                ),
                encoding="utf-8",
            )
            with self.assertRaisesRegex(DocForgeError, "must not overlap"):
                Project.open(overlap_root)

            invalid_root = self.copy_fixture(parent / "invalid")
            invalid = invalid_root / ".docforge/project.toml"
            invalid.write_text(
                invalid.read_text().replace(
                    'operations = ["create", "update", "move", "delete"]',
                    'operations = ["create", "execute"]',
                ),
                encoding="utf-8",
            )
            with self.assertRaisesRegex(DocForgeError, "operations are empty or invalid"):
                Project.open(invalid_root)

            symlink_root = self.copy_fixture(parent / "symlink")
            outside = parent / "outside-changesets"
            outside.mkdir()
            (symlink_root / ".docforge/changesets").symlink_to(outside, target_is_directory=True)
            with self.assertRaisesRegex(DocForgeError, "outside the project root"):
                Project.open(symlink_root)

    def test_toml_anchors_and_mid_write_source_set_changes_fail_atomically(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            parent = Path(directory)

            toml_root = self.copy_fixture(parent / "toml")
            toml_store = ChangesetStore(Project.open(toml_root), "alpha-editor")
            changeset = toml_store.create("toml-create")
            path = toml_root / ".docforge/changesets/toml-create.json"
            before = path.read_bytes()
            with self.assertRaisesRegex(DocForgeError, "explicit source_anchor"):
                toml_store.propose_create(
                    changeset_id="toml-create",
                    expected_changeset_hash=changeset["changeset_hash"],
                    node_id="proof.second",
                    target_source="docs/content/proof.toml",
                    metadata=self.new_metadata(family="proof"),
                    content="A TOML proposal needs an explicit anchor.",
                    relationship_changes=[],
                    rationale="Reject ambiguous TOML placement.",
                )
            self.assertEqual(before, path.read_bytes())

            duplicate_anchor = {
                **self.new_metadata(family="proof"),
                "source_anchor": "node-1",
            }
            with self.assertRaisesRegex(DocForgeError, "anchors must be unique"):
                toml_store.propose_create(
                    changeset_id="toml-create",
                    expected_changeset_hash=changeset["changeset_hash"],
                    node_id="proof.second",
                    target_source="docs/content/proof.toml",
                    metadata=duplicate_anchor,
                    content="A duplicate anchor is still ambiguous.",
                    relationship_changes=[],
                    rationale="Reject duplicate TOML anchors.",
                )
            self.assertEqual(before, path.read_bytes())

            race_root = self.copy_fixture(parent / "race")
            race_project = Project.open(race_root)
            race_store = ChangesetStore(race_project, "alpha-editor")
            sources = race_project.canonical_source_paths()
            invented = race_root / "docs/content/invented.md"
            with (
                mock.patch.object(
                    race_project,
                    "canonical_source_paths",
                    side_effect=[sources, (*sources, invented)],
                ),
                self.assertRaisesRegex(DocForgeError, "changed during changeset storage"),
            ):
                race_store.create("source-race")
            self.assertFalse((race_root / ".docforge/changesets/source-race.json").exists())


if __name__ == "__main__":
    unittest.main()