diff --git a/app/startup/composition/system.py b/app/startup/composition/system.py index 4bf277cbb..b8d3f4353 100644 --- a/app/startup/composition/system.py +++ b/app/startup/composition/system.py @@ -121,15 +121,27 @@ class _FileLogAdapter(SystemLogPort): async with aiofiles.open(path, mode="r", encoding="utf-8", errors="replace") as file: await file.seek(0, 2) previous_size = (await asyncio.to_thread(path.stat)).st_size + pending = "" while not runtime_stop_state.is_system_stopped: if await disconnected(): break current_size = (await asyncio.to_thread(path.stat)).st_size + if current_size < previous_size: + await file.seek(0) + previous_size = 0 + pending = "" if current_size > previous_size: - line = (await file.readline()).strip() - if line: - yield line - previous_size = current_size + pending += await file.read() + previous_size = (await asyncio.to_thread(path.stat)).st_size + fragments = pending.splitlines(keepends=True) + if fragments and not fragments[-1].endswith(("\n", "\r")): + pending = fragments.pop() + else: + pending = "" + for fragment in fragments: + line = fragment.strip() + if line: + yield line else: await asyncio.sleep(0.5) diff --git a/tests/test_agent_realtime_log.py b/tests/test_agent_realtime_log.py new file mode 100644 index 000000000..d6a830488 --- /dev/null +++ b/tests/test_agent_realtime_log.py @@ -0,0 +1,49 @@ +"""验证 Agent 多行日志在实时日志轮询中的传输完整性。""" + +import asyncio +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from app.startup.composition.system import _FileLogAdapter + + +@pytest.mark.asyncio +async def test_file_log_follow_drains_all_lines_from_one_append(tmp_path: Path) -> None: + """一次写入包含多行时,实时跟随器应逐行转发全部新增内容。""" + log_path = tmp_path / "moviepilot.log" + log_path.touch() + adapter = _FileLogAdapter(SimpleNamespace(get=lambda key: tmp_path)) + + async def is_disconnected() -> bool: + """保持测试连接,直到断言收集到全部新增日志。""" + return False + + stream = await adapter.follow("moviepilot.log", 50, is_disconnected) + collected: list[str] = [] + + async def collect_lines() -> None: + """收集同一次写入产生的三行日志。""" + async for line in stream: + collected.append(line) + if len(collected) == 3: + break + + collector = asyncio.create_task(collect_lines()) + await asyncio.sleep(0.05) + log_path.write_text( + "【INFO】 2026-09-11 11:39:36,300 callback - Agent消息: 首段\n" + "...\n" + "⚙️ => 读取技能说明:moviepilot-api\n", + encoding="utf-8", + ) + + await asyncio.wait_for(collector, timeout=2) + await stream.aclose() + + assert collected == [ + "【INFO】 2026-09-11 11:39:36,300 callback - Agent消息: 首段", + "...", + "⚙️ => 读取技能说明:moviepilot-api", + ]