Blame for src/python/tom_world/records.py

"""Typed content-addressed records used by the TOM world kernel."""
from __future__ import annotations

import re
from collections.abc import Mapping, Sequence
from typing import Any

from .canonical import attach_hash, verify_hash

RECORD_SCHEMA = "TOM-WORLD-RECORD-0.1"
RECORD_TYPES = frozenset({
    "definition",
    "instance",
    "relation",
    "support",
    "compatibility",
    "transition",
    "event_spec",
    "grammar",
    "observation",
    "hypothesis",
    "goal",
    "policy",
    "event",
    "lineage",
    "checkpoint",
})
DEFINITION_PHASES = (
    "parse",
    "normalize",
    "resolve",
    "construct",
    "transform",
    "support",
    "compatibility",
    "guard",
    "transition",
    "lineage",
)
_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/@+\-]*$")


def _require_mapping(value: Any, name: str) -> Mapping[str, Any]:
    if not isinstance(value, Mapping):
        raise ValueError(f"{name} must be an object")
    return value


def _require_string(value: Any, name: str) -> str:
    if not isinstance(value, str) or not value:
        raise ValueError(f"{name} must be a nonempty string")
    return value


def _require_string_list(value: Any, name: str) -> list[str]:
    if not isinstance(value, list) or any(not isinstance(x, str) or not x for x in value):
        raise ValueError(f"{name} must be an array of nonempty strings")
    if len(value) != len(set(value)):
        raise ValueError(f"{name} must not contain duplicates")
    return value




def _require_sha256(value: Any, name: str) -> str:
    text = _require_string(value, name)
    if not re.fullmatch(r"sha256:[0-9a-f]{64}", text):
        raise ValueError(f"{name} must be a lowercase sha256 identifier")
    return text


def _validate_interval(value: Any, name: str) -> None:
    interval = _require_mapping(value, name)
    start = interval.get("start")
    end = interval.get("end")
    if isinstance(start, bool) or not isinstance(start, int) or start < 0:
        raise ValueError(f"{name}.start must be a nonnegative integer")
    if isinstance(end, bool) or not isinstance(end, int) or end < start:
        raise ValueError(f"{name}.end must be an integer >= start")


def _validate_topology_sheet(payload: Mapping[str, Any], name: str) -> None:
    if "topology_sheet" in payload:
        sheet = payload["topology_sheet"]
        if isinstance(sheet, bool) or not isinstance(sheet, int) or not 0 <= sheet <= 0xFFFFFFFF:
            raise ValueError(f"{name}.topology_sheet must be a u32 integer")


