Blame for crosswire/core.py

"""Finite Mealy adapters, exact behavior comparisons and bounded task admission.



Only public reference data and raw transcripts enter CandidateSpace. Work counts

are cumulative: new reference transition executions, new adapter/word expansions,

and represented candidate-pair comparisons, including histogram-aggregated pairs.

Cached results incur none of those costs; pair counts are not primitive CPU operations.

"""

from collections import Counter, deque
from dataclasses import dataclass
from fractions import Fraction
from itertools import permutations, product
import math
from random import Random


def _integer(value, name, minimum=0, maximum=None):
    if type(value) is not int:
        raise TypeError(f'{name} must be an integer')
    if value < minimum or (maximum is not None and value > maximum):
        raise ValueError(f'{name} out of bounds')
    return value


def _word(value, inputs):
    try:
        result = tuple(value)
    except TypeError as exc:
        raise TypeError('word must be a sequence') from exc
    for symbol in result:
        _integer(symbol, 'input', maximum=inputs - 1)
    return result


@dataclass(frozen=True)
class Machine:
    name: str
    transitions: tuple
    inputs: int = 3
    outputs: int = 4
    initial: int = 0

    def __post_init__(self):
        if not isinstance(self.name, str) or not self.name:
            raise ValueError('machine name must be a nonempty string')
        _integer(self.inputs, 'inputs', 1)
        _integer(self.outputs, 'outputs', 1)
        rows = tuple(tuple(tuple(cell) for cell in row) for row in self.transitions)
        if not rows:
            raise ValueError('machine needs at least one state')
        _integer(self.initial, 'initial', maximum=len(rows) - 1)
        for row in rows:
            if len(row) != self.inputs:
                raise ValueError('transition rows must match input alphabet')
            for cell in row:
                if len(cell) != 2:
                    raise ValueError('transition cell must contain next state and output')
                _integer(cell[0], 'next state', maximum=len(rows) - 1)
                _integer(cell[1], 'output', maximum=self.outputs - 1)
        object.__setattr__(self, 'transitions', rows)

    def trace(self, word):
        state, result = self.initial, []
        for symbol in _word(word, self.inputs):
            state, output = self.transitions[state][symbol]
            result.append(output)
        return tuple(result)

    def to_dict(self):
        return {'name': self.name, 'transitions': [[list(c) for c in r] for r in self.transitions],
                'inputs': self.inputs, 'outputs': self.outputs, 'initial': self.initial}

    @classmethod
    def from_dict(cls, data):
        return cls(**data)


@dataclass(frozen=True)
class Contract:
    goal: int = 1
    hazard: int = 2
    horizon: int = 4

    def __post_init__(self):
        _integer(self.goal, 'goal', maximum=3)
        _integer(self.hazard, 'hazard', maximum=3)
        _integer(self.horizon, 'horizon', 1, 6)
        if self.goal == self.hazard:
            raise ValueError('goal and hazard must differ')

    def accepts(self, outputs):
        outputs = tuple(outputs)
        for symbol in outputs:
            # The owning Machine validates its alphabet; contracts may also be
            # used with larger alphabets than the original four-symbol fixtures.
            _integer(symbol, 'semantic output')
        return bool(outputs) and outputs[-1] == self.goal and self.hazard not in outputs

    def to_dict(self):
        return {'goal': self.goal, 'hazard': self.hazard, 'horizon': self.horizon}

    @classmethod
    def from_dict(cls, data):
        return cls(**data)


@dataclass(frozen=True)
class Adapter:
    id: int
    input_map: tuple
    output_map: tuple
    delay: int = 0
    init_output: int = 0

    def __post_init__(self):
        _integer(self.id, 'adapter id')
        _integer(self.delay, 'delay', minimum=0, maximum=1)
        for name in ('input_map', 'output_map'):
            mapping = tuple(getattr(self, name))
            if not mapping:
                raise ValueError('adapter alphabets cannot be empty')
            for symbol in mapping:
                _integer(symbol, name, maximum=len(mapping) - 1)
            if len(set(mapping)) != len(mapping):
                raise ValueError(f'{name} must be bijective')
            object.__setattr__(self, name, mapping)
        _integer(self.init_output, 'init_output', maximum=len(self.output_map) - 1)

    def to_dict(self):
        result = {'id': self.id, 'input_map': list(self.input_map), 'output_map': list(self.output_map)}
        if self.delay:
            result['delay'] = self.delay
            result['init_output'] = self.init_output
        return result


def words(inputs, max_length):
    _integer(inputs, 'inputs', 1)
    _integer(max_length, 'max_length')
    return tuple(word for length in range(1, max_length + 1) for word in product(range(inputs), repeat=length))


