fix(plugin): persist v4 runtime state and tolerate runner registration gaps (#2774)

Persist the v4 plugin runtime state so recreating the runtime container no
longer discards installed plugins' artifacts, dependency environments and
installation-private data, and stop a transient Runner registry miss during
steering from dropping the message.

- docker-compose: bind-mount ./data/plugin-runtime for the plugin runtime
  alongside ./data/plugins.
- kubernetes: add the langbot-plugin-runtime-state claim (5Gi) mounted at
  /app/data/plugin-runtime, kept separate from the plugin data claim so
  runtime state does not land inside the plugin directory.
- chat handler: treat RunnerNotFoundError while claiming steering as
  "steering unavailable" and continue with a normal run; every other error
  keeps propagating. The runner error imports move to module scope, which
  the new early reference requires.
- tests: cover the steering fallback, the claimed-steering interrupt, and
  the lasting-gap path that still reports runner-not-found.

The steering fallback buys one extra uncached registry lookup while a
runtime reconnects; it does not recover a gap that outlasts the message.

docker/docker-compose.yaml

@@ -9,6 +9,11 @@ services:
     container_name: langbot_plugin_runtime
     volumes:
       - ./data/plugins:/app/data/plugins
+      # Persist the v4 plugin artifact cache, dependency environments, and
+      # per-installation data. Without this, recreating the container throws
+      # away every installed plugin's environment and private data while
+      # data/plugins survives.
+      - ./data/plugin-runtime:/app/data/plugin-runtime
     ports:
       - 5401:5401
     restart: on-failure

docker/kubernetes.yaml

@@ -81,6 +81,25 @@ spec:
   # Uncomment and modify if you have a specific StorageClass
   # storageClassName: your-storage-class
 
+---
+# PersistentVolumeClaim for the v4 Plugin Runtime state
+# Holds the plugin artifact cache, per-plugin dependency environments, and
+# per-installation data. It must survive Pod restarts: rebuilding it forces
+# every installed plugin to be re-downloaded and its environment rebuilt.
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: langbot-plugin-runtime-state
+  namespace: langbot
+spec:
+  accessModes:
+    - ReadWriteOnce
+  resources:
+    requests:
+      storage: 5Gi
+  # Uncomment and modify if you have a specific StorageClass
+  # storageClassName: your-storage-class
+
 ---
 # ConfigMap for environment configuration
 apiVersion: v1
@@ -160,6 +179,8 @@ spec:
         volumeMounts:
         - name: plugin-data
           mountPath: /app/data/plugins
+        - name: plugin-runtime-state
+          mountPath: /app/data/plugin-runtime
         resources:
           requests:
             memory: "512Mi"
@@ -189,6 +210,9 @@ spec:
       - name: plugin-data
         persistentVolumeClaim:
           claimName: langbot-plugin-runtime-data
+      - name: plugin-runtime-state
+        persistentVolumeClaim:
+          claimName: langbot-plugin-runtime-state
       restartPolicy: Always
 
 ---

src/langbot/pkg/pipeline/process/handlers/chat.py

@@ -15,6 +15,11 @@ from ... import plugin_diagnostics
 import langbot_plugin.api.entities.events as events
 from ....agent.runner.config_resolver import RunnerConfigResolver
 from ....agent.runner import config_schema
+from ....agent.runner.errors import (
+    RunnerExecutionError,
+    RunnerNotFoundError,
+    RunnerNotAuthorizedError,
+)
 from ....utils import runner as runner_utils
 import langbot_plugin.api.entities.builtin.provider.session as provider_session
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
@@ -119,9 +124,24 @@ class ChatMessageHandler(handler.MessageHandler):
                     'try_claim_steering_from_query',
                     None,
                 )
-                if try_claim_steering and await try_claim_steering(query):
-                    yield entities.StageProcessResult(result_type=entities.ResultType.INTERRUPT, new_query=query)
-                    return
+                if try_claim_steering:
+                    try:
+                        claimed = await try_claim_steering(query)
+                    except RunnerNotFoundError as steering_error:
+                        # Steering is an optimization for an already-running
+                        # agent. A runtime reconnect can temporarily remove the
+                        # selected Runner from the registry; that must not
+                        # prevent the message from starting a normal run.
+                        self.ap.logger.warning(
+                            'Skipping steering for query %s because Runner %s is not registered',
+                            query.query_id,
+                            steering_error.runner_id,
+                        )
+                        claimed = False
+
+                    if claimed:
+                        yield entities.StageProcessResult(result_type=entities.ResultType.INTERRUPT, new_query=query)
+                        return
 
                 try:
                     is_stream = await query.adapter.is_stream_output_supported()
