"""Reusable contracts for explicit project adapters and shadow verification.""" from __future__ import annotations import hashlib import json from collections.abc import Callable, Mapping from dataclasses import dataclass from pathlib import Path from typing import Protocol from .config_validation import AUTHORITIES, ID_PATTERN from .errors import DocForgeError from .models import ( Edge, Limits, Node, ProjectDescriptor, ProjectSnapshot, ProposalWriter, RenderConfig, ) from .project import validate_graph @dataclass(frozen=True) class AdapterNode: """One core node plus deterministic adapter-owned metadata.""" node: Node metadata: tuple[tuple[str, str], ...] = () def as_dict(self) -> dict[str, object]: return {"node": self.node.as_dict(), "metadata": dict(self.metadata)} @dataclass(frozen=True) class AdapterEdge: """One core edge plus deterministic adapter-owned metadata.""" edge: Edge metadata: tuple[tuple[str, str], ...] = () def as_dict(self) -> dict[str, object]: return {"edge": self.edge.as_dict(), "metadata": dict(self.metadata)} @dataclass(frozen=True) class AdapterProjection: """A complete immutable graph projection supplied by one project adapter.""" project_id: str title: str adapter_id: str adapter_version: str root: Path revision: str source_hash: str nodes: tuple[AdapterNode, ...] edges: tuple[AdapterEdge, ...] def core_nodes(self) -> tuple[Node, ...]: return tuple(item.node for item in self.nodes) def core_edges(self) -> tuple[Edge, ...]: return tuple(item.edge for item in self.edges) def identity(self) -> str: payload = { "project_id": self.project_id, "adapter_id": self.adapter_id, "adapter_version": self.adapter_version, "revision": self.revision, "source_hash": self.source_hash, "nodes": [item.as_dict() for item in self.nodes], "edges": [item.as_dict() for item in self.edges], } encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() return hashlib.sha256(encoded).hexdigest() class AdapterLoader(Protocol): """Load one current, deterministic, project-confined adapter projection.""" def load_projection(self) -> AdapterProjection: ... ProposalValidator = Callable[ [ ProjectSnapshot, ProjectSnapshot, tuple[Mapping[str, object], ...], ], None, ] @dataclass(frozen=True) class AdapterProjectSettings: """Optional confined proposal and preview policy supplied by an explicit adapter.""" descriptor_path: Path | None = None content_roots: tuple[Path, ...] = () authority_files: tuple[Path, ...] = () canonical_sources: tuple[Path, ...] = () changeset_root: Path | None = None proposal_writers: tuple[ProposalWriter, ...] = () render: RenderConfig | None = None limits: Limits | None = None proposal_validator: ProposalValidator | None = None def validate_projection(projection: AdapterProjection) -> None: """Validate generic invariants without interpreting adapter metadata.""" root = projection.root.resolve(strict=True) if not root.is_dir() or projection.root != root: raise DocForgeError("invalid_adapter", "Adapter root must be a resolved directory") for label, value in ( ("project_id", projection.project_id), ("title", projection.title), ("adapter_id", projection.adapter_id), ("adapter_version", projection.adapter_version), ("revision", projection.revision), ("source_hash", projection.source_hash), ): if not value.strip(): raise DocForgeError("invalid_adapter", f"Adapter {label} must not be empty") if ID_PATTERN.fullmatch(projection.project_id) is None: raise DocForgeError("invalid_adapter", "Adapter project ID is invalid") if len(projection.source_hash) != 64 or any( character not in "0123456789abcdef" for character in projection.source_hash ): raise DocForgeError("invalid_adapter", "Adapter source hash must be lowercase SHA-256") ordered_nodes = tuple(sorted(projection.nodes, key=lambda item: item.node.node_id)) ordered_edges = tuple( sorted( projection.edges, key=lambda item: ( item.edge.source_id, item.edge.relation, item.edge.target_id, ), ) ) if projection.nodes != ordered_nodes or projection.edges != ordered_edges: raise DocForgeError( "invalid_adapter", "Adapter projection must be deterministically ordered" ) for item in (*projection.nodes, *projection.edges): keys = [key for key, _ in item.metadata] if keys != sorted(keys) or len(keys) != len(set(keys)): raise DocForgeError( "invalid_adapter", "Adapter metadata keys must be unique and ordered" ) for item in projection.nodes: node = item.node source = Path(node.source_path) if ID_PATTERN.fullmatch(node.node_id) is None: raise DocForgeError("invalid_adapter", "Adapter node ID is invalid", id=node.node_id) if node.authority not in AUTHORITIES: raise DocForgeError( "invalid_adapter", "Adapter node authority is invalid", id=node.node_id ) if ( not node.title.strip() or not node.family.strip() or not node.status.strip() or not node.summary.strip() or not node.content.strip() ): raise DocForgeError( "invalid_adapter", "Adapter node has empty required content", id=node.node_id ) if source.is_absolute() or ".." in source.parts or not node.source_path: raise DocForgeError( "invalid_adapter", "Adapter node source path is unsafe", id=node.node_id ) if len(node.tags) != len(set(node.tags)) or any(not tag for tag in node.tags): raise DocForgeError("invalid_adapter", "Adapter node tags are invalid", id=node.node_id) if len(node.content_hash) != 64 or any( character not in "0123456789abcdef" for character in node.content_hash ): raise DocForgeError( "invalid_adapter", "Adapter node content hash is invalid", id=node.node_id ) for item in projection.edges: if ID_PATTERN.fullmatch(item.edge.relation) is None: raise DocForgeError("invalid_adapter", "Adapter relationship type is invalid") validate_graph(projection.core_nodes(), projection.core_edges()) class AdapterProject: """Expose a validated adapter projection through the standard index boundary.""" def __init__( self, loader: AdapterLoader, *, cache_root: Path, settings: AdapterProjectSettings | None = None, ) -> None: self.loader = loader self.settings = settings or AdapterProjectSettings() initial = loader.load_projection() validate_projection(initial) root = initial.root resolved_cache = cache_root.resolve(strict=False) if ( resolved_cache == root or not resolved_cache.is_relative_to(root) or resolved_cache.is_symlink() ): raise DocForgeError( "path_escape", "Adapter cache must be a confined project subdirectory" ) self._identity = ( initial.project_id, initial.adapter_id, initial.adapter_version, initial.root, ) content_roots = self._resolved_directories( root, self.settings.content_roots, label="content root" ) authority_files = self._resolved_files( root, self.settings.authority_files, label="authority file" ) canonical_sources = self._resolved_files( root, self.settings.canonical_sources, label="canonical source" ) if canonical_sources and any( not any(source.is_relative_to(content_root) for content_root in content_roots) for source in canonical_sources ): raise DocForgeError( "invalid_adapter", "Adapter canonical sources must be inside a declared content root", ) changeset_root = self._resolved_output( root, self.settings.changeset_root or resolved_cache / "changesets-disabled", label="changeset root", ) if any( changeset_root == content_root or changeset_root.is_relative_to(content_root) or content_root.is_relative_to(changeset_root) for content_root in content_roots ): raise DocForgeError( "invalid_adapter", "Adapter changesets must not overlap canonical content" ) descriptor_path = self.settings.descriptor_path or ( root / ".docforge" / "shadow-adapter.toml" ) if self.settings.proposal_writers and ( not descriptor_path.is_file() or descriptor_path.is_symlink() or not descriptor_path.resolve().is_relative_to(root) ): raise DocForgeError( "invalid_adapter", "Proposal-enabled adapters require a confined descriptor file", ) limits = self.settings.limits or Limits( max_nodes=max(10_000, len(initial.nodes)), max_context_tokens=64_000, ) self._validate_proposal_writers(initial, self.settings.proposal_writers) self._validate_render(root, self.settings.render, content_roots, changeset_root) self.descriptor = ProjectDescriptor( schema_version=1, project_id=initial.project_id, title=initial.title, adapter=f"{initial.adapter_id}@{initial.adapter_version}", root=root, descriptor_path=descriptor_path, descriptor_hash=initial.identity(), content_roots=content_roots, authority_files=authority_files, cache_root=resolved_cache, index_path=resolved_cache / "index.sqlite3", changeset_root=changeset_root, proposal_writers=self.settings.proposal_writers, render=self.settings.render, allowed_relations=tuple(sorted({item.edge.relation for item in initial.edges})), profiles=(), limits=limits, ) self._canonical_sources = canonical_sources def load(self) -> ProjectSnapshot: canonical_sources = self.canonical_source_paths() captured = {path: path.read_bytes() for path in canonical_sources} projection = self.loader.load_projection() validate_projection(projection) identity = ( projection.project_id, projection.adapter_id, projection.adapter_version, projection.root, ) if identity != self._identity: raise DocForgeError("adapter_changed", "Adapter identity changed during the operation") if self.canonical_source_paths() != canonical_sources or any( not path.is_file() or path.read_bytes() != raw for path, raw in captured.items() ): raise DocForgeError( "source_changed", "Adapter canonical source changed during the operation" ) source_hash = projection.source_hash if captured: digest = hashlib.sha256(projection.source_hash.encode("ascii")) for path in canonical_sources: relative = path.relative_to(projection.root).as_posix().encode() digest.update(len(relative).to_bytes(8, "big")) digest.update(relative) digest.update(hashlib.sha256(captured[path]).digest()) source_hash = digest.hexdigest() return ProjectSnapshot( descriptor=self.descriptor, nodes=projection.core_nodes(), edges=projection.core_edges(), source_hash=source_hash, revision=projection.revision, ) def canonical_source_paths(self) -> tuple[Path, ...]: """Adapters validate their own source sets before producing a projection.""" return self._resolved_files( self.descriptor.root, self._canonical_sources, label="canonical source", ) def validate_proposal( self, base: ProjectSnapshot, projected: ProjectSnapshot, operations: tuple[Mapping[str, object], ...], ) -> None: validator = self.settings.proposal_validator if validator is None: if operations: raise DocForgeError( "proposal_policy_missing", "Adapter does not define proposal validation policy", ) return validator(base, projected, operations) @staticmethod def _resolved_directories( root: Path, paths: tuple[Path, ...], *, label: str ) -> tuple[Path, ...]: resolved: list[Path] = [] for path in paths: if path.is_symlink(): raise DocForgeError("path_escape", f"Adapter {label} is unsafe") current = path.resolve(strict=True) if not current.is_dir() or current.is_symlink() or not current.is_relative_to(root): raise DocForgeError("path_escape", f"Adapter {label} is unsafe") resolved.append(current) if len(resolved) != len(set(resolved)): raise DocForgeError("invalid_adapter", f"Adapter {label}s repeat") return tuple(sorted(resolved)) @staticmethod def _resolved_files(root: Path, paths: tuple[Path, ...], *, label: str) -> tuple[Path, ...]: resolved: list[Path] = [] for path in paths: if path.is_symlink(): raise DocForgeError("path_escape", f"Adapter {label} is unsafe") current = path.resolve(strict=True) if not current.is_file() or current.is_symlink() or not current.is_relative_to(root): raise DocForgeError("path_escape", f"Adapter {label} is unsafe") resolved.append(current) if len(resolved) != len(set(resolved)): raise DocForgeError("invalid_adapter", f"Adapter {label}s repeat") return tuple(sorted(resolved)) @staticmethod def _resolved_output(root: Path, path: Path, *, label: str) -> Path: if path.is_symlink(): raise DocForgeError("path_escape", f"Adapter {label} is unsafe") current = path.resolve(strict=False) if current == root or current.is_symlink() or not current.is_relative_to(root): raise DocForgeError("path_escape", f"Adapter {label} is unsafe") return current @staticmethod def _validate_proposal_writers( projection: AdapterProjection, writers: tuple[ProposalWriter, ...] ) -> None: writer_ids = [writer.writer_id for writer in writers] families = {item.node.family for item in projection.nodes} if len(writer_ids) != len(set(writer_ids)): raise DocForgeError("invalid_adapter", "Adapter proposal writer IDs repeat") for writer in writers: if ( ID_PATTERN.fullmatch(writer.writer_id) is None or not writer.families or not set(writer.families).issubset(families) or not writer.operations or not set(writer.operations).issubset({"create", "update", "move", "delete"}) ): raise DocForgeError("invalid_adapter", "Adapter proposal writer policy is invalid") @classmethod def _validate_render( cls, root: Path, render: RenderConfig | None, content_roots: tuple[Path, ...], changeset_root: Path, ) -> None: if render is None: return template_roots = cls._resolved_directories( root, (render.template_root,), label="template root" ) template_root = template_roots[0] preview_root = cls._resolved_output(root, render.preview_root, label="preview root") protected = (*content_roots, changeset_root, template_root) if any( preview_root == path or preview_root.is_relative_to(path) or path.is_relative_to(preview_root) for path in protected ): raise DocForgeError("invalid_adapter", "Adapter preview root overlaps a protected path") view_ids: set[str] = set() outputs: set[Path] = set() for view in render.views: template = cls._resolved_files(root, (view.template_path,), label="render template")[0] output = cls._resolved_output(root, view.output_path, label="render output") if ( ID_PATTERN.fullmatch(view.view_id) is None or view.view_id in view_ids or view.renderer != "generic_html" or not template.is_relative_to(template_root) or output.suffix != ".html" or output in outputs or any(output == path or output.is_relative_to(path) for path in protected) or output == preview_root or output.is_relative_to(preview_root) ): raise DocForgeError("invalid_adapter", "Adapter render view is invalid") view_ids.add(view.view_id) outputs.add(output) @dataclass(frozen=True) class ShadowArtifact: """One named deterministic byte artifact used by a shadow comparison.""" artifact_id: str content: bytes @property def sha256(self) -> str: return hashlib.sha256(self.content).hexdigest() def compare_artifacts( reference: tuple[ShadowArtifact, ...], candidate: tuple[ShadowArtifact, ...] ) -> dict[str, object]: """Compare complete artifact sets without writing either side.""" reference_by_id = {item.artifact_id: item for item in reference} candidate_by_id = {item.artifact_id: item for item in candidate} if len(reference_by_id) != len(reference) or len(candidate_by_id) != len(candidate): raise DocForgeError("duplicate_artifact", "Shadow artifact IDs must be unique") missing = sorted(set(reference_by_id) - set(candidate_by_id)) unexpected = sorted(set(candidate_by_id) - set(reference_by_id)) changed = sorted( artifact_id for artifact_id in set(reference_by_id) & set(candidate_by_id) if reference_by_id[artifact_id].content != candidate_by_id[artifact_id].content ) return { "status": "ok" if not missing and not unexpected and not changed else "mismatch", "count": len(reference), "missing": missing, "unexpected": unexpected, "changed": changed, "hashes": { artifact_id: reference_by_id[artifact_id].sha256 for artifact_id in sorted(reference_by_id) if artifact_id in candidate_by_id and reference_by_id[artifact_id].content == candidate_by_id[artifact_id].content }, }