Examples and CLI¶

The example library is a registry of runnable stories (fraud review, clinical triage, autoscaling and about sixty more), each returning its decision records and a summary. list_examples, run_example and run_all let you run them from Python instead of parsing terminal output, and interface_map returns a machine-readable summary of the public surface. The cli module is the grounded-decision command.

Public here: list_examples, run_example, run_all, interface_map from grounded_decision.examples.catalog. The command line, with real output: Command line; what each example teaches: Example library.

catalog ¶

Registry of every runnable example, and the public interface map.

This is the one place that knows which examples exist. The command line (uv run grounded-decision examples), the test suite and the documentation all go through it, so adding an example means adding it here.

How to use it::

from grounded_decision.examples.catalog import list_examples, run_example

list_examples(track="live")          # [{"id", "title", "track"}, ...]
run_example("fraud_review")          # prints the human-readable report
run_example("fraud_review", as_json=True)["records"]

Each example's runner is a function run(*, as_json: bool) -> dict that returns {"id", "title", "records", "summary"}. Product examples expose it as a module-level run next to ID and TITLE; the step, adaptive, combo, live, markets and games tracks expose a RUNNERS dict of id -> (title, runner). See :mod:grounded_decision.examples for what each track teaches and the recommended reading order.

:func:interface_map returns a machine-readable summary of the public surface (decide inputs and outputs, the six checks, builders, router subgoals, compare helpers, integration entry points) and is what uv run grounded-decision interfaces prints.

INTERFACE_MAP_VERSION `module-attribute` ¶

INTERFACE_MAP_VERSION: Final = __version__

The interface map is versioned with the package.

TRACKS `module-attribute` ¶

TRACKS: Final = (
    "product",
    "step",
    "adaptive",
    "combo",
    "live",
    "markets",
    "games",
)

Every track name, in the order the catalog lists them.

ExampleRunner ¶

Bases: Protocol

runner(*, as_json: bool) -> {"id", "title", "records", "summary"}.

CatalogEntry ¶

Bases: NamedTuple

One registered example.

Attributes:

Name	Type	Description
`title`	`str`	Human heading, e.g. `"Fraud review"`.
`runner`	`ExampleRunner`	The function that runs the example.
`track`	`str`	Which track it belongs to (one of :data:`TRACKS`).

list_examples ¶

list_examples(
    track: str | None = None,
) -> list[dict[str, str]]

List the registered examples, optionally for one track.

Parameters:

Name	Type	Description	Default
`track`	`str \| None`	One of :data:`TRACKS`, or `None` for every example.	`None`

Returns:

Type	Description
`list[dict[str, str]]`	`[{"id": ..., "title": ..., "track": ...}, ...]` in catalog order.

Source code in src/grounded_decision/examples/catalog.py

def list_examples(track: str | None = None) -> list[dict[str, str]]:
    """List the registered examples, optionally for one track.

    Args:
        track: One of :data:`TRACKS`, or ``None`` for every example.

    Returns:
        ``[{"id": ..., "title": ..., "track": ...}, ...]`` in catalog order.
    """

    return [
        {"id": example_id, "title": entry.title, "track": entry.track}
        for example_id, entry in _EXAMPLES.items()
        if track is None or entry.track == track
    ]

run_example ¶

run_example(
    name: str, *, as_json: bool = False
) -> dict[str, Any]

Run one example by id and return its result.

Parameters:

Name	Type	Description	Default
`name`	`str`	The example id, e.g. `"fraud_review"`.	required
`as_json`	`bool`	Print one line of JSON instead of the human-readable report.	`False`

Returns:

Type	Description
`dict[str, Any]`	`{"id", "title", "records", "summary"}` from the example's runner.

Raises:

Type	Description
`SystemExit`	If `name` is not a registered example id.

Source code in src/grounded_decision/examples/catalog.py

