feat(runtime): record terminal artifact delivery receipts (slice 1 of #4272) (#4365)

* feat(runtime): record terminal artifact delivery receipts (#4272)

* fix(runtime): persist delivery receipts across recovery

* test(runtime): cover delivery receipt invariants

* fix(runtime): preserve terminal status on receipt outages

README.md

@@ -275,7 +275,7 @@ Browser login uses `HttpOnly` session cookies. The login page offers a "keep me
 DeerFlow still uses `Forwarded` / `X-Forwarded-*` headers to recover the browser-facing scheme and origin behind a proxy. The bundled nginx sets `X-Forwarded-Proto`, but preserves an upstream HTTPS value and does not overwrite every forwarded header. Configure the outer trusted proxy to replace or strip client-supplied forwarding headers before traffic reaches DeerFlow.
 
 > [!IMPORTANT]
-> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), and `run_ownership.heartbeat_enabled: true`. The bridge shares SSE delivery and `Last-Event-ID` replay across workers, while lease reconciliation marks runs from dead workers as errors, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. SSE and `/wait` consumers also refresh durable status on heartbeats as a fallback if terminal publication fails. The rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. IM channel state and other process-local services still need their own multi-worker coordination.
+> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`; process-local memory/JSONL event stores cannot enforce singleton delivery receipts across workers. The bridge shares SSE delivery and `Last-Event-ID` replay across workers, while lease reconciliation marks runs from dead workers as errors, persists their delivery receipts, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. SSE and `/wait` consumers also refresh durable status on heartbeats as a fallback if terminal publication fails. Malformed reconnect IDs live-tail new events instead of replaying the retained buffer, and the rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. IM channel state and other process-local services still need their own multi-worker coordination.
 >
 > Reconciliation uses an atomic takeover claim that re-checks the lease after candidate selection, so a successful owner renewal wins over orphan recovery and only one reconciler can report a run as recovered.
 
@@ -574,6 +574,17 @@ logging:
 
 When enabled, every Gateway HTTP response includes `X-Trace-Id`, logs include `trace_id`, and Langfuse traces created by that request include `metadata.deerflow_trace_id` with the same value.
 
+Gateway run history also records one terminal `run.delivery` receipt per run,
+including zero-output and crash-recovered runs. The receipt is persisted before
+the durable terminal run status during normal execution. Orphan recovery first
+atomically claims an expired lease and then idempotently backfills the receipt,
+so a stale recovery scan cannot overwrite a live run's detailed delivery facts.
+Receipt persistence remains best-effort during an event-store outage. Runs that
+fail checkpoint preflight (or are cancelled while waiting for prior
+finalization) keep the existing completion-data behavior: they receive the
+zero-delivery receipt but do not overwrite RunStore completion fields with an
+empty snapshot.
+
 #### LangSmith Tracing
 
 DeerFlow has built-in [LangSmith](https://smith.langchain.com) integration for observability. When enabled, all LLM calls, agent runs, and tool executions are traced and visible in the LangSmith dashboard.

backend/AGENTS.md

@@ -219,7 +219,9 @@ Blocking-IO runtime gate (`tests/blocking_io/`):
   for #1917); `test_sqlite_lifespan.py` (locks the offload around
   SQLite path resolution plus `ensure_sqlite_parent_dir`, fix for #1912);
   `test_jsonl_run_event_store.py` (locks `JsonlRunEventStore`'s async
-  API offloading its file IO via `asyncio.to_thread`);
+  API — including idempotent singleton-event writes — offloading its file IO
+  via `asyncio.to_thread`); `test_run_journal_callbacks.py` (locks
+  `RunJournal.run_inline` tool callbacks to in-memory/event-loop-safe work);
   `test_integrations_router.py` (locks Lark integration install and auth
   completion route handlers offloading archive filesystem work and `lark-cli`
   subprocesses);
@@ -451,6 +453,44 @@ captures a pre-run and post-run snapshot of the thread-owned `workspace` and
 are size-limited; binary, large, and sensitive-looking paths are persisted as
 metadata only.
 
+**Run delivery receipts**: `RunJournal` records each non-empty artifact update
+once per tool `Command` for the terminal `run.delivery` event. When a command
+contains multiple messages, a unique tool name resolved from matching
+`ToolMessage` entries supplies attribution; additional command messages do not
+duplicate artifact paths or counts. If multiple different tool names resolve
+for one flat artifact update, the paths remain counted but unattributed because
+the command does not carry a per-path mapping. `RunJournal` callbacks set
+`run_inline=True`: they do only in-memory bookkeeping or schedule async writes,
+and staying on the run's event-loop thread serializes parallel tool callbacks
+before terminal delivery recording and flushing. Each worker creates a separate
+journal per run before cancellable/fallible preflight work, so checkpoint
+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
+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
+does not undo the terminal takeover. An existing detailed receipt is preserved
+when a worker crashed after writing it. Event stores
+serialize `put_if_absent` with ordinary thread writers: memory and JSONL provide
+the documented single-process guarantee, while the DB store adds per-thread
+in-process locks and PostgreSQL advisory locks for cross-process writers.
+Moving journal construction ahead of preflight is receipt-only on early failure
+paths: a separate boundary flag preserves the previous completion-data
+semantics, so checkpoint incompatibility or cancellation while waiting for an
+older finalizing run does not persist an empty completion snapshot. Worker tests
+pin one accumulated receipt across multiple goal-continuation `_stream_once`
+calls; journal tests drive LangChain's real async callback dispatcher against a
+single journal to pin serialized, deduplicated parallel tool callbacks.
+Multi-worker deployments therefore require `run_events.backend: db` for shared,
+ordered delivery events; the startup gate rejects process-local memory and
+JSONL event stores when `GATEWAY_WORKERS > 1`.
+
 **RunManager / RunStore contract**:
 - LangGraph-compatible run requests validate their supported subset before creating a run. `runtime/stream_modes.py` is the shared backend contract for public stream modes and the worker's `graph.astream` mapping; the public `messages-tuple` mode maps to LangGraph's internal `messages` mode, while public `messages`, `events`, and other unsupported modes are rejected instead of being dropped or replaced with `values`. `app/gateway/run_models.py::RunCreateRequest` is shared by HTTP and internal scheduled launch paths, retains only truthful compatibility defaults for unimplemented options (`if_not_exists="create"` plus `None` placeholders), returns 422 for unsupported values including `on_completion="complete"`, `on_completion="continue"`, and `multitask_strategy="enqueue"`, and forbids undeclared SDK options so fields such as `checkpoint_during` and `durability` cannot be silently discarded. A placeholder must still accept the stock SDK's own default: `langgraph_sdk` drops only `None` from its run payload, so `stream_resumable=False` reaches every request and means "non-resumable", which is what DeerFlow serves — rejecting it 422'd every IM channel run (#4466). `tests/test_run_request_validation.py::test_gateway_accepts_langgraph_sdk_default_payload` pins the real SDK payload against this boundary; channel tests mock the SDK client and cannot catch this class of drift.
 - `RunManager.get()` is async; direct callers must `await` it.

backend/app/gateway/deps.py

@@ -58,14 +58,16 @@ def _browser_tools_enabled_in_config(config: AppConfig) -> bool:
 def _enforce_postgres_for_multi_worker(config: AppConfig) -> None:
     """Refuse unsafe multi-worker configurations before persistence starts.
 
-    Three checks (all must pass for multi-worker):
+    Four checks (all must pass for multi-worker):
 
     1. Process-local browser sessions must be disabled. Browser tools keep
        Chromium and Playwright objects in one worker's memory, while ordinary
        uvicorn dispatch provides no thread-id affinity.
     2. The DB backend must be Postgres — SQLite write-locks cannot support
        concurrent multi-process access.
-    3. ``run_ownership.heartbeat_enabled`` must be True — without heartbeat,
+    3. ``run_events.backend`` must be ``db``. Memory and JSONL stores are
+       process-local, so workers cannot enforce a shared singleton receipt.
+    4. ``run_ownership.heartbeat_enabled`` must be True — without heartbeat,
        every run has a NULL lease, so reconciliation treats all inflight
        runs as orphans and Worker B would kill Worker A's live runs on
        every rolling update or scale-up.
@@ -89,6 +91,14 @@ def _enforce_postgres_for_multi_worker(config: AppConfig) -> None:
     if backend != "postgres":
         raise SystemExit(f"GATEWAY_WORKERS={workers} requires database.backend='postgres', but database.backend is '{backend}'. SQLite cannot support concurrent multi-process access. Set GATEWAY_WORKERS=1 or switch to Postgres.")
 
+    run_events_backend = getattr(getattr(config, "run_events", None), "backend", None)
+    if run_events_backend != "db":
+        raise SystemExit(
+            f"GATEWAY_WORKERS={workers} requires run_events.backend='db', but run_events.backend is '{run_events_backend}'. "
+            "Memory and JSONL event stores are process-local, so delivery receipt singleton guarantees cannot hold across workers. "
+            "Set GATEWAY_WORKERS=1 or configure run_events.backend: db."
+        )
+
     run_ownership = getattr(config, "run_ownership", None)
     if run_ownership is None or not run_ownership.heartbeat_enabled:
         raise SystemExit(
@@ -448,6 +458,7 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen
         app.state.run_manager = RunManager(
             store=app.state.run_store,
             run_ownership_config=run_ownership_config,
+            event_store=app.state.run_event_store,
             on_orphans_recovered=terminalize_recovered_runs,
         )
         # Startup recovery: mark inflight runs whose lease has expired as error.

backend/docs/CONFIGURATION.md

@@ -248,7 +248,7 @@ Notes:
 - `enabled: false` keeps background polling off by default.
 - `max_concurrent_runs` is a global cap on active scheduled runs (queued/running run rows); each poll cycle claims only into the remaining budget, so long runs accumulating across cycles cannot exceed it.
 - All scheduler fields are restart-required; edits need a Gateway restart.
-- Multi-worker deployments (`GATEWAY_WORKERS > 1`) must use the Postgres database backend. SQLite silently ignores row-level locks, so multiple workers can double-fire the same task. The process-local agentic browser tool group is incompatible with multiple Gateway workers; keep `GATEWAY_WORKERS=1` while `browser_navigate` is enabled. Browser control also requires the backend `browser` extra (`cd backend && uv sync --extra browser && uv run playwright install chromium`); startup detects enabled browser config and fails fast when Playwright is missing, and `/api/features` reports `browser_control.enabled=false` until the runtime is available.
+- Multi-worker deployments (`GATEWAY_WORKERS > 1`) must use the Postgres database backend, enable run ownership heartbeats, and set `run_events.backend: db`. SQLite silently ignores row-level locks, while memory and JSONL run-event stores are process-local and cannot enforce singleton delivery receipts across workers; startup rejects these combinations. The process-local agentic browser tool group is incompatible with multiple Gateway workers; keep `GATEWAY_WORKERS=1` while `browser_navigate` is enabled. Browser control also requires the backend `browser` extra (`cd backend && uv sync --extra browser && uv run playwright install chromium`); startup detects enabled browser config and fails fast when Playwright is missing, and `/api/features` reports `browser_control.enabled=false` until the runtime is available.
 - The MVP supports thread reuse and fresh-thread-per-run execution modes.
 - The MVP supports only `once` and `cron`.
 - Manual trigger uses the same scheduled-task resource and run lifecycle.

backend/packages/harness/deerflow/runtime/events/store/base.py

@@ -51,6 +51,26 @@ class RunEventStore(abc.ABC):
         Returns complete records with seq assigned.
         """
 
+    @abc.abstractmethod
+    async def put_if_absent(
+        self,
+        *,
+        thread_id: str,
+        run_id: str,
+        event_type: str,
+        category: str,
+        content: str | dict = "",
+        metadata: dict | None = None,
+        created_at: str | None = None,
+    ) -> tuple[dict, bool]:
+        """Write one event unless this run already has the same event type.
+
+        The check and write must be serialized with ordinary writers for the
+        thread. Returns ``(record, created)``. This is the durability primitive
+        used by terminal run receipts, whose recovery path may safely retry
+        after a worker crash.
+        """
+
     @abc.abstractmethod
     async def list_messages(
         self,

backend/packages/harness/deerflow/runtime/events/store/db.py

@@ -194,6 +194,59 @@ class DbRunEventStore(RunEventStore):
                         rows.append(row)
                 return [self._row_to_dict(r) for r in rows]
 
+    async def put_if_absent(
+        self,
+        *,
+        thread_id,
+        run_id,
+        event_type,
+        category,
+        content="",
+        metadata=None,
+        created_at=None,
+    ):
+        """Idempotently insert a run-scoped singleton event.
+
+        ``_max_seq_for_thread`` takes the same PostgreSQL advisory lock used by
+        every normal writer (and the in-process lock covers SQLite), so the
+        existence check cannot race another ``put_if_absent`` or journal write.
+        Terminal delivery receipts use this method on both the worker and
+        recovery paths; ordinary event types remain append-only.
+        """
+        content, metadata = self._truncate_trace(category, content, metadata)
+        db_content, metadata = self._content_to_db(content, metadata)
+        user_id = self._user_id_from_context()
+        async with self._get_write_lock(thread_id):
+            async with self._sf() as session:
+                async with session.begin():
+                    max_seq = await self._max_seq_for_thread(session, thread_id)
+                    stmt = (
+                        select(RunEventRow)
+                        .where(
+                            RunEventRow.thread_id == thread_id,
+                            RunEventRow.run_id == run_id,
+                            RunEventRow.event_type == event_type,
+                        )
+                        .order_by(RunEventRow.seq.asc())
+                        .limit(1)
+                    )
+                    existing = await session.scalar(stmt)
+                    if existing is not None:
+                        return self._row_to_dict(existing), False
+                    row = RunEventRow(
+                        thread_id=thread_id,
+                        run_id=run_id,
+                        user_id=user_id,
+                        event_type=event_type,
+                        category=category,
+                        content=db_content,
+                        event_metadata=metadata,
+                        seq=(max_seq or 0) + 1,
+                        created_at=datetime.fromisoformat(created_at) if created_at else datetime.now(UTC),
+                    )
+                    session.add(row)
+                return self._row_to_dict(row), True
+
     async def list_messages(
         self,
         thread_id,

backend/packages/harness/deerflow/runtime/events/store/jsonl.py

@@ -178,6 +178,36 @@ class JsonlRunEventStore(RunEventStore):
             results.extend(records)
         return results
 
+    async def put_if_absent(
+        self,
+        *,
+        thread_id,
+        run_id,
+        event_type,
+        category,
+        content="",
+        metadata=None,
+        created_at=None,
+    ):
+        async with self._get_write_lock(thread_id):
+            existing = await asyncio.to_thread(self._read_run_events, thread_id, run_id)
+            for event in existing:
+                if event.get("event_type") == event_type:
+                    return event, False
+            await self._ensure_seq_loaded(thread_id)
+            record = {
+                "thread_id": thread_id,
+                "run_id": run_id,
+                "event_type": event_type,
+                "category": category,
+                "content": content,
+                "metadata": metadata or {},
+                "seq": self._next_seq(thread_id),
+                "created_at": created_at or datetime.now(UTC).isoformat(),
+            }
+            await asyncio.to_thread(self._write_record, record)
+            return record, True
+
     async def _write_batch_async(self, thread_id: str, batch: list[dict[str, Any]]) -> list[dict[str, Any]]:
         async with self._get_write_lock(thread_id):
             await self._ensure_seq_loaded(thread_id)

backend/packages/harness/deerflow/runtime/events/store/memory.py

@@ -93,6 +93,35 @@ class MemoryRunEventStore(RunEventStore):
             results.append(record)
         return results
 
+    async def put_if_absent(
+        self,
+        *,
+        thread_id,
+        run_id,
+        event_type,
+        category,
+        content="",
+        metadata=None,
+        created_at=None,
+    ):
+        # No await occurs between the lookup and append, so this is atomic for
+        # the backend's documented single-event-loop concurrency model.
+        for event in self._events_by_run.get(thread_id, {}).get(run_id, []):
+            if event["event_type"] == event_type:
+                return event, False
+        return (
+            self._put_one(
+                thread_id=thread_id,
+                run_id=run_id,
+                event_type=event_type,
+                category=category,
+                content=content,
+                metadata=metadata,
+                created_at=created_at,
+            ),
+            True,
+        )
+
     async def list_messages(self, thread_id, *, limit=50, before_seq=None, after_seq=None, user_id: str | None | _AutoSentinel = AUTO):
         # ``messages`` is messages-only and seq-sorted, so the seq window is a
         # contiguous slice located with bisect (O(log m)) rather than a full scan.

backend/packages/harness/deerflow/runtime/journal.py

@@ -178,6 +178,12 @@ def build_branch_history_seed_events(
 class RunJournal(BaseCallbackHandler):
     """LangChain callback handler that captures events to RunEventStore."""
 
+    # Every callback only updates in-memory run state or schedules async IO.
+    # Keeping callbacks on the run's event-loop thread serializes mutations
+    # from parallel tool calls and prevents cancelled executor callbacks from
+    # racing terminal delivery recording and flush.
+    run_inline = True
+
     def __init__(
         self,
         run_id: str,
@@ -242,6 +248,11 @@ class RunJournal(BaseCallbackHandler):
         self._current_run_tool_call_names: dict[str, str] = {}
         self._persisted_tool_message_identities: set[str] = set()
 
+        # Artifact-production tracking for the terminal run.delivery event
+        # (#4272 slice 1). Deduped by (path, tool_name); insertion order kept.
+        self._produced_artifacts: list[tuple[str, str | None]] = []
+        self._produced_artifact_keys: set[tuple[str, str | None]] = set()
+
     # -- Lifecycle callbacks --
 
     @staticmethod
@@ -489,11 +500,26 @@ class RunJournal(BaseCallbackHandler):
             elif isinstance(output, Command):
                 cmd = cast(Command, output)
                 messages = cmd.update.get("messages", [])
+                # A non-empty ``artifacts`` update is only produced on the
+                # success path (e.g. present_files returns an error ToolMessage
+                # without touching state when validation fails), so its
+                # presence is the artifact-production signal (#4272 slice 1).
+                artifacts = cmd.update.get("artifacts")
+                artifact_tool_names: set[str] = set()
                 for message in messages:
                     if isinstance(message, BaseMessage):
                         self._persist_tool_result_message(message)
+                        if artifacts and isinstance(message, ToolMessage):
+                            tool_call_id = getattr(message, "tool_call_id", None)
+                            if isinstance(tool_call_id, str):
+                                tool_name = self._current_run_tool_call_names.get(tool_call_id)
+                                if tool_name:
+                                    artifact_tool_names.add(tool_name)
                     else:
                         logger.warning(f"on_tool_end {run_id}: command update message is not BaseMessage: {type(message)}")
+                if artifacts:
+                    artifact_tool_name = next(iter(artifact_tool_names)) if len(artifact_tool_names) == 1 else None
+                    self._record_produced_artifacts(artifacts, artifact_tool_name)
             else:
                 logger.warning(f"on_tool_end {run_id}: output is not ToolMessage: {type(output)}")
         finally:
@@ -784,6 +810,44 @@ class RunJournal(BaseCallbackHandler):
         )
         self._memory_context_recorded = True
 
+    def _record_produced_artifacts(self, artifacts: Any, tool_name: str | None) -> None:
+        """Accumulate produced artifact paths, deduped by (path, tool_name)."""
+        if not isinstance(artifacts, list):
+            return
+        for path in artifacts:
+            if not isinstance(path, str) or not path:
+                continue
+            key = (path, tool_name)
+            if key not in self._produced_artifact_keys:
+                self._produced_artifact_keys.add(key)
+                self._produced_artifacts.append(key)
+
+    def get_delivery_content(self) -> dict[str, Any]:
+        """Return the terminal delivery fact accumulated for this run.
+
+        This is a fact record, not a verdict: runs that produced no artifacts
+        emit ``presented: 0``.
+        """
+        by_tool: dict[str, list[str]] = {}
+        paths: list[str] = []
+        for path, tool_name in self._produced_artifacts:
+            paths.append(path)
+            if tool_name:
+                by_tool.setdefault(tool_name, []).append(path)
+        return {"presented": len(paths), "paths": paths, "by_tool": by_tool}
+
+    def record_delivery(self) -> None:
+        """Buffer the terminal ``run.delivery`` event for this run (#4272 slice 1).
+
+        Kept for direct journal users. The worker uses the event store's
+        idempotent singleton write so crash recovery can safely backfill it.
+        """
+        self._put(
+            event_type="run.delivery",
+            category="outputs",
+            content=self.get_delivery_content(),
+        )
+
     async def flush(self) -> None:
         """Force flush remaining buffer. Called in worker's finally block."""
         if self._pending_flush_tasks:

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

@@ -24,6 +24,7 @@ from .schemas import DisconnectMode, RunStatus, ThreadOperationKind
 
 if TYPE_CHECKING:
     from deerflow.config.run_ownership_config import RunOwnershipConfig
+    from deerflow.runtime.events.store.base import RunEventStore
     from deerflow.runtime.runs.store.base import RunStore
 
 logger = logging.getLogger(__name__)
@@ -221,6 +222,7 @@ class RunManager:
         persistence_retry_policy: PersistenceRetryPolicy | None = None,
         worker_id: str | None = None,
         run_ownership_config: RunOwnershipConfig | None = None,
+        event_store: RunEventStore | None = None,
         on_orphans_recovered: OrphanRecoveryCallback | None = None,
     ) -> None:
         self._runs: dict[str, RunRecord] = {}
@@ -234,6 +236,7 @@ class RunManager:
         self._persistence_retry_policy = persistence_retry_policy or PersistenceRetryPolicy()
         self._worker_id = worker_id or _generate_worker_id()
         self._run_ownership_config = run_ownership_config
+        self._event_store = event_store
         self._on_orphans_recovered = on_orphans_recovered
         self._heartbeat_task: asyncio.Task | None = None
         self._heartbeat_stop: asyncio.Event | None = None
@@ -757,7 +760,15 @@ class RunManager:
                 logger.warning("Failed to map store row for run %s", run_id, exc_info=True)
         return records_by_id
 
-    async def set_status(self, run_id: str, status: RunStatus, *, error: str | None = None, stop_reason: str | None = None) -> None:
+    async def set_status(
+        self,
+        run_id: str,
+        status: RunStatus,
+        *,
+        error: str | None = None,
+        stop_reason: str | None = None,
+        persist: bool = True,
+    ) -> None:
         """Transition a run to a new status."""
         async with self._lock:
             record = self._runs.get(run_id)
@@ -770,9 +781,43 @@ class RunManager:
                 record.error = error
             if stop_reason is not None:
                 record.stop_reason = stop_reason
-        await self._persist_status(record, status, error=error, stop_reason=stop_reason)
+        if persist:
+            await self._persist_status(record, status, error=error, stop_reason=stop_reason)
         logger.info("Run %s -> %s", run_id, status.value)
 
+    async def persist_current_status(self, run_id: str) -> bool:
+        """Persist the status already staged on the in-memory run record."""
+        async with self._lock:
+            record = self._runs.get(run_id)
+            if record is None:
+                logger.warning("persist_current_status called for unknown run %s", run_id)
+                return False
+            status = record.status
+            error = record.error
+            stop_reason = record.stop_reason
+        return await self._persist_status(record, status, error=error, stop_reason=stop_reason)
+
+    async def _ensure_delivery_receipt(self, record: RunRecord) -> bool:
+        """Idempotently persist a zero-delivery receipt during recovery."""
+        if self._event_store is None:
+            return True
+        try:
+            await self._event_store.put_if_absent(
+                thread_id=record.thread_id,
+                run_id=record.run_id,
+                event_type="run.delivery",
+                category="outputs",
+                content={"presented": 0, "paths": [], "by_tool": {}},
+            )
+            return True
+        except Exception:
+            logger.warning(
+                "Failed to backfill delivery receipt for recovered run %s; preserving its terminal status",
+                record.run_id,
+                exc_info=True,
+            )
+            return False
+
     async def set_finalizing(self, run_id: str, finalizing: bool) -> None:
         """Mark whether a run is performing post-cancel cleanup."""
         async with self._lock:
@@ -1355,6 +1400,12 @@ class RunManager:
             record.stop_reason = stop_reason
             record.updated_at = now
             if record.operation_kind == ThreadOperationKind.run:
+                # The atomic takeover above must win before writing a zero-delivery
+                # receipt; otherwise a stale scan could race a heartbeat renewal and
+                # permanently overwrite a live run's later detailed receipt. The
+                # receipt remains best-effort, matching normal terminal delivery
+                # when its event store is unavailable.
+                await self._ensure_delivery_receipt(record)
                 recovered.append(record)
 
         if recovered:

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

@@ -103,6 +103,57 @@ async def _checkpoint_thread_lock(thread_id: str) -> AsyncIterator[None]:
         yield
 
 
+_DELIVERY_RECEIPT_RETRY_DELAYS_SECONDS = (0.1, 0.5)
+
+
+async def _persist_delivery_receipt(
+    event_store: Any,
+    *,
+    thread_id: str,
+    run_id: str,
+    content: dict[str, Any],
+) -> bool:
+    """Persist a terminal receipt with short bounded retries.
+
+    The owning worker still knows the real terminal outcome and renews its
+    lease while this coroutine runs. Retrying here handles transient event
+    store failures without handing a successful run to orphan recovery, which
+    cannot reconstruct either the terminal status or the detailed receipt.
+    """
+    attempts = len(_DELIVERY_RECEIPT_RETRY_DELAYS_SECONDS) + 1
+    for attempt in range(attempts):
+        try:
+            await event_store.put_if_absent(
+                thread_id=thread_id,
+                run_id=run_id,
+                event_type="run.delivery",
+                category="outputs",
+                content=content,
+            )
+            return True
+        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",
+                    run_id,
+                    attempts,
+                    exc_info=True,
+                )
+                return False
+            delay = _DELIVERY_RECEIPT_RETRY_DELAYS_SECONDS[attempt]
+            logger.warning(
+                "Failed to persist delivery receipt for run %s (attempt %d/%d); retrying in %.1fs",
+                run_id,
+                attempt + 1,
+                attempts,
+                delay,
+                exc_info=True,
+            )
+            await asyncio.sleep(delay)
+
+    return False  # pragma: no cover - loop always returns
+
+
 # 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
@@ -393,6 +444,7 @@ async def run_agent(
     event_store = ctx.event_store
     run_events_config = ctx.run_events_config
     thread_store = ctx.thread_store
+    terminal_status_kwargs = {"persist": False} if event_store is not None else {}
 
     run_id = record.run_id
     thread_id = record.thread_id
@@ -413,6 +465,12 @@ async def run_agent(
     accessor: CheckpointStateAccessor | None = None
     rollback_point: RollbackPoint | None = None
     journal = 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
+    # checkpoint failures / cancellation while waiting did not write an empty
+    # completion snapshot into RunStore.
+    persist_completion = False
     # Buffers subagent step events for batched persistence (#3779); assigned once
     # streaming starts and flushed in the finally block. Pre-bound to None so the
     # finally is safe even if an exception fires before streaming begins.
@@ -423,6 +481,22 @@ async def run_agent(
         normalized_stream_modes = normalize_stream_modes(stream_modes)
         requested_modes: set[str] = set(normalized_stream_modes)
         lg_modes = to_langgraph_stream_modes(normalized_stream_modes)
+        # Initialize the run-scoped journal before any fallible or cancellable
+        # preflight work. Every terminal run with an event store must reach the
+        # shared finally block with a journal available for its run.delivery
+        # receipt, including checkpoint validation failures and cancellation
+        # while waiting for an earlier run to finish finalizing.
+        if event_store is not None:
+            from deerflow.runtime.journal import RunJournal
+
+            journal = RunJournal(
+                run_id=run_id,
+                thread_id=thread_id,
+                event_store=event_store,
+                track_token_usage=getattr(run_events_config, "track_token_usage", True),
+                progress_reporter=lambda snapshot: run_manager.update_run_progress(run_id, **snapshot),
+            )
+
         await run_manager.wait_for_prior_finalizing(
             thread_id,
             run_id,
@@ -439,7 +513,6 @@ async def run_agent(
                 await thread_store.update_status(thread_id, "running")
             except Exception:
                 logger.debug("Failed to update thread_meta status for %s (non-fatal)", thread_id)
-
         mode = ctx.checkpoint_channel_mode
         inject_checkpoint_mode(config, mode)
         checkpoint_config = {
@@ -472,22 +545,7 @@ async def run_agent(
                     mode,
                 )
 
-        # Initialize RunJournal + write human_message event.
-        # These are inside the try block so any exception (e.g. a DB
-        # error writing the event) flows through the except/finally
-        # path that publishes an "end" event to the SSE bridge —
-        # otherwise a failure here would leave the stream hanging
-        # with no terminator.
-        if event_store is not None:
-            from deerflow.runtime.journal import RunJournal
-
-            journal = RunJournal(
-                run_id=run_id,
-                thread_id=thread_id,
-                event_store=event_store,
-                track_token_usage=getattr(run_events_config, "track_token_usage", True),
-                progress_reporter=lambda snapshot: run_manager.update_run_progress(run_id, **snapshot),
-            )
+        persist_completion = True
 
         if event_store is not None:
             workspace_changes_user_id = get_effective_user_id()
@@ -731,7 +789,12 @@ async def run_agent(
             await run_manager.set_finalizing(run_id, True)
             action = record.abort_action
             if action == "rollback":
-                await run_manager.set_status(run_id, RunStatus.error, error="Rolled back by user")
+                await run_manager.set_status(
+                    run_id,
+                    RunStatus.error,
+                    error="Rolled back by user",
+                    **terminal_status_kwargs,
+                )
                 try:
                     await _rollback_to_pre_run_checkpoint(
                         accessor=accessor,
@@ -745,13 +808,22 @@ async def run_agent(
                 except Exception:
                     logger.warning("Failed to rollback checkpoint for run %s", run_id, exc_info=True)
             else:
-                await run_manager.set_status(run_id, RunStatus.interrupted)
+                await run_manager.set_status(
+                    run_id,
+                    RunStatus.interrupted,
+                    **terminal_status_kwargs,
+                )
         elif llm_error_fallback_message or (journal is not None and journal.had_llm_error_fallback):
             error_msg = llm_error_fallback_message
             if error_msg is None and journal is not None:
                 error_msg = journal.llm_error_fallback_message
             error_msg = error_msg or "LLM provider failed after retries"
-            await run_manager.set_status(run_id, RunStatus.error, error=error_msg)
+            await run_manager.set_status(
+                run_id,
+                RunStatus.error,
+                error=error_msg,
+                **terminal_status_kwargs,
+            )
         else:
             runtime_context = runtime.context if isinstance(runtime.context, dict) else None
             # Guard middlewares that hard-stop a run by stripping tool_calls
@@ -769,13 +841,23 @@ 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
-            await run_manager.set_status(run_id, RunStatus.success, stop_reason=stop_reason)
+            await run_manager.set_status(
+                run_id,
+                RunStatus.success,
+                stop_reason=stop_reason,
+                **terminal_status_kwargs,
+            )
 
     except asyncio.CancelledError:
         await run_manager.set_finalizing(run_id, True)
         action = record.abort_action
         if action == "rollback":
-            await run_manager.set_status(run_id, RunStatus.error, error="Rolled back by user")
+            await run_manager.set_status(
+                run_id,
+                RunStatus.error,
+                error="Rolled back by user",
+                **terminal_status_kwargs,
+            )
             try:
                 await _rollback_to_pre_run_checkpoint(
                     accessor=accessor,
@@ -789,13 +871,22 @@ async def run_agent(
             except Exception:
                 logger.warning("Run %s cancellation rollback failed", run_id, exc_info=True)
         else:
-            await run_manager.set_status(run_id, RunStatus.interrupted)
+            await run_manager.set_status(
+                run_id,
+                RunStatus.interrupted,
+                **terminal_status_kwargs,
+            )
             logger.info("Run %s was cancelled", run_id)
 
     except Exception as exc:
         error_msg = f"{exc}"
         logger.exception("Run %s failed: %s", run_id, error_msg)
-        await run_manager.set_status(run_id, RunStatus.error, error=error_msg)
+        await run_manager.set_status(
+            run_id,
+            RunStatus.error,
+            error=error_msg,
+            **terminal_status_kwargs,
+        )
         await bridge.publish(
             run_id,
             "error",
@@ -823,13 +914,34 @@ async def run_agent(
             except Exception:
                 logger.warning("Failed to record workspace changes for run %s", run_id, exc_info=True)
 
-        # Flush any buffered journal events and persist completion data
+        # Flush buffered journal events before the terminal receipt. The
+        # receipt uses a run-scoped idempotent write shared with recovery, then
+        # the staged terminal status is persisted. This ordering closes the
+        # crash window where a terminal run could otherwise outlive its receipt.
         if journal is not None:
             try:
                 await journal.flush()
             except Exception:
                 logger.warning("Failed to flush journal for run %s", run_id, exc_info=True)
 
+            await _persist_delivery_receipt(
+                event_store,
+                thread_id=thread_id,
+                run_id=run_id,
+                content=journal.get_delivery_content(),
+            )
+
+        if event_store is not None:
+            try:
+                # Even after bounded receipt retries are exhausted, persist the
+                # real worker outcome. Leaving a successful row inflight would
+                # let lease recovery rewrite it as an error with a synthetic
+                # zero receipt.
+                await run_manager.persist_current_status(run_id)
+            except Exception:
+                logger.warning("Failed to persist terminal status for run %s after delivery receipt attempts", run_id, exc_info=True)
+
+        if journal is not None and persist_completion:
             try:
                 # Persist token usage + convenience fields to RunStore
                 completion = journal.get_completion_data()

backend/tests/blocking_io/test_jsonl_run_event_store.py

@@ -38,7 +38,7 @@ async def test_jsonl_run_event_store_async_api_does_not_block_event_loop(tmp_pat
     thread_dir.mkdir(parents=True, exist_ok=True)
     (thread_dir / "r0.jsonl").write_text('{"seq": 1, "category": "message", "run_id": "r0"}\n', encoding="utf-8")
 
-    # writes: put + put_batch
+    # writes: put + put_batch + idempotent singleton insert
     record = await store.put(thread_id="t1", run_id="r1", event_type="message", category="message", content="hi")
     assert record["seq"] >= 2
     batch = await store.put_batch(
@@ -48,6 +48,23 @@ async def test_jsonl_run_event_store_async_api_does_not_block_event_loop(tmp_pat
         ]
     )
     assert len(batch) == 2
+    singleton, created = await store.put_if_absent(
+        thread_id="t1",
+        run_id="r1",
+        event_type="run.delivery",
+        category="outputs",
+        content={"presented": 0, "paths": [], "by_tool": {}},
+    )
+    duplicate, duplicate_created = await store.put_if_absent(
+        thread_id="t1",
+        run_id="r1",
+        event_type="run.delivery",
+        category="outputs",
+        content={"presented": 99},
+    )
+    assert created is True
+    assert duplicate_created is False
+    assert duplicate == singleton
 
     # reads: list_messages / list_events / list_messages_by_run / count_messages.
     # list_events is exercised both without and with the event_types filter so

backend/tests/blocking_io/test_run_journal_callbacks.py

@@ -0,0 +1,39 @@
+"""Regression anchor: inline RunJournal callbacks stay event-loop safe."""
+
+from __future__ import annotations
+
+from uuid import uuid4
+
+import pytest
+from langchain_core.callbacks.manager import ahandle_event
+from langchain_core.messages import AIMessage, ToolMessage
+from langgraph.types import Command
+
+from deerflow.runtime.events.store.memory import MemoryRunEventStore
+from deerflow.runtime.journal import RunJournal
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_inline_tool_callback_does_not_block_event_loop() -> None:
+    journal = RunJournal("run-1", "thread-1", MemoryRunEventStore(), flush_threshold=100)
+    journal._remember_current_run_tool_calls(
+        AIMessage(content="", tool_calls=[{"id": "call-1", "name": "present_files", "args": {}}]),
+        caller="lead_agent",
+    )
+    command = Command(
+        update={
+            "artifacts": ["/mnt/user-data/outputs/report.md"],
+            "messages": [ToolMessage("Successfully presented files", tool_call_id="call-1")],
+        }
+    )
+
+    await ahandle_event(
+        [journal],
+        "on_tool_end",
+        "ignore_agent",
+        command,
+        run_id=uuid4(),
+    )
+
+    assert journal.get_delivery_content()["paths"] == ["/mnt/user-data/outputs/report.md"]

backend/tests/test_gateway_run_recovery.py

@@ -34,9 +34,17 @@ class _FakeRunManager:
     recovered_runs = [SimpleNamespace(run_id="run-1", thread_id="thread-1")]
     latest_by_thread: dict[str, list[SimpleNamespace]] = {}
 
-    def __init__(self, *, store, run_ownership_config=None, on_orphans_recovered=None):
+    def __init__(
+        self,
+        *,
+        store,
+        run_ownership_config=None,
+        event_store=None,
+        on_orphans_recovered=None,
+    ):
         self.store = store
         self.run_ownership_config = run_ownership_config
+        self.event_store = event_store
         self.on_orphans_recovered = on_orphans_recovered
         self.reconcile_calls: list[dict] = []
         self.list_by_thread_calls: list[dict] = []

backend/tests/test_multi_worker_postgres_gate.py

@@ -24,10 +24,21 @@ from deerflow.config.database_config import DatabaseConfig
 from deerflow.config.run_ownership_config import RunOwnershipConfig
 
 
-def _config_with_backend(backend: str, *, heartbeat_enabled: bool | None = None, browser_enabled: bool = False) -> SimpleNamespace:
+def _config_with_backend(
+    backend: str,
+    *,
+    heartbeat_enabled: bool | None = None,
+    browser_enabled: bool = False,
+    run_events_backend: str = "db",
+) -> SimpleNamespace:
     run_ownership = RunOwnershipConfig(heartbeat_enabled=heartbeat_enabled) if heartbeat_enabled is not None else None
     tools = [SimpleNamespace(name="browser_navigate")] if browser_enabled else []
-    return SimpleNamespace(database=DatabaseConfig(backend=backend), run_ownership=run_ownership, tools=tools)
+    return SimpleNamespace(
+        database=DatabaseConfig(backend=backend),
+        run_ownership=run_ownership,
+        run_events=SimpleNamespace(backend=run_events_backend),
+        tools=tools,
+    )
 
 
 # ---------------------------------------------------------------------------
@@ -54,6 +65,34 @@ def test_gate_allows_multi_worker_with_postgres_and_heartbeat(monkeypatch):
     _enforce_postgres_for_multi_worker(_config_with_backend("postgres", heartbeat_enabled=True))
 
 
+@pytest.mark.parametrize("run_events_backend", ["memory", "jsonl"])
+def test_gate_rejects_process_local_run_events_with_multi_worker(monkeypatch, run_events_backend):
+    monkeypatch.setenv("GATEWAY_WORKERS", "2")
+    with pytest.raises(SystemExit) as exc_info:
+        _enforce_postgres_for_multi_worker(
+            _config_with_backend(
+                "postgres",
+                heartbeat_enabled=True,
+                run_events_backend=run_events_backend,
+            )
+        )
+    msg = str(exc_info.value)
+    assert "run_events.backend='db'" in msg
+    assert run_events_backend in msg
+    assert "GATEWAY_WORKERS=1" in msg
+
+
+def test_gate_allows_process_local_run_events_for_single_worker(monkeypatch):
+    monkeypatch.setenv("GATEWAY_WORKERS", "1")
+    for run_events_backend in ("memory", "jsonl"):
+        _enforce_postgres_for_multi_worker(
+            _config_with_backend(
+                "sqlite",
+                run_events_backend=run_events_backend,
+            )
+        )
+
+
 def test_gate_rejects_process_local_browser_with_multi_worker(monkeypatch):
     monkeypatch.setenv("GATEWAY_WORKERS", "2")
     with pytest.raises(SystemExit) as exc_info:

backend/tests/test_run_event_store.py

@@ -58,6 +58,28 @@ class TestPutAndSeq:
         record = await store.put(thread_id="t1", run_id="r1", event_type="llm_end", category="trace", metadata=meta)
         assert record["metadata"] == meta
 
+    @pytest.mark.anyio
+    async def test_put_if_absent_preserves_first_run_scoped_event(self, store):
+        first, created = await store.put_if_absent(
+            thread_id="t1",
+            run_id="r1",
+            event_type="run.delivery",
+            category="outputs",
+            content={"presented": 1},
+        )
+        duplicate, duplicate_created = await store.put_if_absent(
+            thread_id="t1",
+            run_id="r1",
+            event_type="run.delivery",
+            category="outputs",
+            content={"presented": 0},
+        )
+
+        assert created is True
+        assert duplicate_created is False
+        assert duplicate == first
+        assert [event["content"] for event in await store.list_events("t1", "r1") if event["event_type"] == "run.delivery"] == [{"presented": 1}]
+
 
 # -- list_messages --
 
@@ -343,6 +365,24 @@ class TestDbRunEventStore:
 
         await close_engine()
 
+    @pytest.mark.anyio
+    async def test_put_if_absent_is_idempotent(self, tmp_path):
+        from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
+        from deerflow.runtime.events.store.db import DbRunEventStore
+
+        url = f"sqlite+aiosqlite:///{tmp_path / 'test.db'}"
+        await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path))
+        s = DbRunEventStore(get_session_factory())
+
+        first, created = await s.put_if_absent(thread_id="t1", run_id="r1", event_type="run.delivery", category="outputs", content={"presented": 2})
+        duplicate, duplicate_created = await s.put_if_absent(thread_id="t1", run_id="r1", event_type="run.delivery", category="outputs", content={"presented": 0})
+
+        assert created is True
+        assert duplicate_created is False
+        assert duplicate["seq"] == first["seq"]
+        assert duplicate["content"] == {"presented": 2}
+        await close_engine()
+
     @pytest.mark.anyio
     async def test_trace_content_truncation(self, tmp_path):
         from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
@@ -695,6 +735,19 @@ class TestJsonlRunEventStore:
         messages = await s.list_messages("t1")
         assert len(messages) == 1
 
+    @pytest.mark.anyio
+    async def test_put_if_absent_is_idempotent(self, tmp_path):
+        from deerflow.runtime.events.store.jsonl import JsonlRunEventStore
+
+        s = JsonlRunEventStore(base_dir=tmp_path / "jsonl")
+        first, created = await s.put_if_absent(thread_id="t1", run_id="r1", event_type="run.delivery", category="outputs", content={"presented": 2})
+        duplicate, duplicate_created = await s.put_if_absent(thread_id="t1", run_id="r1", event_type="run.delivery", category="outputs", content={"presented": 0})
+
+        assert created is True
+        assert duplicate_created is False
+        assert duplicate == first
+        assert len(await s.list_events("t1", "r1", event_types=["run.delivery"])) == 1
+
     @pytest.mark.anyio
     async def test_file_at_correct_path(self, tmp_path):
         from deerflow.runtime.events.store.jsonl import JsonlRunEventStore

backend/tests/test_run_journal.py

@@ -1259,3 +1259,277 @@ class TestChatModelStartHumanMessage:
         j.on_chat_model_start({}, [], run_id=uuid4(), tags=["lead_agent"])
         await j.flush()
         assert j._first_human_msg is None
+
+
+class TestDeliveryTracking:
+    """Slice 1 (#4272): journal records artifact production for run.delivery."""
+
+    @staticmethod
+    def _register_tool_call(j: RunJournal, tool_call_id: str, name: str) -> None:
+        from langchain_core.messages import AIMessage
+
+        ai = AIMessage(content="", tool_calls=[{"id": tool_call_id, "name": name, "args": {}}])
+        j._remember_current_run_tool_calls(ai, caller="lead_agent")
+
+    def test_callbacks_run_inline_to_serialize_parallel_mutations(self, journal_setup):
+        j, _ = journal_setup
+
+        # LangChain dispatches synchronous handlers with run_inline=False via
+        # run_in_executor, allowing parallel tool callbacks to mutate one
+        # journal from different threads.
+        assert j.run_inline is True
+
+    @pytest.mark.anyio
+    async def test_concurrent_callbacks_on_one_journal_are_serialized(self, journal_setup):
+        from langchain_core.callbacks.manager import ahandle_event
+        from langchain_core.messages import ToolMessage
+        from langgraph.types import Command
+
+        j, _ = journal_setup
+        commands = []
+        for index, path in enumerate(("report.md", "report.md", "appendix.md"), start=1):
+            tool_call_id = f"call_{index}"
+            self._register_tool_call(j, tool_call_id, "present_files")
+            commands.append(
+                Command(
+                    update={
+                        "artifacts": [f"/mnt/user-data/outputs/{path}"],
+                        "messages": [ToolMessage("Successfully presented files", tool_call_id=tool_call_id)],
+                    }
+                )
+            )
+
+        # This is the real LangChain async callback dispatcher. Because the
+        # journal is run_inline, each synchronous mutation completes on the
+        # event-loop thread instead of racing in executor threads.
+        await asyncio.gather(
+            *(
+                ahandle_event(
+                    [j],
+                    "on_tool_end",
+                    "ignore_agent",
+                    command,
+                    run_id=uuid4(),
+                )
+                for command in commands
+            )
+        )
+
+        content = j.get_delivery_content()
+        assert content["presented"] == 2
+        assert set(content["paths"]) == {
+            "/mnt/user-data/outputs/report.md",
+            "/mnt/user-data/outputs/appendix.md",
+        }
+        assert set(content["by_tool"]["present_files"]) == set(content["paths"])
+
+    @pytest.mark.anyio
+    async def test_concurrent_runs_keep_delivery_accumulators_isolated(self):
+        from langchain_core.messages import ToolMessage
+        from langgraph.types import Command
+
+        store = MemoryRunEventStore()
+        journals = [RunJournal(run_id, "t1", store, flush_threshold=100) for run_id in ("r1", "r2")]
+
+        async def finish_run(journal: RunJournal, index: int) -> None:
+            tool_call_id = f"call_run_{index}"
+            self._register_tool_call(journal, tool_call_id, "present_files")
+            journal.on_tool_end(
+                Command(
+                    update={
+                        "artifacts": [f"/mnt/user-data/outputs/report-{index}.md"],
+                        "messages": [ToolMessage("Successfully presented files", tool_call_id=tool_call_id)],
+                    }
+                ),
+                run_id=uuid4(),
+            )
+            await asyncio.sleep(0)
+            journal.record_delivery()
+            await journal.flush()
+
+        await asyncio.gather(*(finish_run(journal, index) for index, journal in enumerate(journals, start=1)))
+
+        for index in (1, 2):
+            events = await store.list_events("t1", f"r{index}")
+            content = next(e for e in events if e["event_type"] == "run.delivery")["content"]
+            assert content == {
+                "presented": 1,
+                "paths": [f"/mnt/user-data/outputs/report-{index}.md"],
+                "by_tool": {"present_files": [f"/mnt/user-data/outputs/report-{index}.md"]},
+            }
+
+    @pytest.mark.anyio
+    async def test_present_files_success_command_recorded_with_attribution(self, journal_setup):
+        from langchain_core.messages import ToolMessage
+        from langgraph.types import Command
+
+        j, store = journal_setup
+        self._register_tool_call(j, "call_1", "present_files")
+        cmd = Command(
+            update={
+                "artifacts": ["/mnt/user-data/outputs/report.md"],
+                "messages": [ToolMessage("Successfully presented files", tool_call_id="call_1")],
+            }
+        )
+        j.on_tool_end(cmd, run_id=uuid4())
+        j.record_delivery()
+        await j.flush()
+
+        events = await store.list_events("t1", "r1")
+        delivery = [e for e in events if e["event_type"] == "run.delivery"]
+        assert len(delivery) == 1
+        content = delivery[0]["content"]
+        assert content["presented"] == 1
+        assert content["paths"] == ["/mnt/user-data/outputs/report.md"]
+        assert content["by_tool"] == {"present_files": ["/mnt/user-data/outputs/report.md"]}
+        assert delivery[0]["category"] == "outputs"
+
+    @pytest.mark.anyio
+    async def test_command_with_multiple_messages_records_artifacts_once(self, journal_setup):
+        from langchain_core.messages import ToolMessage
+        from langgraph.types import Command
+
+        j, store = journal_setup
+        self._register_tool_call(j, "call_multi", "present_files")
+        cmd = Command(
+            update={
+                "artifacts": ["/mnt/user-data/outputs/report.md"],
+                "messages": [
+                    ToolMessage("Successfully presented files", tool_call_id="call_multi"),
+                    HumanMessage("Additional command message"),
+                ],
+            }
+        )
+        j.on_tool_end(cmd, run_id=uuid4())
+        j.record_delivery()
+        await j.flush()
+
+        events = await store.list_events("t1", "r1")
+        content = next(e for e in events if e["event_type"] == "run.delivery")["content"]
+        assert content == {
+            "presented": 1,
+            "paths": ["/mnt/user-data/outputs/report.md"],
+            "by_tool": {"present_files": ["/mnt/user-data/outputs/report.md"]},
+        }
+
+    @pytest.mark.anyio
+    async def test_command_with_multiple_tool_names_leaves_artifacts_unattributed(self, journal_setup):
+        from langchain_core.messages import ToolMessage
+        from langgraph.types import Command
+
+        j, store = journal_setup
+        self._register_tool_call(j, "call_present", "present_files")
+        self._register_tool_call(j, "call_browser", "browser_screenshot")
+        cmd = Command(
+            update={
+                "artifacts": [
+                    "/mnt/user-data/outputs/report.md",
+                    "/mnt/user-data/outputs/shot.png",
+                ],
+                "messages": [
+                    ToolMessage("Successfully presented files", tool_call_id="call_present"),
+                    ToolMessage("Saved browser screenshot", tool_call_id="call_browser"),
+                ],
+            }
+        )
+        j.on_tool_end(cmd, run_id=uuid4())
+        j.record_delivery()
+        await j.flush()
+
+        events = await store.list_events("t1", "r1")
+        content = next(e for e in events if e["event_type"] == "run.delivery")["content"]
+        assert content == {
+            "presented": 2,
+            "paths": [
+                "/mnt/user-data/outputs/report.md",
+                "/mnt/user-data/outputs/shot.png",
+            ],
+            "by_tool": {},
+        }
+
+    @pytest.mark.anyio
+    async def test_error_command_without_artifacts_not_recorded(self, journal_setup):
+        from langchain_core.messages import ToolMessage
+        from langgraph.types import Command
+
+        j, store = journal_setup
+        self._register_tool_call(j, "call_2", "present_files")
+        cmd = Command(update={"messages": [ToolMessage("Error: Only files in /mnt/user-data/outputs can be presented", tool_call_id="call_2")]})
+        j.on_tool_end(cmd, run_id=uuid4())
+        j.record_delivery()
+        await j.flush()
+
+        events = await store.list_events("t1", "r1")
+        delivery = [e for e in events if e["event_type"] == "run.delivery"]
+        assert len(delivery) == 1
+        assert delivery[0]["content"] == {"presented": 0, "paths": [], "by_tool": {}}
+
+    @pytest.mark.anyio
+    async def test_browser_tool_artifacts_recorded_under_producing_tool(self, journal_setup):
+        from langchain_core.messages import ToolMessage
+        from langgraph.types import Command
+
+        j, store = journal_setup
+        self._register_tool_call(j, "call_3", "browser_screenshot")
+        cmd = Command(
+            update={
+                "artifacts": ["/mnt/user-data/outputs/shot.png"],
+                "messages": [ToolMessage("Saved browser screenshot", tool_call_id="call_3")],
+            }
+        )
+        j.on_tool_end(cmd, run_id=uuid4())
+        j.record_delivery()
+        await j.flush()
+
+        events = await store.list_events("t1", "r1")
+        content = next(e for e in events if e["event_type"] == "run.delivery")["content"]
+        assert content["presented"] == 1
+        assert content["by_tool"] == {"browser_screenshot": ["/mnt/user-data/outputs/shot.png"]}
+
+    @pytest.mark.anyio
+    async def test_duplicate_path_tool_pair_recorded_once(self, journal_setup):
+        from langchain_core.messages import ToolMessage
+        from langgraph.types import Command
+
+        j, store = journal_setup
+        self._register_tool_call(j, "call_4", "present_files")
+        for _ in range(2):
+            j.on_tool_end(
+                Command(
+                    update={
+                        "artifacts": ["/mnt/user-data/outputs/report.md"],
+                        "messages": [ToolMessage("Successfully presented files", tool_call_id="call_4")],
+                    }
+                ),
+                run_id=uuid4(),
+            )
+        j.record_delivery()
+        await j.flush()
+
+        events = await store.list_events("t1", "r1")
+        content = next(e for e in events if e["event_type"] == "run.delivery")["content"]
+        assert content["presented"] == 1
+        assert content["paths"] == ["/mnt/user-data/outputs/report.md"]
+
+    @pytest.mark.anyio
+    async def test_unattributed_artifacts_counted_without_by_tool_entry(self, journal_setup):
+        from langchain_core.messages import ToolMessage
+        from langgraph.types import Command
+
+        j, store = journal_setup
+        # No _register_tool_call: attribution missing (e.g. tool_call names map miss).
+        cmd = Command(
+            update={
+                "artifacts": ["/mnt/user-data/outputs/anon.txt"],
+                "messages": [ToolMessage("ok", tool_call_id="call_unknown")],
+            }
+        )
+        j.on_tool_end(cmd, run_id=uuid4())
+        j.record_delivery()
+        await j.flush()
+
+        events = await store.list_events("t1", "r1")
+        content = next(e for e in events if e["event_type"] == "run.delivery")["content"]
+        assert content["presented"] == 1
+        assert content["paths"] == ["/mnt/user-data/outputs/anon.txt"]
+        assert content["by_tool"] == {}

backend/tests/test_run_manager.py

@@ -11,6 +11,7 @@ from sqlalchemy.exc import DatabaseError as SQLAlchemyDatabaseError
 
 from deerflow.config.run_ownership_config import RunOwnershipConfig
 from deerflow.runtime import DisconnectMode, RunManager, RunStatus, ThreadOperationKind
+from deerflow.runtime.events.store.memory import MemoryRunEventStore
 from deerflow.runtime.runs.manager import CancelOutcome, ConflictError, PersistenceRetryPolicy, RunStartOutcome
 from deerflow.runtime.runs.store.memory import MemoryRunStore
 
@@ -506,6 +507,66 @@ async def test_reconcile_orphaned_inflight_runs_marks_stale_rows_error():
     }
 
 
+@pytest.mark.anyio
+async def test_reconcile_orphaned_run_backfills_delivery_after_atomic_takeover():
+    """Lease recovery must durably backfill the terminal receipt exactly once."""
+    store = MemoryRunStore()
+    events = MemoryRunEventStore()
+    await store.put("running-run", thread_id="thread-1", status="running", created_at="2026-01-01T00:00:00+00:00")
+    manager = RunManager(store=store, event_store=events)
+
+    first = await manager.reconcile_orphaned_inflight_runs(error="worker crashed", before="2026-01-01T00:00:01+00:00")
+    second = await manager.reconcile_orphaned_inflight_runs(error="worker crashed", before="2026-01-01T00:00:01+00:00")
+
+    assert [record.run_id for record in first] == ["running-run"]
+    assert second == []
+    delivery = await events.list_events("thread-1", "running-run", event_types=["run.delivery"])
+    assert len(delivery) == 1
+    assert delivery[0]["content"] == {"presented": 0, "paths": [], "by_tool": {}}
+    assert (await store.get("running-run"))["status"] == "error"
+
+
+@pytest.mark.anyio
+async def test_reconcile_preserves_delivery_written_before_worker_crash():
+    """A crash after the receipt but before status persistence keeps its facts."""
+    store = MemoryRunStore()
+    events = MemoryRunEventStore()
+    await store.put("running-run", thread_id="thread-1", status="running", created_at="2026-01-01T00:00:00+00:00")
+    await events.put_if_absent(
+        thread_id="thread-1",
+        run_id="running-run",
+        event_type="run.delivery",
+        category="outputs",
+        content={"presented": 1, "paths": ["report.md"], "by_tool": {"present_files": ["report.md"]}},
+    )
+    manager = RunManager(store=store, event_store=events)
+
+    recovered = await manager.reconcile_orphaned_inflight_runs(error="worker crashed", before="2026-01-01T00:00:01+00:00")
+
+    assert [record.run_id for record in recovered] == ["running-run"]
+    delivery = await events.list_events("thread-1", "running-run", event_types=["run.delivery"])
+    assert len(delivery) == 1
+    assert delivery[0]["content"]["presented"] == 1
+
+
+@pytest.mark.anyio
+async def test_reconcile_preserves_terminal_takeover_when_delivery_backfill_fails():
+    """A receipt-store outage must not undo an atomically claimed orphan."""
+
+    class FailingReceiptStore(MemoryRunEventStore):
+        async def put_if_absent(self, **kwargs):
+            raise RuntimeError("event store unavailable")
+
+    store = MemoryRunStore()
+    await store.put("running-run", thread_id="thread-1", status="running", created_at="2026-01-01T00:00:00+00:00")
+    manager = RunManager(store=store, event_store=FailingReceiptStore())
+
+    recovered = await manager.reconcile_orphaned_inflight_runs(error="worker crashed", before="2026-01-01T00:00:01+00:00")
+
+    assert [record.run_id for record in recovered] == ["running-run"]
+    assert (await store.get("running-run"))["status"] == "error"
+
+
 @pytest.mark.anyio
 async def test_reconcile_orphaned_inflight_runs_skips_live_local_run():
     """Startup recovery should not mark an active row orphaned when this worker owns it."""

backend/tests/test_run_worker_delivery.py

@@ -0,0 +1,362 @@
+"""Worker-level regression tests for the terminal run.delivery event (#4272 slice 1)."""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+from uuid import uuid4
+
+import pytest
+from langchain_core.messages import AIMessage, ToolMessage
+from langgraph.types import Command
+
+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
+
+
+def _make_bridge():
+    return SimpleNamespace(publish=AsyncMock(), publish_end=AsyncMock(), cleanup=AsyncMock())
+
+
+async def _delivery_events(store: MemoryRunEventStore, thread_id: str, run_id: str) -> list[dict]:
+    events = await store.list_events(thread_id, run_id)
+    return [e for e in events if e["event_type"] == "run.delivery"]
+
+
+@pytest.mark.anyio
+async def test_delivery_event_records_present_files_paths_on_success():
+    run_manager = RunManager()
+    record = await run_manager.create("thread-1")
+    store = MemoryRunEventStore()
+
+    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={},
+    )
+    await asyncio.sleep(0)
+
+    delivery = await _delivery_events(store, "thread-1", record.run_id)
+    assert len(delivery) == 1
+    assert delivery[0]["content"]["presented"] == 1
+    assert delivery[0]["content"]["paths"] == ["/mnt/user-data/outputs/report.md"]
+    assert delivery[0]["content"]["by_tool"] == {"present_files": ["/mnt/user-data/outputs/report.md"]}
+    fetched = await run_manager.get(record.run_id)
+    assert fetched.status == RunStatus.success
+
+
+@pytest.mark.anyio
+async def test_delivery_event_presented_zero_without_artifact_production():
+    run_manager = RunManager()
+    record = await run_manager.create("thread-1")
+    store = MemoryRunEventStore()
+
+    class DummyAgent:
+        async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False):
+            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={},
+    )
+    await asyncio.sleep(0)
+
+    delivery = await _delivery_events(store, "thread-1", record.run_id)
+    assert len(delivery) == 1
+    assert delivery[0]["content"] == {"presented": 0, "paths": [], "by_tool": {}}
+    fetched = await run_manager.get(record.run_id)
+    assert fetched.status == RunStatus.success
+
+
+@pytest.mark.anyio
+async def test_delivery_event_is_singleton_across_goal_continuations(monkeypatch):
+    run_manager = RunManager()
+    record = await run_manager.create("thread-1")
+    store = MemoryRunEventStore()
+    stream_calls = 0
+    continuation_calls = 0
+
+    class ContinuingAgent:
+        async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False):
+            nonlocal stream_calls
+            stream_calls += 1
+            journal = config["context"]["__run_journal"]
+            tool_call_id = f"call_{stream_calls}"
+            journal._remember_current_run_tool_calls(
+                AIMessage(content="", tool_calls=[{"id": tool_call_id, "name": "present_files", "args": {}}]),
+                caller="lead_agent",
+            )
+            artifacts = ["/mnt/user-data/outputs/report.md"]
+            if stream_calls == 2:
+                artifacts.append("/mnt/user-data/outputs/appendix.md")
+            journal.on_tool_end(
+                Command(
+                    update={
+                        "artifacts": artifacts,
+                        "messages": [ToolMessage("Successfully presented files", tool_call_id=tool_call_id)],
+                    }
+                ),
+                run_id=uuid4(),
+            )
+            yield {"messages": []}
+
+    async def prepare_continuation(**kwargs):
+        nonlocal continuation_calls
+        continuation_calls += 1
+        if continuation_calls == 1:
+            return {"messages": []}
+        return None
+
+    monkeypatch.setattr("deerflow.runtime.runs.worker._prepare_goal_continuation_input", prepare_continuation)
+
+    await run_agent(
+        _make_bridge(),
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=None, event_store=store),
+        agent_factory=lambda *, config: ContinuingAgent(),
+        graph_input={},
+        config={},
+    )
+
+    delivery = await _delivery_events(store, "thread-1", record.run_id)
+    assert stream_calls == 2
+    assert len(delivery) == 1
+    assert delivery[0]["content"] == {
+        "presented": 2,
+        "paths": [
+            "/mnt/user-data/outputs/report.md",
+            "/mnt/user-data/outputs/appendix.md",
+        ],
+        "by_tool": {
+            "present_files": [
+                "/mnt/user-data/outputs/report.md",
+                "/mnt/user-data/outputs/appendix.md",
+            ]
+        },
+    }
+
+
+@pytest.mark.anyio
+async def test_delivery_event_emitted_exactly_once_on_error_path():
+    run_manager = RunManager()
+    record = await run_manager.create("thread-1")
+    store = MemoryRunEventStore()
+
+    class FailingAgent:
+        async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False):
+            raise RuntimeError("boom")
+            yield  # pragma: no cover - make this an async generator
+
+    await run_agent(
+        _make_bridge(),
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=None, event_store=store),
+        agent_factory=lambda *, config: FailingAgent(),
+        graph_input={},
+        config={},
+    )
+    await asyncio.sleep(0)
+
+    delivery = await _delivery_events(store, "thread-1", record.run_id)
+    assert len(delivery) == 1
+    assert delivery[0]["content"]["presented"] == 0
+    fetched = await run_manager.get(record.run_id)
+    assert fetched.status == RunStatus.error
+
+
+@pytest.mark.anyio
+async def test_delivery_is_durable_before_terminal_run_status():
+    events = MemoryRunEventStore()
+
+    class OrderingRunStore(MemoryRunStore):
+        async def update_status(self, run_id, status, *, error=None, stop_reason=None):
+            if status not in {"pending", "running"}:
+                receipt = await events.list_events("thread-1", run_id, event_types=["run.delivery"])
+                assert len(receipt) == 1
+            return await super().update_status(run_id, status, error=error, stop_reason=stop_reason)
+
+    run_store = OrderingRunStore()
+    run_manager = RunManager(store=run_store)
+    record = await run_manager.create("thread-1")
+
+    class DummyAgent:
+        async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False):
+            yield {"messages": []}
+
+    await run_agent(
+        _make_bridge(),
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=None, event_store=events),
+        agent_factory=lambda *, config: DummyAgent(),
+        graph_input={},
+        config={},
+    )
+
+    assert (await run_store.get(record.run_id))["status"] == "success"
+
+
+@pytest.mark.anyio
+async def test_delivery_write_retries_before_persisting_success():
+    class FlakyReceiptStore(MemoryRunEventStore):
+        def __init__(self):
+            super().__init__()
+            self.attempts = 0
+
+        async def put_if_absent(self, **kwargs):
+            self.attempts += 1
+            if self.attempts == 1:
+                raise RuntimeError("transient event store outage")
+            return await super().put_if_absent(**kwargs)
+
+    event_store = FlakyReceiptStore()
+    run_store = MemoryRunStore()
+    run_manager = RunManager(store=run_store)
+    record = await run_manager.create("thread-1")
+
+    class DummyAgent:
+        async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False):
+            yield {"messages": []}
+
+    await run_agent(
+        _make_bridge(),
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=None, event_store=event_store),
+        agent_factory=lambda *, config: DummyAgent(),
+        graph_input={},
+        config={},
+    )
+
+    assert event_store.attempts == 2
+    assert len(await _delivery_events(event_store, "thread-1", record.run_id)) == 1
+    assert (await run_store.get(record.run_id))["status"] == "success"
+
+
+@pytest.mark.anyio
+async def test_delivery_write_failure_preserves_real_durable_terminal_status():
+    class FailingReceiptStore(MemoryRunEventStore):
+        def __init__(self):
+            super().__init__()
+            self.attempts = 0
+
+        async def put_if_absent(self, **kwargs):
+            self.attempts += 1
+            raise RuntimeError("event store unavailable")
+
+    run_store = MemoryRunStore()
+    run_manager = RunManager(store=run_store)
+    record = await run_manager.create("thread-1")
+    event_store = FailingReceiptStore()
+
+    class DummyAgent:
+        async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False):
+            yield {"messages": []}
+
+    await run_agent(
+        _make_bridge(),
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=None, event_store=event_store),
+        agent_factory=lambda *, config: DummyAgent(),
+        graph_input={},
+        config={},
+    )
+
+    # A receipt outage must not let lease recovery rewrite a genuine success
+    # as an error. After bounded retries, preserve the worker's real outcome.
+    assert event_store.attempts > 1
+    assert record.status == RunStatus.success
+    assert (await run_store.get(record.run_id))["status"] == "success"
+
+
+@pytest.mark.anyio
+async def test_delivery_event_emitted_when_checkpoint_preflight_fails(monkeypatch):
+    run_manager = RunManager()
+    run_manager.update_run_completion = AsyncMock(wraps=run_manager.update_run_completion)
+    record = await run_manager.create("thread-1")
+    store = MemoryRunEventStore()
+    compatibility_check = AsyncMock(side_effect=RuntimeError("incompatible checkpoint"))
+    monkeypatch.setattr("deerflow.runtime.runs.worker.aensure_checkpoint_mode_compatible", compatibility_check)
+
+    def unexpected_agent_factory(**kwargs):
+        raise AssertionError("agent must not be built after preflight failure")
+
+    await run_agent(
+        _make_bridge(),
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=object(), event_store=store),
+        agent_factory=unexpected_agent_factory,
+        graph_input={},
+        config={},
+    )
+
+    delivery = await _delivery_events(store, "thread-1", record.run_id)
+    assert len(delivery) == 1
+    assert delivery[0]["content"] == {"presented": 0, "paths": [], "by_tool": {}}
+    fetched = await run_manager.get(record.run_id)
+    assert fetched.status == RunStatus.error
+    run_manager.update_run_completion.assert_not_awaited()
+
+
+@pytest.mark.anyio
+async def test_delivery_event_emitted_when_cancelled_waiting_for_prior_finalization(monkeypatch):
+    run_manager = RunManager()
+    run_manager.update_run_completion = AsyncMock(wraps=run_manager.update_run_completion)
+    record = await run_manager.create("thread-1")
+    store = MemoryRunEventStore()
+    monkeypatch.setattr(
+        run_manager,
+        "wait_for_prior_finalizing",
+        AsyncMock(side_effect=asyncio.CancelledError()),
+    )
+
+    def unexpected_agent_factory(**kwargs):
+        raise AssertionError("agent must not be built after preflight cancellation")
+
+    await run_agent(
+        _make_bridge(),
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=None, event_store=store),
+        agent_factory=unexpected_agent_factory,
+        graph_input={},
+        config={},
+    )
+
+    delivery = await _delivery_events(store, "thread-1", record.run_id)
+    assert len(delivery) == 1
+    assert delivery[0]["content"] == {"presented": 0, "paths": [], "by_tool": {}}
+    fetched = await run_manager.get(record.run_id)
+    assert fetched.status == RunStatus.interrupted
+    run_manager.update_run_completion.assert_not_awaited()