"""Immutable exact-rational interval candidate index for the rebuilt 0.4 line.""" from __future__ import annotations import hashlib from typing import Any, Iterable, Mapping, Sequence from tom_world03.canonical import attach_hash, canonical_bytes, require_hash from tom_world03.interval import ClosedInterval from tom_world03.rational import Q from .model import ContinuationRelation INDEX_SCHEMA = "TOM-RELATION-INTERVAL-INDEX-0.4.1" INDEX_VERSION = "0.4.1" PLAN_SCHEMA = "TOM-INTERVAL-CANDIDATE-PLAN-0.4.1" def _relation_set(relations: Iterable[ContinuationRelation]) -> list[dict[str, Any]]: return [ { "id": relation.id, "content_hash": relation.content_hash, "active_time": relation.active_time.to_record(), "support_id": relation.support_id, "compatibility_id": relation.compatibility_id, "fire_policy": relation.fire_policy, } for relation in sorted(relations, key=lambda item: item.id) ] def build_interval_index( relations: Iterable[ContinuationRelation], *, seed_sha256: str, ) -> dict[str, Any]: relation_list = list(relations) entries = [ { "lower": relation.active_time.lower.to_record(), "upper": relation.active_time.upper.to_record(), "relation_id": relation.id, "relation_hash": relation.content_hash, "support_id": relation.support_id, "compatibility_id": relation.compatibility_id, "fire_policy": relation.fire_policy, } for relation in relation_list ] entries.sort(key=lambda item: ( Q.from_value(item["lower"]).as_fraction(), Q.from_value(item["upper"]).as_fraction(), str(item["relation_id"]), str(item["relation_hash"]), )) postings_support: dict[str, list[str]] = {} postings_compatibility: dict[str, list[str]] = {} for relation in relation_list: postings_support.setdefault(relation.support_id, []).append(relation.id) postings_compatibility.setdefault(relation.compatibility_id, []).append(relation.id) for mapping in (postings_support, postings_compatibility): for key in mapping: mapping[key].sort() relation_set = _relation_set(relation_list) relation_set_hash = "sha256:" + hashlib.sha256(canonical_bytes(relation_set)).hexdigest() return attach_hash({ "schema": INDEX_SCHEMA, "version": INDEX_VERSION, "seed_sha256": seed_sha256, "relation_count": len(relation_list), "relation_set_hash": relation_set_hash, "algorithm": "start-sorted exact closed-interval overlap with safe support/compatibility postings", "ordering": ["active_lower", "active_upper", "relation_id", "relation_hash"], "entries": entries, "by_support": {key: postings_support[key] for key in sorted(postings_support)}, "by_compatibility": {key: postings_compatibility[key] for key in sorted(postings_compatibility)}, }) def validate_interval_index( index: Mapping[str, Any], relations: Sequence[ContinuationRelation], *, seed_sha256: str, ) -> None: require_hash(index, label="0.4.1 interval index") if index.get("schema") != INDEX_SCHEMA or index.get("version") != INDEX_VERSION: raise ValueError("unsupported 0.4.1 interval index schema/version") if index.get("seed_sha256") != seed_sha256: raise ValueError("interval index seed binding mismatch") expected = build_interval_index(relations, seed_sha256=seed_sha256) if canonical_bytes(index) != canonical_bytes(expected): raise ValueError("interval index does not match the relation set") def _prefix_limit(entries: Sequence[Mapping[str, Any]], upper: Q) -> int: lo, hi = 0, len(entries) while lo < hi: mid = (lo + hi) // 2 if Q.from_value(entries[mid]["lower"]) <= upper: lo = mid + 1 else: hi = mid return lo def query_interval_index( index: Mapping[str, Any], interval: ClosedInterval, *, allowed_support_ids: set[str] | None = None, allowed_compatibility_ids: set[str] | None = None, excluded_relation_ids: set[str] | None = None, ) -> tuple[list[str], dict[str, Any]]: """Return a complete candidate set for one exact query bracket. Posting filters are used only with caller-supplied *safe* allowed-ID sets. The solver constructs those sets conservatively from exact state intervals: a support or compatibility ID is removed only when it is impossible over the whole bracket. This preserves no-false-negative semantics. """ require_hash(index, label="0.4.1 interval index") entries_raw = index.get("entries") if not isinstance(entries_raw, list): raise ValueError("interval index entries must be an array") entries: list[Mapping[str, Any]] = [] for item in entries_raw: if not isinstance(item, Mapping): raise ValueError("interval index entry must be an object") entries.append(item) excluded = excluded_relation_ids or set() limit = _prefix_limit(entries, interval.upper) selected: list[str] = [] rejected_upper = 0 rejected_support = 0 rejected_compatibility = 0 rejected_fired = 0 for item in entries[:limit]: if Q.from_value(item["upper"]) < interval.lower: rejected_upper += 1 continue relation_id = str(item["relation_id"]) if relation_id in excluded: rejected_fired += 1 continue support_id = str(item["support_id"]) if allowed_support_ids is not None and support_id not in allowed_support_ids: rejected_support += 1 continue compatibility_id = str(item["compatibility_id"]) if allowed_compatibility_ids is not None and compatibility_id not in allowed_compatibility_ids: rejected_compatibility += 1 continue selected.append(relation_id) selected = sorted(set(selected)) plan = attach_hash({ "schema": PLAN_SCHEMA, "planner": "indexed", "index_hash": index["content_hash"], "query_interval": interval.to_record(), "relation_count": int(index["relation_count"]), "lower_endpoint_prefix": limit, "rejected_by_upper_endpoint": rejected_upper, "allowed_support_ids": None if allowed_support_ids is None else sorted(allowed_support_ids), "rejected_by_support": rejected_support, "allowed_compatibility_ids": None if allowed_compatibility_ids is None else sorted(allowed_compatibility_ids), "rejected_by_compatibility": rejected_compatibility, "excluded_fired_relations": sorted(excluded), "rejected_as_already_fired": rejected_fired, "candidate_count": len(selected), "candidate_ids": selected, "completeness_rule": "closed overlap plus only proven-impossible support/compatibility filters", }) return selected, plan def exhaustive_candidates( relations: Sequence[ContinuationRelation], interval: ClosedInterval, *, excluded_relation_ids: set[str] | None = None, ) -> tuple[list[str], dict[str, Any]]: excluded = excluded_relation_ids or set() selected = sorted(relation.id for relation in relations if relation.id not in excluded) overlap = sorted( relation.id for relation in relations if relation.id not in excluded and relation.active_time.intersection(interval) is not None ) plan = attach_hash({ "schema": PLAN_SCHEMA, "planner": "exhaustive", "index_hash": None, "query_interval": interval.to_record(), "relation_count": len(relations), "lower_endpoint_prefix": len(relations), "rejected_by_upper_endpoint": 0, "allowed_support_ids": None, "rejected_by_support": 0, "allowed_compatibility_ids": None, "rejected_by_compatibility": 0, "excluded_fired_relations": sorted(excluded), "rejected_as_already_fired": len(excluded.intersection({r.id for r in relations})), "candidate_count": len(selected), "candidate_ids": selected, "overlap_count_after_exhaustive_test": len(overlap), "overlap_ids_after_exhaustive_test": overlap, "completeness_rule": "every unfired relation enters the exhaustive solver loop", }) return selected, plan