@@ -228,13 +248,6 @@ class ChatMessageHandler(handler.MessageHandler):
                     yield entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
 
             except Exception as e:
-                # Import orchestrator errors for specific handling
-                from ....agent.runner.errors import (
-                    RunnerNotFoundError,
-                    RunnerNotAuthorizedError,
-                    RunnerExecutionError,
-                )
-
                 error_info = f'{traceback.format_exc()}'
                 self.ap.logger.error(f'Conversation({query.query_id}) Request Failed: {error_info}')
 

tests/unit_tests/agent/test_chat_handler.py

@@ -140,6 +140,39 @@ class MockAgentRunOrchestrator:
         return 'plugin:langbot-team/LocalAgent/default'
 
 
+class SteeringRunnerNotFoundOrchestrator(MockAgentRunOrchestrator):
+    async def try_claim_steering_from_query(self, query):
+        raise RunnerNotFoundError('plugin:langbot-team/LocalAgent/default')
+
+
+class SteeringClaimedOrchestrator(MockAgentRunOrchestrator):
+    def __init__(self):
+        super().__init__()
+        self.run_called = False
+
+    async def try_claim_steering_from_query(self, query):
+        return True
+
+    async def run_from_query(self, query):
+        self.run_called = True
+        if False:
+            yield None
+
+
+class SteeringUnexpectedErrorOrchestrator(MockAgentRunOrchestrator):
+    async def try_claim_steering_from_query(self, query):
+        raise RuntimeError('steering lookup failed')
+
+
+class PersistentRunnerGapOrchestrator(SteeringRunnerNotFoundOrchestrator):
+    """Steering and the normal run both miss the Runner (a lasting gap)."""
+
+    async def run_from_query(self, query):
+        raise RunnerNotFoundError('plugin:langbot-team/LocalAgent/default')
+        if False:
+            yield None
+
+
 class MockApplication:
     """Mock Application for testing."""
 
@@ -441,6 +474,141 @@ class TestChatHandlerAsyncBehavior:
         assert results[0].result_type == entities.ResultType.INTERRUPT
         assert 'not found' in results[0].user_notice
 