def run_example(name: str, *, as_json: bool = False) -> dict[str, Any]:
    """Run one example by id and return its result.

    Args:
        name: The example id, e.g. ``"fraud_review"``.
        as_json: Print one line of JSON instead of the human-readable report.

    Returns:
        ``{"id", "title", "records", "summary"}`` from the example's runner.

    Raises:
        SystemExit: If ``name`` is not a registered example id.
    """

    if name not in _EXAMPLES:
        known_ids = ", ".join(_EXAMPLES)
        raise SystemExit(f"unknown example {name!r}. known: {known_ids}")
    return _EXAMPLES[name].runner(as_json=as_json)

run_all ¶

run_all(*, as_json: bool = False) -> list[dict[str, Any]]

Run every example in catalog order.

Parameters:

Name	Type	Description	Default
`as_json`	`bool`	Print JSON lines instead of human-readable reports (which are separated by a blank line).	`False`

Returns:

Type	Description
`list[dict[str, Any]]`	One result dict per example, in catalog order.

Source code in src/grounded_decision/examples/catalog.py

def run_all(*, as_json: bool = False) -> list[dict[str, Any]]:
    """Run every example in catalog order.

    Args:
        as_json: Print JSON lines instead of human-readable reports (which
            are separated by a blank line).

    Returns:
        One result dict per example, in catalog order.
    """

    results: list[dict[str, Any]] = []
    for example_id in _EXAMPLES:
        if not as_json:
            print()
        results.append(run_example(example_id, as_json=as_json))
    return results

interface_map ¶

interface_map() -> dict[str, Any]

Describe the public surface of the library as plain data.

Returns:

Type	Description
`dict[str, Any]`	A JSON-serialisable dict with the `decide` contract (inputs,
`dict[str, Any]`	output, verdicts and the six checks with their catalog
`dict[str, Any]`	references), the policy `builders`, the `router_subgoals`, the
`dict[str, Any]`	`compare` helpers, every registered example, the track names and
`dict[str, Any]`	the `integrate` entry points.

Source code in src/grounded_decision/examples/catalog.py

def interface_map() -> dict[str, Any]:
    """Describe the public surface of the library as plain data.

    Returns:
        A JSON-serialisable dict with the ``decide`` contract (inputs,
        output, verdicts and the six checks with their catalog
        references), the policy ``builders``, the ``router_subgoals``, the
        ``compare`` helpers, every registered example, the track names and
        the ``integrate`` entry points.
    """

    return {
        "version": INTERFACE_MAP_VERSION,
        "decide": {
            "input": ["Event(name, payload)", "EvidenceSnapshot(...)"],
            "output": "DecisionRecord.to_dict()",
            "verdicts": ["act", "abstain", "degrade"],
            "channels": [
                "certificate (star-height ≤ 3, Lean 134)",
                "game (parity + mean-payoff, Lean 104; Truffet banned)",
                "reconstruction (d λ² > 1+margin, Lean 229 supercritical)",
                "threshold (Friedgut–Kalai window, Lean 186; operational prefactor)",
                "conformal (split conformal, operational)",
                "support (share of related cases agreeing with the proposed action ≥ min_support, operational)",
            ],
        },
        "builders": {
            "Module.from_expr": "parse a policy string or Expr, compile DFA/statechart",
            "Statechart": "explicit states, transitions, payoffs, priorities",
            "Pipeline": "conservative AND of child modules",
        },
        "router_subgoals": [
            "max_cut",
            "vertex_cover",
            "k_median",
            "wl",
            "rational_zeros",
        ],
        "compare": [
            "best_coordinate (Courtade–Kumar 119)",
            "compare_outcome_laws (Brenier 1/3, 374)",
            "sample_two_row / sample_switch (tables 115)",
            "single_item_max_of_samples (prophet rule with the guaranteed 1/2; Lean 2^{-310} unused)",
            "single_item_one_sample (a scalar bar; no guaranteed factor)",
        ],
        "examples": list_examples(),
        "tracks": list(TRACKS),
        "integrate": {
            "handle_request": "JSON in, DecisionRecord.to_dict() out; restore extras['state']",
            "DecisionSession": "sequential decide over ticks",
            "EvidencePrior": "mean graph; Volatility = per-tick noise",
            "from_dict": "Event, EvidenceSnapshot, DecisionRecord",
        },
    }

