fix(runtime): preserve explicit replies and report bot configuration errors

tests/unit_tests/agent/test_reply_stream.py

@@ -0,0 +1,270 @@
+"""Explicit streaming delivery across SDK, Host lifecycle, and adapter boundaries."""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+from uuid import uuid4
+
+import pytest
+from langbot_plugin.api.entities.builtin.platform import events, entities, message
+from langbot_plugin.api.entities.builtin.agent_runner.context_access import ContextAPICapabilities
+from langbot_plugin.api.proxies.agent_run import AgentRunAPIProxy
+from langbot_plugin.api.proxies.agent_run.common import PermissionDeniedError
+from langbot_plugin.entities.io.actions.enums import PluginToRuntimeAction
+
+from langbot.pkg.agent.runner.reply_stream import ReplyStreamRequest, ReplyStreamSession
+
+
+def make_session(*, native=True, source=True, mock=False):
+    event = SimpleNamespace(
+        delivery=SimpleNamespace(
+            surface='webui' if mock else 'platform',
+            platform_capabilities={'debug_mock': mock},
+            reply_target={'target_type': 'person', 'target_id': 'user-1'},
+        )
+    )
+    incoming = (
+        events.MessageReceivedEvent(
+            message_id='source-1',
+            sender=entities.User(id='user-1'),
+            chat_id='user-1',
+            chat_type=entities.ChatType.PRIVATE,
+            message_chain=message.MessageChain([message.Plain(text='hello')]),
+            source_platform_object=object(),
+        )
+        if source
+        else None
+    )
+    adapter = SimpleNamespace(
+        is_stream_output_supported=AsyncMock(return_value=native),
+        create_message_card=AsyncMock(return_value=True),
+        reply_message_chunk=AsyncMock(),
+        send_message=AsyncMock(),
+        reply_message=AsyncMock(),
+    )
+    return ReplyStreamSession(event, adapter, incoming), adapter, incoming
+
+
+def request(key, operation='update', text='hello'):
+    return ReplyStreamRequest(stream_id=key, operation=operation, text=text)
+
+
+def proxy_for(session, *, allowed=True, advertised=True):
+    if not hasattr(AgentRunAPIProxy, 'reply_stream'):
+        pytest.skip('SDK does not provide the optional streaming reply API')
+    context = SimpleNamespace(
+        run_id='run-1',
+        runtime=SimpleNamespace(deadline_at=None),
+        context=SimpleNamespace(available_apis=ContextAPICapabilities(reply_stream=advertised)),
+        resources=SimpleNamespace(
+            models=[],
+            knowledge_bases=[],
+            tools=[SimpleNamespace(tool_name='event_reply', operations=['call'])] if allowed else [],
+        ),
+    )
+
+    async def action(action, data, timeout):
+        assert action == PluginToRuntimeAction.REPLY_STREAM
+        assert data['run_id'] == 'run-1'
+        return {
+            'result': await session.apply(
+                ReplyStreamRequest.model_validate({k: v for k, v in data.items() if k != 'run_id'})
+            )
+        }
+
+    transport = SimpleNamespace(call_action=AsyncMock(side_effect=action))
+    return AgentRunAPIProxy(context, transport), transport
+
+
+@pytest.mark.parametrize(
+    'native,source,mock',
+    [
+        (True, True, False),
+        (False, True, False),
+        (True, False, False),
+        (True, True, True),
+    ],
+)
+async def test_sdk_stream_reuses_adapter_or_sends_one_final_message(native, source, mock):
+    session, adapter, incoming = make_session(native=native, source=source, mock=mock)
+    api, transport = proxy_for(session)
+    async with api.reply_stream() as stream:
+        await stream.update('hello')
+        await stream.update('hello world')
+        adapter.send_message.assert_not_awaited()
+        adapter.reply_message.assert_not_awaited()
+    assert stream.result['status'] == 'completed'
+    assert stream.result['text'] == 'hello world'
+    assert transport.call_action.await_count == 3
+    if mock:
+        adapter.create_message_card.assert_not_awaited()
+        adapter.reply_message_chunk.assert_not_awaited()
+        adapter.send_message.assert_not_awaited()
+        assert stream.result['mock'] is True
+    elif native and source:
+        adapter.create_message_card.assert_awaited_once()
+        delivered_source = adapter.create_message_card.await_args.args[1]
+        assert delivered_source.source_platform_object is incoming.source_platform_object
+        chunks = adapter.reply_message_chunk.await_args_list
+        assert [c.kwargs['bot_message'].all_content for c in chunks] == ['hello', 'hello world', 'hello world']
+        assert [c.kwargs['is_final'] for c in chunks] == [False, False, True]
+        assert chunks[-1].kwargs['bot_message'].tool_calls is None
+    else:
+        adapter.reply_message_chunk.assert_not_awaited()
+        if source:
+            adapter.reply_message.assert_awaited_once()
+            assert adapter.reply_message.await_args.kwargs['message'][0].text == 'hello world'
+        else:
+            adapter.send_message.assert_awaited_once()
+            assert adapter.send_message.await_args.args[2][0].text == 'hello world'
+    await session.close()
+
+
+@pytest.mark.parametrize('native', [True, False])
+@pytest.mark.parametrize('error', [RuntimeError, asyncio.CancelledError])
+async def test_exception_finalizes_visible_card_without_sending_buffered_partial(native, error):
+    session, adapter, _ = make_session(native=native)
+    api, _ = proxy_for(session)
+    with pytest.raises(error):
+        async with api.reply_stream() as stream:
+            await stream.update('partial')
+            raise error()
+    if native:
+        assert adapter.reply_message_chunk.await_args.kwargs['is_final'] is True
+    adapter.reply_message.assert_not_awaited()
+    adapter.send_message.assert_not_awaited()
+    count = adapter.reply_message_chunk.await_count
+    await session.close()
+    assert adapter.reply_message_chunk.await_count == count
+
+
+@pytest.mark.parametrize('allowed,advertised', [(False, True), (True, False)])
+async def test_missing_permission_or_old_host_fails_before_delivery(allowed, advertised):
+    session, _, _ = make_session()
+    api, transport = proxy_for(session, allowed=allowed, advertised=advertised)
+    with pytest.raises(PermissionDeniedError):
+        async with api.reply_stream():
+            pytest.fail('Not authorized')
+    transport.call_action.assert_not_awaited()
+
+
+async def test_empty_stream_and_duplicate_finish_do_not_send_twice():
+    session, adapter, _ = make_session(native=False)
+    empty = uuid4()
+    await session.apply(request(empty, 'finish', ''))
+    adapter.reply_message.assert_not_awaited()
+    key = uuid4()
+    await session.apply(request(key))
+    first = await session.apply(request(key, 'finish'))
+    assert await session.apply(request(key, 'finish')) == first
+    adapter.reply_message.assert_awaited_once()
+    with pytest.raises(ValueError, match='closed'):
+        await session.apply(request(key))
+    await session.close()
+    with pytest.raises(ValueError, match='ended'):
+        await session.apply(request(uuid4()))
+
+
+async def test_host_cleanup_closes_stream_when_plugin_disappears():
+    session, adapter, _ = make_session()
+    await session.apply(request(uuid4()))
+    await session.close()
+    assert adapter.reply_message_chunk.await_args.kwargs['is_final'] is True
+
+
+async def test_uncertain_final_send_is_not_retried_by_finish_or_cleanup():
+    session, adapter, _ = make_session(native=False)
+    adapter.reply_message.side_effect = TimeoutError('Response lost')
+    key = uuid4()
+    with pytest.raises(TimeoutError):
+        await session.apply(request(key, 'finish'))
+    with pytest.raises(ValueError, match='closed'):
+        await session.apply(request(key, 'finish'))
+    await session.close()
+    adapter.reply_message.assert_awaited_once()
+
+
+async def test_failed_update_closes_existing_card_during_run_cleanup():
+    session, adapter, _ = make_session()
+    adapter.reply_message_chunk.side_effect = [RuntimeError('update failed'), None]
+    with pytest.raises(RuntimeError):
+        await session.apply(request(uuid4()))
+    await session.close()
+    assert adapter.reply_message_chunk.await_args.kwargs['is_final'] is True
+
+
+async def test_streams_are_isolated_by_run_and_bounded():
+    first, a, _ = make_session(native=False)
+    second, b, _ = make_session(native=False)
+    key = uuid4()
+    await first.apply(request(key, 'update', 'first'))
+    await second.apply(request(key, 'finish', 'second'))
+    assert b.reply_message.await_args.kwargs['message'][0].text == 'second'
+    a.reply_message.assert_not_awaited()
+    for _ in range(15):
+        await first.apply(request(uuid4(), 'finish', ''))
+    with pytest.raises(ValueError, match='at most'):
+        await first.apply(request(uuid4()))
+    await first.close()
+    a.reply_message.assert_not_awaited()
+
+
+async def test_event_processor_uses_shared_sdk_api_and_emits_one_trace_for_the_stream():
+    from unittest.mock import Mock
+    from langbot_plugin.api.definition.components.event_processor import EventProcessor
+
+    session, adapter, incoming = make_session()
+    api, _ = proxy_for(session)
+    processor = EventProcessor()
+    processor.get_run_api = Mock(return_value=api)
+
+    @processor.handler(events.MessageReceivedEvent)
+    async def handle(ctx):
+        async with ctx.reply_stream() as stream:
+            await stream.update('one')
+            await stream.update('one two')
+
+    context = SimpleNamespace(
+        run_id='run-1',
+        config={},
+        event=SimpleNamespace(
+            data=incoming.model_dump(mode='json', exclude={'source_platform_object', 'legacy_event'})
+        ),
+    )
+    results = [result async for result in processor.run(context)]
+    assert [r.type for r in results] == ['tool.call.started', 'tool.call.completed', 'run.completed']
+    assert results[1].data['result']['text'] == 'one two'
+    assert adapter.reply_message_chunk.await_count == 3
+    await session.close()
+
+
+async def test_shared_adapter_uses_host_ids_to_isolate_identical_plugin_stream_ids():
+    first, adapter, _ = make_session()
+    second, _, _ = make_session()
+    second.adapter = adapter
+    key = uuid4()
+    await first.apply(request(key, text='first'))
+    await second.apply(request(key, text='second'))
+    ids = [c.args[0] for c in adapter.create_message_card.await_args_list]
+    assert len(set(ids)) == 2
+    assert str(key) not in ids
+    await first.close()
+    await second.close()
+
+
+async def test_run_cleanup_cancels_inflight_update_and_finalizes_card():
+    session, adapter, _ = make_session()
+    started = asyncio.Event()
+
+    async def update(**kwargs):
+        if not kwargs['is_final']:
+            started.set()
+            await asyncio.Event().wait()
+
+    adapter.reply_message_chunk.side_effect = update
+    task = asyncio.create_task(session.apply(request(uuid4())))
+    await asyncio.wait_for(started.wait(), 1)
+    await asyncio.wait_for(session.close(), 1)
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert adapter.reply_message_chunk.await_args.kwargs['is_final'] is True

