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

Add versioned independent projection contracts

Browse source
This commit is contained in:
Andraxion 2026-07-29 10:50:05 -04:00
parent 4c5773c865
commit 96e3965855
22 changed files with 3561 additions and 133 deletions
Download patch file Download diff file Expand all files Collapse all files

497
src/docforge/graph_projection.py Normal file
View file

@ -0,0 +1,497 @@
"""Pure, bounded portable-graph planning over one immutable graph generation."""
from __future__ import annotations
import re
from dataclasses import dataclass
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
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",
"title",
"family",
"authority",
"status",
"tags",
"summary",
"content_hash",
)
@dataclass(frozen=True)
class GraphViewRequestV1:
"""One closed, inert graph selection request."""
view_id: str
title: str
root_node_id: str | None = None
query: str | None = None
initial_mode: GraphViewMode = "nodes"
depth: int = 1
max_nodes: int = 100
max_edges: int = 400
max_work: int = 100_000
families: tuple[str, ...] = ()
relations: tuple[str, ...] = ()
authorities: tuple[str, ...] = ()
statuses: tuple[str, ...] = ()
tags: tuple[str, ...] = ()
include_logic: bool = False
@dataclass(frozen=True)
class _ValidatedRequest:
view_id: str
title: str
root_node_id: str | None
query: str | None
initial_mode: GraphViewMode
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
def _invalid(message: str, **details: object) -> DocForgeError:
return DocForgeError("invalid_graph_view_request", message, **details)
def _string(value: object, *, field: str, maximum: int = MAX_GRAPH_VIEW_STRING_CHARS) -> str:
if not isinstance(value, str) or not value.strip() or len(value) > maximum or "\0" in value:
raise _invalid("Graph view request string is invalid", field=field)
return value.strip()
def _filter_values(values: object, *, field: str) -> tuple[str, ...]:
if not isinstance(values, tuple):
raise _invalid("Graph view filter is invalid", field=field)
tuple_values = cast(tuple[object, ...], values)
if not all(isinstance(value, str) for value in tuple_values):
raise _invalid("Graph view filter is invalid", field=field)
if len(tuple_values) > MAX_GRAPH_VIEW_FILTERS:
raise _invalid("Graph view filter is invalid", field=field)
normalized = tuple(_string(value, field=field) for value in cast(tuple[str, ...], tuple_values))
if len(normalized) != len(set(normalized)):
raise _invalid("Graph view filter contains duplicates", field=field)
return tuple(sorted(normalized))
def _validated_request(request: GraphViewRequestV1) -> _ValidatedRequest:
view_id = _string(request.view_id, field="view_id")
title = _string(request.title, field="title")
if (request.root_node_id is None) == (request.query is None):
raise _invalid("Choose exactly one exact root or lexical query")
root_node_id = (
None
if request.root_node_id is None
else _string(request.root_node_id, field="root_node_id")
)
query = (
None
if request.query is None
else _string(
request.query,
field="query",
maximum=MAX_GRAPH_VIEW_QUERY_CHARS,
)
)
if request.initial_mode not in {"nodes", "flow", "web", "logic"}:
raise _invalid("Graph view initial mode is unsupported")
if type(request.depth) is not int or not 1 <= request.depth <= MAX_GRAPH_VIEW_DEPTH:
raise _invalid(
"Graph view depth is outside the fixed boundary",
maximum=MAX_GRAPH_VIEW_DEPTH,
)
for field, value, minimum, maximum in (
("max_nodes", request.max_nodes, 1, MAX_GRAPH_VIEW_NODES),
("max_edges", request.max_edges, 0, MAX_GRAPH_VIEW_EDGES),
("max_work", request.max_work, 1, MAX_GRAPH_VIEW_WORK),
):
if type(value) is not int or not minimum <= value <= maximum:
raise _invalid(
"Graph view bound is outside the fixed boundary",
field=field,
minimum=minimum,
maximum=maximum,
)
if type(request.include_logic) is not bool:
raise _invalid("Graph view Logic selection must be Boolean")
return _ValidatedRequest(
view_id=view_id,
title=title,
root_node_id=root_node_id,
query=query,
initial_mode=request.initial_mode,
depth=request.depth,
max_nodes=request.max_nodes,
max_edges=request.max_edges,
max_work=request.max_work,
families=_filter_values(request.families, field="families"),
relations=_filter_values(request.relations, field="relations"),
authorities=_filter_values(request.authorities, field="authorities"),
statuses=_filter_values(request.statuses, field="statuses"),
tags=_filter_values(request.tags, field="tags"),
include_logic=request.include_logic,
)
def _validated_graph(
snapshot: ProjectSnapshot,
) -> tuple[tuple[Node, ...], tuple[Edge, ...], dict[str, Node]]:
nodes = tuple(sorted(snapshot.nodes, key=lambda node: node.node_id))
node_by_id = {node.node_id: node for node in nodes}
if len(node_by_id) != len(nodes):
raise DocForgeError(
"invalid_projection",
"Graph view snapshot contains duplicate node identities",
)
edges = tuple(
sorted(
snapshot.edges,
key=lambda edge: (edge.source_id, edge.relation, edge.target_id),
)
)
edge_keys = {(edge.source_id, edge.relation, edge.target_id) for edge in edges}
if len(edge_keys) != len(edges) or any(
edge.source_id not in node_by_id or edge.target_id not in node_by_id for edge in edges
):
raise DocForgeError(
"invalid_projection",
"Graph view snapshot contains invalid relationships",
)
return nodes, edges, node_by_id
def _eligible(node: Node, request: _ValidatedRequest) -> bool:
return (
(not request.families or node.family in request.families)
and (not request.authorities or node.authority in request.authorities)
and (not request.statuses or node.status in request.statuses)
and (not request.tags or set(request.tags).issubset(node.tags))
)
def _relation_allowed(edge: Edge, request: _ValidatedRequest) -> bool:
return not request.relations or edge.relation in request.relations
def _lexical_text(node: Node) -> str:
return " ".join(
(
node.node_id,
node.title,
node.summary,
node.family,
node.authority,
node.status,
*node.tags,
)
).casefold()
def _lexical_nodes(
nodes: tuple[Node, ...],
request: _ValidatedRequest,
omissions: list[dict[str, object]],
) -> tuple[set[str], int, bool]:
query = request.query
maximum_nodes = request.max_nodes
maximum_work = request.max_work
assert query is not None
terms = tuple(dict.fromkeys(_QUERY_TOKEN.findall(query.casefold())))
if not terms:
raise _invalid("Lexical graph scope contains no searchable text")
selected: set[str] = set()
work = 0
work_limited = False
for node in nodes:
if work >= maximum_work:
work_limited = True
break
work += 1
if not _eligible(node, request) or not all(term in _lexical_text(node) for term in terms):
continue
if len(selected) >= maximum_nodes:
omissions.append(
{
"code": "node_result_limit",
"subject": "nodes",
"limit": maximum_nodes,
"minimum_omitted": 1,
}
)
break
selected.add(node.node_id)
return selected, work, work_limited
def _root_nodes(
edges: tuple[Edge, ...],
node_by_id: dict[str, Node],
request: _ValidatedRequest,
omissions: list[dict[str, object]],
) -> tuple[set[str], int, bool]:
root_node_id = request.root_node_id
maximum_nodes = request.max_nodes
maximum_work = request.max_work
depth = request.depth
assert root_node_id is not None
root = node_by_id.get(root_node_id)
if root is None:
raise DocForgeError(
"missing_node",
"No node has the requested stable ID",
node_id=root_node_id,
)
if not _eligible(root, request):
raise _invalid("Exact graph root is excluded by the closed node filters")
selected = {root_node_id}
frontier = {root_node_id}
work = 0
work_limited = False
node_limited = False
for _ in range(depth):
if not frontier:
break
next_frontier: set[str] = set()
for edge in edges:
if work >= maximum_work:
work_limited = True
break
work += 1
if not _relation_allowed(edge, request):
continue
candidate: str | None = None
if edge.source_id in frontier:
candidate = edge.target_id
elif edge.target_id in frontier:
candidate = edge.source_id
if candidate is None or candidate in selected:
continue
node = node_by_id[candidate]
if not _eligible(node, request):
continue
if len(selected) >= maximum_nodes:
node_limited = True
continue
selected.add(candidate)
next_frontier.add(candidate)
if work_limited:
break
frontier = next_frontier
if node_limited:
omissions.append(
{
"code": "node_result_limit",
"subject": "nodes",
"limit": maximum_nodes,
"minimum_omitted": 1,
}
)
return selected, work, work_limited
def _selected_edges(
edges: tuple[Edge, ...],
selected_ids: set[str],
request: _ValidatedRequest,
*,
initial_work: int,
omissions: list[dict[str, object]],
) -> tuple[list[Edge], int, bool]:
maximum_edges = request.max_edges
maximum_work = request.max_work
selected: list[Edge] = []
work = initial_work
work_limited = False
edge_limited = False
for edge in edges:
if work >= maximum_work:
work_limited = True
break
work += 1
if (
edge.source_id not in selected_ids
or edge.target_id not in selected_ids
or not _relation_allowed(edge, request)
):
continue
if len(selected) >= maximum_edges:
edge_limited = True
break
selected.append(edge)
if edge_limited:
omissions.append(
{
"code": "edge_result_limit",
"subject": "edges",
"limit": maximum_edges,
"minimum_omitted": 1,
}
)
return selected, work, work_limited
def _node_payload(node: Node) -> dict[str, object]:
return {
"node_id": node.node_id,
"title": node.title,
"family": node.family,
"authority": node.authority,
"status": node.status,
"tags": sorted(node.tags),
"summary": node.summary,
"content_hash": node.content_hash,
}
def _edge_payload(edge: Edge) -> dict[str, str]:
return {
"source_id": edge.source_id,
"relation": edge.relation,
"target_id": edge.target_id,
}
def build_graph_view_plan(
snapshot: ProjectSnapshot,
request: GraphViewRequestV1,
allow_logic: bool,
) -> GraphViewPlanV1:
"""Build one deterministic, path-free graph plan without rendering or storage access."""
if type(allow_logic) is not bool:
raise _invalid("Graph view Logic policy must be Boolean")
normalized = _validated_request(request)
nodes, edges, node_by_id = _validated_graph(snapshot)
omissions: list[dict[str, object]] = []
if normalized.root_node_id is not None:
selected_ids, work, work_limited = _root_nodes(
edges,
node_by_id,
normalized,
omissions,
)
scope: dict[str, object] = {
"kind": "exact_root",
"root_node_id": normalized.root_node_id,
"depth": normalized.depth,
}
else:
selected_ids, work, work_limited = _lexical_nodes(
nodes,
normalized,
omissions,
)
scope = {
"kind": "lexical",
"query": normalized.query,
}
selected_edges, work, edge_work_limited = _selected_edges(
edges,
selected_ids,
normalized,
initial_work=work,
omissions=omissions,
)
work_limited = work_limited or edge_work_limited
if work_limited:
omissions.append(
{
"code": "work_limit",
"subject": "selection",
"limit": normalized.max_work,
"examined": work,
"minimum_omitted": 1,
}
)
logic_requested = normalized.include_logic
if logic_requested and not allow_logic:
omissions.append(
{
"code": "logic_forbidden",
"subject": "logic",
"minimum_omitted": 1,
}
)
omissions.sort(key=lambda item: (str(item["code"]), str(item["subject"])))
selected_nodes = [node_by_id[node_id] for node_id in sorted(selected_ids)]
filters: dict[str, list[str]] = {
"families": list(normalized.families),
"relations": list(normalized.relations),
"authorities": list(normalized.authorities),
"statuses": list(normalized.statuses),
"tags": list(normalized.tags),
}
return GraphViewPlanV1.create(
{
"project": {
"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": {
"view_id": normalized.view_id,
"title": normalized.title,
"initial_mode": normalized.initial_mode,
"scope": scope,
"filters": filters,
"detail_fields": list(_DETAIL_FIELDS),
},
"bounds": {
"depth": normalized.depth,
"max_nodes": normalized.max_nodes,
"max_edges": normalized.max_edges,
"max_work": normalized.max_work,
},
"policy": {
"visibility": "selected_graph_only",
"source_paths": "excluded",
"source_bodies": "excluded",
"database_queries": "forbidden",
"executable_content": "forbidden",
"logic": "allowed" if allow_logic else "forbidden",
"logic_requested": logic_requested,
},
"graph": {
"root_node_id": normalized.root_node_id,
"nodes": [_node_payload(node) for node in selected_nodes],
"edges": [_edge_payload(edge) for edge in selected_edges],
"logic_projections": [],
},
"omissions": omissions,
"diagnostics": {
"selection": scope["kind"],
"returned_nodes": len(selected_nodes),
"returned_edges": len(selected_edges),
"examined_work_units": work,
"truncated": bool(omissions),
"ordering": "node_id;source_id,relation,target_id",
},
}
)

191
src/docforge/manual_projection.py Normal file
View file

@ -0,0 +1,191 @@
"""Pure manual planning over one immutable validated graph generation."""
from __future__ import annotations
import hashlib
from collections import defaultdict
from .errors import DocForgeError
from .models import Edge, ProjectSnapshot, RenderView
from .project import project_root_fingerprint
from .projection_contract import ManualRenderPlanV1, ProjectionPackageV1
def _edge_dict(edge: Edge) -> dict[str, str]:
return {
"source_id": edge.source_id,
"relation": edge.relation,
"target_id": edge.target_id,
}
def _cycles(node_ids: tuple[str, ...], edges: tuple[Edge, ...]) -> list[list[str]]:
"""Return deterministic strongly connected components that represent cycles."""
adjacency: dict[str, list[str]] = {node_id: [] for node_id in node_ids}
for edge in edges:
adjacency[edge.source_id].append(edge.target_id)
for targets in adjacency.values():
targets.sort()
index = 0
indexes: dict[str, int] = {}
lowlinks: dict[str, int] = {}
stack: list[str] = []
on_stack: set[str] = set()
components: list[list[str]] = []
def visit(node_id: str) -> None:
nonlocal index
indexes[node_id] = index
lowlinks[node_id] = index
index += 1
stack.append(node_id)
on_stack.add(node_id)
for target_id in adjacency[node_id]:
if target_id not in indexes:
visit(target_id)
lowlinks[node_id] = min(lowlinks[node_id], lowlinks[target_id])
elif target_id in on_stack:
lowlinks[node_id] = min(lowlinks[node_id], indexes[target_id])
if lowlinks[node_id] != indexes[node_id]:
return
component: list[str] = []
while stack:
member = stack.pop()
on_stack.remove(member)
component.append(member)
if member == node_id:
break
component.sort()
if len(component) > 1 or component[0] in adjacency[component[0]]:
components.append(component)
for node_id in node_ids:
if node_id not in indexes:
visit(node_id)
return sorted(components)
def build_manual_render_plan(
snapshot: ProjectSnapshot,
view: RenderView,
*,
changeset_hash: str | None,
) -> ManualRenderPlanV1:
"""Select and describe a complete manual without rendering markup."""
selected = tuple(
node for node in snapshot.nodes if not view.families or node.family in view.families
)
selected_ids = {node.node_id for node in selected}
edges = tuple(
edge
for edge in snapshot.edges
if edge.source_id in selected_ids and edge.target_id in selected_ids
)
outgoing: dict[str, list[Edge]] = defaultdict(list)
incoming: dict[str, list[Edge]] = defaultdict(list)
for edge in edges:
outgoing[edge.source_id].append(edge)
incoming[edge.target_id].append(edge)
for values in (*outgoing.values(), *incoming.values()):
values.sort(key=lambda edge: (edge.source_id, edge.relation, edge.target_id))
pages = [
{
"node_id": node.node_id,
"title": node.title,
"family": node.family,
"authority": node.authority,
"status": node.status,
"tags": list(node.tags),
"summary": node.summary,
"content": node.content,
"content_hash": node.content_hash,
"components": [
"manual.node-metadata@1",
"manual.summary@1",
"manual.commonmark@1",
"manual.relationships@1",
],
"breadcrumbs": [],
"cross_references": [_edge_dict(edge) for edge in outgoing[node.node_id]],
"backlinks": [_edge_dict(edge) for edge in incoming[node.node_id]],
}
for node in selected
]
node_ids = tuple(node.node_id for node in selected)
connected = {endpoint for edge in edges for endpoint in (edge.source_id, edge.target_id)}
return ManualRenderPlanV1.create(
{
"project": {
"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": {
"view_id": view.view_id,
"title": view.title,
"families": list(view.families),
"renderer": view.renderer,
},
"changeset_hash": changeset_hash,
"pages": pages,
"navigation": [{"node_id": node.node_id, "title": node.title} for node in selected],
"search_documents": [
{
"node_id": node.node_id,
"title": node.title,
"summary": node.summary,
"family": node.family,
"status": node.status,
"tags": list(node.tags),
}
for node in selected
],
"diagnostics": {
"orphans": [node_id for node_id in node_ids if node_id not in connected],
"cycles": _cycles(node_ids, edges),
},
}
)
def build_manual_projection_package(
plan: ManualRenderPlanV1,
template_bytes: bytes,
*,
renderer_id: str,
renderer_version: str,
max_output_bytes: int,
) -> ProjectionPackageV1:
"""Bind one plan and inert template asset for a path-free manual renderer."""
try:
template = template_bytes.decode("utf-8")
except UnicodeDecodeError as error:
raise DocForgeError("invalid_template", "Render template is not valid UTF-8") from error
return ProjectionPackageV1.create(
kind="manual",
plan=plan,
renderer={"renderer_id": renderer_id, "renderer_version": renderer_version},
components=[
{"component_id": "manual.document@1"},
{"component_id": "manual.commonmark@1"},
],
assets=[
{
"asset_id": "manual.template",
"media_type": "text/html; charset=utf-8",
"sha256": hashlib.sha256(template_bytes).hexdigest(),
"text": template,
}
],
output_policy={
"artifact_ids": ["manual.html"],
"max_total_bytes": max_output_bytes,
},
)

502
src/docforge/projection_contract.py Normal file
View file

@ -0,0 +1,502 @@
"""Versioned, canonical contracts shared by independent projection renderers."""
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass
from typing import Literal, cast
from .errors import DocForgeError
MANUAL_RENDER_PLAN_CONTRACT = "docforge.manual-render-plan"
GRAPH_VIEW_PLAN_CONTRACT = "docforge.graph-view-plan"
PROJECTION_PACKAGE_CONTRACT = "docforge.projection-package"
PROJECTION_RECEIPT_CONTRACT = "docforge.projection-receipt"
PROJECTION_SCHEMA_VERSION = 1
MAX_PLAN_BYTES = 16_000_000
MAX_PACKAGE_BYTES = 24_000_000
MAX_RECEIPT_BYTES = 128_000
MAX_PROJECTION_ARTIFACTS = 32
ProjectionKind = Literal["manual", "graph"]
def canonical_projection_bytes(value: object) -> bytes:
"""Return the one canonical UTF-8 representation used for projection identities."""
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode(
"utf-8"
)
def projection_hash(value: object) -> str:
return hashlib.sha256(canonical_projection_bytes(value)).hexdigest()
def _is_hash(value: object) -> bool:
return (
isinstance(value, str)
and len(value) == 64
and all(character in "0123456789abcdef" for character in value)
)
def _bounded(document: dict[str, object], maximum: int, *, kind: str) -> None:
size = len(canonical_projection_bytes(document))
if size > maximum:
raise DocForgeError(
"projection_too_large",
f"{kind} exceeds its fixed serialized-size limit",
maximum_bytes=maximum,
actual_bytes=size,
)
def _validated_identity(
document: dict[str, object],
*,
identity_field: str,
maximum: int,
kind: str,
) -> dict[str, object]:
_bounded(document, maximum, kind=kind)
identity = document.get(identity_field)
if not _is_hash(identity):
raise DocForgeError("invalid_projection", f"{kind} identity is invalid")
body = dict(document)
body.pop(identity_field)
if projection_hash(body) != identity:
raise DocForgeError("invalid_projection", f"{kind} identity does not match its content")
return document
def _reject_runtime_authority(value: object) -> None:
"""Reject structural capabilities while treating selected content as inert data."""
forbidden_keys = {
"command",
"database",
"database_path",
"index_path",
"project_root",
"project_path",
"sql",
}
if isinstance(value, dict):
payload = cast(dict[object, object], value)
for key, item in payload.items():
if isinstance(key, str) and key in forbidden_keys:
raise DocForgeError(
"invalid_projection",
"Projection package contains forbidden runtime authority",
field=key,
)
if (
isinstance(key, str)
and (key == "path" or key.endswith("_path"))
and isinstance(item, str)
and item.startswith("/")
):
raise DocForgeError(
"invalid_projection",
"Projection package contains an absolute runtime path",
field=key,
)
_reject_runtime_authority(item)
elif isinstance(value, list):
for item in cast(list[object], value):
_reject_runtime_authority(item)
@dataclass(frozen=True)
class ManualRenderPlanV1:
"""One immutable, bounded manual plan prepared from a validated graph generation."""
document: dict[str, object]
@property
def plan_id(self) -> str:
return cast(str, self.document["plan_id"])
def as_dict(self) -> dict[str, object]:
return dict(self.document)
@classmethod
def create(cls, payload: dict[str, object]) -> ManualRenderPlanV1:
body = {
"schema_version": PROJECTION_SCHEMA_VERSION,
"contract": MANUAL_RENDER_PLAN_CONTRACT,
**payload,
}
document = {**body, "plan_id": projection_hash(body)}
return cls(validate_manual_render_plan(document))
@classmethod
def from_dict(cls, document: dict[str, object]) -> ManualRenderPlanV1:
return cls(validate_manual_render_plan(dict(document)))
@dataclass(frozen=True)
class GraphViewPlanV1:
"""One immutable, bounded portable-graph plan."""
document: dict[str, object]
@property
def plan_id(self) -> str:
return cast(str, self.document["plan_id"])
def as_dict(self) -> dict[str, object]:
return dict(self.document)
@classmethod
def create(cls, payload: dict[str, object]) -> GraphViewPlanV1:
body = {
"schema_version": PROJECTION_SCHEMA_VERSION,
"contract": GRAPH_VIEW_PLAN_CONTRACT,
**payload,
}
document = {**body, "plan_id": projection_hash(body)}
return cls(validate_graph_view_plan(document))
@classmethod
def from_dict(cls, document: dict[str, object]) -> GraphViewPlanV1:
return cls(validate_graph_view_plan(dict(document)))
@dataclass(frozen=True)
class ProjectionPackageV1:
"""Path-free package supplied to one capability-isolated renderer."""
document: dict[str, object]
@property
def package_id(self) -> str:
return cast(str, self.document["package_id"])
@property
def kind(self) -> ProjectionKind:
return cast(ProjectionKind, self.document["kind"])
def as_dict(self) -> dict[str, object]:
return dict(self.document)
@classmethod
def create(
cls,
*,
kind: ProjectionKind,
plan: ManualRenderPlanV1 | GraphViewPlanV1,
renderer: dict[str, object],
components: list[dict[str, object]],
assets: list[dict[str, object]],
output_policy: dict[str, object],
) -> ProjectionPackageV1:
body: dict[str, object] = {
"schema_version": PROJECTION_SCHEMA_VERSION,
"contract": PROJECTION_PACKAGE_CONTRACT,
"kind": kind,
"plan_id": plan.plan_id,
"plan": plan.as_dict(),
"renderer": renderer,
"components": components,
"assets": assets,
"output_policy": output_policy,
}
document = {**body, "package_id": projection_hash(body)}
return cls(validate_projection_package(document))
@classmethod
def from_dict(cls, document: dict[str, object]) -> ProjectionPackageV1:
return cls(validate_projection_package(dict(document)))
@dataclass(frozen=True)
class ProjectionReceiptV1:
"""Renderer evidence that contains identities and sizes, never artifact bytes."""
document: dict[str, object]
@property
def receipt_id(self) -> str:
return cast(str, self.document["receipt_id"])
def as_dict(self) -> dict[str, object]:
return dict(self.document)
@classmethod
def create(
cls,
*,
kind: ProjectionKind,
package_id: str,
plan_id: str,
renderer: dict[str, object],
artifacts: list[dict[str, object]],
diagnostics: dict[str, object],
timing: dict[str, object],
peak_memory_bytes: int | None,
) -> ProjectionReceiptV1:
body: dict[str, object] = {
"schema_version": PROJECTION_SCHEMA_VERSION,
"contract": PROJECTION_RECEIPT_CONTRACT,
"kind": kind,
"package_id": package_id,
"plan_id": plan_id,
"renderer": renderer,
"artifacts": artifacts,
"diagnostics": diagnostics,
"timing": timing,
"peak_memory_bytes": peak_memory_bytes,
}
document = {**body, "receipt_id": projection_hash(body)}
return cls(validate_projection_receipt(document))
@classmethod
def from_dict(cls, document: dict[str, object]) -> ProjectionReceiptV1:
return cls(validate_projection_receipt(dict(document)))
@dataclass(frozen=True)
class ProjectionArtifact:
"""One renderer-produced artifact addressed by a logical identifier."""
artifact_id: str
media_type: str
content: bytes
def evidence(self) -> dict[str, object]:
return {
"artifact_id": self.artifact_id,
"media_type": self.media_type,
"sha256": hashlib.sha256(self.content).hexdigest(),
"bytes": len(self.content),
}
@dataclass(frozen=True)
class ProjectionRenderResult:
"""Artifact bytes plus the bounded renderer receipt that attests them."""
artifacts: tuple[ProjectionArtifact, ...]
receipt: ProjectionReceiptV1
def _validate_project_identity(value: object) -> None:
if not isinstance(value, dict):
raise DocForgeError("invalid_projection", "Projection project identity is invalid")
project = cast(dict[str, object], value)
if set(project) != {
"project_id",
"project_root_fingerprint",
"adapter",
"revision",
"source_hash",
}:
raise DocForgeError("invalid_projection", "Projection project identity is invalid")
if not all(
isinstance(project.get(key), str) and bool(project[key])
for key in ("project_id", "project_root_fingerprint", "adapter", "revision")
) or not _is_hash(project.get("source_hash")):
raise DocForgeError("invalid_projection", "Projection project identity is invalid")
def validate_manual_render_plan(document: dict[str, object]) -> dict[str, object]:
required = {
"schema_version",
"contract",
"plan_id",
"project",
"view",
"changeset_hash",
"pages",
"navigation",
"search_documents",
"diagnostics",
}
if set(document) != required:
raise DocForgeError("invalid_projection", "Manual render plan fields are invalid")
if (
document.get("schema_version") != PROJECTION_SCHEMA_VERSION
or document.get("contract") != MANUAL_RENDER_PLAN_CONTRACT
):
raise DocForgeError("invalid_projection", "Manual render plan version is unsupported")
_validate_project_identity(document.get("project"))
if not all(
isinstance(document.get(key), expected)
for key, expected in (
("view", dict),
("pages", list),
("navigation", list),
("search_documents", list),
("diagnostics", dict),
)
):
raise DocForgeError("invalid_projection", "Manual render plan structure is invalid")
changeset_hash = document.get("changeset_hash")
if changeset_hash is not None and not _is_hash(changeset_hash):
raise DocForgeError("invalid_projection", "Manual render plan changeset hash is invalid")
return _validated_identity(
document,
identity_field="plan_id",
maximum=MAX_PLAN_BYTES,
kind="Manual render plan",
)
def validate_graph_view_plan(document: dict[str, object]) -> dict[str, object]:
required = {
"schema_version",
"contract",
"plan_id",
"project",
"view",
"bounds",
"policy",
"graph",
"omissions",
"diagnostics",
}
if set(document) != required:
raise DocForgeError("invalid_projection", "Graph view plan fields are invalid")
if (
document.get("schema_version") != PROJECTION_SCHEMA_VERSION
or document.get("contract") != GRAPH_VIEW_PLAN_CONTRACT
):
raise DocForgeError("invalid_projection", "Graph view plan version is unsupported")
_validate_project_identity(document.get("project"))
if not all(
isinstance(document.get(key), expected)
for key, expected in (
("view", dict),
("bounds", dict),
("policy", dict),
("graph", dict),
("omissions", list),
("diagnostics", dict),
)
):
raise DocForgeError("invalid_projection", "Graph view plan structure is invalid")
return _validated_identity(
document,
identity_field="plan_id",
maximum=MAX_PLAN_BYTES,
kind="Graph view plan",
)
def validate_projection_package(document: dict[str, object]) -> dict[str, object]:
required = {
"schema_version",
"contract",
"package_id",
"kind",
"plan_id",
"plan",
"renderer",
"components",
"assets",
"output_policy",
}
if set(document) != required:
raise DocForgeError("invalid_projection", "Projection package fields are invalid")
kind = document.get("kind")
if (
document.get("schema_version") != PROJECTION_SCHEMA_VERSION
or document.get("contract") != PROJECTION_PACKAGE_CONTRACT
or kind not in {"manual", "graph"}
):
raise DocForgeError("invalid_projection", "Projection package version or kind is invalid")
plan = document.get("plan")
if not isinstance(plan, dict):
raise DocForgeError("invalid_projection", "Projection package plan is invalid")
validated_plan = (
validate_manual_render_plan(cast(dict[str, object], plan))
if kind == "manual"
else validate_graph_view_plan(cast(dict[str, object], plan))
)
if document.get("plan_id") != validated_plan.get("plan_id"):
raise DocForgeError("invalid_projection", "Projection package plan identity is invalid")
if not all(
isinstance(document.get(key), expected)
for key, expected in (
("renderer", dict),
("components", list),
("assets", list),
("output_policy", dict),
)
):
raise DocForgeError("invalid_projection", "Projection package structure is invalid")
if len(cast(list[object], document["assets"])) > MAX_PROJECTION_ARTIFACTS:
raise DocForgeError("projection_too_large", "Projection package has too many assets")
_reject_runtime_authority(document)
return _validated_identity(
document,
identity_field="package_id",
maximum=MAX_PACKAGE_BYTES,
kind="Projection package",
)
def validate_projection_receipt(document: dict[str, object]) -> dict[str, object]:
required = {
"schema_version",
"contract",
"receipt_id",
"kind",
"package_id",
"plan_id",
"renderer",
"artifacts",
"diagnostics",
"timing",
"peak_memory_bytes",
}
if set(document) != required:
raise DocForgeError("invalid_projection", "Projection receipt fields are invalid")
if (
document.get("schema_version") != PROJECTION_SCHEMA_VERSION
or document.get("contract") != PROJECTION_RECEIPT_CONTRACT
or document.get("kind") not in {"manual", "graph"}
or not _is_hash(document.get("package_id"))
or not _is_hash(document.get("plan_id"))
):
raise DocForgeError("invalid_projection", "Projection receipt identity is invalid")
artifacts_value = document.get("artifacts")
if not isinstance(artifacts_value, list):
raise DocForgeError("invalid_projection", "Projection receipt structure is invalid")
artifacts = cast(list[object], artifacts_value)
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)
):
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")
for artifact in artifacts:
if not isinstance(artifact, dict):
raise DocForgeError("invalid_projection", "Projection receipt artifact is invalid")
item = cast(dict[str, object], artifact)
if (
set(item) != {"artifact_id", "media_type", "sha256", "bytes"}
or not isinstance(item.get("artifact_id"), str)
or not item["artifact_id"]
or "/" in cast(str, item["artifact_id"])
or not isinstance(item.get("media_type"), str)
or not item["media_type"]
or not _is_hash(item.get("sha256"))
or type(item.get("bytes")) is not int
or cast(int, item["bytes"]) < 0
):
raise DocForgeError("invalid_projection", "Projection receipt artifact is invalid")
return _validated_identity(
document,
identity_field="receipt_id",
maximum=MAX_RECEIPT_BYTES,
kind="Projection receipt",
)

