Cross-side sync-delta applier registry (bot-side).

This module is the bot-side endpoint of the cross-side state-propagation pipeline introduced for dual-side scenarios (telecom and any future domain that opts into Scenario.sync_state).

The pipeline at a glance:

A write tool on bot A fires and calls WriteScenarioTool._record_action, which (in addition to appending to shared_state["actions"]) emits an action-applied RTVI server message.
The bridge picks that up, replays the action onto an in-process shadow copy of both DBs (using the scenario’s tool map + each tool’s sync invoke method), then calls scenario.sync_state(agent_db, user_db).
sync_state returns a per-side delta dict describing the cross-side field changes that must land on the OTHER bot. The bridge dispatches each delta via the apply_sync_delta RTVI action.
The receiving bot’s handler calls into apply_sync_delta(domain, db, delta) in this module to mutate its own shared_state["db"].

The default applier handles “dotted path → value” deltas (e.g. "surroundings.payment_request": {...}) and is sufficient for any domain whose cross-side propagation is pure field assignment. Domains needing more (list-by-id lookups, post-apply re-derivation hooks, domain-specific validation) register a per-domain applier via @register_sync_applier(domain="...").

This is structurally parallel to initialization_functions and db_state_predicates — same registry-by-domain pattern, same opt-in shape.

Module Contents

Functions

Name	Description
`_default_sync_applier`	Generic dotted-path field setter.
`apply_sync_delta`	Dispatch a sync delta to the registered applier for `domain`.
`register_sync_applier`	Decorator: register a domain-specific applier.

Data

SYNC_APPLIERS

API

nemo_voice_agent.evaluation.sync_appliers._default_sync_applier(
    db: dict,
    delta: dict
) -> None

Generic dotted-path field setter.

Each delta key is a dotted path into db; the corresponding value is assigned at that path. The path components are followed verbatim — no list-index or by-id-match support. Domains needing those (e.g. bills[B1002].status) must register their own applier.

Example::

delta = {“surroundings.payment_request”: {“bill_id”: “B1002”, …}}

→ db[“surroundings”][“payment_request”] = {“bill_id”: “B1002”, …}

Raises:

KeyError: if an intermediate path component doesn’t exist in db.

nemo_voice_agent.evaluation.sync_appliers.apply_sync_delta(
    domain: str,
    db: dict,
    delta: dict
) -> None

Dispatch a sync delta to the registered applier for domain.

Falls back to _default_sync_applier (dotted-path field set) when no per-domain applier exists. The default suffices for any future domain whose sync deltas are pure field assignments.

Parameters:

domain

str

Scenario.domain value of the active scenario.

dict

The bot’s live shared_state["db"] (mutated in place).

delta

dict

Cross-side delta from the bridge — shape is domain-defined.

nemo_voice_agent.evaluation.sync_appliers.register_sync_applier(
    domain: str
)

Decorator: register a domain-specific applier.

The applier signature is func(db: dict, delta: dict) -> None and must mutate db in place. The exact shape of delta is a contract between the scenario’s sync_state (which produces it) and the applier (which consumes it) — the bridge transports it verbatim.

Usage::

@register_sync_applier(domain=“tau2_telecom”) def apply_telecom_sync_delta(db: dict, delta: dict) -> None: …

nemo_voice_agent.evaluation.sync_appliers.SYNC_APPLIERS: Dict[str, Callable[[dict, dict], None]] = {}