diff --git a/src/fi/simulate/agent/definition.py b/src/fi/simulate/agent/definition.py index ffc3263..de795c0 100644 --- a/src/fi/simulate/agent/definition.py +++ b/src/fi/simulate/agent/definition.py @@ -108,6 +108,16 @@ class TelephonyTransport(BaseModel): None, description="Dispatch rule that routes the inbound call (sip_inbound only).", ) + sip_inbound_trunk_id: Optional[str] = Field( + None, + description=( + "LiveKit inbound SIP trunk the leased dispatch rule belongs to " + "(sip_inbound only). When set, the SDK validates the named rule binds " + "this trunk and skips the LIVEKIT_INBOUND_TRUNK_ID env fallback — this " + "is how a run pins the carrier's (e.g. Telnyx) pool trunk instead of " + "inheriting the worker's default inbound trunk." + ), + ) readiness_timeout_seconds: Optional[float] = Field( None, gt=0, @@ -128,6 +138,8 @@ def _check_kind_fields(self) -> "TelephonyTransport": if self.kind == "sip_outbound": if self.inbound_call_originator is not None: raise ValueError("sip_outbound cannot set inbound_call_originator") + if self.sip_inbound_trunk_id is not None: + raise ValueError("sip_outbound cannot set sip_inbound_trunk_id") if not self.sip_trunk_id or not self.sip_trunk_id.strip(): raise ValueError("sip_outbound requires sip_trunk_id") if not self.sip_call_to or not _E164.match(self.sip_call_to): @@ -153,6 +165,7 @@ def _check_kind_fields(self) -> "TelephonyTransport": self.sip_call_to, self.sip_number, self.dispatch_rule_name, + self.sip_inbound_trunk_id, self.inbound_call_originator, ] ): diff --git a/src/fi/simulate/endpoints/profiles.py b/src/fi/simulate/endpoints/profiles.py index a853f54..6dd15b3 100644 --- a/src/fi/simulate/endpoints/profiles.py +++ b/src/fi/simulate/endpoints/profiles.py @@ -139,7 +139,13 @@ def _sip_inbound_required_env(agent_definition: Any) -> list[str]: transport = agent_definition.transport target = agent_definition.target names: list[str] = [] - if transport is not None and not transport.dispatch_rule_name: + if ( + transport is not None + and not transport.dispatch_rule_name + and not transport.sip_inbound_trunk_id + ): + # Only needed when the SDK must self-provision a rule: with a named rule + # or an explicit inbound trunk id, the trunk is already pinned. names.append("LIVEKIT_INBOUND_TRUNK_ID") if transport is not None and transport.inbound_call_originator == "vapi": names.extend( diff --git a/src/fi/simulate/simulation/engines/livekit.py b/src/fi/simulate/simulation/engines/livekit.py index 4e3abbc..466f0d3 100644 --- a/src/fi/simulate/simulation/engines/livekit.py +++ b/src/fi/simulate/simulation/engines/livekit.py @@ -2289,6 +2289,10 @@ async def _ensure_sip_inbound_dispatch( """ existing = await api_client.sip.list_sip_dispatch_rule(ListSIPDispatchRuleRequest()) + # An explicit inbound trunk (e.g. the leased carrier pool trunk) pins the + # carrier; it takes precedence over the LIVEKIT_INBOUND_TRUNK_ID env, which + # is only a fallback for a self-provisioned rule. + explicit_trunk = transport.sip_inbound_trunk_id if transport.dispatch_rule_name: for rule in existing.items: if rule.name != transport.dispatch_rule_name: @@ -2307,12 +2311,21 @@ async def _ensure_sip_inbound_dispatch( "sip_inbound_rule_mismatch: " f"{transport.dispatch_rule_name} targets a different room" ) + # When the caller pins a trunk, the reused rule must belong to it — + # otherwise a stale/other-carrier rule of the same name would route + # the call onto the wrong (e.g. Twilio) trunk. + if explicit_trunk and explicit_trunk not in list(rule.trunk_ids): + raise RuntimeError( + "sip_inbound_rule_trunk_mismatch: " + f"{transport.dispatch_rule_name} is not bound to {explicit_trunk}" + ) return rule.sip_dispatch_rule_id, False raise RuntimeError(f"sip_inbound_rule_missing: {transport.dispatch_rule_name}") - trunk_id = os.environ.get(_LIVEKIT_INBOUND_TRUNK_ENV) + trunk_id = explicit_trunk or os.environ.get(_LIVEKIT_INBOUND_TRUNK_ENV) if not trunk_id: raise RuntimeError( - f"sip_inbound_trunk_missing: set {_LIVEKIT_INBOUND_TRUNK_ENV}" + "sip_inbound_trunk_missing: set transport.sip_inbound_trunk_id or " + f"{_LIVEKIT_INBOUND_TRUNK_ENV}" ) for rule in existing.items: if trunk_id and trunk_id in rule.trunk_ids: diff --git a/src/fi/simulate/simulation/matrix.py b/src/fi/simulate/simulation/matrix.py index 12f6394..31954d7 100644 --- a/src/fi/simulate/simulation/matrix.py +++ b/src/fi/simulate/simulation/matrix.py @@ -64,8 +64,18 @@ def _leg_manifest(base: Mapping[str, Any], leg: MatrixLeg) -> dict[str, Any]: transport = dict(agent_definition.get("transport") or {}) transport["kind"] = leg.channel if leg.channel == "webrtc": - for legacy in ("sip_trunk_id", "sip_number", "sip_call_to", "dispatch_rule_name"): - transport.pop(legacy, None) + for sip_only in ( + "sip_trunk_id", + "sip_number", + "sip_call_to", + "participant_identity", + "dispatch_rule_name", + "sip_inbound_trunk_id", + "readiness_timeout_seconds", + "answer_timeout_seconds", + "inbound_call_originator", + ): + transport.pop(sip_only, None) agent_definition["transport"] = transport if leg.provider_evidence_overrides is not None: agent_definition["provider_evidence"] = leg.provider_evidence_overrides diff --git a/tests/runtime/test_livekit_engine.py b/tests/runtime/test_livekit_engine.py index 5e2d8cd..8332574 100644 --- a/tests/runtime/test_livekit_engine.py +++ b/tests/runtime/test_livekit_engine.py @@ -1615,6 +1615,76 @@ def open_conversation(self): assert result.metadata["sip_dispatch_rule_created"] is False +def test_sip_inbound_named_rule_with_pinned_trunk_is_reused() -> None: + created = [] + + class _Sip: + async def list_sip_dispatch_rule(self, _request): + return SimpleNamespace( + items=[ + SimpleNamespace( + name="telnyx-rule", + sip_dispatch_rule_id="SD_telnyx", + trunk_ids=["ST_telnyx"], + rule=SimpleNamespace( + dispatch_rule_direct=SimpleNamespace(room_name="room-1") + ), + ) + ] + ) + + async def create_sip_dispatch_rule(self, _request): + created.append(True) + raise AssertionError("a matching pinned rule must be reused") + + transport = livekit.TelephonyTransport( + kind="sip_inbound", + dispatch_rule_name="telnyx-rule", + sip_inbound_trunk_id="ST_telnyx", + ) + result = asyncio.run( + livekit._ensure_sip_inbound_dispatch( + SimpleNamespace(sip=_Sip()), + transport=transport, + room_name="room-1", + ) + ) + + assert result == ("SD_telnyx", False) + assert created == [] + + +def test_sip_inbound_named_rule_rejects_mismatched_pinned_trunk() -> None: + class _Sip: + async def list_sip_dispatch_rule(self, _request): + return SimpleNamespace( + items=[ + SimpleNamespace( + name="telnyx-rule", + sip_dispatch_rule_id="SD_twilio", + trunk_ids=["ST_twilio"], + rule=SimpleNamespace( + dispatch_rule_direct=SimpleNamespace(room_name="room-1") + ), + ) + ] + ) + + transport = livekit.TelephonyTransport( + kind="sip_inbound", + dispatch_rule_name="telnyx-rule", + sip_inbound_trunk_id="ST_telnyx", + ) + with pytest.raises(RuntimeError, match="sip_inbound_rule_trunk_mismatch"): + asyncio.run( + livekit._ensure_sip_inbound_dispatch( + SimpleNamespace(sip=_Sip()), + transport=transport, + room_name="room-1", + ) + ) + + def test_cleanup_logging_redacts_exception_details(caplog) -> None: secret = "-".join(("provider", "secret", "value")) errors = [] @@ -1898,9 +1968,7 @@ def test_case_crash_yields_dense_failed_result_without_shifting_order( statuses = [r.metadata["status"] for r in report.results] assert statuses[2] == CaseStatus.FAILED.value assert report.results[2].metadata["failure"]["code"] == "case_execution_error" - assert all( - statuses[i] == CaseStatus.COMPLETED.value for i in (0, 1, 3, 4) - ) + assert all(statuses[i] == CaseStatus.COMPLETED.value for i in (0, 1, 3, 4)) def test_on_case_complete_streams_every_index_including_failed_slot( @@ -1969,7 +2037,9 @@ def test_dispatch_metadata_empty_by_default(): # A real target agent flips to outbound/no-greet on any dispatch metadata, # so the default must be an empty string (not our simulation context). assert livekit._dispatch_metadata_json(_agent()) == "" - assert livekit._dispatch_metadata_json(SimpleNamespace(dispatch_metadata=None)) == "" + assert ( + livekit._dispatch_metadata_json(SimpleNamespace(dispatch_metadata=None)) == "" + ) assert livekit._dispatch_metadata_json(SimpleNamespace(dispatch_metadata={})) == "" diff --git a/tests/runtime/test_manifest_engine_dispatch.py b/tests/runtime/test_manifest_engine_dispatch.py index e8ed69e..b729381 100644 --- a/tests/runtime/test_manifest_engine_dispatch.py +++ b/tests/runtime/test_manifest_engine_dispatch.py @@ -7,7 +7,9 @@ import pytest from fi.simulate import cli +from fi.simulate.agent.definition import AgentDefinition from fi.simulate.manifest import ManifestError, run_manifest_file +from fi.simulate.simulation.matrix import MatrixLeg, _leg_manifest from fi.simulate.simulation.models import ( Persona, TestCaseResult as CaseResult, @@ -389,6 +391,39 @@ async def run_test(self, **kwargs): assert captured["agent_definition"].transport.dispatch_rule_name is None +def test_matrix_webrtc_leg_strips_sip_fields_and_validates() -> None: + base = { + "agent_definition": { + "name": "phone-agent", + "url": "ws://127.0.0.1:7880", + "room_name": "sdk-{test_case_id}", + "system_prompt": "Help.", + "transport": { + "kind": "sip_inbound", + "sip_trunk_id": "ST_outbound", + "sip_number": "+12068956991", + "sip_call_to": "+14155551234", + "participant_identity": "sip-{test_case_id}", + "dispatch_rule_name": "inbound-rule", + "sip_inbound_trunk_id": "ST_telnyx", + "readiness_timeout_seconds": 30, + "answer_timeout_seconds": 30, + "inbound_call_originator": "vapi", + }, + } + } + + manifest = _leg_manifest( + base, + MatrixLeg(provider="livekit", channel="webrtc"), + ) + transport = manifest["agent_definition"]["transport"] + + assert transport == {"kind": "webrtc"} + validated = AgentDefinition.model_validate(manifest["agent_definition"]) + assert validated.transport.kind == "webrtc" + + def test_livekit_manifest_rejects_sip_inbound_empty_dispatch_rule( tmp_path: Path, ) -> None: