DocForge2/tests/test_changesets.py

from __future__ import annotations

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

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_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_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()