272 lines
9.9 KiB
Python
272 lines
9.9 KiB
Python
|
"""Reusable contracts for explicit project adapters and shadow verification."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
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,
|
||
|
|
)
|
||
|
|
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: ...
|
||
|
|
|
||
|
|
|
||
|
|
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) -> None:
|
||
|
|
self.loader = loader
|
||
|
|
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,
|
||
|
|
)
|
||
|
|
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=root / ".docforge" / "shadow-adapter.toml",
|
||
|
|
descriptor_hash=initial.identity(),
|
||
|
|
content_roots=(),
|
||
|
|
authority_files=(),
|
||
|
|
cache_root=resolved_cache,
|
||
|
|
index_path=resolved_cache / "index.sqlite3",
|
||
|
|
changeset_root=resolved_cache / "changesets-disabled",
|
||
|
|
proposal_writers=(),
|
||
|
|
render=None,
|
||
|
|
allowed_relations=tuple(sorted({item.edge.relation for item in initial.edges})),
|
||
|
|
profiles=(),
|
||
|
|
limits=Limits(
|
||
|
|
max_nodes=max(10_000, len(initial.nodes)),
|
||
|
|
max_context_tokens=64_000,
|
||
|
|
),
|
||
|
|
)
|
||
|
|
|
||
|
|
def load(self) -> ProjectSnapshot:
|
||
|
|
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")
|
||
|
|
return ProjectSnapshot(
|
||
|
|
descriptor=self.descriptor,
|
||
|
|
nodes=projection.core_nodes(),
|
||
|
|
edges=projection.core_edges(),
|
||
|
|
source_hash=projection.source_hash,
|
||
|
|
revision=projection.revision,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@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
|
||
|
|
},
|
||
|
|
}
|