File size: 8,531 Bytes
fc3e70a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | # This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
"""Phase 3: Decoder FSM — explicit state machine with typed transitions.
Every transition is explicit. Every invalid transition fails closed.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Callable, Optional
from .types import FSMState, FSMTransition, PipelineError
# ── Transition table ──────────────────────────────────────────────────────────
TRANSITION_TABLE: list[FSMTransition] = [
FSMTransition(FSMState.INPUT, "raw_input", "normalize", ("text_not_empty",), FSMState.NORMALIZE, "input preserved"),
FSMTransition(FSMState.NORMALIZE, "normalized_input", "lex", ("tokens_non_empty",), FSMState.LEXICAL_ANALYSIS, "normalization idempotent"),
FSMTransition(FSMState.LEXICAL_ANALYSIS, "token_stream", "detect_intent", ("tokens_recognized",), FSMState.INTENT_DETECTION, "lexical coverage"),
FSMTransition(FSMState.INTENT_DETECTION, "intent_class", "bind_entities", ("intent_not_unknown",), FSMState.ENTITY_BINDING, "intent classified"),
FSMTransition(FSMState.ENTITY_BINDING, "entities", "bind_semantics", ("entities_typed",), FSMState.SEMANTIC_BINDING, "entities resolved"),
FSMTransition(FSMState.SEMANTIC_BINDING, "semantic_candidate","lock_temporal", ("candidate_valid",), FSMState.TEMPORAL_LOCK, "semantic candidate valid"),
FSMTransition(FSMState.TEMPORAL_LOCK, "temporal_state", "bind_geometry", ("tau_monotonic",), FSMState.GEOMETRY_BINDING, "tau assigned"),
FSMTransition(FSMState.GEOMETRY_BINDING, "geometry", "build_matrix", ("geometry_consistent",), FSMState.MATRIX_FORM, "dimensions compatible"),
FSMTransition(FSMState.MATRIX_FORM, "semantic_matrix", "resolve_boolean", ("matrix_well_formed",), FSMState.BOOLEAN_ROOT, "matrix non-degenerate"),
FSMTransition(FSMState.BOOLEAN_ROOT, "boolean_expr", "normalize_boolean", ("boolean_valid",), FSMState.BOOLEAN_NORMALIZATION, "boolean well-formed"),
FSMTransition(FSMState.BOOLEAN_NORMALIZATION,"normalized_boolean","tokenize", ("normal_form_stable",), FSMState.TOKENIZATION, "normalization idempotent"),
FSMTransition(FSMState.TOKENIZATION, "token_stream", "canonicalize_latin", ("tokens_round_trip",), FSMState.LATIN_CANONICALIZATION, "encode/decode invariant"),
FSMTransition(FSMState.LATIN_CANONICALIZATION,"latin_form", "lex_latin", ("vocabulary_closed",), FSMState.LATIN_LEXING, "vocabulary complete"),
FSMTransition(FSMState.LATIN_LEXING, "latin_tokens", "map_rosetta", ("latin_tokens_valid",), FSMState.ROSETTA_MAPPING, "all tokens mapped"),
FSMTransition(FSMState.ROSETTA_MAPPING, "rosetta_mapping", "build_subleq", ("mapping_complete",), FSMState.SUBLEQ_IR, "all operations mapped"),
FSMTransition(FSMState.SUBLEQ_IR, "subleq_program", "verify", ("subleq_valid",), FSMState.VERIFICATION, "program halts"),
FSMTransition(FSMState.VERIFICATION, "verified_ir", "select_artifact", ("all_checks_pass",), FSMState.ARTIFACT_SELECTION, "verification passed"),
FSMTransition(FSMState.ARTIFACT_SELECTION, "artifact_type", "emit", ("template_bound",), FSMState.EMISSION, "template resolved"),
FSMTransition(FSMState.EMISSION, "artifact", "complete", ("artifact_provenance_ok",), FSMState.COMPLETE, "artifact complete"),
FSMTransition(FSMState.COMPLETE, "done", "noop", (), FSMState.COMPLETE, "terminal"),
# Error transitions — any state can go to ERROR
FSMTransition(FSMState.INPUT, "invalid", "fail", ("error_logged",), FSMState.ERROR, "fail closed"),
]
# Build lookup: (source_state, input_class) → transition
_TRANSITION_MAP: dict[tuple[FSMState, str], FSMTransition] = {
(t.source_state, t.input_class): t for t in TRANSITION_TABLE
}
# ── FSM context ───────────────────────────────────────────────────────────────
@dataclass
class FSMContext:
state: FSMState = FSMState.INPUT
history: list[tuple[FSMState, str, FSMState]] = field(default_factory=list)
data: dict[str, object] = field(default_factory=dict)
error: Optional[str] = None
def transition(self, input_class: str) -> FSMTransition:
key = (self.state, input_class)
t = _TRANSITION_MAP.get(key)
if t is None:
# Also try any-state error transition
err = _TRANSITION_MAP.get((self.state, "invalid"))
if err:
self.history.append((self.state, input_class, FSMState.ERROR))
self.state = FSMState.ERROR
self.error = f"No transition from {self.state.name} on input '{input_class}'"
raise PipelineError(self.state, f"No valid transition on '{input_class}'")
raise PipelineError(
self.state,
f"No transition from {self.state.name} on '{input_class}' — fail closed",
)
self.history.append((self.state, input_class, t.destination_state))
self.state = t.destination_state
return t
def set(self, key: str, value: object) -> None:
self.data[key] = value
def get(self, key: str) -> object:
return self.data.get(key)
def is_terminal(self) -> bool:
return self.state in (FSMState.COMPLETE, FSMState.ERROR)
def dump_history(self) -> list[str]:
return [f"{src.name} --[{inp}]--> {dst.name}" for src, inp, dst in self.history]
# ── Constraint validators ─────────────────────────────────────────────────────
ConstraintValidator = Callable[[FSMContext], bool]
_VALIDATORS: dict[str, ConstraintValidator] = {
"text_not_empty": lambda ctx: bool(str(ctx.get("raw_text") or "").strip()),
"tokens_non_empty": lambda ctx: bool(ctx.get("tokens")),
"tokens_recognized": lambda ctx: True, # lexer failure raises
"intent_not_unknown": lambda ctx: ctx.get("intent") != "unknown",
"entities_typed": lambda ctx: True,
"candidate_valid": lambda ctx: ctx.get("candidate") is not None,
"tau_monotonic": lambda ctx: True, # TemporalLock.record enforces this
"geometry_consistent": lambda ctx: True, # GeometryError raised on violation
"matrix_well_formed": lambda ctx: True,
"boolean_valid": lambda ctx: True,
"normal_form_stable": lambda ctx: True,
"tokens_round_trip": lambda ctx: True,
"vocabulary_closed": lambda ctx: True,
"latin_tokens_valid": lambda ctx: True,
"mapping_complete": lambda ctx: True,
"subleq_valid": lambda ctx: True,
"all_checks_pass": lambda ctx: ctx.get("verified") is True,
"template_bound": lambda ctx: ctx.get("template") is not None,
"artifact_provenance_ok": lambda ctx: ctx.get("artifact") is not None,
"error_logged": lambda ctx: True,
}
def validate_constraints(transition: FSMTransition, ctx: FSMContext) -> None:
for c in transition.constraints:
validator = _VALIDATORS.get(c)
if validator and not validator(ctx):
raise PipelineError(
ctx.state,
f"Invariant '{c}' violated on transition {transition.source_state.name} "
f"→ {transition.destination_state.name}",
)
|