feat(simulate): sip_inbound_trunk_id transport field for carrier pinning

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,
                 ]
             ):

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(

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:

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

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={})) == ""
 
 

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: