DocForge2/src/docforge/changesets.py

"""Confined atomic storage for project-bound proposal changesets."""

from __future__ import annotations

import fcntl
import json
import os
import tempfile
from collections.abc import Callable, Generator, Mapping
from contextlib import contextmanager
from pathlib import Path
from typing import Any, cast

from .changeset_contract import (
    document_hash,
    normalize_operation,
    validate_document,
    validate_hash,
    validate_id,
)
from .errors import DocForgeError
from .models import Edge, Node, ProjectService, ProjectSnapshot, ProposalWriter
from .project import project_root_fingerprint
from .proposal_projection import ProposalProjector


class ChangesetStore:
    """One project-bound proposal store with an optional immutable writer identity."""

    def __init__(self, project: ProjectService, writer_id: str | None = None) -> None:
        self.project = project
        self.projector = ProposalProjector(project)
        writers = {writer.writer_id: writer for writer in project.descriptor.proposal_writers}
        if writer_id is not None and writer_id not in writers:
            raise DocForgeError(
                "unknown_writer",
                "Proposal writer is not configured for this project",
                writer=writer_id,
            )
        self.writer = writers.get(writer_id) if writer_id is not None else None

    @property
    def writer_id(self) -> str | None:
        return self.writer.writer_id if self.writer is not None else None

    def access(self) -> dict[str, object]:
        if self.writer is None:
            return {"enabled": False, "writer": None, "families": [], "operations": []}
        return {
            "enabled": True,
            "writer": self.writer.writer_id,
            "families": list(self.writer.families),
            "operations": list(self.writer.operations),
        }

    def create(self, changeset_id: str) -> dict[str, object]:
        writer = self._require_writer()
        validate_id(changeset_id, "changeset_id")
        with self._lock():
            path = self._path(changeset_id)
            if path.exists():
                raise DocForgeError(
                    "changeset_exists", "Changeset ID already exists", changeset_id=changeset_id
                )
            existing = tuple(self._root().glob("*.json"))
            if len(existing) >= self.project.descriptor.limits.max_changesets:
                raise DocForgeError("changeset_limit", "Project changeset limit has been reached")
            snapshot = self.project.load()
            document: dict[str, Any] = {
                "schema_version": 1,
                "changeset_id": changeset_id,
                "project_id": snapshot.descriptor.project_id,
                "root_fingerprint": project_root_fingerprint(snapshot.descriptor.root),
                "base_revision": snapshot.revision,
                "base_source_hash": snapshot.source_hash,
                "creator": writer.writer_id,
                "operations": [],
            }
            self._write(path, document)
            return self._result(snapshot, document, valid=True)

    def register(
        self,
        changeset_id: str,
        operations: list[dict[str, Any]],
    ) -> dict[str, object]:
        """Create and validate one complete proposal in a single atomic write."""

        writer = self._require_writer()
        validate_id(changeset_id, "changeset_id")
        if not operations:
            raise DocForgeError(
                "empty_changeset",
                "Registered changes require at least one operation",
            )
        with self._lock():
            path = self._path(changeset_id)
            if path.exists():
                raise DocForgeError(
                    "changeset_exists",
                    "Changeset ID already exists",
                    changeset_id=changeset_id,
                )
            existing = tuple(self._root().glob("*.json"))
            if len(existing) >= self.project.descriptor.limits.max_changesets:
                raise DocForgeError("changeset_limit", "Project changeset limit has been reached")
            if len(operations) > self.project.descriptor.limits.max_changeset_operations:
                raise DocForgeError(
                    "changeset_operation_limit",
                    "Changeset operation limit has been reached",
                )
            snapshot = self.project.load()
            nodes = {node.node_id: node for node in snapshot.nodes}
            normalized = [
                normalize_operation(
                    self._complete_operation(operation, nodes),
                    sequence=sequence,
                )
                for sequence, operation in enumerate(operations, start=1)
            ]
            if len({item["node_id"] for item in normalized}) != len(normalized):
                raise DocForgeError(
                    "duplicate_operation",
                    "A changeset may touch a node only once",
                )
            document: dict[str, Any] = {
                "schema_version": 1,
                "changeset_id": changeset_id,
                "project_id": snapshot.descriptor.project_id,
                "root_fingerprint": project_root_fingerprint(snapshot.descriptor.root),
                "base_revision": snapshot.revision,
                "base_source_hash": snapshot.source_hash,
                "creator": writer.writer_id,
                "operations": normalized,
            }
            projected_nodes, projected_edges = self.projector.project(snapshot, document)
            self._check_proposal_conflicts(document, snapshot)
            self._write(path, document)
            return self._result(
                snapshot,
                document,
                valid=True,
                lifecycle="ready",
                ready_for_review=True,
                projected_node_count=len(projected_nodes),
                projected_edge_count=len(projected_edges),
            )

    def propose_create(
        self,
        *,
        changeset_id: str,
        expected_changeset_hash: str,
        node_id: str,
        target_source: str,
        metadata: dict[str, Any],
        content: str,
        relationship_changes: list[dict[str, Any]],
        rationale: str,
    ) -> dict[str, object]:
        return self._append(
            changeset_id,
            expected_changeset_hash,
            {
                "operation": "create",
                "node_id": node_id,
                "expected_content_hash": None,
                "target_source": target_source,
                "metadata": metadata,
                "content": content,
                "relationship_changes": relationship_changes,
                "rationale": rationale,
            },
        )

    def propose_update(
        self,
        *,
        changeset_id: str,
        expected_changeset_hash: str,
        node_id: str,
        expected_content_hash: str,
        metadata: dict[str, Any] | None,
        content: str | None,
        relationship_changes: list[dict[str, Any]],
        rationale: str,
    ) -> dict[str, object]:
        return self._append(
            changeset_id,
            expected_changeset_hash,
            {
                "operation": "update",
                "node_id": node_id,
                "expected_content_hash": expected_content_hash,
                "target_source": None,
                "metadata": metadata,
                "content": content,
                "relationship_changes": relationship_changes,
                "rationale": rationale,
            },
        )

    def propose_move(
        self,
        *,
        changeset_id: str,
        expected_changeset_hash: str,
        node_id: str,
        expected_content_hash: str,
        target_source: str,
        rationale: str,
    ) -> dict[str, object]:
        return self._append(
            changeset_id,
            expected_changeset_hash,
            {
                "operation": "move",
                "node_id": node_id,
                "expected_content_hash": expected_content_hash,
                "target_source": target_source,
                "metadata": None,
                "content": None,
                "relationship_changes": [],
                "rationale": rationale,
            },
        )

    def propose_relationship_update(
        self,
        *,
        changeset_id: str,
        expected_changeset_hash: str,
        node_id: str,
        expected_content_hash: str,
        relationship_changes: list[dict[str, Any]],
        rationale: str,
    ) -> dict[str, object]:
        """Queue relationship-only changes against one exact existing node."""

        if not relationship_changes:
            raise DocForgeError(
                "invalid_operation", "Relationship-only updates require at least one change"
            )
        return self.propose_update(
            changeset_id=changeset_id,
            expected_changeset_hash=expected_changeset_hash,
            node_id=node_id,
            expected_content_hash=expected_content_hash,
            metadata=None,
            content=None,
            relationship_changes=relationship_changes,
            rationale=rationale,
        )

    def propose_delete(
        self,
        *,
        changeset_id: str,
        expected_changeset_hash: str,
        node_id: str,
        expected_content_hash: str,
        relationship_changes: list[dict[str, Any]],
        rationale: str,
    ) -> dict[str, object]:
        return self._append(
            changeset_id,
            expected_changeset_hash,
            {
                "operation": "delete",
                "node_id": node_id,
                "expected_content_hash": expected_content_hash,
                "target_source": None,
                "metadata": None,
                "content": None,
                "relationship_changes": relationship_changes,
                "rationale": rationale,
            },
        )

    def validate(self, changeset_id: str) -> dict[str, object]:
        validate_id(changeset_id, "changeset_id")
        with self._lock():
            snapshot, document, nodes, edges = self._validate_locked(changeset_id)
            return self._result(
                snapshot,
                document,
                valid=True,
                projected_node_count=len(nodes),
                projected_edge_count=len(edges),
            )

    def list_changesets(
        self,
        *,
        include_history: bool = True,
        status: str | None = None,
    ) -> dict[str, object]:
        with self._lock():
            snapshot = self.project.load()
            records: list[dict[str, object]] = []
            for path in sorted(self._root().glob("*.json"), key=lambda item: item.name):
                document = self._read(path)
                base_state = self._base_state(document, snapshot)
                lifecycle = self._lifecycle(document, base_state)
                if status is not None and lifecycle["status"] != status:
                    continue
                if (
                    status is None
                    and not include_history
                    and lifecycle["status"] in {"abandoned", "applied", "stale"}
                ):
                    continue
                records.append(
                    {
                        "changeset_id": document["changeset_id"],
                        "changeset_hash": document_hash(document),
                        "creator": document["creator"],
                        "base_revision": document["base_revision"],
                        "base_source_hash": document["base_source_hash"],
                        "base_state": base_state,
                        "lifecycle": lifecycle,
                        "operation_count": len(document["operations"]),
                    }
                )
            return self._base_result(snapshot, count=len(records), changesets=records)

    def inspect(self, changeset_id: str) -> dict[str, object]:
        validate_id(changeset_id, "changeset_id")
        with self._lock():
            document = self._read(self._path(changeset_id))
            snapshot = self.project.load()
            return self._result(
                snapshot,
                document,
                base_state=self._base_state(document, snapshot),
                lifecycle=self._lifecycle(
                    document,
                    self._base_state(document, snapshot),
                ),
            )

    def rebase(
        self,
        changeset_id: str,
        expected_changeset_hash: str,
    ) -> dict[str, object]:
        """Move a proposal to the current base when every touched fact is unchanged."""

        validate_id(changeset_id, "changeset_id")
        validate_hash(expected_changeset_hash, "expected_changeset_hash")
        with self._lock():
            path = self._path(changeset_id)
            document = self._read(path)
            actual_hash = document_hash(document)
            if actual_hash != expected_changeset_hash:
                raise DocForgeError(
                    "changeset_conflict",
                    "Changeset changed after the caller read it",
                    changeset_id=changeset_id,
                    expected=expected_changeset_hash,
                    actual=actual_hash,
                )
            snapshot = self.project.load()
            self._require_mutable(document, snapshot)
            if self._base_state(document, snapshot) == "current":
                return self._result(
                    snapshot,
                    document,
                    valid=True,
                    rebased=False,
                    lifecycle=self._lifecycle(document, "current"),
                )
            candidate = {
                **document,
                "base_revision": snapshot.revision,
                "base_source_hash": snapshot.source_hash,
            }
            nodes, edges = self.projector.project(snapshot, candidate)
            self._check_proposal_conflicts(candidate, snapshot)
            self._write(path, candidate)
            return self._result(
                snapshot,
                candidate,
                valid=True,
                rebased=True,
                lifecycle="ready",
                projected_node_count=len(nodes),
                projected_edge_count=len(edges),
            )

    def abandon(
        self,
        changeset_id: str,
        expected_changeset_hash: str,
        reason: str,
    ) -> dict[str, object]:
        """Mark one proposal as abandoned without deleting its audit record."""

        validate_id(changeset_id, "changeset_id")
        validate_hash(expected_changeset_hash, "expected_changeset_hash")
        if not reason.strip():
            raise DocForgeError("invalid_operation", "Abandon reason must be non-empty")
        with self._lock():
            document = self._read(self._path(changeset_id))
            actual_hash = document_hash(document)
            if actual_hash != expected_changeset_hash:
                raise DocForgeError(
                    "changeset_conflict",
                    "Changeset changed after the caller read it",
                    changeset_id=changeset_id,
                    expected=expected_changeset_hash,
                    actual=actual_hash,
                )
            snapshot = self.project.load()
            self._require_mutable(document, snapshot)
            receipt = self._write_state(
                changeset_id,
                {
                    "status": "abandoned",
                    "changeset_hash": actual_hash,
                    "reason": reason.strip(),
                    "revision": snapshot.revision,
                    "source_hash": snapshot.source_hash,
                },
            )
            return self._result(
                snapshot,
                document,
                base_state=self._base_state(document, snapshot),
                lifecycle=receipt,
            )

    def diff(self, changeset_id: str) -> dict[str, object]:
        validate_id(changeset_id, "changeset_id")
        with self._lock():
            snapshot, document, _, _ = self._validate_locked(changeset_id)
            nodes = {node.node_id: node for node in snapshot.nodes}
            edges = {(edge.source_id, edge.relation, edge.target_id) for edge in snapshot.edges}
            changes: list[dict[str, object]] = []
            for operation in document["operations"]:
                before = nodes.get(operation["node_id"])
                before_edges = set(edges)
                self.projector.apply_operation(
                    snapshot, nodes, edges, operation, self.projector.writer(document)
                )
                after = nodes.get(operation["node_id"])
                changes.append(
                    self.projector.operation_diff(
                        operation,
                        before,
                        after,
                        sorted(edges - before_edges),
                        sorted(before_edges - edges),
                    )
                )
            return self._result(snapshot, document, valid=True, changes=changes)

    def projected_snapshot(self, changeset_id: str) -> tuple[ProjectSnapshot, str]:
        """Return a validated in-memory proposal projection for derived preview use."""

        validate_id(changeset_id, "changeset_id")
        with self._lock():
            snapshot, document, nodes, edges = self._validate_locked(changeset_id)
            projected = ProjectSnapshot(
                descriptor=snapshot.descriptor,
                nodes=tuple(sorted(nodes.values(), key=lambda node: node.node_id)),
                edges=tuple(Edge(*edge) for edge in sorted(edges)),
                source_hash=snapshot.source_hash,
                revision=snapshot.revision,
            )
            return projected, document_hash(document)

    def apply(
        self,
        *,
        changeset_id: str,
        expected_changeset_hash: str,
        applier_id: str,
        application: Callable[
            [ProjectSnapshot, ProjectSnapshot, tuple[Mapping[str, object], ...]],
            dict[str, object],
        ],
    ) -> dict[str, object]:
        """Apply one exact validated proposal through a project-owned canonical serializer."""

        validate_id(changeset_id, "changeset_id")
        validate_hash(expected_changeset_hash, "expected_changeset_hash")
        if self.writer is None or self.writer.writer_id != applier_id:
            raise DocForgeError(
                "canonical_application_disabled",
                "Canonical applier identity is not configured for this store",
            )
        with self._lock():
            document = self._read(self._path(changeset_id))
            snapshot = self.project.load()
            self._require_mutable(document, snapshot)
            self._check_base(document, snapshot)
            nodes, edges = self.projector.project(snapshot, document)
            self._check_proposal_conflicts(document, snapshot)
            actual_hash = document_hash(document)
            if actual_hash != expected_changeset_hash:
                raise DocForgeError(
                    "changeset_conflict",
                    "Changeset changed after the caller approved it",
                    changeset_id=changeset_id,
                    expected=expected_changeset_hash,
                    actual=actual_hash,
                )
            if document["creator"] != applier_id:
                raise DocForgeError(
                    "changeset_owner_conflict",
                    "Canonical applier does not own this changeset",
                    changeset_id=changeset_id,
                    owner=document["creator"],
                    applier=applier_id,
                )
            projected = ProjectSnapshot(
                descriptor=snapshot.descriptor,
                nodes=tuple(sorted(nodes.values(), key=lambda node: node.node_id)),
                edges=tuple(Edge(*edge) for edge in sorted(edges)),
                source_hash=snapshot.source_hash,
                revision=snapshot.revision,
            )
            payload = application(
                snapshot,
                projected,
                tuple(cast(Mapping[str, object], item) for item in document["operations"]),
            )
            current = self.project.load()
            lifecycle = self._write_state(
                changeset_id,
                {
                    "status": "applied",
                    "changeset_hash": actual_hash,
                    "revision": current.revision,
                    "source_hash": current.source_hash,
                },
            )
            return self._result(
                current,
                document,
                valid=True,
                applied=True,
                lifecycle=lifecycle,
                applied_from_revision=snapshot.revision,
                applied_from_source_hash=snapshot.source_hash,
                **payload,
            )

    def _append(
        self,
        changeset_id: str,
        expected_changeset_hash: str,
        operation: dict[str, Any],
    ) -> dict[str, object]:
        writer = self._require_writer()
        validate_id(changeset_id, "changeset_id")
        validate_hash(expected_changeset_hash, "expected_changeset_hash")
        with self._lock():
            path = self._path(changeset_id)
            document = self._read(path)
            actual_hash = document_hash(document)
            if actual_hash != expected_changeset_hash:
                raise DocForgeError(
                    "changeset_conflict",
                    "Changeset changed after the caller read it",
                    changeset_id=changeset_id,
                    expected=expected_changeset_hash,
                    actual=actual_hash,
                )
            if document["creator"] != writer.writer_id:
                raise DocForgeError(
                    "changeset_owner_conflict",
                    "Proposal writer does not own this changeset",
                    changeset_id=changeset_id,
                    owner=document["creator"],
                    writer=writer.writer_id,
                )
            snapshot = self.project.load()
            self._require_mutable(document, snapshot)
            if (
                len(document["operations"])
                >= self.project.descriptor.limits.max_changeset_operations
            ):
                raise DocForgeError(
                    "changeset_operation_limit", "Changeset operation limit has been reached"
                )
            normalized = normalize_operation(operation, sequence=len(document["operations"]) + 1)
            if any(item["node_id"] == normalized["node_id"] for item in document["operations"]):
                raise DocForgeError(
                    "duplicate_operation",
                    "A changeset may touch a node only once",
                    node_id=normalized["node_id"],
                )
            candidate = {**document, "operations": [*document["operations"], normalized]}
            self._check_base(candidate, snapshot)
            nodes, edges = self.projector.project(snapshot, candidate)
            self._check_proposal_conflicts(candidate, snapshot)
            self._write(path, candidate)
            return self._result(
                snapshot,
                candidate,
                valid=True,
                projected_node_count=len(nodes),
                projected_edge_count=len(edges),
            )

    @staticmethod
    def _complete_operation(
        operation: dict[str, Any],
        nodes: dict[str, Node],
    ) -> dict[str, Any]:
        allowed = {
            "operation",
            "node_id",
            "expected_content_hash",
            "target_source",
            "metadata",
            "content",
            "relationship_changes",
            "rationale",
        }
        unknown = sorted(set(operation) - allowed)
        if unknown:
            raise DocForgeError(
                "invalid_operation",
                "Operation has unknown fields",
                fields=unknown,
            )
        kind = operation.get("operation")
        node_id = operation.get("node_id")
        expected = operation.get("expected_content_hash")
        if kind != "create" and expected is None and isinstance(node_id, str):
            node = nodes.get(node_id)
            if node is None:
                raise DocForgeError(
                    "missing_node",
                    "No node has the requested stable ID",
                    node_id=node_id,
                )
            expected = node.content_hash
        return {
            "operation": kind,
            "node_id": node_id,
            "expected_content_hash": expected,
            "target_source": operation.get("target_source"),
            "metadata": operation.get("metadata"),
            "content": operation.get("content"),
            "relationship_changes": operation.get("relationship_changes", []),
            "rationale": operation.get("rationale"),
        }

    def _validate_locked(
        self, changeset_id: str
    ) -> tuple[ProjectSnapshot, dict[str, Any], dict[str, Node], set[tuple[str, str, str]]]:
        document = self._read(self._path(changeset_id))
        snapshot = self.project.load()
        self._check_base(document, snapshot)
        nodes, edges = self.projector.project(snapshot, document)
        self._check_proposal_conflicts(document, snapshot)
        return snapshot, document, nodes, edges

    def _check_base(self, document: dict[str, Any], snapshot: ProjectSnapshot) -> None:
        if self._base_state(document, snapshot) != "current":
            raise DocForgeError(
                "base_conflict",
                "Canonical project changed after the changeset was created",
                expected_revision=document["base_revision"],
                actual_revision=snapshot.revision,
                expected_source_hash=document["base_source_hash"],
                actual_source_hash=snapshot.source_hash,
            )

    @staticmethod
    def _base_state(document: dict[str, Any], snapshot: ProjectSnapshot) -> str:
        if (
            document["base_revision"] == snapshot.revision
            and document["base_source_hash"] == snapshot.source_hash
        ):
            return "current"
        return "stale"

    def _lifecycle(
        self,
        document: dict[str, Any],
        base_state: str,
    ) -> dict[str, object]:
        state_path = self._state_root() / f"{document['changeset_id']}.json"
        if state_path.is_file() and not state_path.is_symlink():
            try:
                parsed: object = json.loads(state_path.read_text(encoding="utf-8"))
            except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error:
                raise DocForgeError(
                    "invalid_changeset_state",
                    "Changeset lifecycle record is unreadable",
                    changeset_id=document["changeset_id"],
                ) from error
            if not isinstance(parsed, dict):
                raise DocForgeError(
                    "invalid_changeset_state",
                    "Changeset lifecycle record does not match its proposal",
                    changeset_id=document["changeset_id"],
                )
            payload = cast(dict[str, object], parsed)
            if payload.get("changeset_hash") != document_hash(document) or payload.get(
                "status"
            ) not in {"applied", "abandoned"}:
                raise DocForgeError(
                    "invalid_changeset_state",
                    "Changeset lifecycle record does not match its proposal",
                    changeset_id=document["changeset_id"],
                )
            return payload
        if base_state == "stale":
            return {"status": "stale"}
        if not document["operations"]:
            return {"status": "draft"}
        return {"status": "ready"}

    def _require_mutable(
        self,
        document: dict[str, Any],
        snapshot: ProjectSnapshot,
    ) -> None:
        lifecycle = self._lifecycle(document, self._base_state(document, snapshot))
        if lifecycle["status"] in {"applied", "abandoned"}:
            raise DocForgeError(
                "changeset_closed",
                "Applied or abandoned changesets cannot be modified",
                changeset_id=document["changeset_id"],
                lifecycle=lifecycle["status"],
            )

    def _check_proposal_conflicts(
        self, document: dict[str, Any], snapshot: ProjectSnapshot
    ) -> None:
        nodes, sources = self.projector.touches(document, snapshot)
        conflicts: list[dict[str, object]] = []
        for path in sorted(self._root().glob("*.json"), key=lambda item: item.name):
            if path.name == f"{document['changeset_id']}.json":
                continue
            other = self._read(path)
            if other["base_source_hash"] != document["base_source_hash"]:
                continue
            other_lifecycle = self._lifecycle(
                other,
                self._base_state(other, snapshot),
            )
            if other_lifecycle["status"] in {"applied", "abandoned"}:
                continue
            other_nodes, other_sources = self.projector.touches(other, snapshot)
            shared_nodes = sorted(nodes & other_nodes)
            shared_sources = sorted(sources & other_sources)
            if shared_nodes or shared_sources:
                conflicts.append(
                    {
                        "changeset_id": other["changeset_id"],
                        "nodes": shared_nodes,
                        "sources": shared_sources,
                    }
                )
        if conflicts:
            raise DocForgeError(
                "proposal_conflict",
                "Changeset overlaps another proposal from the same canonical base",
                conflicts=conflicts,
            )

    def _result(
        self,
        snapshot: ProjectSnapshot,
        document: dict[str, Any],
        **payload: object,
    ) -> dict[str, object]:
        return self._base_result(
            snapshot,
            changeset_id=document["changeset_id"],
            changeset_hash=document_hash(document),
            creator=document["creator"],
            base_revision=document["base_revision"],
            base_source_hash=document["base_source_hash"],
            operation_count=len(document["operations"]),
            operations=document["operations"],
            **payload,
        )

    @staticmethod
    def _base_result(snapshot: ProjectSnapshot, **payload: object) -> dict[str, object]:
        return {
            "status": "ok",
            "project_id": snapshot.descriptor.project_id,
            "project_root_fingerprint": project_root_fingerprint(snapshot.descriptor.root),
            "adapter": snapshot.descriptor.adapter,
            "revision": snapshot.revision,
            "source_hash": snapshot.source_hash,
            **payload,
        }

    def _require_writer(self) -> ProposalWriter:
        if self.writer is None:
            raise DocForgeError(
                "proposal_access_disabled",
                "Server has no configured proposal writer identity",
            )
        return self.writer

    def _read(self, path: Path) -> dict[str, Any]:
        if path.is_symlink():
            raise DocForgeError("path_escape", "Changeset files may not be symbolic links")
        if not path.is_file():
            raise DocForgeError("missing_changeset", "Changeset does not exist", path=path.name)
        if path.stat().st_size > self.project.descriptor.limits.max_changeset_bytes:
            raise DocForgeError("changeset_too_large", "Changeset exceeds configured size limit")
        raw = path.read_bytes()
        if len(raw) > self.project.descriptor.limits.max_changeset_bytes:
            raise DocForgeError("changeset_too_large", "Changeset exceeds configured size limit")
        try:
            document = json.loads(raw)
        except (UnicodeDecodeError, json.JSONDecodeError) as error:
            raise DocForgeError("invalid_changeset", "Changeset is not valid UTF-8 JSON") from error
        return validate_document(self.project, document, path=path)

    def _write(self, path: Path, document: dict[str, Any]) -> None:
        raw = json.dumps(document, sort_keys=True, indent=2).encode("utf-8") + b"\n"
        if len(raw) > self.project.descriptor.limits.max_changeset_bytes:
            raise DocForgeError("changeset_too_large", "Changeset exceeds configured size limit")
        root = self._root()
        descriptor = self.project.descriptor
        previous = path.read_bytes() if path.is_file() and not path.is_symlink() else None
        if path.is_symlink():
            raise DocForgeError("path_escape", "Changeset files may not be symbolic links")
        source_paths = self.project.canonical_source_paths()
        canonical_before = {
            source: source.read_bytes()
            for source in (
                descriptor.descriptor_path,
                *descriptor.authority_files,
                *source_paths,
            )
        }
        file_descriptor, temporary_name = tempfile.mkstemp(prefix=".pending-", dir=root)
        temporary = Path(temporary_name)
        try:
            with os.fdopen(file_descriptor, "wb") as handle:
                handle.write(raw)
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(temporary, path)
            directory_descriptor = os.open(root, os.O_RDONLY)
            try:
                os.fsync(directory_descriptor)
            finally:
                os.close(directory_descriptor)
        except Exception:
            temporary.unlink(missing_ok=True)
            raise
        changed_source: str | None = None
        if self.project.canonical_source_paths() != source_paths:
            changed_source = "canonical_source_set"
        else:
            for source, before in canonical_before.items():
                if not source.is_file() or source.read_bytes() != before:
                    changed_source = source.relative_to(descriptor.root).as_posix()
                    break
        if changed_source is not None:
            self._restore(path, previous, root)
            raise DocForgeError(
                "canonical_write_detected",
                "Canonical source changed during changeset storage",
                source=changed_source,
            )

    @staticmethod
    def _restore(path: Path, previous: bytes | None, root: Path) -> None:
        if previous is None:
            path.unlink(missing_ok=True)
            return
        restore_descriptor, restore_name = tempfile.mkstemp(prefix=".rollback-", dir=root)
        restore = Path(restore_name)
        try:
            with os.fdopen(restore_descriptor, "wb") as handle:
                handle.write(previous)
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(restore, path)
        except Exception:
            restore.unlink(missing_ok=True)
            raise

    def _path(self, changeset_id: str) -> Path:
        validate_id(changeset_id, "changeset_id")
        path = self._root() / f"{changeset_id}.json"
        if path.parent.resolve() != self._root():
            raise DocForgeError("path_escape", "Changeset path escaped its configured root")
        return path

    def _root(self) -> Path:
        root = self.project.descriptor.changeset_root
        root.mkdir(parents=True, exist_ok=True)
        if not root.is_dir() or root.resolve() != root:
            raise DocForgeError("path_escape", "Changeset root changed or resolves unexpectedly")
        return root

    def _state_root(self) -> Path:
        root = self._root() / ".state"
        root.mkdir(parents=True, exist_ok=True)
        if (
            not root.is_dir()
            or root.is_symlink()
            or not root.resolve().is_relative_to(self._root())
        ):
            raise DocForgeError("path_escape", "Changeset state root is not safe")
        return root

    def _write_state(
        self,
        changeset_id: str,
        payload: dict[str, object],
    ) -> dict[str, object]:
        root = self._state_root()
        path = root / f"{changeset_id}.json"
        raw = json.dumps(payload, sort_keys=True, indent=2).encode("utf-8") + b"\n"
        descriptor, temporary_name = tempfile.mkstemp(prefix=".state-", dir=root)
        temporary = Path(temporary_name)
        try:
            with os.fdopen(descriptor, "wb") as handle:
                handle.write(raw)
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(temporary, path)
            directory_descriptor = os.open(root, os.O_RDONLY)
            try:
                os.fsync(directory_descriptor)
            finally:
                os.close(directory_descriptor)
        except Exception:
            temporary.unlink(missing_ok=True)
            raise
        return payload

    @contextmanager
    def _lock(self) -> Generator[None]:
        root = self._root()
        lock_path = root / ".lock"
        try:
            descriptor = os.open(
                lock_path,
                os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW,
                0o600,
            )
        except OSError as error:
            raise DocForgeError("path_escape", "Changeset lock path is not safe") from error
        with os.fdopen(descriptor, "a+b") as handle:
            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
            try:
                yield
            finally:
                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)