> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.nvidia.com/nemo/labs-voice-agent/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.nvidia.com/nemo/labs-voice-agent/_mcp/server.

# nemo_voice_agent.evaluation.db_state_predicates

DB-state predicate registry + dispatcher for `Scenario.db_state_assertions`.

A *db-state predicate* is a deterministic function over the final DB state of a
scenario, used as a per-predicate scoring signal alongside `db_state_match`
(whole-DB hash equality) and `nl_assertions` (LLM-judged transcript predicates).
Concretely:

* Predicate signature: `(db: dict, **arguments) -> bool`. Pure: same DB →
  same bool. No I/O, no randomness. Predicates are *side-agnostic* — they
  don't know if they're checking the "agent DB" or "user DB", and they
  don't need to: the caller (runner) picks the right DB based on the
  assertion record's `side` field before invoking.
* Predicates are registered per `(domain, func_name)` only — no `side`.
  Function names are unique within a domain by upstream construction (in
  tau2 telecom, the 4 user-side and 2 agent-side assertion names are
  disjoint), so adding `side` to the registry key would namespace
  against a non-collision.
* Dispatch is runner-side via `evaluate_db_state_assertion(...)`. The bot
  never sees the predicates — it just ships the inline DB dicts back to the
  runner via the `get_scenario_summary` action with `include_db=True`.

Why runner-side and not bot-side:

* Predicates are pure; they belong in a shared module, not behind an RTVI call.
* Synthetic tests can call predicates directly on dict fixtures without
  standing up a fake RTVI bot.
* Verdict aggregation lives next to the existing `nl_assertion` and
  `db_state_match` aggregation in `runner.py` — uniform code path for
  all three scoring signals.

The mirror-symmetric `initialization_actions` surface goes the *other* way
(bot-side dispatch via the `apply_initialization_actions` RTVI action), because init
actions mutate the live DB through real toolkit methods. Predicates only read.

Upstream tau2-bench calls this surface `env_assertions`; we renamed to
`db_state_assertions` so the name parallels the existing `db_state_match`
metric (same artifact, finer granularity) and mirrors `nl_assertions` in
shape (per-predicate verdicts). Same upstream JSON shape, just renamed at the
Scenario field + metric layer.

## Module Contents

### Functions

| Name                                                                                                          | Description                                                              |
| ------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------ |
| [`evaluate_db_state_assertion`](#nemo_voice_agent-evaluation-db_state_predicates-evaluate_db_state_assertion) | Run one `db_state_assertion` against the pulled DB and return a verdict. |
| [`list_registered_predicates`](#nemo_voice_agent-evaluation-db_state_predicates-list_registered_predicates)   | Diagnostic helper. Returns `{domain: [func_names]}`.           |
| [`register_db_state_predicate`](#nemo_voice_agent-evaluation-db_state_predicates-register_db_state_predicate) | Decorator: register a predicate function under `(domain, fn.__name__)`.  |

### Data

[`ALL_DB_STATE_PREDICATES`](#nemo_voice_agent-evaluation-db_state_predicates-ALL_DB_STATE_PREDICATES)

[`Predicate`](#nemo_voice_agent-evaluation-db_state_predicates-Predicate)

### API

```python
nemo_voice_agent.evaluation.db_state_predicates.evaluate_db_state_assertion(
    domain: str,
    assertion: typing.Dict[str, typing.Any],
    db: typing.Optional[typing.Dict[str, typing.Any]],
    user_db: typing.Optional[typing.Dict[str, typing.Any]]
) -> typing.Dict[str, typing.Any]
```

Run one `db_state_assertion` against the pulled DB and return a verdict.

Verdict shape mirrors `nl_assertion_verdicts` so the runner can
aggregate both via the same code path::

\{
"func\_name": str,
"side": str,
"passed": bool,           # True iff predicate(db, \*\*arguments) == assert\_value
"expected": Any,          # the assertion's assert\_value
"actual": Optional\[bool], # what the predicate returned (None on error)
"message": Optional\[str], # upstream's optional human-readable label
"error": Optional\[str],   # set to the failure mode when predicate

# is missing or raises; passed=False then.

}

Never raises — always returns a verdict dict. A missing predicate or
raising predicate becomes `passed=False` with `error` set, so the
runner can keep iterating the assertion list and surface all failures at
once rather than aborting on the first.

**Parameters:**

**`domain`**

Scenario domain (e.g., `"tau2_telecom"`). Determines which
registry bucket to look up the predicate in.

---

**`assertion`**

One entry from `scenario.db_state_assertions`. Required keys:
`side`, `func_name`, `arguments`, `assert_value`. Optional:
`message`. `side` is used here only to pick which DB to pass to
the predicate — it is NOT part of the registry key.

---

**`db`**

The agent-side DB dict pulled back by the bridge
(`shared_state["db"]`). Used when `side=="agent"`. May
be `None` if the scenario only emits user-side assertions.

---

**`user_db`**

The user-side DB dict pulled back by the bridge
(`shared_state["user_db"]`). Used when `side=="user"`.

---

```python
nemo_voice_agent.evaluation.db_state_predicates.list_registered_predicates(
    domain: typing.Optional[str] = None
) -> typing.Dict[str, typing.Any]
```

Diagnostic helper. Returns `{domain: [func_names]}`.

With `domain=None` returns the whole registry; with a specific domain
returns only that subtree. Useful for test setup verification and
debugging missing-predicate errors.

```python
nemo_voice_agent.evaluation.db_state_predicates.register_db_state_predicate(
    domain: str
)
```

Decorator: register a predicate function under `(domain, fn.__name__)`.

Usage::

@register\_db\_state\_predicate(domain="tau2\_telecom")
def assert\_mobile\_data\_status(db: dict, expected\_status: bool) -> bool:
return \_get\_mobile\_data\_working(db) == expected\_status

The function name (`fn.__name__`) becomes the registry key — matching
the upstream `func_name` field in
`evaluation_criteria.env_assertions`. Renames at the source require
updating the upstream task JSON, so don't.

**No `side` parameter.** Predicate names are unique within a domain;
`side` is purely caller-side metadata (the runner uses it on each
assertion record to pick which pulled DB to pass — agent's `db` or
user's `user_db`). The predicate itself is side-agnostic.

**Parameters:**

**`domain`**

Registry namespace (matching `Scenario.domain`, e.g.
`"tau2_telecom"`).

---

**Raises:**

* `ValueError`: if a predicate with the same name is already registered
  in this domain.

```python
nemo_voice_agent.evaluation.db_state_predicates.ALL_DB_STATE_PREDICATES: Dict[str, Dict[str, Predicate]] = {}
```

```python
nemo_voice_agent.evaluation.db_state_predicates.Predicate = Callable[..., bool]
```