From d1f1c49dcd872aeb7aec6114ca36f6ee6760fed6 Mon Sep 17 00:00:00 2001 From: Jun <84921700+Amazingjun-j@users.noreply.github.com> Date: Mon, 7 Sep 2026 15:18:58 +0800 Subject: [PATCH] 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 --- backend/packages/harness/deerflow/AGENTS.md | 56 ++++---- .../deerflow/workspace_changes/recorder.py | 53 +++++--- .../test_workspace_changes_cancellation.py | 120 ++++++++++++++++++ 3 files changed, 184 insertions(+), 45 deletions(-) create mode 100644 backend/tests/blocking_io/test_workspace_changes_cancellation.py diff --git a/backend/packages/harness/deerflow/AGENTS.md b/backend/packages/harness/deerflow/AGENTS.md index 13cb70a2e..279542f2d 100644 --- a/backend/packages/harness/deerflow/AGENTS.md +++ b/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. - -**Architecture**: Imports the same `deerflow` modules that Gateway API uses. Shares the same config files and data directories. No FastAPI dependency. +`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. **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 -`TestGatewayConformance`), `tests/test_client_live.py` (live integration tests, -requires a root `config.yaml`, valid API credentials, and explicit opt-in via -`make test-live` or `DEER_FLOW_RUN_LIVE_TESTS=1`). The live suite calls real -external APIs and may incur API costs or create local sandboxes, artifacts, and -files. It is marked `live`, excluded from `make test`, and skipped in default -CI. +**Tests**: `tests/test_client.py` is offline, including `TestGatewayConformance`. +`tests/test_client_live.py` requires root `config.yaml`, valid API credentials, +and opt-in via `make test-live` or `DEER_FLOW_RUN_LIVE_TESTS=1`. It calls real +APIs (possible costs) and may create local sandboxes, artifacts, and files. +Marked `live`, it is 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. - -Each mount has these fixed limits: - -- 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. +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 +pass shares a 512 MiB / 2,000-file budget across skill projections and mounts. 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. ``result.truncated`` is ``True`` only when the upload pass -was stopped early by a resource limit (deadline, file count cap, or byte -budget). Individual mount failures (missing host path, SDK errors) are -logged but do NOT set ``truncated``. ``None`` on a reclaimed sandbox -means "not available" — the result was recorded at creation time and is -preserved within the same Gateway process lifetime via a provider-level -map. +After creation, ``E2BSandbox.mount_upload_result`` holds a ``MountUploadResult``. +``result.truncated`` is true only for resource-limit stops (deadline, file count, +bytes), not logged mount failures (missing paths, SDK errors). A provider-level +map preserves creation results within the Gateway process; ``None`` on a +reclaimed sandbox means unavailable. + +### Workspace Snapshot Cancellation (`workspace_changes/recorder.py`) + +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. diff --git a/backend/packages/harness/deerflow/workspace_changes/recorder.py b/backend/packages/harness/deerflow/workspace_changes/recorder.py index 67ca4573d..0439470b1 100644 --- a/backend/packages/harness/deerflow/workspace_changes/recorder.py +++ b/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: - scan.result() - except asyncio.CancelledError: - pass - except Exception: - pass - - if text_cache_dir is None: - return + _consume_cancelled_scan_outcome(scan, thread_id=thread_id) 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 - # this coroutine. Keep the cache alive until that worker has finished, - # then remove it before propagating cancellation. Repeated cancellation - # must not abandon either phase. - await _drain_scan_and_cleanup(scan, text_cache_dir) + # A metadata-only scan has no cache resource to protect. It still runs in + # the worker after caller cancellation, so retain ownership only long + # enough to consume/log its eventual outcome instead of delaying the + # cancellation until a full workspace scan finishes. + 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: diff --git a/backend/tests/blocking_io/test_workspace_changes_cancellation.py b/backend/tests/blocking_io/test_workspace_changes_cancellation.py new file mode 100644 index 000000000..589e41d9c --- /dev/null +++ b/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"