Deterministic Fallback State Machines in Python for Route Degradation

This page solves one specific production failure: when telemetry drops or the solver times out, dispatch code degrades service “somehow” through scattered if branches, and no two incidents degrade the same way or leave the same trail.

When a refuse fleet loses live position or a re-optimization run fails to converge, the routing service must fall back — but a fallback that is decided by ad-hoc conditionals is neither reproducible nor auditable. The fix is to model degradation as an explicit deterministic finite state machine with four declared states — NORMAL, DEGRADED, LAST_KNOWN_GOOD, and MANUAL — and a single transition table that is the only place a state change can originate. Every hop is logged as a state-transition record, so an incident review can replay the exact event sequence and land on the exact same state. This is the decision core of the Fallback Routing Logic subsystem within the Core Architecture & Compliance Mapping framework.

Environment & Data Prerequisites

The state machine is pure standard library — enum, dataclasses, json, logging, and datetime on Python 3.10+. There is no solver dependency here: this layer only decides which route strategy is authoritative. The single third-party dependency is pytest, used to prove determinism:

python -m pip install pytest==8.2.0

Transitions are driven by events raised elsewhere in the stack — a telemetry-gap detector, the solver’s own timeout, or an operator action. Each event is one flat trigger record:

One rule matters before any code: the detail field never influences a transition. Only the (state, event) pair decides the next state, because a machine that peeks at free-text notes is no longer deterministic — it is the very ad-hoc branching this design exists to abolish.

Step-by-Step Implementation

Step 1 — Declare the states and events as enums

Both the states and the events are closed sets. Modeling them as Enum members — not strings — means a typo like "telemetery_gap" fails at construction instead of silently falling through to a default branch.

import json
import logging
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from enum import Enum
from typing import Optional


class State(Enum):
    NORMAL = "NORMAL"
    DEGRADED = "DEGRADED"
    LAST_KNOWN_GOOD = "LAST_KNOWN_GOOD"
    MANUAL = "MANUAL"


class Event(Enum):
    TELEMETRY_GAP = "telemetry_gap"
    SOLVER_FAIL = "solver_fail"
    OPERATOR_OVERRIDE = "operator_override"
    RECOVERED = "recovered"

Step 2 — The transition table is the single source of truth

Every legal (state, event) -> state edge lives in one immutable mapping. Because it is a plain dict keyed by (State, Event) tuples, there is no ordering ambiguity: a lookup either hits a declared edge or it does not. There is no wildcard and no default — an omission is a deliberate “this cannot happen,” not an oversight.

TRANSITIONS: dict[tuple[State, Event], State] = {
    (State.NORMAL, Event.TELEMETRY_GAP): State.DEGRADED,
    (State.NORMAL, Event.SOLVER_FAIL): State.DEGRADED,
    (State.DEGRADED, Event.SOLVER_FAIL): State.LAST_KNOWN_GOOD,
    (State.DEGRADED, Event.TELEMETRY_GAP): State.LAST_KNOWN_GOOD,
    (State.DEGRADED, Event.RECOVERED): State.NORMAL,
    (State.LAST_KNOWN_GOOD, Event.OPERATOR_OVERRIDE): State.MANUAL,
    (State.LAST_KNOWN_GOOD, Event.RECOVERED): State.NORMAL,
    (State.MANUAL, Event.RECOVERED): State.NORMAL,
}


class IllegalTransition(Exception):
    """Raised when no declared edge exists for a (state, event) pair."""

Step 3 — A pure transition function

The core is a single pure function: same inputs, same output, no side effects, no clock read. Purity is what makes the machine testable and replayable — a caller can fold an entire event log through it and get a deterministic final state. An undeclared pair raises IllegalTransition rather than returning the current state, so an illegal recovered on an already-NORMAL machine is surfaced, not swallowed.

