fix(plugin): validate runtime timeout before startup

src/langbot/pkg/api/http/controller/groups/plugins.py

@@ -15,6 +15,7 @@ import posixpath
 import sqlalchemy
 
 from .....core import taskmgr
+from .....core.task_boundary import run_in_workspace_uow
 from .....entity.persistence import plugin as persistence_plugin
 from ...authz import Permission
 from ...context import ExecutionContext, RequestContext
@@ -310,13 +311,11 @@ class PluginsRouterGroup(group.RouterGroup):
     ):
         """Revalidate a captured task context immediately before Runtime I/O."""
 
-        persistence_mgr = getattr(self.ap, 'persistence_mgr', None)
-        tenant_scope = getattr(persistence_mgr, 'tenant_scope', None)
-        if callable(tenant_scope):
-            async with tenant_scope(execution_context.workspace_uuid):
-                await self.ap.plugin_connector.require_workspace_context(execution_context)
-                return await operation()
-        await self.ap.plugin_connector.require_workspace_context(execution_context)
+        await run_in_workspace_uow(
+            self.ap,
+            execution_context.workspace_uuid,
+            lambda: self.ap.plugin_connector.require_workspace_context(execution_context),
+        )
         return await operation()
 
     async def _require_authenticated_plugin_runtime_context(

src/langbot/pkg/plugin/connector.py

@@ -841,6 +841,8 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
             runtime_id=self._runtime_id,
         )
         self.worker_policy = self._load_worker_policy()
+        plugin_config = self.ap.instance_config.data.get('plugin', {})
+        connect_timeout_seconds = self._runtime_connect_timeout(plugin_config)
 
         async with self._lifecycle_lock:
             if self._closing:
@@ -981,7 +983,6 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
                 task_coro = self.ctrl.run(new_connection_callback)
 
             self._transport_task = asyncio.create_task(task_coro)
-            connect_timeout_seconds = self._runtime_connect_timeout(self.ap.instance_config.data.get('plugin', {}))
             try:
                 await asyncio.wait_for(self._connected.wait(), timeout=connect_timeout_seconds)
             except asyncio.TimeoutError as exc:

tests/unit_tests/api/test_plugin_runtime_route_fence.py

@@ -124,37 +124,24 @@ async def test_background_plugin_operation_refences_captured_generation(plugin_r
 
 
 @pytest.mark.asyncio
-async def test_background_plugin_operation_revalidates_and_runs_inside_tenant_uow(plugin_router_cls):
+async def test_background_plugin_operation_revalidates_inside_short_tenant_uow(plugin_router_cls):
     scopes = []
-    active_scope = None
-
-    transaction_active = False
 
     @asynccontextmanager
-    async def tenant_scope(workspace_uuid):
-        nonlocal active_scope
+    async def tenant_uow(workspace_uuid):
         scopes.append(workspace_uuid)
-        active_scope = workspace_uuid
-        try:
-            yield
-        finally:
-            active_scope = None
+        yield
 
     connector = SimpleNamespace(
         require_workspace_context=AsyncMock(side_effect=lambda context: context),
     )
-
-    async def operation():
-        assert active_scope == CONTEXT.workspace_uuid
-        assert transaction_active is False
-        return 'done'
-
+    operation = AsyncMock(return_value='done')
     router = object.__new__(plugin_router_cls)
     router.ap = SimpleNamespace(
         plugin_connector=connector,
         persistence_mgr=SimpleNamespace(
             mode=SimpleNamespace(value='cloud_runtime'),
-            tenant_scope=tenant_scope,
+            tenant_uow=tenant_uow,
         ),
     )
 
@@ -163,3 +150,4 @@ async def test_background_plugin_operation_revalidates_and_runs_inside_tenant_uo
     assert result == 'done'
     assert scopes == [CONTEXT.workspace_uuid]
     connector.require_workspace_context.assert_awaited_once_with(CONTEXT)
+    operation.assert_awaited_once()

tests/unit_tests/plugin/test_connector_ping.py

@@ -132,6 +132,49 @@ async def test_stdio_runtime_connection_does_not_capture_unconsumed_stderr(
     await connector.aclose()
 
 
+@pytest.mark.asyncio
+async def test_invalid_connect_timeout_is_rejected_before_transport_startup(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    connector = make_connector()
+    connector.ap.instance_config.data['plugin']['connect_timeout_seconds'] = 0
+    stdio_controller = Mock()
+    websocket_controller = Mock()
+    create_task = Mock()
+    get_platform = Mock(return_value='linux')
+    use_websocket = Mock(return_value=False)
+    connector._start_runtime_subprocess = AsyncMock()
+    monkeypatch.setattr(connector_module.constants, 'instance_id', 'instance-a')
+    monkeypatch.setattr(connector_module.asyncio, 'create_task', create_task)
+    monkeypatch.setattr(connector_module.platform, 'get_platform', get_platform)
+    monkeypatch.setattr(
+        connector_module.platform,
+        'use_websocket_to_connect_plugin_runtime',
+        use_websocket,
+    )
+    monkeypatch.setattr(
+        connector_module.stdio_client_controller,
+        'StdioClientController',
+        stdio_controller,
+    )
+    monkeypatch.setattr(
+        connector_module.ws_client_controller,
+        'WebSocketClientController',
+        websocket_controller,
+    )
+
+    with pytest.raises(ValueError, match='plugin.connect_timeout_seconds'):
+        await connector.initialize()
+
+    get_platform.assert_not_called()
+    use_websocket.assert_not_called()
+    stdio_controller.assert_not_called()
+    websocket_controller.assert_not_called()
+    connector._start_runtime_subprocess.assert_not_awaited()
+    create_task.assert_not_called()
+    assert connector._transport_task is None
+
+
 @pytest.mark.asyncio
 async def test_runtime_disconnect_notifies_once_and_clears_handler(
     monkeypatch: pytest.MonkeyPatch,