fix(core): allow RPC handler registration after connect

livekit-portal/src/portal.rs

@@ -301,6 +301,15 @@ pub struct Portal {
     // attribute-change and participant-connect events can update operators,
     // robot_identity, and active_operator without taking a Portal-level lock.
     controller: Arc<ControllerState>,
+
+    // Handle to the tokio runtime `connect()` ran on. Captured because
+    // `register_rpc_method` can be called from a foreign thread with no
+    // runtime context (e.g. a binding's asyncio loop); registering an RPC
+    // method on a live LocalParticipant triggers publisher negotiation,
+    // which `tokio::spawn`s and would otherwise panic with "no reactor
+    // running". We enter this handle around that call. `None` before the
+    // first connect.
+    runtime_handle: Mutex<Option<tokio::runtime::Handle>>,
 }
 
 impl Portal {
@@ -379,11 +388,15 @@ impl Portal {
             rpc_handlers: Arc::new(Mutex::new(HashMap::new())),
             local_participant: Arc::new(Mutex::new(None)),
             controller: Arc::new(ControllerState::new()),
+            runtime_handle: Mutex::new(None),
         }
     }
 
     pub async fn connect(&self, url: &str, token: &str) -> PortalResult<()> {
         let _lifecycle = self.lifecycle.lock().await;
+        // Capture the runtime we're running on so `register_rpc_method` can
+        // enter it when called later from a non-runtime (foreign) thread.
+        *self.runtime_handle.lock() = Some(tokio::runtime::Handle::current());
         if self.conn.lock().room.is_some() {
             return Err(PortalError::AlreadyConnected);
         }
@@ -852,7 +865,20 @@ impl Portal {
             map.insert(method.to_string(), handler.clone());
         }
         if let Some(lp) = self.local_participant.lock().clone() {
-            register_handler_on(&lp, method.to_string(), handler);
+            // The SDK's `register_rpc_method` kicks off publisher negotiation,
+            // which `tokio::spawn`s internally. If we were called from a
+            // foreign thread with no runtime context (a binding's asyncio
+            // loop), that spawn panics. Enter the runtime `connect()` ran on
+            // so the spawn lands on it. The handle is always present here
+            // (an LP exists only after a successful connect set it), but fall
+            // back to a bare call rather than panicking if it somehow isn't.
+            match self.runtime_handle.lock().clone() {
+                Some(handle) => {
+                    let _guard = handle.enter();
+                    register_handler_on(&lp, method.to_string(), handler);
+                }
+                None => register_handler_on(&lp, method.to_string(), handler),
+            }
         }
     }
 

python/packages/livekit-portal/tests/integration/conftest.py

@@ -44,6 +44,7 @@
         "test_multi_operator.py",
         "test_action_subscription.py",
         "test_webrtc_codecs.py",
+        "test_rpc.py",
     ]
 )
 

python/packages/livekit-portal/tests/integration/test_rpc.py

@@ -0,0 +1,99 @@
+"""RPC integration tests.
+
+Covers handler registration both before and after `connect()`, plus error
+propagation. The post-connect case is the regression guard: registering an
+RPC handler on a live participant triggers SDK publisher negotiation, which
+spawns onto the tokio runtime. When `register_rpc_method` is called from a
+binding's asyncio thread (no runtime context), that spawn used to panic with
+"there is no reactor running". The core now enters the runtime captured at
+connect time around the registration.
+
+Skipped automatically without `LIVEKIT_URL` (see conftest).
+"""
+from __future__ import annotations
+
+import asyncio
+import os
+
+import pytest
+
+from integration.conftest import URL, _make_token
+from livekit.portal import (
+    DType,
+    Operator,
+    OperatorConfig,
+    PortalError,
+    Robot,
+    RobotConfig,
+    RpcError,
+)
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_rpc_register_after_connect(pair):
+    """Regression: registering a handler AFTER connect must not panic and the
+    method must be invocable from the peer."""
+    await pair.start()
+
+    async def handler(data):
+        return f"pong:{data.payload}"
+
+    pair.robot.register_rpc_method("ping", handler)
+    await asyncio.sleep(0.1)
+
+    result = await pair.operator.perform_rpc(
+        "ping", "hi", destination=pair.robot.local_identity()
+    )
+    assert result == "pong:hi"
+
+
+async def test_rpc_register_before_connect():
+    """The pre-connect path keeps working: handlers registered before connect
+    are applied on connect and invocable afterwards. Built manually rather
+    than via the `pair` fixture so the robot exists before `connect()`.
+    """
+    room = f"rpc-pre-{os.urandom(4).hex()}"
+    robot_cfg = RobotConfig(room)
+    robot_cfg.add_state_typed([("j", DType.F32)])
+    operator_cfg = OperatorConfig(room)
+    operator_cfg.add_state_typed([("j", DType.F32)])
+    robot = Robot(robot_cfg)
+    operator = Operator(operator_cfg)
+
+    async def handler(data):
+        return f"echo:{data.payload}"
+
+    robot.register_rpc_method("echo", handler)  # before connect
+
+    try:
+        await robot.connect(URL, _make_token("robot", room))
+        await asyncio.sleep(0.2)
+        await operator.connect(URL, _make_token("operator", room))
+        await asyncio.sleep(0.2)
+        result = await operator.perform_rpc(
+            "echo", "x", destination=robot.local_identity()
+        )
+        assert result == "echo:x"
+    finally:
+        for side in (operator, robot):
+            try:
+                await side.disconnect()
+            except Exception:  # noqa: BLE001
+                pass
+
+
+async def test_rpc_handler_error_propagates(pair):
+    """An application error raised by the handler surfaces on the caller."""
+    await pair.start()
+
+    async def handler(data):
+        raise RpcError.Error(code=1234, message="boom", data=None)
+
+    pair.robot.register_rpc_method("fail", handler)
+    await asyncio.sleep(0.1)
+
+    with pytest.raises(PortalError):
+        await pair.operator.perform_rpc(
+            "fail", "", destination=pair.robot.local_identity()
+        )