tests/unit_tests/api/service/test_agent_service.py

@@ -151,7 +151,18 @@ class TestAgentServiceMetadata:
 
 class TestAgentServiceDebug:
     @pytest.mark.parametrize('streaming', [False, True])
-    async def test_debug_agent_runs_configured_runner_with_synthetic_event(self, streaming):
+    @pytest.mark.parametrize(
+        'result_type',
+        [
+            'tool.call.started',
+            'tool.call.completed',
+            'message.delta',
+            'message.completed',
+            'processor.log',
+            'run.completed',
+        ],
+    )
+    async def test_debug_agent_runs_configured_runner_with_synthetic_event(self, streaming, result_type):
         app = _make_app()
         agent_config = _agent_row().config
         agent_config['allowed_platform_tools'] = ['platform_get_user_info']
@@ -162,7 +173,7 @@ class TestAgentServiceDebug:
         agent_config['allowed_tools'] = ['exec', 'weather']
 
         visible_event = {
-            'type': 'tool.call.started',
+            'type': result_type,
             'data': {'tool_name': 'exec', 'parameters': {'command': 'echo hi'}},
         }
         observer = AsyncMock() if streaming else None
@@ -171,6 +182,9 @@ class TestAgentServiceDebug:
             assert binding.delivery_policy.enable_streaming is streaming
             await adapter_context['_result_observer']({**visible_event, 'private_context': 'must not leak'})
             await adapter_context['_result_observer']({'type': 'state.updated', 'data': {'private': True}})
