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] 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