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`	Run one `db_state_assertion` against the pulled DB and return a verdict.
`list_registered_predicates`	Diagnostic helper. Returns `{domain: [func_names]}`.
`register_db_state_predicate`	Decorator: register a predicate function under `(domain, fn.__name__)`.

Data

ALL_DB_STATE_PREDICATES

Predicate

API

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

str

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

assertion

Dict[str, Any]

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.

Optional[Dict[str, Any]]

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

Optional[Dict[str, Any]]

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

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.

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

str

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

Raises:

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

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

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