DocForge2/src/docforge/index.py

"""Atomic derived SQLite index and deterministic read operations."""

from __future__ import annotations

import fcntl
import hashlib
import json
import os
import sqlite3
import tempfile
import time
from collections import deque
from collections.abc import Callable, Generator
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import cast

from .errors import DocForgeError
from .models import (
    BuildReportingProject,
    Edge,
    GenerationRecordingProject,
    IncrementalStateProject,
    LogicEdge,
    LogicNode,
    LogicProject,
    LogicProjection,
    Node,
    ProjectDescriptor,
    ProjectService,
    ProjectSnapshot,
    ProjectState,
)
from .project import project_root_fingerprint

INDEX_SCHEMA_VERSION = 3
APPLICATION_ID = 1_146_683_778


def _node_hash(nodes: tuple[Node, ...]) -> str:
    payload = json.dumps(
        [node.as_dict() for node in nodes], sort_keys=True, separators=(",", ":")
    ).encode()
    return hashlib.sha256(payload).hexdigest()


def _edge_hash(edges: tuple[Edge, ...]) -> str:
    payload = json.dumps(
        [edge.as_dict() for edge in edges], sort_keys=True, separators=(",", ":")
    ).encode()
    return hashlib.sha256(payload).hexdigest()


def _logic_hash(projections: tuple[LogicProjection, ...]) -> str:
    payload = json.dumps(
        [projection.as_dict() for projection in projections],
        sort_keys=True,
        separators=(",", ":"),
    ).encode()
    return hashlib.sha256(payload).hexdigest()


def _connect_read_only(path: Path) -> sqlite3.Connection:
    if not path.is_file():
        raise DocForgeError("missing_index", "Derived index does not exist; run build first")
    connection = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
    connection.row_factory = sqlite3.Row
    return connection


@contextmanager
def _read_connection(path: Path) -> Generator[sqlite3.Connection, None, None]:
    connection: sqlite3.Connection | None = None
    try:
        connection = _connect_read_only(path)
        yield connection
    except DocForgeError:
        raise
    except (sqlite3.Error, json.JSONDecodeError, KeyError, TypeError) as error:
        raise DocForgeError("invalid_index", "Derived index is corrupt or unreadable") from error
    finally:
        if connection is not None:
            connection.close()


def _status(
    snapshot: ProjectSnapshot,
    logic: tuple[LogicProjection, ...],
) -> dict[str, object]:
    return {
        "project_id": snapshot.descriptor.project_id,
        "project_root_fingerprint": project_root_fingerprint(snapshot.descriptor.root),
        "revision": snapshot.revision,
        "source_hash": snapshot.source_hash,
        "node_hash": _node_hash(snapshot.nodes),
        "node_count": len(snapshot.nodes),
        "edge_hash": _edge_hash(snapshot.edges),
        "edge_count": len(snapshot.edges),
        "logic_hash": _logic_hash(logic),
        "logic_projection_count": len(logic),
        "logic_node_count": sum(len(projection.nodes) for projection in logic),
        "logic_edge_count": sum(len(projection.edges) for projection in logic),
        "index_schema_version": INDEX_SCHEMA_VERSION,
        "adapter": snapshot.descriptor.adapter,
        "status": "ok",
    }


@dataclass(frozen=True)
class _IndexReadSnapshot:
    """One request-scoped read transaction over a verified immutable generation."""

    connection: sqlite3.Connection
    checked: dict[str, object]

    def project_snapshot(self, descriptor: ProjectDescriptor) -> ProjectSnapshot:
        nodes = tuple(
            _row_to_node(row)
            for row in self.connection.execute("SELECT * FROM nodes ORDER BY node_id")
        )
        edges = tuple(
            Edge(*row)
            for row in self.connection.execute(
                "SELECT source_id, relation, target_id FROM edges "
                "ORDER BY source_id, relation, target_id"
            )
        )
        return ProjectSnapshot(
            descriptor=descriptor,
            nodes=nodes,
            edges=edges,
            source_hash=cast(str, self.checked["source_hash"]),
            revision=cast(str, self.checked["revision"]),
        )

    def result(self, **payload: object) -> dict[str, object]:
        return {
            "status": "ok",
            "project_id": self.checked["project_id"],
            "project_root_fingerprint": self.checked["project_root_fingerprint"],
            "revision": self.checked["revision"],
            "source_hash": self.checked["source_hash"],
            "adapter": self.checked["adapter"],
            **payload,
        }


