1
0
Fork 0
Code Issues Pull requests Projects Releases 2 Packages Wiki Activity Actions Pages

Add durable portable graph publication

This commit is contained in:
Andraxion 2026-07-29 11:37:33 -04:00
parent 96e3965855
commit 1134c2d375
19 changed files with 2542 additions and 13 deletions

View file

@ -3,7 +3,10 @@
from __future__ import annotations
import os
import secrets
import stat
from collections.abc import Callable
from contextlib import suppress
from pathlib import Path
from .errors import DocForgeError
@ -69,3 +72,177 @@ def require_bound_directory(path: Path, directory_fd: int) -> None:
"path_escape",
"Derived cache root disappeared during publication",
) from error
def open_confined_directory(root: Path, path: Path, *, create: bool) -> int:
"""Open a descendant directory through stable no-follow directory descriptors."""
try:
unsafe = (
root.is_symlink()
or root.resolve(strict=True) != root
or not path.is_relative_to(root)
or path == root
)
except OSError as error:
raise DocForgeError("path_escape", "Project root cannot be resolved safely") from error
if unsafe:
raise DocForgeError("path_escape", "Derived output directory is not confined")
relative = path.relative_to(root)
try:
descriptor = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
except OSError as error:
raise DocForgeError("path_escape", "Project root cannot be opened safely") from error
try:
for part in relative.parts:
if part in {"", ".", ".."}:
raise DocForgeError("path_escape", "Derived output directory is not confined")
if create:
try:
os.mkdir(part, mode=0o700, dir_fd=descriptor)
except FileExistsError:
pass
except OSError as error:
raise DocForgeError(
"publication_failure",
"Derived output directory could not be created",
) from error
try:
next_descriptor = os.open(
part,
os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
dir_fd=descriptor,
)
except OSError as error:
raise DocForgeError(
"path_escape",
"Derived output directory is missing or unsafe",
) from error
os.close(descriptor)
descriptor = next_descriptor
require_bound_directory(path, descriptor)
return descriptor
except Exception:
os.close(descriptor)
raise
def safe_file_identity_at(
directory: Path,
directory_fd: int,
name: str,
) -> dict[str, object] | None:
"""Return one no-follow regular-file identity relative to a bound directory."""
del directory
if not name or "/" in name or name in {".", ".."}:
raise DocForgeError("path_escape", "Derived artifact name is unsafe")
try:
current = os.stat(name, dir_fd=directory_fd, follow_symlinks=False)
except FileNotFoundError:
return None
except OSError as error:
raise DocForgeError("path_escape", "Derived artifact cannot be inspected") from error
if not stat.S_ISREG(current.st_mode):
raise DocForgeError("path_escape", "Derived artifact is not a safe regular file")
return {
"path": name,
"device": current.st_dev,
"inode": current.st_ino,
"mode": current.st_mode,
"size": current.st_size,
"mtime_ns": current.st_mtime_ns,
"ctime_ns": current.st_ctime_ns,
}
def read_bounded_file_at(
directory_fd: int,
name: str,
maximum_bytes: int,
) -> bytes | None:
"""Read one regular file through a bound directory without following links."""
try:
descriptor = os.open(name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=directory_fd)
except FileNotFoundError:
return None
except OSError as error:
raise DocForgeError("path_escape", "Derived artifact cannot be opened safely") from error
with os.fdopen(descriptor, "rb") as handle:
current = os.fstat(handle.fileno())
if not stat.S_ISREG(current.st_mode) or current.st_size > maximum_bytes:
raise DocForgeError("invalid_projection", "Derived artifact is invalid or oversized")
content = handle.read(maximum_bytes + 1)
if len(content) > maximum_bytes:
raise DocForgeError("invalid_projection", "Derived artifact is oversized")
return content
def atomic_replace_bytes_at(
path: Path,
directory_fd: int,
name: str,
content: bytes,
*,
verify: Callable[[], None],
) -> dict[str, object]:
"""Durably replace one file inside an already bound directory."""
if not name or "/" in name or name in {".", ".."}:
raise DocForgeError("path_escape", "Derived artifact name is unsafe")
existing = safe_file_identity_at(path, directory_fd, name)
del existing
temporary = f".docforge-projection-{secrets.token_hex(12)}"
descriptor: int | None = None
committed = False
try:
descriptor = os.open(
temporary,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
0o600,
dir_fd=directory_fd,
)
with os.fdopen(descriptor, "wb") as handle:
descriptor = None
handle.write(content)
handle.flush()
os.fsync(handle.fileno())
verify()
require_bound_directory(path, directory_fd)
os.replace(
temporary,
name,
src_dir_fd=directory_fd,
dst_dir_fd=directory_fd,
)
committed = True
os.fsync(directory_fd)
identity = safe_file_identity_at(path, directory_fd, name)
if identity is None:
raise DocForgeError(
"publication_failure",
"Derived artifact disappeared after publication",
mutation_committed=True,
)
return identity
except DocForgeError as error:
if committed:
raise DocForgeError(
"publication_failure",
"Derived artifact was replaced but final publication verification failed",
mutation_committed=True,
cause=error.code,
) from error
raise
except OSError as error:
raise DocForgeError(
"publication_failure",
"Derived artifact publication failed",
mutation_committed=committed,
) from error
finally:
if descriptor is not None:
os.close(descriptor)
with suppress(OSError):
os.unlink(temporary, dir_fd=directory_fd)

View file

@ -13,6 +13,7 @@ from .client_config import CLIENT_NAMES, generate_client_configuration
from .context import compile_context
from .doctor import run_doctor
from .errors import DocForgeError
from .graph_rendering import GraphRenderService
from .index import ProjectIndex
from .onboarding import assess_project, scaffold_project
from .project import Project, project_root_fingerprint
@ -95,6 +96,12 @@ def _parser() -> argparse.ArgumentParser:
render_status = commands.add_parser("render-status")
render_status.add_argument("view_id", nargs="?")
render_status.add_argument("--deep", action="store_true")
graph_plan = commands.add_parser("graph-plan")
graph_plan.add_argument("view_id")
graph_render = commands.add_parser("graph-render")
graph_render.add_argument("view_id")
graph_render_status = commands.add_parser("graph-render-status")
graph_render_status.add_argument("view_id", nargs="?")
preview = commands.add_parser("preview")
preview.add_argument("changeset_id")
preview.add_argument("view_id")
@ -258,6 +265,12 @@ def _run(arguments: argparse.Namespace) -> dict[str, object]:
if arguments.deep
else rendering.status(arguments.view_id)
)
if arguments.command == "graph-plan":
return GraphRenderService(project).plan(arguments.view_id)
if arguments.command == "graph-render":
return GraphRenderService(project).render(arguments.view_id)
if arguments.command == "graph-render-status":
return GraphRenderService(project).status(arguments.view_id)
if arguments.command == "preview":
return RenderService(project).preview(arguments.changeset_id, arguments.view_id)
if arguments.command == "apply":

View file

