747 lines
28 KiB
Python
747 lines
28 KiB
Python
"""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_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, ...] = ()
|
|
|
|
|
|
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: ...
|
|
|
|
|
|
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 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)
|
|
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,
|
|
),
|
|
)
|
|
),
|
|
)
|
|
validate_projection(projection)
|
|
logic_projections = tuple(
|
|
sorted(
|
|
(logic for contribution in contributions for logic in contribution.logic),
|
|
key=lambda projection: projection.owner_node_id,
|
|
)
|
|
)
|
|
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"
|
|
)
|
|
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
|
|
},
|
|
}
|