1
src/docforge/py.typed Normal file
View file

@ -0,0 +1 @@
# PEP 561 marker for the typed DocForge public package.

107
src/docforge/render_contract.py
View file

@ -1,31 +1,17 @@
"""Deterministic built-in renderer contract and safe template primitives."""
"""Compatibility shim over the versioned manual projection boundary."""
from __future__ import annotations
import hashlib
import html
import json
import re
from dataclasses import dataclass
from importlib.metadata import version
from pathlib import Path
from typing import Protocol
from markdown_it import MarkdownIt
from .errors import DocForgeError
from .models import Edge, Node, ProjectSnapshot, RenderView
_TEMPLATE_TOKEN = re.compile(r"{{\s*([a-z_][a-z0-9_]*)\s*}}")
_ALLOWED_TOKENS = frozenset(
{
"docforge_content",
"docforge_project_id",
"docforge_render_identity",
"docforge_title",
"docforge_view_id",
}
)
from .manual_projection import build_manual_projection_package, build_manual_render_plan
from .models import ProjectSnapshot, RenderView
@dataclass(frozen=True)
@ -55,13 +41,12 @@ class Renderer(Protocol):
class GenericHtmlRenderer:
"""Render validated nodes through escaped CommonMark and a strict token template."""
"""Preserve the public v1 renderer API over the plan-only manual renderer."""
renderer_id = "generic_html"
contract_version = "1"
def __init__(self) -> None:
self.markdown = MarkdownIt("commonmark", {"html": False, "typographer": False})
self.renderer_version = (
f"{self.contract_version}+markdown-it-py-{version('markdown-it-py')}"
)
@ -74,22 +59,6 @@ class GenericHtmlRenderer:
*,
changeset_hash: str | None,
) -> PreparedRender:
try:
template = template_bytes.decode("utf-8")
except UnicodeDecodeError as error:
raise DocForgeError("invalid_template", "Render template is not valid UTF-8") from error
tokens = _TEMPLATE_TOKEN.findall(template)
unknown = sorted(set(tokens) - _ALLOWED_TOKENS)
remainder = _TEMPLATE_TOKEN.sub("", template)
if unknown or "{{" in remainder or "}}" in remainder:
raise DocForgeError(
"invalid_template", "Render template contains unsupported tokens", tokens=unknown
)
if tokens.count("docforge_content") != 1:
raise DocForgeError(
"invalid_template", "Render template must contain docforge_content exactly once"
)
selected = tuple(
node for node in snapshot.nodes if not view.families or node.family in view.families
)
@ -128,16 +97,26 @@ class GenericHtmlRenderer:
render_identity = hashlib.sha256(
json.dumps(identity_payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
).hexdigest()
content = self._content(selected, selected_edges)
replacements = {
"docforge_content": content,
"docforge_project_id": html.escape(snapshot.descriptor.project_id, quote=True),
"docforge_render_identity": render_identity,
"docforge_title": html.escape(view.title, quote=True),
"docforge_view_id": html.escape(view.view_id, quote=True),
}
rendered = _TEMPLATE_TOKEN.sub(lambda match: replacements[match.group(1)], template)
output = rendered.rstrip().encode("utf-8") + b"\n"
plan = build_manual_render_plan(snapshot, view, changeset_hash=changeset_hash)
package = build_manual_projection_package(
plan,
template_bytes,
renderer_id=self.renderer_id,
renderer_version=self.renderer_version,
max_output_bytes=snapshot.descriptor.limits.max_render_bytes,
)
from docforge_renderers.manual import ManualHtmlRenderer
result = ManualHtmlRenderer(self.renderer_version).render(
package,
render_identity=render_identity,
)
if len(result.artifacts) != 1:
raise DocForgeError(
"invalid_projection",
"Manual renderer returned an unsupported artifact set",
)
output = result.artifacts[0].content
return PreparedRender(
render_identity=render_identity,
output_hash=hashlib.sha256(output).hexdigest(),
@ -147,44 +126,6 @@ class GenericHtmlRenderer:
template_hash=template_hash,
)
def _content(self, nodes: tuple[Node, ...], edges: tuple[Edge, ...]) -> str:
navigation = ['<nav aria-label="Documentation"><ul>']
for node in nodes:
navigation.append(
f'<li><a href="#node-{html.escape(node.node_id, quote=True)}">'
f"{html.escape(node.title)}</a></li>"
)
navigation.append("</ul></nav>")
sections = [*navigation]
edge_map: dict[str, list[Edge]] = {}
for edge in edges:
edge_map.setdefault(edge.source_id, []).append(edge)
for node in nodes:
sections.extend(
[
f'<section id="node-{html.escape(node.node_id, quote=True)}">',
f"<h2>{html.escape(node.title)}</h2>",
'<dl class="docforge-node-meta">',
f"<dt>ID</dt><dd>{html.escape(node.node_id)}</dd>",
f"<dt>Family</dt><dd>{html.escape(node.family)}</dd>",
f"<dt>Status</dt><dd>{html.escape(node.status)}</dd>",
f"<dt>Authority</dt><dd>{html.escape(node.authority)}</dd>",
"</dl>",
f'<p class="docforge-summary">{html.escape(node.summary)}</p>',
self.markdown.render(node.content).rstrip(),
]
)
relationships = edge_map.get(node.node_id, [])
if relationships:
sections.append('<ul class="docforge-relationships">')
for edge in relationships:
sections.append(
f"<li>{html.escape(edge.relation)}: {html.escape(edge.target_id)}</li>"
)
sections.append("</ul>")
sections.append("</section>")
return "\n".join(sections)
_RENDERERS: dict[str, type[GenericHtmlRenderer]] = {
GenericHtmlRenderer.renderer_id: GenericHtmlRenderer

46
src/docforge/visualization.py
View file

@ -506,11 +506,11 @@ class VisualizationIndexSnapshot:
)
def source(self, node_id: str) -> dict[str, object]:
"""Return one node's bounded, project-confined UTF-8 source file."""
"""Return bounded source evidence stored in the pinned index generation."""
with self._connection() as connection:
row = connection.execute(
"SELECT node_id, source_path, source_anchor FROM nodes WHERE node_id = ?",
"SELECT node_id, source_path, source_anchor, content FROM nodes WHERE node_id = ?",
(node_id,),
).fetchone()
if row is None:
@ -518,7 +518,8 @@ class VisualizationIndexSnapshot:
"""
SELECT logic.logic_id AS node_id,
owner.source_path AS source_path,
logic.source_anchor AS source_anchor
logic.source_anchor AS source_anchor,
owner.content AS content
FROM logic_nodes AS logic
JOIN nodes AS owner ON owner.node_id = logic.owner_node_id
WHERE logic.logic_id = ?
@ -533,51 +534,26 @@ class VisualizationIndexSnapshot:
"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:
content = row["content"]
if not isinstance(content, str):
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",
"invalid_index",
"Pinned source evidence is invalid",
node_id=node_id,
)
raw = source.read_bytes()
raw = content.encode("utf-8")
if len(raw) > self.max_source_bytes:
raise DocForgeError(
"source_too_large",
"Node source exceeds the configured source limit",
"Pinned source evidence 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,
source_provenance="index_snapshot",
snapshot=True,
)

1
src/docforge_renderers/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Capability-isolated renderer implementations for DocForge projection packages."""

172
src/docforge_renderers/manual.py Normal file
View file

@ -0,0 +1,172 @@
"""Plan-only renderer for the built-in DocForge manual artifact."""
from __future__ import annotations
import hashlib
import html
import re
from time import perf_counter_ns
from typing import cast
from markdown_it import MarkdownIt
from docforge.errors import DocForgeError
from docforge.projection_contract import (
ProjectionArtifact,
ProjectionPackageV1,
ProjectionReceiptV1,
ProjectionRenderResult,
)
_TEMPLATE_TOKEN = re.compile(r"{{\s*([a-z_][a-z0-9_]*)\s*}}")
_ALLOWED_TOKENS = frozenset(
{
"docforge_content",
"docforge_project_id",
"docforge_render_identity",
"docforge_title",
"docforge_view_id",
}
)
_ACTIVE_TEMPLATE_CONTENT = re.compile(
r"<\s*(?:script|iframe|object|embed)\b"
r"|\son[a-z0-9_-]+\s*="
r"|javascript\s*:"
r"|<\s*meta\b[^>]*\bhttp-equiv\s*=\s*[\"']?\s*refresh\b",
re.IGNORECASE,
)
class ManualHtmlRenderer:
"""Transform one validated path-free package without graph-selection authority."""
renderer_id = "generic_html"
def __init__(self, renderer_version: str) -> None:
self.renderer_version = renderer_version
self.markdown = MarkdownIt("commonmark", {"html": False, "typographer": False})
def render(
self,
package: ProjectionPackageV1,
*,
render_identity: str | None = None,
) -> ProjectionRenderResult:
started = perf_counter_ns()
package = ProjectionPackageV1.from_dict(package.as_dict())
document = package.document
if package.kind != "manual":
raise DocForgeError("invalid_projection", "Manual renderer requires a manual package")
renderer = cast(dict[str, object], document["renderer"])
if renderer != {
"renderer_id": self.renderer_id,
"renderer_version": self.renderer_version,
}:
raise DocForgeError("unsupported_renderer", "Manual renderer identity is incompatible")
plan = cast(dict[str, object], document["plan"])
assets = cast(list[object], document["assets"])
if len(assets) != 1 or not isinstance(assets[0], dict):
raise DocForgeError("invalid_projection", "Manual template asset is invalid")
asset = cast(dict[str, object], assets[0])
if (
set(asset) != {"asset_id", "media_type", "sha256", "text"}
or asset.get("asset_id") != "manual.template"
or asset.get("media_type") != "text/html; charset=utf-8"
or not isinstance(asset.get("text"), str)
):
raise DocForgeError("invalid_projection", "Manual template asset is invalid")
template = cast(str, asset["text"])
template_bytes = template.encode("utf-8")
if hashlib.sha256(template_bytes).hexdigest() != asset.get("sha256"):
raise DocForgeError("invalid_projection", "Manual template asset hash is invalid")
if _ACTIVE_TEMPLATE_CONTENT.search(template):
raise DocForgeError(
"invalid_template",
"Render template contains active or executable content",
)
tokens = _TEMPLATE_TOKEN.findall(template)
unknown = sorted(set(tokens) - _ALLOWED_TOKENS)
remainder = _TEMPLATE_TOKEN.sub("", template)
if unknown or "{{" in remainder or "}}" in remainder:
raise DocForgeError(
"invalid_template",
"Render template contains unsupported tokens",
tokens=unknown,
)
if tokens.count("docforge_content") != 1:
raise DocForgeError(
"invalid_template",
"Render template must contain docforge_content exactly once",
)
identity = render_identity or cast(str, plan["plan_id"])
project = cast(dict[str, object], plan["project"])
view = cast(dict[str, object], plan["view"])
replacements = {
"docforge_content": self._content(plan),
"docforge_project_id": html.escape(cast(str, project["project_id"]), quote=True),
"docforge_render_identity": identity,
"docforge_title": html.escape(cast(str, view["title"]), quote=True),
"docforge_view_id": html.escape(cast(str, view["view_id"]), quote=True),
}
rendered = _TEMPLATE_TOKEN.sub(lambda match: replacements[match.group(1)], template)
output = rendered.rstrip().encode("utf-8") + b"\n"
policy = cast(dict[str, object], document["output_policy"])
maximum = policy.get("max_total_bytes")
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="manual.html",
media_type="text/html; charset=utf-8",
content=output,
)
receipt = ProjectionReceiptV1.create(
kind="manual",
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)
def _content(self, plan: dict[str, object]) -> str:
navigation = ['<nav aria-label="Documentation"><ul>']
for value in cast(list[object], plan["navigation"]):
item = cast(dict[str, object], value)
navigation.append(
f'<li><a href="#node-{html.escape(cast(str, item["node_id"]), quote=True)}">'
f"{html.escape(cast(str, item['title']))}</a></li>"
)
navigation.append("</ul></nav>")
sections = [*navigation]
for value in cast(list[object], plan["pages"]):
page = cast(dict[str, object], value)
node_id = cast(str, page["node_id"])
sections.extend(
[
f'<section id="node-{html.escape(node_id, quote=True)}">',
f"<h2>{html.escape(cast(str, page['title']))}</h2>",
'<dl class="docforge-node-meta">',
f"<dt>ID</dt><dd>{html.escape(node_id)}</dd>",
f"<dt>Family</dt><dd>{html.escape(cast(str, page['family']))}</dd>",
f"<dt>Status</dt><dd>{html.escape(cast(str, page['status']))}</dd>",
f"<dt>Authority</dt><dd>{html.escape(cast(str, page['authority']))}</dd>",
"</dl>",
f'<p class="docforge-summary">{html.escape(cast(str, page["summary"]))}</p>',
self.markdown.render(cast(str, page["content"])).rstrip(),
]
)
relationships = cast(list[object], page["cross_references"])
if relationships:
sections.append('<ul class="docforge-relationships">')
for relationship_value in relationships:
relationship = cast(dict[str, object], relationship_value)
sections.append(
f"<li>{html.escape(cast(str, relationship['relation']))}: "
f"{html.escape(cast(str, relationship['target_id']))}</li>"
)
sections.append("</ul>")
sections.append("</section>")
return "\n".join(sections)

1
src/docforge_renderers/py.typed Normal file
View file

@ -0,0 +1 @@
# PEP 561 marker for the typed DocForge renderer package.