@ -9,18 +9,20 @@ from typing import Literal, cast
from .errors import DocForgeError
from .models import Edge, Node, ProjectSnapshot
from .project import project_root_fingerprint
from .projection_contract import GraphViewPlanV1
from .projection_contract import (
MAX_GRAPH_VIEW_DEPTH,
MAX_GRAPH_VIEW_EDGES,
MAX_GRAPH_VIEW_FILTERS,
MAX_GRAPH_VIEW_NODES,
MAX_GRAPH_VIEW_QUERY_CHARS,
MAX_GRAPH_VIEW_STRING_CHARS,
MAX_GRAPH_VIEW_WORK,
GraphViewPlanV1,
ProjectionPackageV1,
)
GraphViewMode = Literal["nodes", "flow", "web", "logic"]
MAX_GRAPH_VIEW_DEPTH = 32
MAX_GRAPH_VIEW_NODES = 1_000
MAX_GRAPH_VIEW_EDGES = 4_000
MAX_GRAPH_VIEW_WORK = 1_000_000
MAX_GRAPH_VIEW_FILTERS = 64
MAX_GRAPH_VIEW_STRING_CHARS = 1_024
MAX_GRAPH_VIEW_QUERY_CHARS = 10_000
_QUERY_TOKEN = re.compile(r"\w+", re.UNICODE)
_DETAIL_FIELDS = (
"node_id",
@ -495,3 +497,29 @@ def build_graph_view_plan(
},
}
)
def build_graph_projection_package(
plan: GraphViewPlanV1,
*,
renderer_id: str,
renderer_version: str,
max_output_bytes: int,
) -> ProjectionPackageV1:
"""Bind a graph plan to the fixed portable renderer without adding runtime authority."""
return ProjectionPackageV1.create(
kind="graph",
plan=plan,
renderer={"renderer_id": renderer_id, "renderer_version": renderer_version},
components=[
{"component_id": "graph.portable-document@1"},
{"component_id": "graph.accessible-list@1"},
{"component_id": "graph.relationship-table@1"},
],
assets=[],
output_policy={
"artifact_ids": ["portable-graph.html"],
"max_total_bytes": max_output_bytes,
},
)

View file

@ -0,0 +1,285 @@
"""Strict parsing and confinement for optional portable graph artifacts."""
from __future__ import annotations
from pathlib import Path
from typing import Literal, cast
from .config_validation import (
ID_PATTERN,
confined_path,
positive_int,
require_string,
string_list,
)
from .errors import DocForgeError
from .models import GraphRenderConfig, GraphRenderView, Limits, RenderConfig
from .projection_contract import (
MAX_GRAPH_VIEW_DEPTH,
MAX_GRAPH_VIEW_EDGES,
MAX_GRAPH_VIEW_FILTERS,
MAX_GRAPH_VIEW_NODES,
MAX_GRAPH_VIEW_QUERY_CHARS,
MAX_GRAPH_VIEW_STRING_CHARS,
MAX_GRAPH_VIEW_WORK,
)
_CONFIG_KEYS = frozenset({"output_root", "views"})
_VIEW_KEYS = frozenset(
{
"id",
"renderer",
"output",
"title",
"root",
"query",
"initial_mode",
"depth",
"max_nodes",
"max_edges",
"max_work",
"families",
"relations",
"authorities",
"statuses",
"tags",
"include_logic",
}
)
_MODES = frozenset({"nodes", "flow", "web"})
def _overlaps(first: Path, second: Path) -> bool:
return first == second or first.is_relative_to(second) or second.is_relative_to(first)
def _optional_string(document: dict[str, object], key: str, source: Path) -> str | None:
if key not in document:
return None
return require_string(document, key, source)
def _bounded_string(
document: dict[str, object],
key: str,
source: Path,
*,
maximum: int,
) -> str:
value = require_string(document, key, source)
if len(value) > maximum:
raise DocForgeError("invalid_config", f"{key} exceeds its fixed character limit")
return value
def _bounded_strings(
value: object,
*,
key: str,
source: Path,
) -> tuple[str, ...]:
values = string_list(value, key=key, source=source)
if len(values) > MAX_GRAPH_VIEW_FILTERS or any(
len(item) > MAX_GRAPH_VIEW_STRING_CHARS for item in values
):
raise DocForgeError("invalid_config", f"{key} exceeds its fixed bounds")
return values
def load_graph_render_config(
root: Path,
document: object,
*,
descriptor_path: Path,
content_roots: tuple[Path, ...],
authority_files: tuple[Path, ...],
cache_root: Path,
index_path: Path,
changeset_root: Path,
manual_render: RenderConfig | None,
limits: Limits,
) -> GraphRenderConfig | None:
if document is None:
return None
if not isinstance(document, dict):
raise DocForgeError("invalid_config", "graph_render must be a table")
document = cast(dict[str, object], document)
unknown = sorted(set(document) - _CONFIG_KEYS)
if unknown:
raise DocForgeError("invalid_config", "graph_render has unknown fields", fields=unknown)
output_root = confined_path(
root,
document.get("output_root"),
field="graph_render.output_root",
must_exist=False,
)
protected = [*content_roots, cache_root, changeset_root]
if manual_render is not None:
protected.extend((manual_render.template_root, manual_render.preview_root))
protected.extend(view.output_path for view in manual_render.views)
if any(_overlaps(output_root, path) for path in protected):
raise DocForgeError(
"invalid_config",
"Portable graph output must not overlap canonical or other derived roots",
)
protected_files = (descriptor_path, index_path, *authority_files)
if any(path == output_root or path.is_relative_to(output_root) for path in protected_files):
raise DocForgeError(
"invalid_config",
"Portable graph output overlaps a protected project path",
)
view_values_value = document.get("views")
if not isinstance(view_values_value, list) or not view_values_value:
raise DocForgeError(
"invalid_config",
"graph_render.views must contain at least one view",
)
view_values = cast(list[object], view_values_value)
if len(view_values) > limits.max_render_views:
raise DocForgeError("invalid_config", "graph_render.views exceeds the configured limit")
views: list[GraphRenderView] = []
view_ids: set[str] = set()
outputs: set[Path] = set()
for value in view_values:
if not isinstance(value, dict):
raise DocForgeError("invalid_config", "Each portable graph view must be a table")
view = cast(dict[str, object], value)
unknown_view = sorted(set(view) - _VIEW_KEYS)
if unknown_view:
raise DocForgeError(
"invalid_config",
"Portable graph view has unknown fields",
fields=unknown_view,
)
view_id = require_string(view, "id", descriptor_path)
if ID_PATTERN.fullmatch(view_id) is None or view_id in view_ids:
raise DocForgeError(
"invalid_config",
"Portable graph view ID is invalid or duplicated",
id=view_id,
)
view_ids.add(view_id)
renderer = require_string(view, "renderer", descriptor_path)
if renderer != "portable_graph_html":
raise DocForgeError(
"unsupported_renderer",
"Portable graph view names an unsupported built-in renderer",
renderer=renderer,
)
output = confined_path(
output_root,
view.get("output"),
field="graph_render.view.output",
must_exist=False,
)
if output.suffix != ".html" or output in outputs:
raise DocForgeError(
"invalid_config",
"Portable graph outputs must be unique HTML files",
)
outputs.add(output)
root_node_id = _optional_string(view, "root", descriptor_path)
query = _optional_string(view, "query", descriptor_path)
if (root_node_id is None) == (query is None):
raise DocForgeError(
"invalid_config",
"Portable graph view requires exactly one root or query",
)
if root_node_id is not None and len(root_node_id) > MAX_GRAPH_VIEW_STRING_CHARS:
raise DocForgeError("invalid_config", "Portable graph root exceeds its fixed limit")
if query is not None and len(query) > MAX_GRAPH_VIEW_QUERY_CHARS:
raise DocForgeError("invalid_config", "Portable graph query exceeds its fixed limit")
initial_mode_value = view.get("initial_mode", "nodes")
if not isinstance(initial_mode_value, str):
raise DocForgeError(
"invalid_config",
"Portable graph initial mode is unsupported",
)
initial_mode = cast(
Literal["nodes", "flow", "web", "logic"],
initial_mode_value,
)
if initial_mode not in _MODES:
raise DocForgeError(
"invalid_config",
"Portable graph initial mode is unsupported",
)
depth = positive_int(view.get("depth", 1), "graph_render.view.depth")
max_nodes = positive_int(view.get("max_nodes", 100), "graph_render.view.max_nodes")
max_edges = positive_int(
view.get("max_edges", 400),
"graph_render.view.max_edges",
allow_zero=True,
)
max_work = positive_int(view.get("max_work", 100_000), "graph_render.view.max_work")
if (
depth > min(limits.max_traversal_depth, MAX_GRAPH_VIEW_DEPTH)
or max_nodes > min(limits.max_nodes, MAX_GRAPH_VIEW_NODES)
or max_edges > MAX_GRAPH_VIEW_EDGES
or max_work > MAX_GRAPH_VIEW_WORK
):
raise DocForgeError(
"invalid_config",
"Portable graph view exceeds project or fixed safety limits",
)
include_logic = view.get("include_logic", False)
if type(include_logic) is not bool:
raise DocForgeError(
"invalid_config",
"Portable graph include_logic must be Boolean",
)
if include_logic:
raise DocForgeError(
"unsupported_renderer",
"Portable graph renderer version 1 does not support Logic projections",
)
views.append(
GraphRenderView(
view_id=view_id,
renderer=renderer,
output_path=output,
title=_bounded_string(
view,
"title",
descriptor_path,
maximum=MAX_GRAPH_VIEW_STRING_CHARS,
),
root_node_id=root_node_id,
query=query,
initial_mode=initial_mode,
depth=depth,
max_nodes=max_nodes,
max_edges=max_edges,
max_work=max_work,
families=_bounded_strings(
view.get("families", []),
key="graph_render.view.families",
source=descriptor_path,
),
relations=_bounded_strings(
view.get("relations", []),
key="graph_render.view.relations",
source=descriptor_path,
),
authorities=_bounded_strings(
view.get("authorities", []),
key="graph_render.view.authorities",
source=descriptor_path,
),
statuses=_bounded_strings(
view.get("statuses", []),
key="graph_render.view.statuses",
source=descriptor_path,
),
tags=_bounded_strings(
view.get("tags", []),
key="graph_render.view.tags",
source=descriptor_path,
),
include_logic=include_logic,
)
)
return GraphRenderConfig(
output_root=output_root,
views=tuple(sorted(views, key=lambda item: item.view_id)),
)

