From b5d921c25556f29de93e6d665d81c6bbbf80e2e3 Mon Sep 17 00:00:00 2001 From: IronMurphy Date: Thu, 8 Oct 2026 21:30:07 +0800 Subject: [PATCH] fix(scanner): cache decoded text atomically (#6510) * fix(scanner): cache decoded text atomically _cache_text_file keyed the cache on the virtual path and returned the path, and the caller serves that path as the file's content, so a write that failed part-way left a truncated entry that every later read handed back as the preview. Publish through a temporary file plus os.replace. * test(scanner): pin the cache entry against a failed publish The helper-level case cannot distinguish the call site, so this drives the real entry point with a failing publish and asserts the cached entry is untouched. * docs(sandbox): keep agent guidance within its size limit * fix(scanner): preserve publish errors during cleanup --------- Co-authored-by: Willem Jiang --- .../harness/deerflow/sandbox/AGENTS.md | 2 +- .../deerflow/workspace_changes/scanner.py | 21 +++- backend/tests/AGENTS.md | 7 ++ .../tests/test_scanner_text_cache_atomic.py | 115 ++++++++++++++++++ 4 files changed, 143 insertions(+), 2 deletions(-) create mode 100644 backend/tests/test_scanner_text_cache_atomic.py diff --git a/backend/packages/harness/deerflow/sandbox/AGENTS.md b/backend/packages/harness/deerflow/sandbox/AGENTS.md index 340c092bb7..33f940e88e 100644 --- a/backend/packages/harness/deerflow/sandbox/AGENTS.md +++ b/backend/packages/harness/deerflow/sandbox/AGENTS.md @@ -96,7 +96,7 @@ cancellation before propagating it. Health probes require an exact `ok` line and reject `Error:`/`Exit Code:` diagnostic lines; other login-shell stdout/stderr is tolerated. Failures log sandbox id and output before eviction. -**Shared warm-pool lifecycle:** community sandbox providers that keep released sandboxes alive for fast reuse share `deerflow.community.warm_pool_lifecycle.WarmPoolLifecycleMixin`. The mixin owns the common `DEFAULT_IDLE_TIMEOUT=600`, `IDLE_CHECK_INTERVAL=60`, `DEFAULT_REPLICAS=3`, idle-checker loop, warm-pool expiry, oldest-warm eviction, replica counting, and soft-cap logging. Providers remain responsible for their own active registries, creation/discovery, health checks, and destroy hook (`_destroy_warm_entry`): AIO destroys `SandboxInfo` through its backend; Boxlite closes loop-affine `BoxliteBox` handles; Tenki closes the microVM session (`TenkiSandbox.close`, which terminates the remote sandbox). AIO keeps active-idle cleanup outside the mixin and delegates only warm-pool expiry to the shared helper. `_stop_idle_checker()` must confirm thread exit before provider teardown. A bounded-join timeout raises; reset the provider shutdown guard so shutdown can retry. +**Shared warm-pool lifecycle:** `deerflow.community.warm_pool_lifecycle.WarmPoolLifecycleMixin` owns `DEFAULT_IDLE_TIMEOUT=600`, `IDLE_CHECK_INTERVAL=60`, `DEFAULT_REPLICAS=3`, the idle-checker loop, warm expiry, oldest-warm eviction, replica counts and soft-cap logs. Providers own active registries, creation/discovery, health checks and `_destroy_warm_entry`: AIO destroys `SandboxInfo` via its backend; Boxlite closes loop-affine `BoxliteBox`; Tenki terminates the remote VM with `TenkiSandbox.close`. AIO delegates only warm expiry; active-idle cleanup stays outside the mixin. Shutdown must await the idle-checker thread before tearing down provider resources. `_stop_idle_checker()` uses a bounded join and raises on timeout; reset the provider's shutdown guard on failure so shutdown can retry. **Virtual Path System**: - Agent sees: `/mnt/user-data/{workspace,uploads,outputs}`, `/mnt/skills` diff --git a/backend/packages/harness/deerflow/workspace_changes/scanner.py b/backend/packages/harness/deerflow/workspace_changes/scanner.py index f4c28931ac..33bfe6750d 100644 --- a/backend/packages/harness/deerflow/workspace_changes/scanner.py +++ b/backend/packages/harness/deerflow/workspace_changes/scanner.py @@ -3,6 +3,7 @@ from __future__ import annotations import fnmatch import hashlib import os +import tempfile from codecs import BOM_UTF16_BE, BOM_UTF16_LE, getincrementaldecoder from pathlib import Path @@ -323,10 +324,28 @@ def _snapshot_symlink(root: WorkspaceRoot, host_file: Path) -> FileSnapshot | No ) +def _publish_text_atomically(path: Path, content: str) -> None: + """Publish text through a temporary file so a partial write cannot survive.""" + fd, temporary = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(content) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, path) + except BaseException: + try: + Path(temporary).unlink(missing_ok=True) + except OSError: + pass + raise + + def _cache_text_file(text: str, virtual_path: str, cache_dir: Path) -> Path: + """Cache decoded text atomically; the caller serves this path as file content.""" cache_name = hashlib.sha256(virtual_path.encode("utf-8")).hexdigest() target = cache_dir / cache_name - target.write_text(text, encoding="utf-8") + _publish_text_atomically(target, text) return target diff --git a/backend/tests/AGENTS.md b/backend/tests/AGENTS.md index cdf2c9dc08..4f4bcd2aff 100644 --- a/backend/tests/AGENTS.md +++ b/backend/tests/AGENTS.md @@ -199,3 +199,10 @@ or when the filesystem does not enforce those bits. Cover both persisted secrets, shell/Compose dotenv overrides, readable read-only secrets, and teardown without secret exports. `down` must not probe, read, or generate secrets. Writable-directory cases remain portable. + +## Workspace text cache + +`test_scanner_text_cache_atomic.py` verifies complete publication and preservation +of an existing entry on failure. A cleanup failure must preserve the original +publish exception, including interruption. Import the scanner during fixture +setup, after the autouse fixtures initialize runtime, to avoid the package cycle. diff --git a/backend/tests/test_scanner_text_cache_atomic.py b/backend/tests/test_scanner_text_cache_atomic.py new file mode 100644 index 0000000000..12f010178b --- /dev/null +++ b/backend/tests/test_scanner_text_cache_atomic.py @@ -0,0 +1,115 @@ +# SPDX-License-Identifier: MIT +"""A cached preview must never be served half written.""" + +from __future__ import annotations + +import hashlib +import os +from pathlib import Path + +import pytest + + +@pytest.fixture +def scanner(): + # Import after the autouse fixtures initialize the runtime package. + from deerflow.workspace_changes import scanner + + return scanner + + +def test_publishes_the_text_and_leaves_no_temp_file(tmp_path, scanner): + target = tmp_path / "entry" + + scanner._publish_text_atomically(target, "decoded body") + + assert target.read_text(encoding="utf-8") == "decoded body" + assert list(tmp_path.glob("*.tmp")) == [] + + +def test_a_failed_publish_keeps_the_previous_entry(tmp_path, monkeypatch, scanner): + target = tmp_path / "entry" + target.write_text("previous body", encoding="utf-8") + + def _boom(*args, **kwargs): + raise OSError("disk full") + + monkeypatch.setattr(os, "replace", _boom) + + with pytest.raises(OSError): + scanner._publish_text_atomically(target, "new body") + + assert target.read_text(encoding="utf-8") == "previous body" + assert list(tmp_path.glob("*.tmp")) == [] + + +@pytest.mark.parametrize("publish_error", [OSError("publish failed"), KeyboardInterrupt("publish interrupted")], ids=["io-error", "interrupt"]) +def test_cleanup_failure_preserves_the_publish_error(tmp_path, monkeypatch, scanner, publish_error): + target = tmp_path / "entry" + target.write_text("previous body", encoding="utf-8") + + def _fail_publish(*args, **kwargs): + raise publish_error + + def _fail_cleanup(*args, **kwargs): + raise PermissionError("cleanup failed") + + with monkeypatch.context() as patches: + patches.setattr(os, "replace", _fail_publish) + patches.setattr(Path, "unlink", _fail_cleanup) + + with pytest.raises(type(publish_error)) as caught: + scanner._publish_text_atomically(target, "new body") + + assert caught.value is publish_error + assert target.read_text(encoding="utf-8") == "previous body" + + +def test_truncating_write_would_have_destroyed_it(tmp_path): + """Control: the previous implementation loses the entry in the same failure.""" + target = tmp_path / "entry" + target.write_text("previous body", encoding="utf-8") + + try: + with open(target, "w", encoding="utf-8") as handle: + handle.write("new ") + raise OSError("disk full") + except OSError: + pass + + assert target.read_text(encoding="utf-8") != "previous body" + + +def test_cache_text_file_returns_a_complete_entry(tmp_path, scanner): + cache_dir = tmp_path / "cache" + cache_dir.mkdir() + virtual_path = "/workspace/report.csv" + + returned = scanner._cache_text_file("a,b\n1,2\n", virtual_path, cache_dir) + + assert Path(returned).read_text(encoding="utf-8") == "a,b\n1,2\n" + # The key is derived from the virtual path, so a partial entry would be served + # as the file's content on every later read. + assert Path(returned).name == hashlib.sha256(virtual_path.encode()).hexdigest() + + +def test_cache_entry_survives_a_failed_publish(tmp_path, monkeypatch, scanner): + """The real entry point: a failed publish must not damage the cached entry.""" + cache_dir = tmp_path / "cache" + cache_dir.mkdir() + virtual_path = "/workspace/report.csv" + name = hashlib.sha256(virtual_path.encode()).hexdigest() + entry = cache_dir / name + entry.write_text("previous body", encoding="utf-8") + + def _boom(*args, **kwargs): + raise OSError("disk full") + + monkeypatch.setattr(os, "replace", _boom) + + with pytest.raises(OSError): + scanner._cache_text_file("new body", virtual_path, cache_dir) + + # A truncating write would have emptied the entry before failing, and the + # caller serves this path as the file's content on every later read. + assert entry.read_text(encoding="utf-8") == "previous body"