def make_record(
    record_type: str,
    ident: str,
    payload: Mapping[str, Any],
    *,
    dependencies: Sequence[str] = (),
    version: int = 1,
    provenance: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    """Create and hash one source record."""

    record = {
        "schema": RECORD_SCHEMA,
        "record_type": record_type,
        "id": ident,
        "version": version,
        "dependencies": list(dependencies),
        "payload": dict(payload),
        "provenance": dict(provenance or {}),
    }
    result = attach_hash(record)
    validate_record(result)
    return result


def validate_record(record: Mapping[str, Any], *, require_hash: bool = True) -> None:
    if record.get("schema") != RECORD_SCHEMA:
        raise ValueError(f"record schema must be {RECORD_SCHEMA}")
    record_type = _require_string(record.get("record_type"), "record_type")
    if record_type not in RECORD_TYPES:
        raise ValueError(f"unsupported record_type {record_type}")
    ident = _require_string(record.get("id"), "id")
    if not _ID_RE.fullmatch(ident):
        raise ValueError(f"record id contains unsupported characters: {ident}")
    version = record.get("version")
    if isinstance(version, bool) or not isinstance(version, int) or version < 1:
        raise ValueError("version must be a positive integer")
    _require_string_list(record.get("dependencies"), "dependencies")
    payload = _require_mapping(record.get("payload"), "payload")
    _require_mapping(record.get("provenance"), "provenance")
    if require_hash and not verify_hash(record):
        raise ValueError(f"record content hash mismatch: {ident}")

    validator = _TYPE_VALIDATORS[record_type]
    validator(ident, payload)


def _validate_definition(ident: str, payload: Mapping[str, Any]) -> None:
    for field in ("kind", "domain", "codomain", "operation", "phase"):
        _require_string(payload.get(field), f"{ident}.payload.{field}")
    if payload["phase"] not in DEFINITION_PHASES:
        raise ValueError(f"{ident}.payload.phase is not a TOM phase")
    order = payload.get("order", 0)
    if isinstance(order, bool) or not isinstance(order, int) or order < 0:
        raise ValueError(f"{ident}.payload.order must be a nonnegative integer")
    if "parameters" in payload:
        _require_mapping(payload["parameters"], f"{ident}.payload.parameters")
    if "capabilities" in payload:
        _require_string_list(payload["capabilities"], f"{ident}.payload.capabilities")
    if "invariants" in payload and not isinstance(payload["invariants"], list):
        raise ValueError(f"{ident}.payload.invariants must be an array")


def _validate_instance(ident: str, payload: Mapping[str, Any]) -> None:
    _require_string(payload.get("program_blob_id"), f"{ident}.payload.program_blob_id")
    if "initial_state" in payload:
        state = _require_mapping(payload["initial_state"], f"{ident}.payload.initial_state")
        valid = {
            "rho", "theta", "tick", "phi", "vrho", "vtheta", "vtick", "vphi",
            "orientation", "sheet", "branch", "cell", "lineage", "output", "residual", "status",
        }
        unknown = sorted(set(state) - valid)
        if unknown:
            raise ValueError(f"{ident}.payload.initial_state has unknown fields: {', '.join(unknown)}")
    if "context" in payload:
        _require_mapping(payload["context"], f"{ident}.payload.context")
    if "time_interval" in payload:
        _validate_interval(payload["time_interval"], f"{ident}.payload.time_interval")
    _validate_topology_sheet(payload, f"{ident}.payload")


def _validate_expression_record(ident: str, payload: Mapping[str, Any]) -> None:
    if "expression" not in payload:
        raise ValueError(f"{ident}.payload.expression is required")


def _validate_relation(ident: str, payload: Mapping[str, Any]) -> None:
    _validate_expression_record(ident, payload)
    _require_string(payload.get("instance_id"), f"{ident}.payload.instance_id")
    zero_test = payload.get("zero_test", "equal_zero")
    if zero_test not in {"equal_zero", "contains_zero", "less_equal_zero"}:
        raise ValueError(f"{ident}.payload.zero_test is unsupported")
    trigger = payload.get("trigger", "enter_zero")
    if trigger not in {"zero", "enter_zero", "crossing", "enter_nonpositive"}:
        raise ValueError(f"{ident}.payload.trigger is unsupported")
    for field in ("support_ids", "compatibility_ids"):
        if field in payload:
            _require_string_list(payload[field], f"{ident}.payload.{field}")
    if "event_spec_id" in payload:
        _require_string(payload["event_spec_id"], f"{ident}.payload.event_spec_id")
    priority = payload.get("priority", 0)
    if isinstance(priority, bool) or not isinstance(priority, int):
        raise ValueError(f"{ident}.payload.priority must be an integer")
    if "active_interval" in payload:
        _validate_interval(payload["active_interval"], f"{ident}.payload.active_interval")
    if "time_interval" in payload:
        _validate_interval(payload["time_interval"], f"{ident}.payload.time_interval")
    _validate_topology_sheet(payload, f"{ident}.payload")


def _validate_support(ident: str, payload: Mapping[str, Any]) -> None:
    _validate_expression_record(ident, payload)


def _validate_compatibility(ident: str, payload: Mapping[str, Any]) -> None:
    _validate_expression_record(ident, payload)


def _validate_transition(ident: str, payload: Mapping[str, Any]) -> None:
    for field in ("set", "add", "xor"):
        if field in payload:
            _require_mapping(payload[field], f"{ident}.payload.{field}")
    if not any(field in payload for field in ("set", "add", "xor")):
        raise ValueError(f"{ident}.payload requires set, add, or xor")
    if "normalize_periodic" in payload and not isinstance(payload["normalize_periodic"], bool):
        raise ValueError(f"{ident}.payload.normalize_periodic must be boolean")


def _validate_event_spec(ident: str, payload: Mapping[str, Any]) -> None:
    _require_string(payload.get("relation_id"), f"{ident}.payload.relation_id")
    if "transition_id" in payload:
        _require_string(payload["transition_id"], f"{ident}.payload.transition_id")
    if "route" in payload:
        _require_string(payload["route"], f"{ident}.payload.route")
    if "confidence" in payload:
        confidence = _require_mapping(payload["confidence"], f"{ident}.payload.confidence")
        if "numerator" in confidence or "denominator" in confidence:
            numerator = confidence.get("numerator")
            denominator = confidence.get("denominator")
            if not isinstance(numerator, int) or not isinstance(denominator, int) or denominator <= 0:
                raise ValueError(f"{ident}.payload.confidence rational is invalid")


def _validate_grammar(ident: str, payload: Mapping[str, Any]) -> None:
    axiom = payload.get("axiom")
    if not isinstance(axiom, list) or any(not isinstance(x, str) for x in axiom):
        raise ValueError(f"{ident}.payload.axiom must be a string array")
    productions = _require_mapping(payload.get("productions"), f"{ident}.payload.productions")
    for symbol, production in productions.items():
        _require_string(symbol, f"{ident}.payload.productions key")
        if isinstance(production, list):
            if any(not isinstance(x, str) for x in production):
                raise ValueError(f"{ident} production {symbol} must contain strings")
        elif isinstance(production, Mapping):
            for branch in ("zero", "one"):
                value = production.get(branch)
                if not isinstance(value, list) or any(not isinstance(x, str) for x in value):
                    raise ValueError(f"{ident} branched production {symbol}.{branch} must be a string array")
        else:
            raise ValueError(f"{ident} production {symbol} must be an array or zero/one object")
    budgets = _require_mapping(payload.get("budgets"), f"{ident}.payload.budgets")
    for field in ("max_depth", "max_symbols", "max_stack"):
        value = budgets.get(field)
        if isinstance(value, bool) or not isinstance(value, int) or value < 0:
            raise ValueError(f"{ident}.payload.budgets.{field} must be a nonnegative integer")
    if "branch_bits" in payload:
        bits = payload["branch_bits"]
        if not isinstance(bits, list) or any(bit not in (0, 1) for bit in bits):
            raise ValueError(f"{ident}.payload.branch_bits must contain only 0/1")
        if not bits and any(isinstance(p, Mapping) for p in productions.values()):
            raise ValueError(f"{ident} requires branch_bits for branched productions")
    policy = payload.get("branch_policy", "cycle")
    if policy not in {"cycle", "strict"}:
        raise ValueError(f"{ident}.payload.branch_policy is unsupported")


def _validate_checkpoint(ident: str, payload: Mapping[str, Any]) -> None:
    _require_string(payload.get("instance_id"), f"{ident}.payload.instance_id")
    tick = payload.get("tick")
    if isinstance(tick, bool) or not isinstance(tick, int) or tick < 0:
        raise ValueError(f"{ident}.payload.tick must be a nonnegative integer")
    state = _require_mapping(payload.get("state"), f"{ident}.payload.state")
    valid = {
        "rho", "theta", "tick", "phi", "vrho", "vtheta", "vtick", "vphi",
        "orientation", "sheet", "branch", "cell", "lineage", "output", "residual", "status",
    }
    if set(state) != valid:
        missing = sorted(valid - set(state))
        extra = sorted(set(state) - valid)
        raise ValueError(f"{ident}.payload.state field mismatch: missing={missing}, extra={extra}")
    for field, value in state.items():
        if isinstance(value, bool) or not isinstance(value, int):
            raise ValueError(f"{ident}.payload.state.{field} must be an integer")
    _require_sha256(payload.get("instance_hash"), f"{ident}.payload.instance_hash")
    _require_sha256(payload.get("program_blob_hash"), f"{ident}.payload.program_blob_hash")
    _require_sha256(payload.get("source_commit"), f"{ident}.payload.source_commit")
    _require_sha256(payload.get("state_certificate_hash"), f"{ident}.payload.state_certificate_hash")
    _validate_topology_sheet(payload, f"{ident}.payload")


def _validate_generic(ident: str, payload: Mapping[str, Any]) -> None:
    del ident, payload


_TYPE_VALIDATORS = {
    "definition": _validate_definition,
    "instance": _validate_instance,
    "relation": _validate_relation,
    "support": _validate_support,
    "compatibility": _validate_compatibility,
    "transition": _validate_transition,
    "event_spec": _validate_event_spec,
    "grammar": _validate_grammar,
    "observation": _validate_generic,
    "hypothesis": _validate_generic,
    "goal": _validate_generic,
    "policy": _validate_generic,
    "event": _validate_generic,
    "lineage": _validate_generic,
    "checkpoint": _validate_checkpoint,
}


def topological_record_order(records: Sequence[Mapping[str, Any]], existing_ids: set[str] | None = None) -> list[str]:
    """Return a deterministic dependency order for staged records."""

    existing = existing_ids or set()
    by_id: dict[str, Mapping[str, Any]] = {}
    input_rank: dict[str, int] = {}
    for rank, record in enumerate(records):
        validate_record(record)
        ident = str(record["id"])
        if ident in by_id:
            raise ValueError(f"duplicate staged record id: {ident}")
        by_id[ident] = record
        input_rank[ident] = rank

    indegree = {ident: 0 for ident in by_id}
    children: dict[str, list[str]] = {ident: [] for ident in by_id}
    for ident, record in by_id.items():
        for dep in record["dependencies"]:
            if dep in existing:
                continue
            if dep not in by_id:
                raise ValueError(f"unresolved dependency {dep} from {ident}")
            indegree[ident] += 1
            children[dep].append(ident)

    ready = sorted((input_rank[ident], ident) for ident, degree in indegree.items() if degree == 0)
    order: list[str] = []
    while ready:
        _, ident = ready.pop(0)
        order.append(ident)
        for child in sorted(children[ident], key=input_rank.__getitem__):
            indegree[child] -= 1
            if indegree[child] == 0:
                ready.append((input_rank[child], child))
                ready.sort()
    if len(order) != len(by_id):
        cyclic = sorted(ident for ident, degree in indegree.items() if degree > 0)
        raise ValueError("record dependency cycle: " + ", ".join(cyclic))
    return order