| """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:
|
|
|
|
|
| _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):
|
|
|
| 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:
|
|
|
|
|
| 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
|
|
|
|
|
|
|
| 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':
|
|
|
|
|
| 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':
|
|
|
|
|
|
|
|
|
|
|
|
|
| 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:
|
|
|
| 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
|
|
|
|
|
|
|
| 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')
|
|
|
|
|
| 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)
|
|
|
|
|
|
|
| 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))
|
|
|
| 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 = []
|
|
|
|
|
| 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))
|
|
|
|
|
| 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,)
|
|
|
| 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.',
|
| }
|
|
|
