DocForge2/src/docforge/visualization.py

"""Project-bound, read-only graph visualization over a validated DocForge index."""

from __future__ import annotations

import atexit
import json
import os
import secrets
import signal
import socket
import sqlite3
import sys
import tempfile
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from collections.abc import Generator
from contextlib import contextmanager
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from importlib import resources
from pathlib import Path
from time import monotonic
from typing import cast

try:
    import fcntl
except ImportError:  # pragma: no cover - retained only for the deprecated direct runner.
    fcntl = None

from .errors import DocForgeError
from .index import APPLICATION_ID, INDEX_SCHEMA_VERSION, ProjectIndex, re_tokenize
from .project import project_root_fingerprint

VISUALIZATION_TEMPLATE = "graph-browser@14"
DEFAULT_EDGE_LIMIT = 100
MAX_EDGE_LIMIT = 400
MAX_LINEAGE_EDGE_LIMIT = 1_000
FLOW_REVERSED_RELATIONS = frozenset(
    {
        "defined_in",
        "inherits",
        "imports",
        "depends_on",
        "reads",
        "tested_by",
    }
)
FLOW_CONTEXT_RELATIONS = frozenset(
    {
        "documents",
        "governs",
        "relates_to",
    }
)
WEB_ROOT_ADJACENT_RELATIONS = frozenset(
    {
        "activates",
        "calls",
        "contains",
        "defines",
        "dispatches_to",
        "implemented_by",
        "launches",
        "writes",
    }
)
DEFAULT_INITIAL_GRACE_SECONDS = 120.0
DEFAULT_LEASE_SECONDS = 180.0
LEASE_MONITOR_INTERVAL_SECONDS = 1.0
VISUALIZATION_REGISTRY_NAME = ".visualization.json"
VISUALIZATION_LOCK_NAME = ".visualization.lock"
VISUALIZATION_RUNTIME = "persistent-worker@1"


class _VisualizationHttpServer(ThreadingHTTPServer):
    daemon_threads = True
    allow_reuse_address = False


