feat(runtime): enforce artifact delivery from workspace snapshots (#4494)

README.md

@@ -345,6 +345,8 @@ DeerFlow runs the agent runtime inside the Gateway API. Development mode enables
 
 Gateway owns `/api/langgraph/*` and translates those public LangGraph-compatible paths to its native `/api/*` routers behind nginx.
 
+Gateway runs automatically enforce native delivery for artifacts created or modified under `/mnt/user-data/outputs`: `present_files` must present at least one output produced by the current run, and the terminal `run.delivery` receipt must be durably recorded. Runs that do not produce output artifacts keep ordinary conversational behavior.
+
 DeerFlow's built-in custom events are available through both LangGraph streaming interfaces: native clients can continue subscribing to `stream_mode="custom"`, while callback-based integrations can consume the same payloads as `on_custom_event` records from `astream_events(version="v2")`. The callback event name matches the payload's `type` field.
 
 #### Docker Production Deployment

backend/AGENTS.md

@@ -470,10 +470,18 @@ compatibility failures and cancellation while waiting for prior finalization
 still emit a zero-delivery receipt. The worker flushes ordinary journal events,
 idempotently persists the run-scoped receipt, and only then persists the staged
 terminal run status. A receipt failure is retried on a short bounded schedule
-while the owning worker still knows the real outcome and holds the lease. If
-all attempts fail, the worker persists that real terminal status instead of
-leaving a successful run inflight for orphan recovery to rewrite as an error;
-the receipt remains best-effort in that outage case. Orphan recovery first
+while the owning worker still knows the real outcome and holds the lease. The
+worker derives delivery requirements from the run's workspace snapshots rather
+than a client request option: every regular file created or modified under
+`/mnt/user-data/outputs` is a candidate produced artifact. At least one candidate
+must be covered by a path attributed by the journal to `present_files`;
+presenting only an unrelated pre-existing path does not satisfy delivery.
+Receipts for such runs add `produced_paths`, `presented_paths`, `matched_paths`,
+`verification`, `stage`, and `satisfied` to the Slice 1 fact fields. Missing a
+matching presentation becomes a run error; a successful presentation is also
+downgraded to error if its receipt cannot be durably verified. Runs without
+changed outputs preserve ordinary chat behavior and the original receipt shape.
+Orphan recovery first
 atomically claims an expired lease, then uses the same singleton write to
 backfill a zero-delivery receipt. This ordering prevents a stale recovery scan
 from overwriting a live run's later detailed receipt; an event-store outage

backend/app/gateway/services.py

@@ -987,7 +987,6 @@ async def start_run(
 
     body_context = getattr(body, "context", None) or {}
     model_name = body_context.get("model_name")
-
     # Coerce non-string model_name values to str before truncation.
     if model_name is not None and not isinstance(model_name, str):
         model_name = str(model_name)

backend/docs/API.md

@@ -109,10 +109,17 @@ Content-Type: application/json
 **Run Option Compatibility:**
 - Supported concurrency strategies: `reject`, `rollback`, and `interrupt`
 - Compatibility default: `if_not_exists="create"`; this matches DeerFlow's current behavior
+- Artifact delivery is enforced automatically when a run creates or modifies regular files under `/mnt/user-data/outputs`. `present_files` must present at least one path produced by the current run (or a directory containing it), and the terminal receipt must be persisted; presenting only an unrelated file does not satisfy delivery. Runs without changed outputs retain ordinary conversational behavior. `artifact_delivery` is not a client-settable run option.
 - Unsupported options return `422`: `webhook`, `stream_resumable=true`, `after_seconds`, `feedback_keys`, any non-null `on_completion` value (including the SDK values `"complete"` and `"continue"`), `if_not_exists="reject"`, and `multitask_strategy="enqueue"`
 - `stream_resumable=false` is accepted: it is the LangGraph SDK's default and requests the non-resumable stream DeerFlow already serves
 - Undeclared SDK options, including `checkpoint_during` and `durability`, also return `422` instead of being silently discarded
 
+When outputs changed during the run, `run.delivery` events retain the Slice 1
+facts (`presented`, `paths`, and `by_tool`) and add `produced_paths`,
+`presented_paths`, `matched_paths`, plus an explicit verdict: `verification`,
+`stage` (`presented`, `mismatched`, or `not_started`), and `satisfied`. Receipts
+for runs without changed outputs keep their existing shape.
+
 **Recursion Limit:**
 
 `config.recursion_limit` caps the number of graph steps LangGraph will execute

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

@@ -78,7 +78,7 @@ from deerflow.trace_context import (
 )
 from deerflow.tracing import inject_langfuse_metadata
 from deerflow.utils.messages import message_to_text
-from deerflow.workspace_changes import capture_workspace_snapshot, record_workspace_changes
+from deerflow.workspace_changes import capture_workspace_snapshot, get_changed_output_paths, record_workspace_changes
 from deerflow.workspace_changes.types import WorkspaceSnapshot
 
 from .manager import RunManager, RunRecord, RunStartOutcome
@@ -140,7 +140,7 @@ async def _persist_delivery_receipt(
         except Exception:
             if attempt == attempts - 1:
                 logger.warning(
-                    "Failed to persist delivery receipt for run %s after %d attempts; preserving the real terminal status without a receipt",
+                    "Failed to persist delivery receipt for run %s after %d attempts; applying terminal delivery semantics without a receipt",
                     run_id,
                     attempts,
                     exc_info=True,
@@ -160,6 +160,68 @@ async def _persist_delivery_receipt(
     return False  # pragma: no cover - loop always returns
 
 
+_DELIVERY_INCOMPLETE_ERROR = "Artifact delivery incomplete: no produced output artifact was presented"
+_DELIVERY_RECEIPT_FAILED_ERROR = "Artifact delivery verification failed: terminal delivery receipt could not be persisted"
+
+
+def _empty_delivery_content() -> dict[str, Any]:
+    return {"presented": 0, "paths": [], "by_tool": {}}
+
+
+def _presented_path_covers_output(presented_path: str, produced_path: str) -> bool:
+    presented_path = presented_path.rstrip("/")
+    return bool(presented_path) and (produced_path == presented_path or produced_path.startswith(f"{presented_path}/"))
+
+
+def _delivery_content_with_outputs(
+    content: dict[str, Any],
+    produced_paths: list[str],
+) -> dict[str, Any]:
+    """Attach a delivery verdict when this run created or modified outputs."""
+    if not produced_paths:
+        return content
+
+    presented_paths = content.get("by_tool", {}).get("present_files", [])
+    matched_paths = [produced_path for produced_path in produced_paths if any(_presented_path_covers_output(presented_path, produced_path) for presented_path in presented_paths)]
+    satisfied = bool(matched_paths)
+    return {
+        **content,
+        "verification": {
+            "source": "outputs_changed",
+            "requirement": "present_files_matches_produced_output",
+        },
+        "produced_paths": produced_paths,
+        "presented_paths": presented_paths,
+        "matched_paths": matched_paths,
+        "stage": "presented" if satisfied else ("mismatched" if presented_paths else "not_started"),
+        "satisfied": satisfied,
+    }
+
+
+def _delivery_error(content: dict[str, Any]) -> str | None:
+    """Return the terminal error when no changed output was presented."""
+    if not content.get("produced_paths") or content.get("satisfied") is True:
+        return None
+    return _DELIVERY_INCOMPLETE_ERROR
+
+
+async def _produced_output_paths(
+    before: WorkspaceSnapshot | None,
+    *,
+    thread_id: str,
+    user_id: str | None,
+) -> list[str]:
+    """Detect regular output files created or modified by this run."""
+    if before is None:
+        return []
+    try:
+        after = await capture_workspace_snapshot(thread_id, user_id=user_id, include_text=False)
+        return get_changed_output_paths(before, after)
+    except Exception:
+        logger.warning("Could not detect produced output artifacts for run thread %s", thread_id, exc_info=True)
+        return []
+
+
 # Keep this streaming policy separate from middleware write-authorization sets.
 _LARGE_FILE_TOOL_NAMES = frozenset({"str_replace", "write_file"})
 _LARGE_FILE_TOOL_BATCH_SIZE = 32
@@ -471,6 +533,8 @@ async def run_agent(
     accessor: CheckpointStateAccessor | None = None
     rollback_point: RollbackPoint | None = None
     journal = None
+    delivery_content: dict[str, Any] | None = None
+    produced_output_paths: list[str] | None = None
     # Journal construction moved ahead of preflight so every terminal run can
     # emit a receipt. Completion persistence keeps its prior boundary: before
     # #4272 the journal did not exist until preflight had succeeded, so early
@@ -871,9 +935,20 @@ async def run_agent(
             # collects the most severe / first / all reasons) instead of each
             # guard writing directly to the same key.
             stop_reason = runtime_context.get("stop_reason") if runtime_context is not None else None
+            produced_output_paths = await _produced_output_paths(
+                pre_run_workspace_snapshot,
+                thread_id=thread_id,
+                user_id=workspace_changes_user_id,
+            )
+            delivery_content = _delivery_content_with_outputs(
+                journal.get_delivery_content() if journal is not None else _empty_delivery_content(),
+                produced_output_paths,
+            )
+            delivery_error = _delivery_error(delivery_content)
             await run_manager.set_status(
                 run_id,
-                RunStatus.success,
+                RunStatus.error if delivery_error else RunStatus.success,
+                error=delivery_error,
                 stop_reason=stop_reason,
                 **terminal_status_kwargs,
             )
@@ -961,12 +1036,27 @@ async def run_agent(
             except Exception:
                 logger.warning("Failed to flush journal for run %s", run_id, exc_info=True)
 
-            await _persist_delivery_receipt(
+            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,
+                    )
+                delivery_content = _delivery_content_with_outputs(journal.get_delivery_content(), produced_output_paths)
+            receipt_persisted = await _persist_delivery_receipt(
                 event_store,
                 thread_id=thread_id,
                 run_id=run_id,
-                content=journal.get_delivery_content(),
+                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,
+                )
 
         if not record.ownership_lost and event_store is not None:
             try:

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

@@ -1,5 +1,5 @@
 from .api import get_workspace_changes_response
-from .diff import compare_snapshots, get_changed_paths
+from .diff import compare_snapshots, get_changed_output_paths, get_changed_paths
 from .recorder import capture_workspace_snapshot, record_workspace_changes
 from .scanner import scan_workspace_roots
 from .types import (
@@ -26,6 +26,7 @@ __all__ = [
     "WorkspaceSnapshot",
     "capture_workspace_snapshot",
     "compare_snapshots",
+    "get_changed_output_paths",
     "get_changed_paths",
     "get_workspace_changes_response",
     "record_workspace_changes",

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

@@ -109,6 +109,16 @@ def get_changed_paths(before: WorkspaceSnapshot, after: WorkspaceSnapshot) -> se
     return changed
 
 
+def get_changed_output_paths(before: WorkspaceSnapshot, after: WorkspaceSnapshot) -> list[str]:
+    """Return created or modified regular files under the outputs root."""
+    paths: list[str] = []
+    for path in sorted(get_changed_paths(before, after)):
+        snapshot = after.files.get(path)
+        if snapshot is not None and snapshot.root == "outputs" and not snapshot.symlink:
+            paths.append(path)
+    return paths
+
+
 def _status(
     before_file: FileSnapshot | None,
     after_file: FileSnapshot | None,

backend/tests/test_run_request_validation.py

@@ -116,6 +116,28 @@ def test_run_request_keeps_supported_modes_and_compatibility_defaults() -> None:
     assert body.if_not_exists == "create"
 
 
+@pytest.mark.parametrize(
+    "artifact_delivery",
+    [
+        {"required": True, "tool": "present_files"},
+        {},
+        None,
+        "present_files",
+        [],
+    ],
+)
+def test_run_request_rejects_removed_artifact_delivery_override(
+    client: TestClient,
+    artifact_delivery: Any,
+) -> None:
+    response = client.post(
+        "/api/runs/stream",
+        json={"artifact_delivery": artifact_delivery},
+    )
+
+    assert response.status_code == 422
+
+
 def _sdk_default_payload(method: str) -> dict[str, Any]:
     """Capture the body a stock ``langgraph_sdk`` client sends for a default run."""
     from langgraph_sdk.client import get_client

backend/tests/test_run_worker_delivery.py

@@ -13,7 +13,7 @@ from deerflow.runtime.events.store.memory import MemoryRunEventStore
 from deerflow.runtime.runs.manager import RunManager
 from deerflow.runtime.runs.schemas import RunStatus
 from deerflow.runtime.runs.store.memory import MemoryRunStore
-from deerflow.runtime.runs.worker import RunContext, run_agent
+from deerflow.runtime.runs.worker import RunContext, _delivery_content_with_outputs, run_agent
 
 
 def _make_bridge():
@@ -25,6 +25,28 @@ async def _delivery_events(store: MemoryRunEventStore, thread_id: str, run_id: s
     return [e for e in events if e["event_type"] == "run.delivery"]
 
 
+def test_delivery_verification_treats_presented_directory_as_covering_produced_files():
+    content = {
+        "presented": 1,
+        "paths": ["/mnt/user-data/outputs/site"],
+        "by_tool": {"present_files": ["/mnt/user-data/outputs/site"]},
+    }
+
+    delivery = _delivery_content_with_outputs(
+        content,
+        [
+            "/mnt/user-data/outputs/site/index.html",
+            "/mnt/user-data/outputs/site/assets/style.css",
+        ],
+    )
+
+    assert delivery["matched_paths"] == [
+        "/mnt/user-data/outputs/site/index.html",
+        "/mnt/user-data/outputs/site/assets/style.css",
+    ]
+    assert delivery["satisfied"] is True
+
+
 @pytest.mark.anyio
 async def test_delivery_event_records_present_files_paths_on_success():
     run_manager = RunManager()
@@ -95,6 +117,201 @@ async def test_delivery_event_presented_zero_without_artifact_production():
     assert fetched.status == RunStatus.success
 
 
+@pytest.mark.anyio
+async def test_changed_outputs_succeed_when_a_produced_output_is_presented(monkeypatch):
+    run_manager = RunManager()
+    record = await run_manager.create("thread-1")
+    store = MemoryRunEventStore()
+    monkeypatch.setattr(
+        "deerflow.runtime.runs.worker._produced_output_paths",
+        AsyncMock(return_value=["/mnt/user-data/outputs/report.md"]),
+    )
+
+    class DummyAgent:
+        async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False):
+            journal = config["context"]["__run_journal"]
+            ai = AIMessage(content="", tool_calls=[{"id": "call_1", "name": "present_files", "args": {}}])
+            journal._remember_current_run_tool_calls(ai, caller="lead_agent")
+            journal.on_tool_end(
+                Command(
+                    update={
+                        "artifacts": ["/mnt/user-data/outputs/report.md"],
+                        "messages": [ToolMessage("Successfully presented files", tool_call_id="call_1")],
+                    }
+                ),
+                run_id=uuid4(),
+            )
+            yield {"messages": []}
+
+    await run_agent(
+        _make_bridge(),
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=None, event_store=store),
+        agent_factory=lambda *, config: DummyAgent(),
+        graph_input={},
+        config={},
+    )
+
+    delivery = await _delivery_events(store, "thread-1", record.run_id)
+    assert delivery[0]["content"] == {
+        "presented": 1,
+        "paths": ["/mnt/user-data/outputs/report.md"],
+        "by_tool": {"present_files": ["/mnt/user-data/outputs/report.md"]},
+        "verification": {
+            "source": "outputs_changed",
+            "requirement": "present_files_matches_produced_output",
+        },
+        "produced_paths": ["/mnt/user-data/outputs/report.md"],
+        "presented_paths": ["/mnt/user-data/outputs/report.md"],
+        "matched_paths": ["/mnt/user-data/outputs/report.md"],
+        "stage": "presented",
+        "satisfied": True,
+    }
+    assert record.status == RunStatus.success
+
+
+@pytest.mark.anyio
+async def test_changed_outputs_fail_closed_when_not_presented(monkeypatch):
+    run_manager = RunManager()
+    record = await run_manager.create("thread-1")
+    store = MemoryRunEventStore()
+    monkeypatch.setattr(
+        "deerflow.runtime.runs.worker._produced_output_paths",
+        AsyncMock(return_value=["/mnt/user-data/outputs/report.md"]),
+    )
+
+    class ProseOnlyAgent:
+        async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False):
+            yield {"messages": [AIMessage(content="SESSION SUMMARY")]}
+
+    await run_agent(
+        _make_bridge(),
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=None, event_store=store),
+        agent_factory=lambda *, config: ProseOnlyAgent(),
+        graph_input={},
+        config={},
+    )
+
+    delivery = await _delivery_events(store, "thread-1", record.run_id)
+    assert delivery[0]["content"] == {
+        "presented": 0,
+        "paths": [],
+        "by_tool": {},
+        "verification": {
+            "source": "outputs_changed",
+            "requirement": "present_files_matches_produced_output",
+        },
+        "produced_paths": ["/mnt/user-data/outputs/report.md"],
+        "presented_paths": [],
+        "matched_paths": [],
+        "stage": "not_started",
+        "satisfied": False,
+    }
+    assert record.status == RunStatus.error
+    assert record.error == "Artifact delivery incomplete: no produced output artifact was presented"
+    assert record.stop_reason is None
+
+
+@pytest.mark.anyio
+async def test_changed_outputs_succeed_when_one_of_multiple_outputs_is_presented(monkeypatch):
+    run_manager = RunManager()
+    record = await run_manager.create("thread-1")
+    store = MemoryRunEventStore()
+    monkeypatch.setattr(
+        "deerflow.runtime.runs.worker._produced_output_paths",
+        AsyncMock(
+            return_value=[
+                "/mnt/user-data/outputs/report.md",
+                "/mnt/user-data/outputs/appendix.md",
+            ]
+        ),
+    )
+
+    class PartiallyPresentingAgent:
+        async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False):
+            journal = config["context"]["__run_journal"]
+            journal._remember_current_run_tool_calls(
+                AIMessage(content="", tool_calls=[{"id": "call_1", "name": "present_files", "args": {}}]),
+                caller="lead_agent",
+            )
+            journal.on_tool_end(
+                Command(
+                    update={
+                        "artifacts": ["/mnt/user-data/outputs/report.md"],
+                        "messages": [ToolMessage("Successfully presented files", tool_call_id="call_1")],
+                    }
+                ),
+                run_id=uuid4(),
+            )
+            yield {"messages": []}
+
+    await run_agent(
+        _make_bridge(),
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=None, event_store=store),
+        agent_factory=lambda *, config: PartiallyPresentingAgent(),
+        graph_input={},
+        config={},
+    )
+
+    delivery = (await _delivery_events(store, "thread-1", record.run_id))[0]["content"]
+    assert delivery["stage"] == "presented"
+    assert delivery["presented_paths"] == ["/mnt/user-data/outputs/report.md"]
+    assert delivery["matched_paths"] == ["/mnt/user-data/outputs/report.md"]
+    assert delivery["satisfied"] is True
+    assert record.status == RunStatus.success
+
+
+@pytest.mark.anyio
+async def test_changed_outputs_fail_when_present_files_only_presents_an_unrelated_file(monkeypatch):
+    run_manager = RunManager()
+    record = await run_manager.create("thread-1")
+    store = MemoryRunEventStore()
+    monkeypatch.setattr(
+        "deerflow.runtime.runs.worker._produced_output_paths",
+        AsyncMock(return_value=["/mnt/user-data/outputs/report.md"]),
+    )
+
+    class UnrelatedPresentingAgent:
+        async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False):
+            journal = config["context"]["__run_journal"]
+            journal._remember_current_run_tool_calls(
+                AIMessage(content="", tool_calls=[{"id": "call_1", "name": "present_files", "args": {}}]),
+                caller="lead_agent",
+            )
+            journal.on_tool_end(
+                Command(
+                    update={
+                        "artifacts": ["/mnt/user-data/outputs/old-report.md"],
+                        "messages": [ToolMessage("Successfully presented files", tool_call_id="call_1")],
+                    }
+                ),
+                run_id=uuid4(),
+            )
+            yield {"messages": []}
+
+    await run_agent(
+        _make_bridge(),
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=None, event_store=store),
+        agent_factory=lambda *, config: UnrelatedPresentingAgent(),
+        graph_input={},
+        config={},
+    )
+
+    delivery = (await _delivery_events(store, "thread-1", record.run_id))[0]["content"]
+    assert delivery["stage"] == "mismatched"
+    assert delivery["presented_paths"] == ["/mnt/user-data/outputs/old-report.md"]
+    assert delivery["matched_paths"] == []
+    assert delivery["satisfied"] is False
+    assert record.status == RunStatus.error
+
+
 @pytest.mark.anyio
 async def test_fenced_worker_leaves_delivery_receipt_to_peer_recovery():
     """A stale worker must not finalize the singleton delivery receipt."""
@@ -337,6 +554,53 @@ async def test_delivery_write_failure_preserves_real_durable_terminal_status():
     assert (await run_store.get(record.run_id))["status"] == "success"
 
 
+@pytest.mark.anyio
+async def test_produced_artifact_delivery_fails_closed_when_receipt_cannot_be_persisted(monkeypatch):
+    class FailingReceiptStore(MemoryRunEventStore):
+        async def put_if_absent(self, **kwargs):
+            raise RuntimeError("event store unavailable")
+
+    run_store = MemoryRunStore()
+    run_manager = RunManager(store=run_store)
+    record = await run_manager.create("thread-1")
+    monkeypatch.setattr(
+        "deerflow.runtime.runs.worker._produced_output_paths",
+        AsyncMock(return_value=["/mnt/user-data/outputs/report.md"]),
+    )
+
+    class PresentingAgent:
+        async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False):
+            journal = config["context"]["__run_journal"]
+            journal._remember_current_run_tool_calls(
+                AIMessage(content="", tool_calls=[{"id": "call_1", "name": "present_files", "args": {}}]),
+                caller="lead_agent",
+            )
+            journal.on_tool_end(
+                Command(
+                    update={
+                        "artifacts": ["/mnt/user-data/outputs/report.md"],
+                        "messages": [ToolMessage("Successfully presented files", tool_call_id="call_1")],
+                    }
+                ),
+                run_id=uuid4(),
+            )
+            yield {"messages": []}
+
+    await run_agent(
+        _make_bridge(),
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=None, event_store=FailingReceiptStore()),
+        agent_factory=lambda *, config: PresentingAgent(),
+        graph_input={},
+        config={},
+    )
+
+    assert record.status == RunStatus.error
+    assert record.error == "Artifact delivery verification failed: terminal delivery receipt could not be persisted"
+    assert (await run_store.get(record.run_id))["status"] == "error"
+
+
 @pytest.mark.anyio
 async def test_delivery_event_emitted_when_checkpoint_preflight_fails(monkeypatch):
     run_manager = RunManager()

backend/tests/test_workspace_changes.py

@@ -14,6 +14,7 @@ from deerflow.workspace_changes import (
     WorkspaceRoot,
     capture_workspace_snapshot,
     compare_snapshots,
+    get_changed_output_paths,
     record_workspace_changes,
     scan_workspace_roots,
 )
@@ -70,6 +71,27 @@ def test_compare_snapshots_reports_text_file_changes(tmp_path):
     assert changes["/mnt/user-data/workspace/old.txt"].status == "deleted"
 
 
+def test_get_changed_output_paths_returns_only_created_or_modified_regular_outputs(tmp_path):
+    roots = _roots(tmp_path)
+    workspace = roots[0].host_path
+    outputs = roots[1].host_path
+    (workspace / "draft.md").write_text("before", encoding="utf-8")
+    (outputs / "existing.md").write_text("before", encoding="utf-8")
+    (outputs / "deleted.md").write_text("before", encoding="utf-8")
+    before = scan_workspace_roots(roots)
+
+    (workspace / "draft.md").write_text("after", encoding="utf-8")
+    (outputs / "existing.md").write_text("after", encoding="utf-8")
+    (outputs / "created.md").write_text("new", encoding="utf-8")
+    (outputs / "deleted.md").unlink()
+    after = scan_workspace_roots(roots)
+
+    assert get_changed_output_paths(before, after) == [
+        "/mnt/user-data/outputs/created.md",
+        "/mnt/user-data/outputs/existing.md",
+    ]
+
+
 def test_compare_snapshots_treats_utf16_markdown_as_text(tmp_path):
     roots = _roots(tmp_path)
     workspace = roots[0].host_path