diff --git a/CHANGELOG.md b/CHANGELOG.md index 5f6cbbab70..d1ad3499b2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1402,6 +1402,12 @@ As of 2026-10-21, the milestone includes **829 merged pull requests**. ### Fixed +- **channels:** IM replies that present files no longer resolve those files + on the event loop. For every artifact, both reply paths (`runs.wait` and + streaming) resolved the virtual path (`realpath`), resolved the outputs + root and checked the file with `stat` inline, stalling other channels and + requests on network-backed storage. They now do it in a worker. ([#7594]) + - **channels:** A streamed Feishu reply no longer rewrites the same thread bindings on every chunk. Each outbound to the running card stored all of the reply's aliases again (about five `set_thread_id` writes per chunk: a @@ -13196,3 +13202,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#7588]: https://github.com/bytedance/deer-flow/pull/7588 [#7591]: https://github.com/bytedance/deer-flow/pull/7591 [#7593]: https://github.com/bytedance/deer-flow/pull/7593 +[#7594]: https://github.com/bytedance/deer-flow/pull/7594 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 8ce3376e8c..69084adcbb 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -1116,6 +1116,10 @@ ### 修复 +- **渠道:** 展示文件的 IM 回复不再在事件循环上解析这些文件。此前两条回复路径(`runs.wait` + 和流式)都会对每个产物内联解析虚拟路径(`realpath`)、解析输出根目录并用 `stat` 检查文件, + 在网络存储上会让其他渠道和请求停顿。现在这些工作在工作线程中进行。([#7594]) + - **渠道:** 飞书流式回复不再在每个分块上重写相同的线程绑定。此前每次向运行中卡片发送消息都会 重新保存该回复的全部别名(每个分块约五次 `set_thread_id`:每次都是在存储锁下完整重写 `store.json`,或一次数据库 upsert),而实际只有 `updated_at` 变化。现在非最终分块对每个 @@ -10519,9 +10523,9 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子 [#7530]: https://github.com/bytedance/deer-flow/pull/7530 [#7557]: https://github.com/bytedance/deer-flow/pull/7557 [#7560]: https://github.com/bytedance/deer-flow/pull/7560 -[#7580]: https://github.com/bytedance/deer-flow/pull/7580 [#7576]: https://github.com/bytedance/deer-flow/pull/7576 [#7580]: https://github.com/bytedance/deer-flow/pull/7580 [#7588]: https://github.com/bytedance/deer-flow/pull/7588 [#7591]: https://github.com/bytedance/deer-flow/pull/7591 [#7593]: https://github.com/bytedance/deer-flow/pull/7593 +[#7594]: https://github.com/bytedance/deer-flow/pull/7594 diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 6ecd64dfc5..b15c773019 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -60,7 +60,7 @@ The `channel_agent_name` cache reloads after restart before a resumed turn route - `FeishuChannel._receive_single_file(...)` / `DingTalkChannel._receive_single_file(...)` — normalize filenames and claim/write through `write_upload_file_no_symlink(..., exclusive=True)`, retrying `FileExistsError` under the channel lock; the final basename drives agent-visible paths and sandbox sync - `sandbox_files.py` — non-mounted Feishu/DingTalk syncs acquire unique execution holders, drain blocking `update_file` workers across repeated cancellation, and release only after the last sandbox operation, so a parallel run cannot close the shared client mid-upload. Both channels request provider release after the final holder exits, parking standalone temporary sandboxes. - `_ingest_inbound_files(...)` and the underlying `ensure_uploads_dir` / `get_uploads_dir` — owner-scoped via the same kwarg -- `_resolve_attachments` / `_prepare_artifact_delivery` — resolve output artifacts from the bound owner's bucket through `app.gateway.path_utils.resolve_outputs_confined_path`, the same outputs-only rule the artifact editor uses, so a sibling `uploads/`/`workspace/` path or a symlink planted in `outputs/` is skipped with a warning +- `_resolve_attachments` / `_prepare_artifact_delivery` — resolve output artifacts from the bound owner's bucket through `app.gateway.path_utils.resolve_outputs_confined_path`, the same outputs-only rule the artifact editor uses, so a sibling `uploads/`/`workspace/` path or a symlink planted in `outputs/` is skipped with a warning; both reply paths run it via `to_thread` (`blocking_io/test_channel_artifact_delivery.py`) The cached value is reused for both the blocking (`runs.wait`) and streaming (`_handle_streaming_chat`) paths, so uploads and artifact delivery always target the same bucket even if a channel returns a rewritten `InboundMessage` from `receive_file`. The bucket id matches the memory bucket resolved by `_resolve_memory_user_id` (both normalize through `make_safe_user_id`). **Configuration** (`config.yaml` -> `channels`): diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index 829e0d293c..cd1c7741cd 100644 --- a/backend/app/channels/manager.py +++ b/backend/app/channels/manager.py @@ -1416,7 +1416,11 @@ def _prepare_artifact_delivery( *, user_id: str | None = None, ) -> tuple[str, list[ResolvedAttachment]]: - """Resolve attachments and append filename fallbacks to the text response.""" + """Resolve attachments and append filename fallbacks to the text response. + + Blocking: every artifact is resolved through ``realpath`` and checked with + ``stat``, so the async reply paths call this through ``asyncio.to_thread``. + """ attachments: list[ResolvedAttachment] = [] if not artifacts: return response_text, attachments @@ -3472,7 +3476,7 @@ class ChannelManager: # Reuse the storage owner cached at the top of _handle_chat so uploads and # artifact delivery always resolve to the same bucket, even if a future # channel.receive_file returns a rewritten InboundMessage. - response_text, attachments = _prepare_artifact_delivery(thread_id, response_text, artifacts, user_id=storage_user_id) + response_text, attachments = await asyncio.to_thread(_prepare_artifact_delivery, thread_id, response_text, artifacts, user_id=storage_user_id) if not response_text: if attachments: @@ -3646,7 +3650,7 @@ class ChannelManager: # Reuse the storage owner resolved by _handle_chat so artifact delivery # matches the upload bucket and we avoid re-running _safe_user_id_for_run # (and its possible filesystem touch) on the streaming-error path. - response_text, attachments = _prepare_artifact_delivery(thread_id, response_text, artifacts, user_id=storage_user_id) + response_text, attachments = await asyncio.to_thread(_prepare_artifact_delivery, thread_id, response_text, artifacts, user_id=storage_user_id) if step_limit_text: # The last text is mid-task narration: say the run stopped, in # place of the generic failure, so the user knows to continue. diff --git a/backend/tests/blocking_io/test_channel_artifact_delivery.py b/backend/tests/blocking_io/test_channel_artifact_delivery.py new file mode 100644 index 0000000000..180fa6a646 --- /dev/null +++ b/backend/tests/blocking_io/test_channel_artifact_delivery.py @@ -0,0 +1,122 @@ +"""Regression anchors: resolving reply artifacts must not block the event loop. + +When a run presents files, ``ChannelManager`` turns each virtual +``/mnt/user-data/outputs/...`` path into a host path before delivering the +reply: ``resolve_outputs_confined_path`` resolves it (``realpath``) and checks it +against the resolved outputs root, then the file is checked with ``is_file`` and +``stat``. Both reply paths (``runs.wait`` and ``runs.stream``) run that in a +worker through ``asyncio.to_thread``. + +Each case drives the real manager with a real thread outputs directory under a +temporary ``DEER_FLOW_HOME`` and asserts the outbound reply carries the +resolved attachment, so an anchor cannot pass because the artifact was never +resolved. +""" + +from __future__ import annotations + +import asyncio +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest +import pytest_asyncio +from blockbuster import BlockingError + +from app.channels import manager as mgr +from app.channels.message_bus import InboundMessage, MessageBus, OutboundMessage +from app.channels.store import JsonChannelStore +from deerflow.config import paths as paths_mod + +pytestmark = pytest.mark.asyncio + +_THREAD_ID = "thread-artifacts" +_ARTIFACT = "/mnt/user-data/outputs/report.txt" + + +def _final_state() -> dict: + return { + "messages": [ + {"type": "human", "content": "make a report"}, + {"type": "ai", "content": "", "tool_calls": [{"name": "present_files", "args": {"filepaths": [_ARTIFACT]}}]}, + {"type": "tool", "name": "present_files", "content": "ok"}, + {"type": "ai", "content": "Here is the report."}, + ], + "artifacts": [_ARTIFACT], + } + + +def _write_artifact(user_id: str) -> None: + outputs = paths_mod.get_paths().sandbox_outputs_dir(_THREAD_ID, user_id=user_id) + outputs.mkdir(parents=True, exist_ok=True) + (outputs / "report.txt").write_text("report", encoding="utf-8") + + +@pytest_asyncio.fixture +async def home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path)) + monkeypatch.setattr(paths_mod, "_paths", None) + storage_user_id = await asyncio.to_thread(mgr._safe_user_id_for_run, "user-1") + await asyncio.to_thread(_write_artifact, storage_user_id) + return tmp_path + + +async def _deliver(home: Path, channel_name: str, client: MagicMock) -> OutboundMessage: + bus = MessageBus() + manager = mgr.ChannelManager(bus=bus, store=JsonChannelStore(path=home / "store.json")) + manager._client = client + finals: list[OutboundMessage] = [] + + async def capture(msg: OutboundMessage) -> None: + if msg.is_final: + finals.append(msg) + + bus.subscribe_outbound(capture) + await manager.start() + try: + await bus.publish_inbound(InboundMessage(channel_name=channel_name, chat_id="chat-1", user_id="user-1", text="make a report", thread_ts="source-1")) + async with asyncio.timeout(5): + while not finals: + await asyncio.sleep(0.01) + finally: + await manager.stop() + return finals[0] + + +def _client() -> MagicMock: + client = MagicMock() + client.threads.create = AsyncMock(return_value={"thread_id": _THREAD_ID}) + client.threads.get = AsyncMock(return_value={"thread_id": _THREAD_ID}) + client.threads.update = AsyncMock(return_value={"thread_id": _THREAD_ID}) + return client + + +async def test_wait_reply_resolves_artifacts_off_the_loop(home: Path) -> None: + client = _client() + client.runs.wait = AsyncMock(return_value=_final_state()) + + reply = await _deliver(home, "slack", client) + + assert [attachment.filename for attachment in reply.attachments] == ["report.txt"] + assert reply.attachments[0].size == len("report") + + +async def test_streamed_reply_resolves_artifacts_off_the_loop(home: Path) -> None: + async def stream(*_args, **_kwargs): + yield SimpleNamespace(event="values", data=_final_state()) + + client = _client() + client.runs.stream = MagicMock(side_effect=stream) + + reply = await _deliver(home, "telegram", client) + + assert [attachment.filename for attachment in reply.attachments] == ["report.txt"] + assert reply.attachments[0].size == len("report") + + +async def test_blocking_probe_inline_artifact_delivery_actually_trips_the_gate(home: Path) -> None: + """Meta-check: the same resolution inline on the loop is caught by the gate.""" + storage_user_id = await asyncio.to_thread(mgr._safe_user_id_for_run, "user-1") + with pytest.raises(BlockingError): + mgr._prepare_artifact_delivery(_THREAD_ID, "Here is the report.", [_ARTIFACT], user_id=storage_user_id)