fix: balance temporary sandbox leases for uploads (#3261)

* fix: balance temporary sandbox leases for uploads

* fix: address sandbox upload review feedback

---------

Co-authored-by: Bxl <188532127+BXL1015@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

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`).

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)

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)

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 (`<think>...</think>`, including unclosed/truncated blocks from reasoning models like MiniMax-M3) is stripped before JSON parsing |

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:

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
 

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")

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"}

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")