feat(agent-runner): expose effective prompt pull api

src/langbot/pkg/agent/runner/context_builder.py

@@ -422,6 +422,7 @@ class AgentRunContextBuilder:
                 'reason': 'current_event_only',
             },
             'available_apis': {
+                'prompt_get': False,
                 'history_page': history_page_enabled,
                 'history_search': history_search_enabled,
                 'event_get': event_get_enabled,

src/langbot/pkg/agent/runner/orchestrator.py

@@ -96,6 +96,8 @@ class AgentRunOrchestrator:
                     context.get('state', {}),
                 )
             session_query_id = adapter_context.get('query_id')
+            if query is not None or session_query_id is not None:
+                context['context']['available_apis']['prompt_get'] = True
             if 'params' in adapter_context:
                 context['adapter']['extra']['params'] = adapter_context['params']
 

src/langbot/pkg/plugin/handler.py

@@ -1506,6 +1506,40 @@ class RuntimeConnectionHandler(handler.Handler):
 
         # ================= Agent History/Event APIs =================
 
+        @self.action(PluginToRuntimeAction.GET_PROMPT)
+        async def get_prompt(data: dict[str, Any]) -> handler.ActionResponse:
+            """Return the current run's effective prompt after PromptPreProcessing."""
+            run_id = data.get('run_id')
+            caller_plugin_identity = data.get('caller_plugin_identity')
+
+            if not run_id:
+                return handler.ActionResponse.error(message='run_id is required')
+
+            session, error = await _validate_agent_run_session(
+                run_id,
+                caller_plugin_identity,
+                self.ap,
+                'Get prompt',
+                api_capability='prompt_get',
+            )
+            if error:
+                return error
+
+            query = _resolve_action_query(data, session, self.ap)
+            if query is None:
+                return handler.ActionResponse.error(
+                    message=f'Query for run_id {run_id} not found or expired',
+                )
+
+            prompt = getattr(query, 'prompt', None)
+            messages = getattr(prompt, 'messages', []) or []
+            return handler.ActionResponse.success(data={
+                'prompt': [
+                    message.model_dump(mode='json') if hasattr(message, 'model_dump') else message
+                    for message in messages
+                ],
+            })
+
         @self.action(PluginToRuntimeAction.HISTORY_PAGE)
         async def history_page(data: dict[str, Any]) -> handler.ActionResponse:
             """Page through transcript history for a conversation.

tests/unit_tests/agent/test_context_builder_state.py

@@ -283,6 +283,7 @@ class TestContextAccessOtherAPIs:
         # Real call
         context_access = await builder._build_context_access(mock_event, mock_descriptor, binding)
 
+        assert context_access['available_apis']['prompt_get'] is False
         assert context_access['available_apis']['history_page'] is True
         assert context_access['available_apis']['history_search'] is True
 

tests/unit_tests/agent/test_orchestrator_integration.py

@@ -669,7 +669,7 @@ class TestQueryEntryAdapterParams:
 
     @pytest.mark.asyncio
     async def test_prompt_not_pushed_into_adapter_extra(self, clean_agent_state):
-        """Pipeline prompt is not pushed into adapter.extra or exposed via prompt_get."""
+        """Pipeline prompt is not pushed into adapter.extra; runners pull it through prompt_get."""
         from langbot_plugin.api.entities.builtin.provider import prompt as provider_prompt
 
         db_engine = clean_agent_state
@@ -699,7 +699,7 @@ class TestQueryEntryAdapterParams:
         context = plugin_connector.contexts[0]
         assert "prompt" not in context
         assert "prompt" not in context["adapter"]["extra"]
-        assert "prompt_get" not in context["context"]["available_apis"]
+        assert context["context"]["available_apis"]["prompt_get"] is True
 
     @pytest.mark.asyncio
     async def test_params_filtering_keeps_public_param(self, clean_agent_state):

tests/unit_tests/plugin/test_handler_actions.py

@@ -388,8 +388,45 @@ class TestAgentRunProxyActions:
     def query(remove_think=True):
         return SimpleNamespace(
             pipeline_config={'output': {'misc': {'remove-think': remove_think}}},
+            prompt=SimpleNamespace(
+                messages=[provider_message.Message(role='system', content='effective prompt')]
+            ),
         )
 
+    @pytest.mark.asyncio
+    async def test_get_prompt_returns_query_effective_prompt(self, app):
+        """GET_PROMPT returns the preprocessed Query prompt for the active run."""
+        from langbot.pkg.agent.runner.session_registry import get_session_registry
+
+        run_id = 'run_proxy_get_prompt'
+        query = self.query()
+        app.query_pool.cached_queries[900] = query
+
+        registry = get_session_registry()
+        await registry.unregister(run_id)
+        await registry.register(
+            run_id=run_id,
+            runner_id='plugin:test/runner/default',
+            query_id=900,
+            plugin_identity='test/runner',
+            resources=make_agent_resources(),
+            available_apis={'prompt_get': True},
+        )
+
+        runtime_handler = make_handler(app)
+
+        try:
+            response = await runtime_handler.actions[PluginToRuntimeAction.GET_PROMPT.value]({
+                'run_id': run_id,
+                'caller_plugin_identity': 'test/runner',
+            })
+        finally:
+            await registry.unregister(run_id)
+
+        assert response.code == 0
+        assert response.data['prompt'][0]['role'] == 'system'
+        assert response.data['prompt'][0]['content'] == 'effective prompt'
+
     @pytest.mark.asyncio
     async def test_invoke_llm_restores_query_and_model_options(self, app):
         """INVOKE_LLM passes Query, model extra_args and remove-think to provider."""