fix(pipelines): retain debug chat image history

This commit is contained in:
RockChinQ
2026-08-26 00:32:03 +08:00
parent 94fd3d274c
commit 47f5515fa9
7 changed files with 105 additions and 36 deletions
@@ -232,7 +232,7 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
@staticmethod @staticmethod
def _history_message_chain(message_chain: list[dict]) -> list[dict]: def _history_message_chain(message_chain: list[dict]) -> list[dict]:
"""Remove large transient payloads before retaining browser history.""" """Retain renderable references without storing large inline payloads."""
history = [] history = []
for component in message_chain: for component in message_chain:
@@ -551,7 +551,10 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
Image / Voice / File components uploaded from the web client carry a Image / Voice / File components uploaded from the web client carry a
storage key in ``path``. Resolve it to a base64 data URI so downstream storage key in ``path``. Resolve it to a base64 data URI so downstream
stages (multimodal LLM input and the Box sandbox inbox) have a usable stages (multimodal LLM input and the Box sandbox inbox) have a usable
payload, then drop the now-consumed storage object. payload. Keep image objects for the short-lived browser history so the
authenticated image endpoint can render them; normal upload retention
cleanup removes them later. Other attachment types are consumed
immediately because the chat history does not render them by path.
Args: Args:
message_chain_obj: 消息链对象列表 message_chain_obj: 消息链对象列表
@@ -606,12 +609,13 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
mime_type = mimetypes.guess_type(comp_path)[0] or 'application/octet-stream' mime_type = mimetypes.guess_type(comp_path)[0] or 'application/octet-stream'
component['base64'] = f'data:{mime_type};base64,{base64_str}' component['base64'] = f'data:{mime_type};base64,{base64_str}'
await storage_mgr.delete_scoped_object_key( if comp_type != 'Image':
execution_context, await storage_mgr.delete_scoped_object_key(
comp_path, execution_context,
expected_owner_type='upload_image', comp_path,
) expected_owner_type='upload_image',
component['path'] = '' )
component['path'] = ''
except Exception as e: except Exception as e:
await self.logger.error(f'Failed to load {comp_type} file {comp_path}: {e}') await self.logger.error(f'Failed to load {comp_type} file {comp_path}: {e}')
raise raise
@@ -2,9 +2,10 @@
The web debug client uploads Image / Voice / File components carrying a storage The web debug client uploads Image / Voice / File components carrying a storage
key in ``path``. This helper resolves each to a base64 data URI (so multimodal key in ``path``. This helper resolves each to a base64 data URI (so multimodal
LLM input and the Box sandbox inbox have usable bytes), then deletes the LLM input and the Box sandbox inbox have usable bytes). Image uploads remain as
consumed upload. Covers mimetype selection per type and fail-closed error authenticated history references until storage retention cleanup, while other
handling. consumed uploads are deleted. Covers mimetype selection per type and
fail-closed error handling.
""" """
from __future__ import annotations from __future__ import annotations
@@ -52,7 +53,7 @@ def _make_adapter(load_return=b'hello', load_side_effect=None):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_image_jpeg_mimetype_and_consumed_storage_key(): async def test_image_jpeg_mimetype_and_retained_history_key():
adapter, storage_mgr, _ = _make_adapter(load_return=b'\xff\xd8\xff') adapter, storage_mgr, _ = _make_adapter(load_return=b'\xff\xd8\xff')
path = f'{_UPLOAD_PREFIX}photo.jpg' path = f'{_UPLOAD_PREFIX}photo.jpg'
chain = [{'type': 'Image', 'path': path}] chain = [{'type': 'Image', 'path': path}]
@@ -61,12 +62,11 @@ async def test_image_jpeg_mimetype_and_consumed_storage_key():
expected_b64 = base64.b64encode(b'\xff\xd8\xff').decode('utf-8') expected_b64 = base64.b64encode(b'\xff\xd8\xff').decode('utf-8')
assert chain[0]['base64'] == f'data:image/jpeg;base64,{expected_b64}' assert chain[0]['base64'] == f'data:image/jpeg;base64,{expected_b64}'
assert chain[0]['path'] == '' assert chain[0]['path'] == path
storage_mgr.delete_scoped_object_key.assert_awaited_once_with( storage_mgr.delete_scoped_object_key.assert_not_awaited()
_CONTEXT,
path, history = adapter._history_message_chain(chain)
expected_owner_type='upload_image', assert history == [{'type': 'Image', 'path': path, 'base64': ''}]
)
def test_history_retains_storage_key_without_large_base64_payload(): def test_history_retains_storage_key_without_large_base64_payload():
@@ -95,18 +95,22 @@ async def test_image_defaults_to_png():
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_voice_uses_guessed_or_wav_mimetype(): async def test_voice_uses_guessed_or_wav_mimetype():
adapter, _, _ = _make_adapter() adapter, storage_mgr, _ = _make_adapter()
chain = [{'type': 'Voice', 'path': f'{_UPLOAD_PREFIX}clip.wav'}] chain = [{'type': 'Voice', 'path': f'{_UPLOAD_PREFIX}clip.wav'}]
await adapter._process_image_components(_make_connection(), chain) await adapter._process_image_components(_make_connection(), chain)
assert chain[0]['base64'].startswith('data:audio/') assert chain[0]['base64'].startswith('data:audio/')
assert chain[0]['path'] == ''
storage_mgr.delete_scoped_object_key.assert_awaited_once()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_file_uses_octet_stream_fallback(): async def test_file_uses_octet_stream_fallback():
adapter, _, _ = _make_adapter() adapter, storage_mgr, _ = _make_adapter()
chain = [{'type': 'File', 'path': f'{_UPLOAD_PREFIX}unknownblob'}] chain = [{'type': 'File', 'path': f'{_UPLOAD_PREFIX}unknownblob'}]
await adapter._process_image_components(_make_connection(), chain) await adapter._process_image_components(_make_connection(), chain)
assert chain[0]['base64'].startswith('data:application/octet-stream;base64,') assert chain[0]['base64'].startswith('data:application/octet-stream;base64,')
assert chain[0]['path'] == ''
storage_mgr.delete_scoped_object_key.assert_awaited_once()
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -482,7 +482,7 @@ async def test_attachment_key_must_belong_to_connection_upload_scope():
await adapter._process_image_components(connection, message_chain) await adapter._process_image_components(connection, message_chain)
assert message_chain[0]['base64'].startswith('data:image/png;base64,') assert message_chain[0]['base64'].startswith('data:image/png;base64,')
assert message_chain[0]['path'] == '' assert message_chain[0]['path'] == 'v1/current/upload_image/key.png'
storage_mgr.scoped_prefix.assert_called_once_with( storage_mgr.scoped_prefix.assert_called_once_with(
connection.execution_context, connection.execution_context,
owner_type='upload_image', owner_type='upload_image',
@@ -496,11 +496,7 @@ async def test_attachment_key_must_belong_to_connection_upload_scope():
'v1/current/upload_image/key.png', 'v1/current/upload_image/key.png',
expected_owner_type='upload_image', expected_owner_type='upload_image',
) )
storage_mgr.delete_scoped_object_key.assert_awaited_once_with( storage_mgr.delete_scoped_object_key.assert_not_awaited()
connection.execution_context,
'v1/current/upload_image/key.png',
expected_owner_type='upload_image',
)
with pytest.raises(ValueError, match='does not belong'): with pytest.raises(ValueError, match='does not belong'):
await adapter._process_image_components( await adapter._process_image_components(
@@ -457,7 +457,7 @@ export default function DebugDialog({
return; return;
} }
const messageChain = []; const messageChain: MessageChainComponent[] = [];
// Add quoted message if present // Add quoted message if present
if (quotedMessage) { if (quotedMessage) {
@@ -511,17 +511,21 @@ export default function DebugDialog({
type: 'Image', type: 'Image',
path: result.file_key, path: result.file_key,
}); });
} else { } else if (attachment.kind === 'voice') {
// Voice / File go through the generic document upload endpoint, // Voice / File go through the generic document upload endpoint,
// which returns a storage key the backend resolves into the // which returns a storage key the backend resolves into the
// sandbox inbox just like images. // sandbox inbox just like images.
const result = await httpClient.uploadDocumentFile(attachment.file); const result = await httpClient.uploadDocumentFile(attachment.file);
messageChain.push({ messageChain.push({
type: attachment.kind === 'voice' ? 'Voice' : 'File', type: 'Voice',
path: result.file_id, path: result.file_id,
...(attachment.kind === 'file' });
? { name: attachment.file.name } } else {
: {}), const result = await httpClient.uploadDocumentFile(attachment.file);
messageChain.push({
type: 'File',
path: result.file_id,
name: attachment.file.name,
}); });
} }
} catch (error) { } catch (error) {
+1 -1
View File
@@ -19,7 +19,7 @@ export interface Plain extends MessageComponent {
// Quote component // Quote component
export interface Quote extends MessageComponent { export interface Quote extends MessageComponent {
type: 'Quote'; type: 'Quote';
id?: number; id?: number | string;
group_id?: number | string; group_id?: number | string;
sender_id?: number | string; sender_id?: number | string;
target_id?: number | string; target_id?: number | string;
@@ -3,12 +3,13 @@
* 用于管理WebSocket连接和消息处理 * 用于管理WebSocket连接和消息处理
*/ */
import { getActiveWorkspaceUuid } from '@/app/infra/http/workspaceContext'; import { getActiveWorkspaceUuid } from '@/app/infra/http/workspaceContext';
import type { MessageChainComponent } from '@/app/infra/entities/message';
export interface WebSocketMessage { export interface WebSocketMessage {
id: number; id: number;
role: 'user' | 'assistant'; role: 'user' | 'assistant';
content: string; content: string;
message_chain: Array<{ type: string; text?: string; target?: string }>; message_chain: MessageChainComponent[];
timestamp: string; timestamp: string;
is_final?: boolean; is_final?: boolean;
connection_id?: string; connection_id?: string;
@@ -16,7 +17,12 @@ export interface WebSocketMessage {
export interface WebSocketResponse { export interface WebSocketResponse {
type: type:
'connected' | 'response' | 'user_message' | 'pong' | 'broadcast' | 'error'; | 'connected'
| 'response'
| 'user_message'
| 'pong'
| 'broadcast'
| 'error';
connection_id?: string; connection_id?: string;
pipeline_uuid?: string; pipeline_uuid?: string;
session_type?: string; session_type?: string;
@@ -262,7 +268,7 @@ export class WebSocketClient {
* 发送消息 * 发送消息
*/ */
public sendMessage( public sendMessage(
messageChain: Array<{ type: string; text?: string; target?: string }>, messageChain: MessageChainComponent[],
stream: boolean = true, stream: boolean = true,
) { ) {
if (!this.ws || this.ws.readyState !== WebSocket.OPEN) { if (!this.ws || this.ws.readyState !== WebSocket.OPEN) {
@@ -226,6 +226,31 @@ test.describe('processor detail workbench', () => {
page, page,
}) => { }) => {
await installLangBotApiMocks(page, { authenticated: true }); await installLangBotApiMocks(page, { authenticated: true });
const debugImageKey =
'v1/mock-instance/workspace-default/1/upload_image/mock-owner/debug-image.png';
const debugImageBytes = Buffer.from(
'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAusB9Y9Zl1sAAAAASUVORK5CYII=',
'base64',
);
await page.route('**/api/v1/files/images', async (route) => {
await route.fulfill({
status: 200,
contentType: 'application/json',
body: JSON.stringify({
code: 0,
message: 'ok',
data: { file_key: debugImageKey },
timestamp: Date.now(),
}),
});
});
await page.route('**/api/v1/files/image/**', async (route) => {
await route.fulfill({
status: 200,
contentType: 'image/png',
body: debugImageBytes,
});
});
await page.routeWebSocket('**/api/v1/pipelines/**/ws/connect**', (ws) => { await page.routeWebSocket('**/api/v1/pipelines/**/ws/connect**', (ws) => {
ws.onMessage((raw) => { ws.onMessage((raw) => {
const message = JSON.parse(String(raw)); const message = JSON.parse(String(raw));
@@ -238,6 +263,21 @@ test.describe('processor detail workbench', () => {
session_type: 'person', session_type: 'person',
}), }),
); );
} else if (message.type === 'message') {
ws.send(
JSON.stringify({
type: 'user_message',
session_type: 'person',
data: {
id: 1,
role: 'user',
content: 'Describe this image',
message_chain: message.message,
timestamp: new Date().toISOString(),
is_final: true,
},
}),
);
} }
}); });
}); });
@@ -303,6 +343,21 @@ test.describe('processor detail workbench', () => {
expect(Math.abs(sendBox!.y - inputBox!.y)).toBeLessThanOrEqual(1); expect(Math.abs(sendBox!.y - inputBox!.y)).toBeLessThanOrEqual(1);
expect(toolbarBox!.y).toBeLessThan(emptyStateBox!.y); expect(toolbarBox!.y).toBeLessThan(emptyStateBox!.y);
await composer.locator('input[type="file"]').setInputFiles({
name: 'debug-image.png',
mimeType: 'image/png',
buffer: debugImageBytes,
});
await expect(
debugPanel.locator('[data-debug-chat-attachment-preview="true"]'),
).toBeVisible();
await messageInput.fill('Describe this image');
await sendButton.click();
await expect(
debugPanel.locator('[data-debug-chat-message-image="true"]'),
).toBeVisible();
await expect(debugPanel.getByText('Describe this image')).toBeVisible();
const streamSwitchBox = await composer.getByRole('switch').boundingBox(); const streamSwitchBox = await composer.getByRole('switch').boundingBox();
const resetBox = await resetButton.boundingBox(); const resetBox = await resetButton.boundingBox();
expect(streamSwitchBox).not.toBeNull(); expect(streamSwitchBox).not.toBeNull();