From 7bb92cea10bb45c63b97a2cb7ae9457a576dc75b Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Mon, 9 Nov 2026 12:53:56 +0900 Subject: [PATCH] fix(threads): keep the owner row when deleting checkpoints fails (#7992) * fix(threads): keep the owner row when deleting checkpoints fails DELETE /api/threads/{id} treated the checkpoint delete as best-effort: a failure was logged at DEBUG, the threads_meta row was removed anyway and the request returned success. Read routes allow a thread without a meta row as an untracked legacy thread, so the surviving checkpoints (title, messages, goal) became readable by every signed-in user through GET /threads/{id}, /state, /history and /goal, and the owner could not retry because DELETE requires the row. A failed checkpoint delete now logs the exception and returns a generic 500 before the meta row is touched, so the thread stays owned and the owner can retry. The remaining cleanup steps stay best-effort; their reads are owner-scoped. * docs(changelog): reference #7992 for the thread delete checkpoint fix * fix(threads): require checkpoint deletion before dropping ownership --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 10 ++++++ CHANGELOG_zh.md | 8 +++++ README.md | 7 +++- backend/app/gateway/AGENTS.md | 2 +- backend/app/gateway/routers/threads.py | 11 +++--- backend/docs/API.md | 10 +++++- backend/tests/test_threads_router.py | 49 ++++++++++++++++++++++++++ 7 files changed, 89 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 38e51dc766..90f5abd74c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,15 @@ This section accumulates work toward the **2.3.0** milestone ### Fixed +- **gateway:** Deleting a thread no longer reports success while its + conversation stays readable. When the checkpoint store failed during + `DELETE /api/threads/{id}`, the route logged it at DEBUG, removed the + thread's metadata row and returned success; any signed-in user could then + read the leftover title and messages through `GET /threads/{id}`, `/state`, + `/history` and `/goal`, and the owner could not retry. The delete now + returns a 500 and keeps the thread, so the owner can delete it again. + ([#7992]) + - **middleware:** With PII redaction enabled, a long run of text without spaces (such as a pasted hex dump) no longer stalls the Gateway. The email detector was quadratic in the length of such a run, and redaction re-scans @@ -16165,5 +16174,6 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#7953]: https://github.com/bytedance/deer-flow/issues/7953 [#7958]: https://github.com/bytedance/deer-flow/pull/7958 [#7981]: https://github.com/bytedance/deer-flow/pull/7981 +[#7992]: https://github.com/bytedance/deer-flow/pull/7992 [#7993]: https://github.com/bytedance/deer-flow/pull/7993 [#8006]: https://github.com/bytedance/deer-flow/pull/8006 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 49c1d9fb58..d9f6e80596 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -26,10 +26,17 @@ ### 修复 +- **网关：** 删除线程时不再在对话仍可读取的情况下返回成功。此前 + `DELETE /api/threads/{id}` 在检查点存储失败时只记录 DEBUG 日志，仍删除线程元数据行并 + 返回成功；之后任何已登录用户都能通过 `GET /threads/{id}`、`/state`、`/history` 和 + `/goal` 读取残留的标题和消息，所有者也无法重试。现在删除会返回 500 并保留线程，所有者 + 可以再次删除。([#7992]) + - **中间件：** 启用 PII 脱敏时，不含空格的长文本（例如粘贴的十六进制转储）不再让 Gateway 卡住。此前邮箱检测器的耗时与这类文本长度呈平方关系，而脱敏会在每次模型调用时重新扫描全部 用户消息，因此一次 100 KB 的粘贴会让每次模型调用阻塞所有线程约 6 秒。现在扫描为线性耗时， 且脱敏的地址与此前完全相同。([#8006]) + - **调度器：** 一个已保存的计划无法再计算的定时任务，不再让其他所有任务停止运行。此前 已保存的时区或 cron 表达式失效时（例如升级后时区数据库中缺少该时区，或新版 croniter 不再接受该表达式），在调度器轮询中计算其下一次运行时间会抛出异常；其启动租约过期后， @@ -12827,5 +12834,6 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#7953]: https://github.com/bytedance/deer-flow/issues/7953 [#7958]: https://github.com/bytedance/deer-flow/pull/7958 [#7981]: https://github.com/bytedance/deer-flow/pull/7981 +[#7992]: https://github.com/bytedance/deer-flow/pull/7992 [#7993]: https://github.com/bytedance/deer-flow/pull/7993 [#8006]: https://github.com/bytedance/deer-flow/pull/8006 diff --git a/README.md b/README.md index 3d2a5e9d5d..8f9a362f20 100644 --- a/README.md +++ b/README.md @@ -3745,7 +3745,12 @@ editing the saved agent configuration to refresh the selection. `DeerFlowClient.stream()` includes `summary_text` in each `values` event. This is the current compacted context summary, or `None` when absent. Consumers can record changes without reading checkpoint internals; repeated snapshots may carry the same summary, and an initial snapshot may already contain one from an earlier turn. -DeerFlow can be used as an embedded Python library without running the full HTTP services. The `DeerFlowClient` provides direct in-process access to all agent and Gateway capabilities, returning the same response schemas as the HTTP Gateway API. The HTTP Gateway also exposes `DELETE /api/threads/{thread_id}` to remove DeerFlow-managed local thread data after the LangGraph thread itself has been deleted: +DeerFlow can be used as an embedded Python library without running the full HTTP services. The `DeerFlowClient` provides direct in-process access to all agent and Gateway capabilities, returning the same response schemas as the HTTP Gateway API. + +The HTTP Gateway exposes `DELETE /api/threads/{thread_id}` to remove a thread's +files, checkpoints, and metadata. If checkpoint deletion fails, the Gateway +returns 500 and retains the owner record for retry. +[Thread cleanup API](backend/docs/API.md#thread-cleanup) An embedded `DeerFlowClient(config_path="custom/config.yaml")` keeps the process config singleton on that file, even after a working-directory change. Later edits diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index a62fc2871e..18e8d55a2f 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -223,7 +223,7 @@ or `deployment.multi_instance: true`. - 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 (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. +- `DELETE /api/threads/{id}` holds its durable `delete` reservation through filesystem cleanup (off-loop, drained on cancel), checkpoint deletion and best-effort owner-scoped run, event, feedback and metadata cleanup. If the configured checkpointer cannot delete checkpoints, return 500 and keep the owner row so legacy reads cannot bypass ownership. Historical-run cleanup deletes only `operation_kind="run"` rows, preserving reservations until `RunManager.reserve_thread_operation()` exits. Fencing already-admitted writes against deleted state is a separate thread-incarnation contract. - 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 data and first-END writes) 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. Terminal `orphan_recovered` and `scheduled_task_orphan_recovered` rows permit heartbeat fallback because their producer is gone; ordinary terminal rows still await END. 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 71c79d5166..529281352b 100644 --- a/backend/app/gateway/routers/threads.py +++ b/backend/app/gateway/routers/threads.py @@ -831,14 +831,15 @@ async def _delete_thread_data_with_reservation(thread_id: str, request: Request) # 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) + # Retain ownership if a configured checkpointer cannot delete the thread. + # Read routes treat rowless checkpoints as untracked legacy threads. checkpointer = getattr(request.app.state, "checkpointer", None) if checkpointer is not None: try: - if hasattr(checkpointer, "adelete_thread"): - await checkpointer.adelete_thread(thread_id) - except Exception: - logger.debug("Could not delete checkpoints for thread %s (not critical)", sanitize_log_param(thread_id)) + await checkpointer.adelete_thread(thread_id) + except Exception as exc: + logger.exception("Failed to delete checkpoints for thread %s", sanitize_log_param(thread_id)) + raise HTTPException(status_code=500, detail="Failed to delete thread checkpoints.") from exc # Remove historical runs (best-effort). Only ``operation_kind == "run"`` rows # are deleted, so the durable thread-operation reservation protecting this diff --git a/backend/docs/API.md b/backend/docs/API.md index 62d5a3a404..de08fa3202 100644 --- a/backend/docs/API.md +++ b/backend/docs/API.md @@ -1458,7 +1458,7 @@ DELETE /api/threads/{thread_id}/uploads/{filename} ### Thread Cleanup -Remove DeerFlow-managed local thread files under `.deer-flow/threads/{thread_id}` after the LangGraph thread itself has been deleted. +Delete a thread's checkpoint data and DeerFlow-managed files and metadata. ```http DELETE /api/threads/{thread_id} @@ -1475,6 +1475,14 @@ DELETE /api/threads/{thread_id} **Error behavior:** - `422` for invalid thread IDs - `500` returns a generic `{"detail": "Failed to delete local thread data."}` response while full exception details stay in server logs +- `500` with `{"detail": "Failed to delete thread checkpoints."}` when a configured + checkpointer cannot delete the thread's checkpoints. The owner row is retained + for a retry. +- `409` with `Thread has work in flight. Delete it after the work finishes.` when + thread work blocks deletion. After a checkpoint error, a retry can also return + this response until the durable delete reservation is released or reconciled. + In a single-worker deployment with ownership heartbeats disabled, a stranded + reservation is reconciled at Gateway startup. Deleting a thread also removes every user's read marker for it. diff --git a/backend/tests/test_threads_router.py b/backend/tests/test_threads_router.py index dec305f33e..c641827925 100644 --- a/backend/tests/test_threads_router.py +++ b/backend/tests/test_threads_router.py @@ -1,4 +1,5 @@ import asyncio +import logging import re import threading from contextlib import asynccontextmanager @@ -5638,3 +5639,51 @@ def test_delete_thread_route_removes_read_markers(tmp_path) -> None: assert asyncio.run(_count()) is None finally: asyncio.run(engine.dispose()) + + +@pytest.mark.parametrize("delete_supported", [True, False], ids=["delete-raises", "delete-unsupported"]) +def test_delete_thread_route_keeps_owner_row_when_checkpoint_delete_fails(tmp_path, monkeypatch, caplog, delete_supported) -> None: + """A failed checkpoint delete must not leave a checkpoint-only thread behind. + + Read routes treat a thread without a ``threads_meta`` row as an untracked + legacy thread open to every user, so dropping the row while the checkpoints + survive would publish the deleted conversation. The delete fails instead, + keeps the owner's row, and succeeds when the owner retries. + """ + app = _build_project_threads_app(tmp_path) + owner, other = {"x-test-user": "user-a"}, {"x-test-user": "user-b"} + checkpointer = app.state.checkpointer + + with patch("app.gateway.routers.threads.get_paths", return_value=Paths(tmp_path)), TestClient(app) as client: + thread_id = client.post("/api/threads", json={}, headers=owner).json()["thread_id"] + anyio.run(lambda: _write_checkpoint(checkpointer, thread_id, str(uuid6()), [HumanMessage(id="human-1", content="private question")], step=1)) + + with monkeypatch.context() as failed_checkpoint: + if delete_supported: + failed_checkpoint.setattr(checkpointer, "adelete_thread", AsyncMock(side_effect=RuntimeError("database is locked: /secret/checkpoints.db"))) + else: + read_only_checkpointer = SimpleNamespace(aget_tuple=checkpointer.aget_tuple, alist=checkpointer.alist) + failed_checkpoint.setattr(app.state, "checkpointer", read_only_checkpointer) + readable = client.get(f"/api/threads/{thread_id}", headers=owner) + assert readable.status_code == 200, readable.text + assert readable.json()["values"]["messages"][0]["id"] == "human-1" + with caplog.at_level(logging.ERROR, logger=threads.logger.name): + failed = client.delete(f"/api/threads/{thread_id}", headers=owner) + + assert failed.status_code == 500 + assert failed.json() == {"detail": "Failed to delete thread checkpoints."} + assert any(record.levelno >= logging.ERROR and "Failed to delete checkpoints" in record.getMessage() for record in caplog.records) + assert anyio.run(lambda: app.state.thread_store.check_access(thread_id, "user-a", require_existing=True)) + for read in ( + lambda: client.get(f"/api/threads/{thread_id}", headers=other), + lambda: client.get(f"/api/threads/{thread_id}/state", headers=other), + lambda: client.post(f"/api/threads/{thread_id}/history", json={}, headers=other), + lambda: client.get(f"/api/threads/{thread_id}/goal", headers=other), + ): + assert read().status_code == 404 + + retried = client.delete(f"/api/threads/{thread_id}", headers=owner) + + assert retried.status_code == 200, retried.text + assert anyio.run(lambda: checkpointer.aget_tuple({"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}})) is None + assert client.get(f"/api/threads/{thread_id}", headers=other).status_code == 404