class VisualizationIndexSnapshot:
    """Fast read model pinned to one index file validated by ``ProjectIndex.check``."""

    _IDENTITY_KEYS = (
        "project_id",
        "project_root_fingerprint",
        "revision",
        "source_hash",
        "adapter",
        "node_count",
        "edge_count",
    )

    def __init__(self, index: ProjectIndex, checked: dict[str, object]) -> None:
        self.path = index.path
        self.project_root = index.project.descriptor.root
        self.title = index.project.descriptor.title
        self.max_source_bytes = index.project.descriptor.limits.max_source_bytes
        self.max_query_chars = index.project.descriptor.limits.max_query_chars
        self.max_results = index.project.descriptor.limits.max_results
        self.max_depth = index.project.descriptor.limits.max_traversal_depth
        self.identity: dict[str, object] = {key: checked[key] for key in self._IDENTITY_KEYS}
        self._stat = self._safe_stat()

    @classmethod
    def from_spec(cls, spec: dict[str, object]) -> VisualizationIndexSnapshot:
        snapshot = cls.__new__(cls)
        path = spec["path"]
        title = spec["title"]
        project_root = spec["project_root"]
        max_source_bytes = spec["max_source_bytes"]
        max_query_chars = spec["max_query_chars"]
        max_results = spec["max_results"]
        max_depth = spec["max_depth"]
        if (
            not isinstance(path, str)
            or not isinstance(title, str)
            or not isinstance(project_root, str)
            or type(max_source_bytes) is not int
            or type(max_query_chars) is not int
            or type(max_results) is not int
            or type(max_depth) is not int
        ):
            raise DocForgeError("invalid_index", "Visualization snapshot is invalid")
        snapshot.path = Path(path)
        try:
            snapshot.project_root = Path(project_root).resolve(strict=True)
        except OSError as error:
            raise DocForgeError("invalid_index", "Visualization project root is invalid") from error
        if not snapshot.project_root.is_dir():
            raise DocForgeError("invalid_index", "Visualization project root is invalid")
        snapshot.title = title
        snapshot.max_source_bytes = max_source_bytes
        snapshot.max_query_chars = max_query_chars
        snapshot.max_results = max_results
        snapshot.max_depth = max_depth
        identity = spec["identity"]
        if not isinstance(identity, dict):
            raise DocForgeError("invalid_index", "Visualization identity is invalid")
        typed_identity = cast(dict[str, object], identity)
        snapshot.identity = {key: typed_identity[key] for key in cls._IDENTITY_KEYS}
        snapshot._stat = snapshot._safe_stat()
        return snapshot

    def spec(self) -> dict[str, object]:
        return {
            "path": str(self.path),
            "project_root": str(self.project_root),
            "title": self.title,
            "max_source_bytes": self.max_source_bytes,
            "max_query_chars": self.max_query_chars,
            "max_results": self.max_results,
            "max_depth": self.max_depth,
            "identity": dict(self.identity),
        }

    def overview(self) -> dict[str, object]:
        with self._connection() as connection:
            return self._result(
                title=self.title,
                node_count=connection.execute("SELECT COUNT(*) FROM nodes").fetchone()[0],
                edge_count=connection.execute("SELECT COUNT(*) FROM edges").fetchone()[0],
                families=_facet_rows(connection, "nodes", "family"),
                authorities=_facet_rows(connection, "nodes", "authority"),
                statuses=_facet_rows(connection, "nodes", "status"),
                relations=_facet_rows(connection, "edges", "relation"),
                max_results=self.max_results,
                max_depth=self.max_depth,
                snapshot=True,
            )

    def search(
        self,
        *,
        query: str,
        family: str | None,
        limit: int,
    ) -> dict[str, object]:
        bounded = self._bounded_limit(limit)
        with self._connection() as connection:
            if query:
                if len(query) > self.max_query_chars:
                    raise DocForgeError(
                        "invalid_query", "Search query exceeds the configured limit"
                    )
                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
                )
                family_clause = "AND nodes.family = ?" if family else ""
                values: tuple[object, ...] = (
                    (expression, family, bounded) if family else (expression, bounded)
                )
                rows = 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 ?
                    """
                    + family_clause
                    + " ORDER BY rank, nodes.node_id LIMIT ?",
                    values,
                ).fetchall()
                results: list[dict[str, object]] = []
                for row in rows:
                    item = _node_dict(row, include_content=False)
                    item.update({"rank": row["rank"], "snippet": row["snippet"]})
                    results.append(item)
            else:
                family_clause = "WHERE family = ?" if family else ""
                values = (family, bounded) if family else (bounded,)
                rows = connection.execute(
                    f"SELECT * FROM nodes {family_clause} ORDER BY node_id LIMIT ?",
                    values,
                ).fetchall()
                results = [_node_dict(row, include_content=False) for row in rows]
        return self._result(
            query=query,
            family=family,
            count=len(results),
            results=results,
            snapshot=True,
        )

    def filter_nodes(
        self,
        *,
        category: str,
        value: str,
        limit: int,
    ) -> dict[str, object]:
        bounded = self._bounded_limit(limit)
        if category not in {"family", "authority", "status", "tag"}:
            raise DocForgeError("invalid_filter", "Descriptor filter category is unsupported")
        if not value or len(value) > self.max_query_chars:
            raise DocForgeError("invalid_filter", "Descriptor filter value is invalid")
        if category == "tag":
            clause = "EXISTS (SELECT 1 FROM json_each(tags_json) WHERE value = ?)"
        else:
            clause = f"{category} = ?"
        with self._connection() as connection:
            total = connection.execute(
                f"SELECT COUNT(*) FROM nodes WHERE {clause}",
                (value,),
            ).fetchone()[0]
            rows = connection.execute(
                f"SELECT * FROM nodes WHERE {clause} ORDER BY node_id LIMIT ?",
                (value, bounded),
            ).fetchall()
        results = [_node_dict(row, include_content=False) for row in rows]
        return self._result(
            category=category,
            value=value,
            count=len(results),
            total=total,
            truncated=total > len(results),
            results=results,
            snapshot=True,
        )

    def node(self, node_id: str, *, depth: int, limit: int) -> dict[str, object]:
        if type(depth) is not int or depth < 1 or depth > self.max_depth:
            raise DocForgeError("invalid_depth", "Traversal depth is outside the configured limit")
        if type(limit) is not int or limit < 1 or limit > MAX_EDGE_LIMIT:
            raise DocForgeError(
                "invalid_limit",
                "Visualization edge limit exceeds the fixed safety boundary",
                maximum=MAX_EDGE_LIMIT,
            )
        with self._connection() as connection:
            root_row = connection.execute(
                "SELECT * FROM nodes WHERE node_id = ?", (node_id,)
            ).fetchone()
            if root_row is None:
                raise DocForgeError(
                    "missing_node",
                    "No node has the requested stable ID",
                    node_id=node_id,
                )
            visited = {node_id}
            frontier = {node_id}
            selected: list[dict[str, str]] = []
            selected_keys: set[tuple[str, str, str]] = set()
            truncated = False
            for _ in range(depth):
                if not frontier or len(selected) >= limit:
                    break
                placeholders = ",".join("?" for _ in frontier)
                values = tuple(sorted(frontier))
                rows = connection.execute(
                    "SELECT source_id, relation, target_id FROM edges "
                    f"WHERE source_id IN ({placeholders}) OR target_id IN ({placeholders}) "
                    "ORDER BY source_id, relation, target_id LIMIT ?",
                    (*values, *values, limit + 1),
                ).fetchall()
                next_frontier: set[str] = set()
                for row in rows:
                    key = (row["source_id"], row["relation"], row["target_id"])
                    if key in selected_keys:
                        continue
                    if len(selected) >= limit:
                        truncated = True
                        break
                    selected_keys.add(key)
                    edge = {
                        "source_id": row["source_id"],
                        "relation": row["relation"],
                        "target_id": row["target_id"],
                    }
                    selected.append(edge)
                    for candidate in (edge["source_id"], edge["target_id"]):
                        if candidate not in visited:
                            visited.add(candidate)
                            next_frontier.add(candidate)
                frontier = next_frontier
            placeholders = ",".join("?" for _ in visited)
            node_rows = connection.execute(
                f"SELECT * FROM nodes WHERE node_id IN ({placeholders}) ORDER BY node_id",
                tuple(sorted(visited)),
            ).fetchall()
        return self._result(
            root=node_id,
            depth=depth,
            edge_limit=limit,
            truncated=truncated,
            node=_node_dict(root_row),
            nodes=[_node_dict(row, include_content=False) for row in node_rows],
            edges=selected,
            snapshot=True,
        )

    def source(self, node_id: str) -> dict[str, object]:
        """Return one node's bounded, project-confined UTF-8 source file."""

        with self._connection() as connection:
            row = connection.execute(
                "SELECT node_id, source_path, source_anchor 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,
            )
        relative = Path(row["source_path"])
        if relative.is_absolute() or ".." in relative.parts or not relative.parts:
            raise DocForgeError("path_escape", "Node source path is unsafe", node_id=node_id)
        source = self.project_root / relative
        try:
            resolved = source.resolve(strict=True)
        except OSError as error:
            raise DocForgeError(
                "missing_source",
                "Node source file is unavailable",
                node_id=node_id,
            ) from error
        if (
            source.is_symlink()
            or resolved != source
            or not source.is_relative_to(self.project_root)
            or not source.is_file()
        ):
            raise DocForgeError("path_escape", "Node source file is unsafe", node_id=node_id)
        if source.stat().st_size > self.max_source_bytes:
            raise DocForgeError(
                "source_too_large",
                "Node source exceeds the configured source limit",
                node_id=node_id,
            )
        raw = source.read_bytes()
        if len(raw) > self.max_source_bytes:
            raise DocForgeError(
                "source_too_large",
                "Node source exceeds the configured source limit",
                node_id=node_id,
            )
        try:
            content = raw.decode("utf-8")
        except UnicodeDecodeError as error:
            raise DocForgeError(
                "invalid_source",
                "Node source is not UTF-8",
                node_id=node_id,
            ) from error
        return self._result(
            node_id=node_id,
            source_path=row["source_path"],
            source_anchor=row["source_anchor"],
            content=content,
            snapshot=True,
        )

    def lineage(self, node_id: str, *, limit: int) -> dict[str, object]:
        """Return bounded semantic flow paths terminating at ``node_id``.

        Structural and execution edges retain their stored direction. Dependency,
        import, data-read, inheritance, and ``tested_by`` edges are reversed so their
        prerequisites flow into the consumer. Context-only documentation edges remain
        available in Web but do not clutter Flow.
        """
        if type(limit) is not int or limit < 1 or limit > MAX_LINEAGE_EDGE_LIMIT:
            raise DocForgeError(
                "invalid_limit",
                "Visualization lineage limit exceeds the fixed safety boundary",
                maximum=MAX_LINEAGE_EDGE_LIMIT,
            )
        with self._connection() as connection:
            root_row = connection.execute(
                "SELECT * FROM nodes WHERE node_id = ?", (node_id,)
            ).fetchone()
            if root_row is None:
                raise DocForgeError(
                    "missing_node",
                    "No node has the requested stable ID",
                    node_id=node_id,
                )
            visited = {node_id}
            frontier = {node_id}
            hops = {node_id: 0}
            selected: list[dict[str, object]] = []
            selected_keys: set[tuple[str, str, str]] = set()
            truncated = False
            while frontier and len(selected) < limit:
                placeholders = ",".join("?" for _ in frontier)
                rows = connection.execute(
                    "SELECT source_id, relation, target_id FROM edges "
                    f"WHERE source_id IN ({placeholders}) OR target_id IN ({placeholders}) "
                    "ORDER BY source_id, relation, target_id",
                    (*sorted(frontier), *sorted(frontier)),
                ).fetchall()
                next_frontier: set[str] = set()
                for row in rows:
                    key = (row["source_id"], row["relation"], row["target_id"])
                    if key in selected_keys or row["relation"] in FLOW_CONTEXT_RELATIONS:
                        continue
                    reversed_edge = row["relation"] in FLOW_REVERSED_RELATIONS
                    source_id = row["target_id"] if reversed_edge else row["source_id"]
                    target_id = row["source_id"] if reversed_edge else row["target_id"]
                    if target_id not in frontier:
                        continue
                    if len(selected) >= limit:
                        truncated = True
                        break
                    selected_keys.add(key)
                    edge = _visualization_edge(row, reversed_edge=reversed_edge)
                    selected.append(edge)
                    if source_id not in visited:
                        visited.add(source_id)
                        hops[source_id] = hops[target_id] + 1
                        next_frontier.add(source_id)
                frontier = next_frontier
            if frontier and len(selected) >= limit:
                truncated = True
            placeholders = ",".join("?" for _ in visited)
            node_rows = connection.execute(
                f"SELECT * FROM nodes WHERE node_id IN ({placeholders}) ORDER BY node_id",
                tuple(sorted(visited)),
            ).fetchall()
        return self._result(
            root=node_id,
            lineage=True,
            edge_limit=limit,
            truncated=truncated,
            hops=hops,
            node=_node_dict(root_row),
            nodes=[_node_dict(row, include_content=False) for row in node_rows],
            edges=selected,
            snapshot=True,
        )

    def web(self, node_id: str, *, depth: int, limit: int) -> dict[str, object]:
        """Return a bounded convergence web centered on ``node_id``.

        Web follows every semantic contributor path toward the focus, including the
        context relationships omitted from Flow. It also reverses direct focus-owned
        members and execution dependencies into adjacent contributor branches. Later
        traversal continues only toward those branches, so entering a package or class
        cannot fan out through unrelated siblings.
        """
        if type(depth) is not int or depth < 1 or depth > self.max_depth:
            raise DocForgeError("invalid_depth", "Traversal depth is outside the configured limit")
        if type(limit) is not int or limit < 1 or limit > MAX_LINEAGE_EDGE_LIMIT:
            raise DocForgeError(
                "invalid_limit",
                "Visualization web limit exceeds the fixed safety boundary",
                maximum=MAX_LINEAGE_EDGE_LIMIT,
            )
        with self._connection() as connection:
            root_row = connection.execute(
                "SELECT * FROM nodes WHERE node_id = ?", (node_id,)
            ).fetchone()
            if root_row is None:
                raise DocForgeError(
                    "missing_node",
                    "No node has the requested stable ID",
                    node_id=node_id,
                )
            visited = {node_id}
            frontier = {node_id}
            hops = {node_id: 0}
            selected: list[dict[str, object]] = []
            selected_keys: set[tuple[str, str, str, bool]] = set()
            truncated = False
            for hop in range(1, depth + 1):
                if not frontier or len(selected) >= limit:
                    break
                placeholders = ",".join("?" for _ in frontier)
                values = tuple(sorted(frontier))
                rows = connection.execute(
                    "SELECT source_id, relation, target_id FROM edges "
                    f"WHERE source_id IN ({placeholders}) OR target_id IN ({placeholders}) "
                    "ORDER BY source_id, relation, target_id",
                    (*values, *values),
                ).fetchall()
                next_frontier: set[str] = set()
                for row in rows:
                    semantic_reversed = row["relation"] in FLOW_REVERSED_RELATIONS
                    semantic_source = row["target_id"] if semantic_reversed else row["source_id"]
                    semantic_target = row["source_id"] if semantic_reversed else row["target_id"]
                    candidates: list[tuple[str, bool]] = []
                    if semantic_target in frontier:
                        candidates.append((semantic_source, semantic_reversed))
                    if (
                        semantic_source == node_id
                        and row["relation"] in WEB_ROOT_ADJACENT_RELATIONS
                    ):
                        candidates.append((semantic_target, not semantic_reversed))
                    for source_id, reversed_edge in candidates:
                        key = (
                            row["source_id"],
                            row["relation"],
                            row["target_id"],
                            reversed_edge,
                        )
                        if key in selected_keys:
                            continue
                        existing_hop = hops.get(source_id)
                        if existing_hop is not None and existing_hop < hop:
                            continue
                        if len(selected) >= limit:
                            truncated = True
                            break
                        selected_keys.add(key)
                        selected.append(_visualization_edge(row, reversed_edge=reversed_edge))
                        if source_id not in visited:
                            visited.add(source_id)
                            hops[source_id] = hop
                            next_frontier.add(source_id)
                    if truncated:
                        break
                frontier = next_frontier
            if frontier and len(selected) >= limit:
                truncated = True

            convergent_ids = {node_id}
            for edge in selected:
                convergent_ids.add(cast(str, edge["source_id"]))
                convergent_ids.add(cast(str, edge["target_id"]))
            placeholders = ",".join("?" for _ in convergent_ids)
            node_rows = connection.execute(
                f"SELECT * FROM nodes WHERE node_id IN ({placeholders}) ORDER BY node_id",
                tuple(sorted(convergent_ids)),
            ).fetchall()
        return self._result(
            root=node_id,
            web=True,
            depth=depth,
            edge_limit=limit,
            truncated=truncated,
            hops={node: hops[node] for node in sorted(convergent_ids)},
            node=_node_dict(root_row),
            nodes=[_node_dict(row, include_content=False) for row in node_rows],
            edges=selected,
            snapshot=True,
        )

    def require_node(self, node_id: str) -> None:
        with self._connection() as connection:
            row = connection.execute("SELECT 1 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,
            )

    def _bounded_limit(self, value: int) -> int:
        if type(value) is not int or value < 1 or value > self.max_results:
            raise DocForgeError("invalid_limit", "Result limit is outside the configured range")
        return value

    def _safe_stat(self) -> tuple[int, int, int, int]:
        if (
            self.path.is_symlink()
            or not self.path.is_file()
            or self.path.resolve(strict=True) != self.path
        ):
            raise DocForgeError("missing_index", "Validated visualization index is unavailable")
        stat = self.path.stat()
        return (stat.st_dev, stat.st_ino, stat.st_size, stat.st_mtime_ns)

    @contextmanager
    def _connection(self) -> Generator[sqlite3.Connection, None, None]:
        if self._safe_stat() != self._stat:
            raise DocForgeError(
                "visualization_stale",
                "The validated index changed; invoke docforge_visualize again",
            )
        connection: sqlite3.Connection | None = None
        try:
            connection = sqlite3.connect(f"file:{self.path}?mode=ro", uri=True)
            connection.row_factory = sqlite3.Row
            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", "Visualization index has an unsupported schema"
                )
            metadata = dict(connection.execute("SELECT key, value FROM metadata"))
            if any(metadata.get(key) != str(value) for key, value in self.identity.items()):
                raise DocForgeError(
                    "visualization_stale",
                    "The validated index identity changed; invoke docforge_visualize again",
                )
            yield connection
            if self._safe_stat() != self._stat:
                raise DocForgeError(
                    "visualization_stale",
                    "The validated index changed during the request",
                )
        except DocForgeError:
            raise
        except (json.JSONDecodeError, KeyError, sqlite3.Error, TypeError) as error:
            raise DocForgeError(
                "invalid_index", "Visualization index is corrupt or unreadable"
            ) from error
        finally:
            if connection is not None:
                connection.close()

    def _result(self, **payload: object) -> dict[str, object]:
        return {
            "status": "ok",
            **self.identity,
            **payload,
        }

    def result(self, **payload: object) -> dict[str, object]:
        return self._result(**payload)


