From c4bad508d278bc76590cfb6218a7d192d8593151 Mon Sep 17 00:00:00 2001 From: dadachann <185672915+dadachann@users.noreply.github.com> Date: Wed, 19 Aug 2026 15:44:03 +0000 Subject: [PATCH 1/6] fix(runtime): honor configured cold reconcile timeout --- src/langbot/pkg/plugin/connector.py | 12 ++++++++++-- src/langbot/pkg/plugin/handler.py | 4 +++- .../unit_tests/plugin/test_connector_reconcile.py | 15 ++++++++++++++- tests/unit_tests/plugin/test_handler.py | 12 ++++++++++++ 4 files changed, 39 insertions(+), 4 deletions(-) diff --git a/src/langbot/pkg/plugin/connector.py b/src/langbot/pkg/plugin/connector.py index 0279ab285..bcdf4fe8e 100644 --- a/src/langbot/pkg/plugin/connector.py +++ b/src/langbot/pkg/plugin/connector.py @@ -701,7 +701,11 @@ class PluginRuntimeConnector(ManagedRuntimeConnector): } self._known_desired_states.update({state.binding.installation_uuid: state for state in desired_states}) - result = await runtime_handler.reconcile_plugin_installations(tuple(self._known_desired_states.values())) + reconcile_timeout_seconds = max(300.0, self._runtime_connect_timeout(self.ap.instance_config.data.get("plugin", {}))) + result = await runtime_handler.reconcile_plugin_installations( + tuple(self._known_desired_states.values()), + timeout=reconcile_timeout_seconds, + ) await self._repair_reconcile_missing_artifacts(self._known_desired_states, result) self._record_reconcile_failures(self._known_desired_states, result) @@ -736,7 +740,11 @@ class PluginRuntimeConnector(ManagedRuntimeConnector): if state.binding.installation_uuid in all_states: raise ValueError('Duplicate plugin installation UUID across projected Workspaces') all_states[state.binding.installation_uuid] = state - result = await runtime_handler.reconcile_plugin_installations(tuple(all_states.values())) + reconcile_timeout_seconds = max(300.0, self._runtime_connect_timeout(self.ap.instance_config.data.get("plugin", {}))) + result = await runtime_handler.reconcile_plugin_installations( + tuple(all_states.values()), + timeout=reconcile_timeout_seconds, + ) await self._repair_reconcile_missing_artifacts(all_states, result) self._record_reconcile_failures(all_states, result) for installation_uuid, previous in tuple(self._known_desired_states.items()): diff --git a/src/langbot/pkg/plugin/handler.py b/src/langbot/pkg/plugin/handler.py index 8dae74f6f..8463750f0 100644 --- a/src/langbot/pkg/plugin/handler.py +++ b/src/langbot/pkg/plugin/handler.py @@ -1677,13 +1677,15 @@ class RuntimeConnectionHandler(handler.Handler): async def reconcile_plugin_installations( self, installations: tuple[PluginInstallationDesiredState, ...], + *, + timeout: float = 300, ) -> dict[str, Any]: request = ReconcilePluginInstallationsRequest(installations=installations) with self.installation_scope(None): return await self.call_action( LangBotToRuntimeAction.RECONCILE_PLUGIN_INSTALLATIONS, request.model_dump(), - timeout=300, + timeout=timeout, ) async def apply_plugin_installation( diff --git a/tests/unit_tests/plugin/test_connector_reconcile.py b/tests/unit_tests/plugin/test_connector_reconcile.py index 8759a537e..d4d9b6b5b 100644 --- a/tests/unit_tests/plugin/test_connector_reconcile.py +++ b/tests/unit_tests/plugin/test_connector_reconcile.py @@ -107,6 +107,19 @@ def shared_connector( return connector +@pytest.mark.asyncio +async def test_shared_reconcile_uses_configured_cold_start_timeout(): + binding = execution_binding("workspace-a") + setting = plugin_setting("01", "a" * 64) + connector = shared_connector([[binding]], {"workspace-a": [setting]}) + connector.ap.instance_config.data["plugin"]["connect_timeout_seconds"] = 900 + connector.handler = runtime_handler() + + await connector._prepare_connected_runtime() + + assert connector.handler.reconcile_plugin_installations.await_args.kwargs["timeout"] == 900 + + @pytest.mark.asyncio async def test_shared_reconnect_replays_two_workspaces_and_removes_missing_projection(): binding_a = execution_binding('workspace-a') @@ -150,7 +163,7 @@ async def test_empty_projected_workspaces_do_not_retain_installation_sets(): assert connector._workspace_installations == {} assert connector._known_desired_states == {} - connector.handler.reconcile_plugin_installations.assert_awaited_once_with(()) + connector.handler.reconcile_plugin_installations.assert_awaited_once_with((), timeout=300.0) @pytest.mark.asyncio diff --git a/tests/unit_tests/plugin/test_handler.py b/tests/unit_tests/plugin/test_handler.py index 952d59134..844dcdc12 100644 --- a/tests/unit_tests/plugin/test_handler.py +++ b/tests/unit_tests/plugin/test_handler.py @@ -81,6 +81,18 @@ async def test_reconcile_plugin_installations_allows_cloud_cold_start_to_finish( assert runtime_handler.call_action.await_args.kwargs['timeout'] == 300 +@pytest.mark.asyncio +async def test_reconcile_plugin_installations_accepts_configured_cold_start_timeout(): + runtime_handler = make_handler(SimpleNamespace()) + runtime_handler.call_action = AsyncMock(return_value={}) + binding = next(iter(runtime_handler._installation_bindings.values()))[0] + desired = PluginInstallationDesiredState(binding=binding, enabled=True) + + await runtime_handler.reconcile_plugin_installations((desired,), timeout=900) + + assert runtime_handler.call_action.await_args.kwargs["timeout"] == 900 + + class TestHandlerQueryVariables: """Tests for handler query variable logic.""" From c3fe312a43d965d24034da48f64ac598a7840861 Mon Sep 17 00:00:00 2001 From: dadachann <185672915+dadachann@users.noreply.github.com> Date: Wed, 19 Aug 2026 16:22:08 +0000 Subject: [PATCH 2/6] fix(cloud): refresh directory during plugin startup --- src/langbot/pkg/core/app.py | 6 +++--- src/langbot/pkg/core/stages/build_app.py | 10 ++++++++++ 2 files changed, 13 insertions(+), 3 deletions(-) diff --git a/src/langbot/pkg/core/app.py b/src/langbot/pkg/core/app.py index 4b82b3efc..3c1a74402 100644 --- a/src/langbot/pkg/core/app.py +++ b/src/langbot/pkg/core/app.py @@ -304,10 +304,10 @@ class Application: async def run(self): self.event_loop_monitor.start() try: - if self.directory_projection_service is not None: - self.task_mgr.create_task( + if self.directory_projection_service is not None and getattr(self, "directory_projection_task", None) is None: + self.directory_projection_task = self.task_mgr.create_task( self.directory_projection_service.run(), - name='cloud-directory-projection', + name="cloud-directory-projection", scopes=[core_entities.LifecycleControlScope.APPLICATION], ) if self.cloud_model_catalog_service is not None: diff --git a/src/langbot/pkg/core/stages/build_app.py b/src/langbot/pkg/core/stages/build_app.py index 14d533dcc..001696d45 100644 --- a/src/langbot/pkg/core/stages/build_app.py +++ b/src/langbot/pkg/core/stages/build_app.py @@ -292,6 +292,16 @@ class BuildAppStage(stage.BootingStage): async def runtime_disconnect_callback(connector: plugin_connector.PluginRuntimeConnector) -> None: connector.schedule_reconnect() + if ap.directory_projection_service is not None: + # Keep the projection fresh while shared Runtime cold restore runs. + # BuildApp initializes the connector before Application.run() starts + # its long-lived tasks, so start the single refresh task here. + ap.directory_projection_task = ap.task_mgr.create_task( + ap.directory_projection_service.run(), + name="cloud-directory-projection", + scopes=[core_entities.LifecycleControlScope.APPLICATION], + ) + plugin_connector_inst = plugin_connector.PluginRuntimeConnector(ap, runtime_disconnect_callback) try: await plugin_connector_inst.initialize() From 693c59b726f16895612d0eb12a4e5f1aa37044fc Mon Sep 17 00:00:00 2001 From: dadachann <185672915+dadachann@users.noreply.github.com> Date: Wed, 19 Aug 2026 16:44:36 +0000 Subject: [PATCH 3/6] fix(cloud): import lifecycle task scope --- src/langbot/pkg/core/stages/build_app.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/langbot/pkg/core/stages/build_app.py b/src/langbot/pkg/core/stages/build_app.py index 001696d45..83f3b10fd 100644 --- a/src/langbot/pkg/core/stages/build_app.py +++ b/src/langbot/pkg/core/stages/build_app.py @@ -1,6 +1,6 @@ from __future__ import annotations -from .. import stage, app +from .. import stage, app, entities as core_entities from ...utils import version, proxy, constants from ...pipeline import pool, controller, pipelinemgr from ...pipeline import aggregator as message_aggregator From f36542135ab07049801a03115047b7c109cce0de Mon Sep 17 00:00:00 2001 From: dadachann <185672915+dadachann@users.noreply.github.com> Date: Wed, 19 Aug 2026 17:13:06 +0000 Subject: [PATCH 4/6] fix(cloud): start API before runtime reconcile --- src/langbot/pkg/core/app.py | 19 +++++++++++++++++-- src/langbot/pkg/core/stages/build_app.py | 7 ------- tests/unit_tests/core/test_app_shutdown.py | 15 +++++++++++++++ 3 files changed, 32 insertions(+), 9 deletions(-) diff --git a/src/langbot/pkg/core/app.py b/src/langbot/pkg/core/app.py index 3c1a74402..493e37363 100644 --- a/src/langbot/pkg/core/app.py +++ b/src/langbot/pkg/core/app.py @@ -301,6 +301,22 @@ class Application: async def initialize(self): pass + async def _initialize_plugin_runtime(self) -> None: + try: + await self.plugin_connector.initialize() + except asyncio.CancelledError: + raise + except Exception as exc: + self.logger.warning(f'Plugin runtime unavailable during startup; reconnecting in background: {exc}') + self.plugin_connector.schedule_reconnect() + + def _start_plugin_runtime_initialization(self): + return self.task_mgr.create_task( + self._initialize_plugin_runtime(), + name='plugin-runtime-initialization', + scopes=[core_entities.LifecycleControlScope.APPLICATION], + ) + async def run(self): self.event_loop_monitor.start() try: @@ -322,8 +338,6 @@ class Application: name='cloud-manifest-refresh', scopes=[core_entities.LifecycleControlScope.APPLICATION], ) - await self.plugin_connector.initialize_plugins() - # 后续可能会允许动态重启其他任务 # 故为了防止程序在非 Ctrl-C 情况下退出，这里创建一个不会结束的协程 async def never_ending(): @@ -348,6 +362,7 @@ class Application: name='http-api-controller', scopes=[core_entities.LifecycleControlScope.APPLICATION], ) + self._start_plugin_runtime_initialization() # Telemetry instance heartbeat (startup + daily); respects # space.disable_telemetry via TelemetryManager.send(). diff --git a/src/langbot/pkg/core/stages/build_app.py b/src/langbot/pkg/core/stages/build_app.py index 83f3b10fd..fed6031fc 100644 --- a/src/langbot/pkg/core/stages/build_app.py +++ b/src/langbot/pkg/core/stages/build_app.py @@ -303,13 +303,6 @@ class BuildAppStage(stage.BootingStage): ) plugin_connector_inst = plugin_connector.PluginRuntimeConnector(ap, runtime_disconnect_callback) - try: - await plugin_connector_inst.initialize() - except Exception as exc: - # Keep the API/UI available while an external or managed runtime is - # starting, then recover in the background with bounded backoff. - ap.logger.warning(f'Plugin runtime unavailable during startup; reconnecting in background: {exc}') - plugin_connector_inst.schedule_reconnect() ap.plugin_connector = plugin_connector_inst workspace_service_inst.release_startup_execution_bindings() diff --git a/tests/unit_tests/core/test_app_shutdown.py b/tests/unit_tests/core/test_app_shutdown.py index ad6dcfad6..d810eb052 100644 --- a/tests/unit_tests/core/test_app_shutdown.py +++ b/tests/unit_tests/core/test_app_shutdown.py @@ -144,3 +144,18 @@ async def test_runtime_resource_stats_are_aggregate_and_constant_time() -> None: assert stats['models']['providers'] == 1 assert stats['runtimes']['plugin_installations'] == 1 assert stats['runtimes']['plugin_runtime_connected'] is True + + +@pytest.mark.asyncio +async def test_start_plugin_runtime_initialization_is_scheduled() -> None: + app = Application() + app.plugin_connector = SimpleNamespace(initialize=AsyncMock()) + captured = {} + app.task_mgr = SimpleNamespace(create_task=lambda coro, **kwargs: captured.update(coro=coro, kwargs=kwargs)) + + app._start_plugin_runtime_initialization() + + assert captured['kwargs']['name'] == 'plugin-runtime-initialization' + assert captured['kwargs']['scopes'] + await captured['coro'] + app.plugin_connector.initialize.assert_awaited_once_with() From 0bec72a3f9702b16069e6a27f9cf12c65a737568 Mon Sep 17 00:00:00 2001 From: dadachann <185672915+dadachann@users.noreply.github.com> Date: Wed, 19 Aug 2026 17:53:43 +0000 Subject: [PATCH 5/6] fix(cloud): run runtime initialization outside transaction gate --- src/langbot/pkg/core/app.py | 17 +++++++++-- tests/unit_tests/core/test_app_shutdown.py | 35 +++++++++++++++++----- 2 files changed, 42 insertions(+), 10 deletions(-) diff --git a/src/langbot/pkg/core/app.py b/src/langbot/pkg/core/app.py index 493e37363..79459b706 100644 --- a/src/langbot/pkg/core/app.py +++ b/src/langbot/pkg/core/app.py @@ -310,12 +310,18 @@ class Application: self.logger.warning(f'Plugin runtime unavailable during startup; reconnecting in background: {exc}') self.plugin_connector.schedule_reconnect() - def _start_plugin_runtime_initialization(self): - return self.task_mgr.create_task( + def _start_plugin_runtime_initialization(self) -> asyncio.Task | None: + task = getattr(self, '_plugin_runtime_initialization_task', None) + if task is not None and not task.done(): + return task + # This is application lifecycle work, not a request side effect. It must + # not wait on PersistenceManager's after-commit gate at boot. + task = asyncio.create_task( self._initialize_plugin_runtime(), name='plugin-runtime-initialization', - scopes=[core_entities.LifecycleControlScope.APPLICATION], ) + self._plugin_runtime_initialization_task = task + return task async def run(self): self.event_loop_monitor.start() @@ -544,6 +550,11 @@ class Application: if self.task_mgr is not None: self.task_mgr.cancel_by_scope(core_entities.LifecycleControlScope.APPLICATION) + plugin_runtime_task = getattr(self, '_plugin_runtime_initialization_task', None) + if plugin_runtime_task is not None and not plugin_runtime_task.done(): + plugin_runtime_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await plugin_runtime_task with contextlib.suppress(Exception): await self.event_loop_monitor.stop() mcp_mount = getattr(self.http_ctrl, 'mcp_mount', None) diff --git a/tests/unit_tests/core/test_app_shutdown.py b/tests/unit_tests/core/test_app_shutdown.py index d810eb052..2c953aaac 100644 --- a/tests/unit_tests/core/test_app_shutdown.py +++ b/tests/unit_tests/core/test_app_shutdown.py @@ -147,15 +147,36 @@ async def test_runtime_resource_stats_are_aggregate_and_constant_time() -> None: @pytest.mark.asyncio -async def test_start_plugin_runtime_initialization_is_scheduled() -> None: +async def test_start_plugin_runtime_initialization_bypasses_after_commit_gate() -> None: app = Application() app.plugin_connector = SimpleNamespace(initialize=AsyncMock()) - captured = {} - app.task_mgr = SimpleNamespace(create_task=lambda coro, **kwargs: captured.update(coro=coro, kwargs=kwargs)) + app.task_mgr = SimpleNamespace(create_task=AsyncMock()) - app._start_plugin_runtime_initialization() + task = app._start_plugin_runtime_initialization() + await task - assert captured['kwargs']['name'] == 'plugin-runtime-initialization' - assert captured['kwargs']['scopes'] - await captured['coro'] app.plugin_connector.initialize.assert_awaited_once_with() + app.task_mgr.create_task.assert_not_called() + + +@pytest.mark.asyncio +async def test_shutdown_cancels_plugin_runtime_initialization_task() -> None: + app = Application() + app._plugin_runtime_initialization_task = asyncio.create_task(asyncio.sleep(60)) + app.task_mgr = SimpleNamespace(cancel_by_scope=lambda *_: None, tasks=[]) + app.event_loop_monitor = SimpleNamespace(stop=AsyncMock()) + app.http_ctrl = SimpleNamespace(mcp_mount=None) + app.platform_mgr = None + app.tool_mgr = None + app.model_mgr = None + app.box_service = None + app.plugin_connector = None + app.telemetry = None + app.vector_db_mgr = None + app.storage_mgr = None + app.persistence_mgr = SimpleNamespace(db=SimpleNamespace(engine=SimpleNamespace(dispose=AsyncMock()))) + app.deployment = None + + await app.shutdown() + + assert app._plugin_runtime_initialization_task.cancelled() From 6bf1546df20350a6df7f8c24fdcd76165dcb6c2a Mon Sep 17 00:00:00 2001 From: dadachann <185672915+dadachann@users.noreply.github.com> Date: Thu, 20 Aug 2026 19:08:41 +0000 Subject: [PATCH 6/6] style: format Cloud runtime readiness fixes --- src/langbot/pkg/core/app.py | 8 ++++++-- src/langbot/pkg/core/stages/build_app.py | 2 +- src/langbot/pkg/plugin/connector.py | 8 ++++++-- 3 files changed, 13 insertions(+), 5 deletions(-) diff --git a/src/langbot/pkg/core/app.py b/src/langbot/pkg/core/app.py index 79459b706..f1bff1c5b 100644 --- a/src/langbot/pkg/core/app.py +++ b/src/langbot/pkg/core/app.py @@ -326,10 +326,13 @@ class Application: async def run(self): self.event_loop_monitor.start() try: - if self.directory_projection_service is not None and getattr(self, "directory_projection_task", None) is None: + if ( + self.directory_projection_service is not None + and getattr(self, 'directory_projection_task', None) is None + ): self.directory_projection_task = self.task_mgr.create_task( self.directory_projection_service.run(), - name="cloud-directory-projection", + name='cloud-directory-projection', scopes=[core_entities.LifecycleControlScope.APPLICATION], ) if self.cloud_model_catalog_service is not None: @@ -344,6 +347,7 @@ class Application: name='cloud-manifest-refresh', scopes=[core_entities.LifecycleControlScope.APPLICATION], ) + # 后续可能会允许动态重启其他任务 # 故为了防止程序在非 Ctrl-C 情况下退出，这里创建一个不会结束的协程 async def never_ending(): diff --git a/src/langbot/pkg/core/stages/build_app.py b/src/langbot/pkg/core/stages/build_app.py index fed6031fc..7941a5553 100644 --- a/src/langbot/pkg/core/stages/build_app.py +++ b/src/langbot/pkg/core/stages/build_app.py @@ -298,7 +298,7 @@ class BuildAppStage(stage.BootingStage): # its long-lived tasks, so start the single refresh task here. ap.directory_projection_task = ap.task_mgr.create_task( ap.directory_projection_service.run(), - name="cloud-directory-projection", + name='cloud-directory-projection', scopes=[core_entities.LifecycleControlScope.APPLICATION], ) diff --git a/src/langbot/pkg/plugin/connector.py b/src/langbot/pkg/plugin/connector.py index bcdf4fe8e..03893cf4c 100644 --- a/src/langbot/pkg/plugin/connector.py +++ b/src/langbot/pkg/plugin/connector.py @@ -701,7 +701,9 @@ class PluginRuntimeConnector(ManagedRuntimeConnector): } self._known_desired_states.update({state.binding.installation_uuid: state for state in desired_states}) - reconcile_timeout_seconds = max(300.0, self._runtime_connect_timeout(self.ap.instance_config.data.get("plugin", {}))) + reconcile_timeout_seconds = max( + 300.0, self._runtime_connect_timeout(self.ap.instance_config.data.get('plugin', {})) + ) result = await runtime_handler.reconcile_plugin_installations( tuple(self._known_desired_states.values()), timeout=reconcile_timeout_seconds, @@ -740,7 +742,9 @@ class PluginRuntimeConnector(ManagedRuntimeConnector): if state.binding.installation_uuid in all_states: raise ValueError('Duplicate plugin installation UUID across projected Workspaces') all_states[state.binding.installation_uuid] = state - reconcile_timeout_seconds = max(300.0, self._runtime_connect_timeout(self.ap.instance_config.data.get("plugin", {}))) + reconcile_timeout_seconds = max( + 300.0, self._runtime_connect_timeout(self.ap.instance_config.data.get('plugin', {})) + ) result = await runtime_handler.reconcile_plugin_installations( tuple(all_states.values()), timeout=reconcile_timeout_seconds,