docs(space): sync Runner usage recommendation contract

Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>

.github/workflows/publish-to-pypi.yml

@@ -2,6 +2,11 @@ name: Build and Publish to PyPI
 
 on:
   workflow_dispatch:
+    inputs:
+      source_ref:
+        description: 'Existing release tag to publish (for example v4.10.11)'
+        required: true
+        type: string
   release:
     types: [published]
 
@@ -11,13 +16,39 @@ jobs:
     permissions:
       contents: read
       id-token: write  # Required for trusted publishing to PyPI
-    
+
     steps:
       - name: Checkout code
         uses: actions/checkout@v4
         with:
+          ref: ${{ inputs.source_ref || github.sha }}
+          fetch-depth: 0
           persist-credentials: false
 
+      - name: Validate release source and version
+        env:
+          RELEASE_TAG: ${{ inputs.source_ref || github.event.release.tag_name }}
+        run: |
+          python3 - <<'PY'
+          import os
+          import re
+          import subprocess
+          import tomllib
+          from pathlib import Path
+
+          tag = os.environ['RELEASE_TAG']
+          if not re.fullmatch(r'v[0-9]+\.[0-9]+\.[0-9]+', tag):
+              raise SystemExit('source_ref must be an existing release tag: vX.Y.Z')
+          def revision(ref):
+              return subprocess.check_output(['git', 'rev-parse', '--verify', ref], text=True).strip()
+          if revision('HEAD') != revision(f'refs/tags/{tag}^{{}}'):
+              raise SystemExit('Checked-out commit does not match the release tag')
+          version = tomllib.loads(Path('pyproject.toml').read_text())['project']['version']
+          if version != tag[1:]:
+              raise SystemExit(f'Package version {version} does not match tag {tag}')
+          print(f'Validated {tag} at {revision("HEAD")} (package {version})')
+          PY
+
       - name: Set up Node.js
         uses: actions/setup-node@v4
         with:
@@ -26,9 +57,9 @@ jobs:
       - name: Build frontend
         run: |
           cd web
-          npm install -g pnpm
-          pnpm install
-          pnpm build
+          # Match the archive/Docker npm path; npm ci rejects older tags' stale npm lockfiles.
+          npm install --include=optional
+          npm run build
           mkdir -p ../src/langbot/web/dist
           cp -r dist ../src/langbot/web/
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "langbot"
-version = "4.10.10"
+version = "4.10.11"
 description = "Production-grade platform for building agentic IM bots"
 readme = "README.md"
 license-files = ["LICENSE"]
