"""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, runtime_checkable from .adapter_validation import ( source_payload, source_projection, validate_logic_projection, validate_manifest, validate_projection, validate_source_projection, ) from .config_validation import ID_PATTERN from .errors import DocForgeError from .incremental import ( CachedSource, ExtractionCache, affected_sources, load_extraction_cache, write_extraction_cache, ) from .models import ( Edge, Limits, LogicProjection, Node, ProjectDescriptor, ProjectSnapshot, ProjectState, ProposalWriter, RenderConfig, ) @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() @dataclass(frozen=True) class AdapterSource: """One fingerprinted extraction unit declared by an incremental adapter.""" source_id: str source_path: str fingerprint: str extractor_version: str dependencies: tuple[str, ...] = () @dataclass(frozen=True) class AdapterManifest: """Cheap project identity and source inventory for incremental compilation.""" project_id: str title: str adapter_id: str adapter_version: str root: Path revision: str source_hash: str families: tuple[str, ...] allowed_relations: tuple[str, ...] sources: tuple[AdapterSource, ...] estimated_nodes: int = 10_000 def identity(self) -> tuple[str, str, str, Path]: return self.project_id, self.adapter_id, self.adapter_version, self.root @dataclass(frozen=True) class AdapterSourceProjection: """Nodes, relationships, and optional logic owned by one source unit.""" source_id: str fingerprint: str nodes: tuple[AdapterNode, ...] edges: tuple[AdapterEdge, ...] logic: tuple[LogicProjection, ...] = () @dataclass(frozen=True) class AdapterAssembly: """One finalized graph and Logic set assembled from cached source contributions.""" projection: AdapterProjection logic: tuple[LogicProjection, ...] = () class AdapterLoader(Protocol): """Load one current, deterministic, project-confined adapter projection.""" def load_projection(self) -> AdapterProjection: ... @runtime_checkable class IncrementalAdapterLoader(AdapterLoader, Protocol): """Opt-in adapter contract for cacheable, source-scoped extraction.""" def load_manifest(self) -> AdapterManifest: ... def extract_source(self, source: AdapterSource) -> AdapterSourceProjection: ... @runtime_checkable class IncrementalAdapterAssembler(Protocol): """Optionally normalize overlapping source evidence into one final projection.""" def assemble_projection( self, manifest: AdapterManifest, contributions: tuple[AdapterSourceProjection, ...], ) -> AdapterAssembly: ... 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 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() self._incremental_loader = loader if isinstance(loader, IncrementalAdapterLoader) else None initial: AdapterProjection | None = None manifest: AdapterManifest | None = None if self._incremental_loader is not None: manifest = self._incremental_loader.load_manifest() validate_manifest(manifest) root = manifest.root project_id = manifest.project_id title = manifest.title adapter_id = manifest.adapter_id adapter_version = manifest.adapter_version descriptor_hash = hashlib.sha256( json.dumps( { "project_id": manifest.project_id, "adapter_id": manifest.adapter_id, "adapter_version": manifest.adapter_version, "families": manifest.families, "allowed_relations": manifest.allowed_relations, }, sort_keys=True, separators=(",", ":"), ).encode() ).hexdigest() families = set(manifest.families) allowed_relations = manifest.allowed_relations estimated_nodes = manifest.estimated_nodes else: initial = loader.load_projection() validate_projection(initial) root = initial.root project_id = initial.project_id title = initial.title adapter_id = initial.adapter_id adapter_version = initial.adapter_version descriptor_hash = initial.identity() families = {item.node.family for item in initial.nodes} allowed_relations = tuple(sorted({item.edge.relation for item in initial.edges})) estimated_nodes = len(initial.nodes) 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 = ( project_id, adapter_id, adapter_version, root, ) self._cache_path = resolved_cache / "extractions.json" self._last_build_report: dict[str, object] = { "mode": "full", "cache_hits": 0, "reparsed_sources": 0, "invalidated_sources": 0, "deleted_sources": 0, "total_sources": 0, } self._last_logic: tuple[LogicProjection, ...] = () 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, estimated_nodes), max_context_tokens=64_000, ) self._validate_proposal_writers(families, self.settings.proposal_writers) self._validate_render(root, self.settings.render, content_roots, changeset_root) self.descriptor = ProjectDescriptor( schema_version=1, project_id=project_id, title=title, adapter=f"{adapter_id}@{adapter_version}", root=root, descriptor_path=descriptor_path, descriptor_hash=descriptor_hash, 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=allowed_relations, 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._load_incremental() if self._incremental_loader is not None else 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 = self._combined_source_hash( projection.source_hash, projection.root, canonical_sources, captured, ) return ProjectSnapshot( descriptor=self.descriptor, nodes=projection.core_nodes(), edges=projection.core_edges(), source_hash=source_hash, revision=projection.revision, ) def incremental_state(self) -> ProjectState | None: """Return current source identity without reconstructing the complete projection.""" loader = self._incremental_loader if loader is None: return None canonical_sources = self.canonical_source_paths() captured = {path: path.read_bytes() for path in canonical_sources} manifest = loader.load_manifest() validate_manifest(manifest) if manifest.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" ) return ProjectState( source_hash=self._combined_source_hash( manifest.source_hash, manifest.root, canonical_sources, captured, ), revision=manifest.revision, ) def build_report(self) -> dict[str, object]: """Return deterministic extraction metrics from the most recent load.""" return dict(self._last_build_report) def logic_projection(self, owner_node_id: str) -> LogicProjection | None: """Load one lazily stored function-scoped logic projection.""" self.load() return next( ( projection for projection in self._last_logic if projection.owner_node_id == owner_node_id ), None, ) def logic_projections(self) -> tuple[LogicProjection, ...]: """Return logic captured by the most recent validated project load.""" return self._last_logic def verify_incremental_equivalence(self) -> dict[str, object]: """Prove the incremental and full loader contracts produce the same graph.""" if self._incremental_loader is None: raise DocForgeError( "incremental_disabled", "Adapter does not implement incremental extraction" ) incremental = self._load_incremental() full = self.loader.load_projection() validate_projection(full) fields = { "project_id": incremental.project_id == full.project_id, "adapter_id": incremental.adapter_id == full.adapter_id, "adapter_version": incremental.adapter_version == full.adapter_version, "revision": incremental.revision == full.revision, "source_hash": incremental.source_hash == full.source_hash, "nodes": incremental.nodes == full.nodes, "edges": incremental.edges == full.edges, } mismatches = [field for field, matches in fields.items() if not matches] if mismatches: raise DocForgeError( "incremental_mismatch", "Incremental extraction does not match a full adapter projection", fields=mismatches, ) return { "status": "ok", "project_id": incremental.project_id, "revision": incremental.revision, "source_hash": incremental.source_hash, "node_count": len(incremental.nodes), "edge_count": len(incremental.edges), } def _load_incremental(self) -> AdapterProjection: loader = self._incremental_loader if loader is None: raise DocForgeError("incremental_disabled", "Incremental adapter is not configured") manifest = loader.load_manifest() validate_manifest(manifest) if manifest.identity() != self._identity: raise DocForgeError("adapter_changed", "Adapter identity changed during the operation") cache = load_extraction_cache( self._cache_path, project_id=manifest.project_id, adapter_id=manifest.adapter_id, adapter_version=manifest.adapter_version, ) cached = {source.source_id: source for source in cache.sources} if cache else {} current = {source.source_id: source for source in manifest.sources} changed = { source_id for source_id, source in current.items() if source_id not in cached or cached[source_id].source_path != source.source_path or cached[source_id].fingerprint != source.fingerprint or cached[source_id].extractor_version != source.extractor_version or cached[source_id].dependencies != source.dependencies } deleted = set(cached) - set(current) invalidated = affected_sources( current_dependencies={ source.source_id: source.dependencies for source in manifest.sources }, cached_dependencies={ source.source_id: source.dependencies for source in cached.values() }, changed=changed | deleted, ) contributions: list[AdapterSourceProjection] = [] cache_records: list[CachedSource] = [] reparsed: list[str] = [] hits: list[str] = [] for source in manifest.sources: if source.source_id in invalidated: contribution = loader.extract_source(source) reparsed.append(source.source_id) cache_record = CachedSource( source_id=source.source_id, source_path=source.source_path, fingerprint=source.fingerprint, extractor_version=source.extractor_version, dependencies=source.dependencies, payload=source_payload(contribution), ) else: cache_record = cached[source.source_id] contribution = source_projection(cache_record.payload) hits.append(source.source_id) validate_source_projection(source, contribution) contributions.append(contribution) cache_records.append(cache_record) if isinstance(loader, IncrementalAdapterAssembler): assembly = loader.assemble_projection(manifest, tuple(contributions)) projection = assembly.projection logic_projections = tuple( sorted( assembly.logic, key=lambda projection: projection.owner_node_id, ) ) else: projection = AdapterProjection( project_id=manifest.project_id, title=manifest.title, adapter_id=manifest.adapter_id, adapter_version=manifest.adapter_version, root=manifest.root, revision=manifest.revision, source_hash=manifest.source_hash, nodes=tuple( sorted( (node for contribution in contributions for node in contribution.nodes), key=lambda item: item.node.node_id, ) ), edges=tuple( sorted( (edge for contribution in contributions for edge in contribution.edges), key=lambda item: ( item.edge.source_id, item.edge.relation, item.edge.target_id, ), ) ), ) logic_projections = tuple( sorted( (logic for contribution in contributions for logic in contribution.logic), key=lambda projection: projection.owner_node_id, ) ) identity = ( projection.project_id, projection.adapter_id, projection.adapter_version, projection.root, ) if ( identity != manifest.identity() or projection.title != manifest.title or projection.revision != manifest.revision or projection.source_hash != manifest.source_hash ): raise DocForgeError( "invalid_adapter", "Incremental assembly changed the manifest-bound project identity", ) validate_projection(projection) owners = [projection.owner_node_id for projection in logic_projections] if len(owners) != len(set(owners)): raise DocForgeError( "invalid_adapter", "A primary graph node may own only one logic projection" ) node_ids = {item.node.node_id for item in projection.nodes} for logic_projection in logic_projections: validate_logic_projection(logic_projection) if logic_projection.owner_node_id not in node_ids: raise DocForgeError( "invalid_adapter", "A Logic projection owner must exist in the assembled primary graph", owner_node_id=logic_projection.owner_node_id, ) stable = loader.load_manifest() validate_manifest(stable) if stable != manifest: raise DocForgeError( "source_changed", "Adapter sources changed during incremental extraction" ) if cache is None or invalidated or deleted: write_extraction_cache( self._cache_path, ExtractionCache( project_id=manifest.project_id, adapter_id=manifest.adapter_id, adapter_version=manifest.adapter_version, sources=tuple(cache_records), ), ) self._last_logic = logic_projections self._last_build_report = { "mode": "incremental", "cache_hits": len(hits), "reparsed_sources": len(reparsed), "invalidated_sources": len(invalidated & set(current)), "deleted_sources": len(deleted), "total_sources": len(current), "cache_hit_ids": hits, "reparsed_source_ids": reparsed, } return projection 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", ) @staticmethod def _combined_source_hash( projection_hash: str, root: Path, canonical_sources: tuple[Path, ...], captured: dict[Path, bytes], ) -> str: if not captured: return projection_hash digest = hashlib.sha256(projection_hash.encode("ascii")) for path in canonical_sources: relative = path.relative_to(root).as_posix().encode() digest.update(len(relative).to_bytes(8, "big")) digest.update(relative) digest.update(hashlib.sha256(captured[path]).digest()) return digest.hexdigest() 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(families: set[str], writers: tuple[ProposalWriter, ...]) -> None: writer_ids = [writer.writer_id for writer in writers] 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 }, }