mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-27 05:30:39 +00:00
fix(channels): release standalone DingTalk upload sandboxes (#7203)
Refs bytedance/deer-flow#3296. Co-authored-by: Jun516 <174629376+Jun516@users.noreply.github.com>
This commit is contained in:
parent
de7307e6b3
commit
8b68832f32
@ -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
|
||||
|
||||
@ -446,6 +446,8 @@ DeerFlow 还可以在 workspace UI 里暴露用户自有的 IM 渠道连接。
|
||||
| QQ | WebSocket(仅文本私聊及群 @;每条来源消息最多回复 4 / 5 条) | 中等 |
|
||||
| Buzz | Nostr relay(WebSocket,NIP-42) | 中等 |
|
||||
|
||||
飞书和钉钉在附件同步完成后,会将独立的临时沙箱归还资源池,不会中断仍在运行的任务。
|
||||
|
||||
**`config.yaml` 中的配置示例:**
|
||||
|
||||
```yaml
|
||||
|
||||
@ -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`).
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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):
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user