fix(workspace): don't infer created or deleted files from truncated snapshots (#7172)

* fix(workspace): don't infer created or deleted files from truncated snapshots

A bounded workspace scan stops at max_scanned_files and marks the snapshot
truncated. Absence from such a snapshot is unknown, so creation is only
inferred from a complete before snapshot and deletion only from a complete
after snapshot. get_changed_paths() is the single filter for the summary,
text capture and output-delivery detection.

The scanner now walks subdirectories in sorted order so the truncation
boundary is reproducible across the before and after scans, and the docs
note that the budget is consumed root by root (workspace before outputs).

Supersedes #5231.

Co-authored-by: liutian <91778998+ImcLiuQian@users.noreply.github.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(runtime): record truncated delivery verification skips

Record verification_skipped=truncated_pre_run_snapshot in run.delivery
receipts when a truncated pre-run scan leaves no provable output changes.
Carry the snapshot into both normal and exceptional receipt finalization.

Keep observed-output verification and terminal statuses unchanged. Cover
success, failure, cancellation and capture failures through the real worker,
and document the per-run receipt marker.

---------

Co-authored-by: liutian <91778998+ImcLiuQian@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -2587,6 +2587,23 @@ storage permissions. An unclean shutdown can leave `.tool-output-*.tmp` files in
 Remove leftovers during thread-data maintenance with all Gateway writers stopped,
 or when deleting the corresponding inactive thread's data.
 
+Workspace change summaries scan up to 2,000 files by default, in sorted order:
+the workspace root first, then outputs. A workspace that alone reaches the limit
+leaves outputs unscanned. When a scan is incomplete, files missing from that
+partial snapshot are not reported as created or deleted. Changes to files
+observed in both snapshots are still reported, and the summary is marked as
+truncated; if no change can be established, no summary is recorded.
+
+Output-delivery verification uses these same snapshots. When the pre-run scan
+is truncated, genuinely new outputs can be omitted even if the post-run scan
+sees them. If no output changes can be established, the run does not fail with
+"Artifact delivery incomplete". If the pre-run scan was truncated, its persisted
+`run.delivery` receipt records `verification_skipped: "truncated_pre_run_snapshot"`
+instead of an output-verification verdict. This distinguishes an incomplete scan
+from a complete scan with no produced outputs; it does not prove that no outputs
+were produced. Outputs modified and observed in both snapshots still participate
+in that check.
+
 `E2BSandboxProvider` uses `wait` as its default overflow policy. It waits for
 `acquire_timeout`, then fails the agent turn. DeerFlow does not retry the turn
 automatically. Clients can use the structured error to schedule a retry.

backend/docs/runtime-guidance-details.md

@@ -108,6 +108,30 @@ been admitted, the old completion leaves the goal alone. This allows duration
 bookkeeping while preserving newer goal or conversation changes. Delivery
 failure does not itself queue a hidden continuation.
 
+**Workspace change detection** (`workspace_changes/diff.py`, `scanner.py`):
+`get_changed_paths()` owns change filtering for summaries, text capture, and
+output delivery detection. Absence from a truncated snapshot is unknown:
+require a complete before snapshot to infer creation and a complete after
+snapshot to infer deletion. Keep comparing files observed in both snapshots and
+preserve the summary's truncation flag. The scanner walks directories and files
+in sorted order so the truncation boundary is reproducible across the before
+and after scans. Its budget is shared and consumed root by root (workspace
+before outputs), so a workspace at the limit starves outputs.
+`get_changed_output_paths()` inherits this uncertainty: newly observed outputs
+are suppressed when the before snapshot was truncated, while modifications
+observed in both snapshots remain eligible. The worker passes this list through
+`_produced_output_paths()` to its delivery check; an empty list leaves
+`produced_paths` and `verification` absent from the delivery receipt, and
+`_delivery_error()` returns no error. When that empty list accompanies a truncated
+pre-run snapshot, the receipt includes
+`verification_skipped: "truncated_pre_run_snapshot"` on both normal completion and
+failure/cancellation cleanup. This marker explains the skipped check without
+asserting that no outputs were produced. Observed modifications still receive the
+ordinary verification verdict, without a skip marker. A missing snapshot or
+capture error alone does not imply this skip reason. These cases are pinned
+through real snapshot capture and `run_agent()` in
+`tests/test_run_worker_delivery.py`.
+
 **Extension changed-run discovery** (`runtime/runs/store/` and
 `extensions/run_evidence.py`) orders public run-record changes by the
 backend-owned `(change_seq, run_id)` key rather than timestamps or per-thread

