diff --git a/CHANGELOG.md b/CHANGELOG.md index 21258f2169..00611fb3a0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -466,6 +466,14 @@ This release closes that milestone with **301 merged pull requests**. ### Fixed +- **gateway:** Deleting a thread with a large workspace no longer freezes every + other Gateway request while its files are removed. `DELETE /api/threads/{id}` + ran `shutil.rmtree` over the thread directory on the event loop, so other + requests and live SSE streams stalled until the whole tree was gone (about + 0.7 seconds for 20,000 small files on a local SSD, longer on mounted + volumes). The removal now runs on the file-IO pool, and a cancelled request + keeps its thread reservation until the removal finishes, so no new run can + start on a thread whose files are still being deleted. ([#6319]) - **sandbox:** Medium-risk audit warnings no longer hide a failed shell exit from subagent evidence. `SandboxAuditMiddleware` appended its warning after the trailing `Exit Code: N` marker and rebuilt the `ToolMessage` from four fields, @@ -7766,3 +7774,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6305]: https://github.com/bytedance/deer-flow/pull/6305 [#6306]: https://github.com/bytedance/deer-flow/pull/6306 [#6307]: https://github.com/bytedance/deer-flow/pull/6307 +[#6319]: https://github.com/bytedance/deer-flow/pull/6319 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 9600d1dffb..f03cdf552f 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -423,6 +423,11 @@ ### 修复 +- **网关:** 删除工作区较大的线程时,移除文件期间不再冻结 Gateway 的其他所有请求。 + `DELETE /api/threads/{id}` 此前在事件循环上对线程目录执行 `shutil.rmtree`, + 因此其他请求与进行中的 SSE 流都要等到整棵目录树删除完毕(本地 SSD 上 20,000 个 + 小文件约 0.7 秒,挂载卷上更久)。现在移除在文件 IO 线程池中执行;请求被取消时, + 线程预留会保持到移除完成,因此文件仍在删除的线程上不会启动新的运行。([#6319]) - **沙箱:** 中风险审计警告不再让子智能体证据丢失失败的 shell 退出码。 `SandboxAuditMiddleware` 原先把警告追加在结尾的 `Exit Code: N` 标记之后,并只用 四个字段重建 `ToolMessage`,导致 `_bash_evidence_status` 找不到标记,退回到报告 @@ -6478,3 +6483,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子 [#6305]: https://github.com/bytedance/deer-flow/pull/6305 [#6306]: https://github.com/bytedance/deer-flow/pull/6306 [#6307]: https://github.com/bytedance/deer-flow/pull/6307 +[#6319]: https://github.com/bytedance/deer-flow/pull/6319 diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index aca1b3a7ab..71cec1c1f6 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -198,7 +198,7 @@ startup gate rejects process-local memory and JSONL event stores when - Run admission and independent writes are first-class thread operations. `runs.operation_kind` distinguishes `run` from `checkpoint_write`, `artifact_write`, `artifact_archive`, `branch`, and `delete`; every active kind shares the durable active-thread uniqueness constraint. New operation kinds must go through `RunStore.create_thread_operation_atomic()` and `RunManager.reserve_thread_operation()` rather than adding another lock or metadata marker. Live and lease-less reservations are non-interruptible; an expired leased reservation can be reclaimed immediately by interrupt/rollback admission without waiting for orphan reconciliation. Lease-less rows stay fail-closed because the store cannot distinguish a stale row from a live writer in another heartbeat-disabled worker; a rare failed delete therefore requires startup reconciliation, and heartbeat-disabled multi-worker deployment remains unsupported. Reservation bodies are attached to their caller task so loss detected by lease renewal cancels the writer before it can continue after takeover; the context manager translates that lease-loss cancellation to `ConflictError` after cleanup so Gateway mutation routes return a retryable 409 instead of dropping the HTTP request. The cleanup scope begins immediately after durable admission, including the await that attaches the caller task, so cancellation cannot strand a locally renewed pending reservation. A failed renewal is revalidated under the manager lock before cancellation; if the reservation completed and unregistered while the store update was in flight, its request task must not be cancelled after the write. Reservations are excluded from run history/reporting and from run-only helpers such as `list_by_thread()` and `has_inflight()`, release uses the captured owner rather than ambient user context, and local cleanup still runs when the best-effort store delete fails. `RunStore.create_run_atomic()` remains a deprecated compatibility shim for external stores that only admit normal runs; new stores must implement `create_thread_operation_atomic()` to support internal operation kinds. - Gateway checkpoint mutations outside run execution must use `services.reserve_checkpoint_write()`, which composes the process-local thread lock with the durable `checkpoint_write` reservation. Manual compaction, `POST /threads/{id}/state`, and both goal mutation routes (`PUT` / `DELETE /threads/{id}/goal`, including creation of a missing goal checkpoint) use this boundary, so an existing run blocks the write and the reservation blocks new reject/interrupt/rollback runs across workers. - Branch/state-update checkpoints copy only the source checkpoint's persisted `deerflow_agent_name`, never request metadata; missing or malformed bindings stay unbound so compaction fails closed. -- `DELETE /api/threads/{id}` holds a durable `delete` reservation for the whole cleanup and removes thread filesystem data, checkpoints, owner-scoped historical runs, run events, feedback and thread metadata best-effort. Historical-run cleanup removes only `operation_kind="run"` rows, so internal thread-operation rows — including the reservation protecting that request — stay durable until `RunManager.reserve_thread_operation()` exits. This cleans persisted rows; preventing an already-admitted write from re-creating state for a deleted thread is a separate thread-incarnation contract, not part of this cleanup. +- `DELETE /api/threads/{id}` holds a durable `delete` reservation for the whole cleanup and removes thread filesystem data (off-loop, drained on cancel), checkpoints, owner-scoped historical runs, run events, feedback and thread metadata best-effort. Historical-run cleanup removes only `operation_kind="run"` rows, so internal thread-operation rows — including the reservation protecting that request — stay durable until `RunManager.reserve_thread_operation()` exits. This cleans persisted rows; preventing an already-admitted write from re-creating state for a deleted thread is a separate thread-incarnation contract, not part of this cleanup. - Creator SSE and `/wait` apply `on_disconnect` only to confirmed request disconnect or cancellation; bridge failures propagate without cancelling the run. `/wait` consumes bridge END; errors return SDK `__error__`, never stale state. - Memory and Redis `StreamBridge` implementations retain only `stream_bridge.queue_maxsize` data events. A syntactically valid `Last-Event-ID` older than the retained watermark, or a live subscriber that falls behind it, yields `StreamGap` before any partial replay. `sse_consumer` maps that control item to an id-less SSE `gap` payload (`stream_replay_gap`) and intentionally leaves the run active; internal `/wait` consumers resume from its latest retained ID because they only need terminal completion. Redis checks bounds plus the non-blocking read in one transaction, using blocking `XREAD` only as a wake-up before repeating the atomic snapshot. For a no-cursor subscriber that established a wait on an empty stream, the first wake response remains provisional until that next snapshot verifies its tail is still retained; this closes the pre-first-delivery trimming window without changing malformed-cursor live tailing. The correctness tradeoff is one three-command snapshot pipeline per poll plus the blocking wake round trip while idle. Malformed cursor behavior remains backend-specific. Memory treats a syntactically numeric cursor below its watermark conservatively as a gap even when the evicted timestamp can no longer be verified; unknown ids at or above the watermark retain the legacy replay-from-earliest policy. - Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on `publish()` / `publish_end()`) as a leak safety net, not as a run timeout. Startup and lease-driven periodic orphan recovery share one Gateway stream-terminalization path: after `RunManager` durably marks a run `error` with `stop_reason=orphan_recovered`, Gateway publishes `END_SENTINEL` and schedules stream cleanup. The periodic store scan, per-row status writes, and Gateway callback run as one supervised single-flight task, so a slow pass is skipped at the next interval instead of piling up or pausing the sole lease-renewal loop. Store retries have bounded attempts/backoff; an individual operation still relies on the database driver/pool timeout. `RunManager.shutdown()` gives active user runs priority within its shared deadline, then drains or cancels orphan recovery. Gateway tracks delayed recovered-stream cleanups and converts unfinished delays to immediate deletes before closing the bridge; the Redis TTL remains the outage safety net. Only startup recovery, before the runtime yields to requests, projects the latest affected thread to `error`; periodic recovery deliberately avoids that non-atomic projection because `ThreadMetaStore` has no `latest_run_id` conditional-update contract. Store-only SSE and `/wait` consumers wait for the bridge's real END marker after an ordinary durable terminal status, because status persistence can precede tail events. The explicit `orphan_recovered` signal is the only heartbeat fallback: its publisher is known to be gone, so it supplies the liveness boundary if END publication fails or the retained key expires. Malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Keep cross-component recovery orchestration in Gateway through the generic `RunManager.on_orphans_recovered` callback; do not introduce a harness-to-app dependency. Callback failure warnings include every recovered `run_id` so operators can identify rows whose Gateway-side terminalization needs inspection. diff --git a/backend/app/gateway/routers/threads.py b/backend/app/gateway/routers/threads.py index f9989c5540..9ca34e0cee 100644 --- a/backend/app/gateway/routers/threads.py +++ b/backend/app/gateway/routers/threads.py @@ -73,7 +73,7 @@ from deerflow.runtime.runs.worker import RUN_MESSAGE_IDS_METADATA_KEY, valid_dur from deerflow.runtime.secret_context import redact_metadata_secrets from deerflow.runtime.user_context import get_effective_user_id from deerflow.uploads.companions import register_companion, resolve_companion -from deerflow.utils.file_io import run_file_io +from deerflow.utils.file_io import await_drained, run_file_io from deerflow.utils.thread_id import ThreadId, resolve_thread_id, validate_thread_id from deerflow.utils.time import coerce_iso, now_iso @@ -778,7 +778,11 @@ async def _delete_thread_data_with_reservation(thread_id: str, request: Request) message="Skipped local data cleanup for legacy thread ID", ) else: - response = _delete_thread_data(thread_id, user_id=user_id) + # rmtree over a large workspace must not stall the loop. Drained, not + # bare-awaited: cancelling the await abandons the worker, not the + # removal, so the reservation would release while files are still + # being deleted under a run that has just been admitted. + response = await await_drained(run_file_io(_delete_thread_data, thread_id, user_id=user_id)) # Remove checkpoints (best-effort) checkpointer = getattr(request.app.state, "checkpointer", None) diff --git a/backend/tests/blocking_io/test_threads_delete.py b/backend/tests/blocking_io/test_threads_delete.py new file mode 100644 index 0000000000..da4d40b3a6 --- /dev/null +++ b/backend/tests/blocking_io/test_threads_delete.py @@ -0,0 +1,60 @@ +"""Regression anchor: deleting a thread must not block the event loop. + +``DELETE /api/threads/{id}`` removes the thread's whole directory tree +(workspace, uploads, outputs). ``shutil.rmtree`` over a large workspace +stalls every other request on the Gateway loop, so the handler offloads it +via ``run_file_io``; if the removal regresses onto the event loop, the strict +Blockbuster gate raises ``BlockingError``, the handler maps it to a 500, and +this test fails. + +The handler is driven through ``__wrapped__`` (past the authz decorator) so +the goal lock, the thread-operation reservation and every cleanup step run as +in production. Imports live at module scope so their file reads happen at +collection time, not on the loop under test; test-side seeding and checks are +offloaded with ``asyncio.to_thread``. +""" + +from __future__ import annotations + +import asyncio +from contextlib import asynccontextmanager +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from app.gateway.routers import threads +from deerflow.config.paths import Paths +from deerflow.runtime.user_context import get_effective_user_id + +pytestmark = pytest.mark.asyncio + + +class _RunManager: + @asynccontextmanager + async def reserve_thread_operation(self, _thread_id: str, **_kwargs): + yield + + +def _seed_workspace(paths: Paths, thread_id: str, user_id: str) -> Path: + paths.ensure_thread_dirs(thread_id, user_id=user_id) + workspace = paths.sandbox_work_dir(thread_id, user_id=user_id) + for index in range(20): + (workspace / f"file-{index}.txt").write_text("content", encoding="utf-8") + return paths.thread_dir(thread_id, user_id=user_id) + + +async def test_delete_thread_does_not_block_event_loop(tmp_path: Path) -> None: + user_id = get_effective_user_id() + # test-side setup and seeding (offloaded; not exercised on the loop) + paths = await asyncio.to_thread(Paths, tmp_path) + thread_dir = await asyncio.to_thread(_seed_workspace, paths, "thread-loop-delete", user_id) + request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(run_manager=_RunManager(), checkpointer=None))) + + with patch("app.gateway.routers.threads.get_paths", return_value=paths): + response = await threads.delete_thread_data.__wrapped__("thread-loop-delete", request) + + assert response.success is True + assert response.message == "Deleted local thread data for thread-loop-delete" + assert not await asyncio.to_thread(thread_dir.exists) diff --git a/backend/tests/test_threads_router.py b/backend/tests/test_threads_router.py index b574176f85..b5d35a8a6e 100644 --- a/backend/tests/test_threads_router.py +++ b/backend/tests/test_threads_router.py @@ -1,5 +1,6 @@ import asyncio import re +import threading from contextlib import asynccontextmanager from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch @@ -747,6 +748,55 @@ def test_delete_thread_route_rejects_active_thread_operation_without_deleting_me assert asyncio.run(store.aget(THREADS_NS, "thread-active-delete")) is not None +def test_delete_thread_route_holds_reservation_until_cancelled_removal_finishes(tmp_path): + """A cancelled delete must not release its reservation mid-rmtree. + + The removal runs on a file-IO worker. Cancelling the request abandons the + await, not the worker, so a bare await would let ``reserve_thread_operation`` + exit -- and admit a new run on the thread -- while its files are still + being deleted. The stalled removal below makes that window observable. + """ + events: list[str] = [] + removal_started = threading.Event() + release_removal = threading.Event() + + class RecordingRunManager(_ThreadTestRunManager): + @asynccontextmanager + async def reserve_thread_operation(self, _thread_id: str, **_kwargs): + try: + yield + finally: + events.append("reservation released") + + def stalled_delete_thread_dir(_thread_id, *, user_id=None): + removal_started.set() + release_removal.wait(timeout=5) + events.append("removal finished") + + paths = Paths(tmp_path) + request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(run_manager=RecordingRunManager(), checkpointer=None))) + + async def scenario() -> None: + task = asyncio.create_task(threads.delete_thread_data.__wrapped__("thread-cancelled-delete", request)) + assert await asyncio.to_thread(removal_started.wait, 5) + task.cancel() + await asyncio.sleep(0.05) + # Cancellation is deferred while the removal is still running. + assert not task.done() + assert events == [] + release_removal.set() + with pytest.raises(asyncio.CancelledError): + await task + + with ( + patch("app.gateway.routers.threads.get_paths", return_value=paths), + patch.object(paths, "delete_thread_dir", side_effect=stalled_delete_thread_dir), + ): + asyncio.run(scenario()) + + assert events == ["removal finished", "reservation released"] + + def test_branch_thread_route_rejects_concurrent_source_operation_without_creating_child(): class RejectingRunManager(_ThreadTestRunManager): @asynccontextmanager