cli ¶

Command-line interface: uv run grounded-decision <subcommand>.

Subcommands¶

demo (also the default when no subcommand is given) Run a short clinical-triage walkthrough: a handful of patients through one grounded module, printed as readable tick cards. The fastest way to see what a decision looks like.

examples [ID] [--track TRACK] [--json] Work with the example library (see :mod:grounded_decision.examples).

* ``examples`` lists every example id with its track and title.
* ``examples --track live`` lists only one track (``product``,
  ``step``, ``adaptive``, ``combo``, ``live`` or ``markets``).
* ``examples fraud_review`` runs one example and prints its report.
* ``examples fraud_review --json`` prints the same run as one line of
  JSON: ``{"id", "title", "records", "summary"}``.
* ``examples --json`` (no id) runs every listed example as JSON lines,
  which is handy for piping into ``jq`` or a regression diff.

learn HISTORY.jsonl --max-risk RATE [--confidence C] [--procedure P] [--output FILE] Learn the check settings from labelled history (see :mod:grounded_decision.learn). Each line of HISTORY.jsonl is one past decision: {"evidence": {...}, "proposed_action": "block", "action_was_correct": true}. Prints what was learned and how far to trust it, and writes the settings with their provenance to FILE. Exits with 2 if no setting could be certified.

play --list Every game (tic-tac-toe, snake, checkers, Chinese checkers), its agents and its default line-up.

play GAME --you [--you-play NAME] [--guarded] Take a seat yourself; with --guarded the game's gate checks your moves and overrides the unsafe ones, explaining why.