+            if streaming:
+                # Debug events reach the client before the runner returns its final output.
+                observer.assert_awaited_once_with(visible_event)
             yield SimpleNamespace(
                 role='assistant',
                 content='debug result',

tests/unit_tests/api/test_bots_controller_errors.py

@@ -0,0 +1,138 @@
+from __future__ import annotations
+
+import sys
+import types
+from importlib import import_module
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+import quart
+
+
+core_app_module = types.ModuleType('langbot.pkg.core.app')
+core_app_module.Application = object
+sys.modules.setdefault('langbot.pkg.core.app', core_app_module)
+
+
+pytestmark = pytest.mark.asyncio
+
+
+async def _create_test_client(bot_service: SimpleNamespace):
+    app = quart.Quart(__name__)
+    account = SimpleNamespace(
+        uuid='account-test',
+        user='test@example.com',
+    )
+    user_service = SimpleNamespace(
+        get_authenticated_account=AsyncMock(return_value=account),
+    )
+    access = SimpleNamespace(
+        workspace=SimpleNamespace(uuid='workspace-test'),
+        membership=SimpleNamespace(
+            uuid='membership-test',
+            role='developer',
+            projection_revision=1,
+        ),
+        execution=SimpleNamespace(
+            instance_uuid='instance-test',
+            placement_generation=1,
+        ),
+    )
+    ap = SimpleNamespace(
+        bot_service=bot_service,
+        user_service=user_service,
+        apikey_service=SimpleNamespace(authenticate_api_key=AsyncMock(return_value=None)),
+        workspace_collaboration_service=SimpleNamespace(resolve_account_workspace=AsyncMock(return_value=access)),
+    )
+    BotsRouterGroup = import_module('langbot.pkg.api.http.controller.groups.platform.bots').BotsRouterGroup
+    group = BotsRouterGroup(ap, app)
+    await group.initialize()
+    return app.test_client()
+
+
+@pytest.mark.parametrize('method,path', [('post', '/api/v1/platform/bots'), ('put', '/api/v1/platform/bots/bot-1')])
+async def test_bot_config_error_preserves_details(method, path):
+    error = ValueError('Lark missing required config: app_id, app_secret, bot_name')
+    service = SimpleNamespace(create_bot=AsyncMock(side_effect=error), update_bot=AsyncMock(side_effect=error))
+    client = await _create_test_client(service)
+    response = await getattr(client, method)(
+        path, json={'adapter_config': {}}, headers={'Authorization': 'Bearer token'}
+    )
+    assert response.status_code == 400
+    body = await response.get_json()
+    assert body['code'] == 'invalid_bot_config'
+    assert body['msg'] == str(error)
+
+
+async def test_bot_apply_failure_identifies_persisted_record():
+    from langbot.pkg.api.http.service.bot_errors import BotApplyError
+
+    service = SimpleNamespace(create_bot=AsyncMock(side_effect=BotApplyError('Missing app_id', 'saved-bot')))
+    client = await _create_test_client(service)
+    response = await client.post('/api/v1/platform/bots', json={}, headers={'Authorization': 'Bearer token'})
+    assert response.status_code == 400
+    body = await response.get_json()
+    assert body.pop('request_id')
+    assert body == {
+        'code': 'bot_apply_failed',
+        'msg': 'Missing app_id',
+        'data': {'uuid': 'saved-bot'},
+    }
+
+
+async def test_unexpected_failure_keeps_request_reference_without_exception_details():
+    client = await _create_test_client(
+        SimpleNamespace(update_bot=AsyncMock(side_effect=RuntimeError('database password')))
+    )
+    response = await client.put('/api/v1/platform/bots/bot-1', json={}, headers={'Authorization': 'Bearer token'})
+    body = await response.get_json()
+    assert response.status_code == 500
+    assert body['request_id']
+    assert 'database password' not in str(body)
+
+
+async def test_invalid_request_body_is_actionable():
+    service = SimpleNamespace(update_bot=AsyncMock())
+    client = await _create_test_client(service)
+    response = await client.put('/api/v1/platform/bots/bot-1', json=[], headers={'Authorization': 'Bearer token'})
+    assert response.status_code == 400
+    service.update_bot.assert_not_awaited()
+
+
+async def test_runtime_error_redacts_persisted_secrets_on_partial_update():
+    from langbot.pkg.api.http.service.bot import BotService
+    from langbot.pkg.api.http.service.bot_errors import BotApplyError
+
+    ap = SimpleNamespace(
+        persistence_mgr=SimpleNamespace(execute_async=AsyncMock(return_value=SimpleNamespace(rowcount=1))),
+        platform_mgr=SimpleNamespace(
+            remove_bot=AsyncMock(),
+            load_bot=AsyncMock(side_effect=ValueError('Invalid app_secret: persisted-secret-value')),
+        ),
+    )
+    service = BotService(ap)
+    service.get_bot = AsyncMock(
+        return_value={'uuid': 'bot-1', 'adapter_config': {'app_secret': 'persisted-secret-value'}}
+    )
+    with pytest.raises(BotApplyError) as captured:
+        await service.update_bot('workspace-test', 'bot-1', {'enable': True})
+    assert str(captured.value) == 'Invalid app_secret: ***'
+    assert captured.value.bot_uuid == 'bot-1'
+    ap.persistence_mgr.execute_async.assert_awaited_once()
+
+
+async def test_validation_error_does_not_include_input_values():
+    import pydantic
+    from langbot.pkg.api.http.service.bot_errors import bot_error_message
+
+    class Config(pydantic.BaseModel):
+        app_secret: int
+
+    try:
+        Config(app_secret='private-value')
+    except pydantic.ValidationError as error:
+        result = bot_error_message(error, {'app_secret': 'private-value'})
+    assert 'app_secret' in result
+    assert 'private-value' not in result
+    assert 'input_value' not in result

tests/unit_tests/platform/test_routing_rules.py

@@ -447,7 +447,7 @@ class TestEBAEventBindings:
             'event_get_actor',
         ]
         assert binding.delivery_policy.enable_streaming is False
