Integrate into a service¶

The library is a function you call inside your own process, not a hosted product: you give it an event and the evidence, it gives back a record saying act, hold or fall back. Everything around that call (remembering where each entity is, surviving retried requests, executing the action exactly once, sending held cases to people, keeping an audit trail) is yours to build. This page shows how to build it by starting from the reference service that ships with the package and swapping its parts for the ones your platform already runs: FastAPI instead of http.server, a queue consumer instead of HTTP, Postgres instead of SQLite.

Who this is for: the engineer wiring decisions into an existing service.

What you need:

To have run Build a decision service, the tutorial for src/grounded_decision/examples/reference_service.py. This page assumes you have seen each of its steps work.
A policy (Write a policy) and an evidence builder (Build evidence) for your domain, or the fraud template to start with.

What every integration must do¶

Whatever the transport, one decision is the same eight steps. The reference service does them in this order, and the order matters:

flowchart TD
    A[request or message] --> B["1. take the entity's lock"]
    B --> C["2. seen this idempotency key? return the stored response"]
    C --> D["3. load the entity's state; policy version must match, else 409"]
    D --> E["4. build evidence (outside any transaction; may be slow)"]
    E --> F["5. restore state, decide (under the module lock; microseconds)"]
    F --> G[6. route the verdict]
    G --> H["7. ONE transaction: new state (version check) + audit row
    + review row or outbox row + idempotency key with response"]
    H --> I[8. one log line, metrics]
    H -.-> J[outbox relay executes the action, idempotently]

Concern	Reference service (`reference_service.py`)	What you might replace it with
Build the policy once, version it	`build_fraud_engine`, `policy_fingerprint`	your own builder; keep the fingerprint
A `Module` is not thread-safe	`DecisionEngine` (one lock around restore + decide)	a pool of modules, or one module per worker process
Two events for one entity at once	`EntityLocks` plus the `version` column	`SELECT ... FOR UPDATE`, or partition the stream by entity
Per-entity state	`StateStore` (`entity_state` table)	a Postgres table, a Redis hash
Evidence from your data	`EvidenceProvider`, `SqliteDeviceGraphEvidenceProvider`	warehouse queries, a feature store, a cache filled by a batch job
Retried requests	`IdempotencyStore`	the same table in your database
Side effects exactly once	`Outbox` plus a relay you write	the same table, Debezium, a worker
Held cases	`ReviewQueue`, `POST /reviews/<id>/resolve`	your case tool and its webhook
Audit trail	`AuditLog` (append-only by trigger)	an append-only table with `UPDATE`/`DELETE` revoked
Observability	`ServiceMetrics` (`GET /metrics`), one JSON log line	your metrics and logging stack
Rollout	`--shadow`, `GET /report`	the same; see Roll out safely

The rest of this page goes through the pieces you are most likely to change.

Step 1: build the policy once, and version it¶

Building a Module runs the expensive analysis of the policy; decide takes microseconds. Build it when the process starts, never per request.

A saved state such as review is a position in one compiled policy. Store a fingerprint of the policy next to every saved state and refuse to restore across a change. The reference service's DecisionEngine keeps two versions: policy_fingerprint(module) hashes what gives a saved state its meaning (the module name, envelope, chart and policy text), and parameters_fingerprint hashes the settings that change what the checks conclude (Limits, your evidence settings, learned parameters). Reuse it:

from grounded_decision import Limits, Module
from grounded_decision.examples.reference_service import DecisionEngine

LIMITS = Limits(require_conformal_calibration=True)  # empty calibration holds


def build_engine(chart, evidence_parameters) -> DecisionEngine:
    module = Module(name="card_auth", chart=chart, limits=LIMITS)
    return DecisionEngine(module, evidence_parameters)

engine.policy_version and engine.parameters_version are the two strings to store.

require_conformal_calibration=True makes an empty calibration set hold instead of pass; choose its other values with Choose thresholds and calibration. Two processes that compile the same policy get the same fingerprint, so a rolling deploy of unchanged code is seamless. A policy edit changes the policy version and needs a state migration; a threshold or learned-parameter change changes only the settings version (Roll out safely).

Step 2: build evidence where it is cheap¶

The decide call needs an EvidenceSnapshot. Build it with snapshot_from_graph from your raw rows (the related cases, the links between them, each one's known outcome, and the calibration scores); Build evidence has recipes for several domains.

Where that code runs is an integration choice:

Per request, as the reference service does: evidence_for walks the device graph in SQLite for each request, capped at 500 accounts. Simple; latency is the graph query.
In a batch job, every few minutes: cache EvidenceSnapshot.to_dict() per entity and have evidence_for read the cache and set label_score (this case's model score) from the request. Use this when the graph query is slow.

Either way, three rules hold. Missing data must give thin evidence (an empty graph, which holds), never made-up numbers. Bound the work (cap the neighbourhood, set a query timeout). Return every decision-relevant setting from evidence_parameters(), so the settings version changes when a threshold does.

Step 3: the per-request core¶

If you build your own service instead of copying the reference one, the core is handle_request: it restores the saved state (or resets to the start for a new entity), decides once, and returns DecisionRecord.to_dict(). The entity's new state is DecisionRecord.extras["state"] (in the dict, record["extras"]["state"]). Save it back against the entity on every verdict, including abstain: the policy advances on every event it recognises; only the side effect is held back.

import threading

from grounded_decision import handle_request
from grounded_decision.examples import fraud_review

MODULE = fraud_review.build()  # compile once, at start-up
MODULE_LOCK = threading.Lock()  # a Module is not thread-safe
saved_state: dict[str, str] = {}  # stands in for your state table


def decide_for(entity_id: str, event: dict, evidence: dict) -> dict:
    body = {"state": saved_state.get(entity_id), "event": event, "evidence": evidence}
    with MODULE_LOCK:  # restore + decide must not interleave
        record = handle_request(MODULE, body)
    saved_state[entity_id] = record["extras"]["state"]  # save it, even on abstain
    return record


ring = fraud_review.device_graph_evidence(
    *fraud_review.dense_fraud_ring_logins(), candidate_model_score=0.1
).to_dict()
for _ in range(2):
    record = decide_for("acct_000", {"name": "suspicious"}, ring)
    print(
        record["verdict"],
        record["action"],
        record["proposed_action"],
        record["extras"]["state"],
        record["reason"],
    )

act block block review all channels passed
abstain None None review event 'suspicious' is not handled in state 'review'; nothing proposed

The first suspicious event blocks the card and moves it to review. The second proposes nothing, because the fraud policy has no suspicious transition out of review: the card is not blocked twice.

This core is correct for one process and one thread per entity at a time. A real service adds steps 1, 2, 7 and 8 from the diagram; the reference service's DecisionService.decide is that code, and the simplest production path is to keep it and change only the transport.

Step 4: put the service behind FastAPI¶

DecisionService knows nothing about HTTP, so a FastAPI app is a thin adapter around it. FastAPI is not a dependency of this package; this is a sketch for your service:

# app.py: SKETCH. FastAPI is your dependency, not this package's.
import uuid
from pathlib import Path

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, Response

from grounded_decision.examples.reference_service import (
    ServiceError,
    build_service,
    parse_decide_request,
)

service = build_service(Path("/var/lib/decisions/decisions.sqlite3"))
app = FastAPI()


@app.post("/decide")
def decide(
    body: dict, request: Request
) -> Response:  # plain def: runs in a thread pool
    request_id = request.headers.get("x-request-id") or str(uuid.uuid4())
    try:
        parsed = parse_decide_request(body, service.engine.valid_event_names)
        result = service.decide(parsed, request_id)
    except ServiceError as error:  # 400 bad request, 409 version / key / conflict
        return JSONResponse(error.to_dict(), status_code=int(error.status))
    headers = {"Idempotent-Replay": "true"} if result.replayed else {}
    return Response(
        result.response_json, media_type="application/json", headers=headers
    )

The body of decide is the reference service's own HTTP handler, so it behaves the same. Run without FastAPI, the same calls return 200 execute for a first request and 200 execute with Idempotent-Replay: true for an exact retry, and the outbox holds one block.

Concurrency carries over unchanged. FastAPI runs a plain def handler on a thread pool; DecisionService already serialises restore + decide under the engine lock, serialises one entity's requests with EntityLocks, and opens one short-lived database connection per unit of work. With several uvicorn or gunicorn worker processes, each builds its own service at start-up; same-entity races between processes are caught by the version column (the loser retries up to three times, then returns 409 state_conflict), and idempotency-key races by the key's primary key.

Do not call service.decide from an async def handler: it blocks on SQLite and on the locks. Keep the handler a plain def, or wrap the call in run_in_threadpool.

Step 5: or consume a queue instead¶

The same service sits behind a Kafka (or SQS, or NATS) consumer. Two things change: the idempotency key must come from the message, so a redelivery is a replay, and errors decide whether to retry or dead-letter. Sketch:

# consumer.py: SKETCH. Your Kafka client; topic partitioned by entity id.
import json

from grounded_decision.examples.reference_service import (
    ServiceError,
    parse_decide_request,
)

DEAD_LETTER = {"invalid_request", "policy_version_mismatch", "idempotency_key_reused"}

for message in consumer:
    body = json.loads(message.value)
    body["idempotency_key"] = body.get("event_id") or (
        f"{message.topic}:{message.partition}:{message.offset}"
    )
    try:
        request = parse_decide_request(body, service.engine.valid_event_names)
        service.decide(request, request_id=body["idempotency_key"])
    except ServiceError as error:
        if error.error_code in DEAD_LETTER:
            dead_letter_producer.send(message, error.to_dict())
        else:
            raise  # state_conflict: do not commit; the message is redelivered
    consumer.commit(message)

Partition the topic by entity id so one consumer sees one entity's events in order; a multi-step policy depends on that order. Commit the offset after decide returns: if the process dies in between, the redelivered message hits the idempotency store and returns the stored response, so nothing happens twice.

Step 6: route the verdict¶

The reference service turns each record into one of four routings with a small pure function, route_decision:

Verdict	`proposed_action`	Routing (`reference_service`)	Live examples record it as	What happens
act	set	`execute`	the action name	the action is written to the outbox
degrade	any	`execute_degraded`	the fallback action	the policy's fallback is written to the outbox
abstain	set	`human_review`	`hold_for_human`	a review is opened with the proposal and reason
abstain	none	`await_next_event`	`await_next_event`	nothing; the policy is waiting for its next event (even if a check failed)

A proposed_action on an abstain is what the policy wanted, kept so the reviewer can see it. It is never permission to act. The live desks use the name hold_for_human for the same routing the reference service calls human_review; pick one name for your dashboards.

Show the reviewer why the case was held: the failed checks by name, reason, and the numbers in health. "Reconstruction failed, three links" and "conformal failed, score unlike calibration" call for different decisions. Troubleshoot translates each reason.

Step 7: execute side effects exactly once¶

Never call the card processor (or pager, or loan system) from the request handler: the call can succeed while the database commit fails, or the other way round. The reference service writes the intended action to the outbox table in the same transaction as the state, the audit row and the idempotency key, and leaves execution to a relay. The relay is yours; a minimal one over the reference service's database:

def relay_once(execute) -> int:
    with service.database.connection() as connection:
        pending = connection.execute(
            "SELECT outbox_id, entity_id, action, effect_key FROM outbox"
            " WHERE status = 'pending' ORDER BY outbox_id"
        ).fetchall()
    for row in pending:
        execute(row["action"], row["entity_id"], idempotency_key=row["effect_key"])
        with service.database.connection() as connection:
            connection.execute(
                "UPDATE outbox SET status = 'sent' WHERE outbox_id = ?",
                (row["outbox_id"],),
            )
    return len(pending)

Run once after the block above, it executes block acct_000 decision:1 and returns 1; run again, it returns 0. If the relay dies between execute and the update, the next run sends the same effect_key again, so the downstream system must deduplicate on it.

Step 8: move from SQLite to Postgres¶

SQLite in write-ahead-log mode handles one writer at a time: enough for one host and a few hundred decisions a second. Beyond that, replace the Database class with a connection pool. Its docstring lists the four changes: ? placeholders become %s, INTEGER PRIMARY KEY AUTOINCREMENT becomes an identity column, BEGIN IMMEDIATE becomes BEGIN, and the append-only triggers become REVOKE UPDATE, DELETE on the audit table. On Postgres you may replace the optimistic version check with SELECT ... FOR UPDATE on the entity's row at the start of the transaction; both are correct, and the optimistic check is cheaper when two events for the same entity rarely collide.

The wire format of `handle_request`¶

If another service calls your decide endpoint with the raw kernel inputs (rather than an entity id, as the reference service does), the request body for handle_request is:

{
  "state": "review",
  "event": { "name": "suspicious", "payload": { "amount_usd": 840 } },
  "evidence": {
    "n": 16,
    "p_hat": 0.333,
    "p_threshold": 0.08,
    "degree": 5.0,
    "lambda_hat": 0.55,
    "n_states": 2,
    "scores": [0.4, 0.42, 0.44],
    "label_score": 0.1,
    "observations": {}
  }
}

state is null or absent for a new entity. The response is DecisionRecord.to_dict(); every type round-trips with from_dict, so a replay job can rebuild a record without pickle. The example step_json_roundtrip runs this round trip:

uv run grounded-decision examples step_json_roundtrip

What can go wrong¶

Sharing one Module between threads without a lock. Request A restores state S1, request B restores S2, A decides from B's state. Use the engine lock, a pool of modules, or one module per process.
Not saving state on abstain. The policy moved; your stored state did not. The next event is decided from the wrong place.
Returning early when evidence is missing. Decide with the empty graph instead (it holds) so the state, audit row and review are written like any other decision.
Unknown event names reaching decide. The state does not move, nothing is proposed, and the typo is invisible. Validate event names at the edge, as parse_decide_request does, and return 400.
Calling the router or a slow query inside the decide lock. Keep everything except restore + decide outside it. Call the router beside decide and combine the results yourself.
Reading an act as "the evidence supports this action". Only the support check looks at direction, and only when your evidence carries a support share; see It acts on the wrong side.

Limits.decide_ms_budget (50 ms) and compile_ms_budget (200 ms) are targets, not timers: the library never interrupts a slow call (tests/test_performance_budget.py checks them on the library itself). In the reference service a decision takes about 2 ms end to end, almost all of it evidence I/O.

Next¶

Roll out safely: shadow mode, comparing with your current rule, and switching enforcement on.
Operate and monitor: metrics, alerts and the runbook.
State and policy versions: why the fingerprint exists.
Composing modules: when one decision needs several policies to agree.