"""Validation and hashing for the versioned isolated changeset contract.""" from __future__ import annotations import hashlib import json import re from pathlib import Path from typing import Any from .errors import DocForgeError from .models import Node, ProjectService from .project import project_root_fingerprint ID_PATTERN = re.compile(r"[a-z0-9][a-z0-9._-]{1,127}") HASH_PATTERN = re.compile(r"[0-9a-f]{64}") CHANGESET_KEYS = frozenset( { "schema_version", "changeset_id", "project_id", "root_fingerprint", "base_revision", "base_source_hash", "creator", "operations", } ) OPERATION_KEYS = frozenset( { "sequence", "operation", "node_id", "expected_content_hash", "target_source", "metadata", "content", "relationship_changes", "rationale", } ) METADATA_KEYS = frozenset( {"title", "family", "authority", "status", "tags", "summary", "source_anchor"} ) REQUIRED_CREATE_METADATA = frozenset({"title", "family", "authority", "status", "tags", "summary"}) RELATIONSHIP_KEYS = frozenset({"action", "source_id", "relation", "target_id"}) OPERATIONS = frozenset({"create", "update", "move", "delete"}) def canonical_bytes(document: dict[str, Any]) -> bytes: return json.dumps(document, sort_keys=True, separators=(",", ":")).encode("utf-8") def document_hash(document: dict[str, Any]) -> str: return hashlib.sha256(canonical_bytes(document)).hexdigest() def node_metadata(node: Node) -> dict[str, Any]: return { "title": node.title, "family": node.family, "authority": node.authority, "status": node.status, "tags": list(node.tags), "summary": node.summary, "source_anchor": node.source_anchor, } def edge_key(change: dict[str, Any]) -> tuple[str, str, str]: return change["source_id"], change["relation"], change["target_id"] def edge_dict(edge: tuple[str, str, str], *, action: str) -> dict[str, str]: return { "action": action, "source_id": edge[0], "relation": edge[1], "target_id": edge[2], } def validate_id(value: Any, field: str) -> None: if not isinstance(value, str) or ID_PATTERN.fullmatch(value) is None: raise DocForgeError("invalid_id", f"{field} must be a stable ID", field=field) def validate_hash(value: Any, field: str) -> None: if not isinstance(value, str) or HASH_PATTERN.fullmatch(value) is None: raise DocForgeError("invalid_hash", f"{field} must be a lowercase SHA-256 hash") def normalize_operation(operation: dict[str, Any], *, sequence: int) -> dict[str, Any]: allowed = OPERATION_KEYS if "sequence" in operation else OPERATION_KEYS - {"sequence"} unknown = sorted(set(operation) - allowed) if unknown: raise DocForgeError("invalid_operation", "Operation has unknown fields", fields=unknown) missing = sorted((OPERATION_KEYS - {"sequence"}) - set(operation)) if missing: raise DocForgeError("invalid_operation", "Operation is missing fields", fields=missing) kind = operation["operation"] if kind not in OPERATIONS: raise DocForgeError("invalid_operation", "Operation type is invalid", operation=kind) validate_id(operation["node_id"], "node_id") rationale = operation["rationale"] if not isinstance(rationale, str) or not rationale.strip(): raise DocForgeError("invalid_operation", "Operation rationale must be non-empty") relationships = operation["relationship_changes"] if not isinstance(relationships, list): raise DocForgeError("invalid_operation", "relationship_changes must be an array") normalized_relationships: list[dict[str, str]] = [] for change in relationships: if not isinstance(change, dict) or set(change) != RELATIONSHIP_KEYS: raise DocForgeError( "invalid_operation", "Relationship changes require exact structured fields" ) if change["action"] not in {"add", "remove"}: raise DocForgeError("invalid_operation", "Relationship action is invalid") for field in ("source_id", "relation", "target_id"): validate_id(change[field], field) normalized_relationships.append( { "action": change["action"], "source_id": change["source_id"], "relation": change["relation"], "target_id": change["target_id"], } ) keys = [ (item["action"], item["source_id"], item["relation"], item["target_id"]) for item in normalized_relationships ] if len(keys) != len(set(keys)): raise DocForgeError("invalid_operation", "Relationship changes contain duplicates") metadata = operation["metadata"] if metadata is not None and not isinstance(metadata, dict): raise DocForgeError("invalid_operation", "metadata must be an object or null") if isinstance(metadata, dict): unknown_metadata = sorted(set(metadata) - METADATA_KEYS) if unknown_metadata: raise DocForgeError( "invalid_operation", "Node metadata has unknown fields", fields=unknown_metadata ) content = operation["content"] if content is not None and not isinstance(content, str): raise DocForgeError("invalid_operation", "content must be text or null") target = operation["target_source"] if target is not None and not isinstance(target, str): raise DocForgeError("invalid_operation", "target_source must be text or null") expected = operation["expected_content_hash"] if expected is not None: validate_hash(expected, "expected_content_hash") return { "sequence": sequence, "operation": kind, "node_id": operation["node_id"], "expected_content_hash": expected, "target_source": target, "metadata": metadata, "content": content, "relationship_changes": sorted( normalized_relationships, key=lambda item: ( item["source_id"], item["relation"], item["target_id"], item["action"], ), ), "rationale": rationale.strip(), } def validate_document(project: ProjectService, document: Any, *, path: Path) -> dict[str, Any]: if not isinstance(document, dict) or set(document) != CHANGESET_KEYS: raise DocForgeError( "invalid_changeset", "Changeset has missing or unknown fields", path=path.name ) if document["schema_version"] != 1: raise DocForgeError("invalid_changeset", "Changeset schema_version must be 1") validate_id(document["changeset_id"], "changeset_id") if path.name != f"{document['changeset_id']}.json": raise DocForgeError("invalid_changeset", "Changeset ID does not match its file name") descriptor = project.descriptor identity = { "project_id": descriptor.project_id, "root_fingerprint": project_root_fingerprint(descriptor.root), } for field, expected in identity.items(): if document[field] != expected: raise DocForgeError( "changeset_identity_mismatch", "Changeset belongs to another project", field=field, expected=expected, actual=document[field], ) if not isinstance(document["base_revision"], str) or not document["base_revision"]: raise DocForgeError("invalid_changeset", "base_revision must be non-empty") validate_hash(document["base_source_hash"], "base_source_hash") validate_id(document["creator"], "creator") writers = {writer.writer_id for writer in descriptor.proposal_writers} if document["creator"] not in writers: raise DocForgeError( "unknown_writer", "Changeset creator is not configured", writer=document["creator"] ) operations = document["operations"] if not isinstance(operations, list): raise DocForgeError("invalid_changeset", "operations must be an array") if len(operations) > descriptor.limits.max_changeset_operations: raise DocForgeError("changeset_operation_limit", "Changeset has too many operations") normalized: list[dict[str, Any]] = [] for index, operation in enumerate(operations, start=1): if not isinstance(operation, dict) or set(operation) != OPERATION_KEYS: raise DocForgeError( "invalid_changeset", "Stored operation has missing or unknown fields" ) if operation["sequence"] != index: raise DocForgeError("invalid_changeset", "Operation sequence is not contiguous") normalized.append(normalize_operation(operation, sequence=index)) if len({item["node_id"] for item in normalized}) != len(normalized): raise DocForgeError("duplicate_operation", "A changeset may touch a node only once") return {**document, "operations": normalized}