+    @pytest.mark.asyncio
+    async def test_runner_not_found_during_steering_falls_back_to_normal_run(self):
+        """A transient registry miss during steering must not drop the message."""
+        from langbot.pkg.pipeline.process.handlers.chat import ChatMessageHandler
+        from langbot.pkg.pipeline import entities
+
+        orchestrator = SteeringRunnerNotFoundOrchestrator(
+            chunks=[MagicMock(content='ok', tool_calls=[], is_final=True)]
+        )
+        mock_ap = MockApplication(orchestrator=orchestrator)
+        mock_ap.plugin_connector.emit_event = AsyncMock(return_value=MockEventContext(prevented=False))
+        handler = ChatMessageHandler(mock_ap)
+        query = MockQuery()
+
+        mock_event = MagicMock()
+        mock_event.return_value = MagicMock()
+
+        def make_result(*args, **kwargs):
+            return MagicMock(
+                result_type=kwargs.get('result_type'),
+                user_notice=kwargs.get('user_notice'),
+            )
+
+        with (
+            patch('langbot.pkg.pipeline.process.handlers.chat.events') as mock_events_module,
+            patch('langbot.pkg.pipeline.entities.StageProcessResult', side_effect=make_result),
+        ):
+            mock_events_module.PersonNormalMessageReceived = mock_event
+            mock_events_module.GroupNormalMessageReceived = mock_event
+            results = [result async for result in handler.handle(query)]
+
+        assert results
+        assert all(result.result_type != entities.ResultType.INTERRUPT for result in results)
+        mock_ap.logger.warning.assert_any_call(
+            'Skipping steering for query %s because Runner %s is not registered',
+            query.query_id,
+            'plugin:langbot-team/LocalAgent/default',
+        )
+
+    @pytest.mark.asyncio
+    async def test_runner_missing_for_steering_and_run_reports_runner_not_found(self):
+        """A lasting gap still fails: the fallback buys one retry, not recovery."""
+        from langbot.pkg.pipeline.process.handlers.chat import ChatMessageHandler
+        from langbot.pkg.pipeline import entities
+
+        orchestrator = PersistentRunnerGapOrchestrator()
+        mock_ap = MockApplication(orchestrator=orchestrator)
+        mock_ap.plugin_connector.emit_event = AsyncMock(return_value=MockEventContext(prevented=False))
+        handler = ChatMessageHandler(mock_ap)
+        query = MockQuery()
+
+        mock_event = MagicMock()
+        mock_event.return_value = MagicMock()
+
+        def make_result(*args, **kwargs):
+            return MagicMock(
+                result_type=kwargs.get('result_type'),
+                user_notice=kwargs.get('user_notice'),
+            )
+
+        with (
+            patch('langbot.pkg.pipeline.process.handlers.chat.events') as mock_events_module,
+            patch('langbot.pkg.pipeline.entities.StageProcessResult', side_effect=make_result),
+        ):
+            mock_events_module.PersonNormalMessageReceived = mock_event
+            mock_events_module.GroupNormalMessageReceived = mock_event
+            results = [result async for result in handler.handle(query)]
+
+        assert len(results) == 1
+        assert results[0].result_type == entities.ResultType.INTERRUPT
+        assert 'not found' in results[0].user_notice
+
+    @pytest.mark.asyncio
+    async def test_successful_steering_interrupts_normal_run(self):
+        """A claimed steering message must not start a duplicate runner turn."""
+        from langbot.pkg.pipeline.process.handlers.chat import ChatMessageHandler
+        from langbot.pkg.pipeline import entities
+
+        orchestrator = SteeringClaimedOrchestrator()
+        mock_ap = MockApplication(orchestrator=orchestrator)
+        mock_ap.plugin_connector.emit_event = AsyncMock(return_value=MockEventContext(prevented=False))
+        handler = ChatMessageHandler(mock_ap)
+        query = MockQuery()
+
+        mock_event = MagicMock()
+        mock_event.return_value = MagicMock()
+
+        def make_result(*args, **kwargs):
+            return MagicMock(result_type=kwargs.get('result_type'))
+
+        with (
+            patch('langbot.pkg.pipeline.process.handlers.chat.events') as mock_events_module,
+            patch('langbot.pkg.pipeline.entities.StageProcessResult', side_effect=make_result),
+        ):
+            mock_events_module.PersonNormalMessageReceived = mock_event
+            mock_events_module.GroupNormalMessageReceived = mock_event
+            results = [result async for result in handler.handle(query)]
+
+        assert len(results) == 1
+        assert results[0].result_type == entities.ResultType.INTERRUPT
+        assert orchestrator.run_called is False
+
+    @pytest.mark.asyncio
+    async def test_unexpected_steering_error_uses_normal_error_handling(self):
+        """Unexpected steering failures must not be silently swallowed."""
+        from langbot.pkg.pipeline.process.handlers.chat import ChatMessageHandler
+        from langbot.pkg.pipeline import entities
+
+        orchestrator = SteeringUnexpectedErrorOrchestrator()
+        mock_ap = MockApplication(orchestrator=orchestrator)
+        mock_ap.plugin_connector.emit_event = AsyncMock(return_value=MockEventContext(prevented=False))
+        handler = ChatMessageHandler(mock_ap)
+        query = MockQuery()
+
+        mock_event = MagicMock()
+        mock_event.return_value = MagicMock()
+
+        def make_result(*args, **kwargs):
+            return MagicMock(
+                result_type=kwargs.get('result_type'),
+                user_notice=kwargs.get('user_notice'),
+            )
+
+        with (
+            patch('langbot.pkg.pipeline.process.handlers.chat.events') as mock_events_module,
+            patch('langbot.pkg.pipeline.entities.StageProcessResult', side_effect=make_result),
+        ):
+            mock_events_module.PersonNormalMessageReceived = mock_event
+            mock_events_module.GroupNormalMessageReceived = mock_event
+            results = [result async for result in handler.handle(query)]
+
+        assert len(results) == 1
+        assert results[0].result_type == entities.ResultType.INTERRUPT
+        assert 'Request failed' in results[0].user_notice
+
     @pytest.mark.asyncio
     async def test_runner_not_authorized_error(self):
         """Handler should catch RunnerNotAuthorizedError and return INTERRUPT."""