Blame for src/python/tom_world04r/solver.py

"""Certified next-event solving for open piecewise-affine continuations.

The solver reuses the corrected 0.3 crossing certifier but changes the 0.4
continuation architecture: it searches forward over an open segment whose upper
bound is the world horizon.  The next event time is therefore a solver output,
not an input hidden in a relation's ``continuation_until`` field.
"""
from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Mapping, Sequence

from tom_world03.canonical import attach_hash
from tom_world03.expression import affine_coefficients
from tom_world03.interval import ClosedInterval
from tom_world03.rational import Q
from tom_world03.solver import CrossingResult, certify_crossing

from .index import exhaustive_candidates, query_interval_index
from .model import ContinuationRelation, ContinuationWorld, OpenSegment


class UnresolvedContinuation(ValueError):
    """Raised when earliest-event ordering cannot be certified exactly."""


@dataclass(frozen=True, slots=True)
class ContinuationCrossing:
    relation: ContinuationRelation
    segment: OpenSegment
    exact_root: Q
    source: CrossingResult
    certificate: Mapping[str, Any]


def event_order_key(result: ContinuationCrossing) -> tuple[Any, ...]:
    relation = result.relation
    return (
        result.exact_root.as_fraction(),
        relation.priority,
        relation.id,
        relation.event_id,
        relation.content_hash,
    )


def _unit_brackets(start: Q, end: Q) -> list[ClosedInterval]:
    """Partition ``[start,end]`` at integer boundaries without zero-width gaps."""
    if end < start:
        raise ValueError("search end is before start")
    if end == start:
        return [ClosedInterval.point(start)]
    result: list[ClosedInterval] = []
    cursor = start
    while cursor < end:
        next_integer = Q(cursor.floor() + 1)
        if next_integer <= cursor:
            next_integer = cursor + 1
        upper = min(end, next_integer)
        result.append(ClosedInterval(cursor, upper))
        cursor = upper
    return result


def _safe_gate_ids(
    world: ContinuationWorld,
    segment: OpenSegment,
    bracket: ClosedInterval,
) -> tuple[set[str], set[str], dict[str, Any]]:
    """Return IDs not proven impossible over the whole bracket.

    Removing a posting is sound only when exact interval evidence proves that
    no point in the bracket can satisfy that gate.  Ambiguous intervals remain
    candidates and are evaluated at the certified root by the corrected 0.3
    solver.
    """
    state_iv = segment.state_interval(bracket)
    supports: set[str] = set()
    support_reasons: dict[str, str] = {}
    for ident, support in world.supports.items():
        impossible = False
        for field, allowed in support.bounds.items():
            actual = state_iv.get(field)
            if actual is None or actual.intersection(allowed) is None:
                impossible = True
                break
        if not impossible:
            supports.add(ident)
        else:
            support_reasons[ident] = "proven-disjoint-state-interval"

    compatibilities: set[str] = set()
    compatibility_reasons: dict[str, str] = {}
    for ident, compatibility in world.compatibilities.items():
        impossible = False
        for field, expected in compatibility.equals.items():
            actual = state_iv.get(field)
            if actual is None or not actual.contains(expected):
                impossible = True
                break
        if not impossible:
            compatibilities.add(ident)
        else:
            compatibility_reasons[ident] = "expected-value-excluded-by-state-interval"

    audit = {
        "state_interval": {name: value.to_record() for name, value in sorted(state_iv.items())},
        "allowed_support_ids": sorted(supports),
        "excluded_supports": support_reasons,
        "allowed_compatibility_ids": sorted(compatibilities),
        "excluded_compatibilities": compatibility_reasons,
        "soundness_rule": "only exact interval impossibility may remove a posting",
    }
    return supports, compatibilities, audit


