From fc26204debe2808fa1b4064f8fddfea6db2c71f8 Mon Sep 17 00:00:00 2001
From: NanPan <111261006+poijygfdyy@users.noreply.github.com>
Date: Wed, 23 Sep 2026 23:26:40 +0800
Subject: [PATCH] fix(gateway): drain artifact updates across cancellation
(#5755)
* fix(gateway): drain artifact updates across cancellation
* fix(gateway): log cancelled artifact commit failures
* refactor(gateway): tighten artifact cancellation follow-up
* docs(gateway): trim artifact guidance
* docs(gateway): retain XSL download guidance
---
backend/app/gateway/AGENTS.md | 2 +-
backend/app/gateway/routers/artifacts.py | 68 ++++++++++++-----
backend/tests/test_artifacts_router.py | 97 ++++++++++++++++++++++++
3 files changed, 146 insertions(+), 21 deletions(-)
diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md
index d224ecbc1..5e70a1228 100644
--- a/backend/app/gateway/AGENTS.md
+++ b/backend/app/gateway/AGENTS.md
@@ -105,7 +105,7 @@ owner-scoped assistant version selection remains enabled.
| **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) |
| **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}` - stream regular text and binary artifacts with `FileResponse`, including byte-`Range` 206/416 behavior used by bounded text previews and media seeking; active content (`text/html`, `text/xml`, `application/xml`, `text/xsl`, any `+xml` type such as XHTML/SVG; `.skill` members too) is always forced as a download attachment to reduce XSS risk; `?download=true` still forces download for other file types. `PUT /{path}` atomically replaces an existing UTF-8 text file under `/mnt/user-data/outputs` when its expected SHA-256 still matches; active runs conflict, and non-mounted sandbox providers receive the same update under a request lease. The outputs-only rule is `path_utils.resolve_outputs_confined_path`, shared with IM-channel attachment delivery: it collapses `..` before the prefix check and re-checks the resolved host path against the resolved outputs root, since `resolve_thread_virtual_path` only confines to `user-data/`; a percent-encoded `..` or a symlink planted in `outputs/` must not reach a sibling `uploads/` file. Atomic replacement applies the existing POSIX permission handling when descriptor-based APIs are available and otherwise keeps the platform-native temporary-file permissions (Windows). |
+| **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 |
| **Input Polish** (`/api/input-polish`) | `POST /` - rewrite a composer draft before it is sent. This is a short authenticated `runs:create` LLM request using `input_polish` config; it does not create a LangGraph run, persist a message, or modify thread state. Shares the non-graph one-shot LLM path (`deerflow.utils.oneshot_llm.run_oneshot_llm`) with the suggestions route so model build + Langfuse metadata + invoke stay in one place; validates the same stripped view of the draft it sends to the model, and preserves literal `` substrings in the rewrite (`strip_think_blocks(truncate_unclosed=False)`) |
| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, seed an empty feed from a checkpoint so legacy checkpoint-only history keeps its order and visibility; skip absent checkpoints or populated feeds. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, treating a protocol-valid hidden `human_input_response` as confirmed replay input while rejecting other hidden/control humans, preserving that response's hidden/correlation metadata, and carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - newest 100 runs as an array; `GET /page` - keyset history page `{data, has_more, next_before_created_at, next_before_run_id}`; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/stream` hides action/wait; GET action 405 pre-owner; POST needs `runs:cancel`; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET/POST /{rid}/artifacts/archive` - receipt manifest / bounded ZIP; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens plus an optional `context_usage` percentage. Context usage approximately counts messages from the latest materialized thread state through `build_thread_checkpoint_state_accessor`, so full and delta checkpoint modes expose the same input. The percentage uses the latest run's model and its `context_window`. |
diff --git a/backend/app/gateway/routers/artifacts.py b/backend/app/gateway/routers/artifacts.py
index 1ec4d40a1..5100eea66 100644
--- a/backend/app/gateway/routers/artifacts.py
+++ b/backend/app/gateway/routers/artifacts.py
@@ -25,6 +25,7 @@ from deerflow.config.paths import make_safe_user_id
from deerflow.runtime import ConflictError, ThreadOperationKind
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.sandbox.sandbox_provider import get_sandbox_provider
+from deerflow.utils.file_io import await_drained
from deerflow.utils.text_detection import _is_active_content_mime_type, is_text_file_by_content
from deerflow.utils.thread_id import ThreadId
@@ -152,6 +153,34 @@ def _sync_artifact_to_sandbox(sandbox, virtual_path: str, content: bytes) -> Non
sandbox.update_file(virtual_path, content)
+async def _commit_artifact_update(
+ *,
+ sandbox,
+ virtual_path: str,
+ actual_path: Path,
+ current: bytes,
+ updated: bytes,
+ file_stat: os.stat_result,
+) -> None:
+ """Keep remote/local artifact mutation ownership until commit or rollback."""
+ try:
+ if sandbox is not None:
+ await asyncio.to_thread(_sync_artifact_to_sandbox, sandbox, virtual_path, updated)
+ await asyncio.to_thread(_replace_artifact_atomically, actual_path, updated, file_stat)
+ except Exception:
+ # Non-cancelled failures are logged again by the outer route handler.
+ # Keep this inner log because await_drained re-raises caller cancellation
+ # after consuming the drained task's exception, which would otherwise make
+ # a cancelled-then-failed commit silent.
+ logger.exception("Failed to commit artifact update before rollback: %s", virtual_path)
+ if sandbox is not None:
+ try:
+ await asyncio.to_thread(_sync_artifact_to_sandbox, sandbox, virtual_path, current)
+ except Exception:
+ logger.exception("Failed to roll back remote artifact after artifact update failure: %s", virtual_path)
+ raise
+
+
def _build_content_disposition(disposition_type: str, filename: str) -> str:
"""Build an RFC 5987 encoded Content-Disposition header value."""
return f"{disposition_type}; filename*=UTF-8''{quote(filename)}"
@@ -491,11 +520,11 @@ async def update_artifact(
) -> ArtifactUpdateResponse:
"""Update an existing text artifact while the thread has no active run.
- The host-side artifact file is updated first; when the sandbox provider is
- not thread-mounted, the new content is also synced into the thread's
- sandbox. Under ``authorization.enabled``, a caller denied
- ``sandbox:execute`` skips that sandbox sync (the host-side update still
- completes).
+ For non-mounted providers, the sandbox copy is written before the host file
+ so a local replacement failure can restore the previous remote bytes. The
+ complete remote/local mutation is drained across caller cancellation before
+ either reservation is released. Under ``authorization.enabled``, a caller
+ denied ``sandbox:execute`` skips sandbox sync and updates only the host file.
"""
virtual_path = _normalize_editable_artifact_path(path)
raw_owner_user_id = get_trusted_internal_owner_user_id(request)
@@ -537,21 +566,20 @@ async def update_artifact(
if not sandbox_lease.denied and sandbox is None:
raise RuntimeError("Failed to acquire sandbox for artifact update")
- try:
- if sandbox is not None:
- await asyncio.to_thread(_sync_artifact_to_sandbox, sandbox, virtual_path, updated)
- await asyncio.to_thread(_replace_artifact_atomically, actual_path, updated, file_stat)
- # Invalidate any cached digest for this path so a subsequent GET
- # serves the fresh SHA-256. The (path, mtime_ns, size) LRU key can
- # collide on a same-size, sub-nanosecond re-write (review nit).
- _sha256_of_file_cached.cache_clear()
- except Exception:
- if sandbox is not None:
- try:
- await asyncio.to_thread(_sync_artifact_to_sandbox, sandbox, virtual_path, current)
- except Exception:
- logger.exception("Failed to roll back remote artifact after artifact update failure: %s", virtual_path)
- raise
+ # A cancelled request must not release the thread-operation reservation
+ # or sandbox request lease while either mutation is still running in a
+ # worker thread. Drain the complete remote/local transaction so it
+ # reaches a coherent commit or rollback before cancellation propagates.
+ await await_drained(
+ _commit_artifact_update(
+ sandbox=sandbox,
+ virtual_path=virtual_path,
+ actual_path=actual_path,
+ current=current,
+ updated=updated,
+ file_stat=file_stat,
+ )
+ )
except ConflictError:
raise HTTPException(status_code=409, detail="Thread has a run in flight. Save after the run finishes.") from None
except HTTPException:
diff --git a/backend/tests/test_artifacts_router.py b/backend/tests/test_artifacts_router.py
index 72b3c028f..6773a114c 100644
--- a/backend/tests/test_artifacts_router.py
+++ b/backend/tests/test_artifacts_router.py
@@ -1,6 +1,7 @@
import asyncio
import hashlib
import stat
+import threading
import zipfile
from contextlib import asynccontextmanager
from pathlib import Path
@@ -426,6 +427,102 @@ def test_update_artifact_rolls_back_remote_when_local_replace_fails(tmp_path, mo
assert artifact_path.read_text(encoding="utf-8") == "before"
+def test_update_artifact_cancellation_drains_remote_sync_before_releasing_write(tmp_path, monkeypatch) -> None:
+ artifact_path = tmp_path / "note.txt"
+ artifact_path.write_text("before", encoding="utf-8")
+ provider = _RemoteSandboxProvider()
+ _patch_artifact_update_dependencies(monkeypatch, artifact_path, provider)
+
+ sync_started = threading.Event()
+ allow_sync = threading.Event()
+ sync_finished = threading.Event()
+ original_sync = artifacts_router._sync_artifact_to_sandbox
+
+ def blocking_sync(sandbox, virtual_path: str, content: bytes) -> None:
+ if content == b"after":
+ sync_started.set()
+ assert allow_sync.wait(timeout=2)
+ try:
+ original_sync(sandbox, virtual_path, content)
+ finally:
+ if content == b"after":
+ sync_finished.set()
+
+ monkeypatch.setattr(artifacts_router, "_sync_artifact_to_sandbox", blocking_sync)
+
+ async def run_cancelled_update() -> None:
+ task = asyncio.create_task(
+ call_unwrapped(
+ artifacts_router.update_artifact,
+ "thread-1",
+ "mnt/user-data/outputs/note.txt",
+ artifacts_router.ArtifactUpdateRequest(content="after", expected_sha256=_artifact_sha256("before")),
+ _make_request(),
+ )
+ )
+ assert await asyncio.to_thread(sync_started.wait, 2)
+ task.cancel()
+ allow_sync.set()
+ assert await asyncio.to_thread(sync_finished.wait, 2)
+ with pytest.raises(asyncio.CancelledError):
+ await task
+
+ asyncio.run(run_cancelled_update())
+
+ # The coherent final remote/host state is the regression oracle. Merely
+ # observing that the task is still pending after one loop turn is not:
+ # the old rollback/release path also needed additional scheduling turns.
+ assert provider.sandbox.updates == [("/mnt/user-data/outputs/note.txt", b"after")]
+ assert artifact_path.read_text(encoding="utf-8") == "after"
+
+
+def test_update_artifact_logs_primary_failure_when_cancelled_commit_fails(tmp_path, monkeypatch, caplog) -> None:
+ artifact_path = tmp_path / "note.txt"
+ artifact_path.write_text("before", encoding="utf-8")
+ provider = _RemoteSandboxProvider()
+ _patch_artifact_update_dependencies(monkeypatch, artifact_path, provider)
+
+ sync_started = threading.Event()
+ allow_sync = threading.Event()
+ original_sync = artifacts_router._sync_artifact_to_sandbox
+
+ def blocking_failing_sync(sandbox, virtual_path: str, content: bytes) -> None:
+ if content == b"after":
+ sync_started.set()
+ assert allow_sync.wait(timeout=2)
+ raise RuntimeError("sandbox sync failed after cancellation")
+ original_sync(sandbox, virtual_path, content)
+
+ monkeypatch.setattr(artifacts_router, "_sync_artifact_to_sandbox", blocking_failing_sync)
+ caplog.set_level("ERROR", logger=artifacts_router.logger.name)
+
+ async def run_cancelled_failure() -> None:
+ task = asyncio.create_task(
+ call_unwrapped(
+ artifacts_router.update_artifact,
+ "thread-1",
+ "mnt/user-data/outputs/note.txt",
+ artifacts_router.ArtifactUpdateRequest(content="after", expected_sha256=_artifact_sha256("before")),
+ _make_request(),
+ )
+ )
+ assert await asyncio.to_thread(sync_started.wait, 2)
+ task.cancel()
+ await asyncio.sleep(0)
+ assert not task.done()
+
+ allow_sync.set()
+ with pytest.raises(asyncio.CancelledError):
+ await task
+
+ asyncio.run(run_cancelled_failure())
+
+ assert provider.sandbox.updates == [("/mnt/user-data/outputs/note.txt", b"before")]
+ assert provider.released == ["sandbox-1"]
+ assert artifact_path.read_text(encoding="utf-8") == "before"
+ assert any("Failed to commit artifact update before rollback" in record.getMessage() and record.exc_info is not None for record in caplog.records)
+
+
def test_update_artifact_rejects_oversized_content(tmp_path, monkeypatch) -> None:
artifact_path = tmp_path / "note.txt"
artifact_path.write_text("before", encoding="utf-8")