From 8c5ee20283a4e6118c8bc3a08832c089941c1538 Mon Sep 17 00:00:00 2001 From: RockChinQ Date: Sat, 31 Oct 2026 21:34:04 +0800 Subject: [PATCH] fix(api-tools): accept runtime_list cursor and float stale_after_seconds, verify 69 operation dispatches LANG-231: Fix SDK 0.7.19 input compatibility regressions by accepting optional cursor in ExecutionRuntimesInput and float | None stale_after_seconds in ExecutionRuntimeReconcileInput. Update test_all_69_operations_route_through_execution_api to drive real SDK 0.7.19 payloads through handler and assert raw execution dispatch. Add autouse fixture in plugin tests to isolate transfer root environment leaks. Co-authored-by: multica-agent --- src/langbot/pkg/api/application/execution.py | 3 +- .../plugin/test_execution_api_routing.py | 320 +++++++++++++++++- tests/unit_tests/plugin/conftest.py | 12 + 3 files changed, 328 insertions(+), 7 deletions(-) diff --git a/src/langbot/pkg/api/application/execution.py b/src/langbot/pkg/api/application/execution.py index 9f651dec4..787392633 100644 --- a/src/langbot/pkg/api/application/execution.py +++ b/src/langbot/pkg/api/application/execution.py @@ -584,6 +584,7 @@ EXECUTION_OPERATIONS = ( 'ExecutionRuntimesInput', statuses=(list[str] | None, None), labels=(Object | None, None), + cursor=(str | None, None), limit=(Positive, 50), ), PageOutput, @@ -591,7 +592,7 @@ EXECUTION_OPERATIONS = ( ), ExecutionOperation( 'runtime_reconcile', - _input('ExecutionRuntimeReconcileInput', stale_after_seconds=(Positive, 60)), + _input('ExecutionRuntimeReconcileInput', stale_after_seconds=(float | None, None)), JSONOutput, 'runner_admin', True, diff --git a/tests/integration/plugin/test_execution_api_routing.py b/tests/integration/plugin/test_execution_api_routing.py index 45abc23fe..c644695df 100644 --- a/tests/integration/plugin/test_execution_api_routing.py +++ b/tests/integration/plugin/test_execution_api_routing.py @@ -4,13 +4,32 @@ from __future__ import annotations import contextlib import contextvars +import types from types import SimpleNamespace +import typing from unittest.mock import AsyncMock, Mock import pytest +from langbot_plugin.api.entities.builtin.platform import message as platform_message +from langbot_plugin.api.entities.builtin.provider import message as provider_message +from langbot_plugin.api.entities.builtin.runner.context import RunnerContext +from langbot_plugin.api.entities.builtin.runner.resources import ( + AgentResources, + KnowledgeBaseResource, + ModelResource, + ToolResource, +) +from langbot_plugin.api.entities.builtin.runner.result import RunnerResult +from langbot_plugin.api.proxies.langbot_api import LangBotAPIProxy +from langbot_plugin.api.proxies.query_based_api import QueryBasedAPIProxy +from langbot_plugin.api.proxies.runner.admin import RunnerAdminAPIProxy +from langbot_plugin.api.proxies.runner.api import RunnerAPIProxy +from langbot_plugin.entities.io.actions.enums import PluginToRuntimeAction, RuntimeToLangBotAction from langbot_plugin.entities.io.context import InstallationBinding from langbot_plugin.entities.io.resp import ActionResponse +from langbot_plugin.runtime.io.handler import Handler +from langbot.pkg.api.application.execution import ExecutionOperation from langbot.pkg.plugin.handler import RuntimeConnectionHandler @@ -65,6 +84,205 @@ def test_host(): return SimpleNamespace(ap=ap, handler=handler, binding=binding, api=handler.execution_api, tenant=tenant) +def _is_union(t: typing.Any) -> bool: + return isinstance(t, types.UnionType) or typing.get_origin(t) in (typing.Union, types.UnionType) + + +def _dummy_for_type(annotation: typing.Any) -> typing.Any: + if _is_union(annotation): + args = typing.get_args(annotation) + if type(None) in args: + return None + return _dummy_for_type(args[0]) + origin = typing.get_origin(annotation) + if annotation in (str,): + return 'test' + if annotation in (int,): + return 1 + if annotation in (float,): + return 1.0 + if annotation in (bool,): + return True + if origin is list: + return [] + if origin is dict: + return {} + return {} + + +def _make_dummy_output(op: ExecutionOperation) -> dict[str, typing.Any]: + out_t = op.output_type + if _is_union(out_t): + out_t = typing.get_args(out_t)[0] + data: dict[str, typing.Any] = {} + if hasattr(out_t, 'model_fields'): + for fname, field in out_t.model_fields.items(): + if field.is_required(): + data[fname] = _dummy_for_type(field.annotation) + return data + + +async def _capture_sdk_payloads() -> dict[str, dict[str, typing.Any]]: + """Drive SDK 0.7.19 proxy methods against a mock handler to capture actual outgoing payloads.""" + recorded: dict[str, dict[str, typing.Any]] = {} + + class Interceptor(Handler): + def __init__(self): + super().__init__(Mock()) + + async def call_action(self, action, payload=None, timeout=None, **kwargs): + name = getattr(action, 'value', str(action)) + recorded[name] = payload or {} + if 'page' in name or 'list' in name: + return {'items': [], 'total_count': 0, 'has_more': False, 'next_cursor': None, 'prev_cursor': None} + return {'id': 'dummy', 'status': 'online', 'runtime_id': 'rt-1', 'run_id': 'r-1', 'token': 't'} + + async def call_action_generator(self, action, payload=None, timeout=None, **kwargs): + name = getattr(action, 'value', str(action)) + recorded[name] = payload or {} + if False: + yield {} + + interceptor = Interceptor() + api = LangBotAPIProxy(interceptor) + admin = RunnerAdminAPIProxy(interceptor) + query_proxy = QueryBasedAPIProxy(query_id='123', query_uuid='uuid-123', plugin_runtime_handler=interceptor) + + ctx = RunnerContext.model_construct( + run_id='run-123', + conversation_id='conv-1', + runtime=SimpleNamespace(deadline_at=None), + resources=AgentResources( + models=[ModelResource(model_id='m-1', operations=['invoke', 'stream', 'count_tokens'])], + tools=[ + ToolResource(tool_name='t-1', operations=['call', 'detail']), + ToolResource(tool_name='event_reply', operations=['call']), + ], + knowledge_bases=[KnowledgeBaseResource(kb_id='kb-1', operations=['retrieve', 'list'])], + ), + context={ + 'available_apis': { + 'prompt_get': True, + 'history_page': True, + 'history_search': True, + 'event_get': True, + 'event_page': True, + 'steering_pull': True, + 'state': True, + 'reply_stream': True, + } + }, + state={}, + ) + runner = RunnerAPIProxy(ctx, interceptor) + msg_chain = platform_message.MessageChain([platform_message.Plain(text='hi')]) + llm_msgs = [provider_message.Message(role='user', content='hi')] + + async def try_call(coro): + with contextlib.suppress(Exception): + await coro + + # LangBotAPIProxy + await try_call(api.acquire_box('test-key')) + await try_call(api.list_boxes()) + await try_call(api.get_box_status()) + await try_call(api.call_platform_api('bot-1', 'custom_action', {})) + await try_call(api.call_tool('t-1', parameters={'a': 1}, session={'id': 'sess-1'}, query_id=123)) + await try_call(api.get_tool_detail('t-1')) + await try_call(api.get_bot_info('bot-1')) + await try_call(api.get_bots()) + await try_call(api.get_config_file('cfg-1')) + await try_call(api.get_knowledge_file_stream('storage/path')) + await try_call(api.get_langbot_version()) + await try_call(api.get_llm_models()) + await try_call(api.invoke_embedding('emb-1', ['text'])) + await try_call(api.invoke_llm('m-1', llm_msgs)) + try: + async for _ in api.invoke_llm_stream('m-1', llm_msgs): + pass + except Exception: + pass + await try_call(api.invoke_parser('author', 'plugin', 'storage/path', 'text/plain', 'doc.txt')) + await try_call(api.invoke_rerank('rerank-1', 'query', ['doc'])) + await try_call(api.list_knowledge_bases()) + await try_call(api.list_parsers()) + await try_call(api.retrieve_knowledge('kb-1', 'query')) + await try_call(api.send_message('bot-1', 'person', 'target-1', msg_chain)) + await try_call(api.vector_delete('col-1')) + await try_call(api.vector_list('col-1')) + await try_call(api.vector_search('col-1', [0.1, 0.2], 5)) + await try_call(api.vector_upsert('col-1', [[0.1, 0.2]], ['id1'])) + + # QueryBasedAPIProxy + await try_call(query_proxy.get_bot_uuid()) + await try_call(query_proxy.get_query_var('var1')) + await try_call(query_proxy.get_query_vars()) + await try_call(query_proxy.set_query_var('var1', 'val1')) + await try_call(query_proxy.create_new_conversation()) + await try_call(query_proxy.list_pipeline_knowledge_bases()) + await try_call(query_proxy.retrieve_knowledge('kb-1', 'query')) + await try_call(query_proxy.reply(msg_chain)) + + # RunnerAdminAPIProxy + await try_call(admin.run_append_result(RunnerResult(type='content', data={'text': 'test'}, run_id='run-1'))) + await try_call(admin.run_cancel('run-1')) + await try_call(admin.run_claim('rt-1')) + await try_call(admin.run_renew_claim('run-1', 'rt-1', 'token-1')) + await try_call(admin.run_release_claim('run-1', 'rt-1', 'token-1')) + await try_call(admin.run_events_page('run-1')) + await try_call(admin.run_finalize('run-1', 'success')) + await try_call(admin.run_get('run-1')) + await try_call(admin.run_list()) + await try_call(admin.run_stats()) + await try_call(admin.runner_list()) + await try_call(admin.runner_stats()) + await try_call(admin.runtime_heartbeat('rt-1')) + await try_call(admin.runtime_list()) + await try_call(admin.runtime_reconcile()) + await try_call(admin.runtime_register('rt-1')) + await try_call(admin.runtime_stats()) + + # RunnerAPIProxy + try: + async with runner.reply_stream() as s: + await s.update('chunk') + except Exception: + pass + await try_call(runner.count_tokens('m-1', llm_msgs)) + await try_call(runner.get_prompt()) + await try_call(runner.history_page()) + await try_call(runner.history_search('query')) + await try_call(runner.event_get('evt-1')) + await try_call(runner.event_page()) + await try_call(runner.steering_pull()) + await try_call(runner.state_get('run', 'k')) + await try_call(runner.state_set('run', 'k', 'v')) + await try_call(runner.state_delete('run', 'k')) + await try_call(runner.state_list('run')) + + # Box operations in 0.7.19 + for box_act, payload in [ + (PluginToRuntimeAction.BIND_BOX, {'box_id': 'b-1', 'run_id': 'run-123'}), + (PluginToRuntimeAction.IMPORT_BOX_ATTACHMENTS, {'attachment_ids': ['att-1'], 'run_id': 'run-123'}), + (PluginToRuntimeAction.EXPORT_BOX_FILES, {'paths': ['/p'], 'run_id': 'run-123'}), + ]: + await interceptor.call_action(box_act, payload) + + # 4 binary storage operations forwarded from Runtime to LangBot Core + for stor_act, payload in [ + ( + RuntimeToLangBotAction.SET_BINARY_STORAGE, + {'key': 'k', 'owner_type': 'plugin', 'owner': 'a/p', 'value_base64': 'dGVzdA=='}, + ), + (RuntimeToLangBotAction.GET_BINARY_STORAGE, {'key': 'k', 'owner_type': 'plugin', 'owner': 'a/p'}), + (RuntimeToLangBotAction.DELETE_BINARY_STORAGE, {'key': 'k', 'owner_type': 'plugin', 'owner': 'a/p'}), + (RuntimeToLangBotAction.GET_BINARY_STORAGE_KEYS, {'owner_type': 'plugin', 'owner': 'a/p'}), + ]: + await interceptor.call_action(stor_act, payload) + + return recorded + + def test_action_audit_and_catalog_completeness(test_host): """Audit that all business actions map to ExecutionAPI and transport controls stay internal.""" operations = test_host.api.operations() @@ -97,16 +315,106 @@ def test_all_69_operations_secured_through_facade(test_host): @pytest.mark.asyncio async def test_all_69_operations_route_through_execution_api(test_host): - """Verify that calling any of the 69 operations via handler dispatches through ExecutionAPI.""" + """Drive all 69 operations through handler with real SDK payloads and verify raw dispatch.""" + sdk_payloads = await _capture_sdk_payloads() operations = test_host.api.operations() + assert len(sdk_payloads) == 69, f'Expected 69 SDK payloads, got {len(sdk_payloads)}' for op in operations: - # Mock the raw handler to return a valid dummy response matching output schema - mock_raw = AsyncMock() if not op.streaming else Mock() - test_host.handler._raw_actions[op.action] = mock_raw + raw_called = False + dummy_data = _make_dummy_output(op) - action_fn = test_host.handler.actions[op.action] - assert getattr(action_fn, '_is_secured', False) is True + async def mock_unary(data, *args, _d=dummy_data, **kwargs): + nonlocal raw_called + raw_called = True + return ActionResponse.success(_d) + + def mock_stream(data, *args, _d=dummy_data, **kwargs): + nonlocal raw_called + raw_called = True + + async def gen(): + yield ActionResponse.success(_d) + + return gen() + + test_host.handler._raw_actions[op.action] = mock_stream if op.streaming else mock_unary + + payload = dict(sdk_payloads[op.action]) + payload['caller_plugin_identity'] = 'author-a/plugin-a' + if op.scope == 'run' and 'run_id' not in payload: + payload['run_id'] = 'run-123' + + if op.streaming: + async for _ in test_host.handler.actions[op.action](payload): + pass + else: + resp = await test_host.handler.actions[op.action](payload) + assert resp.code == 0, f'{op.action} failed input validation: {resp.message}' + + assert raw_called is True, f'{op.action} did not call raw handler implementation' + + +@pytest.mark.asyncio +async def test_sdk_0_7_19_runtime_list_cursor_compatibility(test_host): + """SDK 0.7.19 sends cursor (default None) in runtime_list calls.""" + raw_called = False + + async def raw_runtime_list(data): + nonlocal raw_called + raw_called = True + return ActionResponse.success( + {'items': [], 'next_cursor': None, 'prev_cursor': None, 'has_more': False, 'total_count': 0} + ) + + test_host.handler._raw_actions['runtime_list'] = raw_runtime_list + + resp = await test_host.handler.actions['runtime_list']( + { + 'statuses': None, + 'labels': {}, + 'cursor': None, + 'limit': 50, + 'caller_plugin_identity': 'author-a/plugin-a', + } + ) + assert resp.code == 0, f'runtime_list failed: {resp.message}' + assert raw_called is True + + +@pytest.mark.asyncio +async def test_sdk_0_7_19_runtime_reconcile_compatibility(test_host): + """SDK 0.7.19 sends stale_after_seconds as None by default or as float.""" + reconcile_called = False + + async def raw_reconcile(data): + nonlocal reconcile_called + reconcile_called = True + return ActionResponse.success({}) + + test_host.handler._raw_actions['runtime_reconcile'] = raw_reconcile + + # 1. SDK default with stale_after_seconds: None + reconcile_called = False + resp1 = await test_host.handler.actions['runtime_reconcile']( + { + 'stale_after_seconds': None, + 'caller_plugin_identity': 'author-a/plugin-a', + } + ) + assert resp1.code == 0, f'runtime_reconcile None failed: {resp1.message}' + assert reconcile_called is True + + # 2. SDK explicit float value 30.0 + reconcile_called = False + resp2 = await test_host.handler.actions['runtime_reconcile']( + { + 'stale_after_seconds': 30.0, + 'caller_plugin_identity': 'author-a/plugin-a', + } + ) + assert resp2.code == 0, f'runtime_reconcile 30.0 failed: {resp2.message}' + assert reconcile_called is True @pytest.mark.asyncio diff --git a/tests/unit_tests/plugin/conftest.py b/tests/unit_tests/plugin/conftest.py index 3285bb2a0..d10fae515 100644 --- a/tests/unit_tests/plugin/conftest.py +++ b/tests/unit_tests/plugin/conftest.py @@ -1,5 +1,6 @@ """Database fixture for plugin persistence behavior tests.""" +import os import pytest from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine @@ -9,6 +10,17 @@ from langbot.pkg.entity.persistence.bstorage import BinaryStorage from langbot.pkg.entity.persistence.plugin import PluginSetting from langbot.pkg.entity.persistence.user import User from langbot.pkg.entity.persistence.workspace import Workspace +from langbot_plugin.runtime.security import PLUGIN_FILE_STORAGE_DIR_ENV + + +@pytest.fixture(autouse=True) +def clean_file_transfer_env(): + original = os.environ.get(PLUGIN_FILE_STORAGE_DIR_ENV) + yield + if original is None: + os.environ.pop(PLUGIN_FILE_STORAGE_DIR_ENV, None) + else: + os.environ[PLUGIN_FILE_STORAGE_DIR_ENV] = original @pytest.fixture