diff --git a/docker/docker-compose.yaml b/docker/docker-compose.yaml index 5cd3ad0f9..fe1aecba0 100644 --- a/docker/docker-compose.yaml +++ b/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 diff --git a/docker/kubernetes.yaml b/docker/kubernetes.yaml index 53a629e20..189641a96 100644 --- a/docker/kubernetes.yaml +++ b/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 --- diff --git a/src/langbot/pkg/pipeline/process/handlers/chat.py b/src/langbot/pkg/pipeline/process/handlers/chat.py index 5226d56b0..274c6fcee 100644 --- a/src/langbot/pkg/pipeline/process/handlers/chat.py +++ b/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}') diff --git a/tests/unit_tests/agent/test_chat_handler.py b/tests/unit_tests/agent/test_chat_handler.py index 8e7276d8c..9066948dd 100644 --- a/tests/unit_tests/agent/test_chat_handler.py +++ b/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."""