From e69a80f5e97b26b00fc72ecb29c6b81e87b4c230 Mon Sep 17 00:00:00 2001 From: Neos <34212397+iloveaimer@users.noreply.github.com> Date: Sat, 29 Aug 2026 22:34:10 +0800 Subject: [PATCH] =?UTF-8?q?Windows=20=E5=85=BC=E5=AE=B9=E6=80=A7=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=20+=20Dify=20API=20=E9=80=82=E9=85=8D=20+=20=E4=BC=81?= =?UTF-8?q?=E5=BE=AE=20WS=20=E5=8A=A0=E8=BD=BD=E4=BC=98=E5=8C=96=20(#2470)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: Windows fsync requires write access to file descriptor - _fsync_file: os.O_RDONLY -> os.O_RDWR (Windows _commit() requires write access) - _fsync_directory: tolerate OSError on fsync and os.open (chmod 0o700 blocks access on Windows) * fix: Dify SSE empty data lines and upload response format mismatch - Skip malformed/flushed SSE data lines (JSONDecodeError) - Accept HTTP 200 besides 201 for file upload response - Unwrap data wrapper in upload response (resp.get('data', resp)) * feat: WeCom WS mode sends empty initial stream frame for loading spinner Send an empty reply_stream frame immediately after stream session creation so the WeCom client shows its built-in loading indicator while the pipeline processes the message (e.g. RAG retrieval). * fix(compat): harden Windows, Dify, and WeCom changes * fix(migration): clean interrupted SQLite temp files --------- Co-authored-by: Hyu --- .../libs/dify_service_api/v1/client.py | 49 ++++++++--- .../libs/wecom_ai_bot_api/ws_client.py | 7 ++ .../persistence/sqlite_migration_backup.py | 61 ++++++++++---- .../test_sqlite_migration_backup.py | 66 +++++++++++++++ .../platform/test_wecombot_template_card.py | 80 ++++++++++++++++++ .../provider/runners/test_difysvapi_runner.py | 82 +++++++++++++++++++ 6 files changed, 320 insertions(+), 25 deletions(-) diff --git a/src/langbot/libs/dify_service_api/v1/client.py b/src/langbot/libs/dify_service_api/v1/client.py index d1317e00e..4cd03a503 100644 --- a/src/langbot/libs/dify_service_api/v1/client.py +++ b/src/langbot/libs/dify_service_api/v1/client.py @@ -1,13 +1,14 @@ from __future__ import annotations import asyncio -import httpx -import typing import json +import os +import typing +from pathlib import Path + +import httpx from .errors import DifyAPIError -from pathlib import Path -import os _MAX_DIFY_RESPONSE_BYTES = 1024 * 1024 _MAX_DIFY_SSE_LINE_BYTES = 1024 * 1024 @@ -15,6 +16,32 @@ _MAX_DIFY_STREAM_BYTES = 16 * 1024 * 1024 _MAX_DIFY_UPLOAD_BYTES = 10 * 1024 * 1024 +def _decode_sse_data(line: bytes) -> dict[str, typing.Any] | None: + data = line[5:].strip() + if not data or data == b'[DONE]': + return None + try: + payload = json.loads(data.decode('utf-8')) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + raise DifyAPIError('Dify SSE data line is not valid JSON') from exc + if not isinstance(payload, dict): + raise DifyAPIError('Dify SSE event is not a JSON object') + return payload + + +def _decode_upload_response(body: bytes) -> dict[str, typing.Any]: + try: + response = json.loads(body) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + raise DifyAPIError('Dify upload response is not valid JSON') from exc + if not isinstance(response, dict): + raise DifyAPIError('Dify upload response is not a JSON object') + payload = response.get('data', response) + if not isinstance(payload, dict) or not isinstance(payload.get('id'), str) or not payload['id']: + raise DifyAPIError('Dify upload response does not contain a valid file id') + return payload + + async def _read_limited_response( response: httpx.Response, *, @@ -56,16 +83,16 @@ async def _iter_sse_json( line = raw_line.rstrip(b'\r').strip() if not line or not line.startswith(b'data:'): continue - payload = json.loads(line[5:].decode('utf-8', errors='replace')) - if isinstance(payload, dict): + payload = _decode_sse_data(line) + if payload is not None: yield payload if len(buffer) > _MAX_DIFY_SSE_LINE_BYTES: raise DifyAPIError('Dify SSE event exceeds the runtime limit') line = bytes(buffer).rstrip(b'\r').strip() if line.startswith(b'data:'): - payload = json.loads(line[5:].decode('utf-8', errors='replace')) - if isinstance(payload, dict): + payload = _decode_sse_data(line) + if payload is not None: yield payload @@ -242,7 +269,7 @@ class AsyncDifyServiceClient: file: httpx._types.FileTypes, user: str, timeout: float = 30.0, - ) -> str: + ) -> dict[str, typing.Any]: # 处理 Path 对象 if isinstance(file, Path): if not file.exists(): @@ -271,6 +298,6 @@ class AsyncDifyServiceClient: timeout=timeout, ) as response: body = await _read_limited_response(response) - if response.status_code != 201: + if response.status_code not in (200, 201): raise DifyAPIError(f'{response.status_code} {body.decode(errors="replace")}') - return json.loads(body) + return _decode_upload_response(body) diff --git a/src/langbot/libs/wecom_ai_bot_api/ws_client.py b/src/langbot/libs/wecom_ai_bot_api/ws_client.py index 997338211..47ffcae86 100644 --- a/src/langbot/libs/wecom_ai_bot_api/ws_client.py +++ b/src/langbot/libs/wecom_ai_bot_api/ws_client.py @@ -936,6 +936,13 @@ class WecomBotWsClient: 'chat_type': message_data.get('type', 'single'), } self._prune_stream_state() + # Send an initial empty stream frame so the WeCom client + # shows its built-in loading spinner while the pipeline + # processes the message (e.g. RAG retrieval). + try: + await self.reply_stream(req_id, stream_id, '', finish=False) + except Exception: + await self.logger.warning(f'Failed to send initial stream frame: {traceback.format_exc()}') message_data['stream_id'] = stream_id message_data['req_id'] = req_id diff --git a/src/langbot/pkg/persistence/sqlite_migration_backup.py b/src/langbot/pkg/persistence/sqlite_migration_backup.py index 9004b7fcc..913b439a6 100644 --- a/src/langbot/pkg/persistence/sqlite_migration_backup.py +++ b/src/langbot/pkg/persistence/sqlite_migration_backup.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import contextlib import dataclasses import datetime import json @@ -82,7 +83,7 @@ def _verify_connection(connection: sqlite3.Connection, expected_revision: str) - def _verify_file(path: pathlib.Path, expected_revision: str) -> None: - with _open_read_only(path) as connection: + with contextlib.closing(_open_read_only(path)) as connection: _verify_connection(connection, expected_revision) @@ -119,12 +120,16 @@ def _write_manifest(backup: SQLiteMigrationBackup, status: str, **extra: typing. def _fsync_file(path: pathlib.Path, *, reopen_attempts: int = 20) -> None: - """Sync a file, tolerating delayed visibility after replace on bind mounts.""" + """Sync a file, tolerating delayed visibility after replace on bind mounts. + + Uses O_RDWR so os.fsync works on Windows (where _commit requires write + access to the file descriptor). + """ descriptor: int | None = None for attempt in range(reopen_attempts): try: - descriptor = os.open(path, os.O_RDONLY) + descriptor = os.open(path, os.O_RDWR) break except FileNotFoundError: if attempt + 1 >= reopen_attempts: @@ -138,13 +143,37 @@ def _fsync_file(path: pathlib.Path, *, reopen_attempts: int = 20) -> None: def _fsync_directory(path: pathlib.Path) -> None: - descriptor = os.open(path, os.O_RDONLY) + if os.name == 'nt': + # Windows cannot fsync directory handles opened through os.open. + return + descriptor = os.open(path, os.O_RDONLY | getattr(os, 'O_DIRECTORY', 0)) try: os.fsync(descriptor) finally: os.close(descriptor) +def _remove_stale_temporary_files( + directory: pathlib.Path, + *, + prefix: str, + suffix: str, +) -> None: + """Remove temporary files left by an interrupted backup or restore.""" + + for candidate in directory.iterdir(): + if candidate.is_dir() or not candidate.name.startswith(prefix) or not candidate.name.endswith(suffix): + continue + try: + candidate.unlink() + except FileNotFoundError: + continue + except PermissionError: + # Another process may still own this file. Do not turn harmless + # cleanup into a migration failure; its unique name cannot collide. + continue + + def _create_backup( database_path: pathlib.Path, source_revision: str, @@ -153,6 +182,11 @@ def _create_backup( backup_directory = database_path.parent / 'migration-backups' backup_directory.mkdir(mode=0o700, parents=True, exist_ok=True) os.chmod(backup_directory, 0o700) + _remove_stale_temporary_files( + backup_directory, + prefix=f'.{database_path.stem}-pre-', + suffix='.creating', + ) created_at = datetime.datetime.now(datetime.UTC).strftime('%Y-%m-%dT%H-%M-%S.%fZ') stem = ( f'{database_path.stem}-pre-{_safe_label(target_revision)}-' @@ -169,11 +203,8 @@ def _create_backup( temporary_path = pathlib.Path(temporary_name) try: with ( - _open_read_only(database_path) as source, - sqlite3.connect( - temporary_path, - timeout=30, - ) as destination, + contextlib.closing(_open_read_only(database_path)) as source, + contextlib.closing(sqlite3.connect(temporary_path, timeout=30)) as destination, ): source.execute('PRAGMA busy_timeout = 30000') source.backup(destination) @@ -221,6 +252,11 @@ async def create_verified_backup( def _restore_backup(backup: SQLiteMigrationBackup) -> None: _verify_file(backup.backup_path, backup.source_revision) + _remove_stale_temporary_files( + backup.database_path.parent, + prefix=f'.{backup.database_path.name}.', + suffix='.restoring', + ) descriptor, temporary_name = tempfile.mkstemp( prefix=f'.{backup.database_path.name}.', suffix='.restoring', @@ -230,11 +266,8 @@ def _restore_backup(backup: SQLiteMigrationBackup) -> None: temporary_path = pathlib.Path(temporary_name) try: with ( - _open_read_only(backup.backup_path) as source, - sqlite3.connect( - temporary_path, - timeout=30, - ) as destination, + contextlib.closing(_open_read_only(backup.backup_path)) as source, + contextlib.closing(sqlite3.connect(temporary_path, timeout=30)) as destination, ): source.backup(destination) destination.commit() diff --git a/tests/integration/persistence/test_sqlite_migration_backup.py b/tests/integration/persistence/test_sqlite_migration_backup.py index e808cb3fa..89fffdc89 100644 --- a/tests/integration/persistence/test_sqlite_migration_backup.py +++ b/tests/integration/persistence/test_sqlite_migration_backup.py @@ -39,6 +39,35 @@ def _assert_verified_backup(payload: dict) -> None: assert connection.execute('SELECT version_num FROM alembic_version').fetchone()[0] == payload['source_revision'] +def _temporary_sqlite_files(root: pathlib.Path) -> list[pathlib.Path]: + return [*root.rglob('*.creating'), *root.rglob('*.restoring')] + + +async def test_backup_removes_stale_temporary_file_from_interrupted_run(tmp_path): + database_path = tmp_path / 'legacy-stale-backup.db' + engine = create_async_engine(f'sqlite+aiosqlite:///{database_path}') + try: + await create_legacy_resource_schema(engine, instance_uuid='stale-backup') + await alembic_runner.run_alembic_stamp(engine, '0008_mcp_resource_prefs') + backup_directory = tmp_path / 'migration-backups' + backup_directory.mkdir() + stale_path = backup_directory / '.legacy-stale-backup-pre-0009-old.creating' + unrelated_path = backup_directory / '.another-database-pre-0009-old.creating' + stale_path.write_bytes(b'interrupted backup') + unrelated_path.write_bytes(b'unrelated backup') + + await sqlite_migration_backup.create_verified_backup( + engine, + source_revision='0008_mcp_resource_prefs', + target_revision='0009_workspace_tenancy', + ) + + assert not stale_path.exists() + assert unrelated_path.read_bytes() == b'unrelated backup' + finally: + await engine.dispose() + + async def test_tenancy_migrations_retain_verified_boundary_backups(tmp_path): database_path = tmp_path / 'legacy-with-backups.db' engine = create_async_engine(f'sqlite+aiosqlite:///{database_path}') @@ -59,6 +88,7 @@ async def test_tenancy_migrations_retain_verified_boundary_backups(tmp_path): } for payload in payloads: _assert_verified_backup(payload) + assert _temporary_sqlite_files(tmp_path) == [] finally: await engine.dispose() @@ -100,6 +130,7 @@ async def test_failed_tenancy_migration_restores_backup_and_revision( assert restored[0]['status'] == 'restored_after_failure' assert restored[0]['source_revision'] == '0009_workspace_tenancy' _assert_verified_backup(restored[0]) + assert _temporary_sqlite_files(tmp_path) == [] monkeypatch.setattr(alembic_runner, 'run_alembic_upgrade', real_upgrade) await _manager(engine)._run_alembic_migrations() @@ -108,6 +139,41 @@ async def test_failed_tenancy_migration_restores_backup_and_revision( await engine.dispose() +async def test_restore_publish_failure_preserves_current_database(tmp_path, monkeypatch): + database_path = tmp_path / 'restore-publish-failure.db' + engine = create_async_engine(f'sqlite+aiosqlite:///{database_path}') + try: + await create_legacy_resource_schema(engine, instance_uuid='restore-publish-failure') + await alembic_runner.run_alembic_stamp(engine, '0008_mcp_resource_prefs') + backup = await sqlite_migration_backup.create_verified_backup( + engine, + source_revision='0008_mcp_resource_prefs', + target_revision='0009_workspace_tenancy', + ) + stale_restore_path = tmp_path / f'.{database_path.name}.interrupted.restoring' + stale_restore_path.write_bytes(b'interrupted restore') + async with engine.begin() as connection: + await connection.execute(sa.text("UPDATE alembic_version SET version_num = 'failed-revision'")) + await engine.dispose() + database_before_restore = database_path.read_bytes() + real_replace = os.replace + + def fail_restore_publish(source, destination): + if pathlib.Path(destination) == database_path: + raise OSError('simulated atomic publish failure') + return real_replace(source, destination) + + monkeypatch.setattr(sqlite_migration_backup.os, 'replace', fail_restore_publish) + + with pytest.raises(OSError, match='atomic publish failure'): + await sqlite_migration_backup.restore_verified_backup(engine, backup) + + assert database_path.read_bytes() == database_before_restore + assert _temporary_sqlite_files(tmp_path) == [] + finally: + await engine.dispose() + + async def test_backup_retries_transient_reopen_failure_after_replace(tmp_path, monkeypatch): database_path = tmp_path / 'legacy-bind-mount.db' engine = create_async_engine(f'sqlite+aiosqlite:///{database_path}') diff --git a/tests/unit_tests/platform/test_wecombot_template_card.py b/tests/unit_tests/platform/test_wecombot_template_card.py index 03d8791b1..d8b46c538 100644 --- a/tests/unit_tests/platform/test_wecombot_template_card.py +++ b/tests/unit_tests/platform/test_wecombot_template_card.py @@ -44,6 +44,86 @@ def test_webhook_dispatch_tasks_are_bounded(): assert len(client._dispatch_tasks) == 100 +@pytest.mark.asyncio +async def test_ws_initial_stream_frame_precedes_pipeline_dispatch(monkeypatch): + from langbot.libs.wecom_ai_bot_api import ws_client as ws_client_module + + order = [] + logger = types.SimpleNamespace( + debug=Mock(), + error=Mock(), + warning=Mock(), + ) + client = WecomBotWsClient('bot-id', 'secret', logger) + + async def parse_message(*args, **kwargs): + del args, kwargs + return {'msgid': 'msg-1', 'type': 'single', 'userid': 'user-1'} + + async def reply_stream(*args, **kwargs): + del args, kwargs + order.append('initial-frame') + return {} + + async def dispatch_event(event): + del event + order.append('pipeline-dispatch') + + monkeypatch.setattr(ws_client_module, 'parse_wecom_bot_message', parse_message) + monkeypatch.setattr(ws_client_module.wecombotevent, 'WecomBotEvent', lambda data: data) + client.reply_stream = reply_stream + client._dispatch_event = dispatch_event + + await client._handle_message_callback({'headers': {'req_id': 'req-1'}, 'body': {}}) + + assert order == ['initial-frame', 'pipeline-dispatch'] + + +@pytest.mark.asyncio +async def test_ws_initial_stream_failure_still_dispatches_message(monkeypatch): + from langbot.libs.wecom_ai_bot_api import ws_client as ws_client_module + + dispatched = [] + + class Logger: + def __init__(self): + self.warnings = [] + + async def debug(self, message): + del message + + async def error(self, message): + raise AssertionError(message) + + async def warning(self, message): + self.warnings.append(message) + + logger = Logger() + client = WecomBotWsClient('bot-id', 'secret', logger) + + async def parse_message(*args, **kwargs): + del args, kwargs + return {'msgid': 'msg-1', 'type': 'single', 'userid': 'user-1'} + + async def reply_stream(*args, **kwargs): + del args, kwargs + raise ConnectionError('simulated reply failure') + + async def dispatch_event(event): + dispatched.append(event) + + monkeypatch.setattr(ws_client_module, 'parse_wecom_bot_message', parse_message) + monkeypatch.setattr(ws_client_module.wecombotevent, 'WecomBotEvent', lambda data: data) + client.reply_stream = reply_stream + client._dispatch_event = dispatch_event + + await client._handle_message_callback({'headers': {'req_id': 'req-1'}, 'body': {}}) + + assert len(dispatched) == 1 + assert len(logger.warnings) == 1 + assert 'simulated reply failure' in logger.warnings[0] + + def test_extract_template_card_action_supports_nested_button_key(): task_id, event_key, card_type = extract_template_card_action( { diff --git a/tests/unit_tests/provider/runners/test_difysvapi_runner.py b/tests/unit_tests/provider/runners/test_difysvapi_runner.py index 028d278be..aab156697 100644 --- a/tests/unit_tests/provider/runners/test_difysvapi_runner.py +++ b/tests/unit_tests/provider/runners/test_difysvapi_runner.py @@ -79,6 +79,32 @@ class TestDifyWorkflowSubmitClient: with pytest.raises(errors.DifyAPIError, match='SSE event exceeds'): await anext(client._iter_sse_json(FakeResponse())) + @pytest.mark.asyncio + async def test_sse_parser_skips_empty_data_and_done_lines(self): + from langbot.libs.dify_service_api.v1 import client + + class FakeResponse: + async def aiter_bytes(self, chunk_size=None): + del chunk_size + yield b'data:\n\ndata: {"event":"message",' + yield b'"answer":"ok"}\n\ndata: [DONE]\n' + + events = [event async for event in client._iter_sse_json(FakeResponse())] + + assert events == [{'event': 'message', 'answer': 'ok'}] + + @pytest.mark.asyncio + async def test_sse_parser_rejects_malformed_nonempty_data(self): + from langbot.libs.dify_service_api.v1 import client, errors + + class FakeResponse: + async def aiter_bytes(self, chunk_size=None): + del chunk_size + yield b'data: not-json\n' + + with pytest.raises(errors.DifyAPIError, match='not valid JSON'): + await anext(client._iter_sse_json(FakeResponse())) + @pytest.mark.asyncio async def test_upload_rejects_oversized_local_file(self, tmp_path): from langbot.libs.dify_service_api.v1 import client @@ -93,6 +119,62 @@ class TestDifyWorkflowSubmitClient: with pytest.raises(ValueError, match='exceeds the size limit'): await dify_client.upload_file(file_path, 'person_user-1') + @pytest.mark.asyncio + @pytest.mark.parametrize( + ('status_code', 'body', 'expected_id'), + [ + (200, b'{"data":{"id":"wrapped-id"}}', 'wrapped-id'), + (201, b'{"id":"flat-id"}', 'flat-id'), + ], + ) + async def test_upload_accepts_supported_success_responses(self, status_code, body, expected_id): + from langbot.libs.dify_service_api.v1 import client + + class FakeResponse: + headers = {} + + def __init__(self): + self.status_code = status_code + + async def aiter_bytes(self, chunk_size=None): + del chunk_size + yield body + + class FakeStreamContext: + async def __aenter__(self): + return FakeResponse() + + async def __aexit__(self, exc_type, exc, traceback): + del exc_type, exc, traceback + return False + + class FakeClient: + def stream(self, *args, **kwargs): + del args, kwargs + return FakeStreamContext() + + dify_client = client.AsyncDifyServiceClient('test-key', 'https://dify.example/v1') + dify_client._client = FakeClient() + + response = await dify_client.upload_file(('hello.txt', b'hello', 'text/plain'), 'person_user-1') + + assert response['id'] == expected_id + + @pytest.mark.parametrize( + 'body', + [ + b'not-json', + b'[]', + b'{"data":null}', + b'{"data":{}}', + ], + ) + def test_upload_rejects_invalid_success_payload(self, body): + from langbot.libs.dify_service_api.v1 import client, errors + + with pytest.raises(errors.DifyAPIError): + client._decode_upload_response(body) + class TestDifyExtractTextOutput: """Tests for _extract_dify_text_output method."""