Runtime

This is where decisions happen. Module holds one compiled policy and the current state of one entity; Module.decide(event, evidence) takes one step and returns a DecisionRecord. Pipeline runs several modules on the same input and compose_parent combines their records (act only if every child acted; any abstain wins over any degrade). Open this page when you need the exact signature of decide, from_expr, restore or reset.

Public here: Module, Pipeline, compose_parent. A Module is not thread-safe: restore state per request under a lock, or keep one module per worker. Step-by-step walk-through: One decision, step by step; combining modules: Composing modules.

module

Encapsulated decision module.

A module owns a grammar (optional), a statechart, an envelope, and the six-check conjunction. Parent modules compose by feeding child DecisionRecords as evidence, not by reaching into child state.

Module dataclass

Module(
    name: str,
    chart: Statechart,
    envelope: Envelope = P_EXACT,
    limits: Limits = DEFAULT_LIMITS,
    expr: Expr | None = None,
    star_h: int = 0,
)

Encapsulated decision module.

Owns a statechart, an optional source expression, an envelope, and the six-check conjunction. Compile-time solvers (star-height, Zielonka, Karp) run in __post_init__. Per-event work is decide.

state property

state: str

Current chart state. Persist this on the worker; restore on the next request.

from_expr classmethod

from_expr(
    name: str,
    expr: Expr,
    *,
    envelope: Envelope = P_EXACT,
    limits: Limits = DEFAULT_LIMITS,
) -> Module

Compile a grammar expression to a module.

Checks star-height, builds a Thompson NFA / subset DFA, derives a statechart, and solves the game. Raises StarHeightError or CompileError before any decide call.

Source code in src/grounded_decision/runtime/module.py

@classmethod
def from_expr(
    cls,
    name: str,
    expr: Expr,
    *,
    envelope: Envelope = Envelope.P_EXACT,
    limits: Limits = DEFAULT_LIMITS,
) -> Module:
    """Compile a grammar expression to a module.

    Checks star-height, builds a Thompson NFA / subset DFA, derives
    a statechart, and solves the game. Raises ``StarHeightError`` or
    ``CompileError`` before any ``decide`` call.
    """

    dfa = compile_expr(expr, limits)
    chart = dfa_to_statechart(name, dfa)
    return cls(name=name, chart=chart, envelope=envelope, limits=limits, expr=expr)

peek_proposed_action

peek_proposed_action(event_name: str) -> str | None

The action decide would propose for this event, without deciding.

Does not change the state. Useful when the evidence depends on the proposed action, for example LearnedParameters.apply(evidence, proposed_action), which picks the calibration scores for that action.

Source code in src/grounded_decision/runtime/module.py

def peek_proposed_action(self, event_name: str) -> str | None:
    """The action ``decide`` would propose for this event, without deciding.

    Does not change the state. Useful when the evidence depends on the
    proposed action, for example
    ``LearnedParameters.apply(evidence, proposed_action)``, which picks
    the calibration scores for that action.
    """

    transition = self.chart.step(self._state, event_name)
    if transition is None:
        return None
    return transition.action or self.chart.action_of(transition.target)

restore

restore(state: str) -> None

Load a previously persisted state name. Unknown names raise KeyError.

Source code in src/grounded_decision/runtime/module.py

def restore(self, state: str) -> None:
    """Load a previously persisted state name. Unknown names raise ``KeyError``."""

    if state not in self.chart.states():
        raise KeyError(f"unknown state {state!r} for module {self.name!r}")
    self._state = state

decide

decide(
    event: Event, evidence: EvidenceSnapshot
) -> DecisionRecord

Take one step.

Advances the statechart on event.name, then evaluates the six checks (certificate, game, reconstruction, threshold, conformal, support). Any failed channel yields Verdict.ABSTAIN with proposed_action filled. Persist DecisionRecord.to_dict().

Source code in src/grounded_decision/runtime/module.py