View file

@ -0,0 +1,793 @@
"""Declared portable graph planning, publication, and receipt-only status."""
from __future__ import annotations
import fcntl
import json
import os
from collections.abc import Callable, Generator
from contextlib import contextmanager
from pathlib import Path
from typing import cast
from ._fs_safety import (
atomic_replace_bytes_at,
open_confined_directory,
read_bounded_file_at,
require_bound_directory,
safe_file_identity_at,
)
from .errors import DocForgeError
from .graph_projection import (
GraphViewRequestV1,
build_graph_projection_package,
build_graph_view_plan,
)
from .models import (
GenerationRecordingProject,
GraphRenderConfig,
GraphRenderView,
IncrementalStateProject,
ProjectService,
ProjectSnapshot,
ProjectState,
)
from .project import project_root_fingerprint
from .projection_contract import GraphViewPlanV1, ProjectionReceiptV1, projection_hash
GRAPH_RENDERER_ID = "portable_graph_html"
GRAPH_RENDERER_VERSION = "1"
GRAPH_PUBLICATION_MANIFEST_VERSION = 1
GRAPH_PUBLICATION_CONTRACT = "docforge.graph-publication"
MAX_GRAPH_PUBLICATION_BYTES = 256_000
class GraphRenderService:
"""Publish one declared artifact while keeping planning and rendering independent."""
def __init__(self, project: ProjectService, *, allow_logic: bool = False) -> None:
self.project = project
self.allow_logic = allow_logic
def plan(self, view_id: str) -> dict[str, object]:
snapshot = self.project.load()
view = self._view(self._config(snapshot), view_id)
plan = self._plan(snapshot, view)
return {
"status": "ok",
**self._identity(snapshot),
"view_id": view.view_id,
"plan": plan.as_dict(),
}
def status(self, view_id: str | None = None) -> dict[str, object]:
config = self.project.descriptor.graph_render
current = self._current_state()
if config is None:
return self._status_result(
current,
configured=False,
state="not_configured",
outputs=[],
)
views = config.views if view_id is None else (self._view(config, view_id),)
first_outputs = [self._manifest_status(view, current) for view in views]
outputs = [self._manifest_status(view, current) for view in views]
if outputs != first_outputs:
for output in outputs:
if output["state"] == "current":
output["state"] = "stale"
output["reason"] = "publication_changed_during_status"
final = self._current_state()
if final != current:
for output in outputs:
if output["state"] == "current":
output["state"] = "stale"
output["reason"] = "source_changed_during_status"
identity = final if final is not None else current
return self._status_result(
identity,
configured=True,
state="current" if all(item["state"] == "current" for item in outputs) else "stale",
outputs=outputs,
)
def render(self, view_id: str) -> dict[str, object]:
with self._lock():
current_status = self.status(view_id)
current_outputs = cast(list[dict[str, object]], current_status["outputs"])
if current_status["state"] == "current" and current_outputs:
return {
**current_status,
"publication": "unchanged",
"output": current_outputs[0],
}
snapshot = self.project.load()
view = self._view(self._config(snapshot), view_id)
plan = self._plan(snapshot, view)
package = build_graph_projection_package(
plan,
renderer_id=GRAPH_RENDERER_ID,
renderer_version=GRAPH_RENDERER_VERSION,
max_output_bytes=snapshot.descriptor.limits.max_render_bytes,
)
from docforge_renderers.graph import PortableGraphHtmlRenderer
result = PortableGraphHtmlRenderer().render(package)
if len(result.artifacts) != 1:
raise DocForgeError(
"invalid_projection",
"Portable graph renderer returned an unsupported artifact set",
)
artifact = result.artifacts[0]
def verify() -> None:
current = self.project.load()
if (
current.revision != snapshot.revision
or current.source_hash != snapshot.source_hash
):
raise DocForgeError(
"render_input_changed",
"Canonical input changed during portable graph rendering",
)
verify()
if isinstance(self.project, GenerationRecordingProject):
self.project.record_generation(snapshot)
artifact_evidence = artifact.evidence()
try:
store_identity = self._publish_artifact(
snapshot,
artifact_evidence["sha256"],
artifact.content,
verify=verify,
)
except DocForgeError as error:
if self._mutation_committed(error):
return self._degraded_publication(
snapshot,
view,
plan,
package.package_id,
result.receipt.as_dict(),
artifact_evidence,
stage="artifact_store",
error=error,
output_published=False,
)
raise
try:
output_identity = self._publish_output(
snapshot,
view,
artifact.content,
verify=verify,
)
except DocForgeError as error:
if self._mutation_committed(error):
return self._degraded_publication(
snapshot,
view,
plan,
package.package_id,
result.receipt.as_dict(),
artifact_evidence,
stage="output",
error=error,
output_published=True,
)
raise
manifest = self._manifest(
snapshot,
view,
plan,
package.package_id,
result.receipt.as_dict(),
artifact_evidence,
store_identity,
output_identity,
)
try:
self._publish_manifest(snapshot, view, manifest, verify=verify)
except DocForgeError as error:
return self._degraded_publication(
snapshot,
view,
plan,
package.package_id,
result.receipt.as_dict(),
artifact_evidence,
stage="manifest",
error=error,
output_published=True,
)
return {
"status": "ok",
**self._identity(snapshot),
"view_id": view.view_id,
"state": "current",
"publication": "published",
"plan_id": plan.plan_id,
"package_id": package.package_id,
"output": {
**artifact.evidence(),
"path": view.output_path.relative_to(snapshot.descriptor.root).as_posix(),
},
"receipt": result.receipt.as_dict(),
"manifest": {
"state": "current",
"publication_id": manifest["publication_id"],
},
}
@staticmethod
def _mutation_committed(error: DocForgeError) -> bool:
return error.details.get("mutation_committed") is True
def _degraded_publication(
self,
snapshot: ProjectSnapshot,
view: GraphRenderView,
plan: GraphViewPlanV1,
package_id: str,
receipt: dict[str, object],
artifact: dict[str, object],
*,
stage: str,
error: DocForgeError,
output_published: bool,
) -> dict[str, object]:
return {
"status": "ok",
**self._identity(snapshot),
"view_id": view.view_id,
"state": "degraded",
"publication": "published" if output_published else "partial",
"committed_stage": stage,
"plan_id": plan.plan_id,
"package_id": package_id,
"artifact": artifact,
"output": {
**artifact,
"path": view.output_path.relative_to(snapshot.descriptor.root).as_posix(),
"state": "unverified" if output_published else "not_published",
},
"receipt": receipt,
"manifest": {
"state": "failed",
"error": error.as_dict(),
},
}
def _manifest_status(
self,
view: GraphRenderView,
current: ProjectState | None,
) -> dict[str, object]:
manifest = self._read_manifest(view)
base = {
"view_id": view.view_id,
"renderer": GRAPH_RENDERER_ID,
"renderer_version": GRAPH_RENDERER_VERSION,
"path": view.output_path.relative_to(self.project.descriptor.root).as_posix(),
"verification": "manifest",
}
if manifest is None:
return {**base, "state": "missing", "reason": "manifest_missing"}
if not self._valid_manifest(view, manifest):
return {**base, "state": "unverified", "reason": "manifest_invalid"}
if current is None:
reason = (
"source_generation_changed"
if isinstance(self.project, GenerationRecordingProject)
else "source_generation_unavailable"
)
return {
**base,
"state": (
"stale"
if isinstance(self.project, GenerationRecordingProject)
else "unverified"
),
"reason": reason,
"plan_id": manifest.get("plan_id"),
"package_id": manifest.get("package_id"),
}
project = cast(dict[str, object], manifest["project"])
if project["revision"] != current.revision or project["source_hash"] != current.source_hash:
return {
**base,
"state": "stale",
"reason": "source_generation_changed",
"plan_id": manifest["plan_id"],
"package_id": manifest["package_id"],
}
artifact = cast(dict[str, object], manifest["artifact"])
store = cast(dict[str, object], manifest["store"])
artifact_root = self.project.descriptor.cache_root / "projection-artifacts"
if not artifact_root.exists():
return {
**base,
"state": "stale",
"reason": "artifact_store_missing",
"plan_id": manifest["plan_id"],
"package_id": manifest["package_id"],
}
if artifact_root.is_symlink() or not artifact_root.is_dir():
return {
**base,
"state": "unsafe",
"reason": "artifact_store_unsafe",
"plan_id": manifest["plan_id"],
"package_id": manifest["package_id"],
}
artifact_directory: int | None = None
try:
artifact_directory = open_confined_directory(
self.project.descriptor.root,
artifact_root,
create=False,
)
artifact_identity = safe_file_identity_at(
artifact_root,
artifact_directory,
f"{artifact['sha256']}.html",
)
except DocForgeError:
return {
**base,
"state": "unsafe",
"reason": "artifact_store_unsafe",
"plan_id": manifest["plan_id"],
"package_id": manifest["package_id"],
}
finally:
if artifact_directory is not None:
os.close(artifact_directory)
if artifact_identity is None:
return {
**base,
"state": "stale",
"reason": "artifact_store_missing",
"plan_id": manifest["plan_id"],
"package_id": manifest["package_id"],
}
if artifact_identity != store:
return {
**base,
"state": "stale",
"reason": "artifact_store_changed",
"plan_id": manifest["plan_id"],
"package_id": manifest["package_id"],
}
try:
directory = open_confined_directory(
self.project.descriptor.root,
view.output_path.parent,
create=False,
)
except DocForgeError:
return {**base, "state": "unsafe", "reason": "output_root_unsafe"}
try:
identity = safe_file_identity_at(
view.output_path.parent, directory, view.output_path.name
)
except DocForgeError:
return {**base, "state": "unsafe", "reason": "output_unsafe"}
finally:
os.close(directory)
expected = cast(dict[str, object], manifest["output"])
if identity != expected:
return {
**base,
"state": "stale",
"reason": "output_changed",
"plan_id": manifest["plan_id"],
"package_id": manifest["package_id"],
}
return {
**base,
"state": "current",
"reason": None,
"plan_id": manifest["plan_id"],
"package_id": manifest["package_id"],
"publication_id": manifest["publication_id"],
"artifact": manifest["artifact"],
}
def _read_manifest(self, view: GraphRenderView) -> dict[str, object] | None:
root = self._manifest_root()
if not root.is_dir() or root.is_symlink():
return None
try:
descriptor = open_confined_directory(
self.project.descriptor.root,
root,
create=False,
)
except DocForgeError:
return None
try:
raw = read_bounded_file_at(
descriptor,
f"{view.view_id}.json",
MAX_GRAPH_PUBLICATION_BYTES,
)
except DocForgeError:
return None
finally:
os.close(descriptor)
if raw is None:
return None
try:
value: object = json.loads(raw)
except (UnicodeDecodeError, json.JSONDecodeError):
return None
return cast(dict[str, object], value) if isinstance(value, dict) else None
def _valid_manifest(self, view: GraphRenderView, manifest: dict[str, object]) -> bool:
required = {
"schema_version",
"contract",
"publication_id",
"project",
"view_id",
"view_config_hash",
"plan_id",
"package_id",
"renderer",
"receipt",
"artifact",
"store",
"output",
}
try:
if (
set(manifest) != required
or manifest.get("schema_version") != GRAPH_PUBLICATION_MANIFEST_VERSION
or manifest.get("contract") != GRAPH_PUBLICATION_CONTRACT
or manifest.get("view_id") != view.view_id
or manifest.get("view_config_hash") != self._view_hash(view)
or not self._hash(manifest.get("plan_id"))
or not self._hash(manifest.get("package_id"))
):
return False
project = manifest.get("project")
descriptor = self.project.descriptor
if not isinstance(project, dict):
return False
project_document = cast(dict[str, object], project)
if (
set(project_document)
!= {
"project_id",
"project_root_fingerprint",
"adapter",
"revision",
"source_hash",
}
or project_document.get("project_id") != descriptor.project_id
or project_document.get("project_root_fingerprint")
!= project_root_fingerprint(descriptor.root)
or project_document.get("adapter") != descriptor.adapter
or not isinstance(project_document.get("revision"), str)
or not project_document["revision"]
or not self._hash(project_document.get("source_hash"))
):
return False
renderer = manifest.get("renderer")
if renderer != {
"renderer_id": GRAPH_RENDERER_ID,
"renderer_version": GRAPH_RENDERER_VERSION,
}:
return False
receipt_value = manifest.get("receipt")
if not isinstance(receipt_value, dict):
return False
receipt = ProjectionReceiptV1.from_dict(
dict(cast(dict[str, object], receipt_value))
).as_dict()
artifacts = receipt.get("artifacts")
if not isinstance(artifacts, list):
return False
artifact_values = cast(list[object], artifacts)
if (
receipt.get("kind") != "graph"
or receipt.get("plan_id") != manifest["plan_id"]
or receipt.get("package_id") != manifest["package_id"]
or receipt.get("renderer") != renderer
or len(artifact_values) != 1
or manifest.get("artifact") != artifact_values[0]
):
return False
artifact = artifact_values[0]
if not isinstance(artifact, dict):
return False
artifact_document = cast(dict[str, object], artifact)
if (
artifact_document.get("artifact_id") != "portable-graph.html"
or artifact_document.get("media_type") != "text/html; charset=utf-8"
):
return False
artifact_hash = artifact_document.get("sha256")
artifact_bytes = artifact_document.get("bytes")
store = manifest.get("store")
output = manifest.get("output")
if (
not self._file_identity(store, expected_name=f"{artifact_hash}.html")
or not self._file_identity(output, expected_name=view.output_path.name)
or type(artifact_bytes) is not int
or cast(dict[str, object], store)["size"] != artifact_bytes
or cast(dict[str, object], output)["size"] != artifact_bytes
):
return False
body = dict(manifest)
publication_id = body.pop("publication_id", None)
return self._hash(publication_id) and publication_id == projection_hash(body)
except (DocForgeError, KeyError, TypeError, ValueError):
return False
@staticmethod
def _hash(value: object) -> bool:
return (
isinstance(value, str)
and len(value) == 64
and all(character in "0123456789abcdef" for character in value)
)
@staticmethod
def _file_identity(value: object, *, expected_name: str) -> bool:
if not isinstance(value, dict):
return False
document = cast(dict[str, object], value)
required = {"path", "device", "inode", "mode", "size", "mtime_ns", "ctime_ns"}
return (
set(document) == required
and document.get("path") == expected_name
and all(
type(document.get(field)) is int and cast(int, document[field]) >= 0
for field in required - {"path"}
)
)
def _manifest(
self,
snapshot: ProjectSnapshot,
view: GraphRenderView,
plan: GraphViewPlanV1,
package_id: str,
receipt: dict[str, object],
artifact: dict[str, object],
store: dict[str, object],
output: dict[str, object],
) -> dict[str, object]:
body: dict[str, object] = {
"schema_version": GRAPH_PUBLICATION_MANIFEST_VERSION,
"contract": GRAPH_PUBLICATION_CONTRACT,
"project": self._identity(snapshot),
"view_id": view.view_id,
"view_config_hash": self._view_hash(view),
"plan_id": plan.plan_id,
"package_id": package_id,
"renderer": {
"renderer_id": GRAPH_RENDERER_ID,
"renderer_version": GRAPH_RENDERER_VERSION,
},
"receipt": receipt,
"artifact": artifact,
"store": store,
"output": output,
}
return {**body, "publication_id": projection_hash(body)}
def _publish_artifact(
self,
snapshot: ProjectSnapshot,
artifact_hash: object,
content: bytes,
*,
verify: Callable[[], None],
) -> dict[str, object]:
if not isinstance(artifact_hash, str):
raise DocForgeError("invalid_projection", "Artifact hash is invalid")
root = snapshot.descriptor.cache_root / "projection-artifacts"
descriptor = open_confined_directory(snapshot.descriptor.root, root, create=True)
name = f"{artifact_hash}.html"
try:
try:
existing = read_bounded_file_at(descriptor, name, len(content))
except DocForgeError as error:
if error.code != "invalid_projection":
raise
existing = None
if existing == content:
identity = safe_file_identity_at(root, descriptor, name)
assert identity is not None
return identity
return atomic_replace_bytes_at(root, descriptor, name, content, verify=verify)
finally:
os.close(descriptor)
def _publish_output(
self,
snapshot: ProjectSnapshot,
view: GraphRenderView,
content: bytes,
*,
verify: Callable[[], None],
) -> dict[str, object]:
root = view.output_path.parent
descriptor = open_confined_directory(snapshot.descriptor.root, root, create=True)
try:
try:
existing = read_bounded_file_at(
descriptor,
view.output_path.name,
len(content),
)
except DocForgeError as error:
if error.code != "invalid_projection":
raise
existing = None
if existing == content:
identity = safe_file_identity_at(root, descriptor, view.output_path.name)
assert identity is not None
return identity
return atomic_replace_bytes_at(
root,
descriptor,
view.output_path.name,
content,
verify=verify,
)
finally:
os.close(descriptor)
def _publish_manifest(
self,
snapshot: ProjectSnapshot,
view: GraphRenderView,
manifest: dict[str, object],
*,
verify: Callable[[], None],
) -> None:
raw = json.dumps(manifest, sort_keys=True, indent=2).encode() + b"\n"
if len(raw) > MAX_GRAPH_PUBLICATION_BYTES:
raise DocForgeError(
"projection_too_large",
"Portable graph publication manifest exceeds its fixed limit",
)
root = self._manifest_root()
descriptor = open_confined_directory(snapshot.descriptor.root, root, create=True)
try:
atomic_replace_bytes_at(
root,
descriptor,
f"{view.view_id}.json",
raw,
verify=verify,
)
finally:
os.close(descriptor)
@contextmanager
def _lock(self) -> Generator[None]:
root = self.project.descriptor.cache_root
descriptor = open_confined_directory(self.project.descriptor.root, root, create=True)
lock_descriptor: int | None = None
try:
lock_descriptor = os.open(
"graph-render.lock",
os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW,
0o600,
dir_fd=descriptor,
)
fcntl.flock(lock_descriptor, fcntl.LOCK_EX)
require_bound_directory(root, descriptor)
yield
except OSError as error:
raise DocForgeError(
"publication_failure",
"Portable graph render lock is unavailable",
) from error
finally:
if lock_descriptor is not None:
os.close(lock_descriptor)
os.close(descriptor)
def _plan(self, snapshot: ProjectSnapshot, view: GraphRenderView) -> GraphViewPlanV1:
return build_graph_view_plan(
snapshot,
GraphViewRequestV1(
view_id=view.view_id,
title=view.title,
root_node_id=view.root_node_id,
query=view.query,
initial_mode=view.initial_mode,
depth=view.depth,
max_nodes=view.max_nodes,
max_edges=view.max_edges,
max_work=view.max_work,
families=view.families,
relations=view.relations,
authorities=view.authorities,
statuses=view.statuses,
tags=view.tags,
include_logic=view.include_logic,
),
self.allow_logic,
)
def _current_state(self) -> ProjectState | None:
if isinstance(self.project, IncrementalStateProject):
return self.project.incremental_state()
return None
def _config(self, snapshot: ProjectSnapshot) -> GraphRenderConfig:
config = snapshot.descriptor.graph_render
if config is None:
raise DocForgeError(
"graph_render_not_configured",
"Project has no portable graph render configuration",
)
return config
@staticmethod
def _view(config: GraphRenderConfig, view_id: str) -> GraphRenderView:
for view in config.views:
if view.view_id == view_id:
return view
raise DocForgeError(
"unknown_graph_render_view",
"Portable graph view is not declared",
view_id=view_id,
)
def _manifest_root(self) -> Path:
return self.project.descriptor.cache_root / "projection-publications" / "graph"
@staticmethod
def _view_hash(view: GraphRenderView) -> str:
return projection_hash(
{
"view_id": view.view_id,
"renderer": view.renderer,
"title": view.title,
"root_node_id": view.root_node_id,
"query": view.query,
"initial_mode": view.initial_mode,
"depth": view.depth,
"max_nodes": view.max_nodes,
"max_edges": view.max_edges,
"max_work": view.max_work,
"families": list(view.families),
"relations": list(view.relations),
"authorities": list(view.authorities),
"statuses": list(view.statuses),
"tags": list(view.tags),
"include_logic": view.include_logic,
}
)
def _status_result(self, current: ProjectState | None, **payload: object) -> dict[str, object]:
descriptor = self.project.descriptor
return {
"status": "ok",
"project_id": descriptor.project_id,
"project_root_fingerprint": project_root_fingerprint(descriptor.root),
"adapter": descriptor.adapter,
"revision": current.revision if current is not None else "unknown",
"source_hash": current.source_hash if current is not None else None,
**payload,
}
@staticmethod
def _identity(snapshot: ProjectSnapshot) -> dict[str, object]:
return {
"project_id": snapshot.descriptor.project_id,
"project_root_fingerprint": project_root_fingerprint(snapshot.descriptor.root),
"adapter": snapshot.descriptor.adapter,
"revision": snapshot.revision,
"source_hash": snapshot.source_hash,
}