class CandidateSpace:
    def __init__(self, machine, contract, temporal=False):
        if not isinstance(machine, Machine) or not isinstance(contract, Contract):
            raise TypeError('CandidateSpace requires a Machine and Contract')
        if max(contract.goal, contract.hazard) >= machine.outputs:
            raise ValueError('contract symbols exceed source alphabet')
        self.machine, self.contract = machine, contract
        self.temporal = bool(temporal)
        adapters = [Adapter(i, a, b) for i, (a, b) in enumerate(
            product(permutations(range(machine.inputs)), permutations(range(machine.outputs))))]
        if self.temporal:
            idx = len(adapters)
            for a, b in product(permutations(range(machine.inputs)), permutations(range(machine.outputs))):
                for init_o in range(machine.outputs):
                    adapters.append(Adapter(idx, a, b, delay=1, init_output=init_o))
                    idx += 1
        self.adapters = tuple(adapters)
        self.work = {'source_steps': 0, 'candidate_expansions': 0, 'pair_checks': 0}
        self._source = {(): (machine.initial, ())}
        self._predictions = {}
        self._semantics = {}
        self._safe = {}
        self._signatures = {}
        self._pair_weights = {}
        self._deployment = words(machine.inputs, contract.horizon)
        self._probes = words(machine.inputs, 3)

    def _adapter(self, identifier):
        return self.adapters[_integer(identifier, 'adapter id', maximum=len(self.adapters) - 1)]

    def _ids(self, ids):
        ids = tuple(ids)
        for identifier in ids:
            self._adapter(identifier)
        return tuple(sorted(set(ids)))

    def _source_trace(self, mapped):
        # Iterative prefixes avoid a recursion bound for imported long traces.
        prefix = ()
        for symbol in mapped:
            next_prefix = prefix + (symbol,)
            if next_prefix not in self._source:
                state, outputs = self._source[prefix]
                state, output = self.machine.transitions[state][symbol]
                self.work['source_steps'] += 1
                self._source[next_prefix] = (state, outputs + (output,))
            prefix = next_prefix
        return self._source[prefix][1]

    def _evaluate(self, identifier, word):
        adapter = self._adapter(identifier)
        word = _word(word, self.machine.inputs)
        key = (identifier, word)
        if key not in self._predictions:
            semantic = self._source_trace(tuple(adapter.input_map[i] for i in word))
            self._semantics[key] = semantic
            if adapter.delay == 1:
                preds = []
                for t in range(len(semantic)):
                    if t == 0:
                        preds.append(adapter.output_map[adapter.init_output])
                    else:
                        preds.append(adapter.output_map[semantic[t - 1]])
                self._predictions[key] = tuple(preds)
            else:
                self._predictions[key] = tuple(adapter.output_map[o] for o in semantic)
            self.work['candidate_expansions'] += 1
        return key

    def predict(self, identifier, word):
        return self._predictions[self._evaluate(identifier, word)]

    def semantic(self, identifier, word):
        return self._semantics[self._evaluate(identifier, word)]

    def safe_words(self, identifier):
        self._adapter(identifier)
        if identifier not in self._safe:
            self._safe[identifier] = frozenset(w for w in self._deployment if self.contract.accepts(self.semantic(identifier, w)))
        return self._safe[identifier]

    def _evidence(self, transcripts, anchors):
        evidence = []
        for transcript in transcripts:
            if type(transcript.get('complete')) is not bool:
                raise ValueError('transcript needs a boolean complete field')
            if not transcript['complete']:
                continue
            word = _word(transcript['word'], self.machine.inputs)
            outputs = tuple(transcript['outputs'])
            if len(outputs) != len(word):
                raise ValueError('complete transcript word/output lengths differ')
            for symbol in outputs:
                _integer(symbol, 'raw output', maximum=self.machine.outputs - 1)
            evidence.append((word, outputs))
        validated = dict(anchors or {})
        for raw, semantic in validated.items():
            _integer(raw, 'anchor raw output', maximum=self.machine.outputs - 1)
            _integer(semantic, 'anchor semantic output', maximum=self.machine.outputs - 1)
        return evidence, validated

    def losses(self, transcripts, anchors=None):
        evidence, anchors = self._evidence(transcripts, anchors)
        result = {}
        for adapter in self.adapters:
            result[adapter.id] = sum(p != o for word, outputs in evidence for p, o in zip(self.predict(adapter.id, word), outputs))
            result[adapter.id] += sum(adapter.output_map[semantic] != raw for raw, semantic in anchors.items())
        return result

    def survivors(self, transcripts, anchors=None):
        return tuple(identifier for identifier, loss in self.losses(transcripts, anchors).items() if loss == 0)

    def level_ids(self, level):
        """Nested hypothesis classes: 'static' (144 memoryless maps) ⊂ 'temporal' (all)."""
        if level == 'static':
            return tuple(a.id for a in self.adapters if a.delay == 0)
        if level == 'temporal':
            return tuple(a.id for a in self.adapters)
        raise ValueError('unknown hypothesis level')

    def _decision_classes(self, ids):
        """Group survivors by their contract-relevant behaviour (safe deployment set)."""
        classes = {}
        for identifier in ids:
            classes.setdefault(self.safe_words(identifier), []).append(identifier)
        return list(classes.items())

    def distinguishing_word(self, a, b, max_length):
        """Shortest lexicographic input word on which adapters a and b predict different raw outputs.



        Exact BFS over the joint state (reference state under a, under b, and each

        adapter's one-step output buffer). Unlike the fixed probe pool, the witness

        may be longer than three actions; it is bounded only by ``max_length``.

        """
        A, B = self._adapter(a), self._adapter(b)
        start = (self.machine.initial, self.machine.initial,
                 A.init_output if A.delay else None, B.init_output if B.delay else None)
        visited, pending = {start}, deque([(start, ())])
        while pending:
            (qa, qb, ba, bb), prefix = pending.popleft()
            if len(prefix) >= max_length:
                continue
            for symbol in range(self.machine.inputs):
                na, sa = self.machine.transitions[qa][A.input_map[symbol]]
                nb, sb = self.machine.transitions[qb][B.input_map[symbol]]
                self.work['source_steps'] += 2
                oa = A.output_map[ba if A.delay else sa]
                ob = B.output_map[bb if B.delay else sb]
                word = prefix + (symbol,)
                if oa != ob:
                    return word
                state = (na, nb, sa if A.delay else None, sb if B.delay else None)
                if state not in visited:
                    visited.add(state)
                    pending.append((state, word))
        return None

    def _weighted_separation(self, classes, word, weights):
        """Sum separated-pair weights by class histograms, O(N + C^2 P).



        ``pair_checks`` retains the represented candidate-pair count; the

        histogram arithmetic does not execute an individual comparison per pair.

        """
        counts = []
        for _, members in classes:
            tally = {}
            for identifier in members:
                p = self.predict(identifier, word)
                tally[p] = tally.get(p, 0) + 1
            counts.append((len(members), tally))
        total = 0
        for i in range(len(counts)):
            for j in range(i + 1, len(counts)):
                w = weights[i][j]
                if not w:
                    continue
                (na, ta), (nb, tb) = counts[i], counts[j]
                agree = sum(c * tb.get(p, 0) for p, c in ta.items())
                total += w * (na * nb - agree)
                self.work['pair_checks'] += na * nb
        return total

    def probe_is_safe(self, ids, word):
        """Check prefix hazard freedom under every live semantic interpretation.



        This is conditional on inclusion in the declared finite model family;

        raw response agreement alone does not certify the target's semantics.

        """
        ids = self._ids(ids)
        word = _word(word, self.machine.inputs)
        return bool(ids) and all(self.contract.hazard not in self.semantic(i, word) for i in ids)

    def information_profile(self, ids, word) -> dict:
        """Compute exact information-theoretic and minimax profile of a probe word over belief `ids`.



        Under deterministic candidate models and a uniform prior over `ids`, the

        Shannon predictive output entropy H(O | w, V) equals the Bayesian Active

        Learning by Disagreement (BALD) mutual information I(alpha; O | w, V).

        """
        ids = self._ids(ids)
        word = _word(word, self.machine.inputs)
        if not ids:
            return {
                'word': list(word), 'cost': 1 + len(word), 'candidate_count': 0,
                'partition_count': 0, 'mutual_information_bits': 0.0,
                'cost_efficiency_bits_per_call': 0.0, 'expected_survivors': 0.0,
                'worst_case_survivors': 0, 'expected_reduction_ratio': 0.0,
                'model_safe': False,
            }
        n = len(ids)
        partitions = {}
        for i in ids:
            partitions.setdefault(self.predict(i, word), []).append(i)
        counts = {p: len(m) for p, m in partitions.items()}
        h = -sum((c / n) * math.log2(c / n) for c in counts.values())
        expected_surv = sum((c * c) / n for c in counts.values())
        worst_surv = max(counts.values())
        cost = 1 + len(word)
        ready_mass = 0
        for members in partitions.values():
            common = set(self.safe_words(members[0]))
            for mid in members[1:]:
                if not common:
                    break
                common.intersection_update(self.safe_words(mid))
            if common:
                ready_mass += len(members)
        expected_readiness = round(ready_mass / n, 4)
        return {
            'word': list(word),
            'cost': cost,
            'candidate_count': n,
            'partition_count': len(counts),
            'mutual_information_bits': round(h, 6),
            'cost_efficiency_bits_per_call': round(h / cost, 6),
            'expected_survivors': round(expected_surv, 4),
            'worst_case_survivors': worst_surv,
            'expected_reduction_ratio': round(1.0 - (expected_surv / n), 6),
            'expected_task_readiness': expected_readiness,
            'model_safe': self.probe_is_safe(ids, word),
        }

    def rank_probes(self, ids, top_k=5, safe_only=False, max_steps=3) -> list[dict]:
        """Rank candidate probes by information-per-call efficiency, minimax survivor bound, and length."""
        if type(safe_only) is not bool:
            raise TypeError('safe_only must be a boolean')
        ids = self._ids(ids)
        _integer(top_k, 'top_k', 1)
        _integer(max_steps, 'max_steps', 1)
        if not ids:
            return []
        eligible = [w for w in self._probes if len(w) <= max_steps]
        if safe_only:
            eligible = [w for w in eligible if self.probe_is_safe(ids, w)]
        profiles = [self.information_profile(ids, w) for w in eligible]
        profiles.sort(key=lambda p: (
            -p['cost_efficiency_bits_per_call'],
            p['worst_case_survivors'],
            p['cost'],
            p['word'],
        ))
        return profiles[:top_k]

    def choose_branching_step(self, ids, word_prefix=(), safe_only=False, safe_ids=None, seed=0):
        """Select the next within-query action conditioned on the live belief and query prefix.



        Chooses an action maximizing predictive output entropy across surviving models.

        If safe_only=True, only actions where all surviving models in safe_ids predict zero hazard on

        this step are considered eligible.

        """
        import math
        ids = self._ids(ids)
        if not ids or len(ids) <= 1:
            return None
        if safe_ids is None:
            safe_ids = ids
        prefix = _word(word_prefix, self.machine.inputs)
        best_action, best_score = None, -1.0
        n = len(ids)
        order = [(seed + offset) % self.machine.inputs for offset in range(self.machine.inputs)]
        for a in order:
            extended = prefix + (a,)
            if safe_only and not self.probe_is_safe(safe_ids, extended):
                continue
            counts = {}
            for i in ids:
                p = self.predict(i, extended)[-1]
                counts[p] = counts.get(p, 0) + 1
            h = 0.0
            for c in counts.values():
                prob = c / n
                h -= prob * math.log2(prob)
            self.work['pair_checks'] += n
            if h > best_score:
                best_action, best_score = a, h
        if best_score <= 0.0:
            # When models agree on predicted outputs, fall back to the rotating
            # eligible action to test the hypothesis against target observations.
            for a in order:
                extended = prefix + (a,)
                if not safe_only or self.probe_is_safe(safe_ids, extended):
                    return a
            return None
        return best_action

    def choose_probe(self, ids, policy, remaining_steps, seen, seed=0, safe_only=False):
        if policy not in ('fixed', 'random', 'disagreement', 'task', 'counterexample', 'entropy', 'ec2', 'branching', 'voi'):
            raise ValueError('unknown probe policy')
        if type(safe_only) is not bool:
            raise TypeError('safe_only must be a boolean')
        ids = self._ids(ids)
        _integer(remaining_steps, 'remaining steps')
        _integer(seed, 'seed', maximum=2**32 - 1)
        seen = {_word(w, self.machine.inputs) for w in seen}
        if not ids:
            return None
        # remaining_steps is the action allowance after the caller has reserved
        # the reset for this query. Scores still charge reset + actions in their
        # information/cost ratio.
        eligible = [w for w in self._probes if w not in seen and len(w) <= remaining_steps]
        if safe_only:
            eligible = [w for w in eligible if self.probe_is_safe(ids, w)]
        if not eligible and policy != 'counterexample':
            return None
        if policy == 'fixed':
            return eligible[0]
        if policy == 'random':
            return Random(seed).choice(sorted(eligible))
        if policy == 'entropy':
            import math
            best, best_score = None, -1.0
            n = len(ids)
            if not ids:
                return None
            if n == 1:
                return eligible[0] if eligible else None
            for word in eligible:
                counts = {}
                for i in ids:
                    p = self.predict(i, word)
                    counts[p] = counts.get(p, 0) + 1
                h = 0.0
                for c in counts.values():
                    prob = c / n
                    h -= prob * math.log2(prob)
                self.work['pair_checks'] += n
                score = h / (1 + len(word))
                if score > best_score:
                    best, best_score = word, score
            return best
        if policy == 'ec2':
            # Equivalence Class Edge Cutting (EC2):
            # Prioritizes probes that cut edges between differing contract decision classes.
            classes = self._decision_classes(ids)
            if len(classes) <= 1:
                return min(eligible, key=lambda w: (len(w), w)) if eligible else None
            weights = [[len(sa ^ sb) / max(1, len(sa | sb)) for sb, _ in classes] for sa, _ in classes]
            best, best_score = None, -1.0
            for word in eligible:
                score = self._weighted_separation(classes, word, weights) / (1 + len(word))
                if score > best_score:
                    best, best_score = word, score
            if best_score <= 0.0:
                return min(eligible, key=lambda w: (len(w), w)) if eligible else None
            return best
        if policy == 'branching':
            step_action = self.choose_branching_step(ids, (), safe_only=safe_only)
            if step_action is not None and (step_action,) in eligible:
                return (step_action,)
            return eligible[0] if eligible else None
        if policy == 'voi':
            import math
            n = len(ids)
            if not ids:
                return None
            if n == 1:
                return eligible[0] if eligible else None
            best, best_score = None, (-1.0, -1.0, 0, 0)
            for word in eligible:
                partitions = {}
                for i in ids:
                    p = self.predict(i, word)
                    partitions.setdefault(p, []).append(i)
                cost = 1 + len(word)
                ready_mass = 0
                h = 0.0
                worst_cell = 0
                for p, members in partitions.items():
                    m_len = len(members)
                    if m_len > worst_cell:
                        worst_cell = m_len
                    prob = m_len / n
                    h -= prob * math.log2(prob)
                    common_safe = set(self.safe_words(members[0]))
                    for mid in members[1:]:
                        if not common_safe:
                            break
                        common_safe.intersection_update(self.safe_words(mid))
                    if common_safe:
                        ready_mass += m_len
                readiness_rate = ready_mass / n
                readiness_efficiency = readiness_rate / cost
                entropy_efficiency = h / cost
                score = (readiness_efficiency, entropy_efficiency, -worst_cell, -len(word))
                self.work['pair_checks'] += n
                if score > best_score:
                    best, best_score = word, score
            return best if best is not None else (eligible[0] if eligible else None)
        if policy == 'disagreement':
            best, best_score = None, 0
            n = len(ids)
            total_pairs = n * (n - 1) // 2
            for word in eligible:
                counts = {}
                for i in ids:
                    p = self.predict(i, word)
                    counts[p] = counts.get(p, 0) + 1
                agreeing = sum(c * (c - 1) // 2 for c in counts.values())
                separated = total_pairs - agreeing
                self.work['pair_checks'] += total_pairs
                score = separated / (1 + len(word))
                if score > best_score:
                    best, best_score = word, score
            return best
        if policy == 'counterexample':
            # Decision-focused counterexample-guided probing. Candidate words are
            # the fixed pool plus exact shortest witnesses separating one
            # representative pair for every pair of *decision classes* (survivors
            # whose admissible deployment sets differ). Witnesses may exceed the
            # three-action pool. Pairs are weighted as in the task policy
            # (Jaccard distance of safe sets), computed per class, not per pair.
            classes = self._decision_classes(ids)
            if len(classes) > 1:
                weights = [[len(sa ^ sb) / max(1, len(sa | sb)) for sb, _ in classes] for sa, _ in classes]
            else:
                # No decision-relevant disagreement: fall back to plain separation.
                classes = [(frozenset(), [i]) for i in ids]
                weights = [[0 if i == j else 1 for j in range(len(classes))] for i in range(len(classes))]
                if len(classes) > 64:
                    classes = classes[:64]
                    weights = [row[:64] for row in weights[:64]]
            pool = list(eligible)
            if len(classes) > 1:
                limit = min(remaining_steps, self.contract.horizon + 4)
                for i in range(len(classes)):
                    for j in range(i + 1, len(classes)):
                        if not weights[i][j]:
                            continue
                        witness = self.distinguishing_word(classes[i][1][0], classes[j][1][0], limit)
                        if (witness is not None and witness not in seen and witness not in pool
                                and (not safe_only or self.probe_is_safe(ids, witness))):
                            pool.append(witness)
            best, best_score = None, 0.0
            for word in pool:
                score = self._weighted_separation(classes, word, weights) / (1 + len(word))
                if score > best_score:
                    best, best_score = word, score
            return best
        # Task weights are equal within each pair of contract classes. Count
        # separated candidate pairs by response histograms, without allocating
        # the quadratic pair list. Rational scores give deterministic ties.
        classes = self._decision_classes(ids)
        weights = [[Fraction(len(sa ^ sb), max(1, len(sa | sb))) for sb, _ in classes]
                   for sa, _ in classes]
        best, best_score = None, Fraction(0)
        for word in eligible:
            score = self._weighted_separation(classes, word, weights) / (1 + len(word))
            if score > best_score:
                best, best_score = word, score
        return best

    def plan(self, ids, mode='robust', transcripts=None, anchors=None, risk_tolerance=0.0):
        if mode not in ('robust', 'one_best', 'unique', 'fit', 'identity', 'abstain', 'risk_controlled'):
            raise ValueError('unknown admission mode')
        # The bounds reject NaN/infinity as well as unsupported numeric aliases;
        # validating before any empty-belief return keeps the public API strict.
        if type(risk_tolerance) not in (int, float) or not 0 <= risk_tolerance < 1:
            raise ValueError('risk_tolerance must be a finite number in [0, 1)')
        ids = self._ids(ids)
        if mode == 'risk_controlled':
            if not ids:
                return {'status': 'FAMILY_CONTRADICTED', 'admitted': False, 'word': None,
                        'candidate_count': 0, 'risk_metrics': {'hazard_risk': None, 'goal_rate': None, 'tolerance': risk_tolerance}}
            n = len(ids)
            admissible_candidates = []
            for w in self._deployment:
                h_count = sum(self.contract.hazard in self.semantic(i, w) for i in ids)
                g_count = sum(self.contract.accepts(self.semantic(i, w)) for i in ids)
                h_risk = h_count / n
                g_rate = g_count / n
                if h_risk <= risk_tolerance and (n - g_count) / n <= risk_tolerance:
                    admissible_candidates.append((w, h_risk, g_rate))
            if not admissible_candidates:
                return {'status': 'UNRESOLVED', 'admitted': False, 'word': None,
                        'candidate_count': n, 'risk_metrics': {'hazard_risk': None, 'goal_rate': None, 'tolerance': risk_tolerance}}
            best_w, best_h_risk, best_g_rate = min(
                admissible_candidates,
                key=lambda item: (len(item[0]), item[1], -item[2], item[0])
            )
            status = 'IDENTIFIED' if n == 1 else ('TASK_ADMISSIBLE' if best_h_risk == 0.0 and best_g_rate == 1.0 else 'RISK_CONTROLLED_ADMISSIBLE')
            return {
                'status': status,
                'admitted': True,
                'word': list(best_w),
                'candidate_count': n,
                'risk_metrics': {
                    'hazard_risk': best_h_risk,
                    'goal_rate': best_g_rate,
                    'tolerance': risk_tolerance,
                    'candidate_count': n
                }
            }
        chosen = ids
        if mode == 'fit':
            losses = self.losses(transcripts or [], anchors)
            chosen = (min(losses, key=lambda i: (losses[i], i)),)
        elif mode == 'identity':
            chosen = (next(a.id for a in self.adapters if a.input_map == tuple(range(self.machine.inputs))
                           and a.output_map == tuple(range(self.machine.outputs))),)
        elif mode == 'one_best':
            chosen = ids[:1]
        elif mode == 'unique' and len(ids) != 1:
            chosen = ()
        elif mode == 'abstain':
            chosen = ()
        safe = set(self.safe_words(chosen[0])) if chosen else set()
        for identifier in chosen[1:]:
            safe.intersection_update(self.safe_words(identifier))
        word = min(safe, key=lambda w: (len(w), w)) if safe else None
        status = ('IDENTIFIED' if len(ids) == 1 else 'TASK_ADMISSIBLE') if word is not None else (
            'FAMILY_CONTRADICTED' if not ids else 'UNRESOLVED')
        return {'status': status, 'admitted': word is not None, 'word': list(word) if word is not None else None,
                'candidate_count': len(ids)}

    def _signature(self, identifier):
        """Canonical minimized reachable raw transducer, independent of state IDs."""
        if identifier in self._signatures:
            return self._signatures[identifier]
        adapter = self._adapter(identifier)
        # Delayed observations depend on both the source state and the previous
        # semantic output. Minimize that augmented transducer, so raw-equivalent
        # static/delayed models can merge while distinct buffers remain separate.
        initial = (self.machine.initial, adapter.init_output if adapter.delay else None)
        states, pending, rows = {initial}, deque([initial]), {}
        while pending:
            state = pending.popleft()
            source_state, buffered = state
            row = []
            for symbol in adapter.input_map:
                source_next, output = self.machine.transitions[source_state][symbol]
                self.work['source_steps'] += 1
                following = (source_next, output if adapter.delay else None)
                row.append((following, adapter.output_map[buffered if adapter.delay else output]))
                if following not in states:
                    states.add(following)
                    pending.append(following)
            rows[state] = tuple(row)
        classes = {state: 0 for state in states}
        while True:
            labels, refined = {}, {}
            for state in sorted(states):
                signature = tuple((output, classes[following]) for following, output in rows[state])
                refined[state] = labels.setdefault(signature, len(labels))
            # Refinement never merges old classes; unchanged count means stable.
            if len(set(refined.values())) == len(set(classes.values())):
                classes = refined
                break
            classes = refined
        representatives = {classes[s]: s for s in sorted(states)}
        start = classes[initial]
        canonical, pending, result = {start: 0}, deque([start]), []
        while pending:
            current = pending.popleft()
            row = []
            for following, output in rows[representatives[current]]:
                following = classes[following]
                if following not in canonical:
                    canonical[following] = len(canonical)
                    pending.append(following)
                row.append((canonical[following], output))
            result.append(tuple(row))
        self._signatures[identifier] = tuple(result)
        return self._signatures[identifier]

    def ambiguity(self, ids):
        ids = self._ids(ids)
        return {'map_count': len(ids), 'observational_classes': len({self._signature(i) for i in ids}),
                'contract_classes': len({self.safe_words(i) for i in ids})}


def compare_adapter(machine, adapter, target):
    """Exact product BFS; first mismatch gives a shortest lexicographic witness."""
    if machine.inputs != target.inputs or machine.outputs != target.outputs:
        raise ValueError('comparison machine alphabets differ')
    if len(adapter.input_map) != machine.inputs or len(adapter.output_map) != machine.outputs:
        raise ValueError('adapter alphabets differ')
    is_temporal = bool(getattr(adapter, 'delay', 0))
    initial = (machine.initial, target.initial, adapter.init_output) if is_temporal else (machine.initial, target.initial)
    visited, pending = {initial}, deque([(initial, ())])
    steps = 0
    while pending:
        curr, prefix = pending.popleft()
        source_state, target_state = curr[0], curr[1]
        buffered = curr[2] if is_temporal else None
        for symbol in range(machine.inputs):
            source_next, semantic = machine.transitions[source_state][adapter.input_map[symbol]]
            target_next, raw = target.transitions[target_state][symbol]
            steps += 1
            word = prefix + (symbol,)
            pred_raw = adapter.output_map[buffered if is_temporal else semantic]
            if pred_raw != raw:
                return {'equivalent': False, 'counterexample': list(word), 'visited_pairs': len(visited),
                        'source_steps': steps, 'target_model_steps': steps}
            following = (source_next, target_next, semantic) if is_temporal else (source_next, target_next)
            if following not in visited:
                visited.add(following)
                pending.append((following, word))
    return {'equivalent': True, 'counterexample': None, 'visited_pairs': len(visited),
            'source_steps': steps, 'target_model_steps': steps}



class _LazyAdapters:
    def __init__(self, space):
        self._space = space

    def __len__(self):
        return self._space.total_adapters

    def __getitem__(self, idx):
        if isinstance(idx, slice):
            return [self[i] for i in range(*idx.indices(len(self)))]
        return self._space.adapter(idx)

    def __iter__(self):
        for i in range(len(self)):
            yield self._space.adapter(i)


class FactoredCandidateSpace(CandidateSpace):
    """Factored permutation space decomposing candidate adapters into orthogonal input/output factors.



    Instead of enumerating N_I * N_O * (1 + N_O) adapters in flat memory, decomposes

    the hypothesis space into P_I x P_O (and temporal initial delay buffers).

    Semantic evaluation depends strictly on P_I; anchors prune P_O independently;

    and combined survivor filtering scales beyond 17,280+ models with complete survivor semantics.

    """

    def __init__(self, machine: Machine, contract: Contract, temporal: bool = False):
        if not isinstance(machine, Machine) or not isinstance(contract, Contract):
            raise TypeError('FactoredCandidateSpace requires a Machine and Contract')
        if max(contract.goal, contract.hazard) >= machine.outputs:
            raise ValueError('contract symbols exceed source alphabet')
        self.machine, self.contract = machine, contract
        self.temporal = bool(temporal)

        self.input_perms = tuple(permutations(range(machine.inputs)))
        self.output_perms = tuple(permutations(range(machine.outputs)))
        self.num_input_perms = len(self.input_perms)
        self.num_output_perms = len(self.output_perms)
        self.static_count = self.num_input_perms * self.num_output_perms

        if self.temporal:
            self.temporal_count = self.static_count * machine.outputs
            self.total_adapters = self.static_count + self.temporal_count
        else:
            self.temporal_count = 0
            self.total_adapters = self.static_count

        self.work = {'source_steps': 0, 'candidate_expansions': 0, 'pair_checks': 0}
        self._source = {(): (machine.initial, ())}
        self._input_traces = {}
        self._safe_input_factors = {}
        self._adapter_cache = {}
        self._predictions = {}
        self._robust_plans = {}
        self._signatures = {}
        self._pair_weights = {}
        self._deployment_words = None
        self._probes = words(machine.inputs, 3)
        self._adapters_proxy = _LazyAdapters(self)

    @property
    def adapters(self):
        return self._adapters_proxy

    @property
    def _deployment(self):
        """Compatibility word bank, materialized only for exhaustive consumers."""
        if self._deployment_words is None:
            self._deployment_words = words(self.machine.inputs, self.contract.horizon)
        return self._deployment_words

    def decompose_id(self, identifier: int) -> tuple[int, int, int, int]:
        """Convert linear adapter id into (input_perm_idx, output_perm_idx, delay, init_output)."""
        _integer(identifier, 'adapter id', maximum=self.total_adapters - 1)
        if identifier < self.static_count:
            i_idx = identifier // self.num_output_perms
            o_idx = identifier % self.num_output_perms
            return i_idx, o_idx, 0, 0
        else:
            rem = identifier - self.static_count
            init_output = rem % self.machine.outputs
            pair_idx = rem // self.machine.outputs
            i_idx = pair_idx // self.num_output_perms
            o_idx = pair_idx % self.num_output_perms
            return i_idx, o_idx, 1, init_output

    def compose_id(self, i_idx: int, o_idx: int, delay: int = 0, init_output: int = 0) -> int:
        """Compose (input_perm_idx, output_perm_idx, delay, init_output) into canonical linear id."""
        _integer(i_idx, 'input factor', maximum=self.num_input_perms - 1)
        _integer(o_idx, 'output factor', maximum=self.num_output_perms - 1)
        _integer(delay, 'delay', maximum=1)
        _integer(init_output, 'init_output', maximum=self.machine.outputs - 1)
        if delay == 0:
            if init_output:
                raise ValueError('static adapter init_output must be zero')
            return i_idx * self.num_output_perms + o_idx
        if not self.temporal:
            raise ValueError('temporal candidates are not enabled')
        pair_idx = i_idx * self.num_output_perms + o_idx
        return self.static_count + pair_idx * self.machine.outputs + init_output

    def adapter(self, identifier: int) -> Adapter:
        _integer(identifier, 'adapter id', maximum=self.total_adapters - 1)
        if identifier not in self._adapter_cache:
            i_idx, o_idx, delay, init_output = self.decompose_id(identifier)
            self._adapter_cache[identifier] = Adapter(
                id=identifier,
                input_map=self.input_perms[i_idx],
                output_map=self.output_perms[o_idx],
                delay=delay,
                init_output=init_output
            )
        return self._adapter_cache[identifier]

    def _adapter(self, identifier):
        return self.adapter(identifier)

    def _ids(self, ids):
        ids = tuple(ids)
        for identifier in ids:
            _integer(identifier, 'adapter id', maximum=self.total_adapters - 1)
        return tuple(sorted(set(ids)))

    def level_ids(self, level):
        if level == 'static':
            return tuple(range(self.static_count))
        if level == 'temporal':
            return tuple(range(self.total_adapters))
        raise ValueError('unknown hypothesis level')

    def _input_trace(self, i_idx: int, word: tuple[int, ...]) -> tuple[int, ...]:
        key = (i_idx, word)
        if key not in self._input_traces:
            input_map = self.input_perms[i_idx]
            mapped = tuple(input_map[x] for x in word)
            self._input_traces[key] = self._source_trace(mapped)
        return self._input_traces[key]

    def _safe_for_input_factor(self, i_idx: int) -> frozenset:
        if i_idx not in self._safe_input_factors:
            self._safe_input_factors[i_idx] = frozenset(
                w for w in self._deployment if self.contract.accepts(self._input_trace(i_idx, w))
            )
        return self._safe_input_factors[i_idx]

    def semantic(self, identifier: int, word: tuple[int, ...]) -> tuple[int, ...]:
        i_idx, _, _, _ = self.decompose_id(identifier)
        word = _word(word, self.machine.inputs)
        return self._input_trace(i_idx, word)

    def predict(self, identifier: int, word: tuple[int, ...]) -> tuple[int, ...]:
        i_idx, o_idx, delay, init_o = self.decompose_id(identifier)
        word = _word(word, self.machine.inputs)
        key = (identifier, word)
        if key in self._predictions:
            return self._predictions[key]
        semantic = self._input_trace(i_idx, word)
        out_map = self.output_perms[o_idx]
        self.work['candidate_expansions'] += 1
        if delay == 0:
            result = tuple(out_map[s] for s in semantic)
        else:
            result = tuple(out_map[init_o if t == 0 else semantic[t - 1]] for t in range(len(semantic)))
        self._predictions[key] = result
        return result

    def safe_words(self, identifier: int) -> frozenset:
        i_idx, _, _, _ = self.decompose_id(identifier)
        return self._safe_for_input_factor(i_idx)

    def probe_is_safe(self, ids, word):
        """Factored prefix hazard check over unique active input factors O(|I_surv|)."""
        ids = self._ids(ids)
        word = _word(word, self.machine.inputs)
        if not ids:
            return False
        active_i = {self.decompose_id(i)[0] for i in ids}
        return all(self.contract.hazard not in self._input_trace(i_idx, word) for i_idx in active_i)

    def survivors(self, transcripts, anchors=None) -> tuple[int, ...]:
        evidence, anchors = self._evidence(transcripts, anchors)

        valid_o_indices = []
        for o_idx, o_perm in enumerate(self.output_perms):
            if all(o_perm[sem] == raw for raw, sem in anchors.items()):
                valid_o_indices.append(o_idx)

        if not valid_o_indices:
            return ()

        surviving_ids = []

        # Check static adapters
        for i_idx in range(self.num_input_perms):
            sem_traces = [self._input_trace(i_idx, word) for word, _ in evidence]
            for o_idx in valid_o_indices:
                o_perm = self.output_perms[o_idx]
                match = True
                for sem, (_, outputs) in zip(sem_traces, evidence):
                    if any(o_perm[s] != out for s, out in zip(sem, outputs)):
                        match = False
                        break
                if match:
                    surviving_ids.append(self.compose_id(i_idx, o_idx, delay=0))

        # Check temporal adapters if enabled
        if self.temporal:
            for i_idx in range(self.num_input_perms):
                sem_traces = [self._input_trace(i_idx, word) for word, _ in evidence]
                for o_idx in valid_o_indices:
                    o_perm = self.output_perms[o_idx]
                    for init_o in range(self.machine.outputs):
                        match = True
                        for sem, (_, outputs) in zip(sem_traces, evidence):
                            for t, out in enumerate(outputs):
                                pred_s = init_o if t == 0 else sem[t - 1]
                                if o_perm[pred_s] != out:
                                    match = False
                                    break
                            if not match:
                                break
                        if match:
                            surviving_ids.append(self.compose_id(i_idx, o_idx, delay=1, init_output=init_o))

        return tuple(sorted(surviving_ids))

    def plan(self, ids, mode='robust', transcripts=None, anchors=None, risk_tolerance=0.0, *, admission=None,

             node_limit=20000, cancel=None):
        if admission is not None:
            mode = admission
        if mode not in ('robust', 'one_best', 'unique', 'fit', 'identity', 'abstain', 'risk_controlled'):
            raise ValueError('unknown admission mode')
        if type(risk_tolerance) not in (int, float) or not 0 <= risk_tolerance < 1:
            raise ValueError('risk_tolerance must be a finite number in [0, 1)')
        _integer(node_limit, 'node_limit', 1)
        self._check_plan_cancel(cancel)
        ids = self._ids(ids)
        if mode == 'risk_controlled':
            if not ids:
                return {'status': 'FAMILY_CONTRADICTED', 'admitted': False, 'word': None,
                        'candidate_count': 0, 'risk_metrics': {'hazard_risk': None, 'goal_rate': None, 'tolerance': risk_tolerance}}
            n = len(ids)
            i_counts = Counter(self.decompose_id(i)[0] for i in ids)
            admissible_candidates = []
            for w in self._deployment:
                h_count = sum(count for i_idx, count in i_counts.items()
                              if self.contract.hazard in self._input_trace(i_idx, w))
                g_count = sum(count for i_idx, count in i_counts.items()
                              if w in self._safe_for_input_factor(i_idx))
                h_risk = h_count / n
                g_rate = g_count / n
                if h_risk <= risk_tolerance and (n - g_count) / n <= risk_tolerance:
                    admissible_candidates.append((w, h_risk, g_rate))
            if not admissible_candidates:
                return {'status': 'UNRESOLVED', 'admitted': False, 'word': None,
                        'candidate_count': n, 'risk_metrics': {'hazard_risk': None, 'goal_rate': None, 'tolerance': risk_tolerance}}
            best_w, best_h_risk, best_g_rate = min(
                admissible_candidates,
                key=lambda item: (len(item[0]), item[1], -item[2], item[0])
            )
            status = 'IDENTIFIED' if n == 1 else ('TASK_ADMISSIBLE' if best_h_risk == 0.0 and best_g_rate == 1.0 else 'RISK_CONTROLLED_ADMISSIBLE')
            return {
                'status': status,
                'admitted': True,
                'word': list(best_w),
                'candidate_count': n,
                'risk_metrics': {
                    'hazard_risk': best_h_risk,
                    'goal_rate': best_g_rate,
                    'tolerance': risk_tolerance,
                    'candidate_count': n
                }
            }
        if mode == 'robust':
            if not ids:
                return {'status': 'FAMILY_CONTRADICTED', 'admitted': False, 'word': None, 'candidate_count': 0}
            active_i = tuple(sorted({self.decompose_id(i)[0] for i in ids}))
            chosen, complete, expanded = self._robust_word(active_i, node_limit, cancel)
            if not complete:
                return {'status': 'SEARCH_LIMIT', 'admitted': False, 'word': None, 'candidate_count': len(ids),
                        'search': {'complete': False, 'expanded_nodes': expanded, 'node_limit': node_limit,
                                   'horizon': self.contract.horizon, 'strategy': 'input_factor_product_bfs'}}
            status = ('IDENTIFIED' if len(ids) == 1 else 'TASK_ADMISSIBLE') if chosen is not None else 'UNRESOLVED'
            return {'status': status, 'admitted': chosen is not None, 'word': list(chosen) if chosen is not None else None,
                    'candidate_count': len(ids)}
        if mode == 'identity':
            safe = set(self.safe_words(0))
            word = min(safe, key=lambda w: (len(w), w)) if safe else None
            status = ('IDENTIFIED' if len(ids) == 1 else 'TASK_ADMISSIBLE') if word is not None else (
                'FAMILY_CONTRADICTED' if not ids else 'UNRESOLVED')
            return {'status': status, 'admitted': word is not None, 'word': list(word) if word is not None else None,
                    'candidate_count': len(ids)}
        return super().plan(ids, mode=mode, transcripts=transcripts, anchors=anchors, risk_tolerance=risk_tolerance)

    @staticmethod
    def _check_plan_cancel(cancel):
        if cancel and cancel():
            from .study import Cancelled
            raise Cancelled('Factored planning cancelled')

    def _robust_word(self, active_i, node_limit, cancel):
        """Search safe joint input-factor states in shortest/lexicographic order."""
        if active_i in self._robust_plans:
            return self._robust_plans[active_i], True, 0
        input_maps = tuple(self.input_perms[i] for i in active_i)
        start = (self.machine.initial,) * len(input_maps)
        pending, visited = deque([(start, ())]), {start}
        expanded = 0
        while pending:
            self._check_plan_cancel(cancel)
            if expanded >= node_limit:
                return None, False, expanded
            states, prefix = pending.popleft()
            expanded += 1
            for action in range(self.machine.inputs):
                self._check_plan_cancel(cancel)
                next_states, goal = [], True
                for state, input_map in zip(states, input_maps):
                    following, output = self.machine.transitions[state][input_map[action]]
                    self.work['source_steps'] += 1
                    if output == self.contract.hazard:
                        break
                    next_states.append(following)
                    goal = goal and output == self.contract.goal
                else:
                    word = prefix + (action,)
                    # A goal edge can return to an already visited state.
                    if goal:
                        self._robust_plans[active_i] = word
                        return word, True, expanded
                    following = tuple(next_states)
                    if len(word) < self.contract.horizon and following not in visited:
                        visited.add(following)
                        pending.append((following, word))
        self._robust_plans[active_i] = None
        return None, True, expanded

    def factor_summary(self, survivors=None) -> dict:
        """Report structural factor counts; the ratio is not measured memory compression."""
        if survivors is None:
            survivors = tuple(range(self.total_adapters))
        else:
            survivors = self._ids(survivors)
        active_i = set()
        active_o = set()
        active_delay = set()
        for sid in survivors:
            i_idx, o_idx, delay, _ = self.decompose_id(sid)
            active_i.add(i_idx)
            active_o.add(o_idx)
            active_delay.add(delay)
        return {
            'total_models': self.total_adapters,
            'static_models': self.static_count,
            'temporal_models': self.temporal_count,
            'input_alphabet': self.machine.inputs,
            'output_alphabet': self.machine.outputs,
            'input_permutations': self.num_input_perms,
            'output_permutations': self.num_output_perms,
            'survivors_count': len(survivors),
            'active_input_factors': len(active_i),
            'active_output_factors': len(active_o),
            'active_delay_modes': sorted(active_delay),
            'compression_ratio': round(self.total_adapters / max(1, (self.num_input_perms + self.num_output_perms)), 2),
            'compression_ratio_scope': 'Structural candidate/factor ratio; not measured memory compression.',
        }