diff --git a/README.md b/README.md index c00c3a5222..9a50312e3d 100644 --- a/README.md +++ b/README.md @@ -1051,6 +1051,9 @@ embedded downloads keep distinct filenames, including when concurrent uploads choose the same name. The final filename is passed to the agent and used for sandbox sync; an existing conversation file is not overwritten. +Feishu and DingTalk attachment transfers return standalone temporary sandboxes +to the pool after synchronization, without interrupting an active run. + **Configuration in `config.yaml`:** Discord's `channels.discord.allowed_guilds` accepts one positive numeric guild diff --git a/README_zh.md b/README_zh.md index 332163aaa1..7d3b51175e 100644 --- a/README_zh.md +++ b/README_zh.md @@ -446,6 +446,8 @@ DeerFlow 还可以在 workspace UI 里暴露用户自有的 IM 渠道连接。 | QQ | WebSocket（仅文本私聊及群 @；每条来源消息最多回复 4 / 5 条） | 中等 | | Buzz | Nostr relay（WebSocket，NIP-42） | 中等 | +飞书和钉钉在附件同步完成后，会将独立的临时沙箱归还资源池，不会中断仍在运行的任务。 + **`config.yaml` 中的配置示例：** ```yaml diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 9e555d9fd8..c6ee536740 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -52,7 +52,7 @@ The `channel_agent_name` cache reloads after restart before a resumed turn route **Owner-scoped file storage**: inbound files, uploads, and output artifacts are staged under the DeerFlow owner's bucket so they land where the agent run reads/writes (`users/{user_id}/threads/{thread_id}/user-data/{uploads,outputs}`). `ChannelManager._handle_chat` resolves the storage owner once via `_channel_storage_user_id(msg)` (sanitized owner id, falling back to `safe(msg.user_id)` for unbound auth-enabled channels — mirroring `_resolve_run_params`'s run identity; `None` only when no identity is available) and threads it as the `user_id=` kwarg through the file pipeline: - `Channel.receive_file(msg, thread_id, user_id=...)` — owner-bound channels persist downloaded files under the owner's bucket instead of the default bucket - `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. Feishu requests provider release after the final holder exits, parking standalone temporary sandboxes; DingTalk retains its non-releasing holder behavior. +- `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 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`). diff --git a/backend/app/channels/dingtalk.py b/backend/app/channels/dingtalk.py index 901af7108f..4acea4c793 100644 --- a/backend/app/channels/dingtalk.py +++ b/backend/app/channels/dingtalk.py @@ -881,6 +881,7 @@ class DingTalkChannel(Channel): virtual_path=virtual_path, content=content, owner_prefix="dingtalk-upload", + release_on_last=True, ) if not synced: # Mirror Feishu: the agent's non-local sandbox cannot see this diff --git a/backend/tests/test_dingtalk_channel.py b/backend/tests/test_dingtalk_channel.py index 24392c56d7..75b0b6e1f5 100644 --- a/backend/tests/test_dingtalk_channel.py +++ b/backend/tests/test_dingtalk_channel.py @@ -27,6 +27,7 @@ from app.channels.dingtalk import ( ) from app.channels.message_bus import InboundMessageType, MessageBus, OutboundMessage from deerflow.config.paths import VIRTUAL_PATH_PREFIX +from deerflow.sandbox.lease import discard_sandbox_lease_manager, get_sandbox_lease_manager def _run(coro): @@ -2101,6 +2102,29 @@ def _patch_uploads(monkeypatch, uploads_dir, *, sandbox_id="local", sandbox=None release=lambda sid: None, ) monkeypatch.setattr("app.channels.dingtalk.get_sandbox_provider", lambda: provider) + return provider + + +@pytest.fixture +def non_local_upload(tmp_path, monkeypatch): + channel = DingTalkChannel(MessageBus(), config={}) + channel._download_by_code = AsyncMock(return_value=b"BYTES") + uploads = tmp_path / "uploads" + uploads.mkdir() + sandbox = SimpleNamespace(update_file=MagicMock(), release_command_scope=MagicMock()) + provider = _patch_uploads(monkeypatch, uploads, sandbox_id="aio:box1", sandbox=sandbox) + provider.release = MagicMock() + message = channel._make_inbound( + chat_id="c", + user_id="u", + text="hi", + thread_ts="m", + files=[{"type": "file", "download_code": "dc", "filename": "a.pdf"}], + ) + try: + yield SimpleNamespace(channel=channel, provider=provider, sandbox=sandbox, message=message) + finally: + discard_sandbox_lease_manager(provider) class TestExtractFiles: @@ -2699,6 +2723,82 @@ class TestReceiveFile: _run(go()) + def test_non_local_upload_releases_standalone_sandbox(self, non_local_upload): + async def go(): + upload = non_local_upload + result = await asyncio.wait_for(upload.channel.receive_file(upload.message, "t1", user_id="default"), timeout=5) + + assert result.text == f"{VIRTUAL_PATH_PREFIX}/uploads/a.pdf\n\nhi" + upload.sandbox.update_file.assert_called_once_with(f"{VIRTUAL_PATH_PREFIX}/uploads/a.pdf", b"BYTES") + upload.sandbox.release_command_scope.assert_called_once() + upload.provider.release.assert_called_once_with("aio:box1") + + _run(go()) + + @pytest.mark.parametrize("run_finishes_first", [False, True]) + def test_non_local_upload_releases_only_after_upload_and_run_finish(self, non_local_upload, run_finishes_first): + async def go(): + upload = non_local_upload + manager = get_sandbox_lease_manager(upload.provider) + await manager.retain_async("active-run", "aio:box1", thread_id="t1", user_id="default") + started = asyncio.Event() + finish_update = threading.Event() + update_finished = threading.Event() + loop = asyncio.get_running_loop() + + def update_file(path, content): + loop.call_soon_threadsafe(started.set) + try: + if not finish_update.wait(timeout=10): + raise TimeoutError("test did not release sandbox update") + finally: + update_finished.set() + + upload.sandbox.update_file.side_effect = update_file + task = asyncio.create_task(upload.channel.receive_file(upload.message, "t1", user_id="default")) + try: + await asyncio.wait_for(started.wait(), timeout=5) + upload.provider.release.assert_not_called() + assert not update_finished.is_set() + + if run_finishes_first: + await manager.release_async("active-run") + upload.provider.release.assert_not_called() + assert not update_finished.is_set() + + finish_update.set() + result = await asyncio.wait_for(task, timeout=5) + assert result.text == f"{VIRTUAL_PATH_PREFIX}/uploads/a.pdf\n\nhi" + assert update_finished.is_set() + if not run_finishes_first: + upload.provider.release.assert_not_called() + assert manager.binding_for("active-run") == "aio:box1" + await manager.release_async("active-run") + + upload.provider.release.assert_called_once_with("aio:box1") + assert manager.binding_for("active-run") is None + assert upload.sandbox.release_command_scope.call_count == 2 + finally: + finish_update.set() + await asyncio.gather(task, return_exceptions=True) + await manager.release_async("active-run") + + _run(go()) + + def test_failed_non_local_upload_releases_standalone_sandbox(self, non_local_upload): + async def go(): + upload = non_local_upload + upload.sandbox.update_file.side_effect = RuntimeError("sandbox transport down") + + result = await asyncio.wait_for(upload.channel.receive_file(upload.message, "t1", user_id="default"), timeout=5) + + assert result.text == "[failed to load file: a.pdf]\n\nhi" + upload.sandbox.update_file.assert_called_once_with(f"{VIRTUAL_PATH_PREFIX}/uploads/a.pdf", b"BYTES") + upload.sandbox.release_command_scope.assert_called_once() + upload.provider.release.assert_called_once_with("aio:box1") + + _run(go()) + class TestDownloadByCode: def test_two_step_download(self):