diff --git a/src/langbot/pkg/agent/runner/execution_context.py b/src/langbot/pkg/agent/runner/execution_context.py index 439cac40b..4e8415e8c 100644 --- a/src/langbot/pkg/agent/runner/execution_context.py +++ b/src/langbot/pkg/agent/runner/execution_context.py @@ -12,6 +12,8 @@ from langbot_plugin.api.entities.builtin.platform import message as platform_mes from langbot_plugin.api.entities.builtin.provider import message as provider_message from langbot_plugin.api.entities.builtin.provider import session as provider_session +from ...api.http.context import ExecutionContext + from .host_models import AgentEventEnvelope @@ -110,8 +112,11 @@ def prepare_execution_query( def build_execution_query( event: AgentEventEnvelope, authorized_skill_names: list[str], + execution_context: ExecutionContext, ) -> pipeline_query.Query: """Build the minimum Query view required by Host-owned tool loaders.""" + if event.workspace_id and event.workspace_id != execution_context.workspace_uuid: + raise ValueError('Event Workspace does not match execution context') launcher_type, launcher_id, sender_id = _resolve_session_identity(event) message_chain = _build_message_chain(event) message_event = platform_events.MessageEvent( @@ -133,6 +138,10 @@ def build_execution_query( user_content = event.input.text or '' query = pipeline_query.Query( + instance_uuid=execution_context.instance_uuid, + workspace_uuid=execution_context.workspace_uuid, + placement_generation=execution_context.placement_generation, + query_uuid=execution_context.query_uuid, query_id=_synthetic_query_id(event.event_id), launcher_type=launcher_type, launcher_id=launcher_id, diff --git a/src/langbot/pkg/agent/runner/orchestrator.py b/src/langbot/pkg/agent/runner/orchestrator.py index fab4b8ef6..23706f9e9 100644 --- a/src/langbot/pkg/agent/runner/orchestrator.py +++ b/src/langbot/pkg/agent/runner/orchestrator.py @@ -116,16 +116,13 @@ class AgentRunOrchestrator: raise ValueError(f'The selected Runner does not support {usage} usage') if execution_query is None: - execution_query = build_execution_query(event, []) + execution_query = build_execution_query(event, [], execution_context) # Event runs bypass pipeline prompt preparation, just like debug runs. prompt_config = extract_prompt_config(descriptor, binding.runner_config, []) execution_query.prompt = provider_prompt.Prompt( name='default', messages=[provider_message.Message(**item) for item in copy.deepcopy(prompt_config)], ) - # Synthetic events must expose the same trusted scope as pipeline queries. - for field_name in ('instance_uuid', 'workspace_uuid', 'placement_generation', 'query_uuid'): - object.__setattr__(execution_query, field_name, getattr(execution_context, field_name)) project_mcp_resource_config(execution_query, binding.runner_config) object.__setattr__(execution_query, '_execution_context', execution_context) diff --git a/src/langbot/pkg/api/http/service/agent.py b/src/langbot/pkg/api/http/service/agent.py index 442e1dcf7..61a64d3dd 100644 --- a/src/langbot/pkg/api/http/service/agent.py +++ b/src/langbot/pkg/api/http/service/agent.py @@ -316,13 +316,11 @@ class AgentService: # prompt before advertising the Host prompt API. Keep explicit [] intact. descriptor = await self.ap.runner_registry.get(execution_context, runner_id) prompt_config = extract_prompt_config(descriptor, runner_config, []) - execution_query = build_execution_query(event, []) + execution_query = build_execution_query(event, [], execution_context) execution_query.prompt = provider_prompt.Prompt( name='default', messages=[provider_message.Message(**message) for message in copy.deepcopy(prompt_config)], ) - for field_name in ('instance_uuid', 'workspace_uuid', 'placement_generation', 'query_uuid'): - object.__setattr__(execution_query, field_name, getattr(execution_context, field_name)) project_mcp_resource_config(execution_query, runner_config) output_items: list[dict[str, typing.Any]] = [] diff --git a/src/langbot/pkg/pipeline/preproc/preproc.py b/src/langbot/pkg/pipeline/preproc/preproc.py index 37c5d28b8..e2189f3d5 100644 --- a/src/langbot/pkg/pipeline/preproc/preproc.py +++ b/src/langbot/pkg/pipeline/preproc/preproc.py @@ -261,7 +261,9 @@ class PreProcessor(stage.PipelineStage): conversation.update_time = now # Attach resolved session state to the query. - query.session = session + # Keep per-request identity separate from cached conversation state. + query.session = session.model_copy(update={'sender_id': query.sender_id, 'bot_uuid': query.bot_uuid}) + query.session.inherit_execution_scope(get_query_execution_context(query)) query.prompt = conversation.prompt.copy() query.messages = await self._resolve_history_messages( runner_id, diff --git a/tests/unit_tests/agent/test_debug_prompt.py b/tests/unit_tests/agent/test_debug_prompt.py index c2f2d0351..ba743e15a 100644 --- a/tests/unit_tests/agent/test_debug_prompt.py +++ b/tests/unit_tests/agent/test_debug_prompt.py @@ -291,7 +291,7 @@ async def test_non_debug_query_keeps_effective_prompt_instead_of_static_config(h input={'text': 'hello'}, delivery={'surface': 'test'}, ) - query = build_execution_query(event, []) + query = build_execution_query(event, [], ExecutionContext.from_request(request_context())) query.prompt = Prompt(name='preprocessed', messages=[Message(**message) for message in effective]) original = query.prompt binding = AgentBinding( diff --git a/tests/unit_tests/agent/test_execution_context.py b/tests/unit_tests/agent/test_execution_context.py index 8910b5c26..a26ee4d5b 100644 --- a/tests/unit_tests/agent/test_execution_context.py +++ b/tests/unit_tests/agent/test_execution_context.py @@ -2,6 +2,7 @@ from __future__ import annotations +import pytest from langbot_plugin.api.entities.builtin.runner.delivery import DeliveryContext from langbot_plugin.api.entities.builtin.runner.input import AgentInput @@ -15,6 +16,12 @@ from langbot.pkg.agent.runner.execution_context import ( project_mcp_resource_config, ) from langbot.pkg.agent.runner.host_models import AgentEventEnvelope +from langbot.pkg.api.http.context import ExecutionContext + + +CONTEXT = ExecutionContext( + instance_uuid='instance-1', workspace_uuid='workspace-1', placement_generation=2, query_uuid='query-1' +) class PlatformAdapter: @@ -54,7 +61,7 @@ def test_query_preparation_does_not_choose_a_box(): event = make_event() query = pipeline_query.Query.model_construct(variables={}) prepare_execution_query(query, event, ['pdf']) - event_query = build_execution_query(event, ['pdf']) + event_query = build_execution_query(event, ['pdf'], CONTEXT) assert query.variables == {'_pipeline_bound_skills': ['pdf']} assert event_query.variables == {'_pipeline_bound_skills': ['pdf']} assert getattr(query, '_box_binding', None) is None @@ -111,7 +118,7 @@ def test_pinned_context_updates_text_and_structured_input(): def test_event_reply_target_populates_valid_session_identity(): event = make_event(conversation_id='rotating-transcript-id', target_type='group', target_id='room-1') - query = build_execution_query(event, []) + query = build_execution_query(event, [], CONTEXT) assert query.launcher_type.value == 'group' assert query.launcher_id == 'room-1' @@ -131,8 +138,25 @@ def test_non_message_event_without_conversation_uses_event_scope(): ) event.event_type = 'schedule.triggered' - query = build_execution_query(event, []) + query = build_execution_query(event, [], CONTEXT) assert '_host_box_scope' not in query.variables assert query.pipeline_config is None assert query.pipeline_uuid is None + + +def test_execution_scope_reaches_nested_event_and_session_at_construction(): + query = build_execution_query(make_event(target_type='group'), [], CONTEXT) + for scoped in (query, query.session, query.message_event): + assert scoped.instance_uuid == CONTEXT.instance_uuid + assert scoped.workspace_uuid == CONTEXT.workspace_uuid + assert scoped.placement_generation == CONTEXT.placement_generation + assert query.query_uuid == CONTEXT.query_uuid + assert query.session.bot_uuid == 'bot-1' + + +def test_execution_query_rejects_event_from_another_workspace(): + event = make_event() + event.workspace_id = 'another-workspace' + with pytest.raises(ValueError, match='Workspace'): + build_execution_query(event, [], CONTEXT) diff --git a/tests/unit_tests/pipeline/test_preproc.py b/tests/unit_tests/pipeline/test_preproc.py index 1b64e2ca3..c9121e9b1 100644 --- a/tests/unit_tests/pipeline/test_preproc.py +++ b/tests/unit_tests/pipeline/test_preproc.py @@ -11,11 +11,14 @@ Tests cover preprocessing behavior including: from __future__ import annotations +import asyncio + import pytest from unittest.mock import AsyncMock, Mock from importlib import import_module from langbot_plugin.api.entities.builtin.provider import session as provider_session +from langbot_plugin.api.entities.builtin.provider.prompt import Prompt from langbot.pkg.agent.runner.descriptor import RunnerDescriptor from tests.factories import ( @@ -113,24 +116,28 @@ class TestPreProcessorNormalText: assert query.messages[0].content == 'plugin history' @pytest.mark.asyncio - async def test_normal_text_continues(self): + @pytest.mark.parametrize('cross_workspace', [False, True]) + async def test_normal_text_continues(self, cross_workspace): """Normal text message should continue pipeline.""" preproc = get_preproc_module() entities = get_entities_module() app = FakeApp() # Mock session manager to return a session - mock_session = make_session() + mock_session = make_session(provider_session.LauncherTypes.GROUP, 99999) + if cross_workspace: + mock_session.workspace_uuid = 'other-workspace' + mock_session._semaphore = asyncio.Semaphore(1) app.sess_mgr.get_session = AsyncMock(return_value=mock_session) - # Mock conversation - mock_conversation = Mock() - mock_conversation.prompt = Mock() - mock_conversation.prompt.messages = [] - mock_conversation.prompt.copy = Mock(return_value=Mock(messages=[])) - mock_conversation.messages = [] - mock_conversation.update_time = Mock() - mock_conversation.uuid = None + mock_conversation = provider_session.Conversation( + prompt=Prompt(name='default', messages=[]), + messages=[], + pipeline_uuid='test-pipeline-uuid', + bot_uuid='test-bot-uuid', + ) + mock_session.using_conversation = mock_conversation + mock_session.conversations.append(mock_conversation) app.sess_mgr.get_conversation = AsyncMock(return_value=mock_conversation) # Mock model manager @@ -151,13 +158,27 @@ class TestPreProcessorNormalText: app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx) stage = preproc.PreProcessor(app) - query = text_query('hello world') + query = group_text_query('hello world', sender_id=67890) + if cross_workspace: + with pytest.raises(ValueError, match='Execution scope mismatch'): + await stage.process(query, 'PreProcessor') + app.plugin_connector.emit_event.assert_not_awaited() + return result = await stage.process(query, 'PreProcessor') assert result.result_type == entities.ResultType.CONTINUE assert result.new_query is not None + assert query.session is not mock_session + assert query.session.sender_id == 67890 + assert mock_session.sender_id == 12345 + assert query.session.bot_uuid == query.bot_uuid + assert query.session.workspace_uuid == query.workspace_uuid + assert query.session.conversations is mock_session.conversations + assert query.session.using_conversation is mock_session.using_conversation + assert query.session._semaphore is mock_session._semaphore + @pytest.mark.asyncio async def test_normal_text_sets_user_message(self): """PreProcessor should set user_message from text content.""" diff --git a/tests/unit_tests/platform/test_http_bot_agent_boundary.py b/tests/unit_tests/platform/test_http_bot_agent_boundary.py index 7ae206721..3e31924c2 100644 --- a/tests/unit_tests/platform/test_http_bot_agent_boundary.py +++ b/tests/unit_tests/platform/test_http_bot_agent_boundary.py @@ -16,6 +16,7 @@ from langbot_plugin.box.client import BoxRuntimeClient from langbot_plugin.api.proxies.runner.box import RunnerBoxAPIMixin from langbot_plugin.entities.io.actions.enums import PluginToRuntimeAction +from langbot.pkg.api.http.context import ExecutionContext from langbot.pkg.agent.runner.execution_context import build_execution_query from langbot.pkg.agent.runner.platform_tools import ( build_platform_tool_resources, @@ -86,7 +87,15 @@ async def test_signed_http_file_reaches_box_and_authorized_agent_reply_reaches_s async def listener(inbound, live_adapter): event = envelope(bot, live_adapter, inbound) received.append(event) - query = build_execution_query(event, []) + query = build_execution_query( + event, + [], + ExecutionContext( + instance_uuid='instance-test', + workspace_uuid=event.workspace_id or 'workspace-test', + placement_generation=1, + ), + ) prepare_input_files(query, event.input) object.__setattr__(query, '_box_binding', RunBoxBinding('run-test', 'box-test', {}, 'run-test')) imported = await runner_box.import_attachments(query, None) diff --git a/tests/unit_tests/plugin/test_handler_actions.py b/tests/unit_tests/plugin/test_handler_actions.py index 8fd068745..bc101f414 100644 --- a/tests/unit_tests/plugin/test_handler_actions.py +++ b/tests/unit_tests/plugin/test_handler_actions.py @@ -1611,7 +1611,7 @@ class TestAgentRunProxyActions: platform_capabilities={'adapter': 'TestAdapter'}, ), ) - query = build_execution_query(event, []) + query = build_execution_query(event, [], TEST_EXECUTION_CONTEXT) app.box_service = SimpleNamespace( available=True, get_backend_status=AsyncMock(return_value={'backend': {'available': True}}),