nemo_voice_agent.evaluation.tools.tau2_telecom_sync

Tau2-telecom cross-side state-propagation pipeline.

Two pieces:

sync_telecom_state(agent_db, user_db) — pure function (no I/O) that mirrors upstream’s sync_tools. It takes both DBs in dict form, mutates them in place where needed, and returns a per-side delta dict describing the field changes the bridge must push to each bot so their live state catches up.
apply_telecom_sync_delta(db, delta) — the registered bot-side applier (consumed by apply_sync_delta in nemo_voice_agent.evaluation.sync_appliers). Walks a telecom delta, handling list-by-id paths ("bills[B1002].status") that the default dotted-path applier can’t, and triggers _simulate_network_search if any surroundings.* field changed.

Both pieces are imported (with side effects — the applier registers itself on import) from nemo_voice_agent/evaluation/tools/__init__.py.

Propagation paths covered (matching upstream TelecomEnvironment.sync_tools):

Trigger (agent side)	Becomes (user side)
`line.status` changes	`surroundings.line_active`
`line.roaming_enabled` changes	`surroundings.roaming_allowed`
`line.data_used_gb` / `data_refueling_gb` / `plan.data_limit_gb`	`surroundings.mobile_data_usage_exceeded`
Any `bill.status == AWAITING_PAYMENT`	`surroundings.payment_request`
User: `payment_request.paid = True`	Agent: `bill.status = Paid` (reverse direction)

Module Contents

Functions

Name	Description
`_find_bill`	-
`_find_customer_by_phone`	-
`_find_line_by_phone`	-
`_find_plan`	-
`apply_telecom_sync_delta`	Apply a telecom sync delta to the bot’s live `shared_state["db"]`.
`sync_telecom_state`	Reconcile cross-side state after a write action fired on either side.

Data

SyncDeltas

_BILLS_PATH_RE

API

nemo_voice_agent.evaluation.tools.tau2_telecom_sync._find_bill(
    agent_db: dict,
    bill_id: str
) -> typing.Optional[dict]

nemo_voice_agent.evaluation.tools.tau2_telecom_sync._find_customer_by_phone(
    agent_db: dict,
    phone_number: str
) -> typing.Optional[dict]

nemo_voice_agent.evaluation.tools.tau2_telecom_sync._find_line_by_phone(
    agent_db: dict,
    phone_number: str
) -> typing.Optional[dict]

nemo_voice_agent.evaluation.tools.tau2_telecom_sync._find_plan(
    agent_db: dict,
    plan_id: str
) -> typing.Optional[dict]

nemo_voice_agent.evaluation.tools.tau2_telecom_sync.apply_telecom_sync_delta(
    db: dict,
    delta: dict
) -> None

Apply a telecom sync delta to the bot’s live shared_state["db"].

Handles two delta path shapes:

Dotted path (e.g. surroundings.payment_request): assigned at the corresponding nested dict location.
List-by-id (e.g. bills[B1002].status): finds the matching element of db["bills"] by bill_id and sets the field on it.

After applying, if any surroundings.* field changed, _simulate_network_search re-derives network_connection_status / network_technology_connected / network_signal_strength so the user-sim’s next check_network_status / run_speed_test / _get_mobile_data_working calls return values consistent with the new surroundings.

The default applier in sync_appliers.py would handle the dotted paths but NOT the bills[...] path and wouldn’t trigger the network-search re-derivation; hence the per-domain override.

nemo_voice_agent.evaluation.tools.tau2_telecom_sync.sync_telecom_state(
    agent_db: dict,
    user_db: dict
) -> nemo_voice_agent.evaluation.tools.tau2_telecom_sync.SyncDeltas

Reconcile cross-side state after a write action fired on either side.

Pure function — same inputs always produce the same output. No I/O, no logging. Called by the bridge between every action and the next-turn dispatch.

Parameters:

agent_db

dict

Agent-side TelecomDB dict (mutated in place for the user→agent direction: payment_request.paid → bill.status).

user_db

dict

User-side TelecomUserDB dict (mutated in place for the agent→user direction: surroundings.{line_active, roaming_allowed, mobile_data_usage_exceeded, payment_request}).

Returns: SyncDeltas

Per-side delta dict::

{“agent”: {<dotted-path or bills[id].field>: value, …}, “user”: {<dotted-path>: value, …}}

nemo_voice_agent.evaluation.tools.tau2_telecom_sync.SyncDeltas = Dict[str, Dict[str, Any]]

nemo_voice_agent.evaluation.tools.tau2_telecom_sync._BILLS_PATH_RE = re.compile('^bills\\[([^\\]]+)\\]\\.(.+)$')