Search for a node to inspect its neighborhood.
Visualization disconnected
diff --git a/src/docforge/cli.py b/src/docforge/cli.py
index dde293e..9e5f507 100644
--- a/src/docforge/cli.py
+++ b/src/docforge/cli.py
@@ -17,6 +17,7 @@ from .graph_rendering import GraphRenderService
from .index import ProjectIndex
from .onboarding import assess_project, scaffold_project
from .project import Project, project_root_fingerprint
+from .projection_policy import compose_projection_policy
from .rendering import RenderService
from .telemetry import request
from .viewer_manager import ViewerManagerClient
@@ -30,6 +31,18 @@ def _parser() -> argparse.ArgumentParser:
action="store_true",
help="Attach bounded request-local stage timings and counters",
)
+ parser.add_argument(
+ "--manual-render-policy",
+ choices=("auto", "explicit", "disabled"),
+ )
+ parser.add_argument(
+ "--portable-graph-policy",
+ choices=("explicit", "disabled"),
+ )
+ parser.add_argument(
+ "--live-viewer-policy",
+ choices=("on-demand", "disabled"),
+ )
commands = parser.add_subparsers(dest="command", required=True)
configure = commands.add_parser("configure")
configure.add_argument("client", choices=CLIENT_NAMES)
@@ -43,6 +56,21 @@ def _parser() -> argparse.ArgumentParser:
configure.add_argument("--proposal-writer")
configure.add_argument("--canonical-applier")
configure.add_argument("--no-ast", action="store_true")
+ configure.add_argument(
+ "--manual-render-policy",
+ choices=("auto", "explicit", "disabled"),
+ default=argparse.SUPPRESS,
+ )
+ configure.add_argument(
+ "--portable-graph-policy",
+ choices=("explicit", "disabled"),
+ default=argparse.SUPPRESS,
+ )
+ configure.add_argument(
+ "--live-viewer-policy",
+ choices=("on-demand", "disabled"),
+ default=argparse.SUPPRESS,
+ )
configure.add_argument("--startup-timeout", type=int, default=30)
configure.add_argument("--tool-timeout", type=int, default=300)
configure.add_argument("--output", type=Path)
@@ -131,6 +159,9 @@ def _run(arguments: argparse.Namespace) -> dict[str, object]:
proposal_writer=arguments.proposal_writer,
canonical_applier=arguments.canonical_applier,
no_ast=arguments.no_ast,
+ manual_render_policy=arguments.manual_render_policy,
+ portable_graph_policy=arguments.portable_graph_policy,
+ live_viewer_policy=arguments.live_viewer_policy,
startup_timeout=arguments.startup_timeout,
tool_timeout=arguments.tool_timeout,
output=arguments.output,
@@ -159,12 +190,39 @@ def _run(arguments: argparse.Namespace) -> dict[str, object]:
content_root=arguments.content_root,
)
project = Project.open(arguments.project_root)
+ projection_policy = compose_projection_policy(
+ manual=arguments.manual_render_policy,
+ portable_graph=arguments.portable_graph_policy,
+ live_viewer=arguments.live_viewer_policy,
+ manual_configured=project.descriptor.render is not None,
+ portable_graph_configured=project.descriptor.graph_render is not None,
+ application_enabled=False,
+ )
build = ProjectIndex(project).build()
- render = RenderService(project).render("manual")
+ render = (
+ {
+ "status": "ok",
+ "state": "skipped",
+ "reason": "projection_policy_disabled",
+ }
+ if projection_policy.manual == "disabled"
+ else RenderService(
+ project,
+ manual_policy=projection_policy.manual,
+ ).render("manual")
+ )
return {**scaffold, "build": build, "render": render}
return assess_project(arguments.project_root, requested_languages=languages)
project = Project.open(arguments.project_root)
index = ProjectIndex(project)
+ projection_policy = compose_projection_policy(
+ manual=arguments.manual_render_policy,
+ portable_graph=arguments.portable_graph_policy,
+ live_viewer=arguments.live_viewer_policy,
+ manual_configured=project.descriptor.render is not None,
+ portable_graph_configured=project.descriptor.graph_render is not None,
+ application_enabled=arguments.command == "apply",
+ )
if arguments.command == "info":
snapshot = project.load()
return {
@@ -257,30 +315,52 @@ def _run(arguments: argparse.Namespace) -> dict[str, object]:
cursor=arguments.cursor,
)
if arguments.command == "render":
- return RenderService(project).render(arguments.view_id)
+ return RenderService(
+ project,
+ manual_policy=projection_policy.manual,
+ ).render(arguments.view_id)
if arguments.command == "render-status":
- rendering = RenderService(project)
+ rendering = RenderService(
+ project,
+ manual_policy=projection_policy.manual,
+ )
return (
rendering.deep_status(arguments.view_id)
if arguments.deep
else rendering.status(arguments.view_id)
)
if arguments.command == "graph-plan":
- return GraphRenderService(project).plan(arguments.view_id)
+ return GraphRenderService(
+ project,
+ portable_graph_policy=projection_policy.portable_graph,
+ ).plan(arguments.view_id)
if arguments.command == "graph-render":
- return GraphRenderService(project).render(arguments.view_id)
+ return GraphRenderService(
+ project,
+ portable_graph_policy=projection_policy.portable_graph,
+ ).render(arguments.view_id)
if arguments.command == "graph-render-status":
- return GraphRenderService(project).status(arguments.view_id)
+ return GraphRenderService(
+ project,
+ portable_graph_policy=projection_policy.portable_graph,
+ ).status(arguments.view_id)
if arguments.command == "preview":
- return RenderService(project).preview(arguments.changeset_id, arguments.view_id)
+ return RenderService(
+ project,
+ manual_policy=projection_policy.manual,
+ ).preview(arguments.changeset_id, arguments.view_id)
if arguments.command == "apply":
return CanonicalApplicationService(
project,
applier_id=arguments.applier,
applier=GenericCanonicalApplier(project),
+ manual_policy=projection_policy.manual,
).apply(arguments.changeset_id, arguments.changeset_hash)
if arguments.command == "visualize":
- visualization = ViewerManagerClient(index).start(
+ visualization = ViewerManagerClient(
+ index,
+ live_viewer_policy=projection_policy.live_viewer,
+ ).start(
node_id=arguments.node,
query=arguments.query,
depth=arguments.depth,
@@ -297,9 +377,15 @@ def _run(arguments: argparse.Namespace) -> dict[str, object]:
"visualization": visualization,
}
if arguments.command == "visualization-status":
- return ViewerManagerClient(index).status()
+ return ViewerManagerClient(
+ index,
+ live_viewer_policy=projection_policy.live_viewer,
+ ).status()
if arguments.command == "visualization-stop":
- return ViewerManagerClient(index).stop()
+ return ViewerManagerClient(
+ index,
+ live_viewer_policy=projection_policy.live_viewer,
+ ).stop()
raise DocForgeError("invalid_command", "Unknown command")
diff --git a/src/docforge/client_config.py b/src/docforge/client_config.py
index 7b482d5..18c9de6 100644
--- a/src/docforge/client_config.py
+++ b/src/docforge/client_config.py
@@ -18,9 +18,10 @@ from typing import Literal, cast
from .changeset_contract import document_hash
from .errors import DocForgeError
-from .models import ProjectService
+from .models import ProjectDescriptor, ProjectService
from .policy import CapabilityMode, compose_effective_policy
-from .project import project_root_fingerprint, validate_descriptor_binding
+from .project import Project, project_root_fingerprint, validate_descriptor_binding
+from .projection_policy import compose_projection_policy
ClientName = Literal["codex", "claude", "openclaw"]
CLIENT_NAMES: tuple[ClientName, ...] = ("codex", "claude", "openclaw")
@@ -605,11 +606,37 @@ def _atomic_write(
os.close(directory_fd)
-def _validate_configuration_result(result: dict[str, object]) -> None:
+def _validate_configuration_result(
+ result: dict[str, object],
+ *,
+ trusted_descriptor: ProjectDescriptor | None = None,
+) -> None:
artifact = cast(dict[str, object], result["artifact"])
binding = cast(dict[str, object], result["binding"])
policy = cast(dict[str, object], result["effective_policy"])
+ projection_policy = cast(dict[str, object], result["projection_policy"])
+ projection_availability = cast(
+ dict[str, object],
+ result["projection_availability"],
+ )
project = cast(dict[str, object], result["project"])
+ if trusted_descriptor is None:
+ try:
+ bound_descriptor = Project.open(cast(str, project["project_root"])).descriptor
+ except (DocForgeError, KeyError, TypeError) as error:
+ raise AssertionError(
+ "Generated client project binding cannot be independently validated"
+ ) from error
+ else:
+ bound_descriptor = trusted_descriptor
+ if (
+ project["project_id"] != bound_descriptor.project_id
+ or project["project_root"] != str(bound_descriptor.root)
+ or project["project_root_fingerprint"] != project_root_fingerprint(bound_descriptor.root)
+ or project["adapter"] != bound_descriptor.adapter
+ or project["descriptor_hash"] != bound_descriptor.descriptor_hash
+ ):
+ raise AssertionError("Generated client project binding drifted")
content = cast(str, artifact["content"])
if artifact["content_sha256"] != hashlib.sha256(content.encode("utf-8")).hexdigest():
raise AssertionError("Generated client content hash drifted")
@@ -644,6 +671,29 @@ def _validate_configuration_result(result: dict[str, object]) -> None:
if remaining[-1:] != ["--no-ast"] or arguments.count("--no-ast") != 1:
raise AssertionError("Generated no-AST argument layout drifted")
remaining = remaining[:-1]
+ projection_arguments: dict[str, str] = {}
+ authority_arguments: list[str] = []
+ position = 0
+ projection_options = {
+ "--manual-render-policy": "manual",
+ "--portable-graph-policy": "portable_graph",
+ "--live-viewer-policy": "live_viewer",
+ }
+ while position < len(remaining):
+ option = remaining[position]
+ field = projection_options.get(option)
+ if field is None:
+ authority_arguments.append(option)
+ position += 1
+ continue
+ if position + 1 >= len(remaining) or option in projection_arguments:
+ raise AssertionError("Generated projection policy argument layout drifted")
+ value = remaining[position + 1]
+ projection_arguments[option] = value
+ if projection_policy[field] != value:
+ raise AssertionError("Generated projection policy argument drifted")
+ position += 2
+ remaining = authority_arguments
mode = binding["capability_mode"]
if (
(mode == "read" and remaining)
@@ -663,6 +713,34 @@ def _validate_configuration_result(result: dict[str, object]) -> None:
)
):
raise AssertionError("Generated authority argument layout drifted")
+ expected_projection_policy = compose_projection_policy(
+ manual=projection_arguments.get("--manual-render-policy"),
+ portable_graph=projection_arguments.get("--portable-graph-policy"),
+ live_viewer=projection_arguments.get("--live-viewer-policy"),
+ manual_configured=cast(bool, projection_availability["manual_configured"]),
+ portable_graph_configured=cast(
+ bool,
+ projection_availability["portable_graph_configured"],
+ ),
+ application_enabled=cast(
+ bool,
+ projection_availability["application_enabled"],
+ ),
+ live_viewer_available=cast(
+ bool,
+ projection_availability["live_viewer_available"],
+ ),
+ )
+ if (
+ projection_policy != expected_projection_policy.as_dict()
+ or projection_availability["manual_configured"] != (render_policy["manual"] != "disabled")
+ or projection_availability["manual_configured"] != (bound_descriptor.render is not None)
+ or projection_availability["portable_graph_configured"]
+ != (bound_descriptor.graph_render is not None)
+ or projection_availability["application_enabled"] != (mode == "application")
+ or projection_availability["live_viewer_available"] is not True
+ ):
+ raise AssertionError("Generated projection policy drifted from its availability")
composed_policy = compose_effective_policy(
selected_mode=cast(CapabilityMode, mode),
capability_source="explicit",
@@ -698,6 +776,18 @@ def _validate_configuration_result(result: dict[str, object]) -> None:
)
):
raise AssertionError("Generated client policy drifted from its binding")
+ if (
+ result["projection_policy_hash"]
+ != hashlib.sha256(
+ json.dumps(
+ projection_policy,
+ sort_keys=True,
+ separators=(",", ":"),
+ ensure_ascii=False,
+ ).encode("utf-8")
+ ).hexdigest()
+ ):
+ raise AssertionError("Generated projection policy hash drifted")
expected_hash = document_hash(
{
"schema_version": 1,
@@ -706,6 +796,9 @@ def _validate_configuration_result(result: dict[str, object]) -> None:
"project": project,
"binding": binding,
"effective_policy": policy,
+ "projection_policy": projection_policy,
+ "projection_policy_hash": result["projection_policy_hash"],
+ "projection_availability": projection_availability,
"artifact_format": artifact["format"],
"artifact_content_sha256": artifact["content_sha256"],
}
@@ -723,6 +816,9 @@ def generate_client_configuration(
proposal_writer: str | None = None,
canonical_applier: str | None = None,
no_ast: bool = False,
+ manual_render_policy: str | None = None,
+ portable_graph_policy: str | None = None,
+ live_viewer_policy: str | None = None,
startup_timeout: int = 30,
tool_timeout: int = 300,
output: Path | None = None,
@@ -839,9 +935,6 @@ def generate_client_configuration(
arguments.extend(("--proposal-writer", proposal_writer))
if canonical_applier is not None:
arguments.extend(("--canonical-applier", canonical_applier))
- if no_ast:
- arguments.append("--no-ast")
-
policy = compose_effective_policy(
selected_mode=selected_mode,
capability_source="explicit",
@@ -850,6 +943,40 @@ def generate_client_configuration(
render_configured=descriptor.render is not None,
application_enabled=canonical_applier is not None,
)
+ projection_policy = compose_projection_policy(
+ manual=manual_render_policy,
+ portable_graph=portable_graph_policy,
+ live_viewer=live_viewer_policy,
+ manual_configured=descriptor.render is not None,
+ portable_graph_configured=descriptor.graph_render is not None,
+ application_enabled=canonical_applier is not None,
+ )
+ default_projection_policy = compose_projection_policy(
+ manual_configured=descriptor.render is not None,
+ portable_graph_configured=descriptor.graph_render is not None,
+ application_enabled=canonical_applier is not None,
+ )
+ for option, selected, default in (
+ (
+ "--manual-render-policy",
+ projection_policy.manual,
+ default_projection_policy.manual,
+ ),
+ (
+ "--portable-graph-policy",
+ projection_policy.portable_graph,
+ default_projection_policy.portable_graph,
+ ),
+ (
+ "--live-viewer-policy",
+ projection_policy.live_viewer,
+ default_projection_policy.live_viewer,
+ ),
+ ):
+ if selected != default:
+ arguments.extend((option, selected))
+ if no_ast:
+ arguments.append("--no-ast")
artifact_format, content, warning = _artifact(
selected_client,
server_name=selected_name,
@@ -902,6 +1029,7 @@ def generate_client_configuration(
"project_root": str(descriptor.root),
"project_root_fingerprint": fingerprint,
"adapter": descriptor.adapter,
+ "descriptor_hash": descriptor.descriptor_hash,
}
policy_payload = policy.as_dict()
plan_hash = document_hash(
@@ -912,6 +1040,14 @@ def generate_client_configuration(
"project": project_binding,
"binding": binding,
"effective_policy": policy_payload,
+ "projection_policy": projection_policy.as_dict(),
+ "projection_policy_hash": projection_policy.policy_hash,
+ "projection_availability": {
+ "manual_configured": descriptor.render is not None,
+ "portable_graph_configured": descriptor.graph_render is not None,
+ "application_enabled": canonical_applier is not None,
+ "live_viewer_available": True,
+ },
"artifact_format": artifact_format,
"artifact_content_sha256": artifact["content_sha256"],
}
@@ -926,6 +1062,14 @@ def generate_client_configuration(
"project": project_binding,
"binding": binding,
"effective_policy": policy_payload,
+ "projection_policy": projection_policy.as_dict(),
+ "projection_policy_hash": projection_policy.policy_hash,
+ "projection_availability": {
+ "manual_configured": descriptor.render is not None,
+ "portable_graph_configured": descriptor.graph_render is not None,
+ "application_enabled": canonical_applier is not None,
+ "live_viewer_available": True,
+ },
"artifact": artifact,
"configuration_hash": plan_hash,
"warnings": [
@@ -933,5 +1077,5 @@ def generate_client_configuration(
*([] if publication_warning is None else [{"code": publication_warning}]),
],
}
- _validate_configuration_result(result)
+ _validate_configuration_result(result, trusted_descriptor=descriptor)
return result
diff --git a/src/docforge/doctor.py b/src/docforge/doctor.py
index e79ed61..89fd345 100644
--- a/src/docforge/doctor.py
+++ b/src/docforge/doctor.py
@@ -20,6 +20,7 @@ from .project import (
project_root_fingerprint,
validate_descriptor_binding,
)
+from .projection_policy import compose_projection_policy
MAX_CLIENT_CONFIG_BYTES = 1_000_000
MAX_CLIENT_SERVERS = 256
@@ -614,6 +615,9 @@ def _parse_binding(arguments: list[str]) -> dict[str, object]:
"--proposal-writer",
"--canonical-applier",
"--capability-mode",
+ "--manual-render-policy",
+ "--portable-graph-policy",
+ "--live-viewer-policy",
}
flag_options = {"--no-ast", "--diagnostics"}
position = 0
@@ -662,6 +666,9 @@ def _parse_binding(arguments: list[str]) -> dict[str, object]:
"capability_mode_implicit": implicit,
"no_ast": "--no-ast" in flags,
"diagnostics": "--diagnostics" in flags,
+ "manual_render_policy": values.get("--manual-render-policy"),
+ "portable_graph_policy": values.get("--portable-graph-policy"),
+ "live_viewer_policy": values.get("--live-viewer-policy"),
}
@@ -749,6 +756,16 @@ def _runtime_policy_check(
canonical_applier is not None and selected_mode in {"application", "operator"}
),
)
+ compose_projection_policy(
+ manual=cast(str | None, binding["manual_render_policy"]),
+ portable_graph=cast(str | None, binding["portable_graph_policy"]),
+ live_viewer=cast(str | None, binding["live_viewer_policy"]),
+ manual_configured=project.descriptor.render is not None,
+ portable_graph_configured=project.descriptor.graph_render is not None,
+ application_enabled=(
+ canonical_applier is not None and selected_mode in {"application", "operator"}
+ ),
+ )
if cast(bool, binding["capability_mode_implicit"]):
return (
"warning",
diff --git a/src/docforge/graph_rendering.py b/src/docforge/graph_rendering.py
index 55621bd..ee5c1a7 100644
--- a/src/docforge/graph_rendering.py
+++ b/src/docforge/graph_rendering.py
@@ -34,6 +34,11 @@ from .models import (
)
from .project import project_root_fingerprint
from .projection_contract import GraphViewPlanV1, ProjectionReceiptV1, projection_hash
+from .projection_policy import (
+ PortableGraphProjectionMode,
+ validate_portable_graph_projection_mode,
+)
+from .projection_worker import render_projection_in_worker
GRAPH_RENDERER_ID = "portable_graph_html"
GRAPH_RENDERER_VERSION = "1"
@@ -45,11 +50,29 @@ 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:
+ def __init__(
+ self,
+ project: ProjectService,
+ *,
+ allow_logic: bool = False,
+ portable_graph_policy: PortableGraphProjectionMode = "explicit",
+ ) -> None:
self.project = project
self.allow_logic = allow_logic
+ self.portable_graph_policy = validate_portable_graph_projection_mode(portable_graph_policy)
+
+ def _require_rendering(self, operation: str) -> None:
+ if self.portable_graph_policy == "disabled":
+ raise DocForgeError(
+ "projection_policy_forbids_operation",
+ "Portable graph projection policy disables rendering work",
+ projection="portable_graph",
+ mode=self.portable_graph_policy,
+ operation=operation,
+ )
def plan(self, view_id: str) -> dict[str, object]:
+ self._require_rendering("plan")
snapshot = self.project.load()
view = self._view(self._config(snapshot), view_id)
plan = self._plan(snapshot, view)
@@ -93,6 +116,7 @@ class GraphRenderService:
)
def render(self, view_id: str) -> dict[str, object]:
+ self._require_rendering("render")
with self._lock():
current_status = self.status(view_id)
current_outputs = cast(list[dict[str, object]], current_status["outputs"])
@@ -111,9 +135,7 @@ class GraphRenderService:
renderer_version=GRAPH_RENDERER_VERSION,
max_output_bytes=snapshot.descriptor.limits.max_render_bytes,
)
- from docforge_renderers.graph import PortableGraphHtmlRenderer
-
- result = PortableGraphHtmlRenderer().render(package)
+ result = render_projection_in_worker(package)
if len(result.artifacts) != 1:
raise DocForgeError(
"invalid_projection",
diff --git a/src/docforge/manual_projection.py b/src/docforge/manual_projection.py
index 3cee4d6..9e5640c 100644
--- a/src/docforge/manual_projection.py
+++ b/src/docforge/manual_projection.py
@@ -4,6 +4,7 @@ from __future__ import annotations
import hashlib
from collections import defaultdict
+from collections.abc import Sequence
from .errors import DocForgeError
from .models import Edge, ProjectSnapshot, RenderView
@@ -161,6 +162,8 @@ def build_manual_projection_package(
renderer_id: str,
renderer_version: str,
max_output_bytes: int,
+ render_identity: str | None = None,
+ fragment_records: Sequence[dict[str, object]] = (),
) -> ProjectionPackageV1:
"""Bind one plan and inert template asset for a path-free manual renderer."""
@@ -168,6 +171,23 @@ def build_manual_projection_package(
template = template_bytes.decode("utf-8")
except UnicodeDecodeError as error:
raise DocForgeError("invalid_template", "Render template is not valid UTF-8") from error
+ template_asset: dict[str, object] = {
+ "asset_id": "manual.template",
+ "media_type": "text/html; charset=utf-8",
+ "sha256": hashlib.sha256(template_bytes).hexdigest(),
+ "text": template,
+ }
+ if render_identity is not None:
+ template_asset["render_identity"] = render_identity
+ assets = [template_asset]
+ if fragment_records:
+ assets.append(
+ {
+ "asset_id": "manual.fragments",
+ "media_type": "application/vnd.docforge.projection-fragments.v1+json",
+ "records": list(fragment_records),
+ }
+ )
return ProjectionPackageV1.create(
kind="manual",
plan=plan,
@@ -176,14 +196,7 @@ def build_manual_projection_package(
{"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,
- }
- ],
+ assets=assets,
output_policy={
"artifact_ids": ["manual.html"],
"max_total_bytes": max_output_bytes,
diff --git a/src/docforge/mcp_server.py b/src/docforge/mcp_server.py
index d90e7b2..21e6544 100644
--- a/src/docforge/mcp_server.py
+++ b/src/docforge/mcp_server.py
@@ -15,11 +15,13 @@ from .application import CanonicalApplicationService, CanonicalApplier, GenericC
from .changesets import ChangesetStore
from .context import compile_context
from .errors import DocForgeError
+from .graph_rendering import GraphRenderService
from .index import ProjectIndex
from .models import IncrementalStateProject, ProjectService, RuntimeValidatedProject
from .pagination import canonical_hash, decode_cursor, page_limit, page_receipt
from .policy import CapabilityMode, capability_mode, compose_effective_policy
from .project import Project, project_root_fingerprint
+from .projection_policy import compose_projection_policy
from .rendering import RenderService
from .retrieval import MAX_TASK_EVIDENCE, TaskKind, build_retrieval_plan
from .telemetry import request, stage
@@ -47,6 +49,8 @@ READ_TOOLS = (
"docforge_get_task_context",
"docforge_validate_project",
"docforge_render_status",
+ "docforge_graph_plan",
+ "docforge_graph_render_status",
"docforge_visualize",
"docforge_stop_visualization",
"docforge_visualization_status",
@@ -134,6 +138,9 @@ class DocForgeService:
no_ast: bool = False,
diagnostics: bool = False,
capability_mode_name: str | None = None,
+ manual_projection_policy: str | None = None,
+ portable_graph_policy: str | None = None,
+ live_viewer_policy: str | None = None,
) -> None:
self.project = project
default_mode: CapabilityMode = (
@@ -153,19 +160,40 @@ class DocForgeService:
render_configured=project.descriptor.render is not None,
application_enabled=application_enabled,
)
+ self.projection_policy = compose_projection_policy(
+ manual=manual_projection_policy,
+ portable_graph=portable_graph_policy,
+ live_viewer=live_viewer_policy,
+ manual_configured=project.descriptor.render is not None,
+ portable_graph_configured=project.descriptor.graph_render is not None,
+ application_enabled=application_enabled,
+ )
self.index = ProjectIndex(self.project, allow_logic=not self.policy.no_ast)
self.changesets = ChangesetStore(
self.project,
proposal_writer if selected_mode != "read" else None,
)
- self.rendering = RenderService(self.project, self.changesets)
+ self.rendering = RenderService(
+ self.project,
+ self.changesets,
+ manual_policy=self.projection_policy.manual,
+ )
+ self.graph_rendering = GraphRenderService(
+ self.project,
+ allow_logic=not self.policy.no_ast,
+ portable_graph_policy=self.projection_policy.portable_graph,
+ )
self.application = CanonicalApplicationService(
self.project,
applier_id=canonical_applier_id if application_enabled else None,
applier=canonical_applier if application_enabled else None,
index=self.index,
+ manual_policy=self.projection_policy.manual,
+ )
+ self.visualization = ViewerManagerClient(
+ self.index,
+ live_viewer_policy=self.projection_policy.live_viewer,
)
- self.visualization = ViewerManagerClient(self.index)
self.context_provider = context_provider
self.task_context_available = context_provider is compile_context
self.binding_metadata = dict(binding_metadata or {})
@@ -621,6 +649,7 @@ class DocForgeService:
}
capabilities = self.capabilities()
effective_policy = self.policy.as_dict()
+ projection_policy = self.projection_policy.as_dict()
session_contract: dict[str, object] = {
"schema_version": 1,
"binding": binding,
@@ -630,6 +659,8 @@ class DocForgeService:
"freshness": "current",
},
"effective_policy": effective_policy,
+ "projection_policy": projection_policy,
+ "projection_policy_hash": self.projection_policy.policy_hash,
"capabilities": capabilities,
"render_policies": {
"manual": effective_policy["manual_render"],
@@ -652,6 +683,8 @@ class DocForgeService:
"canonical_paths": [str(path) for path in descriptor.content_roots],
"adapter_policy": self.adapter_policy(),
"effective_policy": effective_policy,
+ "projection_policy": projection_policy,
+ "projection_policy_hash": self.projection_policy.policy_hash,
"capabilities": capabilities,
"session_contract": session_contract,
"proposal_access": proposal_access,
@@ -714,6 +747,8 @@ class DocForgeService:
),
"adapter_policy": self.adapter_policy(),
"effective_policy": self.policy.as_dict(),
+ "projection_policy": self.projection_policy.as_dict(),
+ "projection_policy_hash": self.projection_policy.policy_hash,
"capabilities": self.capabilities(),
"canonical_paths": [
*(relative(path) for path in snapshot.descriptor.content_roots),
@@ -829,6 +864,29 @@ class DocForgeService:
operation_name="mcp.render_status",
)
+ def graph_plan(self, view_id: str) -> dict[str, object]:
+ """Plan one declared portable graph without publishing derived output."""
+
+ return self.invoke(
+ lambda: self.graph_rendering.plan(view_id),
+ synchronize=False,
+ load_error_identity=False,
+ operation_name="mcp.graph_plan",
+ )
+
+ def graph_render_status(
+ self,
+ view_id: str | None = None,
+ ) -> dict[str, object]:
+ """Report portable-graph publication state without planning or rendering."""
+
+ return self.invoke(
+ lambda: self.graph_rendering.status(view_id),
+ synchronize=False,
+ load_error_identity=False,
+ operation_name="mcp.graph_render_status",
+ )
+
def context(
self,
profile: str,
@@ -1575,6 +1633,18 @@ def _create_bound_server(service: DocForgeService, *, read_only: bool) -> FastMC
return service.render_status(view_id, deep=deep)
+ @server.tool(name="docforge_graph_plan")
+ def graph_plan(view_id: str) -> dict[str, Any]:
+ """Plan one declared portable graph without publishing derived output."""
+
+ return service.graph_plan(view_id)
+
+ @server.tool(name="docforge_graph_render_status")
+ def graph_render_status(view_id: str | None = None) -> dict[str, Any]:
+ """Report portable-graph publication state without rendering."""
+
+ return service.graph_render_status(view_id)
+
@server.tool(name="docforge_visualize")
def visualize(
node_id: str | None = None,
@@ -1622,6 +1692,8 @@ def _create_bound_server(service: DocForgeService, *, read_only: bool) -> FastMC
get_task_context,
validate_project,
render_status,
+ graph_plan,
+ graph_render_status,
visualize,
stop_visualization,
visualization_status,
@@ -2004,6 +2076,9 @@ def create_server(
no_ast: bool = False,
diagnostics: bool = False,
capability_mode: str | None = None,
+ manual_projection_policy: str | None = None,
+ portable_graph_policy: str | None = None,
+ live_viewer_policy: str | None = None,
) -> FastMCP:
project = Project.open(project_root)
return create_project_server(
@@ -2020,6 +2095,9 @@ def create_server(
no_ast=no_ast,
diagnostics=diagnostics,
capability_mode=capability_mode,
+ manual_projection_policy=manual_projection_policy,
+ portable_graph_policy=portable_graph_policy,
+ live_viewer_policy=live_viewer_policy,
)
@@ -2034,6 +2112,9 @@ def create_project_server(
no_ast: bool = False,
diagnostics: bool = False,
capability_mode: str | None = None,
+ manual_projection_policy: str | None = None,
+ portable_graph_policy: str | None = None,
+ live_viewer_policy: str | None = None,
) -> FastMCP:
"""Create the full fixed MCP surface for one explicitly configured project service."""
@@ -2047,6 +2128,9 @@ def create_project_server(
no_ast=no_ast,
diagnostics=diagnostics,
capability_mode_name=capability_mode,
+ manual_projection_policy=manual_projection_policy,
+ portable_graph_policy=portable_graph_policy,
+ live_viewer_policy=live_viewer_policy,
)
return _create_bound_server(
service,
@@ -2062,6 +2146,9 @@ def create_read_only_server(
no_ast: bool = False,
diagnostics: bool = False,
capability_mode: str | None = None,
+ manual_projection_policy: str | None = None,
+ portable_graph_policy: str | None = None,
+ live_viewer_policy: str | None = None,
) -> FastMCP:
"""Create an adapter-capable MCP server exposing only the fixed read tool surface."""
@@ -2073,6 +2160,9 @@ def create_read_only_server(
no_ast=no_ast,
diagnostics=diagnostics,
capability_mode_name="read" if capability_mode is None else capability_mode,
+ manual_projection_policy=manual_projection_policy,
+ portable_graph_policy=portable_graph_policy,
+ live_viewer_policy=live_viewer_policy,
)
if service.policy.capability_mode != "read":
raise DocForgeError(
@@ -2106,6 +2196,18 @@ def main() -> None:
choices=("read", "proposal", "application", "operator"),
help="Expose the versioned project-bound capability surface",
)
+ parser.add_argument(
+ "--manual-render-policy",
+ choices=("auto", "explicit", "disabled"),
+ )
+ parser.add_argument(
+ "--portable-graph-policy",
+ choices=("explicit", "disabled"),
+ )
+ parser.add_argument(
+ "--live-viewer-policy",
+ choices=("on-demand", "disabled"),
+ )
arguments = parser.parse_args()
create_server(
arguments.project_root,
@@ -2114,6 +2216,9 @@ def main() -> None:
no_ast=arguments.no_ast,
diagnostics=arguments.diagnostics,
capability_mode=arguments.capability_mode,
+ manual_projection_policy=arguments.manual_render_policy,
+ portable_graph_policy=arguments.portable_graph_policy,
+ live_viewer_policy=arguments.live_viewer_policy,
).run(transport="stdio")
diff --git a/src/docforge/project.py b/src/docforge/project.py
index 04dfebb..17052b3 100644
--- a/src/docforge/project.py
+++ b/src/docforge/project.py
@@ -550,7 +550,6 @@ def _load_descriptor(root: Path) -> ProjectDescriptor:
for field in defaults.__dataclass_fields__
}
)
-
render = load_render_config(
root,
document.get("render"),
diff --git a/src/docforge/projection_fragments.py b/src/docforge/projection_fragments.py
new file mode 100644
index 0000000..b2fe3d2
--- /dev/null
+++ b/src/docforge/projection_fragments.py
@@ -0,0 +1,489 @@
+"""Bounded, path-free, disposable projection fragment caching."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Literal, cast
+
+from ._fs_safety import (
+ atomic_replace_bytes_at,
+ open_confined_directory,
+ read_bounded_file_at,
+ require_bound_directory,
+)
+from .errors import DocForgeError
+from .projection_contract import canonical_projection_bytes, projection_hash
+
+FRAGMENT_SCHEMA_VERSION = 1
+FRAGMENT_KEY_CONTRACT = "docforge.projection-fragment-key"
+FRAGMENT_RECORD_CONTRACT = "docforge.projection-fragment-record"
+FRAGMENT_CACHE_DIRECTORY = "projection-fragments-v1"
+MAX_FRAGMENT_CONTENT_BYTES = 4_000_000
+MAX_FRAGMENT_ID_CHARS = 256
+MAX_FRAGMENT_CACHE_ENTRIES = 10_000
+MAX_FRAGMENT_CACHE_BYTES = 64_000_000
+_MAX_RECORD_OVERHEAD_BYTES = 8_192
+
+ProjectionKind = Literal["manual", "graph"]
+
+
+def _is_hash(value: object) -> bool:
+ return (
+ isinstance(value, str)
+ and len(value) == 64
+ and all(character in "0123456789abcdef" for character in value)
+ )
+
+
+def _version_string(value: object, *, field: str) -> str:
+ if (
+ not isinstance(value, str)
+ or not value
+ or value != value.strip()
+ or len(value) > MAX_FRAGMENT_ID_CHARS
+ or any(ord(character) < 32 for character in value)
+ ):
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment identity is invalid",
+ field=field,
+ )
+ return value
+
+
+def fragment_semantic_hash(value: object) -> str:
+ """Hash one complete semantic input using the projection canonical JSON form."""
+
+ try:
+ return hashlib.sha256(canonical_projection_bytes(value)).hexdigest()
+ except (TypeError, ValueError) as error:
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment semantic input is not canonical JSON",
+ ) from error
+
+
+@dataclass(frozen=True)
+class FragmentKey:
+ """Versioned identity for one renderer component's complete semantics."""
+
+ projection_kind: ProjectionKind
+ renderer_id: str
+ renderer_version: str
+ component_version: str
+ semantic_input_hash: str
+ key_id: str
+
+ @classmethod
+ def create(
+ cls,
+ *,
+ projection_kind: ProjectionKind,
+ renderer_id: str,
+ renderer_version: str,
+ component_version: str,
+ semantic_input_hash: str,
+ ) -> FragmentKey:
+ body = cls._body(
+ projection_kind=projection_kind,
+ renderer_id=renderer_id,
+ renderer_version=renderer_version,
+ component_version=component_version,
+ semantic_input_hash=semantic_input_hash,
+ )
+ return cls._from_validated({**body, "key_id": projection_hash(body)})
+
+ @classmethod
+ def from_dict(cls, value: object) -> FragmentKey:
+ if not isinstance(value, dict):
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment key is invalid",
+ )
+ return cls._from_validated(dict(cast(dict[str, object], value)))
+
+ @staticmethod
+ def _body(
+ *,
+ projection_kind: object,
+ renderer_id: object,
+ renderer_version: object,
+ component_version: object,
+ semantic_input_hash: object,
+ ) -> dict[str, object]:
+ if projection_kind not in {"manual", "graph"}:
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment kind is invalid",
+ )
+ if not _is_hash(semantic_input_hash):
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment semantic input hash is invalid",
+ )
+ return {
+ "schema_version": FRAGMENT_SCHEMA_VERSION,
+ "contract": FRAGMENT_KEY_CONTRACT,
+ "projection_kind": projection_kind,
+ "renderer_id": _version_string(renderer_id, field="renderer_id"),
+ "renderer_version": _version_string(
+ renderer_version,
+ field="renderer_version",
+ ),
+ "component_version": _version_string(
+ component_version,
+ field="component_version",
+ ),
+ "semantic_input_hash": semantic_input_hash,
+ }
+
+ @classmethod
+ def _from_validated(cls, value: dict[str, object]) -> FragmentKey:
+ required = {
+ "schema_version",
+ "contract",
+ "projection_kind",
+ "renderer_id",
+ "renderer_version",
+ "component_version",
+ "semantic_input_hash",
+ "key_id",
+ }
+ if (
+ set(value) != required
+ or value.get("schema_version") != FRAGMENT_SCHEMA_VERSION
+ or value.get("contract") != FRAGMENT_KEY_CONTRACT
+ ):
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment key contract is incompatible",
+ )
+ body = cls._body(
+ projection_kind=value.get("projection_kind"),
+ renderer_id=value.get("renderer_id"),
+ renderer_version=value.get("renderer_version"),
+ component_version=value.get("component_version"),
+ semantic_input_hash=value.get("semantic_input_hash"),
+ )
+ key_id = value.get("key_id")
+ if not _is_hash(key_id) or key_id != projection_hash(body):
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment key does not match its semantics",
+ )
+ return cls(
+ projection_kind=cast(ProjectionKind, body["projection_kind"]),
+ renderer_id=cast(str, body["renderer_id"]),
+ renderer_version=cast(str, body["renderer_version"]),
+ component_version=cast(str, body["component_version"]),
+ semantic_input_hash=cast(str, body["semantic_input_hash"]),
+ key_id=cast(str, key_id),
+ )
+
+ def as_dict(self) -> dict[str, object]:
+ return {
+ "schema_version": FRAGMENT_SCHEMA_VERSION,
+ "contract": FRAGMENT_KEY_CONTRACT,
+ "projection_kind": self.projection_kind,
+ "renderer_id": self.renderer_id,
+ "renderer_version": self.renderer_version,
+ "component_version": self.component_version,
+ "semantic_input_hash": self.semantic_input_hash,
+ "key_id": self.key_id,
+ }
+
+
+@dataclass(frozen=True)
+class FragmentRecord:
+ """One path-free fragment payload with complete byte evidence."""
+
+ key: FragmentKey
+ content: bytes
+ byte_count: int
+ content_sha256: str
+ record_id: str
+
+ @classmethod
+ def create(cls, key: FragmentKey, content: bytes) -> FragmentRecord:
+ if len(content) > MAX_FRAGMENT_CONTENT_BYTES:
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment content is invalid or oversized",
+ maximum_bytes=MAX_FRAGMENT_CONTENT_BYTES,
+ )
+ body = cls._body(key, content)
+ return cls(
+ key=key,
+ content=content,
+ byte_count=len(content),
+ content_sha256=hashlib.sha256(content).hexdigest(),
+ record_id=projection_hash(body),
+ )
+
+ @classmethod
+ def from_dict(cls, value: object) -> FragmentRecord:
+ if not isinstance(value, dict):
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment record is invalid",
+ )
+ document = dict(cast(dict[str, object], value))
+ required = {
+ "schema_version",
+ "contract",
+ "record_id",
+ "key",
+ "content_encoding",
+ "content",
+ "byte_count",
+ "content_sha256",
+ }
+ if (
+ set(document) != required
+ or document.get("schema_version") != FRAGMENT_SCHEMA_VERSION
+ or document.get("contract") != FRAGMENT_RECORD_CONTRACT
+ or document.get("content_encoding") != "base64"
+ ):
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment record contract is incompatible",
+ )
+ encoded = document.get("content")
+ if not isinstance(encoded, str):
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment content encoding is invalid",
+ )
+ try:
+ content = base64.b64decode(encoded.encode("ascii"), validate=True)
+ except (UnicodeEncodeError, ValueError) as error:
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment content encoding is invalid",
+ ) from error
+ if len(content) > MAX_FRAGMENT_CONTENT_BYTES:
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment content is oversized",
+ maximum_bytes=MAX_FRAGMENT_CONTENT_BYTES,
+ )
+ key = FragmentKey.from_dict(document.get("key"))
+ body = cls._body(key, content)
+ record_id = document.get("record_id")
+ if (
+ type(document.get("byte_count")) is not int
+ or document.get("byte_count") != len(content)
+ or document.get("content_sha256") != hashlib.sha256(content).hexdigest()
+ or not _is_hash(record_id)
+ or record_id != projection_hash(body)
+ ):
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment byte evidence is invalid",
+ )
+ return cls(
+ key=key,
+ content=content,
+ byte_count=len(content),
+ content_sha256=hashlib.sha256(content).hexdigest(),
+ record_id=cast(str, record_id),
+ )
+
+ @classmethod
+ def from_bytes(cls, raw: bytes) -> FragmentRecord:
+ try:
+ value: object = json.loads(raw)
+ except (UnicodeDecodeError, json.JSONDecodeError) as error:
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment record is not valid JSON",
+ ) from error
+ record = cls.from_dict(value)
+ if record.to_bytes() != raw:
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment record is not canonically serialized",
+ )
+ return record
+
+ @staticmethod
+ def _body(key: FragmentKey, content: bytes) -> dict[str, object]:
+ return {
+ "schema_version": FRAGMENT_SCHEMA_VERSION,
+ "contract": FRAGMENT_RECORD_CONTRACT,
+ "key": key.as_dict(),
+ "content_encoding": "base64",
+ "content": base64.b64encode(content).decode("ascii"),
+ "byte_count": len(content),
+ "content_sha256": hashlib.sha256(content).hexdigest(),
+ }
+
+ def as_dict(self) -> dict[str, object]:
+ return {
+ **self._body(self.key, self.content),
+ "record_id": self.record_id,
+ }
+
+ def to_bytes(self) -> bytes:
+ return canonical_projection_bytes(self.as_dict())
+
+
+class ProjectionFragmentCache:
+ """Confined best-effort storage for immutable projection fragments."""
+
+ def __init__(
+ self,
+ project_root: Path,
+ cache_root: Path,
+ *,
+ maximum_content_bytes: int = MAX_FRAGMENT_CONTENT_BYTES,
+ ) -> None:
+ if (
+ type(maximum_content_bytes) is not int
+ or not 1 <= maximum_content_bytes <= MAX_FRAGMENT_CONTENT_BYTES
+ ):
+ raise DocForgeError(
+ "invalid_projection_fragment",
+ "Projection fragment cache byte limit is invalid",
+ maximum_bytes=MAX_FRAGMENT_CONTENT_BYTES,
+ )
+ self.project_root = project_root
+ self.cache_root = cache_root
+ self.fragment_root = cache_root / FRAGMENT_CACHE_DIRECTORY
+ self.maximum_content_bytes = maximum_content_bytes
+
+ @property
+ def maximum_record_bytes(self) -> int:
+ encoded = ((self.maximum_content_bytes + 2) // 3) * 4
+ return encoded + _MAX_RECORD_OVERHEAD_BYTES
+
+ def get(self, key: FragmentKey) -> FragmentRecord | None:
+ """Return one exact compatible fragment, treating every cache defect as a miss."""
+
+ descriptor: int | None = None
+ try:
+ descriptor = self._open(create=False)
+ raw = read_bounded_file_at(
+ descriptor,
+ f"{key.key_id}.json",
+ self.maximum_record_bytes,
+ )
+ if raw is None:
+ return None
+ record = FragmentRecord.from_bytes(raw)
+ if record.key != key or record.byte_count > self.maximum_content_bytes:
+ return None
+ require_bound_directory(self.fragment_root, descriptor)
+ return record
+ except (DocForgeError, OSError):
+ return None
+ finally:
+ if descriptor is not None:
+ os.close(descriptor)
+
+ def put(self, key: FragmentKey, content: bytes) -> FragmentRecord | None:
+ """Durably publish one fragment, returning ``None`` on disposable cache failure."""
+
+ if len(content) > self.maximum_content_bytes:
+ return None
+ try:
+ record = FragmentRecord.create(key, content)
+ except DocForgeError:
+ return None
+ raw = record.to_bytes()
+ if len(raw) > self.maximum_record_bytes:
+ return None
+ descriptor: int | None = None
+ try:
+ descriptor = self._open(create=True)
+ name = f"{key.key_id}.json"
+ try:
+ existing = read_bounded_file_at(
+ descriptor,
+ name,
+ self.maximum_record_bytes,
+ )
+ except DocForgeError as error:
+ if error.code == "path_escape":
+ return None
+ existing = None
+ if existing == raw:
+ require_bound_directory(self.fragment_root, descriptor)
+ return record
+ atomic_replace_bytes_at(
+ self.fragment_root,
+ descriptor,
+ name,
+ raw,
+ verify=lambda: require_bound_directory(self.fragment_root, descriptor),
+ )
+ return record
+ except (DocForgeError, OSError):
+ return None
+ finally:
+ if descriptor is not None:
+ os.close(descriptor)
+
+ def prune(self, keep: tuple[FragmentKey, ...]) -> bool:
+ """Remove every stale entry and prove the retained inventory is bounded."""
+
+ keep_ids = {key.key_id for key in keep}
+ if len(keep_ids) > MAX_FRAGMENT_CACHE_ENTRIES:
+ return False
+ descriptor: int | None = None
+ try:
+ descriptor = self._open(create=False)
+ retained_entries = 0
+ retained_bytes = 0
+ with os.scandir(descriptor) as entries:
+ for entry in entries:
+ name = entry.name
+ retained = (
+ len(name) == 69
+ and name.endswith(".json")
+ and name[:-5] in keep_ids
+ and all(character in "0123456789abcdef" for character in name[:-5])
+ )
+ if retained:
+ identity = entry.stat(follow_symlinks=False)
+ if not entry.is_file(follow_symlinks=False):
+ return False
+ retained_entries += 1
+ retained_bytes += identity.st_size
+ continue
+ try:
+ os.unlink(name, dir_fd=descriptor)
+ except OSError:
+ return False
+ if (
+ retained_entries > MAX_FRAGMENT_CACHE_ENTRIES
+ or retained_bytes > MAX_FRAGMENT_CACHE_BYTES
+ ):
+ return False
+ require_bound_directory(self.fragment_root, descriptor)
+ os.fsync(descriptor)
+ return True
+ except (DocForgeError, OSError):
+ return False
+ finally:
+ if descriptor is not None:
+ os.close(descriptor)
+
+ def _open(self, *, create: bool) -> int:
+ if self.cache_root == self.project_root or not self.cache_root.is_relative_to(
+ self.project_root
+ ):
+ raise DocForgeError(
+ "path_escape",
+ "Projection fragment cache is not confined to a derived project root",
+ )
+ return open_confined_directory(
+ self.project_root,
+ self.fragment_root,
+ create=create,
+ )
diff --git a/src/docforge/projection_policy.py b/src/docforge/projection_policy.py
new file mode 100644
index 0000000..5c7321d
--- /dev/null
+++ b/src/docforge/projection_policy.py
@@ -0,0 +1,234 @@
+"""Independent version-2 policy for manual, portable graph, and live projections."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from typing import Literal, cast
+
+from .errors import DocForgeError
+
+ManualProjectionMode = Literal["auto", "explicit", "disabled"]
+PortableGraphProjectionMode = Literal["explicit", "disabled"]
+LiveViewerProjectionMode = Literal["on-demand", "disabled"]
+
+MANUAL_PROJECTION_MODES: tuple[ManualProjectionMode, ...] = (
+ "auto",
+ "explicit",
+ "disabled",
+)
+PORTABLE_GRAPH_PROJECTION_MODES: tuple[PortableGraphProjectionMode, ...] = (
+ "explicit",
+ "disabled",
+)
+LIVE_VIEWER_PROJECTION_MODES: tuple[LiveViewerProjectionMode, ...] = (
+ "on-demand",
+ "disabled",
+)
+
+
+def _invalid_mode(field: str, value: object, allowed: tuple[str, ...]) -> DocForgeError:
+ return DocForgeError(
+ "invalid_projection_policy",
+ "Projection policy mode is unsupported",
+ projection=field,
+ mode=value,
+ allowed=list(allowed),
+ )
+
+
+def _select_mode(
+ value: object | None,
+ *,
+ field: str,
+ default: str,
+ allowed: tuple[str, ...],
+) -> str:
+ selected: object = default if value is None else value
+ if not isinstance(selected, str) or selected not in allowed:
+ raise _invalid_mode(field, selected, allowed)
+ return selected
+
+
+def _require_boolean(field: str, value: object) -> bool:
+ if type(value) is not bool:
+ raise DocForgeError(
+ "invalid_projection_policy",
+ "Projection policy availability must be Boolean",
+ field=field,
+ )
+ return value
+
+
+def _unavailable(field: str, mode: str, required: str) -> DocForgeError:
+ return DocForgeError(
+ "projection_policy_unavailable",
+ "Projection policy mode is unavailable",
+ projection=field,
+ mode=mode,
+ required=required,
+ )
+
+
+def validate_manual_projection_mode(value: object) -> ManualProjectionMode:
+ """Validate one direct manual-service policy selection."""
+
+ return cast(
+ ManualProjectionMode,
+ _select_mode(
+ value,
+ field="manual",
+ default="explicit",
+ allowed=MANUAL_PROJECTION_MODES,
+ ),
+ )
+
+
+def validate_portable_graph_projection_mode(
+ value: object,
+) -> PortableGraphProjectionMode:
+ """Validate one direct portable-graph service policy selection."""
+
+ return cast(
+ PortableGraphProjectionMode,
+ _select_mode(
+ value,
+ field="portable_graph",
+ default="explicit",
+ allowed=PORTABLE_GRAPH_PROJECTION_MODES,
+ ),
+ )
+
+
+def validate_live_viewer_projection_mode(value: object) -> LiveViewerProjectionMode:
+ """Validate one direct live-viewer service policy selection."""
+
+ return cast(
+ LiveViewerProjectionMode,
+ _select_mode(
+ value,
+ field="live_viewer",
+ default="on-demand",
+ allowed=LIVE_VIEWER_PROJECTION_MODES,
+ ),
+ )
+
+
+@dataclass(frozen=True)
+class ProjectionPolicyV2:
+ """One immutable policy for three independent projection consumers."""
+
+ manual: ManualProjectionMode
+ portable_graph: PortableGraphProjectionMode
+ live_viewer: LiveViewerProjectionMode
+
+ def __post_init__(self) -> None:
+ if self.manual not in MANUAL_PROJECTION_MODES:
+ raise _invalid_mode("manual", self.manual, MANUAL_PROJECTION_MODES)
+ if self.portable_graph not in PORTABLE_GRAPH_PROJECTION_MODES:
+ raise _invalid_mode(
+ "portable_graph",
+ self.portable_graph,
+ PORTABLE_GRAPH_PROJECTION_MODES,
+ )
+ if self.live_viewer not in LIVE_VIEWER_PROJECTION_MODES:
+ raise _invalid_mode(
+ "live_viewer",
+ self.live_viewer,
+ LIVE_VIEWER_PROJECTION_MODES,
+ )
+
+ def as_dict(self) -> dict[str, object]:
+ return {
+ "schema_version": 2,
+ "manual": self.manual,
+ "portable_graph": self.portable_graph,
+ "live_viewer": self.live_viewer,
+ }
+
+ @property
+ def policy_hash(self) -> str:
+ raw = json.dumps(
+ self.as_dict(),
+ sort_keys=True,
+ separators=(",", ":"),
+ ensure_ascii=False,
+ ).encode("utf-8")
+ return hashlib.sha256(raw).hexdigest()
+
+
+def compose_projection_policy(
+ *,
+ manual: str | None = None,
+ portable_graph: str | None = None,
+ live_viewer: str | None = None,
+ manual_configured: bool,
+ portable_graph_configured: bool,
+ application_enabled: bool,
+ live_viewer_available: bool = True,
+) -> ProjectionPolicyV2:
+ """Compose compatible defaults with explicit availability-checked selections."""
+
+ manual_available = _require_boolean("manual_configured", manual_configured)
+ graph_available = _require_boolean(
+ "portable_graph_configured",
+ portable_graph_configured,
+ )
+ application_available = _require_boolean("application_enabled", application_enabled)
+ viewer_available = _require_boolean("live_viewer_available", live_viewer_available)
+
+ default_manual = (
+ "auto"
+ if manual_available and application_available
+ else ("explicit" if manual_available else "disabled")
+ )
+ default_graph = "explicit" if graph_available else "disabled"
+ default_viewer = "on-demand" if viewer_available else "disabled"
+
+ manual_mode = cast(
+ ManualProjectionMode,
+ _select_mode(
+ manual,
+ field="manual",
+ default=default_manual,
+ allowed=MANUAL_PROJECTION_MODES,
+ ),
+ )
+ graph_mode = cast(
+ PortableGraphProjectionMode,
+ _select_mode(
+ portable_graph,
+ field="portable_graph",
+ default=default_graph,
+ allowed=PORTABLE_GRAPH_PROJECTION_MODES,
+ ),
+ )
+ viewer_mode = cast(
+ LiveViewerProjectionMode,
+ _select_mode(
+ live_viewer,
+ field="live_viewer",
+ default=default_viewer,
+ allowed=LIVE_VIEWER_PROJECTION_MODES,
+ ),
+ )
+
+ if manual_mode != "disabled" and not manual_available:
+ raise _unavailable("manual", manual_mode, "manual_render_config")
+ if manual_mode == "auto" and not application_available:
+ raise _unavailable("manual", manual_mode, "canonical_application")
+ if graph_mode == "explicit" and not graph_available:
+ raise _unavailable(
+ "portable_graph",
+ graph_mode,
+ "portable_graph_render_config",
+ )
+ if viewer_mode == "on-demand" and not viewer_available:
+ raise _unavailable("live_viewer", viewer_mode, "live_viewer_runtime")
+
+ return ProjectionPolicyV2(
+ manual=manual_mode,
+ portable_graph=graph_mode,
+ live_viewer=viewer_mode,
+ )
diff --git a/src/docforge/projection_worker.py b/src/docforge/projection_worker.py
new file mode 100644
index 0000000..558c197
--- /dev/null
+++ b/src/docforge/projection_worker.py
@@ -0,0 +1,436 @@
+"""One-shot detached execution for the fixed built-in projection renderers."""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import json
+import os
+import resource
+import subprocess
+import sys
+import tempfile
+from importlib.metadata import version
+from typing import cast
+
+from .errors import DocForgeError
+from .projection_contract import (
+ MAX_PACKAGE_BYTES,
+ MAX_PROJECTION_ARTIFACTS,
+ MAX_RECEIPT_BYTES,
+ ProjectionArtifact,
+ ProjectionPackageV1,
+ ProjectionReceiptV1,
+ ProjectionRenderResult,
+ canonical_projection_bytes,
+)
+
+WORKER_PROTOCOL_VERSION = 1
+MAX_WORKER_ARTIFACT_BYTES = 20_000_000
+MAX_WORKER_REQUEST_BYTES = MAX_PACKAGE_BYTES + 1
+MAX_WORKER_RESPONSE_BYTES = 4 * ((MAX_WORKER_ARTIFACT_BYTES + 2) // 3) + MAX_RECEIPT_BYTES + 256_000
+WORKER_TIMEOUT_SECONDS = 30
+
+_GENERIC_HTML_RENDERER_ID = "generic_html"
+_PORTABLE_GRAPH_RENDERER_ID = "portable_graph_html"
+_PORTABLE_GRAPH_RENDERER_VERSION = "1"
+
+
+def _generic_html_renderer_version() -> str:
+ return f"1+markdown-it-py-{version('markdown-it-py')}"
+
+
+def _worker_failure(message: str, **details: object) -> DocForgeError:
+ return DocForgeError("projection_worker_failure", message, **details)
+
+
+def _renderer_identity(package: ProjectionPackageV1) -> dict[str, object]:
+ renderer_value = package.document.get("renderer")
+ if not isinstance(renderer_value, dict):
+ raise DocForgeError("unsupported_renderer", "Projection renderer identity is invalid")
+ renderer = cast(dict[str, object], renderer_value)
+ if set(renderer) != {"renderer_id", "renderer_version"}:
+ raise DocForgeError("unsupported_renderer", "Projection renderer identity is invalid")
+ renderer_id = renderer.get("renderer_id")
+ renderer_version = renderer.get("renderer_version")
+ if not isinstance(renderer_id, str) or not isinstance(renderer_version, str):
+ raise DocForgeError("unsupported_renderer", "Projection renderer identity is invalid")
+ supported = (
+ package.kind == "manual"
+ and renderer_id == _GENERIC_HTML_RENDERER_ID
+ and renderer_version == _generic_html_renderer_version()
+ ) or (
+ package.kind == "graph"
+ and renderer_id == _PORTABLE_GRAPH_RENDERER_ID
+ and renderer_version == _PORTABLE_GRAPH_RENDERER_VERSION
+ )
+ if not supported:
+ raise DocForgeError(
+ "unsupported_renderer",
+ "Projection worker supports only the fixed built-in renderer versions",
+ )
+ return dict(renderer)
+
+
+def _output_policy(package: ProjectionPackageV1) -> tuple[tuple[str, ...], int]:
+ policy_value = package.document.get("output_policy")
+ if not isinstance(policy_value, dict):
+ raise DocForgeError("invalid_projection", "Projection output policy is invalid")
+ policy = cast(dict[str, object], policy_value)
+ if set(policy) != {"artifact_ids", "max_total_bytes"}:
+ raise DocForgeError("invalid_projection", "Projection output policy is invalid")
+ artifact_ids_value = policy.get("artifact_ids")
+ maximum = policy.get("max_total_bytes")
+ if not isinstance(artifact_ids_value, list):
+ raise DocForgeError("invalid_projection", "Projection artifact inventory is invalid")
+ artifact_ids_objects = cast(list[object], artifact_ids_value)
+ if (
+ not artifact_ids_objects
+ or len(artifact_ids_objects) > MAX_PROJECTION_ARTIFACTS
+ or not all(
+ isinstance(artifact_id, str)
+ and bool(artifact_id)
+ and "/" not in artifact_id
+ and artifact_id not in {".", ".."}
+ for artifact_id in artifact_ids_objects
+ )
+ ):
+ raise DocForgeError("invalid_projection", "Projection artifact inventory is invalid")
+ artifact_ids = tuple(cast(list[str], artifact_ids_objects))
+ if len(set(artifact_ids)) != len(artifact_ids):
+ raise DocForgeError("invalid_projection", "Projection artifact inventory is duplicated")
+ if type(maximum) is not int or maximum < 1:
+ raise DocForgeError(
+ "invalid_projection",
+ "Projection output byte allowance is invalid",
+ )
+ return artifact_ids, maximum
+
+
+def _validated_package(package: object) -> ProjectionPackageV1:
+ if not isinstance(package, ProjectionPackageV1):
+ raise TypeError("Projection worker requires ProjectionPackageV1")
+ validated = ProjectionPackageV1.from_dict(package.as_dict())
+ _renderer_identity(validated)
+ _output_policy(validated)
+ return validated
+
+
+def _validate_result(
+ package: ProjectionPackageV1,
+ result: ProjectionRenderResult,
+ *,
+ require_peak_memory: bool,
+) -> ProjectionRenderResult:
+ artifact_ids, maximum = _output_policy(package)
+ if (
+ len(result.artifacts) != len(artifact_ids)
+ or tuple(artifact.artifact_id for artifact in result.artifacts) != artifact_ids
+ ):
+ raise _worker_failure("Projection worker returned an invalid artifact inventory")
+ total_bytes = 0
+ evidence: list[dict[str, object]] = []
+ for artifact in result.artifacts:
+ if not artifact.media_type or type(artifact.content) is not bytes:
+ raise _worker_failure("Projection worker returned an invalid artifact")
+ total_bytes += len(artifact.content)
+ if total_bytes > maximum or total_bytes > MAX_WORKER_ARTIFACT_BYTES:
+ raise _worker_failure("Projection worker artifact transfer exceeded its fixed boundary")
+ evidence.append(artifact.evidence())
+
+ try:
+ receipt = ProjectionReceiptV1.from_dict(result.receipt.as_dict())
+ except DocForgeError as error:
+ raise _worker_failure("Projection worker receipt is invalid") from error
+ receipt_document = receipt.document
+ renderer = _renderer_identity(package)
+ timing_value = receipt_document.get("timing")
+ if not isinstance(timing_value, dict):
+ raise _worker_failure("Projection worker receipt timing is invalid")
+ timing = cast(dict[str, object], timing_value)
+ elapsed = timing.get("elapsed_ns")
+ peak_memory = receipt_document.get("peak_memory_bytes")
+ if (
+ receipt_document.get("kind") != package.kind
+ or receipt_document.get("package_id") != package.package_id
+ or receipt_document.get("plan_id") != package.document.get("plan_id")
+ or receipt_document.get("renderer") != renderer
+ or receipt_document.get("artifacts") != evidence
+ or type(elapsed) is not int
+ or elapsed < 0
+ or (require_peak_memory and (type(peak_memory) is not int or peak_memory <= 0))
+ ):
+ raise _worker_failure("Projection worker receipt does not attest the requested package")
+ return ProjectionRenderResult(tuple(result.artifacts), receipt)
+
+
+def _decode_canonical_line(raw: bytes, *, maximum: int, label: str) -> dict[str, object]:
+ if type(raw) is not bytes or len(raw) > maximum:
+ raise _worker_failure(f"{label} exceeded its fixed boundary", maximum_bytes=maximum)
+ if not raw or not raw.endswith(b"\n") or raw.count(b"\n") != 1:
+ raise _worker_failure(f"{label} framing is invalid")
+ payload = raw[:-1]
+ try:
+ value: object = json.loads(payload)
+ except (UnicodeDecodeError, json.JSONDecodeError) as error:
+ raise _worker_failure(f"{label} is not valid JSON") from error
+ if not isinstance(value, dict):
+ raise _worker_failure(f"{label} must be one JSON object")
+ document = cast(dict[str, object], value)
+ if canonical_projection_bytes(document) != payload:
+ raise _worker_failure(f"{label} is not canonical JSON")
+ return document
+
+
+def _encode_request(package: ProjectionPackageV1) -> bytes:
+ encoded = canonical_projection_bytes(package.as_dict()) + b"\n"
+ if len(encoded) > MAX_WORKER_REQUEST_BYTES:
+ raise DocForgeError(
+ "projection_too_large",
+ "Projection worker request exceeds its fixed boundary",
+ maximum_bytes=MAX_WORKER_REQUEST_BYTES,
+ )
+ return encoded
+
+
+def _invoke_worker(request: bytes) -> subprocess.CompletedProcess[bytes]:
+ environment = {key: os.environ[key] for key in ("SYSTEMROOT", "WINDIR") if key in os.environ}
+ environment.update(
+ {
+ "PYTHONIOENCODING": "utf-8",
+ "PYTHONUTF8": "1",
+ }
+ )
+ command = [sys.executable, "-I", "-m", "docforge.projection_worker"]
+ with tempfile.TemporaryFile() as output:
+ completed = subprocess.run(
+ command,
+ input=request,
+ stdout=output,
+ stderr=subprocess.DEVNULL,
+ check=False,
+ timeout=WORKER_TIMEOUT_SECONDS,
+ shell=False,
+ cwd=sys.prefix,
+ env=environment,
+ )
+ output.seek(0)
+ stdout = output.read(MAX_WORKER_RESPONSE_BYTES + 1)
+ return subprocess.CompletedProcess(
+ command,
+ completed.returncode,
+ stdout=stdout,
+ )
+
+
+def _decode_response(package: ProjectionPackageV1, raw: bytes) -> ProjectionRenderResult:
+ document = _decode_canonical_line(
+ raw,
+ maximum=MAX_WORKER_RESPONSE_BYTES,
+ label="Projection worker response",
+ )
+ if (
+ set(document) != {"schema_version", "artifacts", "receipt"}
+ or document.get("schema_version") != WORKER_PROTOCOL_VERSION
+ ):
+ raise _worker_failure("Projection worker response contract is invalid")
+ artifact_values = document.get("artifacts")
+ receipt_value = document.get("receipt")
+ if not isinstance(artifact_values, list) or not isinstance(receipt_value, dict):
+ raise _worker_failure("Projection worker response structure is invalid")
+ artifacts: list[ProjectionArtifact] = []
+ total_bytes = 0
+ for value in cast(list[object], artifact_values):
+ if not isinstance(value, dict):
+ raise _worker_failure("Projection worker artifact envelope is invalid")
+ artifact = cast(dict[str, object], value)
+ if set(artifact) != {"artifact_id", "media_type", "content_base64"}:
+ raise _worker_failure("Projection worker artifact envelope is invalid")
+ artifact_id = artifact.get("artifact_id")
+ media_type = artifact.get("media_type")
+ encoded = artifact.get("content_base64")
+ if (
+ not isinstance(artifact_id, str)
+ or not isinstance(media_type, str)
+ or not isinstance(encoded, str)
+ ):
+ raise _worker_failure("Projection worker artifact envelope is invalid")
+ try:
+ content = base64.b64decode(encoded.encode("ascii"), validate=True)
+ except (UnicodeEncodeError, binascii.Error, ValueError) as error:
+ raise _worker_failure("Projection worker artifact encoding is invalid") from error
+ total_bytes += len(content)
+ if total_bytes > MAX_WORKER_ARTIFACT_BYTES:
+ raise _worker_failure("Projection worker artifact transfer exceeded its fixed boundary")
+ artifacts.append(ProjectionArtifact(artifact_id, media_type, content))
+ try:
+ receipt = ProjectionReceiptV1.from_dict(cast(dict[str, object], receipt_value))
+ except DocForgeError as error:
+ raise _worker_failure("Projection worker receipt is invalid") from error
+ return _validate_result(
+ package,
+ ProjectionRenderResult(tuple(artifacts), receipt),
+ require_peak_memory=True,
+ )
+
+
+def render_projection_in_worker(package: ProjectionPackageV1) -> ProjectionRenderResult:
+ """Render one validated path-free package in a fixed one-shot child process."""
+
+ validated = _validated_package(package)
+ request = _encode_request(validated)
+ try:
+ completed = _invoke_worker(request)
+ except subprocess.TimeoutExpired as error:
+ raise DocForgeError(
+ "projection_worker_timeout",
+ "Detached projection worker exceeded its fixed timeout",
+ timeout_seconds=WORKER_TIMEOUT_SECONDS,
+ ) from error
+ except OSError as error:
+ raise _worker_failure("Detached projection worker could not be launched") from error
+ if completed.returncode != 0:
+ if completed.returncode == 3 and completed.stdout:
+ try:
+ failure = _decode_canonical_line(
+ completed.stdout,
+ maximum=MAX_RECEIPT_BYTES,
+ label="Projection worker error response",
+ )
+ error = failure.get("error")
+ error_document = cast(dict[str, object], error) if isinstance(error, dict) else None
+ if (
+ set(failure) == {"schema_version", "error"}
+ and failure.get("schema_version") == WORKER_PROTOCOL_VERSION
+ and error_document is not None
+ and set(error_document) == {"code", "message", "details"}
+ and isinstance(error_document.get("code"), str)
+ and bool(error_document["code"])
+ and isinstance(error_document.get("message"), str)
+ and bool(error_document["message"])
+ and isinstance(error_document.get("details"), dict)
+ ):
+ raise DocForgeError(
+ cast(str, error_document["code"]),
+ cast(str, error_document["message"]),
+ **cast(dict[str, object], error_document["details"]),
+ )
+ except DocForgeError as error:
+ if error.code != "projection_worker_failure":
+ raise
+ if completed.returncode < 0:
+ raise _worker_failure(
+ "Detached projection worker terminated by signal",
+ signal=-completed.returncode,
+ )
+ raise _worker_failure(
+ "Detached projection worker exited unsuccessfully",
+ exit_code=completed.returncode,
+ )
+ if type(completed.stdout) is not bytes:
+ raise _worker_failure("Detached projection worker returned invalid output")
+ return _decode_response(validated, completed.stdout)
+
+
+def _render_package(package: ProjectionPackageV1) -> ProjectionRenderResult:
+ renderer = _renderer_identity(package)
+ if renderer["renderer_id"] == _GENERIC_HTML_RENDERER_ID:
+ from docforge_renderers.manual import ManualHtmlRenderer
+
+ result = ManualHtmlRenderer(cast(str, renderer["renderer_version"])).render(package)
+ else:
+ from docforge_renderers.graph import PortableGraphHtmlRenderer
+
+ result = PortableGraphHtmlRenderer().render(package)
+ return _validate_result(package, result, require_peak_memory=False)
+
+
+def _peak_memory_bytes() -> int:
+ peak = int(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss)
+ return max(1, peak if sys.platform == "darwin" else peak * 1024)
+
+
+def _child_response(package: ProjectionPackageV1) -> bytes:
+ result = _render_package(package)
+ original = result.receipt.document
+ receipt = ProjectionReceiptV1.create(
+ kind=package.kind,
+ package_id=package.package_id,
+ plan_id=cast(str, package.document["plan_id"]),
+ renderer=cast(dict[str, object], original["renderer"]),
+ artifacts=[artifact.evidence() for artifact in result.artifacts],
+ diagnostics=cast(dict[str, object], original["diagnostics"]),
+ timing=cast(dict[str, object], original["timing"]),
+ peak_memory_bytes=_peak_memory_bytes(),
+ )
+ validated = _validate_result(
+ package,
+ ProjectionRenderResult(result.artifacts, receipt),
+ require_peak_memory=True,
+ )
+ document: dict[str, object] = {
+ "schema_version": WORKER_PROTOCOL_VERSION,
+ "artifacts": [
+ {
+ "artifact_id": artifact.artifact_id,
+ "media_type": artifact.media_type,
+ "content_base64": base64.b64encode(artifact.content).decode("ascii"),
+ }
+ for artifact in validated.artifacts
+ ],
+ "receipt": validated.receipt.as_dict(),
+ }
+ encoded = canonical_projection_bytes(document) + b"\n"
+ if len(encoded) > MAX_WORKER_RESPONSE_BYTES:
+ raise _worker_failure(
+ "Projection worker response exceeded its fixed boundary",
+ maximum_bytes=MAX_WORKER_RESPONSE_BYTES,
+ )
+ return encoded
+
+
+def _read_child_request() -> ProjectionPackageV1:
+ raw = sys.stdin.buffer.read(MAX_WORKER_REQUEST_BYTES + 1)
+ document = _decode_canonical_line(
+ raw,
+ maximum=MAX_WORKER_REQUEST_BYTES,
+ label="Projection worker request",
+ )
+ return _validated_package(ProjectionPackageV1.from_dict(document))
+
+
+def main(argv: list[str] | None = None) -> int:
+ """Run the closed one-request child protocol."""
+
+ arguments = sys.argv[1:] if argv is None else argv
+ if arguments:
+ return 2
+ try:
+ package = _read_child_request()
+ except Exception:
+ return 2
+ try:
+ response = _child_response(package)
+ sys.stdout.buffer.write(response)
+ sys.stdout.buffer.flush()
+ except DocForgeError as error:
+ response = (
+ canonical_projection_bytes(
+ {
+ "schema_version": WORKER_PROTOCOL_VERSION,
+ "error": error.as_dict(),
+ }
+ )
+ + b"\n"
+ )
+ if len(response) <= MAX_RECEIPT_BYTES:
+ sys.stdout.buffer.write(response)
+ sys.stdout.buffer.flush()
+ return 3
+ except Exception:
+ return 2
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/src/docforge/render_contract.py b/src/docforge/render_contract.py
index aaecf9a..430dc26 100644
--- a/src/docforge/render_contract.py
+++ b/src/docforge/render_contract.py
@@ -3,15 +3,28 @@
from __future__ import annotations
import hashlib
+import html
import json
from dataclasses import dataclass
from importlib.metadata import version
from pathlib import Path
-from typing import Protocol
+from typing import Protocol, cast
from .errors import DocForgeError
from .manual_projection import build_manual_projection_package, build_manual_render_plan
from .models import ProjectSnapshot, RenderView
+from .projection_contract import (
+ ManualRenderPlanV1,
+ ProjectionPackageV1,
+ ProjectionRenderResult,
+)
+from .projection_fragments import (
+ FragmentKey,
+ FragmentRecord,
+ ProjectionFragmentCache,
+ fragment_semantic_hash,
+)
+from .projection_worker import render_projection_in_worker
@dataclass(frozen=True)
@@ -22,6 +35,7 @@ class PreparedRender:
renderer: str
renderer_version: str
template_hash: str
+ projection_receipt: dict[str, object] | None = None
class Renderer(Protocol):
@@ -46,10 +60,13 @@ class GenericHtmlRenderer:
renderer_id = "generic_html"
contract_version = "1"
- def __init__(self) -> None:
+ page_component_version = "manual.page@1"
+
+ def __init__(self, *, incremental: bool = True) -> None:
self.renderer_version = (
f"{self.contract_version}+markdown-it-py-{version('markdown-it-py')}"
)
+ self.incremental = incremental
def prepare(
self,
@@ -98,19 +115,24 @@ class GenericHtmlRenderer:
json.dumps(identity_payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
).hexdigest()
plan = build_manual_render_plan(snapshot, view, changeset_hash=changeset_hash)
- package = build_manual_projection_package(
+ full_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 self.incremental:
+ result = self._incremental_result(
+ snapshot,
+ plan,
+ template_bytes,
+ full_package=full_package,
+ render_identity=render_identity,
+ )
+ else:
+ result = render_projection_in_worker(full_package)
if len(result.artifacts) != 1:
raise DocForgeError(
"invalid_projection",
@@ -124,21 +146,128 @@ class GenericHtmlRenderer:
renderer=self.renderer_id,
renderer_version=self.renderer_version,
template_hash=template_hash,
+ projection_receipt=result.receipt.as_dict(),
)
+ def _incremental_result(
+ self,
+ snapshot: ProjectSnapshot,
+ plan: ManualRenderPlanV1,
+ template_bytes: bytes,
+ *,
+ full_package: ProjectionPackageV1,
+ render_identity: str,
+ ) -> ProjectionRenderResult:
+ cache = ProjectionFragmentCache(
+ snapshot.descriptor.root,
+ snapshot.descriptor.cache_root,
+ )
+ pages = cast(list[object], plan.document["pages"])
+ records: list[FragmentRecord | None] = []
+ missing: list[tuple[int, FragmentKey]] = []
+ for value in pages:
+ page = cast(dict[str, object], value)
+ key = FragmentKey.create(
+ projection_kind="manual",
+ renderer_id=self.renderer_id,
+ renderer_version=self.renderer_version,
+ component_version=self.page_component_version,
+ semantic_input_hash=fragment_semantic_hash(page),
+ )
+ record = cache.get(key)
+ if record is None:
+ missing.append((len(records), key))
+ records.append(record)
+ if not records:
+ return render_projection_in_worker(full_package)
+ full_result: ProjectionRenderResult | None = None
+ new_records: list[FragmentRecord] = []
+ if missing:
+ full_result = render_projection_in_worker(full_package)
+ fragments = self._extract_page_fragments(
+ pages,
+ full_result.artifacts[0].content,
+ )
+ try:
+ for position, key in missing:
+ record = FragmentRecord.create(key, fragments[position])
+ records[position] = record
+ new_records.append(record)
+ except DocForgeError:
+ return full_result
+ complete_records = [record for record in records if record is not None]
+ if len(complete_records) != len(records):
+ return full_result or render_projection_in_worker(full_package)
+ try:
+ incremental_package = build_manual_projection_package(
+ plan,
+ template_bytes,
+ fragment_records=[record.as_dict() for record in complete_records],
+ renderer_id=self.renderer_id,
+ renderer_version=self.renderer_version,
+ max_output_bytes=snapshot.descriptor.limits.max_render_bytes,
+ render_identity=render_identity,
+ )
+ incremental_result = render_projection_in_worker(incremental_package)
+ except DocForgeError:
+ cache.prune(())
+ return full_result or render_projection_in_worker(full_package)
+ if full_result is None:
+ cache.prune(tuple(record.key for record in complete_records))
+ return incremental_result
+ if tuple(
+ (artifact.artifact_id, artifact.media_type, artifact.content)
+ for artifact in incremental_result.artifacts
+ ) != tuple(
+ (artifact.artifact_id, artifact.media_type, artifact.content)
+ for artifact in full_result.artifacts
+ ):
+ cache.prune(())
+ return full_result
+ for record in new_records:
+ cache.put(record.key, record.content)
+ cache.prune(tuple(record.key for record in complete_records))
+ return incremental_result
+
+ @staticmethod
+ def _extract_page_fragments(
+ pages: list[object],
+ output: bytes,
+ ) -> list[bytes]:
+ """Extract exact deterministic page sections from one trusted full artifact."""
+
+ fragments: list[bytes] = []
+ cursor = 0
+ closing = b""
+ for value in pages:
+ page = cast(dict[str, object], value)
+ node_id = cast(str, page["node_id"])
+ marker = f'
'.encode()
+ start = output.find(marker, cursor)
+ end = output.find(closing, start + len(marker)) if start >= 0 else -1
+ if start < 0 or end < 0:
+ raise DocForgeError(
+ "invalid_projection",
+ "Full manual artifact does not contain its planned page fragments",
+ )
+ end += len(closing)
+ fragments.append(output[start:end])
+ cursor = end
+ return fragments
+
_RENDERERS: dict[str, type[GenericHtmlRenderer]] = {
GenericHtmlRenderer.renderer_id: GenericHtmlRenderer
}
-def renderer_for(view: RenderView) -> Renderer:
+def renderer_for(view: RenderView, *, incremental: bool = True) -> Renderer:
factory = _RENDERERS.get(view.renderer)
if factory is None:
raise DocForgeError(
"unsupported_renderer", "View does not name a supported built-in renderer"
)
- return factory()
+ return factory(incremental=incremental)
def relative_output(snapshot: ProjectSnapshot, path: Path) -> str:
diff --git a/src/docforge/rendering.py b/src/docforge/rendering.py
index 79e4116..86738d3 100644
--- a/src/docforge/rendering.py
+++ b/src/docforge/rendering.py
@@ -26,6 +26,8 @@ from .models import (
RenderView,
)
from .project import project_root_fingerprint
+from .projection_contract import ProjectionReceiptV1
+from .projection_policy import ManualProjectionMode, validate_manual_projection_mode
from .render_contract import PreparedRender, relative_output, renderer_for
from .telemetry import increment, stage
@@ -36,9 +38,26 @@ MAX_RENDER_RECEIPT_BYTES = 64_000
class RenderService:
"""Render only declared views through fixed built-in renderer implementations."""
- def __init__(self, project: ProjectService, changesets: ChangesetStore | None = None) -> None:
+ def __init__(
+ self,
+ project: ProjectService,
+ changesets: ChangesetStore | None = None,
+ *,
+ manual_policy: ManualProjectionMode = "explicit",
+ ) -> None:
self.project = project
self.changesets = changesets or ChangesetStore(project)
+ self.manual_policy = validate_manual_projection_mode(manual_policy)
+
+ def _require_rendering(self, operation: str) -> None:
+ if self.manual_policy == "disabled":
+ raise DocForgeError(
+ "projection_policy_forbids_operation",
+ "Manual projection policy disables rendering work",
+ projection="manual",
+ mode=self.manual_policy,
+ operation=operation,
+ )
def status(self, view_id: str | None = None) -> dict[str, object]:
"""Report publication state from bounded receipts without rendering canonical content."""
@@ -86,6 +105,7 @@ class RenderService:
def deep_status(self, view_id: str | None = None) -> dict[str, object]:
"""Recompute render output as the explicit side-effect-free equivalence oracle."""
+ self._require_rendering("deep_status")
snapshot = self.project.load()
config = snapshot.descriptor.render
if config is None:
@@ -107,7 +127,12 @@ class RenderService:
snapshot.descriptor.root,
view.output_path,
)
- prepared, _ = self._prepare(snapshot, view, changeset_hash=None)
+ prepared, _ = self._prepare(
+ snapshot,
+ view,
+ changeset_hash=None,
+ incremental=False,
+ )
state = "missing"
actual_hash: str | None = None
output = view.output_path
@@ -151,6 +176,7 @@ class RenderService:
)
def render(self, view_id: str) -> dict[str, object]:
+ self._require_rendering("render")
with self._lock():
snapshot = self.project.load()
config = self._config(snapshot)
@@ -373,6 +399,8 @@ class RenderService:
"template_file": final_template_file,
"output_file": final_output_file,
}
+ if prepared.projection_receipt is not None:
+ payload["projection_receipt"] = prepared.projection_receipt
raw = json.dumps(payload, sort_keys=True, indent=2).encode("utf-8") + b"\n"
if len(raw) > MAX_RENDER_RECEIPT_BYTES:
raise DocForgeError(
@@ -572,8 +600,10 @@ class RenderService:
renderer = renderer_for(view)
template_file = receipt.get("template_file")
output_file = receipt.get("output_file")
+ fields = set(receipt)
+ projection_receipt = receipt.get("projection_receipt")
return (
- set(receipt) == required
+ fields in (required, required | {"projection_receipt"})
and receipt.get("schema_version") == RENDER_RECEIPT_SCHEMA_VERSION
and receipt.get("project_id") == descriptor.project_id
and receipt.get("project_root_fingerprint") == project_root_fingerprint(descriptor.root)
@@ -602,6 +632,50 @@ class RenderService:
descriptor.limits.max_render_bytes,
)
and cast(dict[str, object], output_file)["size"] == receipt.get("output_bytes")
+ and (
+ projection_receipt is None
+ or self._valid_projection_receipt(
+ projection_receipt,
+ renderer_id=renderer.renderer_id,
+ renderer_version=renderer.renderer_version,
+ output_hash=cast(str, receipt["output_hash"]),
+ output_bytes=cast(int, receipt["output_bytes"]),
+ )
+ )
+ )
+
+ @staticmethod
+ def _valid_projection_receipt(
+ value: object,
+ *,
+ renderer_id: str,
+ renderer_version: str,
+ output_hash: str,
+ output_bytes: int,
+ ) -> bool:
+ if not isinstance(value, dict):
+ return False
+ try:
+ receipt = ProjectionReceiptV1.from_dict(cast(dict[str, object], value))
+ except DocForgeError:
+ return False
+ document = receipt.document
+ return (
+ document.get("kind") == "manual"
+ and document.get("renderer")
+ == {
+ "renderer_id": renderer_id,
+ "renderer_version": renderer_version,
+ }
+ and document.get("artifacts")
+ == [
+ {
+ "artifact_id": "manual.html",
+ "media_type": "text/html; charset=utf-8",
+ "sha256": output_hash,
+ "bytes": output_bytes,
+ }
+ ]
)
@staticmethod
@@ -663,6 +737,7 @@ class RenderService:
"reason": reason,
"verification": "receipt",
"receipt_schema_version": payload.get("schema_version"),
+ "projection_receipt": payload.get("projection_receipt"),
}
def _current_state(self) -> ProjectState | None:
@@ -687,6 +762,7 @@ class RenderService:
}
def preview(self, changeset_id: str, view_id: str) -> dict[str, object]:
+ self._require_rendering("preview")
with self._lock():
snapshot, changeset_hash = self.changesets.projected_snapshot(changeset_id)
config = self._config(snapshot)
@@ -731,11 +807,12 @@ class RenderService:
view: RenderView,
*,
changeset_hash: str | None,
+ incremental: bool = True,
) -> tuple[PreparedRender, bytes]:
increment("render_prepare_calls")
template = self._template_bytes(snapshot, view)
with stage("render.prepare"):
- prepared = renderer_for(view).prepare(
+ prepared = renderer_for(view, incremental=incremental).prepare(
snapshot,
view,
template,
@@ -850,6 +927,7 @@ class RenderService:
"expected_output_hash": prepared.output_hash,
"actual_output_hash": actual_hash,
"template_hash": prepared.template_hash,
+ "projection_receipt": prepared.projection_receipt,
"path": relative_output(snapshot, view.output_path),
"state": state,
}
diff --git a/src/docforge/telemetry.py b/src/docforge/telemetry.py
index 1367273..1571c24 100644
--- a/src/docforge/telemetry.py
+++ b/src/docforge/telemetry.py
@@ -79,6 +79,7 @@ OPERATION_NAMES = frozenset(
"test",
"benchmark.m1",
"benchmark.m2",
+ "benchmark.m3",
"mcp.invoke",
"mcp.bootstrap",
"mcp.sync",
@@ -96,6 +97,8 @@ OPERATION_NAMES = frozenset(
"mcp.generation_diff",
"mcp.validate_project",
"mcp.render_status",
+ "mcp.graph_plan",
+ "mcp.graph_render_status",
"mcp.visualize",
"mcp.visualization_status",
"mcp.stop_visualization",
diff --git a/src/docforge/viewer_manager.py b/src/docforge/viewer_manager.py
index cd8ca88..44f0d63 100644
--- a/src/docforge/viewer_manager.py
+++ b/src/docforge/viewer_manager.py
@@ -28,6 +28,10 @@ from .errors import DocForgeError
from .index import ProjectIndex
from .models import IncrementalStateProject
from .project import project_root_fingerprint
+from .projection_policy import (
+ LiveViewerProjectionMode,
+ validate_live_viewer_projection_mode,
+)
from .telemetry import increment, stage
from .visualization import VISUALIZATION_TEMPLATE, VisualizationIndexSnapshot
@@ -623,9 +627,16 @@ class ViewerManager:
class ViewerManagerClient:
"""Project-bound MCP-side client for the separately supervised manager service."""
- def __init__(self, index: ProjectIndex, *, state_path: Path | None = None) -> None:
+ def __init__(
+ self,
+ index: ProjectIndex,
+ *,
+ state_path: Path | None = None,
+ live_viewer_policy: LiveViewerProjectionMode = "on-demand",
+ ) -> None:
self.index = index
self.state_path = state_path or default_state_path()
+ self.live_viewer_policy = validate_live_viewer_projection_mode(live_viewer_policy)
def start(
self,
@@ -634,6 +645,14 @@ class ViewerManagerClient:
query: str | None = None,
depth: int = 1,
) -> dict[str, object]:
+ if self.live_viewer_policy == "disabled":
+ raise DocForgeError(
+ "projection_policy_forbids_operation",
+ "Live viewer projection policy disables viewer startup",
+ projection="live_viewer",
+ mode=self.live_viewer_policy,
+ operation="start",
+ )
if node_id is not None and query is not None:
raise DocForgeError(
"invalid_visualization_target",
diff --git a/src/docforge_renderers/graph.py b/src/docforge_renderers/graph.py
index a95e322..a36528b 100644
--- a/src/docforge_renderers/graph.py
+++ b/src/docforge_renderers/graph.py
@@ -45,7 +45,7 @@ html[data-enhanced="true"] main[data-mode="flow"] [data-panel="nodes"] { display
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; }
+.muted { color: CanvasText; }
dialog {
max-width: min(42rem, calc(100% - 2rem));
border: 1px solid GrayText; border-radius: .5rem;
diff --git a/src/docforge_renderers/manual.py b/src/docforge_renderers/manual.py
index d54be0f..e816919 100644
--- a/src/docforge_renderers/manual.py
+++ b/src/docforge_renderers/manual.py
@@ -17,6 +17,10 @@ from docforge.projection_contract import (
ProjectionReceiptV1,
ProjectionRenderResult,
)
+from docforge.projection_fragments import (
+ FragmentRecord,
+ fragment_semantic_hash,
+)
_TEMPLATE_TOKEN = re.compile(r"{{\s*([a-z_][a-z0-9_]*)\s*}}")
_ALLOWED_TOKENS = frozenset(
@@ -35,6 +39,7 @@ _ACTIVE_TEMPLATE_CONTENT = re.compile(
r"|<\s*meta\b[^>]*\bhttp-equiv\s*=\s*[\"']?\s*refresh\b",
re.IGNORECASE,
)
+_MANUAL_PAGE_COMPONENT = "manual.page@1"
class ManualHtmlRenderer:
@@ -65,11 +70,15 @@ class ManualHtmlRenderer:
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):
+ if len(assets) not in {1, 2} 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"}
+ set(asset)
+ not in (
+ {"asset_id", "media_type", "sha256", "text"},
+ {"asset_id", "media_type", "sha256", "text", "render_identity"},
+ )
or asset.get("asset_id") != "manual.template"
or asset.get("media_type") != "text/html; charset=utf-8"
or not isinstance(asset.get("text"), str)
@@ -98,11 +107,27 @@ class ManualHtmlRenderer:
"invalid_template",
"Render template must contain docforge_content exactly once",
)
- identity = render_identity or cast(str, plan["plan_id"])
+ packaged_identity = asset.get("render_identity")
+ if packaged_identity is not None and (
+ not isinstance(packaged_identity, str)
+ or len(packaged_identity) != 64
+ or any(character not in "0123456789abcdef" for character in packaged_identity)
+ ):
+ raise DocForgeError(
+ "invalid_projection",
+ "Manual render identity is invalid",
+ )
+ if render_identity is not None and packaged_identity not in {None, render_identity}:
+ raise DocForgeError(
+ "invalid_projection",
+ "Manual render identity is inconsistent",
+ )
+ identity = render_identity or packaged_identity or cast(str, plan["plan_id"])
+ fragments = self._fragments(plan, assets[1:] if len(assets) == 2 else [])
project = cast(dict[str, object], plan["project"])
view = cast(dict[str, object], plan["view"])
replacements = {
- "docforge_content": self._content(plan),
+ "docforge_content": self._content(plan, fragments),
"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),
@@ -131,7 +156,98 @@ class ManualHtmlRenderer:
)
return ProjectionRenderResult((artifact,), receipt)
- def _content(self, plan: dict[str, object]) -> str:
+ def _fragments(
+ self,
+ plan: dict[str, object],
+ assets: list[object],
+ ) -> dict[str, str]:
+ if not assets:
+ return {}
+ asset = assets[0]
+ if not isinstance(asset, dict):
+ raise DocForgeError("invalid_projection", "Manual fragment asset is invalid")
+ document = cast(dict[str, object], asset)
+ if (
+ set(document) != {"asset_id", "media_type", "records"}
+ or document.get("asset_id") != "manual.fragments"
+ or document.get("media_type") != "application/vnd.docforge.projection-fragments.v1+json"
+ or not isinstance(document.get("records"), list)
+ ):
+ raise DocForgeError("invalid_projection", "Manual fragment asset is invalid")
+ pages = cast(list[object], plan["pages"])
+ records = cast(list[object], document["records"])
+ if len(records) != len(pages):
+ raise DocForgeError(
+ "invalid_projection",
+ "Manual fragment inventory is incomplete",
+ )
+ fragments: dict[str, str] = {}
+ for page_value, record_value in zip(pages, records, strict=True):
+ if not isinstance(page_value, dict):
+ raise DocForgeError("invalid_projection", "Manual page is invalid")
+ page = cast(dict[str, object], page_value)
+ record = FragmentRecord.from_dict(record_value)
+ if (
+ record.key.projection_kind != "manual"
+ or record.key.renderer_id != self.renderer_id
+ or record.key.renderer_version != self.renderer_version
+ or record.key.component_version != _MANUAL_PAGE_COMPONENT
+ or record.key.semantic_input_hash != fragment_semantic_hash(page)
+ ):
+ raise DocForgeError(
+ "invalid_projection",
+ "Manual fragment identity does not match its page semantics",
+ )
+ try:
+ fragment = record.content.decode("utf-8")
+ except UnicodeDecodeError as error:
+ raise DocForgeError(
+ "invalid_projection",
+ "Manual fragment content is not valid UTF-8",
+ ) from error
+ node_id = cast(str, page["node_id"])
+ if node_id in fragments or fragment != self.render_page_fragment(page):
+ raise DocForgeError(
+ "invalid_projection",
+ "Manual fragment content does not match its page semantics",
+ )
+ fragments[node_id] = fragment
+ return fragments
+
+ def render_page_fragment(self, page: dict[str, object]) -> str:
+ """Render one page from complete plan semantics without project authority."""
+
+ node_id = cast(str, page["node_id"])
+ sections = [
+ f'',
+ f"{html.escape(cast(str, page['title']))}
",
+ '',
+ f"- ID
- {html.escape(node_id)}
",
+ f"- Family
- {html.escape(cast(str, page['family']))}
",
+ f"- Status
- {html.escape(cast(str, page['status']))}
",
+ f"- Authority
- {html.escape(cast(str, page['authority']))}
",
+ "
",
+ f'{html.escape(cast(str, page["summary"]))}
',
+ self.markdown.render(cast(str, page["content"])).rstrip(),
+ ]
+ relationships = cast(list[object], page["cross_references"])
+ if relationships:
+ sections.append('')
+ for relationship_value in relationships:
+ relationship = cast(dict[str, object], relationship_value)
+ sections.append(
+ f"- {html.escape(cast(str, relationship['relation']))}: "
+ f"{html.escape(cast(str, relationship['target_id']))}
"
+ )
+ sections.append("
")
+ sections.append("")
+ return "\n".join(sections)
+
+ def _content(
+ self,
+ plan: dict[str, object],
+ fragments: dict[str, str],
+ ) -> str:
navigation = ['