fix(agent): prepare plugin execution identity at context creation (#2842)

* fix(tools): propagate current execution identity to plugin calls

* fix(agent): construct complete execution identity before tool dispatch

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,

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)
 

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]] = []

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,

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(

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)

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."""

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)

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}}),