View file

@ -5,7 +5,7 @@ from __future__ import annotations
from collections.abc import Mapping
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Protocol, runtime_checkable
from typing import Literal, Protocol, runtime_checkable
@dataclass(frozen=True)
@ -49,6 +49,33 @@ class RenderConfig:
views: tuple[RenderView, ...]
@dataclass(frozen=True)
class GraphRenderView:
view_id: str
renderer: str
output_path: Path
title: str
root_node_id: str | None
query: str | None
initial_mode: Literal["nodes", "flow", "web", "logic"]
depth: int
max_nodes: int
max_edges: int
max_work: int
families: tuple[str, ...]
relations: tuple[str, ...]
authorities: tuple[str, ...]
statuses: tuple[str, ...]
tags: tuple[str, ...]
include_logic: bool
@dataclass(frozen=True)
class GraphRenderConfig:
output_root: Path
views: tuple[GraphRenderView, ...]
@dataclass(frozen=True)
class ContextProfile:
profile_id: str
@ -78,6 +105,7 @@ class ProjectDescriptor:
allowed_relations: tuple[str, ...]
profiles: tuple[ContextProfile, ...]
limits: Limits
graph_render: GraphRenderConfig | None = None
@dataclass(frozen=True)

View file

@ -25,6 +25,7 @@ from .config_validation import (
string_list,
)
from .errors import DocForgeError
from .graph_render_config import load_graph_render_config
from .models import (
ContextProfile,
Edge,
@ -66,6 +67,7 @@ _DESCRIPTOR_KEYS = frozenset(
"derived",
"changesets",
"render",
"graph_render",
"graph",
"limits",
"profiles",
@ -560,6 +562,18 @@ def _load_descriptor(root: Path) -> ProjectDescriptor:
changeset_root=changeset_root,
limits=limits,
)
graph_render = load_graph_render_config(
root,
document.get("graph_render"),
descriptor_path=descriptor_path,
content_roots=content_roots,
authority_files=authority_files,
cache_root=cache_root,
index_path=index_path,
changeset_root=changeset_root,
manual_render=render,
limits=limits,
)
profile_documents = document.get("profiles", [])
if not isinstance(profile_documents, list):
@ -623,6 +637,7 @@ def _load_descriptor(root: Path) -> ProjectDescriptor:
changeset_root=changeset_root,
proposal_writers=tuple(sorted(proposal_writers, key=lambda writer: writer.writer_id)),
render=render,
graph_render=graph_render,
allowed_relations=allowed_relations,
profiles=tuple(profiles),
limits=limits,

View file

@ -19,6 +19,13 @@ MAX_PLAN_BYTES = 16_000_000
MAX_PACKAGE_BYTES = 24_000_000
MAX_RECEIPT_BYTES = 128_000
MAX_PROJECTION_ARTIFACTS = 32
MAX_GRAPH_VIEW_DEPTH = 32
MAX_GRAPH_VIEW_NODES = 1_000
MAX_GRAPH_VIEW_EDGES = 4_000
MAX_GRAPH_VIEW_WORK = 1_000_000
MAX_GRAPH_VIEW_FILTERS = 64
MAX_GRAPH_VIEW_STRING_CHARS = 1_024
MAX_GRAPH_VIEW_QUERY_CHARS = 10_000
ProjectionKind = Literal["manual", "graph"]
@ -468,16 +475,39 @@ def validate_projection_receipt(document: dict[str, object]) -> dict[str, object
if not isinstance(artifacts_value, list):
raise DocForgeError("invalid_projection", "Projection receipt structure is invalid")
artifacts = cast(list[object], artifacts_value)
renderer = document.get("renderer")
diagnostics = document.get("diagnostics")
timing = document.get("timing")
if (
not isinstance(renderer, dict)
or not isinstance(diagnostics, dict)
or not isinstance(timing, dict)
):
raise DocForgeError("invalid_projection", "Projection receipt structure is invalid")
renderer_document = cast(dict[str, object], renderer)
diagnostics_document = cast(dict[str, object], diagnostics)
timing_document = cast(dict[str, object], timing)
warnings = diagnostics_document.get("warnings")
if (
len(artifacts) > MAX_PROJECTION_ARTIFACTS
or not isinstance(document.get("renderer"), dict)
or not isinstance(document.get("diagnostics"), dict)
or not isinstance(document.get("timing"), dict)
or set(renderer_document) != {"renderer_id", "renderer_version"}
or not all(
isinstance(renderer_document.get(field), str) and renderer_document[field]
for field in ("renderer_id", "renderer_version")
)
or set(diagnostics_document) != {"warnings"}
or not isinstance(warnings, list)
or len(cast(list[object], warnings)) > 10_000
or not all(isinstance(item, str) for item in cast(list[object], warnings))
or set(timing_document) != {"elapsed_ns"}
or type(timing_document.get("elapsed_ns")) is not int
or cast(int, timing_document["elapsed_ns"]) < 0
):
raise DocForgeError("invalid_projection", "Projection receipt structure is invalid")
peak = document.get("peak_memory_bytes")
if peak is not None and (type(peak) is not int or peak < 0):
raise DocForgeError("invalid_projection", "Projection receipt memory value is invalid")
artifact_ids: set[str] = set()
for artifact in artifacts:
if not isinstance(artifact, dict):
raise DocForgeError("invalid_projection", "Projection receipt artifact is invalid")
@ -494,6 +524,10 @@ def validate_projection_receipt(document: dict[str, object]) -> dict[str, object
or cast(int, item["bytes"]) < 0
):
raise DocForgeError("invalid_projection", "Projection receipt artifact is invalid")
artifact_id = cast(str, item["artifact_id"])
if artifact_id in artifact_ids:
raise DocForgeError("invalid_projection", "Projection receipt artifacts are duplicated")
artifact_ids.add(artifact_id)
return _validated_identity(
document,
identity_field="receipt_id",

View file

@ -117,6 +117,9 @@ OPERATION_NAMES = frozenset(
"cli.impact",
"cli.context",
"cli.generation-diff",
"cli.graph-plan",
"cli.graph-render",
"cli.graph-render-status",
"cli.configure",
"cli.doctor",
"cli.render",

View file

@ -0,0 +1,321 @@
"""Deterministic self-contained renderer for one portable graph package."""
from __future__ import annotations
import base64
import hashlib
import html
import json
from time import perf_counter_ns
from typing import cast
from docforge.errors import DocForgeError
from docforge.projection_contract import (
ProjectionArtifact,
ProjectionPackageV1,
ProjectionReceiptV1,
ProjectionRenderResult,
)
PORTABLE_GRAPH_CSS = """
:root { color-scheme: light dark; font-family: system-ui, sans-serif; }
* { box-sizing: border-box; }
body { margin: 0; background: Canvas; color: CanvasText; }
.skip { position: absolute; left: -10000px; top: auto; }
.skip:focus { left: 1rem; top: 1rem; z-index: 2; padding: .5rem; background: Canvas; }
header, main { width: min(96%, 1100px); margin: 0 auto; }
header { padding: 1rem 0; }
.controls { display: flex; flex-wrap: wrap; gap: .75rem; align-items: end; }
label { display: grid; gap: .25rem; font-weight: 600; }
input, select, button { font: inherit; min-height: 2.75rem; padding: .45rem .65rem; }
button { cursor: pointer; }
button:focus-visible, input:focus-visible, select:focus-visible { outline: .2rem solid Highlight; }
.summary { margin: 1rem 0; }
.layout { display: grid; grid-template-columns: minmax(16rem, 1fr) minmax(20rem, 2fr); gap: 1rem; }
.panel { border: 1px solid GrayText; border-radius: .5rem; padding: 1rem; overflow: auto; }
html[data-enhanced="true"] main[data-mode="nodes"] .layout,
html[data-enhanced="true"] main[data-mode="flow"] .layout { grid-template-columns: 1fr; }
html[data-enhanced="true"] main[data-mode="nodes"] [data-panel="relationships"] { display: none; }
html[data-enhanced="true"] main[data-mode="flow"] [data-panel="nodes"] { display: none; }
.node-list { list-style: none; padding: 0; margin: 0; display: grid; gap: .5rem; }
.node-list button {
width: 100%; text-align: left; border: 1px solid GrayText; border-radius: .35rem;
}
.node-list button[aria-current="true"] { border-width: .2rem; }
table { border-collapse: collapse; width: 100%; }
th, td { text-align: left; border-bottom: 1px solid GrayText; padding: .5rem; vertical-align: top; }
caption { text-align: left; font-weight: 700; margin-bottom: .5rem; }
.muted { color: GrayText; }
dialog {
max-width: min(42rem, calc(100% - 2rem));
border: 1px solid GrayText; border-radius: .5rem;
}
dialog::backdrop { background: rgb(0 0 0 / 55%); }
@media (max-width: 48rem) { .layout { grid-template-columns: 1fr; } }
@media (prefers-reduced-motion: reduce) {
*, *::before, *::after { scroll-behavior: auto !important; }
}
@media (forced-colors: active) {
.panel, .node-list button, dialog { border: 2px solid CanvasText; }
}
""".strip()
PORTABLE_GRAPH_JAVASCRIPT = r"""
(() => {
"use strict";
const plan = JSON.parse(document.getElementById("docforge-graph-plan").textContent);
const nodes = plan.graph.nodes;
const edges = plan.graph.edges;
const list = document.getElementById("node-list");
const rows = document.getElementById("edge-rows");
const filter = document.getElementById("filter");
const mode = document.getElementById("mode");
const main = document.getElementById("main");
const status = document.getElementById("status");
const dialog = document.getElementById("node-dialog");
const detail = document.getElementById("node-detail");
const close = document.getElementById("close-dialog");
let opener = null;
document.documentElement.dataset.enhanced = "true";
const matches = (node) => {
const query = filter.value.trim().toLocaleLowerCase();
const fields = [
node.node_id, node.title, node.summary, node.family, node.status, ...node.tags
];
return !query || fields
.join(" ").toLocaleLowerCase().includes(query);
};
const selectedIds = () => new Set(nodes.filter(matches).map((node) => node.node_id));
const render = () => {
main.dataset.mode = mode.value;
const visible = nodes.filter(matches);
const ids = selectedIds();
list.replaceChildren(...visible.map((node) => {
const item = document.createElement("li");
const button = document.createElement("button");
button.type = "button";
button.textContent = `${node.title} (${node.node_id})`;
button.dataset.nodeId = node.node_id;
button.addEventListener("click", () => inspect(node, button));
item.append(button);
return item;
}));
const visibleEdges = edges.filter((edge) => ids.has(edge.source_id) && ids.has(edge.target_id));
rows.replaceChildren(...visibleEdges.map((edge) => {
const row = document.createElement("tr");
[edge.source_id, edge.relation, edge.target_id].forEach((value) => {
const cell = document.createElement("td");
cell.textContent = value;
row.append(cell);
});
return row;
}));
status.textContent = `${visible.length} nodes and ${visibleEdges.length} relationships `
+ `shown in ${mode.value} mode.`;
};
const inspect = (node, button) => {
opener = button;
detail.replaceChildren();
const heading = document.createElement("h2");
heading.id = "node-dialog-title";
heading.textContent = node.title;
const identity = document.createElement("p");
identity.textContent = `${node.node_id} · ${node.family} · ${node.status}`;
const summary = document.createElement("p");
summary.textContent = node.summary;
detail.append(heading, identity, summary);
dialog.showModal();
close.focus();
};
close.addEventListener("click", () => dialog.close());
dialog.addEventListener("close", () => opener?.focus());
filter.addEventListener("input", render);
mode.addEventListener("change", render);
render();
})();
""".strip()
def _csp_hash(content: str) -> str:
digest = hashlib.sha256(content.encode("utf-8")).digest()
return base64.b64encode(digest).decode("ascii")
def _embedded_json(value: object) -> str:
return (
json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
.replace("&", "\\u0026")
.replace("<", "\\u003c")
.replace(">", "\\u003e")
)
def _static_node_markup(plan: dict[str, object]) -> str:
graph = cast(dict[str, object], plan["graph"])
nodes = cast(list[dict[str, object]], graph["nodes"])
parts: list[str] = []
for node in nodes:
node_id = html.escape(cast(str, node["node_id"]))
attribute_node_id = html.escape(cast(str, node["node_id"]), quote=True)
title = html.escape(cast(str, node["title"]))
family = html.escape(cast(str, node["family"]))
status = html.escape(cast(str, node["status"]))
summary = html.escape(cast(str, node["summary"]))
parts.append(
"<li><article>"
f'<button type="button" data-node-id="{attribute_node_id}">'
f"{title} ({node_id})</button>"
f'<p class="muted">{family} · {status}</p>'
f"<p>{summary}</p>"
"</article></li>"
)
return "".join(parts)
def _static_edge_markup(plan: dict[str, object]) -> str:
graph = cast(dict[str, object], plan["graph"])
edges = cast(list[dict[str, object]], graph["edges"])
return "".join(
"<tr>"
f"<td>{html.escape(cast(str, edge['source_id']))}</td>"
f"<td>{html.escape(cast(str, edge['relation']))}</td>"
f"<td>{html.escape(cast(str, edge['target_id']))}</td>"
"</tr>"
for edge in edges
)
class PortableGraphHtmlRenderer:
"""Render a validated graph package without querying or publishing project state."""
renderer_id = "portable_graph_html"
renderer_version = "1"
def render(self, package: ProjectionPackageV1) -> ProjectionRenderResult:
started = perf_counter_ns()
package = ProjectionPackageV1.from_dict(package.as_dict())
document = package.document
if package.kind != "graph":
raise DocForgeError("invalid_projection", "Graph renderer requires a graph package")
renderer = cast(dict[str, object], document["renderer"])
if renderer != {
"renderer_id": self.renderer_id,
"renderer_version": self.renderer_version,
}:
raise DocForgeError("unsupported_renderer", "Graph renderer identity is incompatible")
if document["components"] != [
{"component_id": "graph.portable-document@1"},
{"component_id": "graph.accessible-list@1"},
{"component_id": "graph.relationship-table@1"},
]:
raise DocForgeError(
"invalid_projection",
"Portable graph renderer component declarations are incompatible",
)
if document["assets"] != []:
raise DocForgeError(
"invalid_projection",
"Portable graph renderer does not accept project-provided assets",
)
plan = cast(dict[str, object], document["plan"])
view = cast(dict[str, object], plan["view"])
initial_mode = view.get("initial_mode")
policy = cast(dict[str, object], plan["policy"])
if (
initial_mode not in {"nodes", "flow", "web"}
or policy.get("logic_requested") is not False
):
raise DocForgeError(
"unsupported_renderer",
"Portable graph renderer version 1 does not render Logic projections",
)
title = html.escape(cast(str, view["title"]))
project = cast(dict[str, object], plan["project"])
embedded = _embedded_json(plan)
static_nodes = _static_node_markup(plan)
static_edges = _static_edge_markup(plan)
csp = (
"default-src 'none'; "
f"style-src 'sha256-{_csp_hash(PORTABLE_GRAPH_CSS)}'; "
f"script-src 'sha256-{_csp_hash(PORTABLE_GRAPH_JAVASCRIPT)}'; "
"img-src 'none'; connect-src 'none'; object-src 'none'; base-uri 'none'; "
"form-action 'none'; frame-ancestors 'none'"
)
output = (
"<!DOCTYPE html>\n"
'<html lang="en">\n'
"<head>\n"
'<meta charset="utf-8">\n'
'<meta name="viewport" content="width=device-width, initial-scale=1">\n'
'<meta http-equiv="Content-Security-Policy" '
f'content="{html.escape(csp, quote=True)}">\n'
f"<title>{title} · DocForge graph</title>\n"
f"<style>{PORTABLE_GRAPH_CSS}</style>\n"
"</head>\n"
"<body>\n"
'<a class="skip" href="#main">Skip to graph content</a>\n'
"<header>\n"
f"<h1>{title}</h1>\n"
f'<p class="muted">Generation {html.escape(cast(str, project["source_hash"]))}</p>\n'
'<div class="controls" role="group" aria-label="Graph controls">\n'
'<label>Filter nodes<input id="filter" type="search" autocomplete="off"></label>\n'
'<label>View mode<select id="mode">'
f'<option value="nodes"{" selected" if initial_mode == "nodes" else ""}>Nodes</option>'
f'<option value="flow"{" selected" if initial_mode == "flow" else ""}>Flow</option>'
f'<option value="web"{" selected" if initial_mode == "web" else ""}>Web</option>'
"</select></label>\n"
"</div>\n"
'<p id="status" class="summary" role="status" aria-live="polite"></p>\n'
"</header>\n"
f'<main id="main" tabindex="-1" data-mode="{initial_mode}">\n'
'<div class="layout">\n'
'<section class="panel" data-panel="nodes" aria-labelledby="nodes-title">'
'<h2 id="nodes-title">Nodes</h2>'
f'<ul id="node-list" class="node-list">{static_nodes}</ul></section>\n'
'<section class="panel" data-panel="relationships" aria-labelledby="relations-title">'
'<h2 id="relations-title">Relationships</h2>'
"<table><caption>Selected graph facts</caption><thead><tr>"
'<th scope="col">Source</th><th scope="col">Relation</th>'
f'<th scope="col">Target</th></tr></thead><tbody id="edge-rows">{static_edges}</tbody>'
"</table>"
"</section>\n"
"</div>\n"
"</main>\n"
'<dialog id="node-dialog" aria-labelledby="node-dialog-title">'
'<div id="node-detail"><h2 id="node-dialog-title">Node details</h2></div>'
'<button id="close-dialog" type="button">Close</button></dialog>\n'
f'<script id="docforge-graph-plan" type="application/json">{embedded}</script>\n'
f"<script>{PORTABLE_GRAPH_JAVASCRIPT}</script>\n"
"</body>\n"
"</html>\n"
).encode()
policy = cast(dict[str, object], document["output_policy"])
maximum = policy.get("max_total_bytes")
if set(policy) != {"artifact_ids", "max_total_bytes"} or policy.get("artifact_ids") != [
"portable-graph.html"
]:
raise DocForgeError(
"invalid_projection",
"Portable graph output policy is incompatible",
)
if type(maximum) is not int or maximum < 1 or len(output) > maximum:
raise DocForgeError("render_too_large", "Rendered output exceeds the configured limit")
artifact = ProjectionArtifact(
artifact_id="portable-graph.html",
media_type="text/html; charset=utf-8",
content=output,
)
receipt = ProjectionReceiptV1.create(
kind="graph",
package_id=package.package_id,
plan_id=cast(str, document["plan_id"]),
renderer=dict(renderer),
artifacts=[artifact.evidence()],
diagnostics={
"warnings": [],
},
timing={"elapsed_ns": perf_counter_ns() - started},
peak_memory_bytes=None,
)
return ProjectionRenderResult((artifact,), receipt)