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 1/7] 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), )}
); From e3c7e49214f658aa668573e1757a5d9d967bb3d9 Mon Sep 17 00:00:00 2001 From: BiFangKNT <1320414964@qq.com> Date: Fri, 23 Oct 2026 09:02:23 +0800 Subject: [PATCH 2/7] test(provider): initialize Codex requester fixture --- tests/unit_tests/provider/test_codex.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit_tests/provider/test_codex.py b/tests/unit_tests/provider/test_codex.py index cb0e9e046..062130231 100644 --- a/tests/unit_tests/provider/test_codex.py +++ b/tests/unit_tests/provider/test_codex.py @@ -48,7 +48,7 @@ def requester(monkeypatch, handler): monkeypatch.setattr( httpx, 'AsyncClient', lambda **kwargs: real_client(transport=httpx.MockTransport(handler), **kwargs) ) - obj = object.__new__(CodexRequester) + obj = CodexRequester(SimpleNamespace(), {}) obj.workspace, obj.provider = 'w', 'p' obj._replay = OrderedDict() obj.auth = SimpleNamespace(access=AsyncMock(return_value=TOKENS)) From 1be47ae925c4f4eb6467ce43b744d0db3037fbb0 Mon Sep 17 00:00:00 2001 From: BiFangKNT <1320414964@qq.com> Date: Fri, 23 Oct 2026 10:21:54 +0800 Subject: [PATCH 3/7] fix(monitoring): refresh image URLs in execution details - Refresh images in inputs, outputs, deliveries, and conversation history - Reuse OneBot refresh logic without rewriting persisted records - Cover existing image URLs across agent and pipeline execution views --- .../service/monitoring_execution_details.py | 24 +++++ .../api/service/test_monitoring_executions.py | 93 +++++++++++++++++++ 2 files changed, 117 insertions(+) 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 21c5112a7..69d23cb95 100644 --- a/src/langbot/pkg/api/http/service/monitoring_execution_details.py +++ b/src/langbot/pkg/api/http/service/monitoring_execution_details.py @@ -831,6 +831,30 @@ class ExecutionDetailsMixin: {'id': run.event_id or run.run_id, 'content': content, 'timestamp_ms': epoch(run.created_at)} ] + from ....platform.image_urls import refresh_history_items + + image_items = [] + response_items = [] + for name in ('inputs', 'outputs', 'deliveries', 'conversation'): + for item in pages.get(name, {}).get('items', []): + response_items.append(item) + image_items.append( + { + 'workspace_id': workspace, + 'bot_id': bot, + 'event_id': item.get('event_id') or (item['id'] if name == 'inputs' else None), + 'content_json': {'content': item.get('content'), 'metadata': item.get('metadata')}, + 'attachment_refs': item.get('attachments', []), + } + ) + await refresh_history_items(self.ap, workspace, bot, image_items) + for item, refreshed in zip(response_items, image_items): + item['content'] = refreshed['content_json']['content'] + if 'metadata' in item: + item['metadata'] = refreshed['content_json']['metadata'] + if 'attachments' in item: + item['attachments'] = refreshed['attachment_refs'] + if 'related' in requested: # Union of ledger executions and legacy pipeline roots. SQL paging # avoids silently dropping siblings on highly fanned-out events. diff --git a/tests/unit_tests/api/service/test_monitoring_executions.py b/tests/unit_tests/api/service/test_monitoring_executions.py index ca376baa8..2029c1e06 100644 --- a/tests/unit_tests/api/service/test_monitoring_executions.py +++ b/tests/unit_tests/api/service/test_monitoring_executions.py @@ -81,6 +81,99 @@ async def test_event_properties_survive_detail_snapshot_fallback(engine, service assert items[1]['content'] == {'text': 'later'} +@pytest.mark.asyncio +@pytest.mark.parametrize('napcat', [True, False]) +@pytest.mark.parametrize('source', ['agent', 'pipeline']) +async def test_execution_details_refresh_existing_images_without_persisting(engine, service, napcat, source): + import json + from unittest.mock import AsyncMock + import sqlalchemy as sa + from langbot.pkg.entity.persistence.event_log import EventLog + from langbot.pkg.entity.persistence.transcript import Transcript + from langbot.pkg.platform.adapters.aiocqhttp.image_urls import OneBotImageURLs + + old = 'https://multimedia.nt.qq.com.cn/download?appid=1407&fileid=existing&rkey=expired' + new = old.replace('expired', 'renewed') + chain = [{'type': 'Source', 'id': 42}, {'type': 'Image', 'image_id': 'pic.jpg', 'url': old}] + content = {'content': [{'type': 'image_url', 'image_url': {'url': old}}]} + await _seed(engine) + async with AsyncSession(engine) as session: + await session.execute( + sa.update(persistence_agent_run.AgentRun) + .where(persistence_agent_run.AgentRun.run_id == 'run-done') + .values(conversation_id='session-1') + ) + await session.execute( + sa.update(persistence_monitoring.MonitoringMessage) + .where(persistence_monitoring.MonitoringMessage.id == 'msg-user') + .values(message_content=json.dumps(chain)) + ) + await session.execute( + sa.update(persistence_monitoring.MonitoringMessage) + .where(persistence_monitoring.MonitoringMessage.id == 'msg-assistant') + .values(message_content=json.dumps(chain), parent_message_id='msg-user') + ) + session.add( + EventLog( + event_id='evt-1', + event_type='message.received', + source='platform', + workspace_id=WORKSPACE, + bot_id='bot-1', + input_json=json.dumps(content), + metadata_json=json.dumps({'input_event': {'message_chain': chain}}), + ) + ) + session.add( + Transcript( + transcript_id='image-output', + event_id='evt-1', + workspace_id=WORKSPACE, + bot_id='bot-1', + conversation_id='session-1', + role='assistant', + run_id='run-done', + seq=1, + content_json=json.dumps(content), + attachment_refs_json=json.dumps([chain[1]]), + ) + ) + await session.commit() + + async def action(name, **params): + if name == 'get_version_info': + return {'app_name': 'NapCat.Onebot' if napcat else 'standard-peer'} + if napcat: + assert name == 'nc_get_rkey' + return [{'type': 20, 'time': 1, 'ttl': 10**12, 'rkey': 'renewed'}] + assert name == 'get_msg' and params == {'message_id': 42} + return {'message': [{'type': 'image', 'data': {'file': 'pic.jpg', 'url': new}}]} + + peer = SimpleNamespace(closed=False, call_action=AsyncMock(side_effect=action)) + peer.image_urls = OneBotImageURLs(peer) + bot = SimpleNamespace( + workspace_uuid=WORKSPACE, + enable=True, + bot_entity=SimpleNamespace(adapter='onebot-v11'), + adapter=SimpleNamespace(bot=SimpleNamespace(connection=peer)), + ) + service.ap.platform_mgr = SimpleNamespace(resolve_public_bot=AsyncMock(return_value=bot)) + identity = 'run-done' if source == 'agent' else 'msg-user' + detail = await service.get_execution_detail(WORKSPACE, source, identity) + for section in ('inputs', 'conversation'): + assert new in json.dumps(detail['pages'][section]['items']) + assert old not in json.dumps(detail['pages'][section]['items']) + if source == 'agent': + assert new in json.dumps(detail['pages']['outputs']['items']) + else: + assert new in json.dumps(detail['pages']['deliveries']['items']) + await service.get_execution_detail(WORKSPACE, source, identity, section='conversation') + assert peer.call_action.await_count == 2 + async with engine.connect() as db: + assert old in (await db.execute(sa.select(EventLog.input_json))).scalar_one() + assert old in (await db.execute(sa.select(Transcript.content_json))).scalar_one() + + def test_event_content_preserves_conflicting_custom_input_fields(): from langbot.pkg.api.http.service.monitoring_execution_details import event_content From 345666849743352286fce55704aea3426c3e4ffe Mon Sep 17 00:00:00 2001 From: BiFangKNT <1320414964@qq.com> Date: Fri, 23 Oct 2026 12:49:56 +0800 Subject: [PATCH 4/7] feat(media): centralize resource URLs and persist refreshed image URLs - Deduplicate message and history resource URLs using stable catalog references - Persist renewed URLs with concurrent refresh protection - Refresh OneBot images via get_msg intervals or NapCat rkey TTL - Migrate historical payloads and cover SQLite and PostgreSQL behavior --- ARCHITECTURE.md | 3 +- skills/skills/langbot-mcp-ops/SKILL.md | 5 +- .../entity/persistence/agent_interaction.py | 7 +- .../pkg/entity/persistence/assistant.py | 3 +- .../pkg/entity/persistence/monitoring.py | 8 +- .../pkg/entity/persistence/pipeline.py | 3 +- .../pkg/entity/persistence/resource_url.py | 26 ++ .../alembic/versions/0041_resource_urls.py | 92 +++++ src/langbot/pkg/persistence/media.py | 48 ++- src/langbot/pkg/persistence/mgr.py | 9 + src/langbot/pkg/persistence/resource_urls.py | 391 ++++++++++++++++++ src/langbot/pkg/persistence/tenant_uow.py | 18 +- .../platform/adapters/aiocqhttp/image_urls.py | 51 ++- .../platform/adapters/aiocqhttp/transport.py | 1 + src/langbot/pkg/platform/image_urls.py | 11 +- .../persistence/test_monitoring_postgres.py | 39 ++ .../api/service/test_monitoring_executions.py | 12 +- .../persistence/test_resource_urls.py | 367 ++++++++++++++++ .../platform/test_onebot_image_urls.py | 51 ++- .../provider/test_request_policy.py | 18 +- 20 files changed, 1116 insertions(+), 47 deletions(-) create mode 100644 src/langbot/pkg/entity/persistence/resource_url.py create mode 100644 src/langbot/pkg/persistence/alembic/versions/0041_resource_urls.py create mode 100644 src/langbot/pkg/persistence/resource_urls.py create mode 100644 tests/unit_tests/persistence/test_resource_urls.py diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 0cfee2568..a18f99565 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -122,7 +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. +- `pkg/platform/image_urls.py` renews image references in authorized monitoring/history responses and before Host model attempts without downloading. The shared resource URL catalog persists renewed URLs behind stable references. OneBot connection-local state in `adapters/aiocqhttp/image_urls.py` identifies the implementation lazily and reuses NapCat credentials until their returned TTL expires. Other implementations use standard `get_msg` when the URL's creation or last successful check is at least 30 minutes old. URLs can be shared across Workspaces; bot ownership checks, credentials and original message identities remain scoped to their connection. 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. @@ -275,6 +275,7 @@ Persistence is centered on `pkg/persistence/mgr.py`. - SQLite is the default database; PostgreSQL is supported. - Models live under `pkg/entity/persistence/`. - `pkg/persistence/media.py::MediaReferenceText` validates bound conversation payloads on ORM/Core writes while preserving the existing TEXT schema and string API. It rejects recognized structured inline images: retained images go through `MediaCache.externalize_chain_dump()` before serialization, while metadata-only snapshots and transcripts use `redact_inline_images()`. Reads and comparisons preserve historical values for incremental cleanup. Runtime/adapter messages, arbitrary diagnostic text, and opaque strings are outside this contract. +- `pkg/persistence/resource_urls.py` interns structured resource URL fields in retained messages, history and monitoring at the typed SQL boundary, in the source transaction. `MediaReferenceJSON` preserves native JSON monitoring details. Payloads store opaque references to the global `resource_urls` catalog; typed SQL reads and comparisons expand them to current URLs, preserving API shapes and source-record authorization. Configuration URLs and URLs in free text are excluded. The URL hash has a unique index; canonical IDs and check timestamps are indexed. Successful refreshes overwrite the current URL and persist the check time even when standard `get_msg` returns the same URL. Converging URLs retain only ID redirects, with no old URL. PostgreSQL row locks and SQLite's immediate write transaction serialize concurrent refreshes; SQLite holds its write lock during the upstream call. Bounded in-memory aliases let reused Runner payloads resolve old URLs without retaining those URLs in the database. Migration `0041_resource_urls` normalizes existing payloads using their source creation timestamps before producers start; downgrade expands references before dropping the catalog. - Timezone-less SQL `DateTime` columns store UTC-naive values on both backends. Normalize aware values with `pkg/persistence/datetime_utils.py::as_naive_utc` at the bind boundary, including deadlines, leases, event times, and query/retention cutoffs; do not merely strip an offset. Existing naive rows represent UTC. Restore UTC awareness for application comparisons and epoch serialization. Runner ledger/event/transcript stores follow this existing schema contract without a data migration or changes to journal/tenant authorization. - Fresh schemas are created from current metadata, then Alembic migrations run to head. LangBot 4.x does not upgrade 3.x databases. - New schema changes use Alembic under `pkg/persistence/alembic/versions/`; there is no legacy migration chain in 4.x. diff --git a/skills/skills/langbot-mcp-ops/SKILL.md b/skills/skills/langbot-mcp-ops/SKILL.md index 2af15c2f3..dde430706 100644 --- a/skills/skills/langbot-mcp-ops/SKILL.md +++ b/skills/skills/langbot-mcp-ops/SKILL.md @@ -221,10 +221,11 @@ 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 +bot without downloading images. Persisted payloads reference a shared URL catalog; +renewal overwrites its current URL so existing references read the new address. 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 +`get_msg` at most once per URL per 30 minutes 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. diff --git a/src/langbot/pkg/entity/persistence/agent_interaction.py b/src/langbot/pkg/entity/persistence/agent_interaction.py index c698f3b80..349f0be5a 100644 --- a/src/langbot/pkg/entity/persistence/agent_interaction.py +++ b/src/langbot/pkg/entity/persistence/agent_interaction.py @@ -7,6 +7,7 @@ import datetime import sqlalchemy from .base import Base +from ...persistence.media import MediaReferenceText class AgentInteraction(Base): @@ -29,15 +30,15 @@ class AgentInteraction(Base): actor_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True) status = sqlalchemy.Column(sqlalchemy.String(50), nullable=False, index=True) - request_json = sqlalchemy.Column(sqlalchemy.Text, nullable=False) + request_json = sqlalchemy.Column(MediaReferenceText, nullable=False) delivery_target_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - delivery_result_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True) + delivery_result_json = sqlalchemy.Column(MediaReferenceText, nullable=True) replaces_interaction_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True) callback_token_hash = sqlalchemy.Column(sqlalchemy.String(64), nullable=False, unique=True, index=True) expires_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True, index=True) submitted_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True) - submission_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True) + submission_json = sqlalchemy.Column(MediaReferenceText, nullable=True) status_reason = sqlalchemy.Column(sqlalchemy.Text, nullable=True) created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, default=datetime.datetime.utcnow) diff --git a/src/langbot/pkg/entity/persistence/assistant.py b/src/langbot/pkg/entity/persistence/assistant.py index 3a6c6c502..14b7d590a 100644 --- a/src/langbot/pkg/entity/persistence/assistant.py +++ b/src/langbot/pkg/entity/persistence/assistant.py @@ -1,6 +1,7 @@ import sqlalchemy as sa from .base import Base +from ...persistence.media import MediaReferenceJSON class AssistantConversation(Base): @@ -13,7 +14,7 @@ class AssistantConversation(Base): account_uuid = sa.Column(sa.String(36), sa.ForeignKey('users.uuid', ondelete='CASCADE'), nullable=False) revision = sa.Column(sa.Integer, nullable=False, default=0) status = sa.Column(sa.String(20), nullable=False, default='ready') - messages = sa.Column(sa.JSON, nullable=False, default=list) + messages = sa.Column(MediaReferenceJSON, nullable=False, default=list) error = sa.Column(sa.Text, nullable=True) model_name = sa.Column(sa.String(255), nullable=True) model_uuid = sa.Column(sa.String(36), nullable=True) diff --git a/src/langbot/pkg/entity/persistence/monitoring.py b/src/langbot/pkg/entity/persistence/monitoring.py index 6392363da..719823adc 100644 --- a/src/langbot/pkg/entity/persistence/monitoring.py +++ b/src/langbot/pkg/entity/persistence/monitoring.py @@ -1,6 +1,6 @@ import sqlalchemy -from ...persistence.media import MediaReferenceText +from ...persistence.media import MediaReferenceText, MediaReferenceJSON from .base import Base @@ -107,7 +107,7 @@ class MonitoringLLMCall(Base): error_message = sqlalchemy.Column(sqlalchemy.Text, nullable=True) message_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True) # Associated message ID - details = sqlalchemy.Column(sqlalchemy.JSON, nullable=True) + details = sqlalchemy.Column(MediaReferenceJSON, nullable=True) __table_args__ = ( sqlalchemy.Index('ix_monitoring_llm_calls_workspace_timestamp', 'workspace_uuid', 'timestamp'), @@ -137,8 +137,8 @@ class MonitoringToolCall(Base): pipeline_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False) session_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True) message_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True) - arguments = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - result = sqlalchemy.Column(sqlalchemy.Text, nullable=True) + arguments = sqlalchemy.Column(MediaReferenceText, nullable=True) + result = sqlalchemy.Column(MediaReferenceText, nullable=True) error_message = sqlalchemy.Column(sqlalchemy.Text, nullable=True) __table_args__ = ( diff --git a/src/langbot/pkg/entity/persistence/pipeline.py b/src/langbot/pkg/entity/persistence/pipeline.py index 6451ed6b7..bd979493a 100644 --- a/src/langbot/pkg/entity/persistence/pipeline.py +++ b/src/langbot/pkg/entity/persistence/pipeline.py @@ -1,6 +1,7 @@ import sqlalchemy from .base import Base +from ...persistence.media import MediaReferenceJSON class LegacyPipeline(Base): @@ -74,7 +75,7 @@ class PipelineRunRecord(Base): ) started_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False) finished_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False) - result = sqlalchemy.Column(sqlalchemy.JSON, nullable=False) + result = sqlalchemy.Column(MediaReferenceJSON, nullable=False) knowledge_base_uuid = sqlalchemy.Column(sqlalchemy.String(255), nullable=True) __table_args__ = ( diff --git a/src/langbot/pkg/entity/persistence/resource_url.py b/src/langbot/pkg/entity/persistence/resource_url.py new file mode 100644 index 000000000..6d9a19ced --- /dev/null +++ b/src/langbot/pkg/entity/persistence/resource_url.py @@ -0,0 +1,26 @@ +"""Shared, stable identities for resource URLs embedded in conversation payloads.""" + +import sqlalchemy as sa + +from .base import Base + + +class ResourceURL(Base): + __tablename__ = 'resource_urls' + + id = sa.Column(sa.String(36), primary_key=True) + url = sa.Column(sa.Text, nullable=True) + url_hash = sa.Column(sa.String(64), nullable=True, unique=True) + canonical_id = sa.Column(sa.String(36), sa.ForeignKey('resource_urls.id'), nullable=True, index=True) + created_at = sa.Column(sa.DateTime, nullable=False) + last_checked_at = sa.Column(sa.DateTime, nullable=True) + updated_at = sa.Column(sa.DateTime, nullable=False) + + __table_args__ = ( + sa.CheckConstraint( + '(canonical_id IS NULL AND url IS NOT NULL AND url_hash IS NOT NULL) ' + 'OR (canonical_id IS NOT NULL AND url IS NULL AND url_hash IS NULL)', + name='ck_resource_url_canonical', + ), + sa.Index('ix_resource_urls_last_checked', 'last_checked_at'), + ) diff --git a/src/langbot/pkg/persistence/alembic/versions/0041_resource_urls.py b/src/langbot/pkg/persistence/alembic/versions/0041_resource_urls.py new file mode 100644 index 000000000..0030a73b8 --- /dev/null +++ b/src/langbot/pkg/persistence/alembic/versions/0041_resource_urls.py @@ -0,0 +1,92 @@ +"""Normalize retained resource URLs behind stable shared references.""" + +import datetime +import json + +from alembic import op +import sqlalchemy as sa + +revision = '0041_resource_urls' +down_revision = '0040_session_summaries' +branch_labels = None +depends_on = None + + +def upgrade(): + from langbot.pkg.entity.persistence.base import Base + from langbot.pkg.entity.persistence.resource_url import ResourceURL + from langbot.pkg.persistence.media import MediaReferenceText, MediaReferenceJSON + from langbot.pkg.persistence.resource_urls import create_postgres_expansion, intern, install + + connection = op.get_bind() + ResourceURL.__table__.create(connection, checkfirst=True) + install(connection.engine) + if connection.dialect.name == 'postgresql': + create_postgres_expansion(connection) + inspector = sa.inspect(connection) + existing = set(inspector.get_table_names()) + for table in sorted(Base.metadata.tables.values(), key=lambda table: table.name): + if table.name not in existing: + continue + retained = [ + column.name for column in table.columns if isinstance(column.type, (MediaReferenceText, MediaReferenceJSON)) + ] + json_columns = {column.name for column in table.columns if isinstance(column.type, MediaReferenceJSON)} + if not retained: + continue + # Reflection intentionally reads literal stored values without expansion. + source = sa.Table(table.name, sa.MetaData(), autoload_with=connection) + retained = [name for name in retained if name in source.c] + keys = list(source.primary_key.columns) + if len(keys) != 1: + raise RuntimeError(f'Resource URL migration requires one primary key: {table.name}') + key = keys[0] + cursor = None + stamp = source.c.get('created_at') + if stamp is None: + stamp = source.c.get('timestamp') + while True: + query = sa.select(source).order_by(key).limit(100) + if cursor is not None: + query = query.where(key > cursor) + rows = connection.execute(query).mappings().all() + if not rows: + break + for row in rows: + created = row[stamp.name] if stamp is not None else None + if created is not None and created.tzinfo is not None: + created = created.astimezone(datetime.timezone.utc).replace(tzinfo=None) + updates = {} + for name in retained: + raw = json.dumps(row[name]) if name in json_columns else row[name] + converted = intern(connection, raw, created) + updates[name] = json.loads(converted) if name in json_columns else converted + updates = {name: value for name, value in updates.items() if value != row[name]} + if updates: + connection.execute(source.update().where(key == row[key.name]).values(**updates)) + cursor = rows[-1][key.name] + + +def downgrade(): + from langbot.pkg.entity.persistence.base import Base + from langbot.pkg.persistence.media import MediaReferenceText, MediaReferenceJSON + from langbot.pkg.persistence.resource_urls import uninstall + + connection = op.get_bind() + existing = set(sa.inspect(connection).get_table_names()) + for table in sorted(Base.metadata.tables.values(), key=lambda table: table.name): + if table.name not in existing: + continue + columns = {column['name'] for column in sa.inspect(connection).get_columns(table.name)} + for column in table.columns: + if column.name in columns and isinstance(column.type, (MediaReferenceText, MediaReferenceJSON)): + name = connection.dialect.identifier_preparer.quote(table.name) + field = connection.dialect.identifier_preparer.quote(column.name) + expanded = f'langbot_expand_resource_urls(CAST({field} AS TEXT))' + if isinstance(column.type, MediaReferenceJSON) and connection.dialect.name == 'postgresql': + expanded = f'CAST({expanded} AS JSON)' + connection.exec_driver_sql(f'UPDATE {name} SET {field} = {expanded}') + op.drop_table('resource_urls') + if connection.dialect.name == 'postgresql': + connection.exec_driver_sql('DROP FUNCTION langbot_expand_resource_urls(text)') + uninstall(connection.engine) diff --git a/src/langbot/pkg/persistence/media.py b/src/langbot/pkg/persistence/media.py index 337c63834..817aee2ab 100644 --- a/src/langbot/pkg/persistence/media.py +++ b/src/langbot/pkg/persistence/media.py @@ -5,7 +5,8 @@ from __future__ import annotations import json from typing import Any -from sqlalchemy import Text +from sqlalchemy import Text, JSON, cast +from sqlalchemy.sql.elements import ColumnElement from sqlalchemy.engine import Dialect from sqlalchemy.sql.operators import OperatorType from sqlalchemy.types import TypeDecorator @@ -23,9 +24,20 @@ class MediaReferenceText(TypeDecorator[str]): impl = Text cache_ok = True + class comparator_factory(Text.Comparator): + def operate(self, op, *other, **kw): + from .resource_urls import ExpandedResourceURLs + + # ORM identity maps compare annotated copies of mapped columns. + if other and isinstance(other[0], ColumnElement) and self.expr.shares_lineage(other[0]): + return super().operate(op, *other, **kw) + # CAS and textual filters compare the same expanded value readers see. + return op(ExpandedResourceURLs(self.expr), *other, **kw) + def process_bind_param(self, value: str | None, dialect: Dialect) -> str | None: if value is None: return None + reject_internal_references(value) try: payload = json.loads(value) except json.JSONDecodeError: @@ -44,3 +56,37 @@ class MediaReferenceText(TypeDecorator[str]): def coerce_compared_value(self, op: OperatorType | None, value: Any) -> Text: # Legacy payloads must remain usable in searches and compare-and-swap updates. return self.impl + + def column_expression(self, column): + from .resource_urls import ExpandedResourceURLs + + return ExpandedResourceURLs(column) + + +class MediaReferenceJSON(TypeDecorator): + """Preserve native JSON API values while interning structured resource fields.""" + + impl = JSON + cache_ok = True + + def process_bind_param(self, value, dialect): + if value is not None: + reject_internal_references(json.dumps(value)) + return value + + def column_expression(self, column): + from .resource_urls import ExpandedResourceJSON + + return ExpandedResourceJSON(cast(column, Text)) + + +def reject_internal_references(value): + """Only the transactional normalizer may introduce catalog references.""" + from .resource_urls import REFERENCE + + if REFERENCE.search(value): + raise ValueError('Internal resource references cannot be supplied as conversation content') + + +# Register the catalog and its database hooks before engines are opened. +from . import resource_urls as _resource_urls # noqa: E402,F401 diff --git a/src/langbot/pkg/persistence/mgr.py b/src/langbot/pkg/persistence/mgr.py index 9350f5f62..462a7e66e 100644 --- a/src/langbot/pkg/persistence/mgr.py +++ b/src/langbot/pkg/persistence/mgr.py @@ -183,16 +183,25 @@ class PersistenceManager: if self.mode == PersistenceMode.CLOUD_RUNTIME: await self._validate_cloud_runtime() + from .resource_urls import install + + install(engine.sync_engine) return if self.mode == PersistenceMode.RELEASE_MIGRATION: async with self._release_migration_lock(): await self._initialize_managed_schema() await self._validate_release_schema() + from .resource_urls import install + + install(engine.sync_engine) return await self._initialize_managed_schema() await self._enable_sqlite_foreign_keys_after_migration() + from .resource_urls import install + + install(engine.sync_engine) if self.mode == PersistenceMode.OSS_COMPAT: await self.write_space_model_providers() diff --git a/src/langbot/pkg/persistence/resource_urls.py b/src/langbot/pkg/persistence/resource_urls.py new file mode 100644 index 000000000..bdbde85b3 --- /dev/null +++ b/src/langbot/pkg/persistence/resource_urls.py @@ -0,0 +1,391 @@ +"""Normalize resource URLs at the typed SQL boundary, within the source transaction. + +Only typed conversation payload writes are interned. Reads expand references in SQL so +ORM and Core callers retain their existing string/JSON contracts. The catalog +is shared; authorization continues to belong to the enclosing business record. +""" + +import datetime +import hashlib +import json +import re +import uuid +import sqlite3 +from collections import OrderedDict + +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql, sqlite +from sqlalchemy.ext.compiler import compiles +from sqlalchemy.sql.functions import FunctionElement + +from ..entity.persistence.resource_url import ResourceURL + + +PREFIX = 'langbot-resource://' +REFERENCE = re.compile(r'langbot-resource://([a-f0-9-]{36})') +FIELDS = frozenset({'url', 'image_url', 'audio_url', 'file_url', 'video_url', 'original_url'}) + + +def now(): + return datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) + + +def digest(url): + return hashlib.sha256(url.encode()).hexdigest() + + +def fields(value): + if isinstance(value, list): + for child in value: + yield from fields(child) + elif isinstance(value, dict): + for key, child in value.items(): + if key in FIELDS and isinstance(child, str) and child.startswith(('http://', 'https://', '/api/v1/files/')): + yield value, key, child + else: + yield from fields(child) + + +def intern(connection, raw, created_at=None, identities_out=None): + """Intern resource fields synchronously in the current SQLAlchemy transaction.""" + if not isinstance(raw, str): + return raw + try: + payload = json.loads(raw) + except ValueError: + return raw + nodes = list(fields(payload)) + if not nodes: + return raw + table = ResourceURL.__table__ + stamp = now() + urls = {url for _, _, url in nodes} + aliases = connection.engine._langbot_resource_aliases + known = {url: aliases[url] for url in urls if url in aliases} + valid = ( + set(connection.execute(sa.select(table.c.id).where(table.c.id.in_(set(known.values())))).scalars()) + if known + else set() + ) + identities = {url: identity for url, identity in known.items() if identity in valid} + urls -= identities.keys() + hashes = dict(sorted((digest(url), url) for url in urls)) + insert = {'sqlite': sqlite.insert, 'postgresql': postgresql.insert}[connection.dialect.name] + if hashes: + connection.execute( + insert(table) + .values( + [ + dict(id=str(uuid.uuid4()), url=url, url_hash=key, created_at=created_at or stamp, updated_at=stamp) + for key, url in hashes.items() + ] + ) + .on_conflict_do_nothing(index_elements=['url_hash']) + ) + for identity, url, key in connection.execute( + sa.select(table.c.id, table.c.url, table.c.url_hash).where(table.c.url_hash.in_(hashes)) + ): + if hashes[key] != url: + raise ValueError('Resource URL hash collision') + identities[url] = identity + if created_at is not None and identities: + # Migration order must not make an old shared URL appear newly created. + connection.execute( + table.update() + .where(table.c.id.in_(set(identities.values())), table.c.created_at > created_at) + .values(created_at=created_at) + ) + for node, key, url in nodes: + node[key] = PREFIX + identities[url] + aliases[url] = identities[url] + aliases.move_to_end(url) + while len(aliases) > 8192: + aliases.popitem(last=False) + if identities_out is not None: + identities_out.update(identities) + return json.dumps(payload, ensure_ascii=False) + + +class ExpandedResourceURLs(FunctionElement): + type = sa.Text() + inherit_cache = True + + +class ExpandedResourceJSON(FunctionElement): + type = sa.JSON() + inherit_cache = True + + +@compiles(ExpandedResourceJSON) +def compile_json_expansion(element, compiler, **kw): + expression = compile_expansion(element, compiler, **kw) + if compiler.dialect.name == 'postgresql': + return f'CAST({expression} AS JSON)' + return expression + + +@compiles(ExpandedResourceURLs) +def compile_expansion(element, compiler, **kw): + source = compiler.process(list(element.clauses)[0], **kw) + if not getattr(compiler.dialect, '_langbot_resource_urls', False): + return source + return f'langbot_expand_resource_urls({source})' + + +def install(engine): + """Enable normalization after schema initialization, before producers start.""" + engine.dialect._langbot_resource_urls = True + if not hasattr(engine, '_langbot_resource_aliases'): + engine._langbot_resource_aliases = OrderedDict() + engine.clear_compiled_cache() + if not sa.event.contains(engine, 'before_cursor_execute', normalize_parameters): + sa.event.listen(engine, 'before_cursor_execute', normalize_parameters, retval=True) + + +def uninstall(engine): + """Restore literal payload reads and writes after catalog downgrade.""" + if sa.event.contains(engine, 'before_cursor_execute', normalize_parameters): + sa.event.remove(engine, 'before_cursor_execute', normalize_parameters) + engine.dialect._langbot_resource_urls = False + engine.clear_compiled_cache() + engine._langbot_resource_aliases.clear() + + +def normalize_parameters(connection, cursor, statement, parameters, context, executemany): + """Replace only typed payload binds; never inspect arbitrary SQL or configuration.""" + from .media import MediaReferenceText, MediaReferenceJSON + + compiled = context.compiled + if compiled is None or not (context.isinsert or context.isupdate): + return statement, parameters + names = { + name + for bind, name in compiled.bind_names.items() + if isinstance(bind.type, (MediaReferenceText, MediaReferenceJSON)) + } + if not names: + return statement, parameters + + def normalize(row): + if isinstance(row, dict): + return {key: intern(connection, value) if key in names else value for key, value in row.items()} + positions = compiled.positiontup or [] + if not positions: + return row + return tuple( + intern(connection, value) if positions[index % len(positions)] in names else value + for index, value in enumerate(row) + ) + + batched = executemany and getattr(compiled, '_insertmanyvalues', None) is None + return statement, [normalize(row) for row in parameters] if batched else normalize(parameters) + + +def register_sqlite(dbapi, record): + """The UDF runs on aiosqlite's worker, including its catalog SELECTs.""" + driver = getattr(dbapi, 'driver_connection', None) + raw_connection = driver._conn if hasattr(driver, '_conn') else dbapi + if not isinstance(raw_connection, sqlite3.Connection): + return + + def expand(raw): + if not isinstance(raw, str) or PREFIX not in raw: + return raw + + resolved = {} + + def lookup(match): + identity = match.group(1) + if identity in resolved: + return resolved[identity] + visited = set() + while identity not in visited: + visited.add(identity) + row = raw_connection.execute( + 'SELECT url, canonical_id FROM resource_urls WHERE id=?', (identity,) + ).fetchone() + if row is None: + raise ValueError('Missing resource URL') + if row[1] is None: + value = json.dumps(row[0], ensure_ascii=False)[1:-1] + for reference in visited: + resolved[reference] = value + return value + identity = row[1] + raise ValueError('Cyclic resource URL reference') + + return REFERENCE.sub(lookup, raw) + + dbapi.create_function('langbot_expand_resource_urls', 1, expand) + + +POSTGRES_EXPANSION = r""" +CREATE OR REPLACE FUNCTION public.langbot_expand_resource_urls(payload text) +RETURNS text LANGUAGE plpgsql STABLE SECURITY INVOKER SET search_path = public, pg_temp AS $$ +DECLARE token text; identity text; target text; parent text; result text := payload; visited text[]; +BEGIN + IF payload IS NULL OR position('langbot-resource://' in payload) = 0 THEN RETURN payload; END IF; + FOR token IN SELECT DISTINCT m[1] FROM regexp_matches(payload, 'langbot-resource://([a-f0-9-]{36})', 'g') m LOOP + identity := token; visited := ARRAY[]::text[]; + LOOP + IF identity = ANY(visited) THEN RAISE EXCEPTION 'Cyclic resource URL reference'; END IF; + visited := array_append(visited, identity); + SELECT url, canonical_id INTO target, parent FROM resource_urls WHERE id = identity; + IF NOT FOUND THEN RAISE EXCEPTION 'Missing resource URL'; END IF; + EXIT WHEN parent IS NULL; + identity := parent; + END LOOP; + result := replace(result, 'langbot-resource://' || token, substr(to_json(target)::text, 2, length(to_json(target)::text)-2)); + END LOOP; + RETURN result; +END $$; +""" + + +def catalog_created(table, connection, **kw): + if connection.dialect.name == 'postgresql': + create_postgres_expansion(connection) + install(connection.engine) + + +def create_postgres_expansion(connection): + schema = connection.scalar(sa.select(sa.func.current_schema())) + quoted = connection.dialect.identifier_preparer.quote(schema) + connection.exec_driver_sql(POSTGRES_EXPANSION.replace('public', quoted)) + + +async def refresh_images(engine, peer, payload, *, is_current=None): + """Serialize refreshes on catalog identities and overwrite current URLs atomically.""" + from ..platform.adapters.aiocqhttp.image_urls import image_nodes + + nodes = list(image_nodes(payload)) + if not nodes: + return + + def connected(): + return not peer.closed and (is_current is None or is_current()) + + table = ResourceURL.__table__ + # Establish catalog identities for runtime-only inputs too. Existing history + # already has identities created in its original write or migration. + async with engine.begin() as db: + references = {} + await db.run_sync(lambda sync: intern(sync, json.dumps(payload), identities_out=references)) + identities = set(references.values()) + async with engine.connect() as db: + if engine.dialect.name == 'sqlite': + await db.exec_driver_sql('BEGIN IMMEDIATE') + else: + await db.begin() + try: + rows = ( + ( + await db.execute( + sa.select(table).where(table.c.id.in_(identities)).order_by(table.c.id).with_for_update() + ) + ) + .mappings() + .all() + ) + by_id = {row['id']: row for row in rows} + for identity in list(by_id): + row = by_id[identity] + while row['canonical_id'] is not None: + row = ( + (await db.execute(sa.select(table).where(table.c.id == row['canonical_id']).with_for_update())) + .mappings() + .one() + ) + by_id[identity] = row + by_url = {node['url']: by_id[references[node['url']]] for node, _, _ in nodes} + # Capture origins even when renewal is not due or identification fails. + for node, message_id, image_id in nodes: + peer.image_urls.remember(by_url[node['url']]['url'], message_id, image_id) + # Identification and upstream credential handling remain connection-local. + if not connected() or not await peer.image_urls.identify(): + return + napcat = peer.image_urls.app_name == 'NapCat.Onebot' + due = [] + stamp = now() + for node, message_id, image_id in nodes: + row = by_url.get(node['url']) + if row is None: + continue + node['url'] = row['url'] + checked = row['last_checked_at'] or row['created_at'] + if napcat or (stamp - checked).total_seconds() >= 1800: + due.append((node, message_id, image_id, row)) + if due: + # Keep Source/Image identities, including those recovered from events. + batch = [ + dict(message_id=message_id, image={'type': 'Image', 'image_id': image_id, 'url': node['url']}) + for node, message_id, image_id, _ in due + ] + old = [entry['image']['url'] for entry in batch] + completed = await peer.image_urls.refresh(batch) + if not completed: + return + if not connected(): + return + stamp = now() + updated = set() + for (node, _, _, row), entry, previous in zip(due, batch, old): + if previous not in completed: + continue + fresh = entry['image']['url'] + node['url'] = fresh + if napcat and fresh == previous: + continue + if row['id'] in updated: + continue + updated.add(row['id']) + values = dict(last_checked_at=stamp) + if fresh != previous: + fresh_hash = digest(fresh) + # Lock the destination before merging identical current URLs. + destination = ( + await db.execute( + sa.select(table.c.id, table.c.url) + .where(table.c.url_hash == fresh_hash) + .with_for_update() + ) + ).first() + if destination is None: + try: + # Another source may discover the same renewed URL + # concurrently. Let the unique index arbitrate it. + async with db.begin_nested(): + await db.execute( + table.update() + .where(table.c.id == row['id']) + .values(url=fresh, url_hash=fresh_hash, updated_at=stamp, **values) + ) + continue + except sa.exc.IntegrityError as exc: + if getattr(exc.orig, 'sqlstate', None) != '23505': + raise + destination = ( + await db.execute( + sa.select(table.c.id, table.c.url) + .where(table.c.url_hash == fresh_hash) + .with_for_update() + ) + ).one() + if destination is not None and destination.id != row['id']: + if destination.url != fresh: + raise ValueError('Resource URL hash collision') + values.update(url=None, url_hash=None, canonical_id=destination.id, updated_at=stamp) + await db.execute( + table.update().where(table.c.id == destination.id).values(last_checked_at=stamp) + ) + else: + values.update(url=fresh, url_hash=fresh_hash, updated_at=stamp) + await db.execute(table.update().where(table.c.id == row['id']).values(**values)) + await db.commit() + except BaseException: + await db.rollback() + raise + + +sa.event.listen(sa.engine.Engine, 'connect', register_sqlite) +sa.event.listen(ResourceURL.__table__, 'after_create', catalog_created) diff --git a/src/langbot/pkg/persistence/tenant_uow.py b/src/langbot/pkg/persistence/tenant_uow.py index 7de4cdba8..2b6a1093d 100644 --- a/src/langbot/pkg/persistence/tenant_uow.py +++ b/src/langbot/pkg/persistence/tenant_uow.py @@ -19,7 +19,7 @@ from sqlalchemy.dialects.postgresql.dml import OnConflictDoUpdate as PostgreSQLO from sqlalchemy.dialects.sqlite.dml import OnConflictDoNothing as SQLiteOnConflictDoNothing from sqlalchemy.dialects.sqlite.dml import OnConflictDoUpdate as SQLiteOnConflictDoUpdate -from .media import MediaReferenceText +from .media import MediaReferenceText, MediaReferenceJSON TENANT_SETTING = 'langbot.workspace_uuid' @@ -295,7 +295,9 @@ def _validate_scoped_sql_type( if type(sql_type) in {Vector, HALFVEC}: return - if type(sql_type) is not MediaReferenceText and not type(sql_type).__module__.startswith('sqlalchemy.'): + if type(sql_type) not in {MediaReferenceText, MediaReferenceJSON} and not type(sql_type).__module__.startswith( + 'sqlalchemy.' + ): raise ScopedSessionTransactionError('TenantUnitOfWork does not allow custom SQL types in public statements') for identifier_attribute in ('name', 'schema', 'collation'): @@ -501,6 +503,14 @@ def _validate_scoped_statement_call(args: tuple[typing.Any, ...], kwargs: dict[s ) if isinstance(element, sqlalchemy.sql.functions.FunctionElement): + from .resource_urls import ExpandedResourceURLs + + resource_expansion_allowed = ( + type(element) is ExpandedResourceURLs + and len(list(element.clauses)) == 1 + and isinstance(list(element.clauses)[0], sqlalchemy.Column) + and type(list(element.clauses)[0].type) is MediaReferenceText + ) function_name = str(getattr(element, 'name', '')).casefold() package_names = tuple(getattr(element, 'packagenames', ())) generic_function_allowed = ( @@ -508,7 +518,9 @@ def _validate_scoped_statement_call(args: tuple[typing.Any, ...], kwargs: dict[s and function_name in _ALLOWED_SCOPED_GENERIC_FUNCTIONS ) builtin_function_allowed = _ALLOWED_SCOPED_BUILTIN_FUNCTION_TYPES.get(function_name) is type(element) - if package_names or not (generic_function_allowed or builtin_function_allowed): + if package_names or not ( + generic_function_allowed or builtin_function_allowed or resource_expansion_allowed + ): raise ScopedSessionTransactionError( f'TenantUnitOfWork does not allow SQL function {function_name or ""!r}' ) diff --git a/src/langbot/pkg/platform/adapters/aiocqhttp/image_urls.py b/src/langbot/pkg/platform/adapters/aiocqhttp/image_urls.py index 0a8790189..701dafec6 100644 --- a/src/langbot/pkg/platform/adapters/aiocqhttp/image_urls.py +++ b/src/langbot/pkg/platform/adapters/aiocqhttp/image_urls.py @@ -77,6 +77,7 @@ class OneBotImageURLs: self.keys = {} self.references = OrderedDict() self.messages = OrderedDict() + self.urls = OrderedDict() self.retry_after = 0 def remember(self, url, message_id, image_id): @@ -87,23 +88,37 @@ class OneBotImageURLs: while len(self.references) > 2048: self.references.popitem(last=False) + async def identify(self): + """Identify lazily with the same backoff as resource renewal.""" + async with self.lock: + if self.connection.closed or time.monotonic() < self.retry_after: + return False + try: + await self._identify() + return True + except Exception: + self.retry_after = time.monotonic() + 30 + raise + + async def _identify(self): + if self.app_name is None: + info = await self.connection.call_action('get_version_info') + self.app_name = info.get('app_name', '') + async def refresh(self, payload): nodes = list(image_nodes(payload)) if not nodes or self.connection.closed: - return + return False 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 + return False try: - if self.app_name is None: - info = await self.connection.call_action('get_version_info') - self.app_name = info.get('app_name', '') + await self._identify() if self.app_name == 'NapCat.Onebot': - await self._napcat(nodes) - else: - await self._standard(nodes) + return await self._napcat(nodes) + return await self._standard(nodes) except Exception: # Failed/unsupported APIs must not break history reads or hammer the peer. self.retry_after = time.monotonic() + 30 @@ -112,7 +127,7 @@ class OneBotImageURLs: 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 + return set() if any(self.keys.get(kind, ('', 0))[1] <= time.time() for kind in wanted): result = await self.connection.call_action('nc_get_rkey') keys = {} @@ -129,10 +144,12 @@ class OneBotImageURLs: if not wanted.issubset(keys): raise ValueError('NapCat returned no valid image credentials') self.keys.update(keys) + completed = set() for node, _, _ in nodes: kind = nt_image_type(node['url']) if kind not in wanted: continue + completed.add(node['url']) parsed = urlsplit(node['url']) parts = parsed.query.split('&') # Preserve every other parameter byte-for-byte, including signed resource IDs. @@ -143,10 +160,17 @@ class OneBotImageURLs: for part in parts ] node['url'] = urlunsplit(parsed._replace(query='&'.join(parts))) + return completed async def _standard(self, nodes): + completed = set() for node, _, _ in nodes: old_url = node['url'] + refreshed = self.urls.get(old_url) + if refreshed is not None and refreshed[1] > time.monotonic(): + node['url'] = refreshed[0] + completed.add(old_url) + continue ref = self.references.get(old_url) if ref is None: continue @@ -160,12 +184,19 @@ class OneBotImageURLs: 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) + cached = (images, time.monotonic() + 1800) 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: + completed.add(old_url) self.remember(new_url, message_id, image_id) + deadline = cached[1] + self.urls[old_url] = (new_url, deadline) + self.urls[new_url] = (new_url, deadline) + while len(self.urls) > 2048: + self.urls.popitem(last=False) node['url'] = new_url + return completed diff --git a/src/langbot/pkg/platform/adapters/aiocqhttp/transport.py b/src/langbot/pkg/platform/adapters/aiocqhttp/transport.py index b1ac06ee9..63df03dfc 100644 --- a/src/langbot/pkg/platform/adapters/aiocqhttp/transport.py +++ b/src/langbot/pkg/platform/adapters/aiocqhttp/transport.py @@ -102,6 +102,7 @@ class OneBotConnection: self.image_urls.keys.clear() self.image_urls.references.clear() self.image_urls.messages.clear() + self.image_urls.urls.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 index fb6ef7001..6c16993f9 100644 --- a/src/langbot/pkg/platform/image_urls.py +++ b/src/langbot/pkg/platform/image_urls.py @@ -1,4 +1,4 @@ -"""Refresh image references at authorized read/use boundaries, without persistence.""" +"""Refresh image references at authorized read/use boundaries.""" import copy import json @@ -21,7 +21,14 @@ async def refresh_image_payload(ap, workspace_uuid, bot_id, payload): return payload result = copy.deepcopy(payload) try: - await connection.image_urls.refresh(result) + from ..persistence.resource_urls import refresh_images + + await refresh_images( + ap.persistence_mgr.get_db_engine(), + connection, + result, + is_current=lambda: bot.adapter.bot.connection is connection, + ) if bot.adapter.bot.connection is not connection or connection.closed: return payload return result diff --git a/tests/integration/persistence/test_monitoring_postgres.py b/tests/integration/persistence/test_monitoring_postgres.py index d0bfa6b41..b1c9931b0 100644 --- a/tests/integration/persistence/test_monitoring_postgres.py +++ b/tests/integration/persistence/test_monitoring_postgres.py @@ -9,9 +9,11 @@ from __future__ import annotations import datetime import logging +import json import os import uuid from types import SimpleNamespace +from unittest.mock import AsyncMock import pytest import pytest_asyncio @@ -365,6 +367,43 @@ async def test_rls_is_enforced_without_application_workspace_predicates(service, await service.ap.persistence_mgr.execute_async(sa.select(models.MonitoringMessage.id)) +async def test_shared_resource_urls_preserve_cloud_workspace_authorization(service, cloud_database): + from langbot.pkg.entity.persistence.resource_url import ResourceURL + from langbot.pkg.persistence.resource_urls import refresh_images + from langbot.pkg.platform.adapters.aiocqhttp.image_urls import OneBotImageURLs + + _, admin = cloud_database + old, fresh = 'https://example.test/old-image', 'https://example.test/current-image' + chain = [{'type': 'Source', 'id': 42}, {'type': 'Image', 'image_id': 'one.jpg', 'url': old}] + records = {} + for workspace in (WORKSPACE_A, WORKSPACE_B): + records[workspace] = await service.record_message( + _context(workspace), **RESOURCE, session_id='shared-url', message_content=json.dumps(chain) + ) + async with admin.begin() as db: + assert await db.scalar(sa.select(sa.func.count()).select_from(ResourceURL)) == 1 + await db.execute(sa.update(ResourceURL).values(created_at=datetime.datetime(2000, 1, 1))) + + async def action(name, **params): + if name == 'get_version_info': + return {'app_name': 'standard'} + assert name == 'get_msg' and params == {'message_id': 42} + return {'message': [{'type': 'image', 'data': {'file': 'one.jpg', 'url': fresh}}]} + + peer = SimpleNamespace(closed=False, call_action=AsyncMock(side_effect=action)) + peer.image_urls = OneBotImageURLs(peer) + # Exercise catalog writes using the actual non-superuser runtime role. + await refresh_images(service.ap.persistence_mgr.get_db_engine(), peer, chain) + for workspace in (WORKSPACE_A, WORKSPACE_B): + details = await _read(service, 'get_message_details', _context(workspace), records[workspace]) + assert fresh in details['message']['message_content'] + other = WORKSPACE_B if workspace == WORKSPACE_A else WORKSPACE_A + assert (await _read(service, 'get_message_details', _context(workspace), records[other]))['found'] is False + async with admin.connect() as db: + assert await db.scalar(sa.select(ResourceURL.url)) == fresh + assert old not in str((await db.execute(sa.text('SELECT message_content FROM monitoring_messages'))).all()) + + async def test_traffic_series_aggregates_all_rows_under_cloud_rls(service): import datetime from langbot.pkg.api.http.service.monitoring_traffic import get_traffic_series diff --git a/tests/unit_tests/api/service/test_monitoring_executions.py b/tests/unit_tests/api/service/test_monitoring_executions.py index 2029c1e06..45b28ba55 100644 --- a/tests/unit_tests/api/service/test_monitoring_executions.py +++ b/tests/unit_tests/api/service/test_monitoring_executions.py @@ -84,12 +84,13 @@ async def test_event_properties_survive_detail_snapshot_fallback(engine, service @pytest.mark.asyncio @pytest.mark.parametrize('napcat', [True, False]) @pytest.mark.parametrize('source', ['agent', 'pipeline']) -async def test_execution_details_refresh_existing_images_without_persisting(engine, service, napcat, source): +async def test_execution_details_refresh_existing_images_through_shared_catalog(engine, service, napcat, source): import json from unittest.mock import AsyncMock import sqlalchemy as sa from langbot.pkg.entity.persistence.event_log import EventLog from langbot.pkg.entity.persistence.transcript import Transcript + from langbot.pkg.entity.persistence.resource_url import ResourceURL from langbot.pkg.platform.adapters.aiocqhttp.image_urls import OneBotImageURLs old = 'https://multimedia.nt.qq.com.cn/download?appid=1407&fileid=existing&rkey=expired' @@ -140,6 +141,9 @@ async def test_execution_details_refresh_existing_images_without_persisting(engi ) await session.commit() + async with engine.begin() as db: + await db.execute(sa.update(ResourceURL).values(created_at=_dt(10))) + async def action(name, **params): if name == 'get_version_info': return {'app_name': 'NapCat.Onebot' if napcat else 'standard-peer'} @@ -170,8 +174,10 @@ async def test_execution_details_refresh_existing_images_without_persisting(engi await service.get_execution_detail(WORKSPACE, source, identity, section='conversation') assert peer.call_action.await_count == 2 async with engine.connect() as db: - assert old in (await db.execute(sa.select(EventLog.input_json))).scalar_one() - assert old in (await db.execute(sa.select(Transcript.content_json))).scalar_one() + assert new in (await db.execute(sa.select(EventLog.input_json))).scalar_one() + assert new in (await db.execute(sa.select(Transcript.content_json))).scalar_one() + raw = (await db.execute(sa.text('SELECT input_json FROM event_log'))).scalar_one() + assert 'langbot-resource://' in raw and old not in raw and new not in raw def test_event_content_preserves_conflicting_custom_input_fields(): diff --git a/tests/unit_tests/persistence/test_resource_urls.py b/tests/unit_tests/persistence/test_resource_urls.py new file mode 100644 index 000000000..320baefd6 --- /dev/null +++ b/tests/unit_tests/persistence/test_resource_urls.py @@ -0,0 +1,367 @@ +"""URL catalog storage, transactional refresh and migration contracts.""" + +import asyncio +import datetime as dt +import importlib +import json +import os +import uuid +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine + +from langbot.pkg.entity.persistence.resource_url import ResourceURL +from langbot.pkg.entity.persistence.event_log import EventLog +from langbot.pkg.entity.persistence.monitoring import MonitoringLLMCall +from langbot.pkg.persistence.media import MediaReferenceText, MediaReferenceJSON +from langbot.pkg.persistence.resource_urls import digest, refresh_images +from langbot.pkg.platform.adapters.aiocqhttp.image_urls import OneBotImageURLs + + +OLD = 'https://multimedia.nt.qq.com.cn/download?appid=1407&fileid=one&rkey=old' +NEW = OLD.replace('rkey=old', 'rkey=new') + + +def chain(url=OLD, message_id=42): + return [{'type': 'Source', 'id': message_id}, {'type': 'Image', 'image_id': 'one.jpg', 'url': url}] + + +@pytest.fixture(params=['sqlite', 'postgresql']) +async def database(tmp_path, request): + admin = None + schema = None + if request.param == 'postgresql': + url = os.environ.get('TEST_POSTGRES_URL') + if not url: + pytest.skip('TEST_POSTGRES_URL not set') + admin = create_async_engine(url) + schema = 'resource_' + uuid.uuid4().hex + async with admin.begin() as db: + await db.execute(sa.schema.CreateSchema(schema)) + engine = create_async_engine(url, connect_args={'server_settings': {'search_path': schema}}) + else: + engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "catalog.db"}') + try: + async with engine.begin() as db: + await db.run_sync(ResourceURL.__table__.create) + await db.run_sync(EventLog.__table__.create) + yield engine + finally: + await engine.dispose() + if admin is not None: + async with admin.begin() as db: + await db.execute(sa.schema.DropSchema(schema, cascade=True)) + await admin.dispose() + + +async def append(db, identity, url=OLD): + await db.execute( + sa.insert(EventLog).values( + event_id=identity, + event_type='message.received', + source='platform', + workspace_id='workspace', + bot_id='bot', + input_json=json.dumps(chain(url)), + ) + ) + + +@pytest.mark.asyncio +async def test_dedup_expansion_and_rollback(database): + async with database.begin() as db: + await append(db, 'a') + await append(db, 'b') + async with database.connect() as db: + assert (await db.scalar(sa.select(sa.func.count()).select_from(ResourceURL))) == 1 + stored = (await db.execute(sa.text('SELECT input_json FROM event_log'))).scalars().all() + assert stored[0] == stored[1] and OLD not in stored[0] + assert OLD in await db.scalar(sa.select(EventLog.input_json).where(EventLog.event_id == 'a')) + # The original string remains usable for compare-and-swap updates. + value = await db.scalar(sa.select(EventLog.input_json).where(EventLog.event_id == 'a')) + result = await db.execute( + sa.update(EventLog) + .where(EventLog.event_id == 'a', EventLog.input_json == value) + .values(input_json=json.dumps(chain(NEW))) + ) + assert result.rowcount == 1 + await db.rollback() + async with database.connect() as db: + assert await db.scalar(sa.select(sa.func.count()).select_from(ResourceURL)) == 1 + + +@pytest.mark.asyncio +async def test_untrusted_content_cannot_inject_catalog_ids(database): + async with database.begin() as db: + await append(db, 'a') + identity = await db.scalar(sa.select(ResourceURL.id)) + with pytest.raises(sa.exc.StatementError, match='Internal resource references'): + async with database.begin() as db: + await append(db, 'b', 'langbot-resource://' + identity) + + +@pytest.mark.asyncio +async def test_orm_update_keeps_loaded_payload_synchronized(database): + async with database.begin() as db: + await append(db, 'a') + async with AsyncSession(database, expire_on_commit=False) as session: + record = await session.scalar(sa.select(EventLog)) + changed = json.dumps(chain(NEW)) + await session.execute(sa.update(EventLog).where(EventLog.event_id == 'a').values(input_json=changed)) + await session.commit() + assert record.input_json == changed + await session.refresh(record) + assert NEW in record.input_json + + +@pytest.mark.asyncio +async def test_bulk_and_native_json_preserve_api_shapes(database): + meta = sa.MetaData() + sample = sa.Table( + 'sample', + meta, + sa.Column('id', sa.Integer, primary_key=True), + sa.Column('body', MediaReferenceText), + sa.Column('detail', MediaReferenceJSON), + ) + async with database.begin() as db: + await db.run_sync(meta.create_all) + await db.execute( + sample.insert(), + [dict(id=i, body=json.dumps(chain()), detail={'url': OLD, 'flag': False, 'number': 0}) for i in range(3)], + ) + rows = (await db.execute(sa.select(sample))).mappings().all() + assert all(OLD in row['body'] and row['detail'] == {'url': OLD, 'flag': False, 'number': 0} for row in rows), ( + rows + ) + assert await db.scalar(sa.select(sa.func.count()).select_from(ResourceURL)) == 1 + assert ( + len( + ( + await db.execute( + sample.insert().returning(sample.c.id), + [dict(id=i, body=json.dumps(chain()), detail={'url': OLD}) for i in range(3, 6)], + ) + ).all() + ) + == 3 + ) + assert all(OLD in value for value in (await db.execute(sa.select(sample.c.body))).scalars()) + + +@pytest.mark.asyncio +@pytest.mark.parametrize('catalog_url', [OLD, NEW]) +async def test_fresh_image_remembers_origin_for_later_model_history(database, catalog_url): + async with database.begin() as db: + await append(db, 'a') + await db.execute(sa.update(ResourceURL).values(url=catalog_url, url_hash=digest(catalog_url))) + + renewed = NEW + '&renewed=1' + + async def action(name, **params): + if name == 'get_version_info': + return {'app_name': 'standard'} + assert name == 'get_msg' and params == {'message_id': 42} + return {'message': [{'type': 'image', 'data': {'file': 'one.jpg', 'url': renewed}}]} + + peer = SimpleNamespace(closed=False, call_action=AsyncMock(side_effect=action)) + peer.image_urls = OneBotImageURLs(peer) + initial = chain() + await refresh_images(database, peer, initial) + assert initial[1]['url'] == catalog_url + assert peer.image_urls.references[catalog_url] == (42, 'one.jpg') + assert not any(call.args[0] == 'get_msg' for call in peer.call_action.call_args_list) + + async with database.begin() as db: + assert await db.scalar(sa.select(ResourceURL.last_checked_at)) is None + await db.execute(sa.update(ResourceURL).values(created_at=dt.datetime(2000, 1, 1))) + history = [{'type': 'image_url', 'image_url': {'url': catalog_url}}] + await refresh_images(database, peer, history) + assert history[0]['image_url']['url'] == renewed + assert sum(call.args[0] == 'get_msg' for call in peer.call_action.call_args_list) == 1 + async with database.connect() as db: + assert renewed in await db.scalar(sa.select(EventLog.input_json)) + assert await db.scalar(sa.select(ResourceURL.last_checked_at)) is not None + + +@pytest.mark.asyncio +@pytest.mark.parametrize('converge', [False, True]) +async def test_refresh_shared_identity_concurrency_and_half_hour(database, converge): + async with database.begin() as db: + await append(db, 'a') + if converge: + await append(db, 'b', NEW) + await db.execute(sa.update(ResourceURL).values(created_at=dt.datetime(2000, 1, 1))) + + async def action(name, **params): + await asyncio.sleep(0) + if name == 'get_version_info': + return {'app_name': 'other'} + assert name == 'get_msg' + return {'message': [{'type': 'image', 'data': {'file': 'one.jpg', 'url': NEW}}]} + + def peer(): + result = SimpleNamespace(closed=False, call_action=AsyncMock(side_effect=action)) + result.image_urls = OneBotImageURLs(result) + return result + + peers = [peer(), peer()] + batches = [chain(message_id=42), chain(message_id=43)] + await asyncio.gather(*(refresh_images(database, p, batch) for p, batch in zip(peers, batches))) + assert all(batch[1]['url'] == NEW for batch in batches) + assert sum(call.args[0] == 'get_msg' for p in peers for call in p.call_action.call_args_list) == 1 + async with database.connect() as db: + assert NEW in await db.scalar(sa.select(EventLog.input_json).where(EventLog.event_id == 'a')) + assert ( + await db.scalar(sa.select(sa.func.count()).select_from(ResourceURL).where(ResourceURL.url.is_not(None))) + == 1 + ) + assert OLD not in str((await db.execute(sa.select(ResourceURL.url))).scalars().all()) + # A fresh connection still reuses the persisted last-check time. + replacement = peer() + await refresh_images(database, replacement, chain(NEW)) + assert not any(call.args[0] == 'get_msg' for call in replacement.call_action.call_args_list) + async with database.begin() as db: + await db.execute(sa.update(ResourceURL).values(last_checked_at=dt.datetime(2000, 1, 1))) + await refresh_images(database, replacement, chain(NEW)) + assert any(call.args[0] == 'get_msg' for call in replacement.call_action.call_args_list) + + +@pytest.mark.asyncio +async def test_distinct_urls_can_converge_concurrently(database): + old_urls = [OLD, OLD + '&variant=two'] + async with database.begin() as db: + for index, url in enumerate(old_urls): + await append(db, str(index), url) + await db.execute(sa.update(ResourceURL).values(created_at=dt.datetime(2000, 1, 1))) + ready = asyncio.Event() + entered = 0 + + async def action(name, **params): + nonlocal entered + if name == 'get_version_info': + return {'app_name': 'standard'} + entered += 1 + if database.dialect.name == 'postgresql': + if entered == 2: + ready.set() + await asyncio.wait_for(ready.wait(), 5) + return {'message': [{'type': 'image', 'data': {'file': 'one.jpg', 'url': NEW}}]} + + peers = [SimpleNamespace(closed=False, call_action=AsyncMock(side_effect=action)) for _ in old_urls] + for peer in peers: + peer.image_urls = OneBotImageURLs(peer) + await asyncio.gather(*(refresh_images(database, peer, chain(url)) for peer, url in zip(peers, old_urls))) + async with database.connect() as db: + assert all(NEW in value for value in (await db.execute(sa.select(EventLog.input_json))).scalars()) + assert (await db.execute(sa.select(ResourceURL.url).where(ResourceURL.url.is_not(None)))).scalars().all() == [ + NEW + ] + + +@pytest.mark.asyncio +@pytest.mark.parametrize('outcome', ['unchanged', 'missing_origin', 'failure']) +async def test_only_successful_checks_advance_the_persisted_interval(database, outcome): + async with database.begin() as db: + await append(db, 'a') + await db.execute(sa.update(ResourceURL).values(created_at=dt.datetime(2000, 1, 1))) + + async def action(name, **params): + if name == 'get_version_info': + return {'app_name': 'standard'} + if outcome == 'failure': + raise RuntimeError('upstream unavailable') + return {'message': [{'type': 'image', 'data': {'file': 'one.jpg', 'url': OLD}}]} + + peer = SimpleNamespace(closed=False, call_action=AsyncMock(side_effect=action)) + peer.image_urls = OneBotImageURLs(peer) + payload = chain() if outcome != 'missing_origin' else [{'type': 'image_url', 'image_url': {'url': OLD}}] + if outcome == 'failure': + with pytest.raises(RuntimeError, match='upstream unavailable'): + await refresh_images(database, peer, payload) + else: + await refresh_images(database, peer, payload) + async with database.connect() as db: + assert (await db.scalar(sa.select(ResourceURL.last_checked_at)) is not None) == (outcome == 'unchanged') + + +@pytest.mark.asyncio +async def test_napcat_uses_key_ttl_instead_of_the_standard_interval(database): + async with database.begin() as db: + await append(db, 'a') + generation = 0 + + async def action(name, **params): + nonlocal generation + if name == 'get_version_info': + return {'app_name': 'NapCat.Onebot'} + assert name == 'nc_get_rkey' + generation += 1 + return [{'type': 20, 'time': 1, 'ttl': 10**12, 'rkey': f'key{generation}'}] + + peer = SimpleNamespace(closed=False, call_action=AsyncMock(side_effect=action)) + peer.image_urls = OneBotImageURLs(peer) + await refresh_images(database, peer, chain()) + async with database.connect() as db: + checked = await db.scalar(sa.select(ResourceURL.last_checked_at)) + assert 'rkey=key1' in await db.scalar(sa.select(EventLog.input_json)) + await refresh_images(database, peer, chain()) + assert generation == 1 + async with database.connect() as db: + assert await db.scalar(sa.select(ResourceURL.last_checked_at)) == checked + peer.image_urls.keys[20] = ('key1', 0) + await refresh_images(database, peer, chain()) + assert generation == 2 + async with database.connect() as db: + assert 'rkey=key2' in await db.scalar(sa.select(EventLog.input_json)) + + +@pytest.mark.asyncio +async def test_migration_preserves_source_creation_and_downgrade(database): + migration = importlib.import_module('langbot.pkg.persistence.alembic.versions.0041_resource_urls') + async with database.begin() as db: + await append(db, 'newer') + historical = sa.Table( + 'monitoring_llm_calls', + sa.MetaData(), + sa.Column('id', sa.String, primary_key=True), + sa.Column('details', sa.JSON), + ) + await db.run_sync(historical.create) + await db.execute(historical.insert().values(id='old', details={'url': OLD, 'flag': False})) + # Simulate a historical write before catalog normalization existed. + await db.execute( + sa.text( + 'INSERT INTO event_log (event_id,event_type,source,input_json,created_at) ' + "VALUES ('old','message.received','platform',:payload,'2000-01-01 00:00:00')" + ), + {'payload': json.dumps(chain())}, + ) + + def upgrade(sync): + with Operations.context(MigrationContext.configure(sync)): + migration.upgrade() + migration.upgrade() + + await db.run_sync(upgrade) + assert OLD in await db.scalar(sa.select(EventLog.input_json)) + assert (await db.scalar(sa.select(ResourceURL.created_at))).year == 2000 + assert OLD not in await db.scalar(sa.text('SELECT input_json FROM event_log')) + assert await db.scalar(sa.select(MonitoringLLMCall.details)) == {'url': OLD, 'flag': False} + assert 'langbot-resource://' in str(await db.scalar(sa.select(historical.c.details))) + + def downgrade(sync): + with Operations.context(MigrationContext.configure(sync)): + migration.downgrade() + + await db.run_sync(downgrade) + assert OLD in await db.scalar(sa.text('SELECT input_json FROM event_log')) + assert await db.scalar(sa.select(historical.c.details)) == {'url': OLD, 'flag': False} + # The same engine can write and read literal payloads after downgrade. + await append(db, 'after', NEW) + assert NEW in await db.scalar(sa.select(EventLog.input_json).where(EventLog.event_id == 'after')) diff --git a/tests/unit_tests/platform/test_onebot_image_urls.py b/tests/unit_tests/platform/test_onebot_image_urls.py index c11c934af..dd5afe60c 100644 --- a/tests/unit_tests/platform/test_onebot_image_urls.py +++ b/tests/unit_tests/platform/test_onebot_image_urls.py @@ -2,13 +2,17 @@ import asyncio import copy +import datetime import json from types import SimpleNamespace from unittest.mock import AsyncMock, Mock from urllib.parse import parse_qs, urlsplit import pytest +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import create_async_engine +from langbot.pkg.entity.persistence.resource_url import ResourceURL 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 @@ -105,6 +109,7 @@ async def test_other_implementation_uses_original_id_and_caches_messages(): await peer.image_urls.refresh(model_input) assert peer.call_action.await_count == 2 peer.image_urls.messages.clear() + peer.image_urls.urls.clear() await peer.image_urls.refresh(model_input) assert peer.call_action.await_count == 3 @@ -128,21 +133,34 @@ async def test_aggregated_messages_keep_each_original_source(): assert peer.call_action.await_count == 3 -def application(peer, workspace='workspace'): +@pytest.fixture +async def catalog(): + engine = create_async_engine('sqlite+aiosqlite:///:memory:') + async with engine.begin() as db: + await db.run_sync(ResourceURL.__table__.create) + yield engine + await engine.dispose() + + +def application(peer, engine, 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()) + ap = SimpleNamespace( + platform_mgr=SimpleNamespace(resolve_public_bot=AsyncMock(return_value=bot)), + persistence_mgr=SimpleNamespace(get_db_engine=lambda: engine), + logger=Mock(), + ) return ap, bot @pytest.mark.asyncio -async def test_scope_failure_backoff_and_no_persistence(): +async def test_scope_failure_backoff_preserves_current_url(catalog): peer = connection(lambda *_args, **_kwargs: {'app_name': 'NapCat.Onebot'}) - ap, bot = application(peer) + ap, bot = application(peer, catalog) original = payload(url()) saved = copy.deepcopy(original) assert await refresh_image_payload(ap, 'foreign', 'bot', original) is original @@ -158,14 +176,14 @@ async def test_scope_failure_backoff_and_no_persistence(): @pytest.mark.asyncio -async def test_model_reuse_refreshes_with_current_keys_without_mutating_messages(): +async def test_model_reuse_refreshes_with_current_keys_without_mutating_messages(catalog): 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) + ap, _ = application(peer, catalog) messages = [Message(role='user', content=[{'type': 'image_url', 'image_url': {'url': url()}}])] context = SimpleNamespace(workspace_uuid='workspace') query = SimpleNamespace(bot_uuid='bot') @@ -180,7 +198,7 @@ async def test_model_reuse_refreshes_with_current_keys_without_mutating_messages @pytest.mark.asyncio -async def test_disconnected_peer_cannot_return_refreshed_result(): +async def test_disconnected_peer_cannot_return_refreshed_result(catalog): async def action(name, **_params): if name == 'get_version_info': return {'app_name': 'NapCat.Onebot'} @@ -188,18 +206,18 @@ async def test_disconnected_peer_cannot_return_refreshed_result(): return [{'type': 20, 'time': 1, 'ttl': 10**12, 'rkey': 'new'}] peer = connection(action) - ap, bot = application(peer) + ap, bot = application(peer, catalog) original = payload(url()) assert await refresh_image_payload(ap, 'workspace', 'bot', original) is original + async with catalog.connect() as db: + assert await db.scalar(sa.select(ResourceURL.url)) == url() @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 +async def test_history_recovers_scoped_origins_and_updates_shared_url(catalog): from langbot.pkg.entity.persistence.event_log import EventLog - engine = create_async_engine('sqlite+aiosqlite:///:memory:') + engine = catalog original = payload('https://example.test/old') metadata = json.dumps({'input_event': {'message_chain': original}}) async with engine.begin() as db: @@ -214,6 +232,7 @@ async def test_history_recovers_scoped_origins_without_rewriting_event_facts(): metadata_json=metadata, ) ) + await db.execute(sa.update(ResourceURL).values(created_at=datetime.datetime(2000, 1, 1))) async def execute(statement): async with engine.connect() as db: @@ -226,8 +245,8 @@ async def test_history_recovers_scoped_origins_without_rewriting_event_facts(): 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) + ap, _ = application(peer, catalog) + ap.persistence_mgr.execute_async = execute item = { 'bot_id': 'bot', 'workspace_id': 'workspace', @@ -240,7 +259,9 @@ async def test_history_recovers_scoped_origins_without_rewriting_event_facts(): 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 + assert 'https://example.test/new' in result.scalar_one() + async with engine.connect() as db: + assert 'langbot-resource://' in await db.scalar(sa.text('SELECT metadata_json FROM event_log')) finally: await engine.dispose() diff --git a/tests/unit_tests/provider/test_request_policy.py b/tests/unit_tests/provider/test_request_policy.py index 8161a5509..23143da0f 100644 --- a/tests/unit_tests/provider/test_request_policy.py +++ b/tests/unit_tests/provider/test_request_policy.py @@ -5,7 +5,10 @@ from unittest.mock import AsyncMock, Mock, patch import httpx import litellm import pytest +from sqlalchemy.ext.asyncio import create_async_engine +from langbot.pkg.entity.persistence.resource_url import ResourceURL +from langbot.pkg.persistence import resource_urls from langbot.pkg.provider.modelmgr.errors import RequesterError from langbot.pkg.provider.modelmgr.request_policy import ( CallState, @@ -20,10 +23,22 @@ from tests.unit_tests.provider.conftest import TEST_EXECUTION_CONTEXT from tests.unit_tests.provider.test_litellmchat import MockRuntimeModel +@pytest.fixture +async def image_url_engine(tmp_path): + engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "resource_urls.db"}') + try: + async with engine.begin() as db: + await db.run_sync(ResourceURL.__table__.create) + resource_urls.install(engine.sync_engine) + yield engine + finally: + await engine.dispose() + + @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 + runtime_provider, runtime_llm_model, monkeypatch, streaming, image_url_engine ): from langbot.pkg.platform.adapters.aiocqhttp.image_urls import OneBotImageURLs from langbot_plugin.api.entities.builtin.pipeline.query import Query @@ -46,6 +61,7 @@ async def test_image_credentials_are_checked_again_before_retry( adapter=SimpleNamespace(bot=SimpleNamespace(connection=peer)), ) runtime_provider.requester.ap.platform_mgr = SimpleNamespace(resolve_public_bot=AsyncMock(return_value=bot)) + runtime_provider.requester.ap.persistence_mgr.get_db_engine = lambda: image_url_engine 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' From 97204a5d9470e0de07acf75c6b6eecfee8c8efc7 Mon Sep 17 00:00:00 2001 From: BiFangKNT <1320414964@qq.com> Date: Fri, 23 Oct 2026 12:51:28 +0800 Subject: [PATCH 5/7] docs(agents): Update test run instructions - Add recommendation for using Linux containers on Windows - Emphasize running tests on Linux to match CI --- AGENTS.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/AGENTS.md b/AGENTS.md index 5dd1352e2..938eef4d9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -40,6 +40,8 @@ pnpm test:e2e Run the narrowest useful test first, then broader checks when confidence is needed. +Run tests in Linux to match CI; on Windows, use Linux containers through `wslc` of WSL3 to avoid platform-specific failures. + ## Where to Look - Architecture map: `ARCHITECTURE.md`. From 5d774ff555b0c83c42fd1819315261c63ec66a04 Mon Sep 17 00:00:00 2001 From: BiFangKNT <1320414964@qq.com> Date: Fri, 23 Oct 2026 14:39:38 +0800 Subject: [PATCH 6/7] fix(persistence): resolve resource URL migration and scoped SQL failures - Reflect migration tables without resolving foreign-key targets - Allow validated resource URL expansion through the tenant SQL AST guard - Add regression coverage for trusted and malformed expansion expressions --- .../alembic/versions/0041_resource_urls.py | 3 ++- src/langbot/pkg/persistence/tenant_uow.py | 18 ++++++++-------- .../unit_tests/persistence/test_tenant_uow.py | 21 +++++++++++++++++++ 3 files changed, 32 insertions(+), 10 deletions(-) diff --git a/src/langbot/pkg/persistence/alembic/versions/0041_resource_urls.py b/src/langbot/pkg/persistence/alembic/versions/0041_resource_urls.py index 0030a73b8..aaa90bd0a 100644 --- a/src/langbot/pkg/persistence/alembic/versions/0041_resource_urls.py +++ b/src/langbot/pkg/persistence/alembic/versions/0041_resource_urls.py @@ -35,7 +35,8 @@ def upgrade(): if not retained: continue # Reflection intentionally reads literal stored values without expansion. - source = sa.Table(table.name, sa.MetaData(), autoload_with=connection) + # Only local columns are needed; historical schemas may omit FK targets. + source = sa.Table(table.name, sa.MetaData(), autoload_with=connection, resolve_fks=False) retained = [name for name in retained if name in source.c] keys = list(source.primary_key.columns) if len(keys) != 1: diff --git a/src/langbot/pkg/persistence/tenant_uow.py b/src/langbot/pkg/persistence/tenant_uow.py index 2b6a1093d..e8ca31765 100644 --- a/src/langbot/pkg/persistence/tenant_uow.py +++ b/src/langbot/pkg/persistence/tenant_uow.py @@ -412,8 +412,16 @@ def _validate_scoped_statement_call(args: tuple[typing.Any, ...], kwargs: dict[s 'TenantUnitOfWork public SQL must use a structured SQLAlchemy query, values, or DML statement' ) + from .resource_urls import ExpandedResourceURLs + for element in _iter_scoped_statement_elements(statement): - if not type(element).__module__.startswith('sqlalchemy.'): + resource_expansion_allowed = ( + type(element) is ExpandedResourceURLs + and len(list(element.clauses)) == 1 + and isinstance(list(element.clauses)[0], sqlalchemy.Column) + and type(list(element.clauses)[0].type) is MediaReferenceText + ) + if not type(element).__module__.startswith('sqlalchemy.') and not resource_expansion_allowed: raise ScopedSessionTransactionError( 'TenantUnitOfWork does not allow custom SQL AST nodes in public statements' ) @@ -503,14 +511,6 @@ def _validate_scoped_statement_call(args: tuple[typing.Any, ...], kwargs: dict[s ) if isinstance(element, sqlalchemy.sql.functions.FunctionElement): - from .resource_urls import ExpandedResourceURLs - - resource_expansion_allowed = ( - type(element) is ExpandedResourceURLs - and len(list(element.clauses)) == 1 - and isinstance(list(element.clauses)[0], sqlalchemy.Column) - and type(list(element.clauses)[0].type) is MediaReferenceText - ) function_name = str(getattr(element, 'name', '')).casefold() package_names = tuple(getattr(element, 'packagenames', ())) generic_function_allowed = ( diff --git a/tests/unit_tests/persistence/test_tenant_uow.py b/tests/unit_tests/persistence/test_tenant_uow.py index ee4c0b570..9a8b1c827 100644 --- a/tests/unit_tests/persistence/test_tenant_uow.py +++ b/tests/unit_tests/persistence/test_tenant_uow.py @@ -980,6 +980,27 @@ async def test_scoped_sql_structure_allows_only_the_production_vocabulary(statem _validate_scoped_statement_call((statement,), {}) +async def test_resource_expansion_allows_only_the_fixed_function_and_media_column() -> None: + from langbot.pkg.entity.persistence.event_log import EventLog + from langbot.pkg.persistence.resource_urls import ExpandedResourceURLs + + statement = sa.update(EventLog).where(EventLog.input_json == '{}').values(input_json='[]') + _validate_scoped_statement_call((statement,), {}) + + class SpoofedExpansion(ExpandedResourceURLs): + inherit_cache = True + + for expression in ( + ExpandedResourceURLs(), + ExpandedResourceURLs(sa.literal('{}')), + ExpandedResourceURLs(User.uuid), + ExpandedResourceURLs(EventLog.input_json, sa.literal('{}')), + SpoofedExpansion(EventLog.input_json), + ): + with pytest.raises(ScopedSessionTransactionError, match='custom SQL AST'): + _validate_scoped_statement_call((sa.select(expression),), {}) + + async def test_monitoring_json_casts_are_limited_to_journal_metadata() -> None: from langbot.pkg.entity.persistence.agent_run import AgentRun from langbot.pkg.entity.persistence.event_log import EventLog From 24ce4ebe47c0b4ac6a82690dd8c1754a638a0f33 Mon Sep 17 00:00:00 2001 From: BiFangKNT <1320414964@qq.com> Date: Fri, 23 Oct 2026 14:58:22 +0800 Subject: [PATCH 7/7] perf(persistence): remove unused resource URL indexes - Keep only primary-key and URL-hash unique indexes - Fold index cleanup into migration 0041 - Verify SQLite and PostgreSQL migration behavior --- ARCHITECTURE.md | 2 +- .../pkg/entity/persistence/resource_url.py | 3 +- .../alembic/versions/0041_resource_urls.py | 2 ++ .../persistence/test_resource_urls.py | 33 +++++++++++++++++++ 4 files changed, 37 insertions(+), 3 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index a18f99565..8553cdfb0 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -275,7 +275,7 @@ Persistence is centered on `pkg/persistence/mgr.py`. - SQLite is the default database; PostgreSQL is supported. - Models live under `pkg/entity/persistence/`. - `pkg/persistence/media.py::MediaReferenceText` validates bound conversation payloads on ORM/Core writes while preserving the existing TEXT schema and string API. It rejects recognized structured inline images: retained images go through `MediaCache.externalize_chain_dump()` before serialization, while metadata-only snapshots and transcripts use `redact_inline_images()`. Reads and comparisons preserve historical values for incremental cleanup. Runtime/adapter messages, arbitrary diagnostic text, and opaque strings are outside this contract. -- `pkg/persistence/resource_urls.py` interns structured resource URL fields in retained messages, history and monitoring at the typed SQL boundary, in the source transaction. `MediaReferenceJSON` preserves native JSON monitoring details. Payloads store opaque references to the global `resource_urls` catalog; typed SQL reads and comparisons expand them to current URLs, preserving API shapes and source-record authorization. Configuration URLs and URLs in free text are excluded. The URL hash has a unique index; canonical IDs and check timestamps are indexed. Successful refreshes overwrite the current URL and persist the check time even when standard `get_msg` returns the same URL. Converging URLs retain only ID redirects, with no old URL. PostgreSQL row locks and SQLite's immediate write transaction serialize concurrent refreshes; SQLite holds its write lock during the upstream call. Bounded in-memory aliases let reused Runner payloads resolve old URLs without retaining those URLs in the database. Migration `0041_resource_urls` normalizes existing payloads using their source creation timestamps before producers start; downgrade expands references before dropping the catalog. +- `pkg/persistence/resource_urls.py` interns structured resource URL fields in retained messages, history and monitoring at the typed SQL boundary, in the source transaction. `MediaReferenceJSON` preserves native JSON monitoring details. Payloads store opaque references to the global `resource_urls` catalog; typed SQL reads and comparisons expand them to current URLs, preserving API shapes and source-record authorization. Configuration URLs and URLs in free text are excluded. Catalog lookups use only the primary-key and URL-hash unique indexes. Successful refreshes overwrite the current URL and persist the check time even when standard `get_msg` returns the same URL. Converging URLs retain only ID redirects, with no old URL. PostgreSQL row locks and SQLite's immediate write transaction serialize concurrent refreshes; SQLite holds its write lock during the upstream call. Bounded in-memory aliases let reused Runner payloads resolve old URLs without retaining those URLs in the database. Migration `0041_resource_urls` normalizes existing payloads using their source creation timestamps before producers start; downgrade expands references before dropping the catalog. - Timezone-less SQL `DateTime` columns store UTC-naive values on both backends. Normalize aware values with `pkg/persistence/datetime_utils.py::as_naive_utc` at the bind boundary, including deadlines, leases, event times, and query/retention cutoffs; do not merely strip an offset. Existing naive rows represent UTC. Restore UTC awareness for application comparisons and epoch serialization. Runner ledger/event/transcript stores follow this existing schema contract without a data migration or changes to journal/tenant authorization. - Fresh schemas are created from current metadata, then Alembic migrations run to head. LangBot 4.x does not upgrade 3.x databases. - New schema changes use Alembic under `pkg/persistence/alembic/versions/`; there is no legacy migration chain in 4.x. diff --git a/src/langbot/pkg/entity/persistence/resource_url.py b/src/langbot/pkg/entity/persistence/resource_url.py index 6d9a19ced..03c0fa893 100644 --- a/src/langbot/pkg/entity/persistence/resource_url.py +++ b/src/langbot/pkg/entity/persistence/resource_url.py @@ -11,7 +11,7 @@ class ResourceURL(Base): id = sa.Column(sa.String(36), primary_key=True) url = sa.Column(sa.Text, nullable=True) url_hash = sa.Column(sa.String(64), nullable=True, unique=True) - canonical_id = sa.Column(sa.String(36), sa.ForeignKey('resource_urls.id'), nullable=True, index=True) + canonical_id = sa.Column(sa.String(36), sa.ForeignKey('resource_urls.id'), nullable=True) created_at = sa.Column(sa.DateTime, nullable=False) last_checked_at = sa.Column(sa.DateTime, nullable=True) updated_at = sa.Column(sa.DateTime, nullable=False) @@ -22,5 +22,4 @@ class ResourceURL(Base): 'OR (canonical_id IS NOT NULL AND url IS NULL AND url_hash IS NULL)', name='ck_resource_url_canonical', ), - sa.Index('ix_resource_urls_last_checked', 'last_checked_at'), ) diff --git a/src/langbot/pkg/persistence/alembic/versions/0041_resource_urls.py b/src/langbot/pkg/persistence/alembic/versions/0041_resource_urls.py index aaa90bd0a..3c5c8821f 100644 --- a/src/langbot/pkg/persistence/alembic/versions/0041_resource_urls.py +++ b/src/langbot/pkg/persistence/alembic/versions/0041_resource_urls.py @@ -20,6 +20,8 @@ def upgrade(): connection = op.get_bind() ResourceURL.__table__.create(connection, checkfirst=True) + op.drop_index('ix_resource_urls_canonical_id', table_name='resource_urls', if_exists=True) + op.drop_index('ix_resource_urls_last_checked', table_name='resource_urls', if_exists=True) install(connection.engine) if connection.dialect.name == 'postgresql': create_postgres_expansion(connection) diff --git a/tests/unit_tests/persistence/test_resource_urls.py b/tests/unit_tests/persistence/test_resource_urls.py index 320baefd6..45bab81a2 100644 --- a/tests/unit_tests/persistence/test_resource_urls.py +++ b/tests/unit_tests/persistence/test_resource_urls.py @@ -321,6 +321,39 @@ async def test_napcat_uses_key_ttl_instead_of_the_standard_interval(database): assert 'rkey=key2' in await db.scalar(sa.select(EventLog.input_json)) +@pytest.mark.asyncio +async def test_index_migration_preserves_primary_key_unique_hash_and_payload(database): + migration = importlib.import_module('langbot.pkg.persistence.alembic.versions.0041_resource_urls') + removed = {'ix_resource_urls_canonical_id', 'ix_resource_urls_last_checked'} + + def indexes(sync): + return {index['name'] for index in sa.inspect(sync).get_indexes('resource_urls')} + + def upgrade(sync): + with Operations.context(MigrationContext.configure(sync)): + migration.upgrade() + + def create_legacy_indexes(sync): + operations = Operations(MigrationContext.configure(sync)) + operations.create_index('ix_resource_urls_canonical_id', 'resource_urls', ['canonical_id']) + operations.create_index('ix_resource_urls_last_checked', 'resource_urls', ['last_checked_at']) + + async with database.begin() as db: + await append(db, 'a') + assert removed.isdisjoint(await db.run_sync(indexes)) + # Recreate the schema deployed before the indexes were removed. + await db.run_sync(create_legacy_indexes) + assert removed <= await db.run_sync(indexes) + await db.run_sync(upgrade) + await db.run_sync(upgrade) + assert removed.isdisjoint(await db.run_sync(indexes)) + primary_key = await db.run_sync(lambda sync: sa.inspect(sync).get_pk_constraint('resource_urls')) + assert primary_key['constrained_columns'] == ['id'] + constraints = await db.run_sync(lambda sync: sa.inspect(sync).get_unique_constraints('resource_urls')) + assert any(constraint['column_names'] == ['url_hash'] for constraint in constraints) + assert OLD in await db.scalar(sa.select(EventLog.input_json)) + + @pytest.mark.asyncio async def test_migration_preserves_source_creation_and_downgrade(database): migration = importlib.import_module('langbot.pkg.persistence.alembic.versions.0041_resource_urls')