"""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