"""Offline, pure conformance validators for TRF Stage 0 schemas. Validates against the .schema.json files under schemas/ (structural conformance) plus a small set of business rules that JSON Schema cannot express: cross-record currency consistency (working default Q6), manifest immutability across versions (Rule 1), and the extension-contract rule that `allocation.rule` may not redefine core terms (TSD §3.3, PRD FR-4). Every function here is pure: given the same input dict(s), it always produces the same result, with no I/O beyond reading the static schema files at import/call time (TSD §6.1 determinism). """ from __future__ import annotations import json from pathlib import Path from typing import Any import jsonschema _SCHEMAS_DIR = Path(__file__).resolve().parents[2] / "schemas" # TSD §6.2/§3.3: an extension's allocation.rule must not redefine these. CORE_TERMS = [ "Phase", "Milestone Release", "Initial Target", "Development Credit", "Remission Credit", "Outstanding Target", "Conversion Event", "Future License", ] class ConformanceError(ValueError): """Raised with all collected validation errors for a single record.""" def __init__(self, errors: list[str]): self.errors = errors super().__init__("; ".join(errors)) def _load_schema(filename: str) -> dict[str, Any]: with open(_SCHEMAS_DIR / filename, encoding="utf-8") as f: return json.load(f) def _schema_errors(schema_filename: str, record: dict[str, Any]) -> list[str]: schema = _load_schema(schema_filename) validator = jsonschema.Draft202012Validator(schema) return [ f"{'/'.join(str(p) for p in e.path) or ''}: {e.message}" for e in sorted(validator.iter_errors(record), key=lambda e: list(e.path)) ] # --- Phase Manifest -------------------------------------------------------- def validate_phase_manifest(manifest: dict[str, Any]) -> None: """Raise ConformanceError if the manifest is not schema-conformant.""" errors = _schema_errors("phase_manifest.schema.json", manifest) if errors: raise ConformanceError(errors) def check_manifest_immutability( published: dict[str, Any], candidate: dict[str, Any] ) -> list[str]: """Compare a published manifest against a proposed replacement. Returns a list of violations (empty if none). Per Rule 1: phase.id must not change, and phase.initial_target.amount must not change except through an explicit, versioned correction record distinct from this check (Stage 0 has no such record type yet, so any change here is flagged). """ errors: list[str] = [] old_phase = published.get("phase", {}) new_phase = candidate.get("phase", {}) if old_phase.get("id") != new_phase.get("id"): errors.append( f"phase.id changed from {old_phase.get('id')!r} to {new_phase.get('id')!r}" ) old_target = old_phase.get("initial_target", {}) new_target = new_phase.get("initial_target", {}) if old_target.get("amount") != new_target.get("amount"): errors.append( "phase.initial_target.amount changed from " f"{old_target.get('amount')!r} to {new_target.get('amount')!r} " "without an explicit, versioned correction record" ) if old_target.get("currency") != new_target.get("currency"): errors.append( "phase.initial_target.currency changed from " f"{old_target.get('currency')!r} to {new_target.get('currency')!r}" ) return errors # --- Target Ledger Entry ---------------------------------------------------- def validate_ledger_entry(entry: dict[str, Any]) -> None: """Raise ConformanceError if the entry is not schema-conformant.""" errors = _schema_errors("ledger_entry.schema.json", entry) if errors: raise ConformanceError(errors) def check_currency_consistency( manifest: dict[str, Any], entries: list[dict[str, Any]] ) -> list[str]: """Working default Q6: every entry's currency must match the Phase's.""" native = manifest.get("phase", {}).get("initial_target", {}).get("currency") errors = [] for entry in entries: if entry.get("currency") != native: errors.append( f"entry {entry.get('id')!r} currency {entry.get('currency')!r} " f"does not match Phase native currency {native!r}" ) return errors # --- Monetization Extension Contract --------------------------------------- def check_extension_core_term_redefinition(extension: dict[str, Any]) -> list[str]: """Heuristic check that allocation.rule does not redefine a core term. Stage 0 limitation: this is a pattern-based heuristic (assignment-like phrasing immediately following a core term name), not a full semantic check. It is deterministic and sufficient to reject the deliberately non-conforming fixture required by WP-0002-T04; it is not a substitute for human review of new extensions before promotion to `canonical`. """ rule_text = extension.get("allocation", {}).get("rule", "") normalized = rule_text.lower() errors: list[str] = [] forbidden_patterns = ("=", "redefine", "means", "is defined as", "shall be") for term in CORE_TERMS: term_l = term.lower() if term_l not in normalized: continue for pattern in forbidden_patterns: if f"{term_l} {pattern}" in normalized or f"{term_l}{pattern}" in normalized: errors.append( f"allocation.rule appears to redefine core term {term!r} " f"(matched {term_l + ' ' + pattern!r})" ) return errors def validate_extension_contract(extension: dict[str, Any]) -> None: """Raise ConformanceError on structural or core-term-redefinition failure. A non-conforming extension is rejected here, not silently accepted with reduced trust (TSD §3.3 conformance rule). """ errors = _schema_errors("extension_contract.schema.json", extension) errors += check_extension_core_term_redefinition(extension) if errors: raise ConformanceError(errors) # --- Conversion Attestation -------------------------------------------------- def validate_conversion_attestation(attestation: dict[str, Any]) -> None: """Raise ConformanceError if the attestation is not schema-conformant.""" errors = _schema_errors("conversion_attestation.schema.json", attestation) if errors: raise ConformanceError(errors)