def _candidate_relations(
    world: ContinuationWorld,
    segment: OpenSegment,
    bracket: ClosedInterval,
    planner: str,
) -> tuple[list[ContinuationRelation], dict[str, Any], dict[str, Any]]:
    fired = set(segment.fired_relations)
    supports, compatibilities, gate_audit = _safe_gate_ids(world, segment, bracket)
    if planner == "indexed":
        ids, plan = query_interval_index(
            world.interval_index,
            bracket,
            allowed_support_ids=supports,
            allowed_compatibility_ids=compatibilities,
            excluded_relation_ids=fired,
        )
    elif planner == "exhaustive":
        ids, plan = exhaustive_candidates(world.relations, bracket, excluded_relation_ids=fired)
    else:
        raise ValueError("planner must be indexed or exhaustive")
    relation_map = world.relation_map()
    try:
        relations = [relation_map[ident] for ident in ids]
    except KeyError as exc:  # index validation should make this unreachable.
        raise ValueError(f"candidate index references unknown relation {exc.args[0]}") from exc
    return relations, plan, gate_audit


def _certify_relation_on_bracket(
    world: ContinuationWorld,
    segment: OpenSegment,
    relation: ContinuationRelation,
    bracket: ClosedInterval,
    *,
    refine_steps: int | None,
) -> ContinuationCrossing | None:
    active = relation.active_time.intersection(bracket)
    if active is None:
        return None

    proxy = world.interval_world(segment, (relation,))
    coefficients = affine_coefficients(relation.expression, segment)
    if coefficients is None:
        raise UnresolvedContinuation(
            f"relation {relation.id} is not affine in time on segment {segment.id}"
        )
    slope, intercept = coefficients
    if slope == Q(0):
        if intercept == Q(0):
            raise UnresolvedContinuation(
                f"relation {relation.id} is identically zero on segment {segment.id}; isolated event time is undefined"
            )
        return None

    source = certify_crossing(proxy, relation.relation03(), active, refine_steps=refine_steps)
    if not source.accepted:
        return None
    if source.exact_root is None:
        raise UnresolvedContinuation(
            f"relation {relation.id} has an accepted but nonexact bracket; exact continuation ordering is unavailable"
        )
    root = source.exact_root
    certificate = attach_hash({
        "schema": "TOM-CONTINUATION-CROSSING-0.4.1",
        "world_hash": world.content_hash,
        "segment_id": segment.id,
        "segment_hash": segment.content_hash,
        "segment_sequence": segment.sequence,
        "relation_id": relation.id,
        "relation_hash": relation.content_hash,
        "event_id": relation.event_id,
        "priority": relation.priority,
        "fire_policy": relation.fire_policy,
        "exact_root_time": root.to_record(),
        "source_certificate_hash": source.certificate["content_hash"],
        "source_certificate": dict(source.certificate),
        "state_transition": [operation.to_record() for operation in relation.transition],
        "rate_transition": [operation.to_record() for operation in relation.rate_transition],
    })
    return ContinuationCrossing(relation, segment, root, source, certificate)


def bracket_events(
    world: ContinuationWorld,
    segment: OpenSegment,
    bracket: ClosedInterval,
    *,
    after: Q,
    planner: str = "indexed",
    refine_steps: int | None = None,
) -> tuple[list[ContinuationCrossing], dict[str, Any]]:
    if not bracket.subset_of(segment.domain):
        raise ValueError("event bracket lies outside open segment")
    candidates, candidate_plan, gate_audit = _candidate_relations(world, segment, bracket, planner)
    accepted: list[ContinuationCrossing] = []
    rejected_or_absent = 0
    for relation in candidates:
        result = _certify_relation_on_bracket(
            world, segment, relation, bracket, refine_steps=refine_steps,
        )
        if result is None:
            rejected_or_absent += 1
            continue
        if result.exact_root <= after:
            # Strict post-state continuation: a relation fired at the previous
            # boundary cannot fire again merely because it remains zero there.
            rejected_or_absent += 1
            continue
        accepted.append(result)
    accepted.sort(key=event_order_key)
    plan = attach_hash({
        "schema": "TOM-CONTINUATION-BRACKET-PLAN-0.4.1",
        "world_hash": world.content_hash,
        "segment_id": segment.id,
        "segment_hash": segment.content_hash,
        "planner": planner,
        "after_exclusive": after.to_record(),
        "bracket": bracket.to_record(),
        "gate_audit": gate_audit,
        "candidate_plan_hash": candidate_plan["content_hash"],
        "candidate_plan": candidate_plan,
        "candidate_relations": len(candidates),
        "rejected_or_absent": rejected_or_absent,
        "accepted_crossings": len(accepted),
        "accepted_relation_ids": [item.relation.id for item in accepted],
    })
    return accepted, plan


