1578 lines
60 KiB
Python
1578 lines
60 KiB
Python
"""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@15"
|
|
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",
|
|
),
|
|
}
|