From f5907405a042f7a22d6bf754f7b7189c35967313 Mon Sep 17 00:00:00 2001 From: TAI-YUAN GUO Date: Thu, 22 Oct 2026 16:52:53 +0800 Subject: [PATCH] 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 * 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 Co-authored-by: Willem Jiang --- README.md | 17 +++ backend/docs/runtime-guidance-details.md | 24 ++++ .../harness/deerflow/runtime/AGENTS.md | 2 + .../harness/deerflow/runtime/runs/worker.py | 9 +- .../deerflow/workspace_changes/diff.py | 16 ++- .../deerflow/workspace_changes/scanner.py | 2 +- backend/tests/test_run_worker_delivery.py | 118 ++++++++++++++++++ backend/tests/test_workspace_changes.py | 72 +++++++++++ 8 files changed, 252 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 352a92b1c7..835efd0035 100644 --- a/README.md +++ b/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. diff --git a/backend/docs/runtime-guidance-details.md b/backend/docs/runtime-guidance-details.md index 370111f831..892de72c00 100644 --- a/backend/docs/runtime-guidance-details.md +++ b/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 diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 2d19575cd4..8a594ac1d1 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/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 diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index 52328aa063..3a560b955f 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/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) diff --git a/backend/packages/harness/deerflow/workspace_changes/diff.py b/backend/packages/harness/deerflow/workspace_changes/diff.py index 1f81c1b5a1..60fe5bf0de 100644 --- a/backend/packages/harness/deerflow/workspace_changes/diff.py +++ b/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) diff --git a/backend/packages/harness/deerflow/workspace_changes/scanner.py b/backend/packages/harness/deerflow/workspace_changes/scanner.py index fb7944b1db..6b77750761 100644 --- a/backend/packages/harness/deerflow/workspace_changes/scanner.py +++ b/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 diff --git a/backend/tests/test_run_worker_delivery.py b/backend/tests/test_run_worker_delivery.py index 21ec22591a..b446be4a05 100644 --- a/backend/tests/test_run_worker_delivery.py +++ b/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 diff --git a/backend/tests/test_workspace_changes.py b/backend/tests/test_workspace_changes.py index c0f4d5c24e..ce692d7748 100644 --- a/backend/tests/test_workspace_changes.py +++ b/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