class VisualizationRunner:
    """Start one token-protected loopback reader for one immutable project binding."""

    def __init__(
        self,
        index: ProjectIndex | None,
        *,
        snapshot_spec: dict[str, object] | None = None,
        token: str | None = None,
        register_atexit: bool = True,
        persistent: bool = False,
        initial_grace_seconds: float = DEFAULT_INITIAL_GRACE_SECONDS,
        lease_seconds: float = DEFAULT_LEASE_SECONDS,
        monitor_interval_seconds: float = LEASE_MONITOR_INTERVAL_SECONDS,
    ) -> None:
        if (index is None) == (snapshot_spec is None):
            raise ValueError("Provide exactly one visualization index or snapshot")
        if (
            (not persistent and initial_grace_seconds <= 0)
            or lease_seconds <= 0
            or monitor_interval_seconds <= 0
        ):
            raise ValueError("Visualization lease durations must be positive")
        self.index = index
        self._snapshot_spec = snapshot_spec
        self._register_atexit = register_atexit
        self.persistent = persistent
        self.initial_grace_seconds = initial_grace_seconds
        self.lease_seconds = lease_seconds
        self.monitor_interval_seconds = monitor_interval_seconds
        self._lock = threading.Lock()
        self._token = token or secrets.token_urlsafe(24)
        self._server: _VisualizationHttpServer | None = None
        self._thread: threading.Thread | None = None
        self._lease_thread: threading.Thread | None = None
        self._lease_stop = threading.Event()
        self._lease_last_activity = monotonic()
        self._activity_last_seen = time.time()
        self._lease_connected = False
        self._atexit_registered = False
        self._reader: VisualizationIndexSnapshot | None = None

    def start(
        self,
        *,
        node_id: str | None = None,
        query: str | None = None,
        depth: int = 1,
    ) -> dict[str, object]:
        if node_id is not None and query is not None:
            raise DocForgeError(
                "invalid_visualization_target",
                "Choose either one exact node ID or one search query",
            )
        reader = (
            VisualizationIndexSnapshot(self.index, self.index.check())
            if self.index is not None
            else VisualizationIndexSnapshot.from_spec(self._snapshot_spec or {})
        )
        maximum_depth = reader.max_depth
        if type(depth) is not int or depth < 1 or depth > maximum_depth:
            raise DocForgeError(
                "invalid_depth",
                "Visualization depth is outside the configured traversal limit",
            )
        if node_id is not None:
            reader.require_node(node_id)
        elif query is not None:
            reader.search(query=query, family=None, limit=1)

        with self._lock:
            self._reader = reader
            if self._server is None:
                runner = self

                class Handler(BaseHTTPRequestHandler):
                    def do_GET(self) -> None:  # noqa: N802
                        runner._handle_get(self)

                    def do_HEAD(self) -> None:  # noqa: N802
                        runner._handle_get(self, include_body=False)

                    def do_POST(self) -> None:  # noqa: N802
                        runner._respond_error(
                            self,
                            HTTPStatus.METHOD_NOT_ALLOWED,
                            "method_not_allowed",
                            "The visualization service is read-only",
                        )

                    def do_PUT(self) -> None:  # noqa: N802
                        self.do_POST()

                    def do_PATCH(self) -> None:  # noqa: N802
                        self.do_POST()

                    def do_DELETE(self) -> None:  # noqa: N802
                        self.do_POST()

                    def log_message(self, format: str, *args: object) -> None:
                        del format, args
                        return

                self._server = _VisualizationHttpServer(("127.0.0.1", 0), Handler)
                self._lease_last_activity = monotonic()
                self._activity_last_seen = time.time()
                self._lease_connected = False
                self._lease_stop.clear()
                self._thread = threading.Thread(
                    target=self._server.serve_forever,
                    name="docforge-visualization",
                    daemon=False,
                )
                self._thread.start()
                if not self.persistent:
                    self._lease_thread = threading.Thread(
                        target=self._monitor_lease,
                        name="docforge-visualization-lease",
                        daemon=True,
                    )
                    self._lease_thread.start()
                if self._register_atexit and not self._atexit_registered:
                    atexit.register(self.stop)
                    self._atexit_registered = True

            server = self._server
            assert server is not None
            port = int(server.server_address[1])

        parameters: dict[str, str] = {"depth": str(depth)}
        if node_id is not None:
            parameters["node"] = node_id
        if query is not None:
            parameters["q"] = query
        query_string = urllib.parse.urlencode(parameters)
        url = f"http://127.0.0.1:{port}/{self._token}/"
        if query_string:
            url = f"{url}?{query_string}"
        return {
            "state": "running",
            "url": url,
            "bind": "127.0.0.1",
            "port": port,
            "template": VISUALIZATION_TEMPLATE,
            "read_only": True,
            "project_bound": True,
            "lifetime": (
                {"policy": "explicit_stop"}
                if self.persistent
                else {
                    "policy": "browser_lease",
                    "initial_grace_seconds": self.initial_grace_seconds,
                    "lease_seconds": self.lease_seconds,
                }
            ),
            "target": {
                "node_id": node_id,
                "query": query,
                "depth": depth,
            },
            "snapshot": dict(reader.identity),
        }

    def stop(self) -> None:
        with self._lock:
            server = self._server
            thread = self._thread
            self._server = None
            self._thread = None
            self._lease_thread = None
            self._reader = None
            self._lease_stop.set()
        if server is None:
            return
        server.shutdown()
        server.server_close()
        if thread is not None and thread is not threading.current_thread():
            thread.join(timeout=2)

    def is_running(self) -> bool:
        with self._lock:
            return self._server is not None

    def _touch_lease(self) -> None:
        with self._lock:
            if self._server is None:
                return
            self._lease_last_activity = monotonic()
            self._activity_last_seen = time.time()
            self._lease_connected = True

    def _monitor_lease(self) -> None:
        while not self._lease_stop.wait(self.monitor_interval_seconds):
            with self._lock:
                if self._server is None:
                    return
                timeout = (
                    self.lease_seconds if self._lease_connected else self.initial_grace_seconds
                )
                expired = monotonic() - self._lease_last_activity >= timeout
            if expired:
                self.stop()
                return

    def _handle_get(self, handler: BaseHTTPRequestHandler, *, include_body: bool = True) -> None:
        parsed = urllib.parse.urlparse(handler.path)
        prefix = f"/{self._token}"
        page_paths = {prefix, f"{prefix}/"}
        asset_prefix = f"{prefix}/assets/"
        is_asset = parsed.path.startswith(asset_prefix)
        if (
            parsed.path not in page_paths
            and not parsed.path.startswith(f"{prefix}/api/")
            and not is_asset
        ):
            self._respond_error(
                handler,
                HTTPStatus.NOT_FOUND,
                "not_found",
                "Not found",
                include_body=include_body,
            )
            return
        if parsed.path in page_paths:
            self._touch_lease()
            self._respond(
                handler,
                _GRAPH_BROWSER_HTML.encode("utf-8"),
                "text/html; charset=utf-8",
                include_body=include_body,
            )
            return
        if is_asset:
            asset = _GRAPH_BROWSER_ASSETS.get(parsed.path.removeprefix(asset_prefix))
            if asset is None:
                self._respond_error(
                    handler,
                    HTTPStatus.NOT_FOUND,
                    "not_found",
                    "Not found",
                    include_body=include_body,
                )
                return
            payload, content_type = asset
            self._respond(
                handler,
                payload,
                content_type,
                include_body=include_body,
            )
            return

        try:
            params = urllib.parse.parse_qs(parsed.query, keep_blank_values=True)
            reader = self._current_reader()
            if parsed.path == f"{prefix}/api/health":
                with self._lock:
                    last_activity = self._activity_last_seen
                payload = reader.result(viewer="alive", last_activity_at=last_activity)
            elif parsed.path == f"{prefix}/api/overview":
                self._touch_lease()
                payload = reader.overview()
            elif parsed.path == f"{prefix}/api/heartbeat":
                self._touch_lease()
                payload = reader.result(
                    viewer="alive",
                    lease_seconds=self.lease_seconds,
                )
            elif parsed.path == f"{prefix}/api/search":
                self._touch_lease()
                payload = self._search(reader, params)
            elif parsed.path == f"{prefix}/api/filter":
                self._touch_lease()
                payload = self._filter(reader, params)
            elif parsed.path == f"{prefix}/api/node":
                self._touch_lease()
                payload = self._node(reader, params)
            elif parsed.path == f"{prefix}/api/source":
                self._touch_lease()
                payload = self._source(reader, params)
            elif parsed.path == f"{prefix}/api/lineage":
                self._touch_lease()
                payload = self._lineage(reader, params)
            elif parsed.path == f"{prefix}/api/web":
                self._touch_lease()
                payload = self._web(reader, params)
            else:
                self._respond_error(
                    handler,
                    HTTPStatus.NOT_FOUND,
                    "not_found",
                    "Not found",
                    include_body=include_body,
                )
                return
        except DocForgeError as error:
            status = {
                "missing_node": HTTPStatus.NOT_FOUND,
                "stale_index": HTTPStatus.CONFLICT,
                "source_changed": HTTPStatus.CONFLICT,
                "stale_adapter_source": HTTPStatus.CONFLICT,
                "visualization_stale": HTTPStatus.CONFLICT,
                "invalid_query": HTTPStatus.BAD_REQUEST,
                "invalid_filter": HTTPStatus.BAD_REQUEST,
                "invalid_depth": HTTPStatus.BAD_REQUEST,
                "invalid_limit": HTTPStatus.BAD_REQUEST,
                "path_escape": HTTPStatus.FORBIDDEN,
                "missing_source": HTTPStatus.NOT_FOUND,
                "source_too_large": HTTPStatus.REQUEST_ENTITY_TOO_LARGE,
            }.get(error.code, HTTPStatus.SERVICE_UNAVAILABLE)
            self._respond_json(
                handler,
                {"status": "error", "error": error.as_dict()},
                status=status,
                include_body=include_body,
            )
            return
        except (TypeError, ValueError):
            self._respond_error(
                handler,
                HTTPStatus.BAD_REQUEST,
                "invalid_request",
                "Request parameters are invalid",
                include_body=include_body,
            )
            return
        self._respond_json(handler, payload, include_body=include_body)

    def _search(
        self,
        reader: VisualizationIndexSnapshot,
        params: dict[str, list[str]],
    ) -> dict[str, object]:
        query = _one(params, "q").strip()
        family = _one(params, "family").strip() or None
        limit = _integer(_one(params, "limit") or "50")
        return reader.search(query=query, family=family, limit=limit)

    def _node(
        self,
        reader: VisualizationIndexSnapshot,
        params: dict[str, list[str]],
    ) -> dict[str, object]:
        node_id = _one(params, "id").strip()
        if not node_id:
            raise DocForgeError("missing_node", "One exact node ID is required")
        depth = _integer(_one(params, "depth") or "1")
        limit = _integer(_one(params, "limit") or str(DEFAULT_EDGE_LIMIT))
        if limit > MAX_EDGE_LIMIT:
            raise DocForgeError(
                "invalid_limit",
                "Visualization edge limit exceeds the fixed safety boundary",
                maximum=MAX_EDGE_LIMIT,
            )
        return reader.node(node_id, depth=depth, limit=limit)

    @staticmethod
    def _source(
        reader: VisualizationIndexSnapshot,
        params: dict[str, list[str]],
    ) -> dict[str, object]:
        node_id = _one(params, "id").strip()
        if not node_id:
            raise DocForgeError("missing_node", "One exact node ID is required")
        return reader.source(node_id)

    def _lineage(
        self,
        reader: VisualizationIndexSnapshot,
        params: dict[str, list[str]],
    ) -> dict[str, object]:
        node_id = _one(params, "id").strip()
        if not node_id:
            raise DocForgeError("missing_node", "One exact node ID is required")
        limit = _integer(_one(params, "limit") or str(MAX_LINEAGE_EDGE_LIMIT))
        if limit > MAX_LINEAGE_EDGE_LIMIT:
            raise DocForgeError(
                "invalid_limit",
                "Visualization lineage limit exceeds the fixed safety boundary",
                maximum=MAX_LINEAGE_EDGE_LIMIT,
            )
        return reader.lineage(node_id, limit=limit)

    def _web(
        self,
        reader: VisualizationIndexSnapshot,
        params: dict[str, list[str]],
    ) -> dict[str, object]:
        node_id = _one(params, "id").strip()
        if not node_id:
            raise DocForgeError("missing_node", "One exact node ID is required")
        depth = _integer(_one(params, "depth") or str(reader.max_depth))
        limit = _integer(_one(params, "limit") or str(MAX_LINEAGE_EDGE_LIMIT))
        if limit > MAX_LINEAGE_EDGE_LIMIT:
            raise DocForgeError(
                "invalid_limit",
                "Visualization web limit exceeds the fixed safety boundary",
                maximum=MAX_LINEAGE_EDGE_LIMIT,
            )
        return reader.web(node_id, depth=depth, limit=limit)

    def _filter(
        self,
        reader: VisualizationIndexSnapshot,
        params: dict[str, list[str]],
    ) -> dict[str, object]:
        category = _one(params, "category").strip()
        value = _one(params, "value").strip()
        limit = _integer(_one(params, "limit") or "50")
        return reader.filter_nodes(category=category, value=value, limit=limit)

    def _current_reader(self) -> VisualizationIndexSnapshot:
        with self._lock:
            reader = self._reader
        if reader is None:
            raise DocForgeError(
                "visualization_unavailable",
                "The visualization snapshot is unavailable",
            )
        return reader

    @staticmethod
    def _respond(
        handler: BaseHTTPRequestHandler,
        payload: bytes,
        content_type: str,
        *,
        status: HTTPStatus = HTTPStatus.OK,
        include_body: bool = True,
    ) -> None:
        handler.send_response(status)
        handler.send_header("Content-Type", content_type)
        handler.send_header("Content-Length", str(len(payload)))
        handler.send_header("Cache-Control", "no-store")
        handler.send_header(
            "Content-Security-Policy",
            "default-src 'none'; script-src 'self'; style-src 'self'; "
            "connect-src 'self'; img-src 'self' data:; base-uri 'none'; form-action 'none'; "
            "frame-ancestors 'none'",
        )
        handler.send_header("X-Content-Type-Options", "nosniff")
        handler.send_header("X-Frame-Options", "DENY")
        handler.send_header("Referrer-Policy", "no-referrer")
        handler.end_headers()
        if include_body:
            handler.wfile.write(payload)

    @classmethod
    def _respond_json(
        cls,
        handler: BaseHTTPRequestHandler,
        body: dict[str, object],
        *,
        status: HTTPStatus = HTTPStatus.OK,
        include_body: bool = True,
    ) -> None:
        cls._respond(
            handler,
            json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8"),
            "application/json; charset=utf-8",
            status=status,
            include_body=include_body,
        )

    @classmethod
    def _respond_error(
        cls,
        handler: BaseHTTPRequestHandler,
        status: HTTPStatus,
        code: str,
        message: str,
        *,
        include_body: bool = True,
    ) -> None:
        cls._respond_json(
            handler,
            {
                "status": "error",
                "error": {"code": code, "message": message, "details": {}},
            },
            status=status,
            include_body=include_body,
        )


