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