src/semantic_engine/fsm.py

# 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}",
            )