diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index e72144ccf4..50ef8d4a09 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -42,7 +42,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk **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 provider filenames, claim a collision-free basename and write it through `write_upload_file_no_symlink` under the same channel lock; the returned basename drives both the agent-visible virtual path and non-local sandbox sync -- `sandbox_files.py` — non-mounted Feishu/DingTalk syncs acquire unique non-releasing 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 +- `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. - `_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/feishu.py b/backend/app/channels/feishu.py index de271341c9..cc77c6e943 100644 --- a/backend/app/channels/feishu.py +++ b/backend/app/channels/feishu.py @@ -503,6 +503,7 @@ class FeishuChannel(Channel): virtual_path=virtual_path, content=content, owner_prefix="feishu-upload", + release_on_last=True, ) if not synced: logger.warning("[Feishu] sandbox not found for thread_id=%s", thread_id) diff --git a/backend/app/channels/sandbox_files.py b/backend/app/channels/sandbox_files.py index eaa8a5e7ff..62c8e94f38 100644 --- a/backend/app/channels/sandbox_files.py +++ b/backend/app/channels/sandbox_files.py @@ -2,8 +2,12 @@ from __future__ import annotations +import logging + from deerflow.sandbox.lease import acquire_sandbox_client_lease +logger = logging.getLogger(__name__) + async def sync_file_to_thread_sandbox( sandbox_provider, @@ -13,14 +17,16 @@ async def sync_file_to_thread_sandbox( virtual_path: str, content: bytes, owner_prefix: str, + release_on_last: bool = False, ) -> bool: - """Copy one attachment while holding a non-releasing sandbox client lease. + """Copy one attachment while holding a sandbox client lease. Thread-data mount providers already see the persisted upload. Other providers need a unique holder so a parallel run cannot close their client during ``update_file``. The blocking transport worker is drained even when the channel handler is repeatedly cancelled, and only then is the holder - released. + released. ``release_on_last=True`` also parks a standalone temporary + sandbox once all execution holders have finished. """ if getattr(sandbox_provider, "uses_thread_data_mounts", False): return True @@ -30,7 +36,7 @@ async def sync_file_to_thread_sandbox( thread_id, user_id=user_id, owner_prefix=owner_prefix, - release_on_last=False, + release_on_last=release_on_last, ) try: if lease.sandbox_id == "local" or lease.sandbox_id.startswith("local:"): @@ -40,4 +46,7 @@ async def sync_file_to_thread_sandbox( await lease.run_sync(lease.sandbox.update_file, virtual_path, content) return True finally: - await lease.release() + try: + await lease.release() + except Exception: + logger.warning("Failed to release sandbox %s after file sync", lease.sandbox_id, exc_info=True) diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index 4f0af8f9dd..f00ab9e27f 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -105,7 +105,7 @@ owner-scoped assistant version selection remains enabled. | **Subagents** (`/api/subagents`) | Admin managed-worker CRUD and listing. | | **Integrations** (`/api/integrations`) | `GET /lark/status` - inspect managed Lark/Feishu CLI integration state, including `sandbox_runtime_mode` / `sandbox_runtime_ready` (whether `lark-cli` will actually be present in the sandbox at chat time); `POST /lark/install` - admin-only install of the official `lark-*` managed skill pack; `POST /lark/config/start` and `/lark/config/complete` - internal first-time Lark connection setup; `POST /lark/config/credentials` - atomically switch the caller's per-user Lark app after validating the new `app_id`/`app_secret` through the official CLI's live tenant-token probe, revoke/remove the previous OAuth tokens, and restore the prior credential tree if the switch fails; `POST /lark/auth/start` and `/lark/auth/complete` - browser device-flow user authorization without terminal access, with optional `domains` / exact `scope` for incremental permission grants. Config and auth flows carry a server-issued, per-user generation persisted under the credential lock; a rejected direct switch leaves the current generation unchanged, stale completions return 409, and browser re-registration uses the same token-clearing/revocation transaction as direct credential switches. | | **Memory** (`/api/memory`) | `GET /` - memory data; `POST /reload` - force reload; `GET /config` - config; `GET /status` - config + data | -| **Uploads** (`/api/threads/{id}/uploads`) | `POST /` - upload files (auto-converts PDF/PPT/Excel/Word); non-mounted sandbox sync uses a non-releasing request lease; `GET /list` - list; `DELETE /{filename}` - delete a regular file; symlinks 404 and a converted `.md` is kept (#5672) | +| **Uploads** (`/api/threads/{id}/uploads`) | `POST /` - upload files (auto-converts PDF/PPT/Excel/Word); non-mounted sync releases its lease after the last holder exits; `GET /list` - list; `DELETE /{filename}` - delete a regular file; symlinks 404 and a converted `.md` is kept (#5672) | | **Threads** (`/api/threads/{id}`) | `DELETE /` - remove DeerFlow-managed local thread data after LangGraph thread deletion; `POST /branches` - branch a completed assistant turn with a replay checkpoint; inherited titles take next-free displayed sibling suffixes, including explicit/renamed ones, while explicit titles stay unchanged. Durable `branch` admission rejects races. Workspace files are not checkpointed, so the branch only best-effort copies the current workspace when branching from the **latest** turn (`workspace_clone_mode="current_thread_best_effort"`); branching from an older/historical turn skips the copy (`workspace_clone_mode="skipped_historical_turn"`) so the branch never inherits files that only exist in a later timeline. Thread-scoped runtime channels (`sandbox`, `thread_data`) are not copied onto the branch: the parent's `sandbox_id` binds path mappings and the release lifecycle to the parent's workspace, so the branch lazily acquires its own sandbox instead. Branch creation also seeds the new thread's run-event feed from the branch checkpoint's visible messages (`history_seed_mode` in the response): the thread feed reads run_events, not checkpoints, so without the seed the inherited history disappears from the UI after the branch's first run (#4380). Seeded rows are grouped into one synthetic run per inherited turn (`branch-seed-{thread_id}-{n}`, a new turn opening at every persisted human message, including an allowlisted hidden `ask_clarification` reply) because `run_id` is a turn identity to the feed's consumers, not a provenance tag: regenerating an inherited answer supersedes that row's whole `run_id` in `GET /messages/page`, so one shared id for the entire seed deleted the complete inherited history on a branch's first regenerate (#4458); `GET /goal`, `PUT /goal`, `DELETE /goal` - read, set, and clear the active thread goal; `POST /compact` - summarize older active context, deriving memory policy and bucket from the state-producing checkpoint rather than request `agent_name`, and block while a run is in flight; unexpected failures return a generic 500 detail | | **Artifacts** (`/api/threads/{id}/artifacts`) | `GET /{path}` streams text/binary artifacts with byte-`Range` support; active HTML/XML/XSL/`+xml` content (including `.skill` members) is forced to download, and `?download=true` forces other types. `PUT /{path}` atomically replaces an existing UTF-8 file under `/mnt/user-data/outputs` on matching SHA-256; active runs conflict. Non-mounted sandbox sync uses a request lease, and caller cancellation drains the remote/local commit-or-rollback before releasing the thread-operation reservation or lease. `path_utils.resolve_outputs_confined_path` normalizes `..` and re-checks the resolved host path under the outputs root, so encoded traversal or output symlinks cannot reach sibling uploads. POSIX ownership/mode is preserved where descriptor APIs exist; Windows keeps native temp-file permissions. | | **Suggestions** (`/api/suggestions`) | `GET /config` - returns global suggestions config boolean; `POST /threads/{id}/suggestions` - generate follow-up questions; rich list/block model content is normalized and inline reasoning (`...`, including unclosed/truncated blocks from reasoning models like MiniMax-M3) is stripped before JSON parsing | diff --git a/backend/app/gateway/upload_ingestion.py b/backend/app/gateway/upload_ingestion.py index 292bc0c543..6d8caabfe7 100644 --- a/backend/app/gateway/upload_ingestion.py +++ b/backend/app/gateway/upload_ingestion.py @@ -198,7 +198,7 @@ class ThreadUploadIngestionService: user_id=self._user_id, app_config=self._config, owner_prefix="gateway:upload", - release_on_last=False, + release_on_last=True, ) self._sandbox = self._sandbox_lease.sandbox if not self._sandbox_lease.denied and self._sandbox is None: diff --git a/backend/tests/blocking_io/test_feishu_receive_file.py b/backend/tests/blocking_io/test_feishu_receive_file.py index 558e9d5d0e..68aa6a60e3 100644 --- a/backend/tests/blocking_io/test_feishu_receive_file.py +++ b/backend/tests/blocking_io/test_feishu_receive_file.py @@ -54,6 +54,7 @@ class _RemoteProvider: def __init__(self) -> None: self.sandbox = _RemoteSandbox() self.acquire_async_calls: list[tuple[str | None, str | None]] = [] + self.release_calls: list[str] = [] def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: raise AssertionError("Feishu receive_file must use acquire_async") @@ -65,6 +66,9 @@ class _RemoteProvider: def get(self, sandbox_id: str): return self.sandbox if sandbox_id == "remote-sandbox" else None + def release(self, sandbox_id: str) -> None: + self.release_calls.append(sandbox_id) + class _MountedProvider: uses_thread_data_mounts = True @@ -133,6 +137,7 @@ async def test_receive_file_remote_sandbox_does_not_block_event_loop(tmp_path, m assert provider.sandbox.updates == [(result, b"DATA")] assert provider_lookup_thread_id != loop_thread_id assert provider.sandbox.update_thread_id != loop_thread_id + assert provider.release_calls == ["remote-sandbox"] async def test_receive_file_mounted_sandbox_skips_redundant_sync(tmp_path, monkeypatch) -> None: @@ -155,7 +160,8 @@ async def test_receive_file_mounted_sandbox_skips_redundant_sync(tmp_path, monke assert await asyncio.to_thread(uploaded.read_bytes) == b"DATA" -async def test_cancelled_receive_file_holds_sandbox_lease_until_remote_sync_finishes(tmp_path, monkeypatch) -> None: +@pytest.mark.parametrize("has_active_run", [False, True]) +async def test_cancelled_receive_file_holds_sandbox_lease_until_remote_sync_finishes(tmp_path, monkeypatch, has_active_run) -> None: from deerflow.config.paths import Paths from deerflow.sandbox.lease import discard_sandbox_lease_manager, get_sandbox_lease_manager @@ -165,7 +171,8 @@ async def test_cancelled_receive_file_holds_sandbox_lease_until_remote_sync_fini monkeypatch.setattr("app.channels.feishu.get_paths", lambda: paths) monkeypatch.setattr("app.channels.feishu.get_sandbox_provider", lambda: provider) - await manager.acquire_async("active-run", "thread-1", user_id="ou-user") + if has_active_run: + await manager.acquire_async("active-run", "thread-1", user_id="ou-user") receive_task = asyncio.create_task( _channel_with_file()._receive_single_file( "message-1", @@ -182,7 +189,8 @@ async def test_cancelled_receive_file_holds_sandbox_lease_until_remote_sync_fini await asyncio.sleep(0) assert not receive_task.done() - await manager.release_async("active-run") + if has_active_run: + await manager.release_async("active-run") assert provider.release_calls == [] assert not provider.sandbox.closed diff --git a/backend/tests/test_aio_sandbox_provider.py b/backend/tests/test_aio_sandbox_provider.py index 9c3b23d8bf..96af2e05a7 100644 --- a/backend/tests/test_aio_sandbox_provider.py +++ b/backend/tests/test_aio_sandbox_provider.py @@ -1611,6 +1611,29 @@ def _make_provider_with_active_sandbox(tmp_path, sandbox_id: str): return provider, sandbox, aio_mod +def test_reused_active_sandbox_requires_matching_releases(tmp_path): + """The execution lease manager keeps AIO clients active until the final holder exits.""" + from deerflow.sandbox.lease import SandboxLeaseManager + + provider, sandbox, _ = _make_provider_with_active_sandbox(tmp_path, "sandbox-lease") + manager = SandboxLeaseManager(provider) + try: + for owner in ("active-run", "temporary-upload"): + manager.retain(owner, "sandbox-lease", thread_id="thread-lease", user_id="owner-upload") + + manager.release("temporary-upload") + assert "sandbox-lease" in provider._sandboxes + assert "sandbox-lease" not in provider._warm_pool + sandbox.close.assert_not_called() + + manager.release("active-run") + assert "sandbox-lease" not in provider._sandboxes + assert "sandbox-lease" in provider._warm_pool + sandbox.close.assert_called_once_with() + finally: + manager.close() + + def test_release_closes_cached_sandbox_client(tmp_path): """release() must close the host-side client owned by the cached AioSandbox (#2872).""" provider, sandbox, _ = _make_provider_with_active_sandbox(tmp_path, "sandbox-rel") diff --git a/backend/tests/test_feishu_parser.py b/backend/tests/test_feishu_parser.py index 5acc1896f2..1f814f53f8 100644 --- a/backend/tests/test_feishu_parser.py +++ b/backend/tests/test_feishu_parser.py @@ -77,6 +77,54 @@ def _feishu_file_channel(*responses): return channel +@pytest.mark.parametrize("failure", [None, "sync", "release"]) +def test_feishu_receive_single_file_releases_sandbox_after_sync(tmp_path, monkeypatch, failure): + async def go(): + from deerflow.config.paths import Paths + + channel = _feishu_file_channel(_feishu_file_response("note.txt", b"hello uploads")) + provider = MagicMock() + provider.uses_thread_data_mounts = False + provider.acquire_async = AsyncMock(return_value="aio-1") + sandbox = provider.get.return_value + if failure == "sync": + sandbox.update_file.side_effect = RuntimeError("sync failed") + elif failure == "release": + provider.release.side_effect = RuntimeError("release failed") + monkeypatch.setattr(feishu_module, "get_paths", lambda: Paths(base_dir=tmp_path)) + monkeypatch.setattr(feishu_module, "get_sandbox_provider", lambda: provider) + + result = await channel._receive_single_file("message-1", "file-key", "file", "thread-a", user_id="owner-upload") + + assert result == ("Failed to obtain the [file]" if failure == "sync" else "/mnt/user-data/uploads/note.txt") + provider.acquire_async.assert_awaited_once_with("thread-a", user_id="owner-upload") + sandbox.update_file.assert_called_once_with("/mnt/user-data/uploads/note.txt", b"hello uploads") + provider.release.assert_called_once_with("aio-1") + + _run(go()) + + +def test_feishu_receive_single_file_skips_release_for_mounted_sandbox(tmp_path, monkeypatch): + async def go(): + from deerflow.config.paths import Paths + + channel = _feishu_file_channel(_feishu_file_response("note.txt", b"hello uploads")) + provider = MagicMock() + provider.uses_thread_data_mounts = True + monkeypatch.setattr(feishu_module, "get_paths", lambda: Paths(base_dir=tmp_path)) + monkeypatch.setattr(feishu_module, "get_sandbox_provider", lambda: provider) + + result = await channel._receive_single_file("message-1", "file-key", "file", "thread-a", user_id="owner-upload") + + assert result == "/mnt/user-data/uploads/note.txt" + provider.acquire.assert_not_called() + provider.acquire_async.assert_not_called() + provider.get.assert_not_called() + provider.release.assert_not_called() + + _run(go()) + + def test_feishu_on_message_plain_text(): bus = MessageBus() config = {"app_id": "test", "app_secret": "test"} diff --git a/backend/tests/test_uploads_router.py b/backend/tests/test_uploads_router.py index 096ce2de30..9d930e76fb 100644 --- a/backend/tests/test_uploads_router.py +++ b/backend/tests/test_uploads_router.py @@ -74,6 +74,7 @@ def test_upload_files_writes_thread_storage_and_skips_local_sandbox_sync(tmp_pat assert (thread_uploads_dir / "notes.txt").read_bytes() == b"hello uploads" sandbox.update_file.assert_not_called() + provider.release.assert_not_called() def test_upload_and_list_response_models_expose_size_as_int(tmp_path): @@ -616,6 +617,7 @@ def test_upload_files_acquires_non_local_sandbox_before_writing(tmp_path): provider.acquire.assert_not_called() provider.acquire_async.assert_awaited_once_with("thread-aio", user_id="owner-upload") sandbox.update_file.assert_called_once_with("/mnt/user-data/uploads/notes.txt", b"hello uploads") + provider.release.assert_called_once_with("aio-1") def test_upload_files_fails_before_writing_when_non_local_sandbox_unavailable(tmp_path): @@ -639,6 +641,7 @@ def test_upload_files_fails_before_writing_when_non_local_sandbox_unavailable(tm assert file.read_calls == [] provider.acquire.assert_not_called() provider.get.assert_not_called() + provider.release.assert_not_called() def test_upload_files_rejects_too_many_files_before_writing(tmp_path): @@ -830,6 +833,7 @@ def test_upload_files_does_not_sync_non_local_sandbox_when_total_size_exceeds_li provider.acquire_async.assert_awaited_once_with("thread-aio", user_id="owner-upload") assert provider.get.call_count == 2 assert all(call.args == ("aio-1",) for call in provider.get.call_args_list) + provider.release.assert_called_once_with("aio-1") sandbox.update_file.assert_not_called() @@ -860,10 +864,78 @@ def test_upload_files_does_not_sync_non_local_sandbox_when_conversion_fails(tmp_ provider.acquire_async.assert_awaited_once_with("thread-aio", user_id="owner-upload") assert provider.get.call_count == 2 assert all(call.args == ("aio-1",) for call in provider.get.call_args_list) + provider.release.assert_called_once_with("aio-1") sandbox.update_file.assert_not_called() assert not (thread_uploads_dir / "report.pdf").exists() +@pytest.mark.parametrize("failure", ["lookup", "sync"]) +def test_upload_files_releases_non_local_sandbox_when_sync_fails(tmp_path, failure): + provider = MagicMock() + provider.uses_thread_data_mounts = False + provider.acquire_async = AsyncMock(return_value="aio-1") + sandbox = provider.get.return_value + if failure == "lookup": + provider.get.return_value = None + else: + sandbox.update_file.side_effect = RuntimeError("sync failed") + file = ChunkedUpload("notes.txt", [b"hello uploads"]) + + with ( + patch.object(uploads, "ensure_uploads_dir", return_value=tmp_path), + patch.object(uploads, "get_sandbox_provider", return_value=provider), + ): + with pytest.raises(HTTPException if failure == "lookup" else RuntimeError): + asyncio.run(call_unwrapped(uploads.upload_files, "thread-aio", request=MagicMock(), files=[file], config=SimpleNamespace())) + + provider.release.assert_called_once_with("aio-1") + if failure == "lookup": + assert file.read_calls == [] + assert list(tmp_path.iterdir()) == [] + + +@pytest.mark.parametrize("run_finishes_during_sync", [False, True]) +def test_upload_files_keeps_active_run_sandbox_until_last_holder_finishes(tmp_path, run_finishes_during_sync): + from deerflow.sandbox.lease import discard_sandbox_lease_manager, get_sandbox_lease_manager + + async def go(): + provider = MagicMock() + provider.uses_thread_data_mounts = False + provider.acquire_async = AsyncMock(return_value="aio-1") + sandbox = provider.get.return_value + manager = get_sandbox_lease_manager(provider) + await manager.acquire_async("active-run", "thread-aio", user_id="owner-upload") + + def sync_file(path, content): + assert path == "/mnt/user-data/uploads/notes.txt" + assert content == b"hello uploads" + if run_finishes_during_sync: + manager.release("active-run") + provider.release.assert_not_called() + + sandbox.update_file.side_effect = sync_file + try: + with ( + patch.object(uploads, "get_effective_user_id", return_value="owner-upload"), + patch.object(uploads, "ensure_uploads_dir", return_value=tmp_path), + patch.object(uploads, "get_sandbox_provider", return_value=provider), + ): + file = UploadFile(filename="notes.txt", file=BytesIO(b"hello uploads")) + result = await call_unwrapped(uploads.upload_files, "thread-aio", request=MagicMock(), files=[file], config=SimpleNamespace()) + + assert result.success + if not run_finishes_during_sync: + provider.release.assert_not_called() + assert manager.binding_for("active-run") == "aio-1" + await manager.release_async("active-run") + provider.release.assert_called_once_with("aio-1") + finally: + await manager.release_async("active-run") + discard_sandbox_lease_manager(provider) + + asyncio.run(go()) + + def test_make_file_sandbox_writable_adds_write_bits_for_regular_files(tmp_path): file_path = tmp_path / "report.pdf" file_path.write_bytes(b"pdf-bytes")