Integrate¶

These helpers put Module.decide behind a service. handle_request is the main entry point: it takes a JSON-shaped body with an optional saved state, an event and evidence, restores the state (or resets when none is given), decides once and returns the record as a dict. DecisionSession replays a stream of events through one module, and EvidencePrior, Volatility and draw_snapshot generate noisy evidence for simulations and tests.

Public here: handle_request, DecisionSession, EvidencePrior, Volatility, draw_snapshot (top level); dumps_record, loads_event_evidence, trajectory, ramp_snapshots and random_walk_p from grounded_decision.integrate. Guide: Integrate into a service; full working service: Build a decision service.

codec ¶

JSON request/response helpers for embedding Module.decide in a service.

The intended worker pattern is stateless:

Persist DecisionRecord.extras["state"] next to the entity.
On the next event, send that state in the request body.
handle_request restores it, steps once, returns a JSON record.

One Module instance may be reused across requests only if you restore state at the start of each call (or use a dedicated instance per in-flight session). The chart itself is not thread-safe.

loads_event_evidence ¶

loads_event_evidence(
    body: Mapping[str, Any],
) -> tuple[Event, EvidenceSnapshot]

Parse the two kernel inputs from a JSON object.

Source code in src/grounded_decision/integrate/codec.py

def loads_event_evidence(body: Mapping[str, Any]) -> tuple[Event, EvidenceSnapshot]:
    """Parse the two kernel inputs from a JSON object."""

    return Event.from_dict(body["event"]), EvidenceSnapshot.from_dict(body["evidence"])

handle_request ¶

handle_request(
    module: Module,
    body: Mapping[str, Any],
    *,
    restore_state: bool = True,
) -> dict[str, Any]

One HTTP-shaped call: optional state, then event + evidence.

Returns DecisionRecord.to_dict(). Downstream stores extras["state"] and the verdict. This is the integration point for FastAPI/Flask/gRPC wrappers; those frameworks are not a dependency of this package.

Source code in src/grounded_decision/integrate/codec.py

def handle_request(
    module: Module,
    body: Mapping[str, Any],
    *,
    restore_state: bool = True,
) -> dict[str, Any]:
    """One HTTP-shaped call: optional ``state``, then ``event`` + ``evidence``.

    Returns ``DecisionRecord.to_dict()``. Downstream stores
    ``extras["state"]`` and the verdict. This is the integration point
    for FastAPI/Flask/gRPC wrappers; those frameworks are not a
    dependency of this package.
    """

    if restore_state and body.get("state"):
        module.restore(str(body["state"]))
    elif restore_state and not body.get("state"):
        module.reset()
    event, evidence = loads_event_evidence(body)
    return module.decide(event, evidence).to_dict()

session ¶

Sequential decide over a stream of (event, evidence) ticks.

DecisionSession `dataclass` ¶

DecisionSession(
    module: Module,
    reset_each_tick: bool = False,
    records: list[DecisionRecord] = list[DecisionRecord](),
)

One module, many ticks, optional reset between ticks.

reset_each_tick=False is the workflow/incident pattern: the chart keeps its state. reset_each_tick=True is the independent scoring pattern: each tick is a fresh intake.

TrajectoryRow ¶

Bases: TypedDict

One row of :func:trajectory.

trajectory ¶

trajectory(
    records: Sequence[DecisionRecord],
) -> list[TrajectoryRow]

Compact time series for plotting act/abstain against evidence.

Source code in src/grounded_decision/integrate/session.py

def trajectory(records: Sequence[DecisionRecord]) -> list[TrajectoryRow]:
    """Compact time series for plotting act/abstain against evidence."""

    rows: list[TrajectoryRow] = []
    for i, rec in enumerate(records):
        rows.append(
            {
                "t": i,
                "event": rec.event.name,
                "verdict": rec.verdict.value,
                "action": rec.action,
                "proposed_action": rec.proposed_action,
                "state": rec.extras.get("state"),
                "acted": rec.verdict is Verdict.ACT,
                "d_lambda_sq": rec.health.d_lambda_sq,
                "p_hat": rec.health.p_hat,
                "p_threshold": rec.health.p_threshold,
                "in_operational_window": rec.health.in_operational_window,
                "conformal_p": rec.health.conformal_p,
                "failed_channels": [c.name.value for c in rec.channels if not c.passed],
            }
        )
    return rows

priors ¶

Priors and volatility for evidence over time.

A prior is the mean graph the operator believes in. Volatility is independent Gaussian noise on degree, λ, density, and the conformal score, clipped to the legal range. Adaptive examples draw a path of EvidenceSnapshots from these two objects so the same module can be probed under calm, noisy, and regime-shift conditions.

EvidencePrior `dataclass` ¶

EvidencePrior(
    n: int = 80,
    p_threshold: float = 0.1,
    degree: float = 8.0,
    lambda_mean: float = 0.55,
    p_mean: float = 0.35,
    label_mean: float = 0.1,
    n_cal: int = 40,
    n_states: int = 2,
    cal_loc: float = 0.4,
    cal_step: float = 0.02,
)

