Jyotisha/scripts/personal_report_contract.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
ReportDocument v1 contract validator (stdlib only).

Mirrors `contracts/personal-report/report-document.v1.schema.json` and
`frontend/src/lib/personal-report-contract.ts` (Zod). This module deliberately
uses only the Python standard library: the project does not vendor `jsonschema`,
so the validator below implements the schema semantics explicitly so that the
three sides (JSON Schema / Zod / Python) stay aligned and testable.

Public surface:
    SCHEMA_VERSION, REPORT_CONTRACT_VERSION, MAX_SERIALIZED_BYTES
    CLAIM_STATUSES, REPORT_TYPES, PRESENTATION_MODES, BIRTH_TIME_STATUSES
    TECHNIQUE_STATUSES, CONFLICT_STATUSES, CHART_IDS
    FAILURE_CODES (shared stable enum used by the persistence layer)
    compute_evidence_hash(document) -> str
    validate_report_document(document) -> ValidationResult
    is_valid_report_document(document) -> bool
    load_report_document(path) -> dict
    parse_report_document_json(text) -> ValidationResult
    CLI: python3 scripts/personal_report_contract.py <path-to-json>
"""

from __future__ import annotations

import hashlib
import json
import re
import sys
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple

SCHEMA_VERSION = "report_document.v1"
REPORT_CONTRACT_VERSION = "1"
MAX_SERIALIZED_BYTES = 1_572_864  # 1.5 MiB hard cap on UTF-8 JSON serialization.

CLAIM_STATUSES = (
    "multi_system_consensus",
    "single_system_inference",
    "parameter_sensitive",
    "unclosed_divisional_chart",
    "user_history_verification_required",
    "blocked",
)
REPORT_TYPES = ("personal_full", "personal_thematic")
PRESENTATION_MODES = ("default", "research")
BIRTH_TIME_STATUSES = ("reported", "candidate", "accepted", "confirmed")
TECHNIQUE_STATUSES = ("verified", "partial", "blocked")
CONFLICT_STATUSES = ("unresolved", "partial", "resolved")
CHART_IDS = ("D1", "D9", "D10")
HOUSE_NUMBERS = tuple(range(1, 13))

# Stable failure-code enum shared with the personal_reports table check
# constraint and frontend/src/lib/personal-report-service.ts.
FAILURE_CODES = (
    "profile_incomplete",
    "birth_time_not_usable",
    "report_generation_in_progress",
    "report_rate_limited",
    "calculation_unavailable",
    "model_unavailable",
    "report_schema_invalid",
    "report_guard_rejected",
    "report_not_found",
)

UUID_PATTERN = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")
SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")
SHA1_PATTERN = re.compile(r"^[0-9a-f]{40}$")
ISO8601_PATTERN = re.compile(
    r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d{1,9})?(Z|[+-]\d{2}:\d{2})$"
)
EVIDENCE_ID_PATTERN = re.compile(r"^ev-[a-z0-9_-]{1,63}$")
SECTION_ID_PATTERN = re.compile(r"^[a-z][a-z0-9_-]{0,63}$")
TECHNIQUE_ID_PATTERN = re.compile(r"^[a-z0-9_.-]{1,80}$")

# ────────────────────────────────────────────────────────────────────────────
# Semantic guard patterns. These must stay byte-for-byte equivalent to the
# FORBIDDEN_PATTERNS list in frontend/src/lib/personal-report-contract.ts.
# ────────────────────────────────────────────────────────────────────────────

FORBIDDEN_PATTERNS: Tuple[Tuple[str, str], ...] = (
    ("html_tag_open", r"<\s*(?:script|iframe|object|embed|style|link|meta|form|svg|img|video|audio|source|template|base|applet)\b"),
    ("html_tag_close", r"</\s*(?:script|iframe|object|embed|style|link|meta|form|svg|img|video|audio|source|template|base|applet)\s*>"),
    ("event_handler", r"\bon(?:load|error|click|mouseover|mouseout|submit|focus|blur|change|dblclick|keydown|keyup|pointerdown|pointerup)\s*="),
    ("executable_url", r"\b(?:javascript|vbscript|data:text/html|data:text/javascript|file):"),
    ("processing_instruction", r"<\?"),
    ("template_literal", r"\$\{"),
    ("stack_trace", r"(?:Traceback \(most recent call last\)|node:internal/| at (?:Object|async|node)\.)"),
    ("dunder_path", r"__(?:dirname|filename)(?![A-Za-z0-9_])|__proto__"),
    ("process_env", r"\bprocess\.env\b"),
    ("unix_home_path", r"(?:^|[\\/:])(?:Users|home|opt|var|tmp|root|srv)[\\/]"),
    ("windows_drive_path", r"^[a-zA-Z]:[\\/]"),
    ("jwt_token", r"\beyJ[A-Za-z0-9_-]{20,}\b"),
    ("secret_marker", r"\b(?:SUPABASE_SERVICE_ROLE_KEY|AUTH_SECRET|BEGIN RSA PRIVATE KEY|BEGIN EC PRIVATE KEY|BEGIN OPENSSH PRIVATE KEY)\b"),
    ("tool_trace", r"\b(?:tool_call_id|tool_result|assistant_tool_calls|system_prompt)\b"),
    ("chain_of_thought", r"\bchain[\s_-]?of[\s_-]?thought\b"),
)

# Deterministic-prediction phrases that must never appear inside a section whose
# claimStatus is "blocked". Must match the TS side exactly.
DETERMINISTIC_PHRASES: Tuple[str, ...] = (
    "必然",
    "必定",
    "一定会",
    "肯定会",
    "绝对会",
    "保证会",
    "无疑将",
    "百分之百",
    "确定无疑",
    "guaranteed",
    "definitely will",
    "certainly will",
    "will certainly",
    "is certain to",
)

# Bounded text limits shared with the schema.
TEXT_LIMITS: Dict[str, int] = {
    "displayName": 120,
    "birthPlaceLabel": 200,
    "headline": 200,
    "summary": 2000,
    "priority": 200,
    "chartTitle": 120,
    "sign": 40,
    "occupant": 40,
    "planetName": 40,
    "sectionTitle": 160,
    "narrative": 4000,
    "action": 400,
    "caveat": 400,
    "techniqueName": 160,
    "notes": 500,
    "conflictDescription": 1000,
    "conflictImpact": 500,
    "evidenceLabel": 160,
    "evidenceValue": 500,
    "evidenceSource": 200,
    "blockedTechnique": 120,
    "disclaimer": 2000,
}

ARRAY_LIMITS: Dict[str, int] = {
    "priorities": 8,
    "charts": 3,
    "houses": 12,
    "planets": 12,
    "occupants": 12,
    "thematicNarrative": 12,
    "actions": 12,
    "caveats": 12,
    "evidenceRefs": 24,
    "techniqueAudit": 100,
    "conflicts": 50,
    "calculationEvidence": 100,
    "blockedTechniques": 100,
}

_RE = re.compile


def _compile(patterns: Tuple[Tuple[str, str], ...]) -> List[Tuple[str, "re.Pattern[str]"]]:
    return [(name, _RE(expression, re.IGNORECASE)) for name, expression in patterns]


_FORBIDDEN_COMPILED = _compile(FORBIDDEN_PATTERNS)
_DETERMINISTIC_COMPILED = [
    (_RE(phrase, re.IGNORECASE), phrase) for phrase in DETERMINISTIC_PHRASES
]


@dataclass
class ValidationResult:
    valid: bool
    errors: List[str] = field(default_factory=list)

    def add(self, path: str, message: str) -> None:
        self.errors.append(f"{path}: {message}")


# ────────────────────────────────────────────────────────────────────────────
# Evidence hash (deterministic, cross-language).
# ────────────────────────────────────────────────────────────────────────────

def _canonical_evidence(document: Dict[str, Any]) -> Dict[str, Any]:
    """Canonical evidence object. Defensive: malformed rows are reduced to
    empty placeholders so validation never raises on arbitrary input; the
    canonical hash only matches the TS side for structurally valid documents."""
    appendix = document.get("evidenceAppendix")
    appendix = appendix if isinstance(appendix, dict) else {}

    def rows(key: str) -> List[Any]:
        value = appendix.get(key)
        return value if isinstance(value, list) else []

    def safe(row: Any, key: str) -> Any:
        return row.get(key) if isinstance(row, dict) else None

    canonical = {
        "techniqueAudit": [],
        "conflicts": [],
        "calculationEvidence": [],
    }
    for row in rows("techniqueAudit"):
        entry = {
            "id": safe(row, "id"),
            "techniqueId": safe(row, "techniqueId"),
            "techniqueName": safe(row, "techniqueName"),
            "status": safe(row, "status"),
            "used": safe(row, "used"),
        }
        if isinstance(row, dict) and "notes" in row:
            entry["notes"] = row["notes"]
        canonical["techniqueAudit"].append(entry)
    for row in rows("conflicts"):
        canonical["conflicts"].append({
            "id": safe(row, "id"),
            "description": safe(row, "description"),
            "impact": safe(row, "impact"),
            "status": safe(row, "status"),
        })
    for row in rows("calculationEvidence"):
        canonical["calculationEvidence"].append({
            "id": safe(row, "id"),
            "label": safe(row, "label"),
            "value": safe(row, "value"),
            "source": safe(row, "source"),
        })
    return canonical


def _canonical_json(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=False)


def compute_evidence_hash(document: Dict[str, Any]) -> str:
    """Deterministic SHA-256 over the evidence appendix, matching the TS side."""
    canonical = _canonical_evidence(document)
    return hashlib.sha256(_canonical_json(canonical).encode("utf-8")).hexdigest()


def serialized_bytes(document: Dict[str, Any]) -> int:
    return len(_canonical_json(document).encode("utf-8"))


# ────────────────────────────────────────────────────────────────────────────
# Semantic guards.
# ────────────────────────────────────────────────────────────────────────────

def forbidden_content_hits(value: str) -> List[str]:
    hits: List[str] = []
    for name, pattern in _FORBIDDEN_COMPILED:
        if pattern.search(value):
            hits.append(name)
    return hits


def _text_guard(result: ValidationResult, path: str, value: Any) -> None:
    if value is None:
        return
    if not isinstance(value, str):
        result.add(path, f"expected string, got {type(value).__name__}")
        return
    hits = forbidden_content_hits(value)
    if hits:
        result.add(path, "forbidden content: " + ", ".join(sorted(set(hits))))


def _blocked_determinism(result: ValidationResult, path: str, claim_status: str, texts: List[Tuple[str, str]]) -> None:
    if claim_status != "blocked":
        return
    for label, text in texts:
        if not isinstance(text, str):
            continue
        for pattern, phrase in _DETERMINISTIC_COMPILED:
            if pattern.search(text):
                result.add(path, f"blocked section contains deterministic prediction ({label!r} matches {phrase!r})")


def _evidence_refs(result: ValidationResult, document: Dict[str, Any]) -> None:
    appendix = document["evidenceAppendix"]

    def ids(key: str) -> List[Any]:
        value = appendix.get(key)
        return value if isinstance(value, list) else []

    known_ids: List[str] = []
    seen: Dict[str, str] = {}
    for key in ("techniqueAudit", "conflicts", "calculationEvidence"):
        for index, row in enumerate(ids(key)):
            if not isinstance(row, dict) or not isinstance(row.get("id"), str):
                continue
            evidence_id = row["id"]
            if evidence_id in seen:
                result.add(
                    f"evidenceAppendix.{key}[{index}].id",
                    f"duplicate evidence id {evidence_id!r} (also used in {seen[evidence_id]})",
                )
            else:
                seen[evidence_id] = f"{key}[{index}]"
            known_ids.append(evidence_id)

    for index, section in enumerate(document.get("thematicNarrative", [])):
        if not isinstance(section, dict):
            continue
        path = f"thematicNarrative[{index}].evidenceRefs"
        refs = section.get("evidenceRefs")
        if not isinstance(refs, list):
            continue
        for ref in refs:
            if ref not in known_ids:
                result.add(path, f"unknown evidence id {ref!r}")


# ────────────────────────────────────────────────────────────────────────────
# Structural validation (explicit schema implementation).
# ────────────────────────────────────────────────────────────────────────────

def _expect_object(result: ValidationResult, path: str, value: Any) -> Optional[Dict[str, Any]]:
    if not isinstance(value, dict):
        result.add(path, f"expected object, got {type(value).__name__}")
        return None
    return value


def _check_enum(result: ValidationResult, path: str, value: Any, allowed: Tuple[str, ...]) -> None:
    if value not in allowed:
        result.add(path, f"invalid value {value!r}; allowed: {', '.join(allowed)}")


def _check_text(result: ValidationResult, path: str, value: Any, max_length: int, min_length: int = 1) -> None:
    if not isinstance(value, str):
        result.add(path, f"expected string, got {type(value).__name__}")
        return
    if len(value) < min_length:
        result.add(path, f"shorter than minimum length {min_length}")
    if len(value) > max_length:
        result.add(path, f"longer than maximum length {max_length}")
    _text_guard(result, path, value)


def _check_uuid(result: ValidationResult, path: str, value: Any) -> None:
    if not isinstance(value, str) or not UUID_PATTERN.match(value):
        result.add(path, f"invalid uuid {value!r}")


def _check_hash(result: ValidationResult, path: str, value: Any, pattern: "re.Pattern[str]", length: int) -> None:
    if not isinstance(value, str) or not pattern.match(value) or len(value) != length:
        result.add(path, f"invalid hex hash {value!r}")


def _check_array(
    result: ValidationResult,
    path: str,
    value: Any,
    max_items: int,
    min_items: int = 0,
) -> Optional[List[Any]]:
    if not isinstance(value, list):
        result.add(path, f"expected array, got {type(value).__name__}")
        return None
    if len(value) > max_items:
        result.add(path, f"longer than maximum items {max_items}")
    if len(value) < min_items:
        result.add(path, f"shorter than minimum items {min_items}")
    return value


def _check_keys(
    result: ValidationResult,
    path: str,
    value: Dict[str, Any],
    required: Tuple[str, ...],
    allowed: Optional[Tuple[str, ...]] = None,
) -> None:
    permitted = required if allowed is None else allowed
    missing = [key for key in required if key not in value]
    if missing:
        result.add(path, "missing required keys: " + ", ".join(missing))
    extra = [key for key in value if key not in permitted]
    if extra:
        result.add(path, "unexpected keys: " + ", ".join(extra))


def _validate_subject(result: ValidationResult, path: str, subject: Any) -> None:
    value = _expect_object(result, path, subject)
    if value is None:
        return
    _check_keys(result, path, value, ("displayName", "birthTimeStatus", "birthPlaceLabel"))
    _check_text(result, f"{path}.displayName", value.get("displayName"), TEXT_LIMITS["displayName"])
    _check_enum(result, f"{path}.birthTimeStatus", value.get("birthTimeStatus"), BIRTH_TIME_STATUSES)
    _check_text(result, f"{path}.birthPlaceLabel", value.get("birthPlaceLabel"), TEXT_LIMITS["birthPlaceLabel"])


def _validate_provenance(result: ValidationResult, path: str, provenance: Any) -> None:
    value = _expect_object(result, path, provenance)
    if value is None:
        return
    _check_keys(
        result,
        path,
        value,
        ("skillSourceCommit", "skillSnapshotSha256", "calculationHash", "evidenceHash", "reportContractVersion"),
    )
    commit = value.get("skillSourceCommit")
    if commit is not None:
        _check_hash(result, f"{path}.skillSourceCommit", commit, SHA1_PATTERN, 40)
    _check_hash(result, f"{path}.skillSnapshotSha256", value.get("skillSnapshotSha256"), SHA256_PATTERN, 64)
    _check_hash(result, f"{path}.calculationHash", value.get("calculationHash"), SHA256_PATTERN, 64)
    _check_hash(result, f"{path}.evidenceHash", value.get("evidenceHash"), SHA256_PATTERN, 64)
    if value.get("reportContractVersion") != REPORT_CONTRACT_VERSION:
        result.add(f"{path}.reportContractVersion", f"must be {REPORT_CONTRACT_VERSION!r}")


def _validate_executive_summary(result: ValidationResult, path: str, summary: Any) -> None:
    value = _expect_object(result, path, summary)
    if value is None:
        return
    _check_keys(result, path, value, ("headline", "summary", "priorities", "overallClaimStatus"))
    _check_text(result, f"{path}.headline", value.get("headline"), TEXT_LIMITS["headline"])
    _check_text(result, f"{path}.summary", value.get("summary"), TEXT_LIMITS["summary"])
    priorities = _check_array(result, f"{path}.priorities", value.get("priorities"), ARRAY_LIMITS["priorities"])
    if priorities is not None:
        for index, priority in enumerate(priorities):
            _check_text(result, f"{path}.priorities[{index}]", priority, TEXT_LIMITS["priority"])
    _check_enum(result, f"{path}.overallClaimStatus", value.get("overallClaimStatus"), CLAIM_STATUSES)


def _validate_chart(result: ValidationResult, path: str, chart: Any) -> None:
    value = _expect_object(result, path, chart)
    if value is None:
        return
    _check_keys(result, path, value, ("id", "title", "houses", "claimStatus"), ("id", "title", "houses", "claimStatus", "planets"))
    _check_enum(result, f"{path}.id", value.get("id"), CHART_IDS)
    _check_text(result, f"{path}.title", value.get("title"), TEXT_LIMITS["chartTitle"])
    _check_enum(result, f"{path}.claimStatus", value.get("claimStatus"), CLAIM_STATUSES)

    houses = _check_array(result, f"{path}.houses", value.get("houses"), ARRAY_LIMITS["houses"])
    seen_houses: List[int] = []
    if houses is not None:
        for index, house in enumerate(houses):
            house_path = f"{path}.houses[{index}]"
            house_value = _expect_object(result, house_path, house)
            if house_value is None:
                continue
            _check_keys(result, house_path, house_value, ("houseNumber", "sign", "occupants"))
            house_number = house_value.get("houseNumber")
            if house_number in seen_houses:
                result.add(house_path, f"duplicate houseNumber {house_number!r}")
            if isinstance(house_number, int) and not isinstance(house_number, bool) and house_number in HOUSE_NUMBERS:
                seen_houses.append(house_number)
            elif not isinstance(house_number, bool):
                result.add(f"{house_path}.houseNumber", f"must be integer 1..12, got {house_number!r}")
            _check_text(result, f"{house_path}.sign", house_value.get("sign"), TEXT_LIMITS["sign"])
            occupants = _check_array(result, f"{house_path}.occupants", house_value.get("occupants"), ARRAY_LIMITS["occupants"])
            if occupants is not None:
                for occupant_index, occupant in enumerate(occupants):
                    _check_text(result, f"{house_path}.occupants[{occupant_index}]", occupant, TEXT_LIMITS["occupant"])

    planets = None
    if "planets" in value:
        planets = _check_array(result, f"{path}.planets", value.get("planets"), ARRAY_LIMITS["planets"])
    if planets is not None:
        for index, planet in enumerate(planets):
            planet_path = f"{path}.planets[{index}]"
            planet_value = _expect_object(result, planet_path, planet)
            if planet_value is None:
                continue
            _check_keys(result, planet_path, planet_value, ("name", "sign", "longitudeDegrees", "houseNumber", "retrograde"))
            _check_text(result, f"{planet_path}.name", planet_value.get("name"), TEXT_LIMITS["planetName"])
            _check_text(result, f"{planet_path}.sign", planet_value.get("sign"), TEXT_LIMITS["sign"])
            longitude = planet_value.get("longitudeDegrees")
            if not isinstance(longitude, (int, float)) or isinstance(longitude, bool) or not (0 <= longitude < 360):
                result.add(f"{planet_path}.longitudeDegrees", f"must be number 0..360 (360 excluded), got {longitude!r}")
            house_number = planet_value.get("houseNumber")
            if not isinstance(house_number, int) or isinstance(house_number, bool) or house_number not in HOUSE_NUMBERS:
                result.add(f"{planet_path}.houseNumber", f"must be integer 1..12, got {house_number!r}")
            if not isinstance(planet_value.get("retrograde"), bool):
                result.add(f"{planet_path}.retrograde", "must be boolean")


def _validate_thematic_section(result: ValidationResult, path: str, section: Any) -> None:
    value = _expect_object(result, path, section)
    if value is None:
        return
    _check_keys(result, path, value, ("id", "title", "narrative", "actions", "caveats", "claimStatus", "evidenceRefs"))
    section_id = value.get("id")
    if not isinstance(section_id, str) or not SECTION_ID_PATTERN.match(section_id):
        result.add(f"{path}.id", f"invalid section id {section_id!r}")
    _check_text(result, f"{path}.title", value.get("title"), TEXT_LIMITS["sectionTitle"])
    _check_text(result, f"{path}.narrative", value.get("narrative"), TEXT_LIMITS["narrative"])
    actions = _check_array(result, f"{path}.actions", value.get("actions"), ARRAY_LIMITS["actions"])
    if actions is not None:
        for index, action in enumerate(actions):
            _check_text(result, f"{path}.actions[{index}]", action, TEXT_LIMITS["action"])
    caveats = _check_array(result, f"{path}.caveats", value.get("caveats"), ARRAY_LIMITS["caveats"])
    if caveats is not None:
        for index, caveat in enumerate(caveats):
            _check_text(result, f"{path}.caveats[{index}]", caveat, TEXT_LIMITS["caveat"])
    _check_enum(result, f"{path}.claimStatus", value.get("claimStatus"), CLAIM_STATUSES)
    refs = _check_array(result, f"{path}.evidenceRefs", value.get("evidenceRefs"), ARRAY_LIMITS["evidenceRefs"])
    if refs is not None:
        for index, ref in enumerate(refs):
            if not isinstance(ref, str) or not EVIDENCE_ID_PATTERN.match(ref):
                result.add(f"{path}.evidenceRefs[{index}]", f"invalid evidence id {ref!r}")


def _validate_technique_row(result: ValidationResult, path: str, row: Any) -> None:
    value = _expect_object(result, path, row)
    if value is None:
        return
    _check_keys(result, path, value, ("id", "techniqueId", "techniqueName", "status", "used"), ("id", "techniqueId", "techniqueName", "status", "used", "notes"))
    evidence_id = value.get("id")
    if not isinstance(evidence_id, str) or not EVIDENCE_ID_PATTERN.match(evidence_id):
        result.add(f"{path}.id", f"invalid evidence id {evidence_id!r}")
    technique_id = value.get("techniqueId")
    if not isinstance(technique_id, str) or not TECHNIQUE_ID_PATTERN.match(technique_id):
        result.add(f"{path}.techniqueId", f"invalid technique id {technique_id!r}")
    _check_text(result, f"{path}.techniqueName", value.get("techniqueName"), TEXT_LIMITS["techniqueName"])
    _check_enum(result, f"{path}.status", value.get("status"), TECHNIQUE_STATUSES)
    if not isinstance(value.get("used"), bool):
        result.add(f"{path}.used", "must be boolean")
    if "notes" in value:
        _check_text(result, f"{path}.notes", value.get("notes"), TEXT_LIMITS["notes"], min_length=0)


def _validate_conflict_row(result: ValidationResult, path: str, row: Any) -> None:
    value = _expect_object(result, path, row)
    if value is None:
        return
    _check_keys(result, path, value, ("id", "description", "impact", "status"))
    evidence_id = value.get("id")
    if not isinstance(evidence_id, str) or not EVIDENCE_ID_PATTERN.match(evidence_id):
        result.add(f"{path}.id", f"invalid evidence id {evidence_id!r}")
    _check_text(result, f"{path}.description", value.get("description"), TEXT_LIMITS["conflictDescription"])
    _check_text(result, f"{path}.impact", value.get("impact"), TEXT_LIMITS["conflictImpact"])
    _check_enum(result, f"{path}.status", value.get("status"), CONFLICT_STATUSES)


def _validate_calculation_row(result: ValidationResult, path: str, row: Any) -> None:
    value = _expect_object(result, path, row)
    if value is None:
        return
    _check_keys(result, path, value, ("id", "label", "value", "source"))
    evidence_id = value.get("id")
    if not isinstance(evidence_id, str) or not EVIDENCE_ID_PATTERN.match(evidence_id):
        result.add(f"{path}.id", f"invalid evidence id {evidence_id!r}")
    _check_text(result, f"{path}.label", value.get("label"), TEXT_LIMITS["evidenceLabel"])
    _check_text(result, f"{path}.value", value.get("value"), TEXT_LIMITS["evidenceValue"])
    _check_text(result, f"{path}.source", value.get("source"), TEXT_LIMITS["evidenceSource"])


def _validate_evidence_appendix(result: ValidationResult, path: str, appendix: Any) -> None:
    value = _expect_object(result, path, appendix)
    if value is None:
        return
    _check_keys(
        result,
        path,
        value,
        ("expandedByDefault", "techniqueAudit", "conflicts", "calculationEvidence", "blockedTechniques"),
    )
    if not isinstance(value.get("expandedByDefault"), bool):
        result.add(f"{path}.expandedByDefault", "must be boolean")
    rows = _check_array(result, f"{path}.techniqueAudit", value.get("techniqueAudit"), ARRAY_LIMITS["techniqueAudit"])
    if rows is not None:
        for index, row in enumerate(rows):
            _validate_technique_row(result, f"{path}.techniqueAudit[{index}]", row)
    rows = _check_array(result, f"{path}.conflicts", value.get("conflicts"), ARRAY_LIMITS["conflicts"])
    if rows is not None:
        for index, row in enumerate(rows):
            _validate_conflict_row(result, f"{path}.conflicts[{index}]", row)
    rows = _check_array(result, f"{path}.calculationEvidence", value.get("calculationEvidence"), ARRAY_LIMITS["calculationEvidence"])
    if rows is not None:
        for index, row in enumerate(rows):
            _validate_calculation_row(result, f"{path}.calculationEvidence[{index}]", row)
    blocked = _check_array(result, f"{path}.blockedTechniques", value.get("blockedTechniques"), ARRAY_LIMITS["blockedTechniques"])
    if blocked is not None:
        for index, technique in enumerate(blocked):
            _check_text(result, f"{path}.blockedTechniques[{index}]", technique, TEXT_LIMITS["blockedTechnique"])


def _validate_chart_set(result: ValidationResult, charts: List[Any]) -> None:
    """Runtime chart-set semantics: exactly one D1, unique ids, D1 complete houses."""
    seen_ids: List[str] = []
    d1_chart: Optional[Dict[str, Any]] = None
    for index, chart in enumerate(charts):
        if not isinstance(chart, dict):
            continue
        chart_id = chart.get("id")
        if isinstance(chart_id, str):
            if chart_id in seen_ids:
                result.add(f"charts[{index}].id", f"duplicate chart id {chart_id!r}")
            seen_ids.append(chart_id)
            if chart_id == "D1":
                d1_chart = chart
    d1_count = sum(1 for chart_id in seen_ids if chart_id == "D1")
    if d1_count != 1:
        result.add("charts", f"must contain exactly one D1 chart, found {d1_count}")
    if d1_chart is not None:
        houses = d1_chart.get("houses")
        if not isinstance(houses, list):
            result.add("charts[D1].houses", "D1 chart must declare houses")
            return
        numbers = []
        for house in houses:
            if isinstance(house, dict) and isinstance(house.get("houseNumber"), int) \
                    and not isinstance(house.get("houseNumber"), bool):
                numbers.append(house["houseNumber"])
        if sorted(numbers) != list(range(1, 13)):
            result.add("charts[D1].houses", "D1 chart must contain all twelve house numbers 1..12 exactly once")
        elif len(set(numbers)) != 12:
            result.add("charts[D1].houses", "D1 chart contains duplicate house numbers")


def validate_report_document(document: Any) -> ValidationResult:
    result = ValidationResult(valid=True)
    if not isinstance(document, dict):
        result.add("(root)", f"expected object, got {type(document).__name__}")
        result.valid = False
        return result

    _check_keys(
        result,
        "(root)",
        document,
        (
            "schemaVersion",
            "reportId",
            "reportType",
            "presentationMode",
            "generatedAt",
            "subject",
            "provenance",
            "executiveSummary",
            "charts",
            "thematicNarrative",
            "evidenceAppendix",
            "disclaimer",
        ),
    )
    if document.get("schemaVersion") != SCHEMA_VERSION:
        result.add("schemaVersion", f"must be {SCHEMA_VERSION!r}")
    _check_uuid(result, "reportId", document.get("reportId"))
    _check_enum(result, "reportType", document.get("reportType"), REPORT_TYPES)
    _check_enum(result, "presentationMode", document.get("presentationMode"), PRESENTATION_MODES)
    generated_at = document.get("generatedAt")
    if not isinstance(generated_at, str) or not ISO8601_PATTERN.match(generated_at):
        result.add("generatedAt", f"invalid ISO-8601 timestamp {generated_at!r}")

    _validate_subject(result, "subject", document.get("subject"))
    _validate_provenance(result, "provenance", document.get("provenance"))
    _validate_executive_summary(result, "executiveSummary", document.get("executiveSummary"))

    charts = _check_array(result, "charts", document.get("charts"), ARRAY_LIMITS["charts"], min_items=1)
    if charts is not None:
        for index, chart in enumerate(charts):
            _validate_chart(result, f"charts[{index}]", chart)
        _validate_chart_set(result, charts)

    sections = _check_array(result, "thematicNarrative", document.get("thematicNarrative"), ARRAY_LIMITS["thematicNarrative"])
    seen_section_ids: List[str] = []
    if sections is not None:
        for index, section in enumerate(sections):
            _validate_thematic_section(result, f"thematicNarrative[{index}]", section)
            section_id = section.get("id") if isinstance(section, dict) else None
            if isinstance(section_id, str):
                if section_id in seen_section_ids:
                    result.add(f"thematicNarrative[{index}].id", f"duplicate section id {section_id!r}")
                seen_section_ids.append(section_id)

    _validate_evidence_appendix(result, "evidenceAppendix", document.get("evidenceAppendix"))
    _check_text(result, "disclaimer", document.get("disclaimer"), TEXT_LIMITS["disclaimer"])

    # Semantic guards (only when the structural shape is usable; all access is
    # defensive so arbitrary/malformed JSON can never raise).
    if (
        isinstance(document.get("evidenceAppendix"), dict)
        and all(
            isinstance(document["evidenceAppendix"].get(key), list)
            for key in ("techniqueAudit", "conflicts", "calculationEvidence")
        )
        and isinstance(document.get("thematicNarrative"), list)
        and isinstance(document.get("provenance"), dict)
        and isinstance(document.get("charts"), list)
        and isinstance(document.get("executiveSummary"), dict)
    ):
        _evidence_refs(result, document)
        expected_hash = compute_evidence_hash(document)
        if document["provenance"].get("evidenceHash") != expected_hash:
            result.add("provenance.evidenceHash", f"does not match computed evidence hash {expected_hash}")

        blocked_texts: List[Tuple[str, str]] = []
        summary = document["executiveSummary"]
        if summary.get("overallClaimStatus") == "blocked":
            blocked_texts.append(("headline", summary.get("headline")))
            blocked_texts.append(("summary", summary.get("summary")))
            for index, priority in enumerate(summary.get("priorities", [])):
                blocked_texts.append((f"priorities[{index}]", priority))
        for index, chart in enumerate(document["charts"]):
            if isinstance(chart, dict) and chart.get("claimStatus") == "blocked":
                blocked_texts.append((f"charts[{index}].title", chart.get("title")))
        for index, section in enumerate(document["thematicNarrative"]):
            if isinstance(section, dict) and section.get("claimStatus") == "blocked":
                blocked_texts.append((f"thematicNarrative[{index}].title", section.get("title")))
                blocked_texts.append((f"thematicNarrative[{index}].narrative", section.get("narrative")))
                for action_index, action in enumerate(section.get("actions", [])):
                    blocked_texts.append((f"thematicNarrative[{index}].actions[{action_index}]", action))
                for caveat_index, caveat in enumerate(section.get("caveats", [])):
                    blocked_texts.append((f"thematicNarrative[{index}].caveats[{caveat_index}]", caveat))
        for path, text in blocked_texts:
            _blocked_determinism(result, path, "blocked", [(path, text)])

        size = serialized_bytes(document)
        if size > MAX_SERIALIZED_BYTES:
            result.add("(size)", f"serialized document is {size} bytes, exceeding {MAX_SERIALIZED_BYTES}")

    result.valid = not result.errors
    return result


def is_valid_report_document(document: Any) -> bool:
    return validate_report_document(document).valid


def load_report_document(path: str) -> Dict[str, Any]:
    with open(path, "r", encoding="utf-8") as handle:
        return json.load(handle)


def parse_report_document_json(text: str) -> ValidationResult:
    try:
        document = json.loads(text)
    except json.JSONDecodeError as error:
        result = ValidationResult(valid=False)
        result.add("(json)", f"invalid JSON: {error}")
        return result
    return validate_report_document(document)


def main(argv: Optional[List[str]] = None) -> int:
    args = list(sys.argv[1:] if argv is None else argv)
    if not args:
        print("usage: python3 scripts/personal_report_contract.py <report-document.json>", file=sys.stderr)
        return 2
    path = args[0]
    try:
        with open(path, "r", encoding="utf-8") as handle:
            text = handle.read()
    except OSError as error:
        print(f"unable to read {path}: {error}", file=sys.stderr)
        return 2
    result = parse_report_document_json(text)
    if result.valid:
        try:
            size = serialized_bytes(json.loads(text))
        except ValueError:
            size = 0
        print(f"valid: {path} ({size} bytes)")
        return 0
    print(f"invalid: {path}", file=sys.stderr)
    for error in result.errors:
        print(f"  - {error}", file=sys.stderr)
    return 1


if __name__ == "__main__":
    raise SystemExit(main())