audit GAME [--agent NAME] [--opponent NAME] [--oracle NAME] [--oracle-depth N] [--seed N] [--size WxH] [--seat N] [--view history|json] [--no-regimes] [--regime-seeds N] [--world MODE] Replay one match and judge every decision of one agent against a stronger oracle (regret, agreement, the worst decisions with their boards, the gate's overrides), then print the game's information-regime table: what not seeing the future costs, and what insuring against the worst case costs (see :mod:grounded_decision.examples.games.audit).

record GAME [--agents A,B] [--world MODE] [--seed N] [--size WxH] [--max-steps N] [game rule options] -o FILE.gdr Play a match of agents to its end and save every frame as a .gdr replay (see :mod:grounded_decision.session.replay); the options and their validation are play's.

replay-info FILE.gdr [--verify] [--moves] [--bytes] [--json] Print a replay's header and check its frames against its content hash and every move arrow against the position before its move; --verify also records the header's spec again and checks the frames are identical; --moves prints every move with its arrows.

serve [--host H] [--port P] [--database FILE] [limits ...] Run our own service (:mod:grounded_decision.service; needs the service extra): the session protocol over WebSocket (/ws/) and HTTP (POST /api/), the reference decision service under /api/decisions and GET /healthz, with JSON logs on stdout. Every flag also reads an environment variable (--max-sessions: GD_SERVICE_MAX_SESSIONS).

new-game NAME [--root DIR] [--force] Write a new game's skeleton, games/NAME.py (the template game, a race to N, renamed) and tests/test_games_NAME.py, which runs the game-module contract on it (see :mod:grounded_decision.examples.games.scaffold and docs/contributing/add-a-game.md).

interfaces Print :func:grounded_decision.examples.catalog.interface_map as indented JSON: the decide contract, the six checks, the builders, router and compare helpers, and every example id.

--version Print the installed package version.

Examples are imported lazily, so --version and --help stay fast.

SUBCOMMAND_PARSERS `module-attribute` ¶

SUBCOMMAND_PARSERS: dict[str, ArgumentParser] = {}

The play and audit parsers, kept for game-specific --help.

EXIT_REPLAY_MISMATCH `module-attribute` ¶

EXIT_REPLAY_MISMATCH = 3

replay-info: the frames do not match the header's hash or a re-recording.

EXIT_INTERRUPTED `module-attribute` ¶

EXIT_INTERRUPTED = 130

128 + SIGINT, the shell convention for Ctrl-C.

EXIT_TERMINATED `module-attribute` ¶

EXIT_TERMINATED = 143

128 + SIGTERM.

RESTORE_TERMINAL `module-attribute` ¶

RESTORE_TERMINAL = '\x1b[0m\x1b[?25h\x1b[?1049l'

Reset attributes, show the cursor, leave the alternate screen.

build_parser ¶

build_parser() -> ArgumentParser

The argument parser for every subcommand described in the module docstring.

Source code in src/grounded_decision/cli.py

def build_parser() -> argparse.ArgumentParser:
    """The argument parser for every subcommand described in the module docstring."""

    parser = argparse.ArgumentParser(
        prog="grounded-decision",
        description="Grammar-driven grounded decision runtime",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=TOP_LEVEL_EXAMPLES,
    )
    parser.add_argument("--version", action="version", version=__version__)
    subcommands = parser.add_subparsers(dest="cmd")

    examples_parser = subcommands.add_parser(
        "examples",
        help="run the real-world example library",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=EXAMPLES_EXAMPLES,
    )
    examples_parser.add_argument(
        "name", nargs="?", help="example id, or omit to list/run all"
    )
    examples_parser.add_argument(
        "--json", action="store_true", help="emit JSON records"
    )
    examples_parser.add_argument(
        "--all",
        action="store_true",
        dest="run_all",
        help="run every listed example and print every report",
    )
    examples_parser.add_argument(
        "--track",
        choices=EXAMPLE_TRACKS,
        help="filter the catalog by track",
    )

    learn_parser = subcommands.add_parser(
        "learn", help="learn check settings from labelled history, with a guarantee"
    )
    learn_parser.add_argument("history", help="JSON Lines file of past decisions")
    learn_parser.add_argument(
        "--max-risk",
        type=float,
        required=True,
        help="largest acceptable error rate, e.g. 0.02",
    )
    learn_parser.add_argument(
        "--confidence", type=float, default=0.95, help="certificate confidence"
    )
    learn_parser.add_argument(
        "--measure",
        choices=["error_rate_among_acts", "false_act_rate"],
        default="error_rate_among_acts",
        help="which error rate --max-risk limits",
    )
    learn_parser.add_argument(
        "--learn-decision-line",
        action="store_true",
        help="also choose p_threshold from quantiles of the observed density",
    )
    learn_parser.add_argument(
        "--procedure",
        choices=["fixed_sequence", "bonferroni"],
        default="fixed_sequence",
        help="multiple-testing procedure for certifying settings",
    )
    learn_parser.add_argument("--seed", type=int, default=0, help="split seed")
    learn_parser.add_argument(
        "--output", help="write the learned settings and provenance here (JSON)"
    )

    SUBCOMMAND_PARSERS["play"] = _add_play_parser(subcommands)
    SUBCOMMAND_PARSERS["audit"] = _add_audit_parser(subcommands)
    SUBCOMMAND_PARSERS["compare"] = _add_compare_parser(subcommands)
    SUBCOMMAND_PARSERS["record"] = _add_record_parser(subcommands)
    _add_replay_info_parser(subcommands)
    _add_serve_parser(subcommands)
    _add_new_game_parser(subcommands)

    subcommands.add_parser("demo", help="run a short clinical-triage walkthrough")
    subcommands.add_parser("interfaces", help="print the public interface map")
    return parser

record_spec ¶

record_spec(arguments: Namespace, seed: int) -> Any

The :class:~grounded_decision.session.SessionSpec that record's options name.

Raises:

Type	Description
`PlayError`	Any option the game does not accept (`play`'s message).

Source code in src/grounded_decision/cli.py

def record_spec(arguments: argparse.Namespace, seed: int) -> Any:
    """The :class:`~grounded_decision.session.SessionSpec` that ``record``'s options name.

    Raises:
        PlayError: Any option the game does not accept (``play``'s message).
    """

    from grounded_decision.examples.games import play
    from grounded_decision.session import SessionSpec

    return SessionSpec(
        game=arguments.game,
        agents=tuple(play.play_names(arguments.agents) or ()),
        world=arguments.world,
        seed=seed,
        size=arguments.size,
        players=arguments.players,
        until_length=arguments.until_length,
        rules=_rule_options(arguments),
        max_steps=arguments.max_steps,
    )

main ¶

main(argv: list[str] | None = None) -> int

Parse argv (default: sys.argv[1:]) and run the subcommand.

Every subcommand runs inside one wrapper, so the terminal behaves: Ctrl-C prints Interrupted. (no traceback), restores the terminal and exits 130; a closed pipe (... | head) exits quietly; SIGTERM runs every cleanup (finally) and exits 143.

Parameters:

Name	Type	Description	Default
`argv`	`list[str] \| None`	Command-line arguments without the program name.	`None`

Returns:

Type	Description
`int`	The process exit code: 0 on success, 1 for an unknown subcommand,
`int`	2 for bad arguments or when `learn` could not certify any
`int`	setting, 130 after Ctrl-C.

Source code in src/grounded_decision/cli.py

def main(argv: list[str] | None = None) -> int:
    """Parse ``argv`` (default: ``sys.argv[1:]``) and run the subcommand.

    Every subcommand runs inside one wrapper, so the terminal behaves:
    Ctrl-C prints ``Interrupted.`` (no traceback), restores the terminal
    and exits 130; a closed pipe (``... | head``) exits quietly; SIGTERM
    runs every cleanup (``finally``) and exits 143.

    Args:
        argv: Command-line arguments without the program name.

    Returns:
        The process exit code: 0 on success, 1 for an unknown subcommand,
        2 for bad arguments or when ``learn`` could not certify any
        setting, 130 after Ctrl-C.
    """

    previous_handler: Any = None
    try:
        previous_handler = signal.signal(signal.SIGTERM, _raise_on_sigterm)
    except ValueError:  # not the main thread
        previous_handler = None
    try:
        return _dispatch(argv)
    except KeyboardInterrupt:
        _restore_terminal()
        print("\nInterrupted.", file=sys.stderr)
        return EXIT_INTERRUPTED
    except BrokenPipeError:
        _silence_stdout()
        return EXIT_SUCCESS
    finally:
        if previous_handler is not None:
            try:
                signal.signal(signal.SIGTERM, previous_handler)
            except ValueError:
                pass

Examples and CLI¶

catalog ¶

INTERFACE_MAP_VERSION module-attribute ¶

TRACKS module-attribute ¶

ExampleRunner ¶

CatalogEntry ¶

list_examples ¶

run_example ¶

run_all ¶

interface_map ¶

cli ¶

Subcommands¶

SUBCOMMAND_PARSERS module-attribute ¶

EXIT_REPLAY_MISMATCH module-attribute ¶

EXIT_INTERRUPTED module-attribute ¶

EXIT_TERMINATED module-attribute ¶

RESTORE_TERMINAL module-attribute ¶

build_parser ¶

record_spec ¶

main ¶

INTERFACE_MAP_VERSION `module-attribute` ¶

TRACKS `module-attribute` ¶

SUBCOMMAND_PARSERS `module-attribute` ¶

EXIT_REPLAY_MISMATCH `module-attribute` ¶

EXIT_INTERRUPTED `module-attribute` ¶

EXIT_TERMINATED `module-attribute` ¶

RESTORE_TERMINAL `module-attribute` ¶