backend/packages/harness/deerflow/runtime/AGENTS.md

@@ -114,6 +114,8 @@ answers for paginated history. See `docs/skill-usage-ui.md`.
 
 **Run delivery receipts:** Finalize artifact evidence and terminal status before clearing an ordinary satisfied goal, under a checkpoint-write reservation. Delivery failure retains the goal without continuing. Only that clear writes `goal_outcome`; other goal writes remove it. Historical replay keeps the current goal and budget, and drops the old outcome. Scheduled exception below; details: `backend/docs/runtime-guidance-details.md`.
 
+**Workspace change detection:** Truncated snapshots cannot prove absence. Mark skipped delivery verification in the receipt. Details: `backend/docs/runtime-guidance-details.md`.
+
 Unmet goals stand down on run-level token/loop hard stops and retain the original stop reason; hidden continuations must not re-enter that run's shared safety state.
 
 **Deferred terminal commit:** With an event store, the worker stages its terminal

backend/packages/harness/deerflow/runtime/runs/worker.py

@@ -361,9 +361,13 @@ def _presented_path_covers_output(presented_path: str, produced_path: str) -> bo
 def _delivery_content_with_outputs(
     content: dict[str, Any],
     produced_paths: list[str],
+    *,
+    pre_run_snapshot: WorkspaceSnapshot | None = None,
 ) -> dict[str, Any]:
-    """Attach a delivery verdict when this run created or modified outputs."""
+    """Attach a delivery verdict or explain why a truncated scan skipped it."""
     if not produced_paths:
+        if pre_run_snapshot is not None and pre_run_snapshot.truncated:
+            return {**content, "verification_skipped": "truncated_pre_run_snapshot"}
         return content
 
     presented_paths = content.get("by_tool", {}).get("present_files", [])
