"""Query-first world semantics over content-addressed records and TOMAGI. World & Query Kernel 0.2 adds immutable-index planning, checkpoint-aware state replay, and deterministic batch certificates while preserving the 0.1 semantic query results and exact-discrete event model. """ from __future__ import annotations from dataclasses import replace from typing import Any, Iterable, Mapping, Sequence from tomagi.core import ( PHI_STATES, THETA_STATES, TIME_STATES, State, Program, STATUS_HALT, i32, mix32, norm_mod, run, step, u32, ) from tomagi.format import loads from .canonical import attach_hash, canonical_bytes, digest_bytes, verify_hash from .expression import ExpressionBudget, evaluate_expression from .planner import PLANNER_MODES, QueryPlanner, make_plan from .records import make_record from .store import TRANSACTION_SCHEMA, WorldStore _STATE_FIELDS = tuple(State.__dataclass_fields__) _SIGNED_FIELDS = frozenset({ "rho", "theta", "tick", "phi", "vrho", "vtheta", "vtick", "vphi", "residual", }) def state_dict(state: State) -> dict[str, int]: return {name: getattr(state, name) for name in _STATE_FIELDS} def state_from_mapping(value: Mapping[str, Any], *, base: State | None = None) -> State: fields = state_dict(base or State()) unknown = sorted(set(value) - set(fields)) if unknown: raise ValueError("unknown State64 fields: " + ", ".join(unknown)) for name, raw in value.items(): if isinstance(raw, bool) or not isinstance(raw, int): raise TypeError(f"State64 field {name} must be an integer") fields[name] = i32(raw) if name in _SIGNED_FIELDS else u32(raw) return State(**fields) def _record_sources( state: State, *, pre_state: State | None = None, context: Mapping[str, Any] | None = None, left: State | None = None, right: State | None = None, event: Mapping[str, Any] | None = None, ) -> dict[str, Any]: result: dict[str, Any] = { "state": state_dict(state), "pre_state": state_dict(pre_state or state), "context": dict(context or {}), } if left is not None: result["left"] = state_dict(left) if right is not None: result["right"] = state_dict(right) if event is not None: result["event"] = dict(event) return result def _zero_test(value: Any, mode: str) -> bool: if mode == "equal_zero": return isinstance(value, int) and not isinstance(value, bool) and value == 0 if mode == "less_equal_zero": return isinstance(value, int) and not isinstance(value, bool) and value <= 0 if mode == "contains_zero": if not isinstance(value, Mapping): raise TypeError("contains_zero relation must return an interval object") lower = value.get("lower") upper = value.get("upper") if isinstance(lower, bool) or not isinstance(lower, int) or isinstance(upper, bool) or not isinstance(upper, int): raise TypeError("interval bounds must be integers") return lower <= 0 <= upper raise ValueError(f"unsupported zero test {mode}") def _triggered(previous: Any, current: Any, mode: str, zero_mode: str) -> bool: previous_zero = _zero_test(previous, zero_mode) current_zero = _zero_test(current, zero_mode) if mode == "zero": return current_zero if mode == "enter_zero": return current_zero and not previous_zero if mode == "enter_nonpositive": if not isinstance(previous, int) or isinstance(previous, bool) or not isinstance(current, int) or isinstance(current, bool): raise TypeError("enter_nonpositive requires integer residuals") return previous > 0 and current <= 0 if mode == "crossing": if current_zero: return True if isinstance(previous, int) and not isinstance(previous, bool) and isinstance(current, int) and not isinstance(current, bool): return (previous < 0 < current) or (current < 0 < previous) return current_zero and not previous_zero raise ValueError(f"unsupported trigger mode {mode}") def _direction(previous: Any, current: Any) -> str: if isinstance(previous, int) and not isinstance(previous, bool) and isinstance(current, int) and not isinstance(current, bool): if previous > 0 and current == 0: return "positive_to_zero" if previous < 0 and current == 0: return "negative_to_zero" if previous < 0 < current: return "negative_to_positive" if previous > 0 > current: return "positive_to_negative" if current > previous: return "increasing" if current < previous: return "decreasing" return "stationary" return "interval_or_symbolic" def _guard_margin(previous: Any, current: Any) -> int | dict[str, int] | None: if isinstance(previous, int) and not isinstance(previous, bool) and isinstance(current, int) and not isinstance(current, bool): if current == 0: return abs(previous) return min(abs(previous), abs(current)) if isinstance(current, Mapping) and isinstance(current.get("lower"), int) and isinstance(current.get("upper"), int): return {"lower": int(current["lower"]), "upper": int(current["upper"])} return None def _length_prefixed(values: Sequence[Mapping[str, Any]]) -> bytes: output = bytearray() for value in values: data = canonical_bytes(value) output.extend(len(data).to_bytes(8, "little")) output.extend(data) return bytes(output) def _work_sum(value: Any) -> dict[str, int]: totals: dict[str, int] = {} if isinstance(value, Mapping): work = value.get("work") if isinstance(work, Mapping): for key, raw in work.items(): if isinstance(raw, int) and not isinstance(raw, bool): totals[str(key)] = totals.get(str(key), 0) + raw for item in value.values(): nested = _work_sum(item) for key, raw in nested.items(): totals[key] = totals.get(key, 0) + raw elif isinstance(value, list): for item in value: nested = _work_sum(item) for key, raw in nested.items(): totals[key] = totals.get(key, 0) + raw return totals class QueryEngine: """Native exact-discrete TOM-SRS queries for one immutable world commit.""" def __init__( self, store: WorldStore, *, commit: str | None = None, max_query_steps: int = 100_000, max_expression_nodes: int = 10_000, max_expression_depth: int = 64, planner_mode: str = "indexed", use_checkpoints: bool = True, ) -> None: store.validate() if planner_mode not in PLANNER_MODES: raise ValueError(f"planner_mode must be one of {sorted(PLANNER_MODES)}") self.store = store self.commit = commit or store.head if self.commit is None: raise ValueError("query engine requires a committed world") self.commit_record = store.read_commit(self.commit) self.snapshot = store.read_snapshot(str(self.commit_record["snapshot_hash"])) self.max_query_steps = max_query_steps self.max_expression_nodes = max_expression_nodes self.max_expression_depth = max_expression_depth self.planner_mode = planner_mode self.use_checkpoints = use_checkpoints self._program_cache: dict[str, Program] = {} def _planner(self, mode: str | None = None) -> QueryPlanner: return QueryPlanner(self.store, commit=self.commit, mode=mode or self.planner_mode) def _expression_budget(self) -> ExpressionBudget: return ExpressionBudget(self.max_expression_nodes, self.max_expression_depth) def definition_at(self, ident: str) -> dict[str, Any]: return self.store.read_record(ident, commit=self.commit) def verify_definition(self, ident: str) -> dict[str, Any]: result = self.store.verify_record(ident, commit=self.commit) if result.get("valid"): record = self.store.read_record(ident, commit=self.commit) result["is_definition"] = record["record_type"] == "definition" result["commit"] = self.commit return result def _instance(self, instance_id: str) -> dict[str, Any]: record = self.store.read_record(instance_id, commit=self.commit) if record["record_type"] != "instance": raise TypeError(f"{instance_id} is not an instance") return record def _program_for_instance(self, instance_id: str) -> tuple[Program, dict[str, Any], str]: instance = self._instance(instance_id) blob_id = str(instance["payload"]["program_blob_id"]) blob_index = self.snapshot["blobs"] if blob_id not in blob_index: raise ValueError(f"instance {instance_id} program blob is absent from snapshot: {blob_id}") blob_hash = str(blob_index[blob_id]) if blob_hash not in self._program_cache: self._program_cache[blob_hash] = loads(self.store.read_blob(blob_hash)) program = self._program_cache[blob_hash] override = instance["payload"].get("initial_state") if isinstance(override, Mapping): program = Program( cells=program.cells, entry=program.entry, seed=program.seed, default_ticks=program.default_ticks, initial_state=state_from_mapping(override, base=program.initial_state), flags=program.flags, ) return program, instance, blob_hash def _checkpoint_start( self, instance_id: str, target_tick: int, *, planner_mode: str | None = None, ) -> tuple[State | None, int, int, dict[str, Any]]: mode = planner_mode or self.planner_mode if not self.use_checkpoints: plan = make_plan( commit=self.commit, snapshot_hash=str(self.snapshot["content_hash"]), indexes_hash=str(self.snapshot.get("indexes_hash")) if self.snapshot.get("indexes_hash") else None, mode=mode, operation="select_checkpoint", stages=[{ "name": "checkpoint_policy", "mechanism": "disabled", "input_count": 0, "output_count": 0, "detail": {"target_tick": target_tick}, }], selected_ids=[], work={"record_reads": 0, "index_lookups": 0}, ) return None, 0, 0, plan selected, plan = self._planner(mode).checkpoint_ids(instance_id, target_tick) if not selected: return None, 0, 0, plan record = self.store.read_record(selected[0], commit=self.commit) if record["record_type"] != "checkpoint": raise TypeError(f"planned checkpoint {selected[0]} is not a checkpoint record") program, instance, blob_hash = self._program_for_instance(instance_id) del program payload = record["payload"] if payload["instance_hash"] != instance["content_hash"]: raise ValueError(f"checkpoint {record['id']} instance hash does not match current snapshot") if payload["program_blob_hash"] != blob_hash: raise ValueError(f"checkpoint {record['id']} program blob hash does not match current snapshot") source_commit = str(payload["source_commit"]) if not self.store.is_ancestor(source_commit, self.commit): raise ValueError(f"checkpoint {record['id']} source commit is not in query commit ancestry") tick = int(payload["tick"]) if tick > target_tick: raise ValueError("planner selected a checkpoint after the target tick") executed_steps = int(payload.get("executed_steps", tick)) return state_from_mapping(payload["state"]), tick, executed_steps, plan def _replay_state( self, instance_id: str, target_tick: int, *, include_trace: bool, planner_mode: str | None = None, ) -> tuple[State, list[dict[str, int]], dict[str, Any], dict[str, Any]]: if isinstance(target_tick, bool) or not isinstance(target_tick, int) or target_tick < 0: raise ValueError("target tick must be a nonnegative integer") if target_tick > self.max_query_steps: raise ValueError(f"target tick exceeds query budget: {target_tick} > {self.max_query_steps}") mode = planner_mode or self.planner_mode program, instance, _ = self._program_for_instance(instance_id) checkpoint_state, checkpoint_tick, checkpoint_executed, checkpoint_plan = self._checkpoint_start( instance_id, target_tick, planner_mode=mode ) remaining = target_tick - checkpoint_tick state, trace = run(program, ticks=remaining, state=checkpoint_state, trace=True) for item in trace: item["step"] = checkpoint_tick + int(item["step"]) actual_replayed_steps = len(trace) executed_steps = checkpoint_executed + actual_replayed_steps if not include_trace: trace = [] replay_plan = attach_hash({ "schema": "TOM-STATE-REPLAY-PLAN-0.2", "commit": self.commit, "instance_id": instance_id, "instance_hash": instance["content_hash"], "target_tick": target_tick, "checkpoint_tick": checkpoint_tick, "checkpoint_id": checkpoint_plan["selected_ids"][0] if checkpoint_plan["selected_ids"] else None, "checkpoint_selection": checkpoint_plan, "logical_steps": target_tick, "replayed_steps": actual_replayed_steps, "executed_steps": executed_steps, "saved_replay_steps": checkpoint_tick, "work": { "tomagi_steps": actual_replayed_steps, "checkpoint_record_reads": 1 if checkpoint_plan["selected_ids"] else 0, }, }) return state, trace, replay_plan, instance def _state_at_with_plan( self, instance_id: str, tick: int, *, include_trace: bool = False, planner_mode: str | None = None, ) -> tuple[dict[str, Any], dict[str, Any]]: state, trace, plan, instance = self._replay_state( instance_id, tick, include_trace=include_trace, planner_mode=planner_mode ) result: dict[str, Any] = { "schema": "TOM-STATE-AT-CERTIFICATE-0.2", "commit": self.commit, "instance_id": instance_id, "instance_hash": instance["content_hash"], "requested_tick": tick, "executed_steps": int(plan["executed_steps"]), "state": state_dict(state), "status": "exact_discrete_replay", } if include_trace: result["trace"] = trace return attach_hash(result), plan def state_at(self, instance_id: str, tick: int, *, include_trace: bool = False) -> dict[str, Any]: result, _ = self._state_at_with_plan(instance_id, tick, include_trace=include_trace) return result def state_at_with_plan( self, instance_id: str, tick: int, *, include_trace: bool = False, planner_mode: str | None = None, ) -> dict[str, Any]: result, plan = self._state_at_with_plan( instance_id, tick, include_trace=include_trace, planner_mode=planner_mode ) return attach_hash({ "schema": "TOM-PLANNED-QUERY-CERTIFICATE-0.2", "query_kind": "state_at", "commit": self.commit, "result": result, "plan": plan, }) def trace(self, instance_id: str, ticks: int) -> dict[str, Any]: return self.state_at(instance_id, ticks, include_trace=True) def _context(self, instance: Mapping[str, Any], context: Mapping[str, Any] | None) -> dict[str, Any]: base = instance["payload"].get("context", {}) if not isinstance(base, Mapping): raise ValueError("instance context must be an object") merged = dict(base) if context: merged.update(context) return merged def _relation_records( self, instance_id: str, relation_ids: Sequence[str] | None, *, support_id: str | None = None, interval: tuple[int, int] | None = None, topology_sheet: int | None = None, planner_mode: str | None = None, ) -> tuple[list[dict[str, Any]], dict[str, Any]]: ids, plan = self._planner(planner_mode).relation_ids( instance_id, explicit_ids=relation_ids, support_id=support_id, interval=interval, topology_sheet=topology_sheet, ) if relation_ids is not None: requested = {str(value) for value in relation_ids} missing = sorted(requested - set(ids)) for ident in missing: try: record = self.store.read_record(ident, commit=self.commit) except KeyError as exc: raise KeyError(ident) from exc if record["record_type"] != "relation": raise TypeError(f"{ident} is not a relation") if record["payload"]["instance_id"] != instance_id: raise ValueError(f"relation {ident} targets a different instance") # A valid explicit relation can be absent only because a declared # support/interval/sheet filter removed it; that is not an error. relations = [self.store.read_record(ident, commit=self.commit) for ident in ids] relations.sort(key=lambda record: (int(record["payload"].get("priority", 0)), record["id"])) return relations, plan def _predicate_decisions( self, ids: Iterable[str], expected_type: str, *, state: State, pre_state: State, context: Mapping[str, Any], ) -> list[dict[str, Any]]: decisions = [] sources = _record_sources(state, pre_state=pre_state, context=context) for ident in ids: record = self.store.read_record(ident, commit=self.commit) if record["record_type"] != expected_type: raise TypeError(f"{ident} is not a {expected_type} record") value = evaluate_expression(record["payload"]["expression"], sources, budget=self._expression_budget()) if not isinstance(value, bool): raise TypeError(f"{expected_type} {ident} expression did not return bool") decisions.append({ "id": ident, "hash": record["content_hash"], "accepted": value, }) return decisions def _evaluate_relation( self, record: Mapping[str, Any], *, state: State, pre_state: State, context: Mapping[str, Any], ) -> tuple[Any, Any, list[dict[str, Any]], list[dict[str, Any]]]: payload = record["payload"] previous_sources = _record_sources(pre_state, pre_state=pre_state, context=context) current_sources = _record_sources(state, pre_state=pre_state, context=context) previous = evaluate_expression(payload["expression"], previous_sources, budget=self._expression_budget()) current = evaluate_expression(payload["expression"], current_sources, budget=self._expression_budget()) supports = self._predicate_decisions( payload.get("support_ids", []), "support", state=state, pre_state=pre_state, context=context ) compatibilities = self._predicate_decisions( payload.get("compatibility_ids", []), "compatibility", state=state, pre_state=pre_state, context=context ) return previous, current, supports, compatibilities def _apply_transition( self, transition_id: str | None, state: State, *, pre_state: State, context: Mapping[str, Any], event_context: Mapping[str, Any], ) -> tuple[State, str | None]: if transition_id is None: return replace(state), None record = self.store.read_record(transition_id, commit=self.commit) if record["record_type"] != "transition": raise TypeError(f"{transition_id} is not a transition") result = replace(state) sources = _record_sources(result, pre_state=pre_state, context=context, event=event_context) for operation in ("set", "add", "xor"): fields = record["payload"].get(operation, {}) for name in sorted(fields): if name not in _STATE_FIELDS: raise ValueError(f"transition {transition_id} names unknown State64 field {name}") value = evaluate_expression(fields[name], sources, budget=self._expression_budget()) if isinstance(value, bool) or not isinstance(value, int): raise TypeError(f"transition {transition_id}.{operation}.{name} must evaluate to integer") current = getattr(result, name) if operation == "set": updated = value elif operation == "add": updated = current + value else: updated = current ^ value setattr(result, name, i32(updated) if name in _SIGNED_FIELDS else u32(updated)) sources["state"] = state_dict(result) if record["payload"].get("normalize_periodic", True): result.theta = norm_mod(result.theta, THETA_STATES) result.tick = norm_mod(result.tick, TIME_STATES) result.phi = norm_mod(result.phi, PHI_STATES) result.orientation &= 1 result.branch &= 1 salt = record["payload"].get("lineage_salt") if salt is not None: if isinstance(salt, bool) or not isinstance(salt, int): raise TypeError("transition lineage_salt must be an integer") relation_word = int(str(event_context["relation_hash"])[7:15], 16) result.lineage = mix32(result.lineage ^ u32(salt) ^ u32(event_context["event_tick"]) ^ relation_word) return result, str(record["content_hash"]) def _event_certificate( self, *, instance: Mapping[str, Any], relation: Mapping[str, Any], after_tick: int, horizon: int, event_tick: int, pre_state: State, state: State, previous_residual: Any, residual: Any, support_decisions: list[dict[str, Any]], compatibility_decisions: list[dict[str, Any]], context: Mapping[str, Any], ) -> dict[str, Any]: payload = relation["payload"] event_spec_id = payload.get("event_spec_id") event_spec = None transition_id = None route = None confidence: Any = {"kind": "not_declared"} if event_spec_id is not None: event_spec = self.store.read_record(str(event_spec_id), commit=self.commit) if event_spec["record_type"] != "event_spec": raise TypeError(f"{event_spec_id} is not an event_spec") transition_id = event_spec["payload"].get("transition_id") route = event_spec["payload"].get("route") confidence = event_spec["payload"].get("confidence", confidence) event_context = { "event_tick": event_tick, "relation_id": relation["id"], "relation_hash": relation["content_hash"], "residual": residual, "previous_residual": previous_residual, } post_state, transition_hash = self._apply_transition( str(transition_id) if transition_id is not None else None, state, pre_state=pre_state, context=context, event_context=event_context, ) return attach_hash({ "schema": "TOM-EVENT-CERTIFICATE-0.1", "source_commit": self.commit, "query": { "instance_id": instance["id"], "after_tick": after_tick, "horizon": horizon, "relation_ids": [relation["id"]], }, "instance_hash": instance["content_hash"], "event_tick": event_tick, "relation_id": relation["id"], "relation_hash": relation["content_hash"], "event_spec_id": event_spec_id, "event_spec_hash": event_spec["content_hash"] if event_spec else None, "transition_id": transition_id, "transition_hash": transition_hash, "previous_residual": previous_residual, "residual": residual, "zero_test": payload.get("zero_test", "equal_zero"), "trigger": payload.get("trigger", "enter_zero"), "direction": _direction(previous_residual, residual), "guard_margin": _guard_margin(previous_residual, residual), "support": support_decisions, "compatibility": compatibility_decisions, "solver_status": "exact_discrete_scan", "confidence": confidence, "route": route, "pre_state": state_dict(pre_state), "event_state": state_dict(state), "post_state": state_dict(post_state), "context": dict(context), }) @staticmethod def _active_at(relation: Mapping[str, Any], logical_tick: int) -> bool: interval = relation["payload"].get("active_interval", relation["payload"].get("time_interval")) if not isinstance(interval, Mapping): return True return int(interval["start"]) <= logical_tick <= int(interval["end"]) def _scan_events( self, instance_id: str, *, after_tick: int, horizon: int, relation_ids: Sequence[str] | None, context: Mapping[str, Any] | None, stop_after_first: bool, support_id: str | None = None, topology_sheet: int | None = None, planner_mode: str | None = None, ) -> tuple[list[dict[str, Any]], dict[str, Any]]: if isinstance(after_tick, bool) or not isinstance(after_tick, int) or after_tick < 0: raise ValueError("after_tick must be a nonnegative integer") if isinstance(horizon, bool) or not isinstance(horizon, int) or horizon < 1: raise ValueError("horizon must be a positive integer") if after_tick + horizon > self.max_query_steps: raise ValueError( f"event scan exceeds query budget: {after_tick + horizon} > {self.max_query_steps}" ) mode = planner_mode or self.planner_mode program, instance, _ = self._program_for_instance(instance_id) relations, relation_plan = self._relation_records( instance_id, relation_ids, support_id=support_id, interval=(after_tick + 1, after_tick + horizon), topology_sheet=topology_sheet, planner_mode=mode, ) state, _, replay_plan, _ = self._replay_state( instance_id, after_tick, include_trace=False, planner_mode=mode ) merged_context = self._context(instance, context) logical_tick = after_tick events: list[dict[str, Any]] = [] relation_evaluations = 0 predicate_evaluations = 0 inactive_interval_skips = 0 ticks_scanned = 0 for _ in range(horizon): pre_state = replace(state) if not (state.status & STATUS_HALT): step(program, state) logical_tick += 1 ticks_scanned += 1 candidates: list[dict[str, Any]] = [] for relation in relations: if not self._active_at(relation, logical_tick): inactive_interval_skips += 1 continue previous, residual, supports, compatibilities = self._evaluate_relation( relation, state=state, pre_state=pre_state, context=merged_context, ) relation_evaluations += 1 predicate_evaluations += len(supports) + len(compatibilities) support_ok = all(decision["accepted"] for decision in supports) compatibility_ok = all(decision["accepted"] for decision in compatibilities) payload = relation["payload"] if support_ok and compatibility_ok and _triggered( previous, residual, str(payload.get("trigger", "enter_zero")), str(payload.get("zero_test", "equal_zero")), ): candidates.append(self._event_certificate( instance=instance, relation=relation, after_tick=after_tick, horizon=horizon, event_tick=logical_tick, pre_state=pre_state, state=state, previous_residual=previous, residual=residual, support_decisions=supports, compatibility_decisions=compatibilities, context=merged_context, )) if candidates: candidates.sort(key=lambda certificate: ( int(self.store.read_record(certificate["relation_id"], commit=self.commit)["payload"].get("priority", 0)), certificate["relation_id"], )) events.extend(candidates) if stop_after_first: events = [candidates[0]] break scan_plan = attach_hash({ "schema": "TOM-EVENT-SCAN-PLAN-0.2", "commit": self.commit, "instance_id": instance_id, "mode": mode, "after_tick": after_tick, "horizon": horizon, "stop_after_first": stop_after_first, "relation_selection": relation_plan, "state_replay": replay_plan, "selected_relation_count": len(relations), "ticks_scanned": ticks_scanned, "relation_evaluations": relation_evaluations, "predicate_evaluations": predicate_evaluations, "inactive_interval_skips": inactive_interval_skips, "events_found": len(events), "work": { "tomagi_steps": int(replay_plan["work"]["tomagi_steps"]) + ticks_scanned, "relation_evaluations": relation_evaluations, "predicate_evaluations": predicate_evaluations, "record_reads": int(relation_plan["work"].get("record_reads", 0)), "index_lookups": int(relation_plan["work"].get("index_lookups", 0)), }, }) return events, scan_plan def next_event( self, instance_id: str, after_tick: int, *, horizon: int = 1024, relation_ids: Sequence[str] | None = None, context: Mapping[str, Any] | None = None, planner_mode: str | None = None, ) -> dict[str, Any] | None: events, _ = self._scan_events( instance_id, after_tick=after_tick, horizon=horizon, relation_ids=relation_ids, context=context, stop_after_first=True, planner_mode=planner_mode, ) return events[0] if events else None def next_event_with_plan( self, instance_id: str, after_tick: int, *, horizon: int = 1024, relation_ids: Sequence[str] | None = None, context: Mapping[str, Any] | None = None, planner_mode: str | None = None, ) -> dict[str, Any]: events, plan = self._scan_events( instance_id, after_tick=after_tick, horizon=horizon, relation_ids=relation_ids, context=context, stop_after_first=True, planner_mode=planner_mode, ) return attach_hash({ "schema": "TOM-PLANNED-QUERY-CERTIFICATE-0.2", "query_kind": "next_event", "commit": self.commit, "result": events[0] if events else None, "plan": plan, }) def events_in_support( self, instance_id: str, *, start_tick: int, end_tick: int, support_id: str | None = None, relation_ids: Sequence[str] | None = None, context: Mapping[str, Any] | None = None, planner_mode: str | None = None, ) -> dict[str, Any]: result, _ = self._events_in_support_with_plan( instance_id, start_tick=start_tick, end_tick=end_tick, support_id=support_id, relation_ids=relation_ids, context=context, planner_mode=planner_mode, ) return result def _events_in_support_with_plan( self, instance_id: str, *, start_tick: int, end_tick: int, support_id: str | None, relation_ids: Sequence[str] | None, context: Mapping[str, Any] | None, planner_mode: str | None, ) -> tuple[dict[str, Any], dict[str, Any]]: if end_tick <= start_tick: raise ValueError("end_tick must be greater than start_tick") if support_id is not None: support_record = self.store.read_record(support_id, commit=self.commit) if support_record["record_type"] != "support": raise TypeError(f"{support_id} is not a support record") events, plan = self._scan_events( instance_id, after_tick=start_tick, horizon=end_tick - start_tick, relation_ids=relation_ids, context=context, stop_after_first=False, support_id=support_id, planner_mode=planner_mode, ) return attach_hash({ "schema": "TOM-EVENTS-IN-SUPPORT-CERTIFICATE-0.2", "commit": self.commit, "instance_id": instance_id, "support_id": support_id, "interval": {"start_exclusive": start_tick, "end_inclusive": end_tick}, "event_count": len(events), "events": events, }), plan def events_in_support_with_plan( self, instance_id: str, *, start_tick: int, end_tick: int, support_id: str | None = None, relation_ids: Sequence[str] | None = None, context: Mapping[str, Any] | None = None, planner_mode: str | None = None, ) -> dict[str, Any]: result, plan = self._events_in_support_with_plan( instance_id, start_tick=start_tick, end_tick=end_tick, support_id=support_id, relation_ids=relation_ids, context=context, planner_mode=planner_mode, ) return attach_hash({ "schema": "TOM-PLANNED-QUERY-CERTIFICATE-0.2", "query_kind": "events_in_support", "commit": self.commit, "result": result, "plan": plan, }) def _compatible_with_plan( self, left_instance_id: str, right_instance_id: str, compatibility_id: str, *, tick: int, context: Mapping[str, Any] | None, planner_mode: str | None, ) -> tuple[dict[str, Any], dict[str, Any]]: record = self.store.read_record(compatibility_id, commit=self.commit) if record["record_type"] != "compatibility": raise TypeError(f"{compatibility_id} is not a compatibility record") left, _, left_plan, _ = self._replay_state( left_instance_id, tick, include_trace=False, planner_mode=planner_mode ) right, _, right_plan, _ = self._replay_state( right_instance_id, tick, include_trace=False, planner_mode=planner_mode ) sources = _record_sources(left, left=left, right=right, context=context) value = evaluate_expression(record["payload"]["expression"], sources, budget=self._expression_budget()) if not isinstance(value, bool): raise TypeError("compatibility expression must return bool") result = attach_hash({ "schema": "TOM-COMPATIBILITY-CERTIFICATE-0.1", "commit": self.commit, "compatibility_id": compatibility_id, "compatibility_hash": record["content_hash"], "left_instance_id": left_instance_id, "right_instance_id": right_instance_id, "tick": tick, "left_state": state_dict(left), "right_state": state_dict(right), "compatible": value, "context": dict(context or {}), }) plan = attach_hash({ "schema": "TOM-COMPATIBILITY-QUERY-PLAN-0.2", "commit": self.commit, "mode": planner_mode or self.planner_mode, "left_replay": left_plan, "right_replay": right_plan, "expression_nodes_limit": self.max_expression_nodes, "work": { "tomagi_steps": int(left_plan["work"]["tomagi_steps"]) + int(right_plan["work"]["tomagi_steps"]), "compatibility_evaluations": 1, }, }) return result, plan def compatible( self, left_instance_id: str, right_instance_id: str, compatibility_id: str, *, tick: int, context: Mapping[str, Any] | None = None, planner_mode: str | None = None, ) -> dict[str, Any]: result, _ = self._compatible_with_plan( left_instance_id, right_instance_id, compatibility_id, tick=tick, context=context, planner_mode=planner_mode, ) return result def compatible_with_plan( self, left_instance_id: str, right_instance_id: str, compatibility_id: str, *, tick: int, context: Mapping[str, Any] | None = None, planner_mode: str | None = None, ) -> dict[str, Any]: result, plan = self._compatible_with_plan( left_instance_id, right_instance_id, compatibility_id, tick=tick, context=context, planner_mode=planner_mode, ) return attach_hash({ "schema": "TOM-PLANNED-QUERY-CERTIFICATE-0.2", "query_kind": "compatible", "commit": self.commit, "result": result, "plan": plan, }) def make_checkpoint_record(self, instance_id: str, tick: int) -> dict[str, Any]: """Construct a checkpoint by exact replay from the instance root state.""" root_engine = QueryEngine( self.store, commit=self.commit, max_query_steps=self.max_query_steps, max_expression_nodes=self.max_expression_nodes, max_expression_depth=self.max_expression_depth, planner_mode="exhaustive", use_checkpoints=False, ) certificate = root_engine.state_at(instance_id, tick) _, instance, blob_hash = self._program_for_instance(instance_id) safe_instance = instance_id.replace("/", "_") ident = f"checkpoint:{safe_instance}:{tick:012d}:{instance['content_hash'][7:19]}" return make_record( "checkpoint", ident, { "instance_id": instance_id, "tick": tick, "executed_steps": certificate["executed_steps"], "state": certificate["state"], "instance_hash": instance["content_hash"], "program_blob_hash": blob_hash, "source_commit": self.commit, "state_certificate_hash": certificate["content_hash"], "topology_sheet": certificate["state"]["sheet"], "generative_address": {"instance_id": instance_id, "tick": tick}, "time_interval": {"start": tick, "end": tick}, }, dependencies=[instance_id], provenance={"method": "exact_replay_from_root", "source_commit": self.commit}, ) def commit_checkpoints( self, instance_id: str, ticks: Sequence[int], *, message: str = "append exact state checkpoints", ) -> dict[str, Any]: unique = sorted(set(ticks)) if len(unique) != len(ticks): raise ValueError("checkpoint tick list must not contain duplicates") records = [self.make_checkpoint_record(instance_id, tick) for tick in unique] transaction = attach_hash({ "schema": TRANSACTION_SCHEMA, "seed_sha256": self.commit_record["seed_sha256"], "base_commit": self.store.head, "sequence": int(self.commit_record["sequence"]) + 1, "message": message, "records": records, "blobs": [], "provenance": { "checkpoint_instance": instance_id, "source_commit": self.commit, "ticks": unique, }, }) return self.store.commit_transaction(transaction) def batch( self, requests: Sequence[Mapping[str, Any]], *, planner_mode: str | None = None, ) -> dict[str, Any]: """Execute a finite declared batch in array order with stable reduction.""" mode = planner_mode or self.planner_mode if mode not in PLANNER_MODES: raise ValueError(f"planner mode must be one of {sorted(PLANNER_MODES)}") seen: set[str] = set() results: list[dict[str, Any]] = [] semantic_results: list[Mapping[str, Any]] = [] for position, request in enumerate(requests): if not isinstance(request, Mapping): raise ValueError("batch requests must be objects") ident = request.get("id") operation = request.get("operation") parameters = request.get("parameters", {}) if not isinstance(ident, str) or not ident: raise ValueError("batch request id must be a nonempty string") if ident in seen: raise ValueError(f"duplicate batch request id: {ident}") seen.add(ident) if not isinstance(operation, str): raise ValueError(f"batch request {ident} operation must be a string") if not isinstance(parameters, Mapping): raise ValueError(f"batch request {ident} parameters must be an object") if operation == "state_at": planned = self.state_at_with_plan( str(parameters["instance_id"]), int(parameters["tick"]), include_trace=bool(parameters.get("include_trace", False)), planner_mode=mode, ) elif operation == "next_event": planned = self.next_event_with_plan( str(parameters["instance_id"]), int(parameters["after_tick"]), horizon=int(parameters.get("horizon", 1024)), relation_ids=parameters.get("relation_ids"), context=parameters.get("context"), planner_mode=mode, ) elif operation == "events_in_support": planned = self.events_in_support_with_plan( str(parameters["instance_id"]), start_tick=int(parameters["start_tick"]), end_tick=int(parameters["end_tick"]), support_id=parameters.get("support_id"), relation_ids=parameters.get("relation_ids"), context=parameters.get("context"), planner_mode=mode, ) elif operation == "compatible": planned = self.compatible_with_plan( str(parameters["left_instance_id"]), str(parameters["right_instance_id"]), str(parameters["compatibility_id"]), tick=int(parameters["tick"]), context=parameters.get("context"), planner_mode=mode, ) elif operation == "definition_at": semantic = self.definition_at(str(parameters["id"])) trivial_plan = make_plan( commit=self.commit, snapshot_hash=str(self.snapshot["content_hash"]), indexes_hash=str(self.snapshot.get("indexes_hash")) if self.snapshot.get("indexes_hash") else None, mode=mode, operation="definition_at", stages=[{ "name": "primary_id_map", "mechanism": "snapshot:records", "input_count": len(self.snapshot["records"]), "output_count": 1, "detail": {"id": parameters["id"]}, }], selected_ids=[str(parameters["id"])], work={"record_reads": 1}, ) planned = attach_hash({ "schema": "TOM-PLANNED-QUERY-CERTIFICATE-0.2", "query_kind": "definition_at", "commit": self.commit, "result": semantic, "plan": trivial_plan, }) else: raise ValueError(f"unsupported batch operation {operation}") semantic = planned["result"] semantic_results.append(semantic if isinstance(semantic, Mapping) else {"value": semantic}) results.append({ "position": position, "id": ident, "operation": operation, "result": semantic, "result_hash": semantic.get("content_hash") if isinstance(semantic, Mapping) else digest_bytes(canonical_bytes(semantic)), "plan": planned["plan"], "planned_certificate_hash": planned["content_hash"], }) reduction_input = _length_prefixed(semantic_results) work = _work_sum(results) return attach_hash({ "schema": "TOM-BATCH-QUERY-CERTIFICATE-0.2", "version": "0.2.0", "commit": self.commit, "planner_mode": mode, "request_count": len(results), "reduction_order": "declared_array_order", "request_ids": [item["id"] for item in results], "semantic_reduction_hash": digest_bytes(reduction_input), "semantic_result_hashes": [item["result_hash"] for item in results], "results": results, "work": {key: work[key] for key in sorted(work)}, }) def reconstruct(self, certificate_or_lineage: Mapping[str, Any] | str) -> dict[str, Any]: if isinstance(certificate_or_lineage, str): record = self.store.read_record(certificate_or_lineage, commit=self.commit) if record["record_type"] not in {"event", "lineage"}: raise TypeError("reconstruct ID must refer to an event or lineage record") certificate = record["payload"].get("certificate") if not isinstance(certificate, Mapping): raise ValueError("event/lineage record has no embedded certificate") else: certificate = certificate_or_lineage if certificate.get("schema") != "TOM-EVENT-CERTIFICATE-0.1" or not verify_hash(certificate): raise ValueError("event certificate is invalid") source_commit = str(certificate["source_commit"]) query = certificate["query"] engine = QueryEngine( self.store, commit=source_commit, max_query_steps=self.max_query_steps, max_expression_nodes=self.max_expression_nodes, max_expression_depth=self.max_expression_depth, planner_mode=self.planner_mode, use_checkpoints=self.use_checkpoints, ) recomputed = engine.next_event( str(query["instance_id"]), int(query["after_tick"]), horizon=int(query["horizon"]), relation_ids=list(query["relation_ids"]), context=certificate.get("context", {}), ) equal = recomputed is not None and canonical_bytes(recomputed) == canonical_bytes(certificate) return attach_hash({ "schema": "TOM-RECONSTRUCTION-CERTIFICATE-0.1", "requested_certificate_hash": certificate["content_hash"], "source_commit": source_commit, "recomputed_certificate_hash": recomputed["content_hash"] if recomputed else None, "byte_equal": equal, "status": "reconstructed" if equal else "mismatch", }) def event_records(self, certificate: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: if certificate.get("schema") != "TOM-EVENT-CERTIFICATE-0.1" or not verify_hash(certificate): raise ValueError("event certificate is invalid") suffix = str(certificate["content_hash"])[7:23] dependencies = [ str(certificate["query"]["instance_id"]), str(certificate["relation_id"]), ] for field in ("event_spec_id", "transition_id"): value = certificate.get(field) if isinstance(value, str): dependencies.append(value) event_id = f"event:{suffix}" event_record = make_record( "event", event_id, { "certificate": dict(certificate), "event_tick": certificate["event_tick"], "relation_id": certificate["relation_id"], "route": certificate.get("route"), "time_interval": {"start": certificate["event_tick"], "end": certificate["event_tick"]}, "topology_sheet": certificate["event_state"]["sheet"], "generative_address": {"event_certificate": certificate["content_hash"]}, }, dependencies=dependencies, provenance={"source_commit": certificate["source_commit"]}, ) lineage_id = f"lineage:{suffix}" lineage_record = make_record( "lineage", lineage_id, { "certificate": dict(certificate), "event_id": event_id, "parent_state_lineage": certificate["event_state"]["lineage"], "post_state_lineage": certificate["post_state"]["lineage"], "definition_hashes": { "relation": certificate["relation_hash"], "event_spec": certificate.get("event_spec_hash"), "transition": certificate.get("transition_hash"), }, "generative_address": {"event_id": event_id, "lineage": certificate["post_state"]["lineage"]}, "topology_sheet": certificate["post_state"]["sheet"], }, dependencies=[event_id], provenance={"source_commit": certificate["source_commit"]}, ) return event_record, lineage_record def commit_event(self, certificate: Mapping[str, Any], *, message: str = "commit verified event") -> dict[str, Any]: if certificate.get("source_commit") != self.store.head: raise ValueError("event certificate source commit must equal current HEAD") event_record, lineage_record = self.event_records(certificate) sequence = int(self.commit_record["sequence"]) + 1 transaction = attach_hash({ "schema": TRANSACTION_SCHEMA, "seed_sha256": self.commit_record["seed_sha256"], "base_commit": self.store.head, "sequence": sequence, "message": message, "records": [event_record, lineage_record], "blobs": [], "provenance": {"query_certificate": certificate["content_hash"]}, }) return self.store.commit_transaction(transaction)