Mean evidence a service believes before seeing this tick's graph.

Volatility `dataclass` ¶

Volatility(
    degree: float = 0.0,
    lambda_hat: float = 0.0,
    p_hat: float = 0.0,
    label_score: float = 0.0,
)

Per-tick standard deviations. Zero is a frozen prior.

draw_snapshot ¶

draw_snapshot(
    prior: EvidencePrior,
    vol: Volatility = CALM,
    *,
    rng: Generator | None = None,
) -> EvidenceSnapshot

One noisy draw from prior under vol.

Source code in src/grounded_decision/integrate/priors.py

def draw_snapshot(
    prior: EvidencePrior,
    vol: Volatility = CALM,
    *,
    rng: np.random.Generator | None = None,
) -> EvidenceSnapshot:
    """One noisy draw from ``prior`` under ``vol``."""

    rng = rng or np.random.default_rng()
    degree = _clip(prior.degree + rng.normal(0.0, vol.degree), 0.0, 64.0)
    lam = _clip(prior.lambda_mean + rng.normal(0.0, vol.lambda_hat), 0.0, 1.0)
    p_hat = _clip(prior.p_mean + rng.normal(0.0, vol.p_hat), 0.0, 1.0)
    label = _clip(prior.label_mean + rng.normal(0.0, vol.label_score), 0.0, 2.0)
    return EvidenceSnapshot(
        n=prior.n,
        p_hat=p_hat,
        p_threshold=prior.p_threshold,
        degree=degree,
        lambda_hat=lam,
        n_states=prior.n_states,
        scores=prior.calibration(),
        label_score=label,
        observations={
            "prior_d_lambda_sq": prior.mean_d_lambda_sq(),
            "vol_degree": vol.degree,
            "vol_lambda": vol.lambda_hat,
            "vol_p": vol.p_hat,
        },
    )

ramp_snapshots ¶

ramp_snapshots(
    *,
    steps: int,
    start: EvidencePrior,
    end: EvidencePrior,
    vol: Volatility = CALM,
    rng: Generator | None = None,
) -> list[EvidenceSnapshot]

Linear interpolation of the prior, one draw per step.

Source code in src/grounded_decision/integrate/priors.py

def ramp_snapshots(
    *,
    steps: int,
    start: EvidencePrior,
    end: EvidencePrior,
    vol: Volatility = CALM,
    rng: np.random.Generator | None = None,
) -> list[EvidenceSnapshot]:
    """Linear interpolation of the prior, one draw per step."""

    rng = rng or np.random.default_rng()
    out: list[EvidenceSnapshot] = []
    for i in range(steps):
        t = 0.0 if steps == 1 else i / (steps - 1)
        mid = EvidencePrior(
            n=round((1 - t) * start.n + t * end.n),
            p_threshold=(1 - t) * start.p_threshold + t * end.p_threshold,
            degree=(1 - t) * start.degree + t * end.degree,
            lambda_mean=(1 - t) * start.lambda_mean + t * end.lambda_mean,
            p_mean=(1 - t) * start.p_mean + t * end.p_mean,
            label_mean=(1 - t) * start.label_mean + t * end.label_mean,
            n_cal=start.n_cal,
            n_states=start.n_states,
            cal_loc=start.cal_loc,
            cal_step=start.cal_step,
        )
        out.append(draw_snapshot(mid, vol, rng=rng))
    return out

random_walk_p ¶

random_walk_p(
    *,
    steps: int,
    start: float,
    threshold: float,
    sigma: float,
    n: int,
    degree: float,
    lambda_hat: float,
    rng: Generator | None = None,
) -> list[EvidenceSnapshot]

Random walk on density; other fields frozen. Probes the FK window.

Source code in src/grounded_decision/integrate/priors.py

def random_walk_p(
    *,
    steps: int,
    start: float,
    threshold: float,
    sigma: float,
    n: int,
    degree: float,
    lambda_hat: float,
    rng: np.random.Generator | None = None,
) -> list[EvidenceSnapshot]:
    """Random walk on density; other fields frozen. Probes the FK window."""

    rng = rng or np.random.default_rng()
    p = start
    scores = tuple(0.4 + 0.02 * i for i in range(40))
    out: list[EvidenceSnapshot] = []
    for _ in range(steps):
        p = _clip(p + rng.normal(0.0, sigma), 0.0, 1.0)
        out.append(
            EvidenceSnapshot(
                n=n,
                p_hat=p,
                p_threshold=threshold,
                degree=degree,
                lambda_hat=lambda_hat,
                scores=scores,
                label_score=0.1,
                observations={"walk_p": p},
            )
        )
    return out

Integrate¶

codec ¶

loads_event_evidence ¶

handle_request ¶

session ¶

DecisionSession dataclass ¶

TrajectoryRow ¶

trajectory ¶

priors ¶

EvidencePrior dataclass ¶

Volatility dataclass ¶

draw_snapshot ¶

ramp_snapshots ¶

random_walk_p ¶

DecisionSession `dataclass` ¶

EvidencePrior `dataclass` ¶

Volatility `dataclass` ¶