From f7cfb23480420b8e6169a4120466febe07693a8d Mon Sep 17 00:00:00 2001 From: Hyu Date: Tue, 15 Sep 2026 20:40:03 +0800 Subject: [PATCH] feat: collect beta-only adapter acceptance evidence --- docs/beta-adapter-diagnostics.md | 30 +++ .../platform/adapters/aiocqhttp/adapter.py | 1 + .../pkg/platform/adapters/dingtalk/adapter.py | 1 + .../pkg/platform/adapters/discord/adapter.py | 1 + .../pkg/platform/adapters/kook/adapter.py | 1 + .../pkg/platform/adapters/lark/adapter.py | 1 + .../adapters/officialaccount/adapter.py | 1 + .../platform/adapters/qqofficial/adapter.py | 1 + .../pkg/platform/adapters/slack/adapter.py | 1 + .../pkg/platform/adapters/telegram/adapter.py | 1 + .../adapters/telegram/event_converter.py | 8 +- .../pkg/platform/adapters/wecom/adapter.py | 1 + .../pkg/platform/adapters/wecombot/adapter.py | 1 + .../pkg/platform/adapters/wecomcs/adapter.py | 1 + src/langbot/pkg/platform/botmgr.py | 3 +- .../pkg/telemetry/adapter_diagnostics.py | 70 ++++++ .../pkg/telemetry/diagnostic_catalog.py | 26 ++- .../pkg/telemetry/diagnostic_privacy.py | 6 +- src/langbot/pkg/telemetry/diagnostics.py | 74 +++++++ .../telemetry/test_adapter_acceptance.py | 208 ++++++++++++++++++ 20 files changed, 426 insertions(+), 11 deletions(-) create mode 100644 docs/beta-adapter-diagnostics.md create mode 100644 src/langbot/pkg/telemetry/adapter_diagnostics.py create mode 100644 tests/unit_tests/telemetry/test_adapter_acceptance.py diff --git a/docs/beta-adapter-diagnostics.md b/docs/beta-adapter-diagnostics.md new file mode 100644 index 000000000..2bf6c689e --- /dev/null +++ b/docs/beta-adapter-diagnostics.md @@ -0,0 +1,30 @@ +# Beta 适配器诊断开发说明 + +此诊断用于 4.11 适配器事件与 API 的 Beta 验收。仅受支持的 Beta 版本启用；稳定版、Alpha、RC、开发／本地标记版本以及关闭遥测或 Beta 诊断的实例不采集、不上传。 + +## 采集位置 + +- `pkg/platform/botmgr.py`：完成适配器事件监听器注册后产生能力快照。`listener_registered` 表示注册完成，不代表网络连接可用。 +- 各适配器的 `_dispatch_eba_event`：调用 `diagnostics.adapter_event_received(self, event)`，记录一次事件已收到。必须位于调用监听器之前，避免路由、插件或模型失败改变接收事实。 +- `pkg/telemetry/adapter_diagnostics.py`：识别适配器 API 最外层调用、平台专用 API 名、聊天与媒体类型。通过内部上下文标记抑制嵌套 API 重复计数。 +- 转换失败无法可靠确定事件种类时，保留未知事件失败；转换成功本身不重复计算事件接收。 + +新增适配器或原生事件入口时，应检查事件是否经过统一分发入口。只在转换器上添加埋点不足以覆盖直接构造事件及交互回调。 + +## 数据约束 + +`adapter_evidence` 标记可用于验收的边界记录；`listener_registered` 标记注册事实；`chat_type`、`content_type` 仅允许有限类别。消息正文、API 参数、用户／群 ID、异常正文、媒体地址不进入这些字段。内部嵌套调用标记不得序列化。 + +平台专用动作优先匹配清单，名称为 `platform_api.<动作名>`；标准交互 API 保持其标准名称。未知动作不能借通用 `call_platform_api` 包装方法被计为已验收功能。清单读取显式使用 UTF-8，兼容 Windows 默认编码环境。 + +## 联动 Space + +适配器清单或诊断操作变更后，同步生成 Space 的 `internal/diagnostics/catalog_v1.json`，并运行其目录一致性测试。Space 从清单生成完整矩阵，旧诊断记录不作为新增验收证据。部署时应先更新 Space 接收端与数据库迁移，再更新 Core Beta。 + +Core 回归命令： + +```sh +uv run pytest tests/unit_tests/telemetry -q +``` + +重点测试在 `test_adapter_acceptance.py`：版本与开关、隐私、嵌套计数、专用／交互 API、分发入口覆盖、监听器失败。Space 仓库的 `docs/beta-adapter-acceptance.md` 说明验收界面、统计口径及联调方式。 diff --git a/src/langbot/pkg/platform/adapters/aiocqhttp/adapter.py b/src/langbot/pkg/platform/adapters/aiocqhttp/adapter.py index 952818dac..ae51f6c9b 100644 --- a/src/langbot/pkg/platform/adapters/aiocqhttp/adapter.py +++ b/src/langbot/pkg/platform/adapters/aiocqhttp/adapter.py @@ -174,6 +174,7 @@ class AiocqhttpAdapter(AiocqhttpAPIMixin, abstract_platform_adapter.AbstractPlat await self._dispatch_eba_event(eba_event) async def _dispatch_eba_event(self, event: platform_events.EBAEvent): + diagnostics.adapter_event_received(self, event) for event_type in (type(event), platform_events.EBAEvent, platform_events.Event): callback = self.listeners.get(event_type) if callback: diff --git a/src/langbot/pkg/platform/adapters/dingtalk/adapter.py b/src/langbot/pkg/platform/adapters/dingtalk/adapter.py index 632802f62..486421470 100644 --- a/src/langbot/pkg/platform/adapters/dingtalk/adapter.py +++ b/src/langbot/pkg/platform/adapters/dingtalk/adapter.py @@ -293,6 +293,7 @@ class DingTalkAdapter(DingTalkAPIMixin, abstract_platform_adapter.AbstractPlatfo await self.logger.error(f'Error in dingtalk native event: {traceback.format_exc()}') async def _dispatch_eba_event(self, event: platform_events.EBAEvent): + diagnostics.adapter_event_received(self, event) for event_type in (type(event), platform_events.EBAEvent, platform_events.Event): callback = self.listeners.get(event_type) if callback: diff --git a/src/langbot/pkg/platform/adapters/discord/adapter.py b/src/langbot/pkg/platform/adapters/discord/adapter.py index 6691defa4..73da2f1ab 100644 --- a/src/langbot/pkg/platform/adapters/discord/adapter.py +++ b/src/langbot/pkg/platform/adapters/discord/adapter.py @@ -390,6 +390,7 @@ class DiscordAdapter(DiscordAPIMixin, abstract_platform_adapter.AbstractPlatform await self.logger.error(f'Error in discord {kind}: {traceback.format_exc()}') async def _dispatch_eba_event(self, event: platform_events.EBAEvent): + diagnostics.adapter_event_received(self, event) for event_type in (type(event), platform_events.EBAEvent, platform_events.Event): callback = self.listeners.get(event_type) if callback: diff --git a/src/langbot/pkg/platform/adapters/kook/adapter.py b/src/langbot/pkg/platform/adapters/kook/adapter.py index 61258b99d..198d1d186 100644 --- a/src/langbot/pkg/platform/adapters/kook/adapter.py +++ b/src/langbot/pkg/platform/adapters/kook/adapter.py @@ -202,6 +202,7 @@ class KookAdapter(KookAPIMixin, BasePlatformAdapter): await self.logger.error(f'Error handling KOOK event: {traceback.format_exc()}') async def _dispatch_eba_event(self, event: platform_events.EBAEvent): + diagnostics.adapter_event_received(self, event) for event_type in (type(event), platform_events.EBAEvent, platform_events.Event): callback = self.listeners.get(event_type) if callback: diff --git a/src/langbot/pkg/platform/adapters/lark/adapter.py b/src/langbot/pkg/platform/adapters/lark/adapter.py index 0ce7839c6..6e75d79a1 100644 --- a/src/langbot/pkg/platform/adapters/lark/adapter.py +++ b/src/langbot/pkg/platform/adapters/lark/adapter.py @@ -765,6 +765,7 @@ class LarkAdapter(LarkAPIMixin, abstract_platform_adapter.AbstractPlatformAdapte await self.logger.error(f'Error in lark message event: {traceback.format_exc()}') async def _dispatch_eba_event(self, event: platform_events.Event): + diagnostics.adapter_event_received(self, event) for event_type in (type(event), platform_events.EBAEvent, platform_events.Event): callback = self.listeners.get(event_type) if callback: diff --git a/src/langbot/pkg/platform/adapters/officialaccount/adapter.py b/src/langbot/pkg/platform/adapters/officialaccount/adapter.py index 6d5b49973..cd1a27077 100644 --- a/src/langbot/pkg/platform/adapters/officialaccount/adapter.py +++ b/src/langbot/pkg/platform/adapters/officialaccount/adapter.py @@ -192,6 +192,7 @@ class OfficialAccountAdapter(OfficialAccountAPIMixin, abstract_platform_adapter. await self.logger.error(f'Error in officialaccount native event: {traceback.format_exc()}') async def _dispatch_eba_event(self, event: platform_events.EBAEvent): + diagnostics.adapter_event_received(self, event) for event_type in (type(event), platform_events.EBAEvent, platform_events.Event): callback = self.listeners.get(event_type) if callback: diff --git a/src/langbot/pkg/platform/adapters/qqofficial/adapter.py b/src/langbot/pkg/platform/adapters/qqofficial/adapter.py index da0c1547f..d57cca2da 100644 --- a/src/langbot/pkg/platform/adapters/qqofficial/adapter.py +++ b/src/langbot/pkg/platform/adapters/qqofficial/adapter.py @@ -388,6 +388,7 @@ class QQOfficialAdapter(QQOfficialAPIMixin, abstract_platform_adapter.AbstractPl await self.logger.error(f'Error in qqofficial native event: {traceback.format_exc()}') async def _dispatch_eba_event(self, event: platform_events.EBAEvent): + diagnostics.adapter_event_received(self, event) for event_type in (type(event), platform_events.EBAEvent, platform_events.Event): callback = self.listeners.get(event_type) if callback: diff --git a/src/langbot/pkg/platform/adapters/slack/adapter.py b/src/langbot/pkg/platform/adapters/slack/adapter.py index c6e6d5db4..fd5a5dfbf 100644 --- a/src/langbot/pkg/platform/adapters/slack/adapter.py +++ b/src/langbot/pkg/platform/adapters/slack/adapter.py @@ -183,6 +183,7 @@ class SlackAdapter(SlackAPIMixin, abstract_platform_adapter.AbstractPlatformAdap await self.logger.error(f'Error in slack native event: {traceback.format_exc()}') async def _dispatch_eba_event(self, event: platform_events.EBAEvent): + diagnostics.adapter_event_received(self, event) for event_type in (type(event), platform_events.EBAEvent, platform_events.Event): callback = self.listeners.get(event_type) if callback: diff --git a/src/langbot/pkg/platform/adapters/telegram/adapter.py b/src/langbot/pkg/platform/adapters/telegram/adapter.py index 0ea343f88..8d1d8541b 100644 --- a/src/langbot/pkg/platform/adapters/telegram/adapter.py +++ b/src/langbot/pkg/platform/adapters/telegram/adapter.py @@ -482,6 +482,7 @@ class TelegramAdapter(TelegramAPIMixin, abstract_platform_adapter.AbstractPlatfo async def _dispatch_eba_event(self, event: platform_events.EBAEvent): """Dispatch once, preferring the most specific registered listener.""" + diagnostics.adapter_event_received(self, event) for event_type in (type(event), platform_events.EBAEvent, platform_events.Event): callback = self.listeners.get(event_type) if callback: diff --git a/src/langbot/pkg/platform/adapters/telegram/event_converter.py b/src/langbot/pkg/platform/adapters/telegram/event_converter.py index af2828175..f62fb2946 100644 --- a/src/langbot/pkg/platform/adapters/telegram/event_converter.py +++ b/src/langbot/pkg/platform/adapters/telegram/event_converter.py @@ -387,7 +387,13 @@ class LegacyEventConverter(abstract_platform_adapter.AbstractEventConverter): return event.source_platform_object @staticmethod - @diagnostics.observe('event', 'platform.target2yiri', source='platform', stage='convert') + @diagnostics.observe( + 'event', + 'platform.target2yiri', + source='platform', + stage='convert', + fields=lambda _: {'attributes': {'adapter_evidence': False}}, + ) async def target2yiri(event: Update, bot: telegram.Bot, bot_account_id: str): """Convert to legacy format (FriendMessage / GroupMessage).""" import langbot_plugin.api.entities.builtin.platform.events as legacy_events diff --git a/src/langbot/pkg/platform/adapters/wecom/adapter.py b/src/langbot/pkg/platform/adapters/wecom/adapter.py index 8c54cc7c0..447980891 100644 --- a/src/langbot/pkg/platform/adapters/wecom/adapter.py +++ b/src/langbot/pkg/platform/adapters/wecom/adapter.py @@ -198,6 +198,7 @@ class WecomAdapter(WecomAPIMixin, abstract_platform_adapter.AbstractPlatformAdap await self.logger.error(f'Error in wecom native event: {traceback.format_exc()}') async def _dispatch_eba_event(self, event: platform_events.EBAEvent): + diagnostics.adapter_event_received(self, event) for event_type in (type(event), platform_events.EBAEvent, platform_events.Event): callback = self.listeners.get(event_type) if callback: diff --git a/src/langbot/pkg/platform/adapters/wecombot/adapter.py b/src/langbot/pkg/platform/adapters/wecombot/adapter.py index 416af30a2..ce23bab27 100644 --- a/src/langbot/pkg/platform/adapters/wecombot/adapter.py +++ b/src/langbot/pkg/platform/adapters/wecombot/adapter.py @@ -312,6 +312,7 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo await self.logger.error(f'Error in wecombot feedback event: {traceback.format_exc()}') async def _dispatch_eba_event(self, event: platform_events.EBAEvent): + diagnostics.adapter_event_received(self, event) for event_type in (type(event), platform_events.EBAEvent, platform_events.Event): callback = self.listeners.get(event_type) if callback: diff --git a/src/langbot/pkg/platform/adapters/wecomcs/adapter.py b/src/langbot/pkg/platform/adapters/wecomcs/adapter.py index 4fd9f40bb..26f132581 100644 --- a/src/langbot/pkg/platform/adapters/wecomcs/adapter.py +++ b/src/langbot/pkg/platform/adapters/wecomcs/adapter.py @@ -214,6 +214,7 @@ class WecomCSAdapter(WecomCSAPIMixin, abstract_platform_adapter.AbstractPlatform await self.logger.error(f'Error in wecomcs native event: {traceback.format_exc()}') async def _dispatch_eba_event(self, event: platform_events.EBAEvent): + diagnostics.adapter_event_received(self, event) for event_type in (type(event), platform_events.EBAEvent, platform_events.Event): callback = self.listeners.get(event_type) if callback: diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index 85d7bc0a3..47f2c433c 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -1549,7 +1549,6 @@ class RuntimeBot: from ..telemetry.diagnostic_catalog import snapshot_bot - snapshot_bot(self) get_supported_events = getattr(self.adapter, 'get_supported_events', None) supported_events: list[str] = [] if callable(get_supported_events): @@ -1621,6 +1620,8 @@ class RuntimeBot: platform_events.EBAEvent, tenant_scoped_listener(on_eba_event), ) + # Registration is evidence of an installed listener, not a live connection. + snapshot_bot(self, listener_registered=True) async def run(self): async def exception_wrapper(): diff --git a/src/langbot/pkg/telemetry/adapter_diagnostics.py b/src/langbot/pkg/telemetry/adapter_diagnostics.py new file mode 100644 index 000000000..a62ab4e72 --- /dev/null +++ b/src/langbot/pkg/telemetry/adapter_diagnostics.py @@ -0,0 +1,70 @@ +"""Finite adapter acceptance evidence, without message contents or target IDs.""" + +from __future__ import annotations + +from .diagnostic_catalog import catalog + + +def boundary_fields(module, kind, operation, bound, parent): + """Only the actual adapter API / EBA conversion boundary counts as a test.""" + entry = next((v for k, v in catalog().items() if k in module.split('.')), None) + if entry is None or '.platform.adapters.' not in module: + return {} + evidence = kind == 'event' and operation == 'platform.target2yiri' + resolved = operation + if kind == 'api': + evidence = True + if operation == 'call_platform_api': + action = bound.get('action') + if isinstance(action, str) and 'platform_api.' + action in entry['specific_apis']: + resolved = 'platform_api.' + action + elif action in entry['apis'] and action != 'call_platform_api': + resolved = action + else: + # Keep the generic boundary for investigation, not a false named test. + evidence = False + # A forwarding API may call another decorated method. Count the outer call. + if parent and parent.adapter_api_active: + evidence = False + if not evidence: + return {'_adapter_api_active': kind == 'api'} + return { + '_adapter_api_active': kind == 'api', + 'operation': resolved, + 'attributes': {'adapter_evidence': True, **message_scenario(bound)}, + } + + +def message_scenario(bound): + """Read typed routing/media categories only; never serialize a payload.""" + from langbot_plugin.api.entities.builtin.platform.events import MessageReceivedEvent + from langbot_plugin.api.entities.builtin.platform.message import MessageChain + + event = bound.get('event') or bound.get('message_source') + target = bound.get('target_type') or bound.get('chat_type') + message = bound.get('message') or bound.get('new_content') + if isinstance(event, MessageReceivedEvent): + target = event.chat_type + message = event.message_chain + result = {} + if target is not None: + target = getattr(target, 'value', target) + if target == 'private': + target = 'person' + result['chat_type'] = target if target in ('person', 'group') else 'unknown' + if isinstance(message, MessageChain): + types = set() + for item in message: + name = type(item).__name__ + types.add( + { + 'Plain': 'text', + 'Image': 'image', + 'Voice': 'audio', + 'Audio': 'audio', + 'Video': 'video', + 'File': 'file', + }.get(name, 'other') + ) + result['content_type'] = next(iter(types)) if len(types) == 1 else 'mixed' if types else 'unknown' + return result diff --git a/src/langbot/pkg/telemetry/diagnostic_catalog.py b/src/langbot/pkg/telemetry/diagnostic_catalog.py index 3130af636..4eb92b045 100644 --- a/src/langbot/pkg/telemetry/diagnostic_catalog.py +++ b/src/langbot/pkg/telemetry/diagnostic_catalog.py @@ -15,7 +15,7 @@ def catalog(): result = {} base = Path(__file__).resolve().parents[1] / 'platform' for manifest in sorted((base / 'adapters').glob('*/manifest.yaml')): - data = yaml.safe_load(manifest.read_text()) + data = yaml.safe_load(manifest.read_text(encoding='utf-8')) spec = data.get('spec', {}) name = data['metadata']['name'] privacy.code_value('adapter', name) @@ -26,9 +26,16 @@ def catalog(): privacy.code_value('platform_event_type', event) for operation in apis: privacy.code_value('operation', operation) + specific_apis = [] for api in spec.get('platform_specific_apis', []): privacy.code_value('operation', api['action']) - result[manifest.parent.name] = {'adapter': name, 'events': events, 'apis': apis} + specific_apis.append(privacy.code_value('operation', 'platform_api.' + api['action'])) + result[manifest.parent.name] = { + 'adapter': name, + 'events': events, + 'apis': apis, + 'specific_apis': specific_apis, + } return result @@ -38,7 +45,7 @@ def adapter_fields(adapter): return {'adapter': entry['adapter']} if entry else {} -def _snapshot_bot(bot): +def _snapshot_bot(bot, *, listener_registered=False): from . import diagnostics manager = getattr(bot.ap, 'diagnostics', None) @@ -51,13 +58,15 @@ def _snapshot_bot(bot): entry = next(e for e in catalog().values() if e['adapter'] == fields['adapter']) for capability_type, method, declared in ( ('event', 'get_supported_events', entry['events']), - ('api', 'get_supported_apis', entry['apis']), + ('api', 'get_supported_apis', entry['apis'] + entry['specific_apis']), ): try: supported = set(getattr(bot.adapter, method)() or []) except Exception: supported = set() for name in declared: + if name == 'call_platform_api': + continue manager.emit( 'capability', name if capability_type == 'api' else 'platform.receive', @@ -69,15 +78,16 @@ def _snapshot_bot(bot): attributes={ 'capability_type': capability_type, 'capability_name': name, - 'supported': name in supported, + 'supported': name in supported + or (name in entry['specific_apis'] and 'call_platform_api' in supported), 'configured': True, - 'available': True, + **({'listener_registered': listener_registered} if capability_type == 'event' else {}), }, ) -def snapshot_bot(bot): +def snapshot_bot(bot, *, listener_registered=False): try: - _snapshot_bot(bot) + _snapshot_bot(bot, listener_registered=listener_registered) except Exception: pass diff --git a/src/langbot/pkg/telemetry/diagnostic_privacy.py b/src/langbot/pkg/telemetry/diagnostic_privacy.py index 3b8b4a8ba..14dc21609 100644 --- a/src/langbot/pkg/telemetry/diagnostic_privacy.py +++ b/src/langbot/pkg/telemetry/diagnostic_privacy.py @@ -32,8 +32,12 @@ VOCABULARY: dict[str, set[str]] = { 'arch': {'x86_64', 'aarch64', 'arm64', 'amd64'}, 'database': {'sqlite', 'postgresql'}, 'edition': {'community', 'cloud', 'enterprise'}, + 'chat_type': {'person', 'group', 'unknown'}, + 'content_type': {'text', 'image', 'audio', 'video', 'file', 'mixed', 'other', 'unknown'}, } -BOOLS = frozenset('stream synthetic configured available previous_session_unclean recovered supported'.split()) +BOOLS = frozenset( + 'stream synthetic configured available previous_session_unclean recovered supported adapter_evidence listener_registered'.split() +) NUMBERS = frozenset( 'attempts successes failures cancellations timeouts partial unknown generated queued acked dropped retried failed queue_size capacity result_count input_tokens output_tokens'.split() ) diff --git a/src/langbot/pkg/telemetry/diagnostics.py b/src/langbot/pkg/telemetry/diagnostics.py index 179c6f73c..2636528ec 100644 --- a/src/langbot/pkg/telemetry/diagnostics.py +++ b/src/langbot/pkg/telemetry/diagnostics.py @@ -34,6 +34,43 @@ def annotate(**fields): span.fields.update(fields) +def adapter_event_received(owner, event): + """Record one converted event before dispatch, including native callback paths.""" + with contextlib.suppress(Exception): + manager = _manager(owner) + if manager is None: + return + from langbot_plugin.api.entities.builtin.platform.events import EBAEvent + from .adapter_diagnostics import message_scenario + + if not isinstance(event, EBAEvent) or not _context_matches(manager, owner, _owner_context(owner)): + return + fields = _context_fields(owner, {'event': event}) + fields['attributes'] = { + **fields.get('attributes', {}), + 'adapter_evidence': True, + **message_scenario({'event': event}), + } + parent = current_span() + if ( + parent + and parent.manager is manager + and not ( + fields.get('workspace_uuid') + and parent.fields.get('workspace_uuid') + and fields['workspace_uuid'] != parent.fields['workspace_uuid'] + ) + ): + fields['trace_id'] = parent.fields['trace_id'] + fields['parent_span_id'] = parent.fields['span_id'] + if parent.fields.get('source') in ('webui_debug', 'synthetic'): + fields['source'] = parent.fields['source'] + fields['attributes']['synthetic'] = True + fields.setdefault('source', 'platform') + privacy.code_value('operation', 'platform.adapter_event') + manager.emit('event', 'platform.adapter_event', 'succeeded', stage='accepted', **fields) + + def _owner_app(owner): """An explicit owner (even absent/disabled) is an inheritance barrier.""" if owner is None: @@ -148,6 +185,7 @@ class Span: self.kind = kind self.operation = operation self.fields = dict(fields) + self.adapter_api_active = bool(self.fields.pop('_adapter_api_active', False)) if ( parent and parent.fields.get('workspace_uuid') @@ -167,6 +205,7 @@ class Span: ) self.fields['span_id'] = str(uuid4()) if parent and parent.manager is manager: + self.adapter_api_active = self.adapter_api_active or parent.adapter_api_active self.fields['parent_span_id'] = parent.fields['span_id'] for key in ('workspace_uuid', 'adapter', 'processor_type', 'platform_event_type', 'run_id'): if not self.fields.get(key) and parent.fields.get(key): @@ -213,8 +252,25 @@ def result_outcome(value): if span is not None and span.fields.get('stage') == 'convert': if isinstance(value, EBAEvent): annotate(platform_event_type=value.type) + if span.fields.get('attributes', {}).get('adapter_evidence'): + from .adapter_diagnostics import message_scenario + + annotate(attributes={**span.fields['attributes'], **message_scenario({'event': value})}) + # Successful conversion is counted once at dispatch, which also + # covers adapters constructing EBA events in native callbacks. + span.fields['attributes']['adapter_evidence'] = False elif value is None: set_outcome('skipped', reason_code='not_matched') + if not isinstance(value, EBAEvent) and span.fields.get('attributes', {}).get('adapter_evidence'): + span.fields['attributes']['adapter_evidence'] = False + if span is not None and span.kind == 'api' and span.fields.get('attributes', {}).get('adapter_evidence'): + # Common adapter response contracts expose status without inspecting content. + if isinstance(value, dict) and ( + value.get('ok') is False + or value.get('status') == 'failed' + or (type(value.get('retcode')) is int and value['retcode'] != 0) + ): + set_outcome('failed', reason_code='response_error') if isinstance(value, ActionResponse): if value.code != 0: set_outcome('failed', reason_code='response_error') @@ -261,6 +317,24 @@ def observe(kind, operation, *, source='internal', stage='execute', ap=None, fie extra = fields(bound) extra['attributes'] = {**metadata.get('attributes', {}), **extra.get('attributes', {})} metadata.update(extra) + from .adapter_diagnostics import boundary_fields + + if parent and ( + parent.manager is not manager + or ( + metadata.get('workspace_uuid') + and parent.fields.get('workspace_uuid') + and metadata['workspace_uuid'] != parent.fields['workspace_uuid'] + ) + ): + parent = None + evidence = ( + boundary_fields(fn.__module__, kind, operation, bound, parent) + if metadata.get('attributes', {}).get('adapter_evidence') is not False + else {} + ) + metadata['attributes'] = {**metadata.get('attributes', {}), **evidence.pop('attributes', {})} + metadata.update(evidence) return Span(manager, kind, operation, metadata) except Exception: return None diff --git a/tests/unit_tests/telemetry/test_adapter_acceptance.py b/tests/unit_tests/telemetry/test_adapter_acceptance.py new file mode 100644 index 000000000..ceffe2288 --- /dev/null +++ b/tests/unit_tests/telemetry/test_adapter_acceptance.py @@ -0,0 +1,208 @@ +"""Adapter acceptance boundaries and strict Beta-only production.""" + +import json +from types import SimpleNamespace +from uuid import uuid4 + +import pytest +from langbot_plugin.api.entities.builtin.platform.events import MessageReceivedEvent +from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain, Image + +from langbot.pkg.telemetry import diagnostics as d +from langbot.pkg.telemetry import adapter_diagnostics as adapter +from langbot.pkg.telemetry.diagnostic_catalog import snapshot_bot, catalog + + +def make_manager(version='4.11.0b2', **config): + ap = SimpleNamespace(instance_config=SimpleNamespace(data={'space': {'url': 'https://example.invalid', **config}})) + ap.diagnostics = d.DiagnosticsManager(ap, version=version, instance_id='instance-test', capacity=2048) + return ap.diagnostics + + +def boundary(fn, operation='send_message', kind='api'): + fn.__module__ = 'langbot.pkg.platform.adapters.telegram.adapter' + return d.observe(kind, operation, source='platform', stage='convert' if kind == 'event' else 'accepted')(fn) + + +def evidence(m): + return [e for e in m.pending if e['attributes'].get('adapter_evidence') and e['outcome'] != 'started'] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'version,config', + [ + ('4.11.0', {}), + ('4.11.0a1', {}), + ('4.11.0rc1', {}), + ('4.11.0.dev1', {}), + ('4.11.0b2+local', {}), + ('invalid', {}), + ('4.11.0b2', {'disable_telemetry': True}), + ('4.11.0b2', {'disable_beta_diagnostics': True}), + ], +) +async def test_disabled_adapter_produces_nothing(version, config, monkeypatch): + m = make_manager(version, **config) + + def forbidden(*args, **kwargs): + raise AssertionError('disabled producer projected adapter metadata') + + monkeypatch.setattr(adapter, 'boundary_fields', forbidden) + + async def call(self): + return 42 + + assert await boundary(call)(SimpleNamespace(ap=m.ap)) == 42 + snapshot_bot(SimpleNamespace(ap=m.ap), listener_registered=True) + d.adapter_event_received(SimpleNamespace(ap=m.ap), MessageReceivedEvent()) + assert not m.pending + assert m.counters['generated'] == 0 + await m.flush_once() + assert m.counters['acked'] == 0 + + +@pytest.mark.asyncio +async def test_nested_api_counted_once_and_only_finite_scenario(): + m = make_manager() + owner = SimpleNamespace(ap=m.ap) + + async def leaf(self, message, target_type): + return {'ok': True, 'private_result': 'SECRET_CANARY'} + + third = boundary(leaf) + + async def middle(self, message, target_type): + return await third(self, message, target_type) + + second = boundary(middle) + + async def outer(self, message, target_type): + return await second(self, message, target_type) + + await boundary(outer)( + owner, MessageChain([Plain(text='SECRET_CANARY'), Image(url='https://secret.invalid')]), 'person' + ) + rows = evidence(m) + assert len(rows) == 1 + assert rows[0]['attributes'] == {'adapter_evidence': True, 'chat_type': 'person', 'content_type': 'mixed'} + assert len(m.pending) == 6 + assert 'SECRET_CANARY' not in json.dumps(list(m.pending)) + assert '_adapter_api_active' not in json.dumps(list(m.pending)) + + +@pytest.mark.asyncio +async def test_api_specific_interaction_unknown_and_failure(): + m = make_manager() + + async def call(self, action, params): + return {'ok': False, 'description': 'SECRET_CANARY'} + + call = boundary(call, 'call_platform_api') + actions = catalog()['telegram']['specific_apis'] + assert actions + action = actions[0].removeprefix('platform_api.') + for name in (action, 'interaction.request', 'SECRET_CANARY'): + await call(SimpleNamespace(ap=m.ap), name, {'token': 'SECRET_CANARY'}) + rows = evidence(m) + assert [r['operation'] for r in rows] == [actions[0], 'interaction.request'] + assert all(r['outcome'] == 'failed' for r in rows) + assert 'SECRET_CANARY' not in json.dumps(list(m.pending)) + + +@pytest.mark.asyncio +async def test_conversion_count_and_private_message_scenario(): + m = make_manager() + + async def convert(self, event): + return event + + call = boundary(convert, 'platform.target2yiri', 'event') + event = await call( + SimpleNamespace(ap=m.ap), MessageReceivedEvent(message_chain=MessageChain([Plain(text='SECRET_CANARY')])) + ) + assert not evidence(m) + d.adapter_event_received(SimpleNamespace(ap=m.ap), event) + await call(SimpleNamespace(ap=m.ap), None) + await call(SimpleNamespace(ap=m.ap), {'legacy': 'SECRET_CANARY'}) + rows = evidence(m) + assert len(rows) == 1 + assert rows[0]['platform_event_type'] == 'message.received' + assert rows[0]['attributes']['chat_type'] == 'person' + assert rows[0]['attributes']['content_type'] == 'text' + assert 'SECRET_CANARY' not in json.dumps(list(m.pending)) + + +def test_registered_snapshot_includes_specific_apis_without_claiming_connection(): + m = make_manager() + entry = catalog()['telegram'] + cls = type( + 'Adapter', + (), + { + '__module__': 'langbot.pkg.platform.adapters.telegram.adapter', + 'get_supported_events': lambda self: entry['events'], + 'get_supported_apis': lambda self: entry['apis'], + }, + ) + snapshot_bot( + SimpleNamespace(ap=m.ap, adapter=cls(), execution_context=SimpleNamespace(workspace_uuid=str(uuid4()))), + listener_registered=True, + ) + rows = list(m.pending) + assert rows + assert all('available' not in r['attributes'] for r in rows) + assert all(r['attributes']['listener_registered'] for r in rows if r['attributes']['capability_type'] == 'event') + names = {r['attributes']['capability_name'] for r in rows} + assert set(entry['specific_apis']) <= names + + +@pytest.mark.asyncio +async def test_real_dispatch_success_precedes_listener_failure(): + from unittest.mock import AsyncMock + from langbot.pkg.platform.adapters.telegram.adapter import TelegramAdapter + from langbot_plugin.api.entities.builtin.platform.events import EBAEvent + + m = make_manager() + owner = SimpleNamespace(ap=m.ap, listeners={EBAEvent: AsyncMock(side_effect=ValueError('SECRET_CANARY'))}) + with pytest.raises(ValueError): + await TelegramAdapter._dispatch_eba_event(owner, MessageReceivedEvent()) + rows = evidence(m) + assert len(rows) == 1 and rows[0]['outcome'] == 'succeeded' + assert rows[0]['operation'] == 'platform.adapter_event' + + +def test_event_dispatch_does_not_borrow_another_workspace_trace(): + from langbot.pkg.api.http.context import ExecutionContext + + m = make_manager() + workspace = str(uuid4()) + owner = SimpleNamespace( + ap=m.ap, + execution_context=ExecutionContext( + instance_uuid=m.instance_id, workspace_uuid=workspace, placement_generation=1 + ), + ) + with d.Span(m, 'event', 'platform.receive', {'workspace_uuid': str(uuid4()), 'source': 'webui_debug'}).activate(): + d.adapter_event_received(owner, MessageReceivedEvent()) + row = evidence(m)[0] + assert row['workspace_uuid'] == workspace + assert not row.get('parent_span_id') + assert row['source'] == 'platform' + assert not row['attributes'].get('synthetic') + + +def test_every_adapter_records_native_and_interaction_dispatch(): + import ast + from pathlib import Path + + base = Path(__file__).resolve().parents[3] / 'src/langbot/pkg/platform/adapters' + for directory in catalog(): + tree = ast.parse((base / directory / 'adapter.py').read_text(encoding='utf-8')) + dispatch = next( + n for n in ast.walk(tree) if isinstance(n, ast.AsyncFunctionDef) and n.name == '_dispatch_eba_event' + ) + assert any( + isinstance(n, ast.Call) and ast.unparse(n.func) == 'diagnostics.adapter_event_received' + for n in ast.walk(dispatch) + ), directory