-        assert binding.delivery_policy.enable_reply is True
+        assert binding.delivery_policy.enable_reply is False
         assert binding.delivery_policy.enable_interactions is True
         assert binding.state_policy.state_scopes == ['conversation', 'actor', 'subject', 'runner']
         assert binding.agent_id == 'agent-1'
@@ -651,7 +651,9 @@ class TestInteractionResumeRouting:
         assert binding.processor_type == 'agent'
         assert binding.processor_id == 'agent-1'
         assert adapter_context == {'_delivery_adapter': adapter}
-        adapter.send_message.assert_awaited_once()
+        assert envelope.delivery.supports_streaming is False
+        assert binding.delivery_policy.enable_reply is False
+        adapter.send_message.assert_not_awaited()
 
     def test_agent_product_to_binding_does_not_fallback_to_component_ref(self):
         """An empty config runner stays unconfigured even if component_ref is stale."""
@@ -762,3 +764,89 @@ async def test_bound_event_processor_receives_one_complete_typed_event():
     assert envelope.data['type'] == 'group.member_joined'
     assert 'source_platform_object' not in envelope.data
     bot.ap.plugin_connector.emit_event.assert_not_called()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('kind', ['agent', 'event_processor'])
+@pytest.mark.parametrize('output_kind', ['message', 'chunks', 'tool_rounds'])
+@pytest.mark.parametrize('explicit_reply', [False, True])
+@pytest.mark.parametrize('runner_fails', [False, True])
+async def test_processor_outputs_require_explicit_platform_actions(kind, output_kind, explicit_reply, runner_fails):
+    """Draining runner results must not send text, duplicate replies, or stream cards."""
+    from langbot_plugin.api.entities.builtin.platform import entities, events, message
+    from langbot_plugin.api.entities.builtin.provider import message as provider_message
+
+    from langbot.pkg.agent.runner.platform_tools import execute_platform_tool, freeze_platform_context
+
+    bot = TestEventRouteTrace._make_bot(
+        [{'id': 'binding-1', 'event_pattern': 'message.received', 'target_type': kind, 'target_uuid': 'agent-1'}]
+    )
+    adapter = SimpleNamespace(
+        get_supported_apis=lambda: ['send_message'],
+        is_stream_output_supported=AsyncMock(return_value=True),
+        send_message=AsyncMock(return_value='message-2'),
+        create_message_card=AsyncMock(),
+        reply_message_chunk=AsyncMock(),
+    )
+    completed = []
+
+    async def run(envelope, binding, adapter_context):
+        assert envelope.delivery.supports_streaming is False
+        assert binding.delivery_policy.enable_streaming is False
+        assert binding.delivery_policy.enable_reply is False
+        assert adapter_context['_delivery_adapter'] is adapter
+        if explicit_reply:
+            session = {'authorization': {'bot_id': 'bot-1', 'platform_context': freeze_platform_context(envelope)}}
+            await execute_platform_tool(bot.ap, TEST_CONTEXT, session, 'event_reply', {'text': 'Working on it'})
+            # Progress arrives while the runner is still working.
+            adapter.send_message.assert_awaited_once()
+            assert completed == []
+        if output_kind == 'chunks':
+            yield provider_message.MessageChunk(role='assistant', content='Done', all_content='Done')
+            yield provider_message.MessageChunk(role='assistant', content='.', all_content='Done.', is_final=True)
+        elif output_kind == 'tool_rounds':
+            yield provider_message.Message(role='assistant', content='Checking the request')
+            yield provider_message.Message(role='assistant', content='Done.')
+        else:
+            yield provider_message.Message(role='assistant', content='Done.')
+        if runner_fails:
+            raise RuntimeError('Runner failed after producing text')
+        completed.append(True)
+
+    bot.ap = SimpleNamespace(
+        workspace_service=active_workspace_service(),
+        platform_mgr=SimpleNamespace(get_bot_by_uuid=AsyncMock(return_value=SimpleNamespace(adapter=adapter))),
+        agent_service=SimpleNamespace(
+            get_agent=AsyncMock(
+                return_value={
+                    'uuid': 'agent-1',
+                    'kind': kind,
+                    'enabled': True,
+                    'supported_event_patterns': ['message.received'],
+                    'config': {'runner': {'id': 'runner-1'}, 'runner_config': {'runner-1': {}}},
+                }
+            )
+        ),
+        agent_run_orchestrator=SimpleNamespace(run=run),
+    )
+    event = events.MessageReceivedEvent(
+        message_id='message-1',
+        message_chain=message.MessageChain([message.Plain(text='hello')]),
+        sender=entities.User(id='user-1', nickname='QA'),
+        chat_type=entities.ChatType.PRIVATE,
+        chat_id='user-1',
+    )
+    trace = await bot._dispatch_eba_event_to_processor(event, adapter)
+
+    assert trace['status'] == ('failed' if runner_fails else 'delivered')
+    if runner_fails:
+        assert trace['failure_code'] == 'runner_failed'
+    assert completed == ([] if runner_fails else [True])
+    assert adapter.send_message.await_count == int(explicit_reply)
+    if explicit_reply:
+        kwargs = adapter.send_message.await_args.kwargs
+        assert kwargs['target_type'] == 'person'
+        assert kwargs['target_id'] == 'user-1'
+        assert kwargs['message'][0].text == 'Working on it'
+    adapter.create_message_card.assert_not_awaited()
+    adapter.reply_message_chunk.assert_not_awaited()

