From baddaa2b7488503e6d485b62bf7ba41b8461c3f1 Mon Sep 17 00:00:00 2001 From: BiFangKNT <1320414964@qq.com> Date: Fri, 23 Oct 2026 08:41:50 +0800 Subject: [PATCH] feat(onebot): refresh expired image URLs in history and model requests - Cache NapCat rkeys by type and refresh only when their TTL expires - Renew image URLs through get_msg for other OneBot v11 implementations - Refresh session views, Runner history, and each model request attempt - Isolate caches by connection without downloading images or rewriting history --- ARCHITECTURE.md | 1 + skills/skills/langbot-mcp-ops/SKILL.md | 8 + .../pkg/api/http/service/monitoring.py | 9 +- .../service/monitoring_execution_details.py | 3 + src/langbot/pkg/api/mcp/server.py | 2 +- .../platform/adapters/aiocqhttp/image_urls.py | 171 ++++++++++++ .../platform/adapters/aiocqhttp/transport.py | 6 + src/langbot/pkg/platform/image_urls.py | 123 +++++++++ src/langbot/pkg/plugin/agent_pull_actions.py | 6 + src/langbot/pkg/plugin/handler.py | 9 +- .../pkg/provider/modelmgr/requester.py | 15 +- .../platform/test_onebot_image_urls.py | 253 ++++++++++++++++++ .../provider/test_request_policy.py | 64 +++++ .../bot-session/BotSessionMonitor.tsx | 46 +++- 14 files changed, 709 insertions(+), 7 deletions(-) create mode 100644 src/langbot/pkg/platform/adapters/aiocqhttp/image_urls.py create mode 100644 src/langbot/pkg/platform/image_urls.py create mode 100644 tests/unit_tests/platform/test_onebot_image_urls.py diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index deb78d659..0cfee2568 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -122,6 +122,7 @@ Platform code lives under `pkg/platform/`. - Bots store exclusive Agent/Pipeline routes in `event_bindings` and independent plugin subscriptions in `plugin_processors` (`processor_uuid`, `enabled`). Subscriptions resolve event patterns from the installed Runner and fan out alongside the primary route. Configuration, state, debug and run logs belong to the reusable processor instance. - `sources/` contains adapter implementations. Each adapter subclasses `langbot_plugin.api.definition.abstract.platform.adapter.AbstractMessagePlatformAdapter` from the SDK. - Platform entities such as `MessageChain`, `Image`, `At`, `Voice`, and events come from `langbot-plugin-sdk`, not from this repo. +- `pkg/platform/image_urls.py` renews image references in authorized monitoring/history responses and before Host model attempts without downloading or rewriting history. OneBot connection-local state in `adapters/aiocqhttp/image_urls.py` identifies the implementation lazily, reuses NapCat credentials until their returned TTL expires, and caches standard `get_msg` lookups for other implementations. Credentials and source mappings never cross connections or Workspace boundaries. The platform layer should translate between external platform APIs and LangBot's shared message/event model. It should not contain LLM-provider logic or pipeline business logic. diff --git a/skills/skills/langbot-mcp-ops/SKILL.md b/skills/skills/langbot-mcp-ops/SKILL.md index b2009a7fd..2af15c2f3 100644 --- a/skills/skills/langbot-mcp-ops/SKILL.md +++ b/skills/skills/langbot-mcp-ops/SKILL.md @@ -220,6 +220,14 @@ logs and action results. These times are not internal plugin profiling data. ## Unified execution monitoring +Message record reads can renew temporary OneBot image URLs through the connected +bot without downloading images or updating persisted payloads. Explicitly identified +NapCat connections reuse rkeys until their returned TTL expires (appid 1406 uses +type 10; appid 1407 uses type 20). Other OneBot implementations use standard +`get_msg` with a short per-message cache when original message and image identities +are available. Disconnected peers, missing origins, or unsupported APIs leave the +stored URL unchanged; returned URLs are not permanent archival links. + Use `get_inflight_executions` for a bounded current-workspace snapshot of active and recently started executions, including `progress_event`. Progress is a reported stage, not an estimated completion percentage. The UI uses the shared diff --git a/src/langbot/pkg/api/http/service/monitoring.py b/src/langbot/pkg/api/http/service/monitoring.py index 77cf8044e..7ca04ecf4 100644 --- a/src/langbot/pkg/api/http/service/monitoring.py +++ b/src/langbot/pkg/api/http/service/monitoring.py @@ -2113,6 +2113,9 @@ class MonitoringService(ExecutionDetailsMixin): serialized_msg = self.ap.persistence_mgr.serialize_model(persistence_monitoring.MonitoringMessage, msg) serialized.append(serialized_msg) + from ....platform.image_urls import refresh_message_rows + + await refresh_message_rows(self.ap, workspace_uuid, serialized) return (serialized, total) async def get_llm_calls( @@ -2797,10 +2800,14 @@ class MonitoringService(ExecutionDetailsMixin): ] total_llm_calls = int(llm_stats.total_calls or 0) + from ....platform.image_urls import refresh_message_rows + + serialized_message = self.ap.persistence_mgr.serialize_model(persistence_monitoring.MonitoringMessage, message) + await refresh_message_rows(self.ap, workspace_uuid, [serialized_message]) return { 'message_id': message_id, 'found': True, - 'message': self.ap.persistence_mgr.serialize_model(persistence_monitoring.MonitoringMessage, message), + 'message': serialized_message, 'llm_calls': llm_calls, 'llm_stats': { 'total_calls': total_llm_calls, diff --git a/src/langbot/pkg/api/http/service/monitoring_execution_details.py b/src/langbot/pkg/api/http/service/monitoring_execution_details.py index f5753572b..21c5112a7 100644 --- a/src/langbot/pkg/api/http/service/monitoring_execution_details.py +++ b/src/langbot/pkg/api/http/service/monitoring_execution_details.py @@ -229,6 +229,9 @@ class ExecutionDetailsMixin: for key, value in row.items() } ) + from ....platform.image_urls import refresh_message_rows + + await refresh_message_rows(self.ap, workspace, messages) return messages, count def _execution_json_text(self, column, key): diff --git a/src/langbot/pkg/api/mcp/server.py b/src/langbot/pkg/api/mcp/server.py index 27fcfca19..967a964ca 100644 --- a/src/langbot/pkg/api/mcp/server.py +++ b/src/langbot/pkg/api/mcp/server.py @@ -188,7 +188,7 @@ class LangBotMCPServer: return _dump(ap.box_service.get_recent_errors(context)) @mcp.tool( - description='Read bounded Workspace runtime records. LLM calls include optional details (timing in ms, complete provider usage, structured error); older records have null details. Filters follow the monitoring service: bot_ids, pipeline_ids, session_ids, start_time, end_time, mode (all/real/debug), execution_statuses (owning execution status groups), knowledge_base_id, user_query, is_active as supported by the record kind. Sessions read stored conversation summaries and do not support mode or execution_statuses.' + description='Read bounded Workspace runtime records. Message image URLs may be renewed through the connected OneBot peer without downloading images or rewriting stored records; URLs remain temporary. LLM calls include optional details (timing in ms, complete provider usage, structured error); older records have null details. Filters follow the monitoring service: bot_ids, pipeline_ids, session_ids, start_time, end_time, mode (all/real/debug), execution_statuses (owning execution status groups), knowledge_base_id, user_query, is_active as supported by the record kind. Sessions read stored conversation summaries and do not support mode or execution_statuses.' ) async def get_monitoring_records( kind: typing.Literal['messages', 'llm_calls', 'tool_calls', 'embedding_calls', 'sessions', 'errors'], diff --git a/src/langbot/pkg/platform/adapters/aiocqhttp/image_urls.py b/src/langbot/pkg/platform/adapters/aiocqhttp/image_urls.py new file mode 100644 index 000000000..0a8790189 --- /dev/null +++ b/src/langbot/pkg/platform/adapters/aiocqhttp/image_urls.py @@ -0,0 +1,171 @@ +"""Connection-local OneBot image references and expiring download credentials.""" + +import asyncio +import math +import time +from collections import OrderedDict +from urllib.parse import parse_qsl, quote, unquote_plus, urlsplit, urlunsplit + +from aiocqhttp import Message + + +def image_nodes(value, message_id=None): + """Yield structured image nodes with the nearest original message identity.""" + if isinstance(value, list): + source = next((v.get('id') for v in value if isinstance(v, dict) and v.get('type') == 'Source'), None) + current_id = source if source is not None else message_id + for child in value: + if isinstance(child, dict) and child.get('type') == 'Source': + current_id = child.get('id') + yield from image_nodes(child, current_id) + elif isinstance(value, dict): + message_id = value.get('message_id', message_id) + if value.get('type') in ('Image', 'image', 'image_url'): + target = value.get('image_url') if value.get('type') == 'image_url' else value + if isinstance(target, dict) and remote_image_url(target.get('url')): + yield target, message_id, value.get('image_id') or value.get('id') + for key, child in value.items(): + if key == 'origin' and value.get('type') == 'Quote': + yield from image_nodes(child, value.get('id')) + elif key not in ('image_url',): + yield from image_nodes(child, message_id) + + +def remote_image_url(url): + """Exclude inline data, local media paths and malformed URLs from peer refresh.""" + if not isinstance(url, str): + return False + try: + parsed = urlsplit(url) + return ( + parsed.scheme in ('http', 'https') and bool(parsed.hostname) and not parsed.username and not parsed.password + ) + except ValueError: + return False + + +def nt_image_type(url): + """Recognize only QQ NT download URLs with unambiguous resource parameters.""" + try: + parsed = urlsplit(url) + pairs = parse_qsl(parsed.query, keep_blank_values=True) + params = dict(pairs) + if ( + parsed.scheme not in ('http', 'https') + or parsed.hostname != 'multimedia.nt.qq.com.cn' + or parsed.username + or parsed.password + or parsed.port not in (None, 80, 443) + or parsed.path != '/download' + or any(sum(key == name for key, _ in pairs) != 1 for name in ('appid', 'fileid', 'rkey')) + or not params.get('fileid') + or not params.get('rkey') + ): + return None + return {'1406': 10, '1407': 20}.get(params['appid']) + except ValueError: + return None + + +class OneBotImageURLs: + """Keep credentials and bounded source mappings within a single connection.""" + + def __init__(self, connection): + self.connection = connection + self.lock = asyncio.Lock() + self.app_name = None + self.keys = {} + self.references = OrderedDict() + self.messages = OrderedDict() + self.retry_after = 0 + + def remember(self, url, message_id, image_id): + if message_id in (None, '', -1, '-1') or not image_id: + return + self.references[url] = (message_id, str(image_id)) + self.references.move_to_end(url) + while len(self.references) > 2048: + self.references.popitem(last=False) + + async def refresh(self, payload): + nodes = list(image_nodes(payload)) + if not nodes or self.connection.closed: + return + for node, message_id, image_id in nodes: + self.remember(node['url'], message_id, image_id) + async with self.lock: + if self.connection.closed or time.monotonic() < self.retry_after: + return + try: + if self.app_name is None: + info = await self.connection.call_action('get_version_info') + self.app_name = info.get('app_name', '') + if self.app_name == 'NapCat.Onebot': + await self._napcat(nodes) + else: + await self._standard(nodes) + except Exception: + # Failed/unsupported APIs must not break history reads or hammer the peer. + self.retry_after = time.monotonic() + 30 + raise + + async def _napcat(self, nodes): + wanted = {kind for node, _, _ in nodes if (kind := nt_image_type(node['url'])) is not None} + if not wanted: + return + if any(self.keys.get(kind, ('', 0))[1] <= time.time() for kind in wanted): + result = await self.connection.call_action('nc_get_rkey') + keys = {} + for item in result: + kind = int(item['type']) + if kind not in (10, 20): + continue + expires = float(item['time']) + float(item['ttl']) + key = item['rkey'] + if isinstance(key, str) and key.startswith('&rkey='): + key = key[len('&rkey=') :] + if isinstance(key, str) and key and math.isfinite(expires) and expires > time.time(): + keys[kind] = (key, expires) + if not wanted.issubset(keys): + raise ValueError('NapCat returned no valid image credentials') + self.keys.update(keys) + for node, _, _ in nodes: + kind = nt_image_type(node['url']) + if kind not in wanted: + continue + parsed = urlsplit(node['url']) + parts = parsed.query.split('&') + # Preserve every other parameter byte-for-byte, including signed resource IDs. + parts = [ + part.split('=', 1)[0] + '=' + quote(self.keys[kind][0], safe='') + if unquote_plus(part.split('=', 1)[0]) == 'rkey' + else part + for part in parts + ] + node['url'] = urlunsplit(parsed._replace(query='&'.join(parts))) + + async def _standard(self, nodes): + for node, _, _ in nodes: + old_url = node['url'] + ref = self.references.get(old_url) + if ref is None: + continue + message_id, image_id = ref + cached = self.messages.get(str(message_id)) + if cached is None or cached[1] <= time.monotonic(): + raw = await self.connection.call_action('get_msg', message_id=int(message_id)) + images = {} + for segment in Message(raw.get('message', [])): + if segment.type == 'image': + data = segment.data + if data.get('file') and remote_image_url(data.get('url')): + images[str(data['file'])] = data['url'] + cached = (images, time.monotonic() + 60) + self.messages[str(message_id)] = cached + self.messages.move_to_end(str(message_id)) + while len(self.messages) > 256: + self.messages.popitem(last=False) + new_url = cached[0].get(image_id) + if new_url: + self.remember(new_url, message_id, image_id) + node['url'] = new_url diff --git a/src/langbot/pkg/platform/adapters/aiocqhttp/transport.py b/src/langbot/pkg/platform/adapters/aiocqhttp/transport.py index 36e09c3d7..b1ac06ee9 100644 --- a/src/langbot/pkg/platform/adapters/aiocqhttp/transport.py +++ b/src/langbot/pkg/platform/adapters/aiocqhttp/transport.py @@ -9,6 +9,8 @@ import uuid from aiocqhttp.exceptions import ActionFailed, ApiNotAvailable +from .image_urls import OneBotImageURLs + class OneBotConnection: def __init__(self, send, receive, close, on_event, validate, *, self_id='', timeout=30): @@ -23,6 +25,7 @@ class OneBotConnection: self.events = asyncio.Queue(maxsize=256) self.closed = False self.send_lock = asyncio.Lock() + self.image_urls = OneBotImageURLs(self) async def call_action(self, action, **params): await self.validate() @@ -96,6 +99,9 @@ class OneBotConnection: if self.closed: return self.closed = True + self.image_urls.keys.clear() + self.image_urls.references.clear() + self.image_urls.messages.clear() for future in self.pending.values(): if not future.done(): future.set_exception(ApiNotAvailable('OneBot connection closed')) diff --git a/src/langbot/pkg/platform/image_urls.py b/src/langbot/pkg/platform/image_urls.py new file mode 100644 index 000000000..fb6ef7001 --- /dev/null +++ b/src/langbot/pkg/platform/image_urls.py @@ -0,0 +1,123 @@ +"""Refresh image references at authorized read/use boundaries, without persistence.""" + +import copy +import json + +from .adapters.aiocqhttp.image_urls import image_nodes + + +async def refresh_image_payload(ap, workspace_uuid, bot_id, payload): + """Return a refreshed copy, resolving bot ownership exclusively from trusted scope.""" + if not workspace_uuid or not bot_id or not list(image_nodes(payload)): + return payload + manager = getattr(ap, 'platform_mgr', None) + if manager is None: + return payload + bot = await manager.resolve_public_bot(bot_id) + if bot is None or bot.workspace_uuid != workspace_uuid or not bot.enable or bot.bot_entity.adapter != 'onebot-v11': + return payload + connection = bot.adapter.bot.connection + if connection is None or connection.closed: + return payload + result = copy.deepcopy(payload) + try: + await connection.image_urls.refresh(result) + if bot.adapter.bot.connection is not connection or connection.closed: + return payload + return result + except Exception as exc: + # Do not log signed URLs, credentials, or arbitrary upstream response bodies. + ap.logger.warning(f'OneBot image URL refresh failed ({type(exc).__name__})') + return payload + + +async def refresh_message_rows(ap, workspace_uuid, rows): + """Refresh bounded, already-authorized monitoring response rows in place.""" + groups = {} + for row in rows: + raw = row.get('message_content') + if not isinstance(raw, str): + continue + try: + payload = json.loads(raw) + except ValueError: + continue + if list(image_nodes(payload)): + groups.setdefault(row.get('bot_id'), []).append((row, payload)) + for bot_id, group in groups.items(): + payloads = [payload for _, payload in group] + refreshed = await refresh_image_payload(ap, workspace_uuid, bot_id, payloads) + if refreshed is not payloads: + for (row, _), payload in zip(group, refreshed): + row['message_content'] = json.dumps(payload, ensure_ascii=False) + + +async def refresh_model_images(ap, context, query, messages): + """Refresh on every actual model attempt, including reused Runner message lists.""" + bot_id = (getattr(query, 'bot_uuid', None) if query is not None else None) or getattr(context, 'bot_uuid', None) + if not bot_id: + return messages + payload = [message.model_dump(mode='json') for message in messages] + if not list(image_nodes(payload)): + return messages + chain = getattr(query, 'message_chain', None) + if chain is not None and hasattr(chain, 'model_dump'): + # Keep current-message origins available when provider contents omit Source/Image IDs. + payload = {'message_chain': chain.model_dump(), 'messages': payload} + refreshed = await refresh_image_payload(ap, context.workspace_uuid, bot_id, payload) + if refreshed is payload: + return messages + if isinstance(refreshed, dict): + refreshed = refreshed['messages'] + return [ + message.model_copy(update={'content': type(message).model_validate(data).content}) + for message, data in zip(messages, refreshed) + ] + + +async def refresh_history_items(ap, workspace_uuid, bot_id, items): + """Recover original image identities from scoped event facts before returning history.""" + if not items or not workspace_uuid or not bot_id or not any(list(image_nodes(item)) for item in items): + return + import sqlalchemy as sa + from ..entity.persistence.event_log import EventLog + + event_ids = [item['event_id'] for item in items if item.get('event_id')] + result = await ap.persistence_mgr.execute_async( + sa.select(EventLog.event_id, EventLog.metadata_json, EventLog.subject_type, EventLog.subject_id).where( + EventLog.workspace_id == workspace_uuid, + EventLog.bot_id == bot_id, + EventLog.event_id.in_(event_ids), + ) + ) + origins = {} + for event_id, raw, subject_type, subject_id in result.all(): + try: + metadata = json.loads(raw) if raw else {} + except ValueError: + continue + if not isinstance(metadata, dict): + continue + event = metadata.get('input_event', {}) + if isinstance(event, dict): + message_id = event.get('message_id') + if message_id is None and subject_type == 'message' and str(subject_id).lstrip('-').isdigit(): + message_id = subject_id + origins[event_id] = {'message_id': message_id, 'message_chain': event.get('message_chain', [])} + scoped_items = [] + payloads = [] + for item in items: + if item.get('workspace_id') != workspace_uuid or item.get('bot_id') != bot_id: + continue + scoped_items.append(item) + payloads.append( + { + **origins.get(item.get('event_id'), {}), + 'content_json': item.get('content_json'), + 'attachment_refs': item.get('attachment_refs', []), + } + ) + refreshed = await refresh_image_payload(ap, workspace_uuid, bot_id, payloads) + for item, payload in zip(scoped_items, refreshed): + item['content_json'] = payload['content_json'] + item['attachment_refs'] = payload['attachment_refs'] diff --git a/src/langbot/pkg/plugin/agent_pull_actions.py b/src/langbot/pkg/plugin/agent_pull_actions.py index e1e3a1714..cbc9d201c 100644 --- a/src/langbot/pkg/plugin/agent_pull_actions.py +++ b/src/langbot/pkg/plugin/agent_pull_actions.py @@ -4,6 +4,8 @@ from __future__ import annotations from typing import Any +from ..platform.image_urls import refresh_history_items + from langbot_plugin.runtime.io import handler from langbot_plugin.entities.io.actions.enums import ( @@ -88,6 +90,8 @@ def register(h): **_run_scope_filters(session), ) + scope = _run_scope_filters(session) + await refresh_history_items(h.ap, scope['workspace_id'], scope['bot_id'], items) return handler.ActionResponse.success( data={ 'items': items, @@ -159,6 +163,8 @@ def register(h): **_run_scope_filters(session), ) + scope = _run_scope_filters(session) + await refresh_history_items(h.ap, scope['workspace_id'], scope['bot_id'], items) return handler.ActionResponse.success( data={ 'items': items, diff --git a/src/langbot/pkg/plugin/handler.py b/src/langbot/pkg/plugin/handler.py index 413ccd678..e823c9591 100644 --- a/src/langbot/pkg/plugin/handler.py +++ b/src/langbot/pkg/plugin/handler.py @@ -1371,8 +1371,9 @@ class RuntimeConnectionHandler(handler.Handler): run_id = data.get('run_id') caller_plugin_identity = data.get('caller_plugin_identity') + session = None if run_id: - _, error = await _validate_run_authorization( + session, error = await _validate_run_authorization( run_id, 'model', llm_model_uuid, @@ -1417,6 +1418,12 @@ class RuntimeConnectionHandler(handler.Handler): return handler.ActionResponse.error(message='LLM provider does not support token counting') try: + from ..platform.image_urls import refresh_model_images + + query = _resolve_action_query(data, session, self.ap, action_context) + messages_obj = await refresh_model_images( + self.ap, self._execution_context(action_context), query, messages_obj + ) tokens = await count_tokens_method( model=llm_model, messages=messages_obj, diff --git a/src/langbot/pkg/provider/modelmgr/requester.py b/src/langbot/pkg/provider/modelmgr/requester.py index 1aa75e4d9..41dd62a8c 100644 --- a/src/langbot/pkg/provider/modelmgr/requester.py +++ b/src/langbot/pkg/provider/modelmgr/requester.py @@ -11,6 +11,7 @@ from ...core import app from ...api.http.context import ExecutionContext from ...entity.persistence import model as persistence_model from ...workspace.errors import WorkspaceInvariantError +from ...platform.image_urls import refresh_model_images import langbot_plugin.api.entities.builtin.resource.tool as resource_tool from . import token from . import reasoning @@ -164,11 +165,16 @@ class RuntimeProvider: state.metrics['attempts'] += 1 state.attempt_start = time.perf_counter() try: + attempt_messages = await state.wait( + lambda: refresh_model_images(self.requester.ap, context, query, messages), + 'response', + state.policy.total_timeout, + ) result = await state.wait( lambda: self.requester.invoke_llm( query=query, model=model, - messages=messages, + messages=attempt_messages, funcs=funcs, extra_args=extra_args, remove_think=remove_think, @@ -215,11 +221,16 @@ class RuntimeProvider: state.metrics['attempts'] += 1 state.attempt_start = time.perf_counter() try: + attempt_messages = await state.wait( + lambda: refresh_model_images(self.requester.ap, context, query, messages), + 'first_event', + state.policy.total_timeout, + ) async with aclosing( self.requester.invoke_llm_stream( query=query, model=model, - messages=messages, + messages=attempt_messages, funcs=funcs, extra_args=extra_args, remove_think=remove_think, diff --git a/tests/unit_tests/platform/test_onebot_image_urls.py b/tests/unit_tests/platform/test_onebot_image_urls.py new file mode 100644 index 000000000..c11c934af --- /dev/null +++ b/tests/unit_tests/platform/test_onebot_image_urls.py @@ -0,0 +1,253 @@ +"""OneBot image refresh contracts without vendor accounts or image downloads.""" + +import asyncio +import copy +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock +from urllib.parse import parse_qs, urlsplit + +import pytest + +from langbot.pkg.platform.adapters.aiocqhttp.image_urls import OneBotImageURLs, nt_image_type +from langbot.pkg.platform.image_urls import refresh_image_payload, refresh_model_images, refresh_history_items +from langbot_plugin.api.entities.builtin.provider.message import Message + + +def url(appid='1407', key='expired', fileid='resource'): + return f'https://multimedia.nt.qq.com.cn/download?appid={appid}&fileid={fileid}&rkey={key}&spec=0' + + +def payload(*urls, message_id=42): + return [{'type': 'Source', 'id': message_id}] + [ + {'type': 'Image', 'image_id': f'{index}.jpg', 'url': value} for index, value in enumerate(urls) + ] + + +def connection(action): + peer = SimpleNamespace(closed=False, call_action=AsyncMock(side_effect=action)) + peer.image_urls = OneBotImageURLs(peer) + return peer + + +@pytest.mark.asyncio +async def test_napcat_ttl_concurrency_types_and_parameters(monkeypatch): + now = [1000] + monkeypatch.setattr('langbot.pkg.platform.adapters.aiocqhttp.image_urls.time.time', lambda: now[0]) + + async def action(name, **params): + await asyncio.sleep(0) + if name == 'get_version_info': + return {'app_name': 'NapCat.Onebot'} + assert name == 'nc_get_rkey' and not params + # Order is deliberately reversed; TTL may be serialized as a string. + return [ + {'type': 20, 'time': now[0], 'ttl': '100', 'rkey': '&rkey=group+/='}, + {'type': 10, 'time': now[0], 'ttl': 200, 'rkey': 'private'}, + ] + + peer = connection(action) + batches = [payload(url(), url('1406')) for _ in range(10)] + await asyncio.gather(*(peer.image_urls.refresh(batch) for batch in batches)) + assert peer.call_action.await_count == 2 + for batch in batches: + assert parse_qs(urlsplit(batch[1]['url']).query) == { + 'appid': ['1407'], + 'fileid': ['resource'], + 'rkey': ['group+/='], + 'spec': ['0'], + } + assert parse_qs(urlsplit(batch[2]['url']).query)['rkey'] == ['private'] + now[0] = 1100 + escaped = payload(url('1406') + '&other=a%2fb+z&empty=') + await peer.image_urls.refresh(escaped) + assert '&other=a%2fb+z&empty=' in escaped[1]['url'] + await peer.image_urls.refresh(payload(url('1406'))) + assert peer.call_action.await_count == 2 + await peer.image_urls.refresh(payload(url())) + assert peer.call_action.await_count == 3 + + +@pytest.mark.parametrize( + 'value', + [ + url('9999'), + url().replace('fileid=resource&', ''), + url() + '&appid=1407', + url().replace('multimedia.nt.qq.com.cn', 'multimedia.nt.qq.com.cn.evil.test'), + url().replace('https://', 'https://user@'), + 'data:image/png;base64,AA==', + ], +) +def test_only_supported_qq_resources_are_rewritten(value): + assert nt_image_type(value) is None + + +@pytest.mark.asyncio +async def test_other_implementation_uses_original_id_and_caches_messages(): + async def action(name, **params): + if name == 'get_version_info': + return {'app_name': 'another-onebot'} + assert name == 'get_msg' and params == {'message_id': 42} + return { + 'message': [ + {'type': 'image', 'data': {'file': '1.jpg', 'url': 'https://example.test/new-1'}}, + {'type': 'image', 'data': {'file': '0.jpg', 'url': 'https://example.test/new-0'}}, + ] + } + + peer = connection(action) + batch = payload('https://example.test/old-0', 'https://example.test/old-1') + await peer.image_urls.refresh(batch) + assert [node['url'] for node in batch[1:]] == ['https://example.test/new-0', 'https://example.test/new-1'] + # Provider contents have no source identity; the connection remembers the association. + model_input = [{'type': 'image_url', 'image_url': {'url': batch[1]['url']}}] + await peer.image_urls.refresh(model_input) + assert peer.call_action.await_count == 2 + peer.image_urls.messages.clear() + await peer.image_urls.refresh(model_input) + assert peer.call_action.await_count == 3 + + +@pytest.mark.asyncio +async def test_aggregated_messages_keep_each_original_source(): + async def action(name, **params): + if name == 'get_version_info': + return {'app_name': 'another-onebot'} + assert name == 'get_msg' + message_id = params['message_id'] + return {'message': f'[CQ:image,file=0.jpg,url=https://example.test/new-{message_id}]'} + + peer = connection(action) + batch = payload('https://example.test/old-42', message_id=42) + payload( + 'https://example.test/old-43', message_id=43 + ) + await peer.image_urls.refresh(batch) + assert batch[1]['url'] == 'https://example.test/new-42' + assert batch[3]['url'] == 'https://example.test/new-43' + assert peer.call_action.await_count == 3 + + +def application(peer, workspace='workspace'): + bot = SimpleNamespace( + workspace_uuid=workspace, + enable=True, + bot_entity=SimpleNamespace(adapter='onebot-v11'), + adapter=SimpleNamespace(bot=SimpleNamespace(connection=peer)), + ) + ap = SimpleNamespace(platform_mgr=SimpleNamespace(resolve_public_bot=AsyncMock(return_value=bot)), logger=Mock()) + return ap, bot + + +@pytest.mark.asyncio +async def test_scope_failure_backoff_and_no_persistence(): + peer = connection(lambda *_args, **_kwargs: {'app_name': 'NapCat.Onebot'}) + ap, bot = application(peer) + original = payload(url()) + saved = copy.deepcopy(original) + assert await refresh_image_payload(ap, 'foreign', 'bot', original) is original + peer.call_action.assert_not_awaited() + peer.call_action.side_effect = RuntimeError('private credentials must not be logged') + assert await refresh_image_payload(ap, 'workspace', 'bot', original) is original + assert await refresh_image_payload(ap, 'workspace', 'bot', original) == original + assert peer.call_action.await_count == 1 + assert original == saved + assert 'private credentials' not in str(ap.logger.warning.call_args) + bot.adapter.bot.connection = None + assert await refresh_image_payload(ap, 'workspace', 'bot', original) is original + + +@pytest.mark.asyncio +async def test_model_reuse_refreshes_with_current_keys_without_mutating_messages(): + async def action(name, **_params): + if name == 'get_version_info': + return {'app_name': 'NapCat.Onebot'} + return [{'type': 20, 'time': 1, 'ttl': 10**12, 'rkey': '&rkey=current'}] + + peer = connection(action) + ap, _ = application(peer) + messages = [Message(role='user', content=[{'type': 'image_url', 'image_url': {'url': url()}}])] + context = SimpleNamespace(workspace_uuid='workspace') + query = SimpleNamespace(bot_uuid='bot') + refreshed = await refresh_model_images(ap, context, query, messages) + assert 'rkey=current' in refreshed[0].content[0].image_url.url + assert 'rkey=expired' in messages[0].content[0].image_url.url + await refresh_model_images(ap, context, query, messages) + assert peer.call_action.await_count == 2 + # A new connection starts with no identity or credentials. + replacement = connection(action) + assert replacement.image_urls.app_name is None and replacement.image_urls.keys == {} + + +@pytest.mark.asyncio +async def test_disconnected_peer_cannot_return_refreshed_result(): + async def action(name, **_params): + if name == 'get_version_info': + return {'app_name': 'NapCat.Onebot'} + bot.adapter.bot.connection = None + return [{'type': 20, 'time': 1, 'ttl': 10**12, 'rkey': 'new'}] + + peer = connection(action) + ap, bot = application(peer) + original = payload(url()) + assert await refresh_image_payload(ap, 'workspace', 'bot', original) is original + + +@pytest.mark.asyncio +async def test_history_recovers_scoped_origins_without_rewriting_event_facts(): + import sqlalchemy as sa + from sqlalchemy.ext.asyncio import create_async_engine + from langbot.pkg.entity.persistence.event_log import EventLog + + engine = create_async_engine('sqlite+aiosqlite:///:memory:') + original = payload('https://example.test/old') + metadata = json.dumps({'input_event': {'message_chain': original}}) + async with engine.begin() as db: + await db.run_sync(EventLog.__table__.create) + await db.execute( + sa.insert(EventLog).values( + event_id='event', + event_type='message.received', + source='platform', + bot_id='bot', + workspace_id='workspace', + metadata_json=metadata, + ) + ) + + async def execute(statement): + async with engine.connect() as db: + return await db.execute(statement) + + async def action(name, **params): + if name == 'get_version_info': + return {'app_name': 'standard-peer'} + assert params == {'message_id': 42} + return {'message': [{'type': 'image', 'data': {'file': '0.jpg', 'url': 'https://example.test/new'}}]} + + peer = connection(action) + ap, _ = application(peer) + ap.persistence_mgr = SimpleNamespace(execute_async=execute) + item = { + 'bot_id': 'bot', + 'workspace_id': 'workspace', + 'event_id': 'event', + 'content_json': {'content': [{'type': 'image_url', 'image_url': {'url': 'https://example.test/old'}}]}, + } + try: + await refresh_history_items(ap, 'foreign', 'bot', [item]) + peer.call_action.assert_not_awaited() + await refresh_history_items(ap, 'workspace', 'bot', [item]) + assert item['content_json']['content'][0]['image_url']['url'] == 'https://example.test/new' + result = await execute(sa.select(EventLog.metadata_json)) + assert result.scalar_one() == metadata + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_inline_and_local_media_do_not_call_peer(): + peer = connection(None) + for value in ('/api/v1/files/media/image.png', 'data:image/png;base64,AA=='): + await peer.image_urls.refresh(payload(value)) + peer.call_action.assert_not_awaited() diff --git a/tests/unit_tests/provider/test_request_policy.py b/tests/unit_tests/provider/test_request_policy.py index c8b0eb448..8161a5509 100644 --- a/tests/unit_tests/provider/test_request_policy.py +++ b/tests/unit_tests/provider/test_request_policy.py @@ -20,6 +20,70 @@ from tests.unit_tests.provider.conftest import TEST_EXECUTION_CONTEXT from tests.unit_tests.provider.test_litellmchat import MockRuntimeModel +@pytest.mark.asyncio +@pytest.mark.parametrize('streaming', [False, True]) +async def test_image_credentials_are_checked_again_before_retry( + runtime_provider, runtime_llm_model, monkeypatch, streaming +): + from langbot.pkg.platform.adapters.aiocqhttp.image_urls import OneBotImageURLs + from langbot_plugin.api.entities.builtin.pipeline.query import Query + + now = [1000] + monkeypatch.setattr('langbot.pkg.platform.adapters.aiocqhttp.image_urls.time.time', lambda: now[0]) + + async def action(name, **params): + if name == 'get_version_info': + return {'app_name': 'NapCat.Onebot'} + assert name == 'nc_get_rkey' + return [{'type': 20, 'time': now[0], 'ttl': 10, 'rkey': f'key-{now[0]}'}] + + peer = SimpleNamespace(closed=False, call_action=AsyncMock(side_effect=action)) + peer.image_urls = OneBotImageURLs(peer) + bot = SimpleNamespace( + workspace_uuid=TEST_EXECUTION_CONTEXT.workspace_uuid, + enable=True, + bot_entity=SimpleNamespace(adapter='onebot-v11'), + adapter=SimpleNamespace(bot=SimpleNamespace(connection=peer)), + ) + runtime_provider.requester.ap.platform_mgr = SimpleNamespace(resolve_public_bot=AsyncMock(return_value=bot)) + query = Query.model_construct(query_id=1, variables={}, bot_uuid='bot', pipeline_uuid='p', session=None) + object.__setattr__(query, '_execution_context', TEST_EXECUTION_CONTEXT) + original_url = 'https://multimedia.nt.qq.com.cn/download?appid=1407&fileid=file&rkey=old' + messages = [Message(role='user', content=[{'type': 'image_url', 'image_url': {'url': original_url}}])] + seen = [] + + async def attempt(**kwargs): + seen.append(kwargs['messages'][0].content[0].image_url.url) + if len(seen) == 1: + now[0] = 1010 + raise RequesterError('transient', retryable=True) + return Message(role='assistant', content='ok') + + async def stream(**kwargs): + await attempt(**kwargs) + yield MessageChunk(role='assistant', content='ok') + + runtime_provider.requester.invoke_llm = attempt + runtime_provider.requester.invoke_llm_stream = stream + extra = {'request_policy': {'retries': 1, 'backoff_base': 0.001}} + if streaming: + chunks = [ + chunk + async for chunk in runtime_provider.invoke_llm_stream( + query, + runtime_llm_model, + messages, + extra_args=extra, + ) + ] + assert chunks[0].content == 'ok' + else: + assert (await runtime_provider.invoke_llm(query, runtime_llm_model, messages, extra_args=extra)).content == 'ok' + assert 'rkey=key-1000' in seen[0] and 'rkey=key-1010' in seen[1] + assert messages[0].content[0].image_url.url == original_url + assert peer.call_action.await_count == 3 + + def test_policy_precedence_zero_and_no_mutation(): model = {'num_retries': 3, 'request_policy': {'read_timeout': 42}} policy = RequestPolicy.resolve({'request_policy': {'retries': 2}}, model, {'num_retries': 0}) diff --git a/web/src/app/home/bots/components/bot-session/BotSessionMonitor.tsx b/web/src/app/home/bots/components/bot-session/BotSessionMonitor.tsx index bdce4047f..b6edb70dd 100644 --- a/web/src/app/home/bots/components/bot-session/BotSessionMonitor.tsx +++ b/web/src/app/home/bots/components/bot-session/BotSessionMonitor.tsx @@ -200,6 +200,8 @@ const BotSessionMonitor = forwardRef< const messagesContainerRef = useRef(null); const sessionRequestIdRef = useRef(0); const messageRequestIdRef = useRef(0); + const retriedImageMessagesRef = useRef(new Set()); + const imageRefreshRef = useRef | null>(null); const { admins, reload: reloadAdmins } = useBotAdmins(botId); const [adminsDialogOpen, setAdminsDialogOpen] = useState(false); const [togglingAdmin, setTogglingAdmin] = useState(null); @@ -399,6 +401,8 @@ const BotSessionMonitor = forwardRef< }, [appliedUserQuery, botId, endDate, sessionPage, startDate]); useEffect(() => { + retriedImageMessagesRef.current.clear(); + imageRefreshRef.current = null; if (selectedSessionId) { loadMessages(selectedSessionId, messagePage); } else { @@ -479,9 +483,41 @@ const BotSessionMonitor = forwardRef< return !msg.runner_name; } + const refreshFailedImage = (messageId: string) => { + if (!selectedSessionId || retriedImageMessagesRef.current.has(messageId)) + return; + retriedImageMessagesRef.current.add(messageId); + if (imageRefreshRef.current) return; + const requestId = messageRequestIdRef.current; + const request = httpClient + .getSessionMessages( + selectedSessionId, + MESSAGE_PAGE_SIZE, + messagePage * MESSAGE_PAGE_SIZE, + botId, + ) + .then((response) => { + if (requestId !== messageRequestIdRef.current) return; + setMessages( + (response.messages ?? []).sort( + (a, b) => + new Date(a.timestamp).getTime() - new Date(b.timestamp).getTime(), + ), + ); + }) + .catch(() => { + /* Keep the failed image visible until the next manual refresh. */ + }) + .finally(() => { + if (imageRefreshRef.current === request) imageRefreshRef.current = null; + }); + imageRefreshRef.current = request; + }; + const renderMessageComponent = ( component: MessageChainComponent, index: number, + messageId: string, ) => { switch (component.type) { case 'Plain': @@ -529,6 +565,8 @@ const BotSessionMonitor = forwardRef< Image refreshFailedImage(messageId)} className="max-w-full max-h-52 rounded-lg" /> @@ -564,7 +602,11 @@ const BotSessionMonitor = forwardRef< >
{quote.origin?.map((comp, idx) => - renderMessageComponent(comp as MessageChainComponent, idx), + renderMessageComponent( + comp as MessageChainComponent, + idx, + messageId, + ), )}
@@ -597,7 +639,7 @@ const BotSessionMonitor = forwardRef< return (
{chain.map((component, index) => - renderMessageComponent(component, index), + renderMessageComponent(component, index, msg.id), )}
);