fix(workspace-changes): avoid draining metadata scans on cancellation (#5234)

* fix(workspace-changes): keep metadata cancellation responsive

* test(workspace-changes): cover metadata cancellation latency

* docs(workspace-changes): document cancellation ownership

* style(workspace-changes): format cancellation regressions

* docs(workspace-changes): remove unapproved nested guidance

* fix(workspace-changes): log only real cancellation drains

* style(workspace-changes): apply repository ruff format

* docs(harness): record workspace scan cancellation ownership

* docs: compact harness guidance below inherited size limit

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/packages/harness/deerflow/AGENTS.md

@@ -48,9 +48,7 @@ drift.
 
 ### Embedded Client (`packages/harness/deerflow/client.py`)
 
-`DeerFlowClient` provides direct in-process access to all DeerFlow capabilities without HTTP services. All return types align with the Gateway API response schemas, so consumer code works identically in HTTP and embedded modes.
+`DeerFlowClient` provides in-process access without HTTP or a FastAPI dependency. It shares Gateway's `deerflow` modules, config files, data directories, and response schemas for compatible consumers.
-
-**Architecture**: Imports the same `deerflow` modules that Gateway API uses. Shares the same config files and data directories. No FastAPI dependency.
 
 **Agent Conversation**:
 - `chat(message, thread_id)` — synchronous, accumulates streaming deltas per message-id and returns the final AI text
@@ -78,17 +76,15 @@ drift.
 | Uploads | `upload_files(thread_id, files)`, `list_uploads(thread_id)`, `delete_upload(thread_id, filename)` | `{"success": true, "files": [...]}`, `{"files": [...], "count": N}` |
 | Artifacts | `get_artifact(thread_id, path)` → `(bytes, mime_type)` | tuple |
 
-**Key difference from Gateway**: Upload accepts local `Path` objects instead of HTTP `UploadFile`, rejects directory paths before copying, and reuses a single worker when document conversion must run inside an active event loop. Artifact returns `(bytes, mime_type)` instead of HTTP Response. The new Gateway-only thread cleanup route deletes `.deer-flow/threads/{thread_id}` after LangGraph thread deletion; there is no matching `DeerFlowClient` method yet. `update_mcp_config()` and `update_skill()` automatically invalidate the cached agent.
+**Gateway differences**: Upload takes local `Path`, not `UploadFile`, rejects directories before copying, and reuses one conversion worker inside an active event loop. Artifacts return `(bytes, mime_type)`, not HTTP Response. Gateway alone deletes `.deer-flow/threads/{thread_id}` after LangGraph thread deletion; the client has no equivalent. `update_mcp_config()` and `update_skill()` invalidate the cached agent.
 
-**Tests**: `tests/test_client.py` (offline unit tests including
+**Tests**: `tests/test_client.py` is offline, including `TestGatewayConformance`.
-`TestGatewayConformance`), `tests/test_client_live.py` (live integration tests,
+`tests/test_client_live.py` requires root `config.yaml`, valid API credentials,
-requires a root `config.yaml`, valid API credentials, and explicit opt-in via
+and opt-in via `make test-live` or `DEER_FLOW_RUN_LIVE_TESTS=1`. It calls real
-`make test-live` or `DEER_FLOW_RUN_LIVE_TESTS=1`). The live suite calls real
+APIs (possible costs) and may create local sandboxes, artifacts, and files.
-external APIs and may incur API costs or create local sandboxes, artifacts, and
+Marked `live`, it is excluded from `make test` and skipped in default CI.
-files. It is marked `live`, excluded from `make test`, and skipped in default
-CI.
 
-**Gateway Conformance Tests** (`TestGatewayConformance`): Validate that every dict-returning client method conforms to the corresponding Gateway Pydantic response model. Each test parses the client output through the Gateway model — if Gateway adds a required field that the client doesn't provide, Pydantic raises `ValidationError` and CI catches the drift. Covers: `ModelsListResponse`, `ModelResponse`, `SkillsListResponse`, `SkillResponse`, `SkillInstallResponse`, `McpConfigResponse`, `UploadResponse`, `MemoryConfigResponse`, `MemoryStatusResponse`.
+**Gateway Conformance Tests** (`TestGatewayConformance`): Parse every dict-returning client method's output through its Gateway Pydantic model so missing required fields raise `ValidationError` in CI. Covers: `ModelsListResponse`, `ModelResponse`, `SkillsListResponse`, `SkillResponse`, `SkillInstallResponse`, `McpConfigResponse`, `UploadResponse`, `MemoryConfigResponse`, `MemoryStatusResponse`.
 
 ### AIO Sandbox Network Policy
 
@@ -104,16 +100,9 @@ Destroy the sandbox, sidecar, and both networks together.
 
 ### E2B Mount Uploads
 
-The E2B provider uploads host mounts during sandbox creation. It passes binary file objects to the E2B SDK.
+E2B uploads host mounts during sandbox creation using binary file objects.
-
+Per-mount limits: 100 MiB/file, 512 MiB total, 2,000 files. The full creation
-Each mount has these fixed limits:
+pass shares a 512 MiB / 2,000-file budget across skill projections and mounts.
-
-- 100 MiB for one file.
-- 512 MiB for all files.
-- 2,000 files.
-
-The full sandbox creation pass also allows 512 MiB and 2,000 files. Skill
-projections and configured mounts share this budget.
 
 The pass has a cooperative deadline controlled by
 ``mount_upload_deadline_seconds`` (default: 120 seconds). The provider checks it before
@@ -135,11 +124,18 @@ Each successful upload logs its source, destination, file count, byte count, and
 
 A stopped pass logs its limit reason and elapsed time. It reports attempted and completed upload totals separately.
 
-A ``MountUploadResult`` is attached to ``E2BSandbox.mount_upload_result``
+After creation, ``E2BSandbox.mount_upload_result`` holds a ``MountUploadResult``.
-after creation. ``result.truncated`` is ``True`` only when the upload pass
+``result.truncated`` is true only for resource-limit stops (deadline, file count,
-was stopped early by a resource limit (deadline, file count cap, or byte
+bytes), not logged mount failures (missing paths, SDK errors). A provider-level
-budget).  Individual mount failures (missing host path, SDK errors) are
+map preserves creation results within the Gateway process; ``None`` on a
-logged but do NOT set ``truncated``.  ``None`` on a reclaimed sandbox
+reclaimed sandbox means unavailable.
-means "not available" — the result was recorded at creation time and is
+
-preserved within the same Gateway process lifetime via a provider-level
+### Workspace Snapshot Cancellation (`workspace_changes/recorder.py`)
-map.
+
+After `_prepare_capture()` hands off roots, cancellation must drain text scans
+(`include_text=True`) before removing the cache the worker may still access.
+Metadata scans (`include_text=False`) own no cache: cancel promptly, let the worker
+continue, and consume/log its outcome in a completion callback. Prepare-stage
+cancellation retains its handoff/reclaim path. Regressions in
+`tests/blocking_io/test_workspace_changes_cancellation.py` must cover prompt
+metadata cancellation and text-cache drain/cleanup.

backend/packages/harness/deerflow/workspace_changes/recorder.py

@@ -4,6 +4,7 @@ import asyncio
 import logging
 import shutil
 import tempfile
+from functools import partial
 from pathlib import Path
 from typing import Any
 
@@ -75,11 +76,33 @@ async def _reclaim_prepare_and_cleanup(prepare: asyncio.Future[tuple[list[Worksp
         await _remove_text_cache_dir(orphaned)
 
 
+def _consume_cancelled_scan_outcome(scan: asyncio.Future[WorkspaceSnapshot], *, thread_id: str) -> None:
+    """Consume a completed scan and retain diagnostics after caller cancellation."""
+    try:
+        scan.result()
+    except asyncio.CancelledError:
+        return
+    except Exception:
+        logger.warning(
+            "Workspace scan failed after snapshot cancellation for thread %s",
+            thread_id,
+            exc_info=True,
+        )
+
+
 async def _drain_scan_and_cleanup(
     scan: asyncio.Future[WorkspaceSnapshot],
-    text_cache_dir: Path | None,
+    text_cache_dir: Path,
+    *,
+    thread_id: str,
 ) -> None:
     """Let a cancelled scan finish before removing the cache it may still use."""
+    if not scan.done():
+        logger.info(
+            "Waiting for cancelled workspace snapshot scan to finish before text-cache cleanup for thread %s",
+            thread_id,
+        )
+
     while not scan.done():
         try:
             await asyncio.shield(scan)
@@ -90,15 +113,7 @@ async def _drain_scan_and_cleanup(
 
     # Cancellation remains the caller-visible outcome, but consume any late scan
     # failure so the drained task cannot emit an un-retrieved exception warning.
-    try:
+    _consume_cancelled_scan_outcome(scan, thread_id=thread_id)
-        scan.result()
-    except asyncio.CancelledError:
-        pass
-    except Exception:
-        pass
-
-    if text_cache_dir is None:
-        return
 
     cleanup = asyncio.create_task(_remove_text_cache_dir(text_cache_dir))
     while not cleanup.done():
@@ -151,11 +166,19 @@ async def capture_workspace_snapshot(
     try:
         return await asyncio.shield(scan)
     except asyncio.CancelledError:
-        # The scan runs in a worker thread and cannot be stopped by cancelling
+        # A metadata-only scan has no cache resource to protect. It still runs in
-        # this coroutine. Keep the cache alive until that worker has finished,
+        # the worker after caller cancellation, so retain ownership only long
-        # then remove it before propagating cancellation. Repeated cancellation
+        # enough to consume/log its eventual outcome instead of delaying the
-        # must not abandon either phase.
+        # cancellation until a full workspace scan finishes.
-        await _drain_scan_and_cleanup(scan, text_cache_dir)
+        if text_cache_dir is None:
+            scan.add_done_callback(partial(_consume_cancelled_scan_outcome, thread_id=thread_id))
+            raise
+
+        # Text capture is different: the worker may still read/write the cache,
+        # so deleting it immediately would race the scan. Keep the cache alive
+        # until the worker drains, then remove it before propagating cancellation.
+        # Repeated cancellation must not abandon either phase.
+        await _drain_scan_and_cleanup(scan, text_cache_dir, thread_id=thread_id)
         raise
     except Exception:
         if text_cache_dir is not None:

backend/tests/blocking_io/test_workspace_changes_cancellation.py

@@ -0,0 +1,120 @@
+"""Cancellation regressions for workspace snapshot scans.
+
+Text snapshots own a temporary cache that must outlive an already-running scan,
+so cancellation deliberately drains that worker before cleanup. Metadata-only
+snapshots own no such resource and must propagate cancellation promptly while
+still consuming/logging the worker's eventual outcome.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import tempfile
+import threading
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from deerflow.workspace_changes import recorder
+from deerflow.workspace_changes.types import WorkspaceSnapshot
+
+pytestmark = pytest.mark.asyncio
+
+
+async def _reset_paths(tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
+    import deerflow.config.paths as paths_mod
+
+    monkeypatch.setattr(paths_mod, "_paths", None)
+
+
+async def test_metadata_only_cancel_does_not_wait_for_scan_worker(tmp_path: Path, monkeypatch, caplog) -> None:
+    """No text cache means cancellation must not wait for the worker scan."""
+    await _reset_paths(tmp_path, monkeypatch)
+
+    entered = threading.Event()
+    release = threading.Event()
+    finished = threading.Event()
+
+    def _blocking_scan(
+        *_args: Any,
+        text_cache_dir: str | Path | None = None,
+        **_kwargs: Any,
+    ) -> WorkspaceSnapshot:
+        assert text_cache_dir is None
+        entered.set()
+        release.wait(timeout=5)
+        finished.set()
+        raise RuntimeError("late metadata scan failure")
+
+    monkeypatch.setattr(recorder, "scan_workspace_roots", _blocking_scan)
+    caplog.set_level(logging.INFO, logger=recorder.__name__)
+
+    task = asyncio.create_task(recorder.capture_workspace_snapshot("t1", include_text=False))
+    assert await asyncio.to_thread(entered.wait, 5), "metadata scan worker did not start"
+
+    try:
+        task.cancel()
+        for _ in range(5):
+            await asyncio.sleep(0)
+        assert task.done(), "metadata-only cancellation waited for a scan with no cache resource to protect"
+        with pytest.raises(asyncio.CancelledError):
+            await task
+    finally:
+        release.set()
+
+    assert await asyncio.to_thread(finished.wait, 5), "metadata scan worker did not finish after release"
+    for _ in range(100):
+        if any("Workspace scan failed after snapshot cancellation" in record.getMessage() for record in caplog.records):
+            break
+        await asyncio.sleep(0.01)
+
+    assert any("Workspace scan failed after snapshot cancellation" in record.getMessage() for record in caplog.records), "the detached metadata scan's late failure must be consumed and logged"
+
+
+async def test_text_scan_cancel_logs_drain_and_late_failure(tmp_path: Path, monkeypatch, caplog) -> None:
+    """A text-cache scan still drains, and that cancellation latency is observable."""
+    await _reset_paths(tmp_path, monkeypatch)
+
+    cache_root = tmp_path / "tmp"
+    cache_root.mkdir()
+    monkeypatch.setattr(tempfile, "tempdir", str(cache_root))
+
+    entered = threading.Event()
+    release = threading.Event()
+
+    def _blocking_scan(
+        *_args: Any,
+        text_cache_dir: str | Path | None = None,
+        **_kwargs: Any,
+    ) -> WorkspaceSnapshot:
+        assert text_cache_dir is not None
+        cache_dir = Path(text_cache_dir)
+        assert cache_dir.exists()
+        entered.set()
+        release.wait(timeout=5)
+        assert cache_dir.exists(), "text cache was removed while the scan worker was still running"
+        raise RuntimeError("late text scan failure")
+
+    monkeypatch.setattr(recorder, "scan_workspace_roots", _blocking_scan)
+    caplog.set_level(logging.INFO, logger=recorder.__name__)
+
+    task = asyncio.create_task(recorder.capture_workspace_snapshot("t1", include_text=True))
+    assert await asyncio.to_thread(entered.wait, 5), "text scan worker did not start"
+
+    task.cancel()
+    for _ in range(5):
+        await asyncio.sleep(0)
+
+    assert not task.done(), "text-cache cancellation must keep ownership until the scan drains"
+    assert any("Waiting for cancelled workspace snapshot scan to finish before text-cache cleanup" in record.getMessage() for record in caplog.records), "entering the cancellation drain should be observable"
+
+    release.set()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    leftovers = await asyncio.to_thread(lambda: sorted(cache_root.glob("deerflow-workspace-changes-*")))
+    assert leftovers == [], f"cancelled text scan leaked a cache dir: {leftovers}"
+    assert any("Workspace scan failed after snapshot cancellation" in record.getMessage() for record in caplog.records), "a scan failure during cancellation drain must retain diagnostics"