def next_event_set(
    world: ContinuationWorld,
    segment: OpenSegment,
    after: Any | None = None,
    before: Any | None = None,
    *,
    planner: str = "indexed",
    refine_steps: int | None = None,
) -> dict[str, Any]:
    """Find the earliest exact event strictly after ``after``.

    Search proceeds bracket by bracket and stops at the first bracket containing
    an accepted exact root.  This makes candidate work explicit and ensures the
    interval index is used for next-event search rather than merely filtering a
    predeclared segment boundary.
    """
    after_q = segment.start if after is None else Q.from_value(after)
    before_q = segment.horizon if before is None else Q.from_value(before)
    if after_q < segment.start or after_q > segment.horizon:
        raise ValueError("after lies outside open segment")
    if before_q < after_q or before_q > segment.horizon:
        raise ValueError("before lies outside the requested forward interval")

    plans: list[dict[str, Any]] = []
    seen: set[tuple[str, str]] = set()
    selected: list[ContinuationCrossing] = []
    for bracket in _unit_brackets(after_q, before_q):
        events, plan = bracket_events(
            world,
            segment,
            bracket,
            after=after_q,
            planner=planner,
            refine_steps=refine_steps,
        )
        plans.append(plan)
        unique: list[ContinuationCrossing] = []
        for item in events:
            key = (item.relation.id, item.exact_root.to_text())
            if key not in seen:
                seen.add(key)
                unique.append(item)
        if unique:
            earliest = min(item.exact_root for item in unique)
            selected = [item for item in unique if item.exact_root == earliest]
            selected.sort(key=event_order_key)
            break

    total_candidates = sum(int(plan["candidate_relations"]) for plan in plans)
    if not selected:
        return attach_hash({
            "schema": "TOM-NEXT-CONTINUATION-EVENT-SET-0.4.1",
            "world_hash": world.content_hash,
            "segment_id": segment.id,
            "segment_hash": segment.content_hash,
            "segment_sequence": segment.sequence,
            "after_exclusive": after_q.to_record(),
            "before_inclusive": before_q.to_record(),
            "planner": planner,
            "status": "none",
            "fired_relations_before": list(segment.fired_relations),
            "scanned_brackets": len(plans),
            "total_candidate_relations": total_candidates,
            "plan_hashes": [plan["content_hash"] for plan in plans],
            "plans": plans,
            "events": [],
        })

    event_time = selected[0].exact_root
    relation_ids = [item.relation.id for item in selected]
    fired_after = sorted(set(segment.fired_relations).union(relation_ids))
    return attach_hash({
        "schema": "TOM-NEXT-CONTINUATION-EVENT-SET-0.4.1",
        "world_hash": world.content_hash,
        "segment_id": segment.id,
        "segment_hash": segment.content_hash,
        "segment_sequence": segment.sequence,
        "after_exclusive": after_q.to_record(),
        "before_inclusive": before_q.to_record(),
        "planner": planner,
        "status": "accepted",
        "event_time": event_time.to_record(),
        "simultaneity_basis": "exact equal reduced rational root on the same open segment",
        "event_count": len(selected),
        "event_order": [item.relation.event_id for item in selected],
        "relation_order": relation_ids,
        "relation_hashes": [item.relation.content_hash for item in selected],
        "ordering_rule": ["root_time", "priority", "relation_id", "event_id", "relation_hash"],
        "fired_relations_before": list(segment.fired_relations),
        "fired_relations_after": fired_after,
        "scanned_brackets": len(plans),
        "total_candidate_relations": total_candidates,
        "plan_hashes": [plan["content_hash"] for plan in plans],
        "plans": plans,
        "events": [dict(item.certificate) for item in selected],
    })