def decide(self, event: Event, evidence: EvidenceSnapshot) -> DecisionRecord:
    """Take one step.

    Advances the statechart on ``event.name``, then evaluates the
    six checks (certificate, game, reconstruction, threshold,
    conformal, support). Any failed channel yields ``Verdict.ABSTAIN`` with
    ``proposed_action`` filled. Persist ``DecisionRecord.to_dict()``.
    """

    trans = self.chart.step(self._state, event.name)
    proposed: str | None = None
    if trans is None:
        # Unknown event: stay, propose nothing.
        state = self._state
    else:
        self._state = trans.target
        state = self._state
        proposed = trans.action or self.chart.action_of(state)

    cert = self._certificate_channel()
    game = self._game_channel()
    recon = reconstruction_report(
        evidence.degree, evidence.lambda_hat, limits=self.limits
    )
    thresh = threshold_report(
        evidence.n,
        evidence.p_hat,
        evidence.p_threshold,
        limits=self.limits,
    )
    conf = conformal_report(
        evidence.label_score, evidence.scores, limits=self.limits
    )
    support = support_report(evidence.support, self.limits)
    channels = (cert, game, recon, thresh, conf, support)
    health = Health(
        d_lambda_sq=float(recon.metrics["d_lambda_sq"]),
        reconstruction_margin=float(recon.metrics["margin"]),
        p_hat=evidence.p_hat,
        p_threshold=evidence.p_threshold,
        threshold_width_theorem=float(thresh.metrics["theorem_width"]),
        threshold_width_operational=float(thresh.metrics["operational_width"]),
        in_theorem_window=bool(thresh.metrics["in_theorem_window"]),
        in_operational_window=bool(thresh.metrics["in_operational_window"]),
        theorem_window_vacuous=bool(thresh.metrics["theorem_vacuous"]),
        conformal_p=float(conf.metrics["p_value"]),
        star_height=self.star_h,
        game_solved=self._game_ok,
        mixer_status="idle",
        support=evidence.support,
    )

    failed = [c for c in channels if not c.passed]
    state_verdict = self.chart.verdict_of(state)

    if failed:
        reason = "; ".join(c.summary for c in failed)
        return DecisionRecord(
            verdict=Verdict.ABSTAIN,
            action=None,
            proposed_action=proposed,
            reason=reason,
            module=self.name,
            event=event,
            channels=channels,
            health=health,
            envelope=self.envelope,
            extras={"state": state},
        )

    if proposed and (state_verdict in {None, Verdict.ACT}):
        return DecisionRecord(
            verdict=Verdict.ACT,
            action=proposed,
            proposed_action=proposed,
            reason="all channels passed",
            module=self.name,
            event=event,
            channels=channels,
            health=health,
            envelope=self.envelope,
            extras={"state": state},
        )

    if state_verdict is Verdict.DEGRADE:
        return DecisionRecord(
            verdict=Verdict.DEGRADE,
            action=self.chart.action_of(state),
            proposed_action=proposed,
            reason="statechart degrade",
            module=self.name,
            event=event,
            channels=channels,
            health=health,
            envelope=self.envelope,
            extras={"state": state},
        )

    return DecisionRecord(
        verdict=Verdict.ABSTAIN,
        action=None,
        proposed_action=proposed,
        reason=(
            f"event {event.name!r} is not handled in state {state!r}; "
            "nothing proposed"
            if trans is None
            else "no action marked on current state"
        ),
        module=self.name,
        event=event,
        channels=channels,
        health=health,
        envelope=self.envelope,
        extras={"state": state},
    )

compose_parent

compose_parent(
    name: str, children: Sequence[DecisionRecord]
) -> Verdict

Conservative aggregate of child verdicts.

Any child abstained: abstain. Otherwise any child degraded: degrade. Every child acted: act. No children: abstain. Abstain dominates degrade because a held child means part of the decision has no support at all, while a degraded child still ran its fallback.

Source code in src/grounded_decision/runtime/module.py

def compose_parent(
    name: str,
    children: Sequence[DecisionRecord],
) -> Verdict:
    """Conservative aggregate of child verdicts.

    Any child abstained: abstain. Otherwise any child degraded: degrade.
    Every child acted: act. No children: abstain. Abstain dominates
    degrade because a held child means part of the decision has no
    support at all, while a degraded child still ran its fallback.
    """

    if not children:
        return Verdict.ABSTAIN
    if any(c.verdict is Verdict.ABSTAIN for c in children):
        return Verdict.ABSTAIN
    if any(c.verdict is Verdict.DEGRADE for c in children):
        return Verdict.DEGRADE
    return Verdict.ACT

pipeline

A pipeline is an ordered list of modules with a shared evidence snapshot.

Each module sees the same event stream. The pipeline verdict is the conservative aggregate of module verdicts.