class ProjectIndex:
    """A disposable index that always checks current canonical source before queries."""

    def __init__(self, project: ProjectService, *, allow_logic: bool = True) -> None:
        self.project = project
        self.allow_logic = allow_logic
        self._verified_index_signature: tuple[int, int, int, int, int] | None = None

    @property
    def path(self) -> Path:
        return self.project.descriptor.index_path

    @property
    def attestation_path(self) -> Path:
        """Return the project-confined receipt for one fully verified index file."""

        return self.path.with_suffix(f"{self.path.suffix}.attestation.json")

    def build(self) -> dict[str, object]:
        """Build one complete index while excluding concurrent publishers."""

        with self._build_lock():
            return self._build_locked()

    def synchronize(self) -> dict[str, object]:
        """Return a current index, rebuilding disposable state when necessary."""

        started = time.perf_counter()
        try:
            checked = self.check(verify_rows=False)
        except DocForgeError as error:
            if error.code not in {"missing_index", "stale_index", "invalid_index"}:
                raise
            initial_error: dict[str, object] | None = error.as_dict()
        else:
            temporary_indexes = tuple(self.project.descriptor.cache_root.glob("index-*.sqlite3"))
            if temporary_indexes:
                with self._build_lock():
                    removed = self._remove_temporary_indexes()
            else:
                removed = []
            return {
                **checked,
                "synchronization": {
                    "action": "current",
                    "elapsed_seconds": round(time.perf_counter() - started, 6),
                    "initial_error": None,
                    "removed_temporary_indexes": removed,
                },
            }

        with self._build_lock():
            try:
                checked = self.check(verify_rows=False)
            except DocForgeError as error:
                if error.code not in {"missing_index", "stale_index", "invalid_index"}:
                    raise
                removed = self._remove_temporary_indexes()
                built = self._build_locked()
                checked = self.check(verify_rows=False)
                action = "rebuilt"
                build = built.get("build")
            else:
                removed = self._remove_temporary_indexes()
                action = "current_after_wait"
                build = None
        synchronization: dict[str, object] = {
            "action": action,
            "elapsed_seconds": round(time.perf_counter() - started, 6),
            "initial_error": initial_error,
            "removed_temporary_indexes": removed,
        }
        if build is not None:
            synchronization["build"] = build
        return {**checked, "synchronization": synchronization}

    def _build_locked(self) -> dict[str, object]:
        snapshot = self.project.load()
        logic = self._logic_projections()
        status = _status(snapshot, logic)
        build_report = (
            self.project.build_report() if isinstance(self.project, BuildReportingProject) else None
        )
        if build_report is not None and build_report.get("mode") != "incremental":
            build_report = None
        cache_root = snapshot.descriptor.cache_root
        cache_root.mkdir(parents=True, exist_ok=True)
        with tempfile.NamedTemporaryFile(
            prefix="index-", suffix=".sqlite3", dir=cache_root, delete=False
        ) as descriptor:
            temporary = Path(descriptor.name)
        try:
            connection = sqlite3.connect(temporary)
            try:
                connection.execute(f"PRAGMA application_id={APPLICATION_ID}")
                connection.execute(f"PRAGMA user_version={INDEX_SCHEMA_VERSION}")
                connection.executescript(
                    """
                    CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL);
                    CREATE TABLE nodes (
                        node_id TEXT PRIMARY KEY,
                        title TEXT NOT NULL,
                        family TEXT NOT NULL,
                        authority TEXT NOT NULL,
                        status TEXT NOT NULL,
                        tags_json TEXT NOT NULL,
                        summary TEXT NOT NULL,
                        content TEXT NOT NULL,
                        source_path TEXT NOT NULL,
                        source_anchor TEXT,
                        content_hash TEXT NOT NULL
                    );
                    CREATE TABLE edges (
                        source_id TEXT NOT NULL,
                        relation TEXT NOT NULL,
                        target_id TEXT NOT NULL,
                        PRIMARY KEY (source_id, relation, target_id)
                    );
                    CREATE INDEX edges_target ON edges(target_id, relation, source_id);
                    CREATE INDEX edges_target_source
                        ON edges(target_id, source_id, relation);
                    CREATE TABLE logic_owners (
                        owner_node_id TEXT PRIMARY KEY,
                        source_id TEXT NOT NULL
                    );
                    CREATE TABLE logic_nodes (
                        owner_node_id TEXT NOT NULL,
                        logic_id TEXT NOT NULL,
                        kind TEXT NOT NULL,
                        label TEXT NOT NULL,
                        source_anchor TEXT,
                        PRIMARY KEY (owner_node_id, logic_id)
                    );
                    CREATE INDEX logic_nodes_id ON logic_nodes(logic_id, owner_node_id);
                    CREATE TABLE logic_edges (
                        owner_node_id TEXT NOT NULL,
                        source_id TEXT NOT NULL,
                        relation TEXT NOT NULL,
                        target_id TEXT NOT NULL,
                        label TEXT,
                        ordinal INTEGER NOT NULL,
                        PRIMARY KEY (
                            owner_node_id, source_id, ordinal, relation, target_id
                        )
                    );
                    CREATE INDEX logic_edges_target
                        ON logic_edges(owner_node_id, target_id, source_id);
                    CREATE VIRTUAL TABLE node_fts USING fts5(
                        node_id UNINDEXED, title, summary, content, tags
                    );
                    """
                )
                connection.executemany(
                    "INSERT INTO metadata(key, value) VALUES (?, ?)",
                    sorted((key, str(value)) for key, value in status.items()),
                )
                connection.executemany(
                    """
                    INSERT INTO nodes VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    [
                        (
                            node.node_id,
                            node.title,
                            node.family,
                            node.authority,
                            node.status,
                            json.dumps(node.tags, separators=(",", ":")),
                            node.summary,
                            node.content,
                            node.source_path,
                            node.source_anchor,
                            node.content_hash,
                        )
                        for node in snapshot.nodes
                    ],
                )
                connection.executemany(
                    "INSERT INTO edges VALUES (?, ?, ?)",
                    [(edge.source_id, edge.relation, edge.target_id) for edge in snapshot.edges],
                )
                connection.executemany(
                    "INSERT INTO logic_owners VALUES (?, ?)",
                    [(projection.owner_node_id, projection.source_id) for projection in logic],
                )
                connection.executemany(
                    "INSERT INTO logic_nodes VALUES (?, ?, ?, ?, ?)",
                    [
                        (
                            projection.owner_node_id,
                            node.logic_id,
                            node.kind,
                            node.label,
                            node.source_anchor,
                        )
                        for projection in logic
                        for node in projection.nodes
                    ],
                )
                connection.executemany(
                    "INSERT INTO logic_edges VALUES (?, ?, ?, ?, ?, ?)",
                    [
                        (
                            projection.owner_node_id,
                            edge.source_id,
                            edge.relation,
                            edge.target_id,
                            edge.label,
                            edge.ordinal,
                        )
                        for projection in logic
                        for edge in projection.edges
                    ],
                )
                connection.executemany(
                    "INSERT INTO node_fts VALUES (?, ?, ?, ?, ?)",
                    [
                        (
                            node.node_id,
                            node.title,
                            node.summary,
                            node.content,
                            " ".join(node.tags),
                        )
                        for node in snapshot.nodes
                    ],
                )
                connection.commit()
                integrity = connection.execute("PRAGMA integrity_check").fetchone()
                if integrity is None or integrity[0] != "ok":
                    raise DocForgeError("index_failure", "New index failed SQLite integrity check")
            finally:
                connection.close()
            current = self.project.load()
            current_logic = self._logic_projections()
            if (
                current.source_hash != snapshot.source_hash
                or current.revision != snapshot.revision
                or current_logic != logic
            ):
                raise DocForgeError("source_changed", "Canonical source changed during index build")
            os.replace(temporary, self.path)
            self._verified_index_signature = self._index_signature()
            self._write_attestation()
            if isinstance(self.project, GenerationRecordingProject):
                self.project.record_generation(current)
        except sqlite3.Error as error:
            temporary.unlink(missing_ok=True)
            raise DocForgeError("index_failure", "Could not build the derived index") from error
        except Exception:
            temporary.unlink(missing_ok=True)
            raise
        result: dict[str, object] = {**status, "database": str(self.path)}
        if build_report is not None:
            result["build"] = build_report
        return result

    @contextmanager
    def _build_lock(self) -> Generator[None, None, None]:
        cache_root = self.project.descriptor.cache_root
        cache_root.mkdir(parents=True, exist_ok=True)
        lock_path = cache_root / ".index.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", "Index 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)

    def _remove_temporary_indexes(self) -> list[str]:
        removed: list[str] = []
        candidates = (
            *self.project.descriptor.cache_root.glob("index-*.sqlite3"),
            *self.project.descriptor.cache_root.glob(".index-attestation-*"),
        )
        for path in sorted(candidates):
            if path == self.path or path.is_symlink() or not path.is_file():
                continue
            path.unlink()
            removed.append(path.name)
        return removed

    def _logic_projections(self) -> tuple[LogicProjection, ...]:
        if isinstance(self.project, LogicProject):
            projections = self.project.logic_projections()
            self._require_logic_allowed(len(projections))
            return projections
        return ()

    def _require_logic_allowed(self, projection_count: int) -> None:
        if projection_count and not self.allow_logic:
            raise DocForgeError(
                "adapter_policy_forbids_logic",
                (
                    "This index preserves a no-AST adapter and refuses function-Logic "
                    "publication or retrieval"
                ),
                logic_projection_count=projection_count,
            )

    @contextmanager
    def _read_snapshot(self) -> Generator[_IndexReadSnapshot, None, None]:
        """Pin one verified index and source generation for a complete read request."""

        checked = self.check(verify_rows=False)
        signature = self._verified_index_signature
        if signature is None or self._index_signature() != signature:
            raise DocForgeError(
                "invalid_index",
                "Derived index changed after validation",
            )
        with _read_connection(self.path) as connection:
            connection.execute("PRAGMA query_only=ON")
            connection.execute("BEGIN")
            application_id = connection.execute("PRAGMA application_id").fetchone()[0]
            schema_version = connection.execute("PRAGMA user_version").fetchone()[0]
            if application_id != APPLICATION_ID or schema_version != INDEX_SCHEMA_VERSION:
                raise DocForgeError("invalid_index", "Derived index has an unsupported schema")
            metadata = dict(connection.execute("SELECT key, value FROM metadata"))
            for key in (
                "project_id",
                "project_root_fingerprint",
                "revision",
                "source_hash",
                "index_schema_version",
                "adapter",
            ):
                if metadata.get(key) != str(checked[key]):
                    raise DocForgeError(
                        "invalid_index",
                        "Derived index changed after validation",
                        field=key,
                    )
            try:
                logic_projection_count = int(metadata["logic_projection_count"])
            except (KeyError, ValueError) as error:
                raise DocForgeError(
                    "invalid_index",
                    "Derived index has invalid Logic metadata",
                ) from error
            self._require_logic_allowed(logic_projection_count)
            snapshot = _IndexReadSnapshot(connection=connection, checked=checked)
            try:
                yield snapshot
            except Exception:
                raise
            else:
                self._confirm_read(snapshot.checked, signature)

    def _confirm_read(
        self,
        checked: dict[str, object],
        signature: tuple[int, int, int, int, int],
    ) -> None:
        if self._index_signature() != signature:
            raise DocForgeError("invalid_index", "Derived index changed during the query")
        state: ProjectState | None = None
        if isinstance(self.project, IncrementalStateProject):
            state = self.project.incremental_state()
        if state is None:
            current = self.project.load()
            state = ProjectState(source_hash=current.source_hash, revision=current.revision)
        if state.source_hash != checked["source_hash"] or state.revision != checked["revision"]:
            raise DocForgeError("source_changed", "Canonical source changed during the query")

    def read_project_snapshot(
        self,
        reader: Callable[[ProjectSnapshot], dict[str, object]],
    ) -> dict[str, object]:
        """Run one bounded reader against an immutable derived project snapshot."""

        with self._read_snapshot() as snapshot:
            payload = reader(snapshot.project_snapshot(self.project.descriptor))
            return snapshot.result(**payload)

    def check(self, *, verify_rows: bool = True) -> dict[str, object]:
        if isinstance(self.project, IncrementalStateProject):
            state = self.project.incremental_state()
            if state is not None:
                return self._check_incremental_state(state, verify_rows=verify_rows)
        snapshot = self.project.load()
        logic = self._logic_projections()
        expected = _status(snapshot, logic)
        with _read_connection(self.path) as connection:
            application_id = connection.execute("PRAGMA application_id").fetchone()[0]
            schema_version = connection.execute("PRAGMA user_version").fetchone()[0]
            if application_id != APPLICATION_ID or schema_version != INDEX_SCHEMA_VERSION:
                raise DocForgeError("invalid_index", "Derived index has an unsupported schema")
            metadata = dict(connection.execute("SELECT key, value FROM metadata"))
            for key in (
                "project_id",
                "project_root_fingerprint",
                "revision",
                "source_hash",
                "node_hash",
                "node_count",
                "edge_hash",
                "edge_count",
                "logic_hash",
                "logic_projection_count",
                "logic_node_count",
                "logic_edge_count",
                "index_schema_version",
                "adapter",
            ):
                if metadata.get(key) != str(expected[key]):
                    raise DocForgeError(
                        "stale_index", "Derived index does not match canonical source", field=key
                    )
            integrity = connection.execute("PRAGMA integrity_check").fetchone()
            if integrity is None or integrity[0] != "ok":
                raise DocForgeError("invalid_index", "Derived index failed SQLite integrity check")
            indexed_nodes = tuple(
                _row_to_node(row)
                for row in connection.execute("SELECT * FROM nodes ORDER BY node_id")
            )
            indexed_edges = tuple(
                Edge(*row)
                for row in connection.execute(
                    "SELECT source_id, relation, target_id FROM edges "
                    "ORDER BY source_id, relation, target_id"
                )
            )
            indexed_logic = _logic_from_connection(connection)
            fts_count = connection.execute("SELECT COUNT(*) FROM node_fts").fetchone()[0]
            if (
                indexed_nodes != snapshot.nodes
                or indexed_edges != snapshot.edges
                or indexed_logic != logic
                or fts_count != len(snapshot.nodes)
            ):
                raise DocForgeError("invalid_index", "Derived index rows do not match source")
        self._verified_index_signature = self._index_signature()
        if isinstance(self.project, GenerationRecordingProject):
            self.project.record_generation(snapshot)
        return {**expected, "database": str(self.path)}

    def _check_incremental_state(
        self,
        state: ProjectState,
        *,
        verify_rows: bool,
    ) -> dict[str, object]:
        """Validate a published index against cheap current source identity."""

        descriptor = self.project.descriptor
        identity = {
            "project_id": descriptor.project_id,
            "project_root_fingerprint": project_root_fingerprint(descriptor.root),
            "revision": state.revision,
            "source_hash": state.source_hash,
            "index_schema_version": INDEX_SCHEMA_VERSION,
            "adapter": descriptor.adapter,
        }
        with _read_connection(self.path) as connection:
            application_id = connection.execute("PRAGMA application_id").fetchone()[0]
            schema_version = connection.execute("PRAGMA user_version").fetchone()[0]
            if application_id != APPLICATION_ID or schema_version != INDEX_SCHEMA_VERSION:
                raise DocForgeError("invalid_index", "Derived index has an unsupported schema")
            metadata = dict(connection.execute("SELECT key, value FROM metadata"))
            for key, expected in identity.items():
                if metadata.get(key) != str(expected):
                    raise DocForgeError(
                        "stale_index", "Derived index does not match canonical source", field=key
                    )
            try:
                logic_projection_count = int(metadata["logic_projection_count"])
            except (KeyError, ValueError) as error:
                raise DocForgeError(
                    "invalid_index",
                    "Derived index has invalid Logic metadata",
                ) from error
            self._require_logic_allowed(logic_projection_count)
            current_signature = self._index_signature()
            if not verify_rows and (
                current_signature == self._verified_index_signature or self._attestation_matches()
            ):
                self._verified_index_signature = current_signature
                return {
                    **identity,
                    "node_hash": metadata["node_hash"],
                    "node_count": int(metadata["node_count"]),
                    "edge_hash": metadata["edge_hash"],
                    "edge_count": int(metadata["edge_count"]),
                    "logic_hash": metadata["logic_hash"],
                    "logic_projection_count": logic_projection_count,
                    "logic_node_count": int(metadata["logic_node_count"]),
                    "logic_edge_count": int(metadata["logic_edge_count"]),
                    "status": "ok",
                    "database": str(self.path),
                }
            integrity = connection.execute("PRAGMA integrity_check").fetchone()
            if integrity is None or integrity[0] != "ok":
                raise DocForgeError("invalid_index", "Derived index failed SQLite integrity check")
            indexed_nodes = tuple(
                _row_to_node(row)
                for row in connection.execute("SELECT * FROM nodes ORDER BY node_id")
            )
            indexed_edges = tuple(
                Edge(*row)
                for row in connection.execute(
                    "SELECT source_id, relation, target_id FROM edges "
                    "ORDER BY source_id, relation, target_id"
                )
            )
            indexed_logic = _logic_from_connection(connection)
            node_hash = _node_hash(indexed_nodes)
            edge_hash = _edge_hash(indexed_edges)
            logic_hash = _logic_hash(indexed_logic)
            fts_count = connection.execute("SELECT COUNT(*) FROM node_fts").fetchone()[0]
            if (
                metadata.get("node_hash") != node_hash
                or metadata.get("edge_hash") != edge_hash
                or metadata.get("logic_hash") != logic_hash
                or metadata.get("node_count") != str(len(indexed_nodes))
                or metadata.get("edge_count") != str(len(indexed_edges))
                or metadata.get("logic_projection_count") != str(len(indexed_logic))
                or metadata.get("logic_node_count")
                != str(sum(len(projection.nodes) for projection in indexed_logic))
                or metadata.get("logic_edge_count")
                != str(sum(len(projection.edges) for projection in indexed_logic))
                or fts_count != len(indexed_nodes)
            ):
                raise DocForgeError("invalid_index", "Derived index rows do not match metadata")
        self._verified_index_signature = current_signature
        self._write_attestation()
        return {
            **identity,
            "node_hash": node_hash,
            "node_count": len(indexed_nodes),
            "edge_hash": edge_hash,
            "edge_count": len(indexed_edges),
            "logic_hash": logic_hash,
            "logic_projection_count": len(indexed_logic),
            "logic_node_count": sum(len(projection.nodes) for projection in indexed_logic),
            "logic_edge_count": sum(len(projection.edges) for projection in indexed_logic),
            "status": "ok",
            "database": str(self.path),
        }

    def _attestation_matches(self) -> bool:
        """Verify a persisted whole-file digest before trusting a warm derived index."""

        path = self.attestation_path
        if not path.is_file() or path.is_symlink():
            return False
        try:
            parsed: object = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, UnicodeDecodeError, json.JSONDecodeError):
            return False
        if not isinstance(parsed, dict):
            return False
        payload = cast(dict[str, object], parsed)
        expected_size = payload.get("index_size")
        expected_hash = payload.get("index_sha256")
        if (
            payload.get("schema_version") != 1
            or type(expected_size) is not int
            or not isinstance(expected_hash, str)
            or len(expected_hash) != 64
        ):
            return False
        try:
            if self.path.stat().st_size != expected_size:
                return False
            with self.path.open("rb") as handle:
                actual_hash = hashlib.file_digest(handle, "sha256").hexdigest()
        except OSError:
            return False
        return actual_hash == expected_hash

    def _write_attestation(self) -> None:
        """Atomically persist the digest of an index that passed complete verification."""

        root = self.project.descriptor.cache_root
        path = self.attestation_path
        if path.parent != root or path.is_symlink():
            raise DocForgeError("path_escape", "Index attestation path is not safe")
        try:
            size = self.path.stat().st_size
            with self.path.open("rb") as handle:
                index_hash = hashlib.file_digest(handle, "sha256").hexdigest()
        except OSError as error:
            raise DocForgeError("missing_index", "Derived index cannot be attested") from error
        payload = {
            "schema_version": 1,
            "index_size": size,
            "index_sha256": index_hash,
        }
        raw = json.dumps(payload, sort_keys=True, indent=2).encode("utf-8") + b"\n"
        descriptor, temporary_name = tempfile.mkstemp(prefix=".index-attestation-", 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

    def _index_signature(self) -> tuple[int, int, int, int, int]:
        try:
            status = self.path.stat()
        except OSError as error:
            raise DocForgeError(
                "missing_index",
                "Derived index does not exist; run build first",
            ) from error
        return (
            status.st_dev,
            status.st_ino,
            status.st_size,
            status.st_mtime_ns,
            status.st_ctime_ns,
        )

    def get_node(self, node_id: str) -> dict[str, object]:
        with self._read_snapshot() as snapshot:
            row = snapshot.connection.execute(
                "SELECT * FROM nodes WHERE node_id = ?",
                (node_id,),
            ).fetchone()
            if row is None:
                raise DocForgeError(
                    "missing_node", "No node has the requested stable ID", node_id=node_id
                )
            return snapshot.result(node=_row_to_node(row).as_dict())

    def get_logic(self, owner_node_id: str) -> dict[str, object]:
        """Return one function-scoped control-flow projection without expanding the graph."""

        with self._read_snapshot() as snapshot:
            owner = snapshot.connection.execute(
                "SELECT * FROM nodes WHERE node_id = ?", (owner_node_id,)
            ).fetchone()
            projection = _logic_projection_from_connection(snapshot.connection, owner_node_id)
            if owner is None:
                raise DocForgeError(
                    "missing_node",
                    "No node has the requested stable ID",
                    node_id=owner_node_id,
                )
            return snapshot.result(
                owner=_row_to_node(owner).as_dict(include_content=False),
                available=projection is not None,
                projection=projection.as_dict() if projection is not None else None,
            )

    def search(self, query: str, *, limit: int | None = None) -> dict[str, object]:
        limits = self.project.descriptor.limits
        if not query.strip() or len(query) > limits.max_query_chars:
            raise DocForgeError("invalid_query", "Search query is empty or exceeds its limit")
        bounded = _bounded_limit(limit, limits.max_results, default=20)
        terms = re_tokenize(query)
        if not terms:
            raise DocForgeError("invalid_query", "Search query contains no searchable text")
        expression = " AND ".join(f'"{term.replace(chr(34), chr(34) * 2)}"' for term in terms)
        with self._read_snapshot() as snapshot:
            rows = snapshot.connection.execute(
                """
                SELECT nodes.*, bm25(node_fts) AS rank,
                       snippet(node_fts, 3, '[', ']', ' … ', 18) AS snippet
                FROM node_fts JOIN nodes USING(node_id)
                WHERE node_fts MATCH ?
                ORDER BY rank, nodes.node_id
                LIMIT ?
                """,
                (expression, bounded + 1),
            ).fetchall()
            results: list[dict[str, object]] = []
            for row in rows[:bounded]:
                payload = _row_to_node(row).as_dict(include_content=False)
                payload.update({"rank": row["rank"], "snippet": row["snippet"]})
                results.append(payload)
            return snapshot.result(
                query=query,
                count=len(results),
                limit=bounded,
                truncated=len(rows) > bounded,
                truncation_reason="result_limit" if len(rows) > bounded else None,
                results=results,
            )

    def filter_nodes(
        self,
        *,
        family: str | None = None,
        authority: str | None = None,
        status: str | None = None,
        tag: str | None = None,
        limit: int | None = None,
    ) -> dict[str, object]:
        bounded = _bounded_limit(limit, self.project.descriptor.limits.max_results, default=100)
        clauses: list[str] = []
        values: list[object] = []
        for field, value in (("family", family), ("authority", authority), ("status", status)):
            if value is not None:
                clauses.append(f"{field} = ?")
                values.append(value)
        if tag is not None:
            clauses.append("EXISTS (SELECT 1 FROM json_each(tags_json) WHERE value = ?)")
            values.append(tag)
        where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
        with self._read_snapshot() as snapshot:
            rows = snapshot.connection.execute(
                f"SELECT * FROM nodes {where} ORDER BY node_id LIMIT ?",
                (*values, bounded + 1),
            ).fetchall()
            results = [_row_to_node(row).as_dict(include_content=False) for row in rows[:bounded]]
            return snapshot.result(
                count=len(results),
                limit=bounded,
                truncated=len(rows) > bounded,
                truncation_reason="result_limit" if len(rows) > bounded else None,
                results=results,
            )

    def backlinks(
        self,
        node_id: str,
        *,
        relation: str | None = None,
        limit: int | None = None,
    ) -> dict[str, object]:
        return self._edges(node_id, incoming=True, relation=relation, limit=limit)

    def dependencies(
        self,
        node_id: str,
        *,
        depth: int = 2,
        limit: int | None = None,
    ) -> dict[str, object]:
        return self._traverse(
            node_id,
            incoming=False,
            depth=depth,
            relation="depends_on",
            limit=limit,
        )

    def impact(
        self,
        node_id: str,
        *,
        depth: int = 2,
        limit: int | None = None,
    ) -> dict[str, object]:
        return self._traverse(
            node_id,
            incoming=True,
            depth=depth,
            relation=None,
            limit=limit,
        )

    def _edges(
        self,
        node_id: str,
        *,
        incoming: bool,
        relation: str | None,
        limit: int | None,
    ) -> dict[str, object]:
        bounded = _bounded_limit(
            limit,
            self.project.descriptor.limits.max_results,
            default=self.project.descriptor.limits.max_results,
        )
        source_column = "target_id" if incoming else "source_id"
        relation_clause = " AND relation = ?" if relation is not None else ""
        values: tuple[object, ...] = (
            (node_id, relation, bounded + 1) if relation is not None else (node_id, bounded + 1)
        )
        with self._read_snapshot() as snapshot:
            self._require_node(snapshot.connection, node_id)
            rows = snapshot.connection.execute(
                f"SELECT source_id, relation, target_id FROM edges "
                f"WHERE {source_column} = ?{relation_clause} "
                "ORDER BY source_id, relation, target_id LIMIT ?",
                values,
            ).fetchall()
            truncated = len(rows) > bounded
            edges = [Edge(*row).as_dict() for row in rows[:bounded]]
            return snapshot.result(
                root=node_id,
                relation=relation,
                count=len(edges),
                limit=bounded,
                truncated=truncated,
                truncation_reason="result_limit" if truncated else None,
                edges=edges,
            )

    def _traverse(
        self,
        node_id: str,
        *,
        incoming: bool,
        depth: int,
        relation: str | None,
        limit: int | None,
    ) -> dict[str, object]:
        maximum = self.project.descriptor.limits.max_traversal_depth
        if type(depth) is not int or depth < 0 or depth > maximum:
            raise DocForgeError("invalid_depth", "Traversal depth is outside the configured limit")
        bounded = _bounded_limit(
            limit,
            self.project.descriptor.limits.max_results,
            default=self.project.descriptor.limits.max_results,
        )
        examined_limit = (bounded + 1) ** 2
        source_column = "target_id" if incoming else "source_id"
        relation_clause = " AND relation = ?" if relation is not None else ""
        with self._read_snapshot() as snapshot:
            self._require_node(snapshot.connection, node_id)
            queue: deque[tuple[str, int, tuple[str, ...]]] = deque([(node_id, 0, (node_id,))])
            seen = {node_id}
            results: list[dict[str, object]] = []
            truncation_reason: str | None = None
            candidate_edges_consumed = 0
            while queue and truncation_reason is None:
                current, current_depth, path = queue.popleft()
                if current_depth >= depth:
                    continue
                remaining = examined_limit - candidate_edges_consumed
                if remaining <= 0:
                    truncation_reason = "edge_examination_limit"
                    break
                values: tuple[object, ...] = (
                    (current, relation, remaining + 1)
                    if relation is not None
                    else (current, remaining + 1)
                )
                candidates = snapshot.connection.execute(
                    "SELECT source_id, relation, target_id FROM edges "
                    f"WHERE {source_column} = ?{relation_clause} "
                    "ORDER BY source_id, relation, target_id LIMIT ?",
                    values,
                )
                for row in candidates:
                    if candidate_edges_consumed >= examined_limit:
                        truncation_reason = "edge_examination_limit"
                        break
                    candidate_edges_consumed += 1
                    edge = Edge(*row)
                    target = edge.source_id if incoming else edge.target_id
                    if target in seen:
                        continue
                    if len(results) >= bounded:
                        truncation_reason = "result_limit"
                        break
                    seen.add(target)
                    target_path = (*path, target)
                    results.append(
                        {
                            "node_id": target,
                            "depth": current_depth + 1,
                            "relation": edge.relation,
                            "path": target_path,
                        }
                    )
                    queue.append((target, current_depth + 1, target_path))
            return snapshot.result(
                root=node_id,
                depth=depth,
                count=len(results),
                limit=bounded,
                truncated=truncation_reason is not None,
                truncation_reason=truncation_reason,
                candidate_edges_consumed=candidate_edges_consumed,
                candidate_edges_limit=examined_limit,
                results=results,
            )

    @staticmethod
    def _require_node(connection: sqlite3.Connection, node_id: str) -> None:
        exists = connection.execute(
            "SELECT 1 FROM nodes WHERE node_id = ?",
            (node_id,),
        ).fetchone()
        if exists is None:
            raise DocForgeError(
                "missing_node", "No node has the requested stable ID", node_id=node_id
            )


def _row_to_node(row: sqlite3.Row) -> Node:
    return Node(
        node_id=row["node_id"],
        title=row["title"],
        family=row["family"],
        authority=row["authority"],
        status=row["status"],
        tags=tuple(json.loads(row["tags_json"])),
        summary=row["summary"],
        content=row["content"],
        source_path=row["source_path"],
        source_anchor=row["source_anchor"],
        content_hash=row["content_hash"],
    )


def _logic_projection_from_connection(
    connection: sqlite3.Connection,
    owner_node_id: str,
) -> LogicProjection | None:
    owner = connection.execute(
        "SELECT owner_node_id, source_id FROM logic_owners WHERE owner_node_id = ?",
        (owner_node_id,),
    ).fetchone()
    if owner is None:
        return None
    nodes = tuple(
        LogicNode(
            logic_id=row["logic_id"],
            kind=row["kind"],
            label=row["label"],
            source_anchor=row["source_anchor"],
        )
        for row in connection.execute(
            "SELECT logic_id, kind, label, source_anchor FROM logic_nodes "
            "WHERE owner_node_id = ? ORDER BY logic_id",
            (owner_node_id,),
        )
    )
    edges = tuple(
        LogicEdge(
            source_id=row["source_id"],
            relation=row["relation"],
            target_id=row["target_id"],
            label=row["label"],
            ordinal=row["ordinal"],
        )
        for row in connection.execute(
            "SELECT source_id, relation, target_id, label, ordinal FROM logic_edges "
            "WHERE owner_node_id = ? "
            "ORDER BY source_id, ordinal, relation, target_id",
            (owner_node_id,),
        )
    )
    return LogicProjection(
        owner_node_id=owner["owner_node_id"],
        source_id=owner["source_id"],
        nodes=nodes,
        edges=edges,
    )


def _logic_from_connection(
    connection: sqlite3.Connection,
) -> tuple[LogicProjection, ...]:
    owners = connection.execute(
        "SELECT owner_node_id FROM logic_owners ORDER BY owner_node_id"
    ).fetchall()
    projections = [
        _logic_projection_from_connection(connection, row["owner_node_id"]) for row in owners
    ]
    return tuple(projection for projection in projections if projection is not None)


def _bounded_limit(value: int | None, maximum: int, *, default: int) -> int:
    if value is None:
        return min(default, maximum)
    if type(value) is not int or value < 1 or value > maximum:
        raise DocForgeError("invalid_limit", "Result limit is outside the configured range")
    return value


def re_tokenize(query: str) -> tuple[str, ...]:
    token = ""
    tokens: list[str] = []
    for character in query.casefold():
        if character.isalnum() or character in {"_", "-"}:
            token += character
        elif token:
            tokens.append(token)
            token = ""
    if token:
        tokens.append(token)
    return tuple(tokens)