tests/unit_tests/plugin/test_handler_actions.py

@@ -1575,9 +1575,7 @@ class TestAgentRunProxyActions:
         query = build_execution_query(event, [])
         app.box_service = SimpleNamespace(
             available=True,
-            get_backend_status=AsyncMock(
-                return_value={'backend': {'available': True}}
-            ),
+            get_backend_status=AsyncMock(return_value={'backend': {'available': True}}),
             execute_tool=AsyncMock(
                 return_value={
                     'ok': True,
@@ -1683,3 +1681,88 @@ class TestAgentRunProxyActions:
         provider.invoke_rerank.assert_awaited_once()
         kwargs = provider.invoke_rerank.await_args.kwargs
         assert kwargs['extra_args'] == {'top_n': 2, 'return_documents': False}
+
+
+@pytest.mark.parametrize(
+    'case',
+    [
+        'valid',
+        'expired',
+        'plugin',
+        'workspace',
+        'permission',
+        'shadowed',
+        'native',
+        'bot_removed',
+        'adapter_replaced',
+        'api_revoked',
+    ],
+)
+@pytest.mark.skipif(not hasattr(PluginToRuntimeAction, 'REPLY_STREAM'), reason='SDK does not support streaming replies')
+async def test_reply_stream_authorization_is_run_and_workspace_scoped(case):
+    from uuid import uuid4
+    from langbot.pkg.agent.runner.session_registry import get_session_registry
+
+    app = SimpleNamespace(logger=Mock(), _test_plugin_identity='test/runner')
+    runtime_handler = make_handler(app)
+    query = SimpleNamespace(
+        **{
+            field: getattr(TEST_EXECUTION_CONTEXT, field)
+            for field in ('instance_uuid', 'workspace_uuid', 'placement_generation')
+        }
+    )
+    if case == 'workspace':
+        query.workspace_uuid = 'another-workspace'
+    streams = SimpleNamespace(mock=True, apply=AsyncMock(return_value={'status': 'completed'}))
+    if case in {'native', 'bot_removed', 'adapter_replaced', 'api_revoked'}:
+        streams.mock = False
+        streams.adapter = SimpleNamespace(get_supported_apis=lambda: [] if case == 'api_revoked' else ['send_message'])
+        bot = (
+            None
+            if case == 'bot_removed'
+            else SimpleNamespace(adapter=object() if case == 'adapter_replaced' else streams.adapter)
+        )
+        app.platform_mgr = SimpleNamespace(get_bot_by_uuid=AsyncMock(return_value=bot))
+    registry = get_session_registry()
+    run_id = str(uuid4())
+    resources = make_agent_resources(
+        tools=[]
+        if case == 'permission'
+        else [
+            {
+                'tool_name': 'event_reply',
+                'operations': ['call'],
+                'source': 'mcp' if case == 'shadowed' else 'platform',
+                'source_id': 'event_reply',
+            }
+        ]
+    )
+    await registry.register(
+        run_id=run_id,
+        runner_id='plugin:test/runner/default',
+        query_id=None,
+        plugin_identity='other/plugin' if case == 'plugin' else 'test/runner',
+        resources=resources,
+        execution_query=query,
+        reply_streams=streams,
+    )
+    try:
+        if case == 'expired':
+            await registry.unregister(run_id)
+        response = await runtime_handler.actions[PluginToRuntimeAction.REPLY_STREAM.value](
+            {
+                'run_id': run_id,
+                'caller_plugin_identity': 'test/runner',
+                'stream_id': str(uuid4()),
+                'operation': 'finish',
+                'text': 'hello',
+            }
+        )
+        if case in {'valid', 'native'}:
+            assert response.code == 0
+            streams.apply.assert_awaited_once()
+        else:
+            assert response.code != 0
+            streams.apply.assert_not_awaited()
+    finally:
+        await registry.unregister(run_id)