class PersistentVisualizationRunner:
    """Run one project-bound browser until an explicit DocForge stop request."""

    def __init__(self, index: ProjectIndex) -> None:
        self.index = index

    @property
    def _cache_root(self) -> Path:
        return self.index.project.descriptor.cache_root

    @property
    def _registry_path(self) -> Path:
        return self._cache_root / VISUALIZATION_REGISTRY_NAME

    @property
    def _lock_path(self) -> Path:
        return self._cache_root / VISUALIZATION_LOCK_NAME

    @contextmanager
    def _locked_registry(self) -> Generator[None, None, None]:
        if fcntl is None:
            raise DocForgeError(
                "visualization_unavailable",
                "The deprecated direct visualization runner is unavailable on this platform",
            )
        self._cache_root.mkdir(parents=True, exist_ok=True)
        with self._lock_path.open("a+", encoding="utf-8") as handle:
            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
            try:
                yield
            finally:
                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)

    def _read_registry(self) -> dict[str, object] | None:
        path = self._registry_path
        if not path.exists():
            return None
        if path.is_symlink() or not path.is_file():
            raise DocForgeError("visualization_unavailable", "Visualization registry is unsafe")
        try:
            raw_document: object = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            self._remove_registry()
            return None
        if not isinstance(raw_document, dict):
            self._remove_registry()
            return None
        document = cast(dict[str, object], raw_document)
        pid = document.get("pid")
        port = document.get("port")
        token = document.get("token")
        snapshot = document.get("snapshot")
        if (
            document.get("runtime") != VISUALIZATION_RUNTIME
            or document.get("template") != VISUALIZATION_TEMPLATE
            or type(pid) is not int
            or pid <= 1
            or type(port) is not int
            or not 1 <= port <= 65535
            or not isinstance(token, str)
            or len(token) < 20
            or not isinstance(snapshot, dict)
        ):
            self._remove_registry()
            return None
        return document

    def _remove_registry(self) -> None:
        path = self._registry_path
        if not path.exists():
            return
        if path.is_symlink() or not path.is_file():
            raise DocForgeError("visualization_unavailable", "Visualization registry is unsafe")
        path.unlink()

    def _write_registry(self, document: dict[str, object]) -> None:
        descriptor, temporary_name = tempfile.mkstemp(
            prefix=".visualization-", dir=self._cache_root
        )
        temporary = Path(temporary_name)
        try:
            os.fchmod(descriptor, 0o600)
            with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
                json.dump(document, handle, sort_keys=True, separators=(",", ":"))
                handle.write("\n")
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(temporary, self._registry_path)
        except OSError:
            temporary.unlink(missing_ok=True)
            raise

    @staticmethod
    def _worker_process(pid: int) -> bool:
        if pid <= 1:
            return False
        try:
            command = Path(f"/proc/{pid}/cmdline").read_bytes()
        except OSError:
            return False
        return b"docforge.visualization_worker" in command

    @staticmethod
    def _target_url(
        *,
        port: int,
        token: str,
        node_id: str | None,
        query: str | None,
        depth: int,
    ) -> str:
        parameters: dict[str, str] = {"depth": str(depth)}
        if node_id is not None:
            parameters["node"] = node_id
        if query is not None:
            parameters["q"] = query
        return f"http://127.0.0.1:{port}/{token}/?{urllib.parse.urlencode(parameters)}"

    def _matches_snapshot(
        self,
        document: dict[str, object],
        snapshot: VisualizationIndexSnapshot,
    ) -> bool:
        record = document.get("snapshot")
        return isinstance(record, dict) and record == snapshot.identity

    def _is_live(
        self,
        document: dict[str, object],
        snapshot: VisualizationIndexSnapshot,
    ) -> bool:
        if not self._matches_snapshot(document, snapshot):
            return False
        pid = cast(int, document["pid"])
        port = cast(int, document["port"])
        token = cast(str, document["token"])
        if not self._worker_process(pid):
            return False
        request = urllib.request.Request(
            f"http://127.0.0.1:{port}/{token}/api/overview",
            headers={"Accept": "application/json"},
        )
        try:
            with urllib.request.urlopen(request, timeout=1) as response:
                raw_payload: object = json.load(response)
        except (OSError, ValueError, urllib.error.URLError):
            return False
        if not isinstance(raw_payload, dict):
            return False
        payload = cast(dict[str, object], raw_payload)
        return all(payload.get(key) == value for key, value in snapshot.identity.items())

    @staticmethod
    def _terminate_worker(pid: int) -> None:
        if not PersistentVisualizationRunner._worker_process(pid):
            return
        try:
            os.kill(pid, signal.SIGTERM)
        except ProcessLookupError:
            return
        except PermissionError as error:
            raise DocForgeError(
                "visualization_unavailable", "Visualization worker cannot be stopped"
            ) from error
        deadline = monotonic() + 2
        while PersistentVisualizationRunner._worker_process(pid) and monotonic() < deadline:
            time.sleep(0.05)
        if PersistentVisualizationRunner._worker_process(pid):
            try:
                os.kill(pid, signal.SIGKILL)
            except ProcessLookupError:
                return

    def _result(
        self,
        snapshot: VisualizationIndexSnapshot,
        *,
        port: int,
        token: str,
        node_id: str | None,
        query: str | None,
        depth: int,
        reused: bool,
    ) -> dict[str, object]:
        return {
            "state": "running",
            "reused": reused,
            "url": self._target_url(
                port=port,
                token=token,
                node_id=node_id,
                query=query,
                depth=depth,
            ),
            "bind": "127.0.0.1",
            "port": port,
            "template": VISUALIZATION_TEMPLATE,
            "read_only": True,
            "project_bound": True,
            "lifetime": {
                "policy": "explicit_stop",
                "stop_tool": "docforge_stop_visualization",
            },
            "target": {"node_id": node_id, "query": query, "depth": depth},
            "snapshot": dict(snapshot.identity),
        }

    def start(
        self,
        *,
        node_id: str | None = None,
        query: str | None = None,
        depth: int = 1,
    ) -> dict[str, object]:
        if node_id is not None and query is not None:
            raise DocForgeError(
                "invalid_visualization_target",
                "Choose either one exact node ID or one search query",
            )
        snapshot = VisualizationIndexSnapshot(self.index, self.index.check())
        if type(depth) is not int or depth < 1 or depth > snapshot.max_depth:
            raise DocForgeError(
                "invalid_depth",
                "Visualization depth is outside the configured traversal limit",
            )
        if node_id is not None:
            snapshot.require_node(node_id)
        elif query is not None:
            snapshot.search(query=query, family=None, limit=1)

        with self._locked_registry():
            existing = self._read_registry()
            if existing is not None and self._is_live(existing, snapshot):
                return self._result(
                    snapshot,
                    port=cast(int, existing["port"]),
                    token=cast(str, existing["token"]),
                    node_id=node_id,
                    query=query,
                    depth=depth,
                    reused=True,
                )
            if existing is not None:
                self._terminate_worker(cast(int, existing["pid"]))
                self._remove_registry()
            return self._launch(snapshot, node_id=node_id, query=query, depth=depth)

    def _launch(
        self,
        snapshot: VisualizationIndexSnapshot,
        *,
        node_id: str | None,
        query: str | None,
        depth: int,
    ) -> dict[str, object]:
        parent_socket, child_socket = socket.socketpair()
        process_id: int | None = None
        token = secrets.token_urlsafe(24)
        try:
            command = (
                sys.executable,
                "-m",
                "docforge.visualization_worker",
                "--control-fd",
                str(child_socket.fileno()),
            )
            child_socket.set_inheritable(True)
            process_id = os.posix_spawn(
                sys.executable,
                command,
                os.environ,
                setsid=True,
            )
            child_socket.close()
            request = {
                "snapshot": snapshot.spec(),
                "token": token,
                "target": {"node_id": node_id, "query": query, "depth": depth},
            }
            parent_socket.settimeout(10)
            parent_socket.sendall(
                json.dumps(request, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n"
            )
            response = _receive_worker_response(parent_socket)
            if response.get("status") != "ok" or not isinstance(
                response.get("visualization"), dict
            ):
                raise ValueError("Visualization worker rejected the snapshot")
            visualization = cast(dict[str, object], response["visualization"])
            port = visualization.get("port")
            if type(port) is not int or not 1 <= port <= 65535:
                raise ValueError("Visualization worker returned an invalid port")
            self._write_registry(
                {
                    "runtime": VISUALIZATION_RUNTIME,
                    "template": VISUALIZATION_TEMPLATE,
                    "pid": process_id,
                    "port": port,
                    "token": token,
                    "snapshot": dict(snapshot.identity),
                }
            )
            return self._result(
                snapshot,
                port=port,
                token=token,
                node_id=node_id,
                query=query,
                depth=depth,
                reused=False,
            )
        except (OSError, ValueError) as error:
            if process_id is not None:
                self._terminate_worker(process_id)
            raise DocForgeError(
                "visualization_unavailable",
                "The persistent visualization worker failed to start",
            ) from error
        finally:
            child_socket.close()
            parent_socket.close()

    def stop(self) -> dict[str, object]:
        with self._locked_registry():
            existing = self._read_registry()
            if existing is None:
                return self._stop_result("not_running")
            self._terminate_worker(cast(int, existing["pid"]))
            self._remove_registry()
            return self._stop_result("stopped")

    def _stop_result(self, state: str) -> dict[str, object]:
        descriptor = self.index.project.descriptor
        return {
            "status": "ok",
            "state": state,
            "project_id": descriptor.project_id,
            "project_root_fingerprint": project_root_fingerprint(descriptor.root),
            "adapter": descriptor.adapter,
        }


def _receive_worker_response(control: socket.socket) -> dict[str, object]:
    chunks: list[bytes] = []
    size = 0
    while True:
        chunk = control.recv(65536)
        if not chunk:
            break
        chunks.append(chunk)
        size += len(chunk)
        if size > 1_000_000:
            raise ValueError("Visualization worker response exceeded its fixed boundary")
        if b"\n" in chunk:
            break
    payload = b"".join(chunks).split(b"\n", 1)[0]
    result = json.loads(payload)
    if not isinstance(result, dict):
        raise ValueError("Visualization worker returned an invalid response")
    return cast(dict[str, object], result)


def _one(params: dict[str, list[str]], name: str) -> str:
    values = params.get(name) or [""]
    return values[0]


def _integer(value: str) -> int:
    return int(value)


def _visualization_edge(
    row: sqlite3.Row,
    *,
    reversed_edge: bool,
) -> dict[str, object]:
    stored_source_id = cast(str, row["source_id"])
    stored_target_id = cast(str, row["target_id"])
    return {
        "source_id": stored_target_id if reversed_edge else stored_source_id,
        "relation": row["relation"],
        "target_id": stored_source_id if reversed_edge else stored_target_id,
        "stored_source_id": stored_source_id,
        "stored_target_id": stored_target_id,
        "reversed": reversed_edge,
    }


def _node_dict(row: sqlite3.Row, *, include_content: bool = True) -> dict[str, object]:
    result = {
        "node_id": row["node_id"],
        "title": row["title"],
        "family": row["family"],
        "authority": row["authority"],
        "status": row["status"],
        "tags": json.loads(row["tags_json"]),
        "summary": row["summary"],
        "source_path": row["source_path"],
        "source_anchor": row["source_anchor"],
        "content_hash": row["content_hash"],
    }
    if include_content:
        result["content"] = row["content"]
    return result


def _facet_rows(connection: sqlite3.Connection, table: str, column: str) -> list[dict[str, object]]:
    allowed = {
        ("nodes", "family"),
        ("nodes", "authority"),
        ("nodes", "status"),
        ("edges", "relation"),
    }
    if (table, column) not in allowed:
        raise DocForgeError("invalid_index", "Unsupported visualization facet")
    rows = connection.execute(
        f"SELECT {column}, COUNT(*) AS count FROM {table} "
        f"GROUP BY {column} ORDER BY count DESC, {column}"
    ).fetchall()
    return [{"value": row[0], "count": row[1]} for row in rows]


def _read_browser_asset(name: str) -> str:
    return resources.files("docforge.assets").joinpath(name).read_text(encoding="utf-8")


_GRAPH_BROWSER_HTML = _read_browser_asset("graph.html")
_GRAPH_BROWSER_CSS = _read_browser_asset("graph.css")
_GRAPH_BROWSER_JAVASCRIPT = _read_browser_asset("graph.js")
_GRAPH_BROWSER_ASSETS = {
    "graph.css": (_GRAPH_BROWSER_CSS.encode("utf-8"), "text/css; charset=utf-8"),
    "graph.js": (
        _GRAPH_BROWSER_JAVASCRIPT.encode("utf-8"),
        "text/javascript; charset=utf-8",
    ),
}