@@ -70,7 +70,7 @@ dependencies = [
     "langchain-text-splitters>=1.1.2",
     "chromadb>=1.0.0,<2.0.0",
     "qdrant-client (>=1.15.1,<2.0.0)",
-    "langbot-plugin==0.5.7",
+    "langbot-plugin==0.5.8",
     "asyncpg>=0.30.0",
     "line-bot-sdk>=3.19.0",
     "matrix-nio>=0.25.2",

skills/skills/langbot-space-ops/SKILL.md

@@ -25,7 +25,10 @@ CLI uses. Create one in your Space account (Profile → Personal Access Tokens),
 then send it as a Bearer token:
 
 ```
-Authorization: Bearer lbpat_...uests without a valid PAT get `401 Unauthorized`.
+Authorization: Bearer <your-pat>
+```
+
+Requests without a valid PAT get `401 Unauthorized`.
 
 ## Client configuration
 
@@ -66,6 +69,36 @@ All tools are read-only.
    state (available, unprobed, unavailable), then Space recommendation. Each
    item includes `availability.up`, `last_probed_at`, latency, and HTTP status.
 
+## Runner usage recommendations
+
+Use `search_plugins` with `runner_usage: "agent"` for Agent, pipeline, and
+setup-wizard recommendations, or `runner_usage: "event"` for event processors.
+The component kind remains `Runner`. Only these two exact values are accepted;
+omit the optional field to preserve unfiltered browsing.
+
+```json
+{"query":"", "runner_usage":"agent", "page":1, "page_size":100}
+```
+
+Plugin results include `latest_version` and `runner_usages: string[]`, the
+explicit union of usages in that latest installable version. Only recommend a
+plugin when this array explicitly contains the target usage. Missing, empty,
+malformed, or unknown usages must never mean agent-compatible. Event-only
+plugins must never enter Agent recommendations. Empty filtered results are
+valid while legacy packages await corrected releases; never remove the filter
+to fill a recommendation list.
+
+REST callers use `runner_usage` on both
+`POST /api/v1/marketplace/extensions/search` and the compatibility
+`POST /api/v1/marketplace/plugins/search`; preserve it during fallback. Add
+`"type_filter":"plugin", "component_filter":"Runner"` on the unified endpoint.
+Usage is ANDed with other filters before pagination and `total`; MCP/Skill items
+do not match. Invalid REST values return HTTP 400.
+
+Open the same filter in the webpage:
+`https://space.langbot.app/market?type=plugin&component=Runner&runner_usage=agent`
+(or `runner_usage=event`). Switch All / Agent / Event in the Runner usage row.
+
 ## Implementation & maintenance (for Space developers)
 
 - Server: `internal/controller/mcp/server.go` (official Go MCP SDK

src/langbot/pkg/plugin/handler.py

@@ -48,6 +48,7 @@ from ..utils import constants
 
 _DEFAULT_BINARY_STORAGE_VALUE_BYTES = 10 * 1024 * 1024
 _HARD_MAX_BINARY_STORAGE_VALUE_BYTES = 64 * 1024 * 1024
+_UNSET_INSTALLATION_SCOPE = object()
 
 
 def _binary_storage_value_limit(ap: Any) -> int:
@@ -479,7 +480,6 @@ class RuntimeConnectionHandler(handler.Handler):
         self._outbound_installation_context: contextvars.ContextVar[InstallationBinding | None] = (
             contextvars.ContextVar(
                 f'{self.__class__.__name__}_{id(self)}_outbound_installation',
-                default=None,
             )
         )
         self._installation_bindings: dict[
@@ -1631,13 +1631,15 @@ class RuntimeConnectionHandler(handler.Handler):
     ) -> InstallationBinding | ActionContext | None:
         if action_context is not None:
             return super().resolve_outbound_action_context(action_context)
-        inbound_context = self.current_action_context
-        if inbound_context is not None:
-            return inbound_context
-        return self._outbound_installation_context.get()
+        # An explicit scope targets the nested call, not its inbound caller.
+        # None deliberately clears the context for runtime-scoped actions.
+        scoped_context = self._outbound_installation_context.get(_UNSET_INSTALLATION_SCOPE)
+        if scoped_context is not _UNSET_INSTALLATION_SCOPE:
+            return typing.cast(InstallationBinding | None, scoped_context)
+        return self.current_action_context
 
     def require_outbound_installation_context(self) -> InstallationBinding:
-        binding = self._outbound_installation_context.get()
+        binding = self._outbound_installation_context.get(None)
         if not isinstance(binding, InstallationBinding):
             raise ValueError('Host plugin action requires an InstallationBinding scope')
         return binding

tests/integration/plugin/test_rag_file_transfer_protocol.py

@@ -0,0 +1,307 @@
+"""Real Core/SDK protocol regression tests; no subprocesses or external services.
+
+Run against the intended local SDK (``uv run --no-sync`` after local install).
+The in-memory transport carries JSON strings through Handler.run on both sides;
+send_file, envelope validation, base64 decoding and transfer storage are real.
+Only Core's database/object-storage services, parser dispatch/provider and host
+sandbox prerequisite probing are doubles. Worker launch/registration is
+represented by its already-registered state.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from contextlib import asynccontextmanager
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from langbot.pkg.plugin.handler import RuntimeConnectionHandler
+from langbot_plugin.entities.io.actions.enums import CommonAction, LangBotToRuntimeAction, PluginToRuntimeAction
+from langbot_plugin.entities.io.context import ActionContext, InstallationBinding, PluginWorkerPolicy, RuntimeIdentity
+from langbot_plugin.runtime.context import RuntimeContext
+from langbot_plugin.runtime.io.connection import Connection
+from langbot_plugin.entities.io.errors import ActionCallError, ConnectionClosedError
+from langbot_plugin.runtime.io.handler import FILE_CHUNK_LENGTH, Handler
+from langbot_plugin.runtime.io.handlers.control import ControlConnectionHandler
+from langbot_plugin.runtime.io.handlers.plugin import PluginConnectionHandler
+from langbot_plugin.runtime.plugin.mgr import PluginManager
+from langbot_plugin.runtime.security import PLUGIN_FILE_STORAGE_DIR_ENV
+
+
+pytestmark = pytest.mark.asyncio
+PAYLOAD = bytes(range(256)) * 161 + b'\x00original RAG file\xff'
+BINDING = InstallationBinding(
+    instance_uuid='instance-a',
+    workspace_uuid='workspace-a',
+    placement_generation=7,
+    installation_uuid='00000000-0000-4000-8000-000000000001',
+    runtime_revision=3,
+    artifact_digest='a' * 64,
+)
+LEGACY = ActionContext(**BINDING.model_dump(exclude={'runtime_revision', 'artifact_digest'}))
+
+
+class QueueConnection(Connection):
+    """Only the byte transport is replaced, not the request/response machinery."""
+
+    def __init__(self):
+        self.incoming = asyncio.Queue()
+        self.sent = []
+        self.peer = None
+
+    async def send(self, message: str) -> None:
+        assert isinstance(message, str)
+        self.sent.append(json.loads(message))
+        await self.peer.incoming.put(message)
+
+    async def receive(self) -> str:
+        message = await self.incoming.get()
+        if message is None:
+            raise ConnectionClosedError('test transport closed')
+        return message
+
+    async def close(self) -> None:
+        await self.incoming.put(None)
+        await self.peer.incoming.put(None)
+
+
+def connection_pair():
+    left, right = QueueConnection(), QueueConnection()
+    left.peer, right.peer = right, left
+    return left, right
+
+
+@asynccontextmanager
+async def protocol_stack(tmp_path, monkeypatch, profile='oss_dev', binding=LEGACY):
+    monkeypatch.chdir(tmp_path)
+    stored = tmp_path / 'original.bin'
+    stored.write_bytes(PAYLOAD)
+    storage_calls = []
+
+    async def get_file_stream(execution_context, storage_path):
+        storage_calls.append((execution_context, storage_path))
+        assert execution_context.workspace_uuid == BINDING.workspace_uuid
+        assert storage_path == 'knowledge/original.bin'
+        return stored.read_bytes()
+
+    async def get_execution_binding(workspace_uuid, expected_generation):
+        assert workspace_uuid == BINDING.workspace_uuid
+        assert expected_generation == BINDING.placement_generation
+        return BINDING
+
+    setting = SimpleNamespace(
+        plugin_author='tester',
+        plugin_name='engine',
+        installation_uuid=BINDING.installation_uuid,
+        runtime_revision=BINDING.runtime_revision,
+        artifact_digest=BINDING.artifact_digest,
+    )
+    app = SimpleNamespace(
+        deployment=SimpleNamespace(mode='oss' if profile == 'oss_dev' else 'cloud'),
+        logger=logging.getLogger(__name__),
+        persistence_mgr=SimpleNamespace(execute_async=AsyncMock(return_value=SimpleNamespace(first=lambda: setting))),
+        workspace_service=SimpleNamespace(get_execution_binding=get_execution_binding),
+        rag_runtime_service=SimpleNamespace(get_file_stream=get_file_stream),
+    )
+    core_conn, control_conn = connection_pair()
+    monkeypatch.setenv(PLUGIN_FILE_STORAGE_DIR_ENV, str(tmp_path / 'core-transfer'))
+    core = RuntimeConnectionHandler(core_conn, AsyncMock(return_value=False), app)
+    core.register_installation_binding(BINDING, plugin_author='tester', plugin_name='engine')
+    runtime = RuntimeContext()
+    runtime.plugin_mgr = PluginManager(runtime)
+    # No worker is launched: omit only host nsjail/cgroup prerequisite probing.
+    monkeypatch.setattr(runtime.plugin_mgr.worker_launcher, 'configure', lambda policy, profile: None)
+    monkeypatch.setenv(PLUGIN_FILE_STORAGE_DIR_ENV, str(tmp_path / 'runtime-transfer'))
+    control = ControlConnectionHandler(control_conn, runtime)
+    runtime.activate_control_handler(control)
+    bridge_conn, plugin_conn = connection_pair()
+    bridge = PluginConnectionHandler(bridge_conn, runtime, file_storage_dir=str(tmp_path / 'bridge-transfer'))
+    plugin = Handler(plugin_conn, file_storage_dir=str(tmp_path / 'plugin-transfer'))
+    # Trusted state left by registration, not plugin-supplied action data.
+    bridge.bind_action_context(binding)
+    runtime.plugin_mgr.plugin_handlers.append(bridge)
+    runtime.plugin_mgr.plugins.append(SimpleNamespace(_runtime_plugin_handler=bridge))
+    handlers = [core, control, bridge, plugin]
+    tasks = [asyncio.create_task(handler.run()) for handler in handlers]
+    try:
+        await asyncio.wait_for(
+            core.set_runtime_config(
+                runtime_identity=RuntimeIdentity(instance_uuid='instance-a', runtime_id='test-runtime'),
+                worker_policy=PluginWorkerPolicy(
+                    max_cpus=1,
+                    max_memory_mb=128,
+                    max_pids=32,
+                    max_open_files=64,
+                    max_file_size_mb=8,
+                    require_hard_limits=False,
+                ),
+                runtime_profile=profile,
+                cloud_service_url=None,
+            ),
+            5,
+        )
+        if isinstance(binding, InstallationBinding):
+            runtime.activate_installation_binding(binding)
+        else:
+            runtime.bind_workspace(binding)
+        yield SimpleNamespace(
+            core=core,
+            control=control,
+            runtime=runtime,
+            bridge=bridge,
+            plugin=plugin,
+            core_conn=core_conn,
+            control_conn=control_conn,
+            bridge_conn=bridge_conn,
+            plugin_conn=plugin_conn,
+            app=app,
+            storage_calls=storage_calls,
+        )
+    finally:
+        for handler in handlers:
+            await handler.close()
+        await asyncio.wait_for(asyncio.gather(*tasks, return_exceptions=True), 5)
+
+
+def assert_chunks(connection, binding, payload=PAYLOAD):
+    chunks = [message for message in connection.sent if message.get('action') == CommonAction.FILE_CHUNK.value]
+    expected = (len(payload) + FILE_CHUNK_LENGTH - 1) // FILE_CHUNK_LENGTH
+    assert expected > 1
+    assert len(chunks) == expected
+    assert [chunk['data']['chunk_index'] for chunk in chunks] == list(range(expected))
+    assert {chunk['data']['chunk_amount'] for chunk in chunks} == {expected}
+    assert all(chunk['context'] == binding.model_dump() for chunk in chunks)
+    assert len({chunk['data']['file_key'] for chunk in chunks}) == 1
+    return chunks[0]['data']['file_key']
+
+
+@pytest.mark.parametrize(
+    'profile,binding',
+    [('oss_dev', LEGACY), ('oss_dev', BINDING), ('shared', BINDING)],
+    ids=['legacy-oss', 'managed-oss', 'managed-shared'],
+)
+async def test_knowledge_file_roundtrip_reaches_plugin_original_bytes(tmp_path, monkeypatch, profile, binding):
+    async with protocol_stack(tmp_path, monkeypatch, profile, binding) as stack:
+        # Legacy plugin API sends no authority; Runtime supplies its trusted binding.
+        result = await asyncio.wait_for(
+            stack.plugin.call_action(
+                PluginToRuntimeAction.GET_KNOWLEDEGE_FILE_STREAM,
+                {'storage_path': 'knowledge/original.bin'},
+            ),
+            5,
+        )
+        assert await stack.plugin.read_local_file(result['file_key']) == PAYLOAD
+        assert len(stack.storage_calls) == 1
+        core_key = assert_chunks(stack.core_conn, binding)
+        plugin_key = assert_chunks(stack.bridge_conn, binding)
+        assert result['file_key'] == plugin_key != core_key
+        assert not (Path(stack.control.file_storage_dir) / core_key).exists()
+        assert not stack.control._owned_transfer_files
+        callbacks = [
+            message
+            for message in stack.control_conn.sent
+            if message.get('action') == PluginToRuntimeAction.GET_KNOWLEDEGE_FILE_STREAM.value
+        ]
+        assert len(callbacks) == 1
+        assert callbacks[0]['context'] == binding.model_dump()
+        assert callbacks[0]['data'] == {'storage_path': 'knowledge/original.bin'}
+
+
+async def test_shared_control_rejects_legacy_chunks_before_storage(tmp_path, monkeypatch):
+    async with protocol_stack(tmp_path, monkeypatch, 'shared', BINDING) as stack:
+        with stack.core.installation_scope(LEGACY):
+            with pytest.raises(ActionCallError, match='InstallationBinding|Legacy FILE_CHUNK'):
+                await asyncio.wait_for(stack.core.send_file(PAYLOAD, ''), 5)
+        assert not list(Path(stack.control.file_storage_dir).iterdir())
+        assert not stack.control._owned_transfer_files
+
+
+async def test_candidate_artifact_pretransfer_does_not_require_active_installation(tmp_path, monkeypatch):
+    async with protocol_stack(tmp_path, monkeypatch, 'shared', BINDING) as stack:
+        candidate = BINDING.model_copy(
+            update={'installation_uuid': 'candidate-installation', 'runtime_revision': 1, 'artifact_digest': 'c' * 64}
+        )
+        assert not stack.runtime.is_current_installation_binding(candidate)
+        with stack.core.installation_scope(candidate):
+            key = await asyncio.wait_for(stack.core.send_file(PAYLOAD, 'lbp'), 5)
+        assert_chunks(stack.core_conn, candidate)
+        assert await stack.control.read_local_file(key) == PAYLOAD
+        assert not stack.runtime.is_current_installation_binding(candidate)
+
+
+async def test_nested_parser_target_owns_file_and_action_envelopes(tmp_path, monkeypatch):
+    async with protocol_stack(tmp_path, monkeypatch, 'shared', BINDING) as stack:
+        target = BINDING.model_copy(
+            update={
+                'installation_uuid': 'parser-installation',
+                'runtime_revision': 2,
+                'artifact_digest': 'b' * 64,
+            }
+        )
+        stack.runtime.activate_installation_binding(target)
+        parser_calls = []
+        restored = []
+
+        async def parse_document(author, name, context_data, file_bytes):
+            parser_calls.append((stack.control.current_action_context, author, name, context_data, file_bytes))
+            return {'documents': [{'text': 'parsed'}]}
+
+        stack.runtime.plugin_mgr.parse_document = parse_document
+
+        class ParserConnector:
+            async def require_workspace_context(self, context):
+                assert context.workspace_uuid == BINDING.workspace_uuid
+
+            async def call_parser(self, plugin_name, context_data, file_bytes):
+                assert plugin_name == 'tester/parser'
+                assert stack.core.current_action_context == BINDING
+                with stack.core.installation_scope(target):
+                    result = await stack.core.parse_document('tester', 'parser', context_data, file_bytes)
+                restored.append(stack.core.resolve_outbound_action_context(None))
+                return result
+
+        stack.app.plugin_connector = ParserConnector()
+        result = await asyncio.wait_for(
+            stack.plugin.call_action(
+                PluginToRuntimeAction.INVOKE_PARSER,
+                {
+                    'plugin_author': 'tester',
+                    'plugin_name': 'parser',
+                    'storage_path': 'knowledge/original.bin',
+                    'filename': 'original.bin',
+                },
+            ),
+            5,
+        )
+        assert result == {'documents': [{'text': 'parsed'}]}
+        key = assert_chunks(stack.core_conn, target)
+        parse_requests = [
+            message
+            for message in stack.core_conn.sent
+            if message.get('action') == LangBotToRuntimeAction.PARSE_DOCUMENT.value
+        ]
+        assert len(parse_requests) == 1
+        assert parse_requests[0]['context'] == target.model_dump()
+        assert parse_requests[0]['data']['context']['file_key'] == key
+        assert parser_calls == [
+            (
+                target,
+                'tester',
+                'parser',
+                {
+                    'mime_type': 'application/octet-stream',
+                    'filename': 'original.bin',
+                    'metadata': {},
+                },
+                PAYLOAD,
+            )
+        ]
+        assert restored == [BINDING]
+        assert stack.core.current_action_context is None
+        assert stack.core.resolve_outbound_action_context(None) is None
+        assert not (Path(stack.control.file_storage_dir) / key).exists()

tests/unit_tests/plugin/test_handler_invocation_scope.py

@@ -0,0 +1,193 @@
+"""Exercise nested installation routing through real Core/SDK wire envelopes."""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from langbot_plugin.entities.io.actions.enums import CommonAction, LangBotToRuntimeAction, PluginToRuntimeAction
+from langbot_plugin.entities.io.req import ActionRequest
+from langbot_plugin.entities.io.resp import ActionResponse
+from langbot_plugin.runtime.io import handler as sdk_handler
+
+from langbot.pkg.plugin.connector import PluginRuntimeConnector
+from tests.unit_tests.plugin.test_handler_tenancy import RecordingConnection, make_handler, workspace_context
+
+
+class ReplyingConnection(RecordingConnection):
+    """Replace only the transport, retaining serialization and response routing."""
+
+    async def send(self, message: str) -> None:
+        await super().send(message)
+        request = json.loads(message)
+        if 'action' in request:
+            response = ActionResponse.success({'elements': []})
+            response.seq_id = request['seq_id']
+            await self.handler._route_response(response.seq_id, response.model_dump())
+
+    @property
+    def requests(self):
+        return [request for message in self.sent if 'action' in (request := json.loads(message))]
+
+
+@pytest.fixture
+def bridge(monkeypatch):
+    runtime_handler, app, binding_a = make_handler()
+    connection = ReplyingConnection()
+    connection.handler = runtime_handler
+    runtime_handler.conn = connection
+    monkeypatch.setattr(sdk_handler, 'FILE_CHUNK_LENGTH', 4)
+    binding_b = binding_a.model_copy(
+        update={
+            'installation_uuid': '00000000-0000-4000-8000-000000000002',
+            'runtime_revision': 2,
+            'artifact_digest': 'b' * 64,
+        }
+    )
+    return runtime_handler, app, connection, binding_a, binding_b
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('mode', ['managed', 'legacy'])
+async def test_nested_invoke_parser_uses_target_for_every_chunk_and_parse(bridge, mode):
+    runtime_handler, app, connection, binding_a, binding_b = bridge
+    app.instance_config = SimpleNamespace(data={'plugin': {'enable': True}})
+    app.deployment.mode = 'cloud' if mode == 'managed' else 'oss'
+    connector = PluginRuntimeConnector(app, AsyncMock())
+    connector.handler = runtime_handler
+    app.plugin_connector = connector
+    execution_context = runtime_handler._execution_context(binding_a)
+    setting_b = SimpleNamespace(
+        installation_uuid=binding_b.installation_uuid,
+        runtime_revision=binding_b.runtime_revision,
+        artifact_digest=binding_b.artifact_digest,
+        install_info={'_artifact_storage': 'tenant_binary_storage_v1'} if mode == 'managed' else {},
+    )
+    connector._setting_for_plugin = AsyncMock(return_value=(execution_context, setting_b))
+    connector.require_workspace_context = AsyncMock(return_value=execution_context)
+    file_bytes = b'parser document'
+    app.rag_runtime_service = SimpleNamespace(get_file_stream=AsyncMock(return_value=file_bytes))
+    inbound_context = binding_a
+    if mode == 'legacy':
+        inbound_context = workspace_context().for_installation(binding_a.installation_uuid)
+        setting_a = SimpleNamespace(
+            plugin_author='author-a',
+            plugin_name='plugin-a',
+            installation_uuid=binding_a.installation_uuid,
+            runtime_revision=binding_a.runtime_revision,
+            artifact_digest=binding_a.artifact_digest,
+        )
+        app.persistence_mgr.execute_async.return_value = SimpleNamespace(first=lambda: setting_a)
+    expected = binding_b if mode == 'managed' else connector._legacy_oss_bridge_binding(execution_context)
+    request = ActionRequest.make_request(
+        101,
+        PluginToRuntimeAction.INVOKE_PARSER.value,
+        {'plugin_author': 'author-b', 'plugin_name': 'parser-b', 'storage_path': 'file-a'},
+        inbound_context,
+    )
+
+    await runtime_handler._handle_action(request.model_dump())
+
+    response = json.loads(connection.sent[-1])
+    assert response['code'] == 0, response
+    chunks = connection.requests[:-1]
+    parse = connection.requests[-1]
+    assert len(chunks) == 4
+    assert all(chunk['action'] == CommonAction.FILE_CHUNK.value for chunk in chunks)
+    assert parse['action'] == LangBotToRuntimeAction.PARSE_DOCUMENT.value
+    assert all(request['context'] == expected.model_dump() for request in connection.requests)
+    assert b''.join(base64.b64decode(chunk['data']['chunk_base64']) for chunk in chunks) == file_bytes
+    assert {chunk['data']['file_key'] for chunk in chunks} == {parse['data']['context']['file_key']}
+    connector._setting_for_plugin.assert_awaited_once_with('author-b', 'parser-b', require_enabled=True)
+    assert runtime_handler.current_action_context is None
+    assert runtime_handler.resolve_outbound_action_context(None) is None
+
+
+@pytest.mark.asyncio
+async def test_explicit_argument_overrides_scope_and_inbound_falls_back(bridge):
+    runtime_handler, _, connection, binding_a, binding_b = bridge
+    token = runtime_handler._current_action_context.set(binding_a)
+    try:
+        with runtime_handler.installation_scope(binding_b):
+            await runtime_handler.call_action(
+                LangBotToRuntimeAction.LIST_PARSERS, {}, action_context=binding_a.model_dump()
+            )
+        await runtime_handler.list_parsers()
+    finally:
+        runtime_handler._current_action_context.reset(token)
+    assert [request['context'] for request in connection.requests] == [binding_a.model_dump()] * 2
+    assert runtime_handler.resolve_outbound_action_context(None) is None
+
+
+@pytest.mark.asyncio
+async def test_explicit_none_scope_clears_inbound_and_restores_outer_scope(bridge):
+    runtime_handler, _, connection, binding_a, binding_b = bridge
+    token = runtime_handler._current_action_context.set(binding_a)
+    try:
+        with runtime_handler.installation_scope(binding_b):
+            await runtime_handler.ping()
+            await runtime_handler.list_parsers()
+        await runtime_handler.list_parsers()
+    finally:
+        runtime_handler._current_action_context.reset(token)
+    assert [request.get('context') for request in connection.requests] == [
+        None,
+        binding_b.model_dump(),
+        binding_a.model_dump(),
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('failure', [RuntimeError, asyncio.CancelledError])
+async def test_scope_restores_after_exception_or_cancellation(bridge, failure):
+    runtime_handler, _, connection, binding_a, binding_b = bridge
+    with runtime_handler.installation_scope(binding_a):
+        with pytest.raises(failure):
+            with runtime_handler.installation_scope(binding_b):
+                await runtime_handler.list_parsers()
+                raise failure()
+        await runtime_handler.list_parsers()
+    await runtime_handler.list_parsers()
+    assert [request.get('context') for request in connection.requests] == [
+        binding_b.model_dump(),
+        binding_a.model_dump(),
+        None,
+    ]
+
+
+@pytest.mark.asyncio
+async def test_concurrent_nested_scopes_do_not_leak_on_task_cancellation(bridge):
+    runtime_handler, _, connection, binding_a, binding_b = bridge
+    entered = asyncio.Event()
+    release = asyncio.Event()
+
+    async def cancelled_invocation():
+        with runtime_handler.installation_scope(binding_b):
+            await runtime_handler.list_parsers()
+            entered.set()
+            await release.wait()
+
+    token = runtime_handler._current_action_context.set(binding_a)
+    task = asyncio.create_task(cancelled_invocation())
+    try:
+        await asyncio.wait_for(entered.wait(), timeout=2)
+        with runtime_handler.installation_scope(None):
+            await runtime_handler.list_parsers()
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        await runtime_handler.list_parsers()
+    finally:
+        runtime_handler._current_action_context.reset(token)
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+    assert [request.get('context') for request in connection.requests] == [
+        binding_b.model_dump(),
+        None,
+        binding_a.model_dump(),
+    ]
+    assert runtime_handler.resolve_outbound_action_context(None) is None

uv.lock

@@ -2008,7 +2008,7 @@ wheels = [
 
 [[package]]
 name = "langbot"
-version = "4.10.10"
+version = "4.10.11"
 source = { editable = "." }
 dependencies = [
     { name = "aiocqhttp" },
@@ -2129,7 +2129,7 @@ requires-dist = [
     { name = "ebooklib", specifier = ">=0.18" },
     { name = "gewechat-client", specifier = ">=0.1.5" },
     { name = "html2text", specifier = ">=2024.2.26" },
-    { name = "langbot-plugin", specifier = "==0.5.7" },
+    { name = "langbot-plugin", specifier = "==0.5.8" },
     { name = "langchain", specifier = ">=1.3.9" },
     { name = "langchain-core", specifier = ">=1.3.3" },
     { name = "langchain-text-splitters", specifier = ">=1.1.2" },
@@ -2196,7 +2196,7 @@ dev = [
 
 [[package]]
 name = "langbot-plugin"
-version = "0.5.7"
+version = "0.5.8"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "aiofiles" },
@@ -2217,9 +2217,9 @@ dependencies = [
     { name = "watchdog" },
     { name = "websockets" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/d2/7d/b024770f1f52c9dc71ddcab79fc07dfb6147ce8e645f0fed170d758e49cb/langbot_plugin-0.5.7.tar.gz", hash = "sha256:faecd566b7ff57dc5f3a5b1be01e2165d25924031c0a65a829c83b51c65255ee", size = 480635, upload-time = "2026-09-04T13:39:22.505Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/d0/ab/8d8bd6b8355c5b30b4aab2b5322fd28d8f36158f36d6b4ee33f4df4bc861/langbot_plugin-0.5.8.tar.gz", hash = "sha256:46fbdf948f4a2d110607738ab35633c9ab22a30784edce3a4e684cd19bab84ff", size = 487972, upload-time = "2026-09-11T09:27:58.304Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/cd/25/416745039cacace6a0ca3f719a2eff41dc74cdb30ef7ffaec1de0142bd2e/langbot_plugin-0.5.7-py3-none-any.whl", hash = "sha256:b1a20bcb6a2d482019eafbfe0ac628c106b8e915c7afe89df057b4d8e2015f05", size = 310463, upload-time = "2026-09-04T13:39:21.18Z" },
+    { url = "https://files.pythonhosted.org/packages/c2/13/4939205e2f7922ec09113e390e35f9355ce6d93e1b380a4b3c49441130f5/langbot_plugin-0.5.8-py3-none-any.whl", hash = "sha256:4fbbcfa55f1dcb9af8392b48de8b7877ea79c880dfd268d651404702614d182e", size = 311552, upload-time = "2026-09-11T09:27:57.082Z" },
 ]
 
 [[package]]