def transition(state: State, event: Event) -> State:
    """Return the next state for (state, event) or raise IllegalTransition."""
    try:
        return TRANSITIONS[(state, event)]
    except KeyError:
        raise IllegalTransition(f"no transition from {state.value} on {event.value}")

Step 4 — Drive it from events and serialize every hop

The driver wraps the pure function with exactly one impure concern: logging. Each accepted hop produces an immutable TransitionRecord and emits it as single-line JSON, so the audit trail is the by-product of running the machine, never a thing someone must remember to write.

@dataclass(frozen=True)
class TransitionRecord:
    vehicle_id: str
    from_state: str
    to_state: str
    trigger: str
    ts: str
    detail: Optional[str] = None


class JSONFormatter(logging.Formatter):
    """Emit each fallback transition as one JSON line for audit ingestion."""

    def format(self, record: logging.LogRecord) -> str:
        return json.dumps({
            "logged_at": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "event": record.getMessage(),
            "transition": getattr(record, "transition", None),
        })


audit_logger = logging.getLogger("waste_ops.fallback_fsm")
if not audit_logger.handlers:
    _h = logging.StreamHandler()
    _h.setFormatter(JSONFormatter())
    audit_logger.addHandler(_h)
audit_logger.setLevel(logging.INFO)


class FallbackMachine:
    def __init__(self, vehicle_id: str, state: State = State.NORMAL) -> None:
        self.vehicle_id = vehicle_id
        self.state = state

    def dispatch(self, event: Event, ts: str, detail: Optional[str] = None) -> TransitionRecord:
        next_state = transition(self.state, event)  # raises on illegal pairs
        rec = TransitionRecord(
            vehicle_id=self.vehicle_id,
            from_state=self.state.value,
            to_state=next_state.value,
            trigger=event.value,
            ts=ts,
            detail=detail,
        )
        self.state = next_state
        audit_logger.info("fallback_transition", extra={"transition": asdict(rec)})
        return rec


if __name__ == "__main__":
    m = FallbackMachine("truck-114")
    m.dispatch(Event.TELEMETRY_GAP, "2026-07-16T09:00:00+00:00", "3 missed GPS polls")
    m.dispatch(Event.SOLVER_FAIL, "2026-07-16T09:02:30+00:00", "re-opt timeout 30s")
    m.dispatch(Event.OPERATOR_OVERRIDE, "2026-07-16T09:05:00+00:00", "dispatcher took manual control")
    print("final:", m.state.value)  # MANUAL

Compliance Output

Folding the three events above yields NORMAL → DEGRADED → LAST_KNOWN_GOOD → MANUAL and appends three records to the audit stream:

{"logged_at": "2026-07-16T09:05:01.221+00:00", "level": "INFO", "event": "fallback_transition", "transition": {"vehicle_id": "truck-114", "from_state": "NORMAL", "to_state": "DEGRADED", "trigger": "telemetry_gap", "ts": "2026-07-16T09:00:00+00:00", "detail": "3 missed GPS polls"}}
{"logged_at": "2026-07-16T09:05:01.221+00:00", "level": "INFO", "event": "fallback_transition", "transition": {"vehicle_id": "truck-114", "from_state": "DEGRADED", "to_state": "LAST_KNOWN_GOOD", "trigger": "solver_fail", "ts": "2026-07-16T09:02:30+00:00", "detail": "re-opt timeout 30s"}}
{"logged_at": "2026-07-16T09:05:01.221+00:00", "level": "INFO", "event": "fallback_transition", "transition": {"vehicle_id": "truck-114", "from_state": "LAST_KNOWN_GOOD", "to_state": "MANUAL", "trigger": "operator_override", "ts": "2026-07-16T09:05:00+00:00", "detail": "dispatcher took manual control"}}

Each field earns its place in the audit trail:

• from_state / to_state — the declared endpoints of the hop. Because they name enum members, an auditor can confirm the transition against the published table without reading application code.
• trigger — the event that caused the hop, and the only input other than the current state that the machine consulted. Together (from_state, trigger) deterministically imply to_state, which is what makes the record reproducible.
• ts — the tz-aware UTC event time. Under the Hours-of-Service electronic-logging rule (49 CFR Part 395), any duty-affecting event on a driver’s day must be time-ordered and non-back-dated; a fallback to MANUAL changes how a route is worked and therefore belongs on that timeline, which the DOT/FMCSA rule mapping reference resolves into the auditor-facing field.
• detail — a human note carried for diagnosis only. It is deliberately absent from the transition logic so it can never make two identical event sequences diverge.

Because the record is emitted on the same call that applies the state change, no state can move without leaving a trail — the “unlogged state change” failure mode is structurally impossible.

Verification

The tests prove the two properties that matter: the machine is deterministic under replay, and it refuses undeclared transitions.

import pytest


def replay(events: list[Event], start: State = State.NORMAL) -> State:
    state = start
    for e in events:
        state = transition(state, e)
    return state


def test_same_sequence_is_deterministic():
    seq = [Event.TELEMETRY_GAP, Event.SOLVER_FAIL, Event.OPERATOR_OVERRIDE]
    assert replay(seq) == replay(seq) == State.MANUAL


def test_recovered_returns_to_normal_from_any_fallback():
    assert replay([Event.TELEMETRY_GAP, Event.RECOVERED]) == State.NORMAL
    assert replay([Event.SOLVER_FAIL, Event.SOLVER_FAIL, Event.RECOVERED]) == State.NORMAL


def test_illegal_transition_raises():
    with pytest.raises(IllegalTransition):
        transition(State.NORMAL, Event.RECOVERED)   # nothing to recover from
    with pytest.raises(IllegalTransition):
        transition(State.MANUAL, Event.SOLVER_FAIL)  # manual hold ignores solver

Common Errors

IllegalTransition: no transition from NORMAL on recovered — a recovered event arrived while the machine was already NORMAL. Root cause: an upstream recovery detector fires unconditionally instead of only when the machine is in a degraded state. Fix: gate the emitter on machine.state != State.NORMAL, or catch IllegalTransition at the driver and treat a no-op recovery as benign — but never add a (NORMAL, RECOVERED) -> NORMAL self-edge just to silence it, because that hides real double-fires.

The final state depends on which handler ran first. Root cause: transitions were spread across several if event == ... blocks that each mutated self.state, so evaluation order changed the outcome. Fix: delete every branch and route all changes through the single transition table in Step 2 — a pure lookup has no ordering to get wrong.

TypeError: unhashable type when building the transition table. Root cause: keying TRANSITIONS with a list [State.NORMAL, Event.SOLVER_FAIL] instead of a tuple. Fix: use tuple keys (State.NORMAL, Event.SOLVER_FAIL); tuples of enum members are hashable and give the table its stable, order-free lookup.

FAQ

Why raise on an illegal transition instead of staying put?

Staying put hides a bug. If an event fires against a state that has no declared edge, something upstream is wrong — a detector double-fired, or an operator action reached a truck that was already in manual hold. Raising IllegalTransition surfaces that at the seam where it happened, while a silent no-op would let the fault propagate and make the audit trail lie about what the system saw.

How does this relate to what the solver actually does?

This machine only decides which strategy is authoritative; it never routes. When it enters LAST_KNOWN_GOOD it signals the batch recalculation layer to hold the last converged plan instead of re-solving on stale data, and MANUAL hands control to a dispatcher entirely. Separating the decision from the routing keeps both independently testable.

Up: Fallback Routing Logic

Related

• Fallback Routing Logic — the subsystem this state machine is the decision core of.
• Batch Recalculation on Demand-Spike Days — what a LAST_KNOWN_GOOD state signals downstream.
• Quarantining Malformed Telemetry Batches — the upstream detector that raises the telemetry_gap events driving this machine.
• Core Architecture & Compliance Mapping — the framework these fallback decisions live inside.