@@ -1726,6 +1730,7 @@ async def run_agent(
             delivery_content = _delivery_content_with_outputs(
                 journal.get_delivery_content() if journal is not None else _empty_delivery_content(),
                 produced_output_paths,
+                pre_run_snapshot=pre_run_workspace_snapshot,
             )
             delivery_error = _delivery_error(delivery_content)
             cancel_action = await run_manager.set_status_if_not_cancelled(
@@ -1852,7 +1857,7 @@ async def run_agent(
                     if delivery_content is None:
                         if produced_output_paths is None:
                             produced_output_paths = await _produced_output_paths(pre_run_workspace_snapshot, thread_id=thread_id, user_id=workspace_changes_user_id, extra_excluded_dir_names=workspace_excluded_dir_names)
-                        delivery_content = _delivery_content_with_outputs(journal.get_delivery_content(), produced_output_paths)
+                        delivery_content = _delivery_content_with_outputs(journal.get_delivery_content(), produced_output_paths, pre_run_snapshot=pre_run_workspace_snapshot)
                     receipt_persisted = await _persist_delivery_receipt(event_store, thread_id=thread_id, run_id=run_id, content=delivery_content)
                     if produced_output_paths and record.status == RunStatus.success and not receipt_persisted:
                         await run_manager.set_status(run_id, RunStatus.error, error=_DELIVERY_RECEIPT_FAILED_ERROR, persist=False)

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

@@ -21,17 +21,14 @@ def compare_snapshots(
     limits: WorkspaceChangeLimits | None = None,
 ) -> WorkspaceChangeResult:
     resolved_limits = limits or WorkspaceChangeLimits()
-    all_paths = sorted(set(before.files) | set(after.files))
     changes: list[WorkspaceFileChange] = []
     created = modified = deleted = symlink_created = additions = deletions = 0
     total_diff_bytes = 0
     truncated = before.truncated or after.truncated
 
-    for path in all_paths:
+    for path in sorted(get_changed_paths(before, after)):
         before_file = before.files.get(path)
         after_file = after.files.get(path)
-        if before_file and after_file and _same_file(before_file, after_file):
-            continue
 
         status = _status(before_file, after_file)
         if status == "created":
@@ -103,6 +100,9 @@ def get_changed_paths(before: WorkspaceSnapshot, after: WorkspaceSnapshot) -> se
     for path in set(before.files) | set(after.files):
         before_file = before.files.get(path)
         after_file = after.files.get(path)
+        # Absence from a partial scan does not prove creation or deletion.
+        if (before_file is None and before.truncated) or (after_file is None and after.truncated):
+            continue
         if before_file and after_file and _same_file(before_file, after_file):
             continue
         changed.add(path)
@@ -110,7 +110,13 @@ def get_changed_paths(before: WorkspaceSnapshot, after: WorkspaceSnapshot) -> se
 
 
 def get_changed_output_paths(before: WorkspaceSnapshot, after: WorkspaceSnapshot) -> list[str]:
-    """Return created or modified regular files under the outputs root."""
+    """Return regular output changes established by the bounded snapshots.
+
+    Paths absent from a truncated before snapshot are omitted, including
+    genuinely new files; modifications observed in both snapshots are retained.
+    The worker's delivery check relies on this list and is skipped when it is
+    empty, so an empty result does not prove that the run produced no outputs.
+    """
     paths: list[str] = []
     for path in sorted(get_changed_paths(before, after)):
         snapshot = after.files.get(path)

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

@@ -137,7 +137,7 @@ def scan_workspace_roots(
             continue
 
         for dirpath, dirnames, filenames in os.walk(root.host_path, followlinks=False):
-            dirnames[:] = [dirname for dirname in dirnames if dirname not in excluded_dir_names and not (Path(dirpath) / dirname).is_symlink()]
+            dirnames[:] = [dirname for dirname in sorted(dirnames) if dirname not in excluded_dir_names and not (Path(dirpath) / dirname).is_symlink()]
             for filename in sorted(filenames):
                 if scanned >= resolved_limits.max_scanned_files:
                     truncated = True

backend/tests/test_run_worker_delivery.py

@@ -19,6 +19,7 @@ from deerflow.runtime.runs.schemas import RunStatus
 from deerflow.runtime.runs.store.memory import MemoryRunStore
 from deerflow.runtime.runs.worker import RunContext, _delivery_content_with_outputs, run_agent
 from deerflow.runtime.user_context import get_effective_user_id
+from deerflow.workspace_changes import WorkspaceChangeLimits, capture_workspace_snapshot
 from deerflow.workspace_changes.api import get_workspace_changes_response
 
 
@@ -279,6 +280,123 @@ async def test_changed_outputs_fail_closed_when_not_presented(monkeypatch):
     assert record.stop_reason is None
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize("modify_existing", [False, True], ids=["unproven-new-output", "observed-modified-output"])
+@pytest.mark.parametrize("terminal_outcome", ["success", "error", "cancelled"])
+async def test_truncated_snapshots_limit_output_delivery_verification(tmp_path, monkeypatch, modify_existing, terminal_outcome):
+    paths = Paths(base_dir=tmp_path)
+    monkeypatch.setattr("deerflow.workspace_changes.recorder.get_paths", lambda: paths)
+    run_manager = RunManager()
+    record = await run_manager.create("thread-1")
+    store = MemoryRunEventStore()
+    outputs = paths.sandbox_outputs_dir("thread-1", user_id=get_effective_user_id())
+    outputs.mkdir(parents=True)
+    for name in ["b.txt", "c.txt", "d.txt"]:
+        (outputs / name).write_text("before\n", encoding="utf-8")
+
+    snapshots = []
+
+    async def capture_bounded_snapshot(*args, **kwargs):
+        snapshot = await capture_workspace_snapshot(*args, limits=WorkspaceChangeLimits(max_scanned_files=2), **kwargs)
+        snapshots.append(snapshot)
+        return snapshot
+
+    monkeypatch.setattr("deerflow.runtime.runs.worker.capture_workspace_snapshot", capture_bounded_snapshot)
+
+    class ProducingAgent:
+        async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False):
+            (outputs / "a.txt").write_text("new output\n", encoding="utf-8")
+            if modify_existing:
+                (outputs / "b.txt").write_text("modified\n", encoding="utf-8")
+            if terminal_outcome == "error":
+                raise RuntimeError("Agent failed after producing output")
+            if terminal_outcome == "cancelled":
+                raise asyncio.CancelledError
+            yield {"messages": [AIMessage(content="Created the output.")]}
+
+    await run_agent(
+        _make_bridge(),
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=None, event_store=store),
+        agent_factory=lambda *, config: ProducingAgent(),
+        graph_input={},
+        config={},
+    )
+
+    assert snapshots[0].truncated is True
+    assert snapshots[-1].truncated is True
+    assert "/mnt/user-data/outputs/a.txt" not in snapshots[0].files
+    assert "/mnt/user-data/outputs/a.txt" in snapshots[-1].files
+    assert (outputs / "a.txt").is_file()
+    delivery = await _delivery_events(store, "thread-1", record.run_id)
+    assert len(delivery) == 1
+    content = delivery[0]["content"]
+    if modify_existing:
+        assert content["produced_paths"] == ["/mnt/user-data/outputs/b.txt"]
+        assert content["verification"]["source"] == "outputs_changed"
+        assert content["satisfied"] is False
+        assert "verification_skipped" not in content
+    else:
+        assert content == {"presented": 0, "paths": [], "by_tool": {}, "verification_skipped": "truncated_pre_run_snapshot"}
+    if terminal_outcome == "error":
+        assert record.status == RunStatus.error
+        assert record.error == "Agent failed after producing output"
+    elif terminal_outcome == "cancelled":
+        assert record.status == RunStatus.interrupted
+    elif modify_existing:
+        assert record.status == RunStatus.error
+        assert record.error == "Artifact delivery incomplete: no produced output artifact was presented"
+    else:
+        assert record.status == RunStatus.success
+        assert record.error is None
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("failure_stage", ["before", "after"])
+async def test_snapshot_capture_failure_does_not_claim_truncated_verification(tmp_path, monkeypatch, failure_stage):
+    paths = Paths(base_dir=tmp_path)
+    monkeypatch.setattr("deerflow.workspace_changes.recorder.get_paths", lambda: paths)
+    run_manager = RunManager()
+    record = await run_manager.create("thread-1")
+    store = MemoryRunEventStore()
+    outputs = paths.sandbox_outputs_dir("thread-1", user_id=get_effective_user_id())
+    outputs.mkdir(parents=True)
+    captures = 0
+
+    async def capture_or_fail(*args, **kwargs):
+        nonlocal captures
+        captures += 1
+        if captures == (1 if failure_stage == "before" else 2):
+            raise OSError("Snapshot capture unavailable")
+        return await capture_workspace_snapshot(*args, **kwargs)
+
+    monkeypatch.setattr("deerflow.runtime.runs.worker.capture_workspace_snapshot", capture_or_fail)
+
+    class ProducingAgent:
+        async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False):
+            (outputs / "report.md").write_text("Report ready\n", encoding="utf-8")
+            yield {"messages": [AIMessage(content="Created report.md.")]}
+
+    await run_agent(
+        _make_bridge(),
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=None, event_store=store),
+        agent_factory=lambda *, config: ProducingAgent(),
+        graph_input={},
+        config={},
+    )
+
+    assert captures == (1 if failure_stage == "before" else 2)
+    assert (outputs / "report.md").is_file()
+    delivery = await _delivery_events(store, "thread-1", record.run_id)
+    assert len(delivery) == 1
+    assert delivery[0]["content"] == {"presented": 0, "paths": [], "by_tool": {}}
+    assert record.status == RunStatus.success
+    assert record.error is None
+
+
 @pytest.mark.anyio
 async def test_externalized_tool_results_do_not_trigger_delivery_verification(tmp_path, monkeypatch):
     """Oversized tool outputs externalized under outputs/.tool-results/ are

backend/tests/test_workspace_changes.py

@@ -17,6 +17,7 @@ from deerflow.workspace_changes import (
     capture_workspace_snapshot,
     compare_snapshots,
     get_changed_output_paths,
+    get_changed_paths,
     record_workspace_changes,
     scan_workspace_roots,
 )
@@ -174,6 +175,55 @@ def test_get_changed_output_paths_returns_only_created_or_modified_regular_outpu
     ]
 
 
+@pytest.mark.parametrize(
+    ("before_contents", "after_contents", "expected_statuses"),
+    [
+        pytest.param(
+            {"b.txt": "before", "c.txt": "unchanged"},
+            {"a.txt": "new", "b.txt": "before", "c.txt": "unchanged"},
+            {"a.txt": "created"},
+            id="insertion-pushes-existing-file-out-of-scan",
+        ),
+        pytest.param(
+            {"a.txt": "removed", "b.txt": "before", "c.txt": "unchanged"},
+            {"b.txt": "before", "c.txt": "unchanged"},
+            {"a.txt": "deleted"},
+            id="deletion-brings-existing-file-into-scan",
+        ),
+        pytest.param(
+            {"a.txt": "removed", "b.txt": "before", "c.txt": "unchanged"},
+            {"b.txt": "after", "c.txt": "unchanged", "d.txt": "new"},
+            {"b.txt": "modified"},
+            id="both-truncated-preserve-observed-modifications",
+        ),
+    ],
+)
+def test_workspace_changes_do_not_infer_absence_from_truncated_snapshots(tmp_path, before_contents, after_contents, expected_statuses):
+    roots = _roots(tmp_path)
+    outputs = roots[1].host_path
+    limits = WorkspaceChangeLimits(max_scanned_files=2)
+    for name, content in before_contents.items():
+        (outputs / name).write_text(content, encoding="utf-8")
+    before = scan_workspace_roots(roots, limits=limits)
+
+    for name in before_contents.keys() - after_contents.keys():
+        (outputs / name).unlink()
+    for name, content in after_contents.items():
+        (outputs / name).write_text(content, encoding="utf-8")
+    after = scan_workspace_roots(roots, limits=limits)
+
+    result = compare_snapshots(before, after, limits=limits)
+
+    expected = {f"/mnt/user-data/outputs/{name}": status for name, status in expected_statuses.items()}
+    assert {change.path: change.status for change in result.files} == expected
+    assert result.summary.created == list(expected.values()).count("created")
+    assert result.summary.modified == list(expected.values()).count("modified")
+    assert result.summary.deleted == list(expected.values()).count("deleted")
+    assert result.summary.truncated is True
+    assert get_changed_paths(before, after) == set(expected)
+    assert get_changed_output_paths(before, after) == sorted(path for path, status in expected.items() if status != "deleted")
+
+
 def test_compare_snapshots_treats_utf16_markdown_as_text(tmp_path):
     roots = _roots(tmp_path)
     workspace = roots[0].host_path
@@ -313,6 +363,28 @@ def test_count_diff_lines_counts_content_starting_with_dashes_or_pluses():
     assert deletions == 1
 
 
+def test_scan_workspace_roots_walks_subdirectories_in_sorted_order(tmp_path, monkeypatch):
+    # The truncation boundary must not depend on the filesystem's directory order.
+    roots = _roots(tmp_path)
+    workspace = roots[0].host_path
+    for name in ("a", "b"):
+        (workspace / name).mkdir()
+        (workspace / name / "file.txt").write_text(name, encoding="utf-8")
+    real_walk = os.walk
+
+    def reversed_walk(top, *args, **kwargs):
+        for dirpath, dirnames, filenames in real_walk(top, *args, **kwargs):
+            dirnames.sort(reverse=True)
+            yield dirpath, dirnames, filenames
+
+    monkeypatch.setattr(scanner.os, "walk", reversed_walk)
+
+    snapshot = scan_workspace_roots(roots, limits=WorkspaceChangeLimits(max_scanned_files=1))
+
+    assert snapshot.truncated is True
+    assert list(snapshot.files) == ["/mnt/user-data/workspace/a/file.txt"]
+
+
 def test_scan_workspace_roots_skips_excluded_directories(tmp_path):
     roots = _roots(tmp_path)
     workspace = roots[0].host_path