DocForge2/src/docforge/rendering.py

"""Confined preview and derived-output orchestration for declared render views."""

from __future__ import annotations

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

from .changesets import ChangesetStore
from .errors import DocForgeError
from .models import (
    GenerationRecordingProject,
    IncrementalStateProject,
    ProjectDescriptor,
    ProjectService,
    ProjectSnapshot,
    ProjectState,
    RenderConfig,
    RenderView,
)
from .project import project_root_fingerprint
from .render_contract import PreparedRender, relative_output, renderer_for

RENDER_RECEIPT_SCHEMA_VERSION = 1
MAX_RENDER_RECEIPT_BYTES = 64_000


class RenderService:
    """Render only declared views through fixed built-in renderer implementations."""

    def __init__(self, project: ProjectService, changesets: ChangesetStore | None = None) -> None:
        self.project = project
        self.changesets = changesets or ChangesetStore(project)

    def status(self, view_id: str | None = None) -> dict[str, object]:
        """Report publication state from bounded receipts without rendering canonical content."""

        descriptor = self.project.descriptor
        config = descriptor.render
        current_state = self._current_state()
        if config is None:
            return self._status_result(
                descriptor,
                current_state,
                configured=False,
                state="not_configured",
                verification="receipt",
                outputs=[],
            )
        views = self._views(config, view_id)
        first_outputs = [self._receipt_status(descriptor, view, current_state) for view in views]
        outputs = [self._receipt_status(descriptor, view, current_state) for view in views]
        if outputs != first_outputs:
            for output in outputs:
                if output["state"] == "current":
                    output["state"] = "stale"
                    output["reason"] = "publication_changed_during_status"
        final_state = self._current_state()
        if final_state != current_state:
            for output in outputs:
                if output["state"] == "current":
                    output["state"] = "stale"
                    output["reason"] = "source_changed_during_status"
        identity = final_state if final_state is not None else current_state
        return self._status_result(
            descriptor,
            identity,
            configured=True,
            state="current" if all(item["state"] == "current" for item in outputs) else "stale",
            verification="receipt",
            outputs=outputs,
        )

    def deep_status(self, view_id: str | None = None) -> dict[str, object]:
        """Recompute render output as the explicit side-effect-free equivalence oracle."""

        snapshot = self.project.load()
        config = snapshot.descriptor.render
        if config is None:
            return self._result(
                snapshot,
                configured=False,
                state="not_configured",
                verification="deep",
                outputs=[],
            )
        views = self._views(config, view_id)
        outputs: list[dict[str, object]] = []
        for view in views:
            template_before = self._safe_file_identity(
                snapshot.descriptor.root,
                view.template_path,
            )
            output_before = self._safe_file_identity(
                snapshot.descriptor.root,
                view.output_path,
            )
            prepared, _ = self._prepare(snapshot, view, changeset_hash=None)
            state = "missing"
            actual_hash: str | None = None
            output = view.output_path
            if output.is_symlink() or output.resolve(strict=False) != output:
                state = "unsafe"
            elif output.is_file():
                if output.stat().st_size > snapshot.descriptor.limits.max_render_bytes:
                    state = "oversized"
                else:
                    raw = output.read_bytes()
                    actual_hash = hashlib.sha256(raw).hexdigest()
                    state = "current" if actual_hash == prepared.output_hash else "stale"
            result = self._view_result(
                snapshot,
                view,
                prepared,
                state=state,
                actual_hash=actual_hash,
            )
            if template_before != self._safe_file_identity(
                snapshot.descriptor.root, view.template_path
            ) or output_before != self._safe_file_identity(
                snapshot.descriptor.root, view.output_path
            ):
                result["state"] = "stale"
                result["reason"] = "publication_changed_during_deep_status"
            outputs.append(result)
        current = self.project.load()
        if current.source_hash != snapshot.source_hash or current.revision != snapshot.revision:
            for output in outputs:
                output["state"] = "stale"
                output["reason"] = "source_changed_during_deep_status"
        return self._result(
            snapshot,
            configured=True,
            state="current" if all(item["state"] == "current" for item in outputs) else "stale",
            verification="deep",
            outputs=outputs,
        )

    def render(self, view_id: str) -> dict[str, object]:
        with self._lock():
            snapshot = self.project.load()
            config = self._config(snapshot)
            view = self._views(config, view_id)[0]
            prepared, template_bytes = self._prepare(snapshot, view, changeset_hash=None)
            self._atomic_write(
                view.output_path,
                prepared.output,
                verify=lambda: self._verify_canonical(snapshot, view, template_bytes),
            )
            receipt: dict[str, object]
            state = "current"
            try:
                if isinstance(self.project, GenerationRecordingProject):
                    self.project.record_generation(snapshot)
                receipt = self._publish_receipt(snapshot, view, prepared)
            except (DocForgeError, OSError) as error:
                state = "degraded"
                receipt = {
                    "state": "failed",
                    "error": (
                        error.as_dict()
                        if isinstance(error, DocForgeError)
                        else {
                            "code": "render_receipt_failure",
                            "message": "Rendered output was published but its receipt failed",
                            "details": {},
                        }
                    ),
                }
            return self._result(
                snapshot,
                configured=True,
                state=state,
                publication="published",
                receipt=receipt,
                output=self._view_result(
                    snapshot,
                    view,
                    prepared,
                    state="current",
                    actual_hash=prepared.output_hash,
                ),
            )

    def _receipt_status(
        self,
        descriptor: ProjectDescriptor,
        view: RenderView,
        current_state: ProjectState | None,
    ) -> dict[str, object]:
        receipt, receipt_state = self._read_receipt(view)
        output_state = self._safe_file_identity(descriptor.root, view.output_path)
        if output_state is None:
            state = "unsafe" if view.output_path.is_symlink() else "missing"
            return self._receipt_view_result(
                descriptor,
                view,
                receipt,
                state=state,
                reason="output_not_safe" if state == "unsafe" else "output_missing",
            )
        if receipt is None:
            return self._receipt_view_result(
                descriptor,
                view,
                receipt,
                state="unverified",
                reason=receipt_state,
            )
        if not self._receipt_matches_binding(descriptor, view, receipt):
            return self._receipt_view_result(
                descriptor,
                view,
                receipt,
                state="unverified",
                reason="foreign_or_incompatible_receipt",
            )
        template_state = self._safe_file_identity(descriptor.root, view.template_path)
        if template_state is None:
            return self._receipt_view_result(
                descriptor,
                view,
                receipt,
                state="unsafe",
                reason="template_not_safe",
            )
        if receipt.get("template_file") != template_state:
            return self._receipt_view_result(
                descriptor,
                view,
                receipt,
                state="stale",
                reason="template_changed",
            )
        if receipt.get("output_file") != output_state:
            return self._receipt_view_result(
                descriptor,
                view,
                receipt,
                state="stale",
                reason="output_changed",
            )
        if current_state is None:
            reason = (
                "source_generation_unavailable"
                if isinstance(self.project, GenerationRecordingProject)
                else "source_generation_unsupported"
            )
            return self._receipt_view_result(
                descriptor,
                view,
                receipt,
                state=(
                    "stale"
                    if isinstance(self.project, GenerationRecordingProject)
                    else "unverified"
                ),
                reason=reason,
            )
        if (
            receipt.get("source_hash") != current_state.source_hash
            or receipt.get("revision") != current_state.revision
        ):
            return self._receipt_view_result(
                descriptor,
                view,
                receipt,
                state="stale",
                reason="source_generation_changed",
            )
        return self._receipt_view_result(
            descriptor,
            view,
            receipt,
            state="current",
            reason=None,
        )

    def _publish_receipt(
        self,
        snapshot: ProjectSnapshot,
        view: RenderView,
        prepared: PreparedRender,
    ) -> dict[str, object]:
        source_before = self._current_state()
        if isinstance(self.project, GenerationRecordingProject) and (
            source_before is None
            or source_before.source_hash != snapshot.source_hash
            or source_before.revision != snapshot.revision
        ):
            raise DocForgeError(
                "render_receipt_failure",
                "Canonical source changed before render receipt publication",
            )
        if source_before is not None and (
            source_before.source_hash != snapshot.source_hash
            or source_before.revision != snapshot.revision
        ):
            raise DocForgeError(
                "render_receipt_failure",
                "Canonical source changed before render receipt publication",
            )
        template_file, template_hash = self._verified_file_digest(
            snapshot.descriptor.root,
            view.template_path,
            snapshot.descriptor.limits.max_template_bytes,
        )
        output_file, output_hash = self._verified_file_digest(
            snapshot.descriptor.root,
            view.output_path,
            snapshot.descriptor.limits.max_render_bytes,
        )
        if (
            template_hash != prepared.template_hash
            or output_hash != prepared.output_hash
            or output_file["size"] != len(prepared.output)
        ):
            raise DocForgeError(
                "render_receipt_failure",
                "Published render files do not match the verified render",
            )
        final_template_file, final_template_hash = self._verified_file_digest(
            snapshot.descriptor.root,
            view.template_path,
            snapshot.descriptor.limits.max_template_bytes,
        )
        final_output_file, final_output_hash = self._verified_file_digest(
            snapshot.descriptor.root,
            view.output_path,
            snapshot.descriptor.limits.max_render_bytes,
        )
        source_after = self._current_state()
        if (
            template_file != final_template_file
            or output_file != final_output_file
            or template_hash != final_template_hash
            or output_hash != final_output_hash
            or source_before != source_after
        ):
            raise DocForgeError(
                "render_receipt_failure",
                "Render publication changed while its receipt was being prepared",
            )
        payload: dict[str, object] = {
            "schema_version": RENDER_RECEIPT_SCHEMA_VERSION,
            "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,
            "view_id": view.view_id,
            "view_config_hash": self._view_config_hash(snapshot.descriptor, view),
            "renderer": prepared.renderer,
            "renderer_version": prepared.renderer_version,
            "render_identity": prepared.render_identity,
            "template_hash": prepared.template_hash,
            "output_hash": prepared.output_hash,
            "output_bytes": len(prepared.output),
            "template_file": final_template_file,
            "output_file": final_output_file,
        }
        raw = json.dumps(payload, sort_keys=True, indent=2).encode("utf-8") + b"\n"
        if len(raw) > MAX_RENDER_RECEIPT_BYTES:
            raise DocForgeError(
                "render_receipt_failure",
                "Render publication receipt exceeds its fixed size limit",
            )
        root = self._receipt_root(create=True)
        path = root / f"{view.view_id}.json"
        if path.is_symlink():
            raise DocForgeError(
                "path_escape",
                "Render publication receipt path is not safe",
            )
        descriptor, temporary_name = tempfile.mkstemp(prefix=".render-receipt-", dir=root)
        temporary = Path(temporary_name)
        try:
            with os.fdopen(descriptor, "wb") as handle:
                handle.write(raw)
                handle.flush()
                os.fsync(handle.fileno())
            last_template_file, last_template_hash = self._verified_file_digest(
                snapshot.descriptor.root,
                view.template_path,
                snapshot.descriptor.limits.max_template_bytes,
            )
            last_output_file, last_output_hash = self._verified_file_digest(
                snapshot.descriptor.root,
                view.output_path,
                snapshot.descriptor.limits.max_render_bytes,
            )
            if (
                last_template_file != final_template_file
                or last_output_file != final_output_file
                or last_template_hash != final_template_hash
                or last_output_hash != final_output_hash
                or self._current_state() != source_after
            ):
                raise DocForgeError(
                    "render_receipt_failure",
                    "Render publication changed before receipt publication",
                )
            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 {
            "state": "current",
            "schema_version": RENDER_RECEIPT_SCHEMA_VERSION,
            "path": path.relative_to(snapshot.descriptor.root).as_posix(),
        }

    def _read_receipt(
        self,
        view: RenderView,
    ) -> tuple[dict[str, object] | None, str]:
        try:
            root = self._receipt_root(create=False)
        except DocForgeError:
            return None, "receipt_root_unsafe"
        path = root / f"{view.view_id}.json"
        if path.is_symlink():
            return None, "receipt_unsafe"
        if not path.is_file():
            return None, "receipt_missing"
        try:
            if path.stat().st_size > MAX_RENDER_RECEIPT_BYTES:
                return None, "receipt_oversized"
            raw = path.read_bytes()
            if len(raw) > MAX_RENDER_RECEIPT_BYTES:
                return None, "receipt_oversized"
            parsed: object = json.loads(raw)
        except (OSError, UnicodeDecodeError, json.JSONDecodeError):
            return None, "receipt_corrupt"
        if not isinstance(parsed, dict):
            return None, "receipt_corrupt"
        return cast(dict[str, object], parsed), "receipt"

    def _receipt_root(self, *, create: bool) -> Path:
        cache_root = self.project.descriptor.cache_root
        root = cache_root / "render-receipts"
        if (
            cache_root.resolve(strict=False) != cache_root
            or root.is_symlink()
            or root.resolve(strict=False) != root
            or not root.is_relative_to(cache_root)
        ):
            raise DocForgeError("path_escape", "Render receipt root is not safe")
        if create:
            cache_root.mkdir(parents=True, exist_ok=True)
            root.mkdir(parents=True, exist_ok=True)
        if root.exists() and not root.is_dir():
            raise DocForgeError("path_escape", "Render receipt root is not safe")
        return root

    @staticmethod
    def _safe_file_identity(root: Path, path: Path) -> dict[str, object] | None:
        if path.is_symlink() or path.resolve(strict=False) != path or not path.is_relative_to(root):
            return None
        try:
            current = path.lstat()
        except OSError:
            return None
        if not stat.S_ISREG(current.st_mode):
            return None
        return {
            "path": path.relative_to(root).as_posix(),
            "device": current.st_dev,
            "inode": current.st_ino,
            "mode": current.st_mode,
            "size": current.st_size,
            "mtime_ns": current.st_mtime_ns,
            "ctime_ns": current.st_ctime_ns,
        }

    @staticmethod
    def _verified_file_digest(
        root: Path,
        path: Path,
        maximum: int,
    ) -> tuple[dict[str, object], str]:
        if path.is_symlink() or path.resolve(strict=False) != path or not path.is_relative_to(root):
            raise DocForgeError(
                "render_receipt_failure",
                "Render publication file is not safe for verification",
            )
        try:
            descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
        except OSError as error:
            raise DocForgeError(
                "render_receipt_failure",
                "Render publication file is not readable for verification",
            ) from error
        with os.fdopen(descriptor, "rb") as handle:
            current = os.fstat(handle.fileno())
            if not stat.S_ISREG(current.st_mode) or current.st_size > maximum:
                raise DocForgeError(
                    "render_receipt_failure",
                    "Render publication file failed receipt validation",
                )
            digest = hashlib.file_digest(handle, "sha256").hexdigest()
        return (
            {
                "path": path.relative_to(root).as_posix(),
                "device": current.st_dev,
                "inode": current.st_ino,
                "mode": current.st_mode,
                "size": current.st_size,
                "mtime_ns": current.st_mtime_ns,
                "ctime_ns": current.st_ctime_ns,
            },
            digest,
        )

    @staticmethod
    def _view_config_hash(descriptor: ProjectDescriptor, view: RenderView) -> str:
        payload = {
            "view_id": view.view_id,
            "renderer": view.renderer,
            "template": view.template_path.relative_to(descriptor.root).as_posix(),
            "output": view.output_path.relative_to(descriptor.root).as_posix(),
            "title": view.title,
            "families": list(view.families),
        }
        return hashlib.sha256(
            json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
        ).hexdigest()

    def _receipt_matches_binding(
        self,
        descriptor: ProjectDescriptor,
        view: RenderView,
        receipt: dict[str, object],
    ) -> bool:
        required = {
            "schema_version",
            "project_id",
            "project_root_fingerprint",
            "adapter",
            "revision",
            "source_hash",
            "view_id",
            "view_config_hash",
            "renderer",
            "renderer_version",
            "render_identity",
            "template_hash",
            "output_hash",
            "output_bytes",
            "template_file",
            "output_file",
        }
        renderer = renderer_for(view)
        template_file = receipt.get("template_file")
        output_file = receipt.get("output_file")
        return (
            set(receipt) == required
            and receipt.get("schema_version") == RENDER_RECEIPT_SCHEMA_VERSION
            and receipt.get("project_id") == descriptor.project_id
            and receipt.get("project_root_fingerprint") == project_root_fingerprint(descriptor.root)
            and receipt.get("adapter") == descriptor.adapter
            and receipt.get("view_id") == view.view_id
            and receipt.get("view_config_hash") == self._view_config_hash(descriptor, view)
            and receipt.get("renderer") == renderer.renderer_id
            and receipt.get("renderer_version") == renderer.renderer_version
            and self._is_hash(receipt.get("source_hash"))
            and isinstance(receipt.get("revision"), str)
            and bool(receipt.get("revision"))
            and self._is_hash(receipt.get("view_config_hash"))
            and self._is_hash(receipt.get("render_identity"))
            and self._is_hash(receipt.get("template_hash"))
            and self._is_hash(receipt.get("output_hash"))
            and type(receipt.get("output_bytes")) is int
            and 0 <= cast(int, receipt["output_bytes"]) <= descriptor.limits.max_render_bytes
            and self._valid_receipt_file(
                template_file,
                view.template_path.relative_to(descriptor.root).as_posix(),
                descriptor.limits.max_template_bytes,
            )
            and self._valid_receipt_file(
                output_file,
                view.output_path.relative_to(descriptor.root).as_posix(),
                descriptor.limits.max_render_bytes,
            )
            and cast(dict[str, object], output_file)["size"] == receipt.get("output_bytes")
        )

    @staticmethod
    def _is_hash(value: object) -> bool:
        return (
            isinstance(value, str)
            and len(value) == 64
            and all(character in "0123456789abcdef" for character in value)
        )

    @staticmethod
    def _valid_receipt_file(
        value: object,
        expected_path: str,
        maximum: int,
    ) -> bool:
        if not isinstance(value, dict):
            return False
        payload = cast(dict[str, object], value)
        return (
            set(payload)
            == {
                "path",
                "device",
                "inode",
                "mode",
                "size",
                "mtime_ns",
                "ctime_ns",
            }
            and payload.get("path") == expected_path
            and all(
                type(payload.get(key)) is int and cast(int, payload[key]) >= 0
                for key in ("device", "inode", "mode", "size", "mtime_ns", "ctime_ns")
            )
            and cast(int, payload["size"]) <= maximum
        )

    @staticmethod
    def _receipt_view_result(
        descriptor: ProjectDescriptor,
        view: RenderView,
        receipt: dict[str, object] | None,
        *,
        state: str,
        reason: str | None,
    ) -> dict[str, object]:
        payload = receipt or {}
        return {
            "view_id": view.view_id,
            "renderer": payload.get("renderer", view.renderer),
            "renderer_version": payload.get("renderer_version"),
            "render_identity": payload.get("render_identity"),
            "expected_output_hash": payload.get("output_hash"),
            "actual_output_hash": (payload.get("output_hash") if state == "current" else None),
            "template_hash": payload.get("template_hash"),
            "path": view.output_path.relative_to(descriptor.root).as_posix(),
            "state": state,
            "reason": reason,
            "verification": "receipt",
            "receipt_schema_version": payload.get("schema_version"),
        }

    def _current_state(self) -> ProjectState | None:
        if isinstance(self.project, IncrementalStateProject):
            return self.project.incremental_state()
        return None

    @staticmethod
    def _status_result(
        descriptor: ProjectDescriptor,
        identity: ProjectState | None,
        **payload: object,
    ) -> dict[str, object]:
        return {
            "status": "ok",
            "project_id": descriptor.project_id,
            "project_root_fingerprint": project_root_fingerprint(descriptor.root),
            "adapter": descriptor.adapter,
            "revision": identity.revision if identity is not None else "unknown",
            "source_hash": identity.source_hash if identity is not None else None,
            **payload,
        }

    def preview(self, changeset_id: str, view_id: str) -> dict[str, object]:
        with self._lock():
            snapshot, changeset_hash = self.changesets.projected_snapshot(changeset_id)
            config = self._config(snapshot)
            view = self._views(config, view_id)[0]
            prepared, template_bytes = self._prepare(snapshot, view, changeset_hash=changeset_hash)
            preview_path = config.preview_root / changeset_id / f"{view.view_id}.html"
            if not preview_path.is_relative_to(config.preview_root):
                raise DocForgeError("path_escape", "Preview path escaped its configured root")

            def verify() -> None:
                current, current_hash = self.changesets.projected_snapshot(changeset_id)
                if current.source_hash != snapshot.source_hash or current_hash != changeset_hash:
                    raise DocForgeError(
                        "render_input_changed",
                        "Changeset or canonical input changed during preview",
                    )
                self._verify_template(view, template_bytes)

            self._atomic_write(preview_path, prepared.output, verify=verify, preview=True)
            return self._result(
                snapshot,
                configured=True,
                state="current",
                changeset_id=changeset_id,
                changeset_hash=changeset_hash,
                preview_identity=prepared.render_identity,
                preview={
                    **self._view_result(
                        snapshot,
                        view,
                        prepared,
                        state="current",
                        actual_hash=prepared.output_hash,
                    ),
                    "path": relative_output(snapshot, preview_path),
                },
            )

    def _prepare(
        self,
        snapshot: ProjectSnapshot,
        view: RenderView,
        *,
        changeset_hash: str | None,
    ) -> tuple[PreparedRender, bytes]:
        template = self._template_bytes(snapshot, view)
        prepared = renderer_for(view).prepare(
            snapshot,
            view,
            template,
            changeset_hash=changeset_hash,
        )
        if len(prepared.output) > snapshot.descriptor.limits.max_render_bytes:
            raise DocForgeError("render_too_large", "Rendered output exceeds the configured limit")
        return prepared, template

    def _template_bytes(self, snapshot: ProjectSnapshot, view: RenderView) -> bytes:
        path = view.template_path
        config = self._config(snapshot)
        if (
            path.is_symlink()
            or path.resolve(strict=False) != path
            or config.template_root.resolve(strict=False) != config.template_root
            or not path.is_file()
            or not path.is_relative_to(config.template_root)
        ):
            raise DocForgeError("unsafe_template", "Render template is missing or unsafe")
        if path.stat().st_size > snapshot.descriptor.limits.max_template_bytes:
            raise DocForgeError(
                "template_too_large", "Render template exceeds the configured limit"
            )
        raw = path.read_bytes()
        if len(raw) > snapshot.descriptor.limits.max_template_bytes:
            raise DocForgeError(
                "template_too_large", "Render template exceeds the configured limit"
            )
        return raw

    def _verify_canonical(
        self, snapshot: ProjectSnapshot, view: RenderView, template_bytes: bytes
    ) -> None:
        current = self.project.load()
        if current.source_hash != snapshot.source_hash:
            raise DocForgeError("render_input_changed", "Canonical input changed during rendering")
        self._verify_template(view, template_bytes)

    @staticmethod
    def _verify_template(view: RenderView, template_bytes: bytes) -> None:
        if (
            view.template_path.is_symlink()
            or view.template_path.resolve(strict=False) != view.template_path
            or not view.template_path.is_file()
        ):
            raise DocForgeError("render_input_changed", "Render template changed during rendering")
        if view.template_path.read_bytes() != template_bytes:
            raise DocForgeError("render_input_changed", "Render template changed during rendering")

    def _atomic_write(
        self,
        output: Path,
        content: bytes,
        *,
        verify: Callable[[], None],
        preview: bool = False,
    ) -> None:
        root = self.project.descriptor.root
        if output.is_symlink() or not output.is_relative_to(root):
            raise DocForgeError("path_escape", "Render output path is unsafe")
        parent = output.parent
        if parent.resolve(strict=False) != parent:
            raise DocForgeError("path_escape", "Render output directory is unsafe")
        parent.mkdir(parents=True, exist_ok=True)
        if parent.resolve() != parent or not parent.is_relative_to(root):
            raise DocForgeError("path_escape", "Render output directory is unsafe")
        descriptor, temporary_name = tempfile.mkstemp(prefix=".docforge-render-", dir=parent)
        temporary = Path(temporary_name)
        try:
            with os.fdopen(descriptor, "wb") as handle:
                handle.write(content)
                handle.flush()
                os.fsync(handle.fileno())
            verify()
            if output.is_symlink():
                raise DocForgeError("path_escape", "Render output became unsafe")
            os.replace(temporary, output)
        except Exception:
            temporary.unlink(missing_ok=True)
            if preview:
                self._remove_empty_preview_parents(parent)
            raise

    def _remove_empty_preview_parents(self, parent: Path) -> None:
        config = self.project.descriptor.render
        if config is None:
            return
        current = parent
        while current != config.preview_root:
            try:
                current.rmdir()
            except OSError:
                return
            current = current.parent

    def _view_result(
        self,
        snapshot: ProjectSnapshot,
        view: RenderView,
        prepared: PreparedRender,
        *,
        state: str,
        actual_hash: str | None,
    ) -> dict[str, object]:
        return {
            "view_id": view.view_id,
            "renderer": prepared.renderer,
            "renderer_version": prepared.renderer_version,
            "render_identity": prepared.render_identity,
            "expected_output_hash": prepared.output_hash,
            "actual_output_hash": actual_hash,
            "template_hash": prepared.template_hash,
            "path": relative_output(snapshot, view.output_path),
            "state": state,
        }

    @staticmethod
    def _config(snapshot: ProjectSnapshot) -> RenderConfig:
        if snapshot.descriptor.render is None:
            raise DocForgeError("render_not_configured", "Project has no configured render views")
        return snapshot.descriptor.render

    @staticmethod
    def _views(config: RenderConfig, view_id: str | None) -> tuple[RenderView, ...]:
        if view_id is None:
            return config.views
        views = tuple(view for view in config.views if view.view_id == view_id)
        if not views:
            raise DocForgeError(
                "unknown_render_view", "Render view is not declared by this project", view=view_id
            )
        return views

    @staticmethod
    def _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,
        }

    @contextmanager
    def _lock(self) -> Generator[None]:
        root = self.project.descriptor.cache_root
        if root.resolve(strict=False) != root:
            raise DocForgeError("path_escape", "Render lock directory is not safe")
        root.mkdir(parents=True, exist_ok=True)
        if not root.is_dir() or root.resolve(strict=False) != root:
            raise DocForgeError("path_escape", "Render lock directory is not safe")
        lock_path = root / "render.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", "Render 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)