Implement TREV-WP-0002 Stage 0 foundation: schemas, pure fold, golden Phase
Delivers the offline runnable specification foundation for the Trust Layer (TSD §3-§6), not a hosted Trust Service: - JSON Schemas for Phase Manifest, Ledger Entry, Extension Contract, and Conversion Attestation, encoding the Stage 0 working defaults (Q3 future license enum, Q6 single-currency Phases, Q8 required longstop_at). - src/target_revenue: pure Outstanding Target fold, SHA-256 hash-chain verification, Ed25519 signing helpers, extension conformance checks (including a core-term-redefinition heuristic), and conversion detection that never requires an attestation document to determine conversion status. - examples/phase-001: golden Phase package matching the concept doc's worked example, generated via scripts/generate_golden_phase.py so the hash chain is computed by the library itself, not hand-typed. - 32 passing pytest tests covering manifest/ledger/extension conformance, tamper/reorder detection, and the full lifecycle fold to conversion. - docs/adr/ADR-0001: proposed (not accepted) Stage 0 stack choice, per the WP-0002-T01 human-accept gate — implementation proceeded against the proposal as the workplan note permits, but the task stays open. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
parent
658b1af00d
commit
57c7111cbc
28 changed files with 1751 additions and 7 deletions
176
src/target_revenue/validation.py
Normal file
176
src/target_revenue/validation.py
Normal file
|
|
@ -0,0 +1,176 @@
|
|||
"""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 '<root>'}: {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)
|
||||
Loading…
Add table
Add a link
Reference in a new issue