feat(runtime): publish owner-confirmed run.terminal evidence (#7134)

* feat(runtime): publish authoritative run terminal events

* fix(runtime): preserve terminal attribution diagnostics

* docs(runtime): consolidate terminal event docs and trim agent guidance

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -135,6 +135,11 @@ That prompt is intended for coding agents. It tells the agent to clone the repo
 
 With `run_events.backend: db`, caller-supplied ISO 8601 event timestamps are normalized to UTC before storage, preserving their instant when SQLite drops timezone information. Timestamps without an offset are treated as UTC. Previously stored timestamps are not rewritten.
 
+Run event streams include `run.terminal` for the owning worker's confirmed final
+Run status, while `run.end` continues to describe each root graph completion.
+Terminal publication is best effort; event persistence failure does not undo the
+Run outcome or prevent stream completion. See [Run Event Stream](backend/docs/RUN_EVENT_STREAM.md#authoritative-run-outcome).
+
 Operators can extend lead-agent, subagent, and DeerMem extraction prompts with
 literal prepend/append configuration without editing source templates. See
 [prompt overlays](backend/docs/CONFIGURATION.md#prompt-overlays).

backend/AGENTS.md

@@ -4,11 +4,7 @@
 
 The LangGraph backend provides sandbox execution, persistent memory, subagent delegation, and extensible tools, isolated per thread.
 
-**Architecture**:
+Service roles and ports: [root service topology](../AGENTS.md#service-topology).
-- **Gateway API** (port 8001): REST API plus embedded LangGraph-compatible agent runtime
-- **Frontend** (port 3000): Next.js web interface
-- **Nginx** (port 2026): Unified reverse proxy entry point
-- **Provisioner** (port 8002, optional in Docker dev): Started only when sandbox is configured for provisioner/Kubernetes mode
 
 **Runtime**:
 - `make dev`, Docker dev, and production all run the agent runtime in Gateway via `RunManager` + `run_agent()` + `StreamBridge` (`packages/harness/deerflow/runtime/`). Nginx exposes that runtime at `/api/langgraph/*` and rewrites it to Gateway's native `/api/*` routers.
@@ -26,20 +22,12 @@ The LangGraph backend provides sandbox execution, persistent memory, subagent de
 - Gateway MCP/skill updates write `extensions_config.json` at runtime. Production Compose mounts it read-write (`config.yaml` stays `:ro`); Helm seeds a writable home-volume copy. Hold both `extensions_config_write_lock` and the sidecar advisory `extensions_config_file_lock` during read-modify-write to prevent lost updates across workers. `atomic_write_extensions_config` uses temp-file replacement; only mount-point `EBUSY` permits a non-atomic in-place overwrite (crashes can truncate). Warn about this fallback once per target; propagate other errors. `get_extensions_config()` revalidates reads by file signature; see [config caching](packages/harness/deerflow/config/AGENTS.md). Tests: `tests/test_compose_extensions_config_writable.py`, `tests/test_extensions_config_atomic_write.py`, `tests/test_helm_extensions_config_writable.py`.
 - MCP cache reset scope and discovery fencing: [MCP guide](packages/harness/deerflow/mcp/AGENTS.md).
 
-**Project Structure** (relative to `backend/`; see the root guide for the full repo):
+**Project structure:** The [root repository map](../AGENTS.md#repository-map)
-
+indexes the stack. Backend code lives in `app/` (Gateway and channels),
-- `Makefile`: backend commands; `langgraph.json`: LangGraph Studio configuration.
+`packages/harness/deerflow/` (agent runtime), and `packages/extension-api/`
-- `packages/extension-api/`: public extension contracts (`deerflow_extension_api.*`).
+(public extension contracts). Tests, detailed docs, and standalone benchmarks
-- `packages/harness/deerflow/`: publishable framework (`deerflow.*`):
+live in `tests/`, `docs/`, and `scripts/benchmark/` respectively. Module-local
-  `agents/` (lead agent, middleware, memory, thread state), `sandbox/`, `subagents/`
+`AGENTS.md` files describe their implementation boundaries.
-  (builtins, executor, registry), `tools/builtins/` (file presentation, clarification,
-  images, skill review), `mcp/`, `integrations/` (Lark CLI skill pack), `extensions/`,
-  `models/`, `skills/`, `config/`, `community/` (search, fetch, scrape, images,
-  AIO sandbox), `reflection/`, `utils/`, and the embedded `client.py`.
-- `app/`: application layer (`app.*`), including `gateway/` (FastAPI app and routers)
-  and `channels/` (IM integrations).
-- `scripts/benchmark/`: standalone backend benchmarks; `tests/`: test suite;
-  `docs/`: backend documentation.
 
 ## Important Development Guidelines
 
@@ -220,20 +208,6 @@ frontend reconnect can deduplicate the user's message.
 - For lightweight config/utility modules, prefer pure unit tests with no external dependencies
 - If a module causes circular import issues in tests, add a `sys.modules` mock in `tests/conftest.py` (see existing example for `deerflow.subagents.executor`)
 
-```bash
-# Four duration-balanced offline shards
-make test
-
-# Run strict blocking-I/O tests
-make test-blocking-io
-
-# Live API tests: require config.yaml, credentials, and explicit opt-in
-make test-live
-
-# Run a specific test file
-PYTHONPATH=. uv run pytest tests/test_<feature>.py -v
-```
-
 Keep live tests opt-in via `DEER_FLOW_RUN_LIVE_TESTS=1`; guard POSIX-only
 markers with `os.name` for Windows collection.
 

backend/docs/README.md

@@ -19,7 +19,7 @@ This directory contains detailed documentation for the DeerFlow backend.
 | Document | Description |
 |----------|-------------|
 | [STREAMING.md](STREAMING.md) | Token-level streaming design: Gateway vs DeerFlowClient paths, `stream_mode` semantics, per-id dedup |
-| [RUN_EVENT_STREAM.md](RUN_EVENT_STREAM.md) | Persisted run event stream contract: envelope, producers, consumers, and known gaps |
+| [RUN_EVENT_STREAM.md](RUN_EVENT_STREAM.md) | Persisted run event stream: contract, terminal authority, evidence ordering, and known gaps |
 | [FILE_UPLOAD.md](FILE_UPLOAD.md) | File upload functionality |
 | [PATH_EXAMPLES.md](PATH_EXAMPLES.md) | Path types and usage examples |
 | [SANDBOX_MEMORY_PROFILING.md](SANDBOX_MEMORY_PROFILING.md) | Sandbox memory baseline and runtime comparison guide |

backend/docs/RUN_EVENT_STREAM.md

@@ -68,6 +68,106 @@ through run-event or specialized APIs:
 | `subagent` | Subagent lifecycle and step history. |
 | `workspace` | Workspace/output file-change evidence. |
 
+## Authoritative Run Outcome
+
+`run.terminal` (`trace`) records the owning Worker's confirmed final business
+outcome. It is a singleton identified by `(thread_id, run_id, event_type)` and
+confirmed through `put_if_absent()`. Content contains only `status`:
+`success`, `error`, `interrupted`, or `timeout`; graph outputs and error details
+are excluded. Metadata requires `protocol_version: 1` and a boolean `recovered`.
+Current Workers emit `recovered: false`; the marker describes evidence origin
+and grants no publication authority. Field definitions remain in the
+[JSON contract](../../contracts/run_event_stream_contract.json).
+
+`run.start` and `run.end` retain their root-invocation producers, counts,
+categories, callback timing, and opaque output contract. A Run may produce
+several `run.end` events with success metadata before its business outcome
+becomes error. Use `run.terminal` for the confirmed Run outcome.
+
+### Publication Authority
+
+Only the current local Run owner's attributable final Terminal Commit ACK may
+authorize normal publication, with the acknowledged outcome still matching.
+Graph completion, a same-status RunRow read, a store-only hydrated record,
+a legacy `None` write result, or a truthy store response cannot replace that
+proof. Without a backing RunStore, a confirmed local Manager transition is
+authoritative only within that process lifetime.
+
+Rollback's `interrupted` ACK is intermediate while restoration remains pending;
+publication waits for restoration and the final `error` ACK. Ownership loss,
+rejected or unconfirmed commits, and unknown predecessor writes fail closed.
+Reading a historical terminal row does not create local publication authority.
+
+### Evidence Ordering
+
+The Worker holds its finalizing barrier through this sequence:
+
+1. Finish callback-producing completion hooks and lifecycle observers.
+2. Freeze Journal producers and drain accepted writes through the existing
+   `_stop_evidence_producers()` / `_drain_evidence()` machinery.
+3. Validate final commit authority and confirm the terminal singleton.
+4. Capture the Evidence Receipt's sequence boundary and event count.
+5. Persist the Evidence Seal.
+6. Release the finalizing barrier, then publish SSE END.
+
+`finish_evidence(before_receipt=...)` holds the Journal operation lock through
+freeze/drain, terminal confirmation, and Receipt capture, serializing normal
+flush/close. All accepted, confirmed Journal evidence precedes a new terminal
+event. Unfinished or UNKNOWN writes prohibit publication and complete sealing;
+only proven-not-committed Journal batches may be retried. Force close invalidates
+bounds even during Receipt capture. Preserve the existing cancellation and
+quarantine behavior rather than replaying an unsettled batch.
+
+The confirmed event must lie inside the Seal's captured boundary. An event
+appended later cannot be claimed as covered by an earlier Seal; an event alone
+does not prove complete evidence. Barrier release before END preserves immediate
+next-message admission, including when later cleanup is slow.
+
+### Idempotence and Failure Semantics
+
+Confirmation validates the returned event's identity, category, status, version,
+required field types, positive sequence, and database user ownership:
+
+| Outcome | Observation and behavior |
+| --- | --- |
+| `CONFIRMED_NEW` | Insertion is confirmed and the returned event matches; Receipt capture may proceed. |
+| `CONFIRMED_EXISTING` | The existing event itself matches; accept idempotently and preserve its sequence and original source marker. |
+| `CONFLICT` | Identity or protocol fields contradict expectations or are malformed; record the conflict, do not overwrite, and do not seal as complete. |
+| `UNKNOWN` | The operation outcome cannot be established; preserve uncertainty and do not seal as complete. |
+
+`created=False` alone proves neither success nor conflict. Additional optional
+metadata may differ without conflicting. At most two atomic singleton attempts
+resolve a lost ACK while final authority remains valid; this bounded confirmation
+never replays a quarantined Journal batch. Cancellation cannot authorize a Receipt.
+
+Publication or Seal failure cannot rewrite the acknowledged Run business status.
+A confirmed event may remain stored with partial evidence if sealing fails.
+Event-write failure must preserve normal SSE completion and existing `/stream`
+and `/wait` missing-stream fallbacks, including Memory Event Store loss after
+restart. Terminal singleton confirmation does not deduplicate Redis END.
+
+### Compatibility and Storage Guarantees
+
+Historical sealed evidence without `run.terminal` remains valid under its original
+coverage contract. Consumers claiming this protocol must validate the event
+inside the captured boundary; a code upgrade or Seal state alone is insufficient.
+Later recovery appends do not silently upgrade that historical coverage.
+
+Confirmation is limited to the Store's declared failure and concurrency model:
+
+| Backend | Guarantee boundary |
+| --- | --- |
+| Memory | One event loop/process lifetime; no persistence across restart. |
+| JSONL | File-backed within the declared single-process mutation boundary; no multi-process guarantee. |
+| SQLite | Database-backed confirmation with per-thread in-process serialization; this does not establish PostgreSQL or multi-Gateway concurrency semantics. |
+| PostgreSQL | Database-backed confirmation with per-thread transaction advisory locking across independent connections/processes; concurrency validation requires real sessions. |
+
+Singleton insertion and RunRow terminal commit are separate transactions. A crash
+between them can leave a terminal Run without an event. Recovery publication,
+takeover backfill, historical migration, and durable outbox/reconciliation are
+outside the current owner-side implementation; eventual evidence after every
+terminal commit is not guaranteed. Missing events must not gate SSE recovery.
+
 ## Producers
 
 `RunJournal` emits callback-derived events:
@@ -158,10 +258,10 @@ nested representation is not currently identical across storage backends:
   `json.dumps(default=str)`, so nested values that are not directly JSON
   serializable are read back as strings.
 
-Consumers may use `run.end` as completion evidence, but must not depend on
+Consumers may use `run.end` as root-invocation completion evidence, but must
-backend-identical nested output values. Normalizing those values would be a
+not infer the final Run outcome or depend on backend-identical nested output
-separate runtime compatibility change rather than part of this current-state
+values. Normalizing those values would be a separate runtime compatibility
-contract.
+change rather than part of this current-state contract.
 
 `subagents/step_events.py::subagent_run_event()` maps streamed `task_*` chunks
 to persisted events. The worker batches them through `put_batch()`:

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

@@ -54,7 +54,7 @@ Checkpointer storage runs in one of two channel modes, selected by `checkpoint_c
 
 **Replay checkpoint lookup prefers lineage and degrades only for an explicitly missing legacy parent link.** Branch and regenerate paths first walk `parent_config`, which prevents a global chronological scan from selecting a sibling created by regeneration. `CheckpointParentMissingError` alone enables the bounded newest-first history fallback in `app/gateway/checkpoint_lineage.py`; cycles, dangling/non-addressable parents, target mismatches, and depth exhaustion raise `CheckpointLineageIntegrityError` and fail closed instead of selecting a sibling. The compatibility scans request 400 raw checkpoints so up to 200 duration-only entries do not consume the effective branch-history budget; the fallback scans oldest-to-newest internally, skips duration-only checkpoints, and accepts only checkpoints with an addressable id as the replay base. A source history with no discoverable pre-user checkpoint preserves the historical single-checkpoint branch behavior instead of rejecting the branch; regeneration remains unavailable for that inherited response. Existing single-checkpoint branches are not mutated by regenerate preparation, and no raw checkpoint tuple is copied across threads because delta state depends on ancestry and pending writes. Regenerate source-run lookup uses the current thread's exact event, then the server-stamped `run_id` on the copied human message, then verified RunManager content matching; it does not read parent-thread events. When an interrupted response was streamed but never checkpointed, regeneration accepts only the latest visible human message's server-stamped `run_id` after verifying that it belongs to the same thread and still has `interrupted` status. Storage or checkpoint-mode failures are not treated as a missing base and still fail closed.
 
-**A delta-mode run cannot fork; `runtime/runs/worker.py` linearizes the resume instead.** Resuming from an older checkpoint (regenerate, or any client-supplied `checkpoint`) forks the lineage, and delta state for a fork is not materializable: `BaseCheckpointSaver.get_delta_channel_history` — and the bespoke overrides in `InMemorySaver`/`PostgresSaver` — collect **every** `pending_writes` entry stored on each on-path ancestor, but a shared parent also carries the writes of the sibling child that was abandoned. Those writes replay into the fork, so the run starts from a message list still containing the answer it was supposed to replace (#4458: regenerating in a branched thread showed the superseded assistant message beside the new one after a reload; reproduced on postgres, sqlite, and the in-memory saver). Write-to-child ownership belongs to the upstream delta contract, so DeerFlow does not reimplement the walk: `_linearize_delta_checkpoint_resume` materializes the requested checkpoint's complete state and writes every channel onto the **current head** (which has no siblings) through the state mutation graph, using `Overwrite` for reducer channels and resetting newer head-only channels to their schema default (or `None` when no constructible default exists); it then drops the `checkpoint_id` selector and lets the run proceed linearly, while the abandoned turn stays in history as the rewritten head's ancestry. The worker holds `_checkpoint_thread_lock` across `_capture_rollback_point` and the optional linear rewrite, making the rollback snapshot and rewrite atomic with graph streaming and the preceding run's duration-metadata checkpoint write. Capture preserves the complete real pre-run state; cancel-with-rollback then linearly replaces the current delta head with that captured state rather than forking the now-shared pre-run checkpoint, so the abandoned turn is restored without replaying the resume sibling's writes. The worker also recomputes the current-run message boundary from the rewritten state and fails closed (an unreadable resume checkpoint raises rather than falling back to the corrupt fork). `full` mode keeps forking — its checkpoints carry complete `channel_values` and need no replay — so LangGraph branching semantics are unchanged there. Root namespace only; subgraph namespaces are left alone.
+**A delta-mode run cannot fork; `runtime/runs/worker.py` linearizes the resume instead.** An older-checkpoint resume creates siblings whose pending writes can replay into the fork, retaining a superseded answer. Write-to-child ownership belongs to the upstream delta contract, so DeerFlow does not reimplement the walk: `_linearize_delta_checkpoint_resume` materializes the requested checkpoint's complete state and writes every channel onto the **current head** (which has no siblings) through the state mutation graph, using `Overwrite` for reducer channels and resetting newer head-only channels to their schema default (or `None` when no constructible default exists); it then drops the `checkpoint_id` selector and lets the run proceed linearly, while the abandoned turn stays in history as the rewritten head's ancestry. The worker holds `_checkpoint_thread_lock` across `_capture_rollback_point` and the optional linear rewrite, making the rollback snapshot and rewrite atomic with graph streaming and the preceding run's duration-metadata checkpoint write. Capture preserves the complete real pre-run state; cancel-with-rollback then linearly replaces the current delta head with that captured state rather than forking the now-shared pre-run checkpoint, so the abandoned turn is restored without replaying the resume sibling's writes. The worker also recomputes the current-run message boundary from the rewritten state and fails closed (an unreadable resume checkpoint raises rather than falling back to the corrupt fork). `full` mode keeps forking — its checkpoints carry complete `channel_values` and need no replay — so LangGraph branching semantics are unchanged there. Root namespace only; subgraph namespaces are left alone.
 
 **Wholesale state replacement uses a state-only mutation graph + `Overwrite`.** `update_state` values pass through channel reducers (`add_messages` merge in full, append in delta), so replacing reducer values requires `Overwrite` rather than an ordinary update. Full-mode rollback and context compaction replace `messages`; delta resume and delta rollback replace every materialized channel and reset current-head-only channels to their schema default (or `None`). These writes go through `build_state_mutation_graph(as_node, mode, state_schema)`, and `state_schema` MUST be the thread's effective schema (`graph_state_schema(assistant_graph)`), because the base-ThreadState fallback silently discards written channels contributed by custom `AgentMiddleware.state_schema`. Channels absent from a full-mode fork write inherit the parent's channel blobs, so middleware channels survive rollback/compaction (locked by `test_rollback_preserves_middleware_contributed_channels` and `test_compact_thread_context_preserves_middleware_contributed_channels`). The compiled mutation graph has one no-op node (entry = finish) whose checkpoint machinery (channels/versions/metadata) is identical to the agent graph's but schedules no pending tasks, so the restored/compacted head stays idle instead of re-triggering the agent. Never hand-write checkpoints via `checkpointer.aput` for this; raw writers elsewhere must preserve checkpoint parentage — severed ancestry breaks delta replay (see `runtime/runs/worker.py` writer parenting and `checkpoint_patches.py`).
 
@@ -411,3 +411,10 @@ unknown or cancelled write outcomes cannot yield a receipt or be replayed. Only
 proven-not-committed batches may be retried. Normal close stops new evidence
 producers and drains already admitted cross-thread callbacks inside its operation
 lock before flushing and detaching; do not reacquire that lock through `flush()`.
+
+The Worker requires its own final terminal ACK for `run.terminal`; rollback's
+intermediate ACK and matching row reads grant no authority. Keep hooks → Journal
+freeze/drain → singleton confirmation → Receipt → Seal → barrier release → END.
+Force close invalidates Receipt bounds. Preserve ownership checks, four-state
+confirmation, and partial evidence on failure without changing business status
+or SSE liveness. See [Run outcome protocol](../../../../docs/RUN_EVENT_STREAM.md#authoritative-run-outcome).

backend/packages/harness/deerflow/runtime/events/catalog.py

@@ -58,6 +58,7 @@ class RunEventPattern:
 RUN_START_EVENT = RunEventDefinition("run.start", "trace")
 RUN_END_EVENT = RunEventDefinition("run.end", "outputs")
 RUN_ERROR_EVENT = RunEventDefinition("run.error", "error")
+RUN_TERMINAL_EVENT = RunEventDefinition("run.terminal", "trace")
 LLM_HUMAN_INPUT_EVENT = RunEventDefinition("llm.human.input", "message")
 LLM_AI_RESPONSE_EVENT = RunEventDefinition("llm.ai.response", "message")
 LLM_TOOL_RESULT_EVENT = RunEventDefinition("llm.tool.result", "message")
@@ -113,9 +114,11 @@ SUBAGENT_RUN_EVENT_DEFINITIONS = (
 )
 
 WORKSPACE_RUN_EVENT_DEFINITIONS = (WORKSPACE_CHANGES_EVENT,)
+WORKER_RUN_EVENT_DEFINITIONS = (RUN_TERMINAL_EVENT,)
 
 FIXED_RUN_EVENT_DEFINITIONS = (
     *JOURNAL_RUN_EVENT_DEFINITIONS,
     *SUBAGENT_RUN_EVENT_DEFINITIONS,
     *WORKSPACE_RUN_EVENT_DEFINITIONS,
+    *WORKER_RUN_EVENT_DEFINITIONS,
 )

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

@@ -103,6 +103,7 @@ class RunEventStore(abc.ABC):
         content: str | dict = "",
         metadata: dict | None = None,
         created_at: str | None = None,
+        user_id: str | None | _AutoSentinel = AUTO,
     ) -> tuple[dict, bool]:
         """Write one event unless this run already has the same event type.
 
@@ -110,6 +111,10 @@ class RunEventStore(abc.ABC):
         thread. Returns ``(record, created)``. This is the durability primitive
         used by terminal run receipts, whose recovery path may safely retry
         after a worker crash.
+
+        Explicit ``user_id`` stamps the owner for contextless worker writes.
+        Stores without user isolation accept it for interface parity. Existing
+        events retain their original owner and must be validated by the caller.
         """
 
     @abc.abstractmethod

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

@@ -293,6 +293,7 @@ class DbRunEventStore(RunEventStore):
         content="",
         metadata=None,
         created_at=None,
+        user_id: str | None | _AutoSentinel = AUTO,
     ):
         """Idempotently insert a run-scoped singleton event.
 
@@ -304,7 +305,7 @@ class DbRunEventStore(RunEventStore):
         """
         content, metadata = self._truncate_trace(category, content, metadata)
         db_content, metadata = self._content_to_db(content, metadata)
-        user_id = self._user_id_from_context()
+        user_id = self._user_id_from_context() if user_id is AUTO else user_id
         async with self._get_write_lock(thread_id):
             async with self._sf() as session:
                 async with session.begin():

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

@@ -286,6 +286,7 @@ class JsonlRunEventStore(RunEventStore):
         content="",
         metadata=None,
         created_at=None,
+        user_id: str | None | _AutoSentinel = AUTO,
     ):
         async def mutate():
             existing = await asyncio.to_thread(self._read_run_events, thread_id, run_id)

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

@@ -104,6 +104,7 @@ class MemoryRunEventStore(RunEventStore):
         content="",
         metadata=None,
         created_at=None,
+        user_id: str | None | _AutoSentinel = AUTO,
     ):
         # No await occurs between the lookup and append, so this is atomic for
         # the backend's documented single-event-loop concurrency model.

backend/packages/harness/deerflow/runtime/events/terminal.py

@@ -0,0 +1,90 @@
+"""Confirmation of the owning worker's Run terminal evidence."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from enum import StrEnum
+
+from deerflow.runtime.events.catalog import RUN_TERMINAL_EVENT
+from deerflow.runtime.events.store.base import RunEventStore
+from deerflow.runtime.events.store.db import DbRunEventStore
+
+TERMINAL_PROTOCOL_VERSION = 1
+TERMINAL_STATUSES = frozenset({"success", "error", "interrupted", "timeout"})
+
+
+class TerminalEventConfirmation(StrEnum):
+    CONFIRMED_NEW = "CONFIRMED_NEW"
+    CONFIRMED_EXISTING = "CONFIRMED_EXISTING"
+    CONFLICT = "CONFLICT"
+    UNKNOWN = "UNKNOWN"
+
+    @property
+    def confirmed(self) -> bool:
+        return self in (self.CONFIRMED_NEW, self.CONFIRMED_EXISTING)
+
+
+def _matches(event: object, *, thread_id: str, run_id: str, status: str, user_id: str | None, owner_required: bool) -> bool:
+    if not isinstance(event, dict):
+        return False
+    metadata = event.get("metadata")
+    return (
+        event.get("thread_id") == thread_id
+        and event.get("run_id") == run_id
+        and event.get("event_type") == RUN_TERMINAL_EVENT.event_type
+        and event.get("category") == RUN_TERMINAL_EVENT.category
+        and event.get("content") == {"status": status}
+        and isinstance(metadata, dict)
+        and type(metadata.get("protocol_version")) is int
+        and metadata["protocol_version"] == TERMINAL_PROTOCOL_VERSION
+        and type(metadata.get("recovered")) is bool
+        and type(event.get("seq")) is int
+        and event["seq"] > 0
+        and (not owner_required or "user_id" in event)
+        and ("user_id" not in event or event["user_id"] == user_id)
+    )
+
+
+async def confirm_terminal_event(
+    store: RunEventStore,
+    *,
+    thread_id: str,
+    run_id: str,
+    status: str,
+    user_id: str | None,
+    authorized: Callable[[], bool],
+) -> TerminalEventConfirmation:
+    """Confirm a singleton without replaying any Journal batch.
+
+    One bounded retry uses the store's atomic check/insert to resolve a lost
+    ACK. Both attempts require the same final commit authority. Cancellation
+    propagates to the worker's owned finalizer; it cannot authorize a receipt.
+    """
+    if status not in TERMINAL_STATUSES:
+        raise ValueError("Terminal evidence requires a final Run status")
+    for attempt in range(2):
+        if not authorized():
+            raise RuntimeError("Terminal event lacks an attributable final commit ACK")
+        try:
+            result = await store.put_if_absent(
+                thread_id=thread_id,
+                run_id=run_id,
+                event_type=RUN_TERMINAL_EVENT.event_type,
+                category=RUN_TERMINAL_EVENT.category,
+                content={"status": status},
+                metadata={"protocol_version": TERMINAL_PROTOCOL_VERSION, "recovered": False},
+                user_id=user_id,
+            )
+        except Exception:
+            if attempt == 0:
+                continue
+            return TerminalEventConfirmation.UNKNOWN
+        if not authorized():
+            raise RuntimeError("Terminal event confirmation lost its final commit authority")
+        if not isinstance(result, tuple) or len(result) != 2 or type(result[1]) is not bool:
+            return TerminalEventConfirmation.UNKNOWN
+        event, created = result
+        if not _matches(event, thread_id=thread_id, run_id=run_id, status=status, user_id=user_id, owner_required=isinstance(store, DbRunEventStore)):
+            return TerminalEventConfirmation.CONFLICT
+        return TerminalEventConfirmation.CONFIRMED_NEW if created else TerminalEventConfirmation.CONFIRMED_EXISTING
+    return TerminalEventConfirmation.UNKNOWN

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

@@ -1763,27 +1763,40 @@ class RunJournal(BaseCallbackHandler):
             await asyncio.sleep(0)
 
     async def _drain_evidence(self) -> None:
+        """Freeze and drain while the caller holds the operation lock."""
         if self._closed:
             return
         await self._stop_evidence_producers()
-        await self.flush()
+        await self._flush_locked()
 
-    async def finish_evidence(self):
+    async def finish_evidence(self, *, before_receipt: Callable[[], Awaitable[None]] | None = None):
         """Stop declared producers and drain accepted work before capturing bounds.
 
         This is independent of terminal business status and does not detach the
         journal. Proven uncommitted writes remain available for explicit retry;
         unknown outcomes stay quarantined and never yield a durable receipt.
+        The worker's optional terminal confirmation runs after drain and before
+        bounds capture, under the same lock as flush and normal close.
         Unsupported stores return no durable receipt.
         """
-        if self._closed:
+        async with self._operation_lock:
-            return None
+            if self._abandoned:
-        await self._drain_evidence()
+                raise RuntimeError(f"RunJournal for run {self.run_id} was force-closed; cannot finish evidence")
-        from deerflow.runtime.events.store.db import DbRunEventStore
+            if self._closed:
+                return None
+            await self._drain_evidence()
+            if before_receipt is not None:
+                await before_receipt()
+            if self._abandoned:
+                raise RuntimeError(f"RunJournal for run {self.run_id} was force-closed; cannot capture evidence bounds")
+            from deerflow.runtime.events.store.db import DbRunEventStore
 
-        if isinstance(self._store, DbRunEventStore):
+            if isinstance(self._store, DbRunEventStore):
-            return await self._store.evidence_receipt(self.thread_id, self.run_id)
+                receipt = await self._store.evidence_receipt(self.thread_id, self.run_id)
-        return None
+                if self._abandoned:
+                    raise RuntimeError(f"RunJournal for run {self.run_id} was force-closed during evidence bounds capture")
+                return receipt
+            return None
 
     def _detach_runtime_dependencies(self) -> None:
         """Drop every external or potentially cyclic run-scoped reference."""

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

@@ -238,7 +238,7 @@ def _store_supports_terminal_completion(store: RunStore) -> bool:
 
 @dataclass(frozen=True)
 class TerminalCommitProof:
-    """Process-local evidence returned by one completed durable write."""
+    """Process-local proof of an acknowledged terminal transition."""
 
     status: RunStatus
     error: str | None
@@ -528,6 +528,12 @@ class RunManager:
     def _has_final_terminal_ack(self, record: RunRecord) -> bool:
         return not record.rollback_finalization_pending and self._has_terminal_ack(record)
 
+    def terminal_event_status(self, record: RunRecord) -> str | None:
+        """Return an attributable final outcome for this local worker only."""
+        if self._runs.get(record.run_id) is not record or record.store_only or record.owner_worker_id != self._worker_id or record.status in (RunStatus.pending, RunStatus.running) or not self._has_final_terminal_ack(record):
+            return None
+        return record.status.value
+
     def can_release_terminal_barrier(self, record: RunRecord) -> bool:
         """Keep durable ownership barriers through rollback's final error ACK."""
         return not self.heartbeat_enabled or self._store is None or record.ownership_lost or self._has_final_terminal_ack(record)
@@ -548,10 +554,11 @@ class RunManager:
     def _acknowledge_terminal(self, record: RunRecord, outcome: TerminalCommitProof, staged_revision: int) -> None:
         if self._runs.get(record.run_id) is not record or record.ownership_lost:
             return
-        # Publish the acknowledged outcome without an await. An older
+        # Publish the acknowledged outcome without an await. With a backing
-        # interrupted ACK cannot supersede a newer staged/acknowledged error.
+        # store, an older interrupted ACK cannot supersede a newer error.
+        # Storeless transitions are synchronous and retain local mutability.
         previous = record.terminal_commit_proof
-        if previous is None or previous.status == RunStatus.interrupted or outcome.status != RunStatus.interrupted:
+        if self._store is None or previous is None or previous.status == RunStatus.interrupted or outcome.status != RunStatus.interrupted:
             record.terminal_commit_proof = outcome
         authoritative = record.terminal_commit_proof
         if authoritative.status != RunStatus.interrupted or record.terminal_status_revision == staged_revision:
@@ -632,11 +639,14 @@ class RunManager:
                 record.run_id,
             )
             return False
+        terminal = status not in (RunStatus.pending, RunStatus.running)
+        outcome = self._terminal_outcome(record, status, error=error, stop_reason=stop_reason)
+        staged_revision = record.terminal_status_revision
         if self._store is None:
+            if terminal and not record.store_only:
+                self._acknowledge_terminal(record, outcome, staged_revision)
             return True
         if self.heartbeat_enabled and not record.store_only:
-            terminal = status not in (RunStatus.pending, RunStatus.running)
-            outcome = self._terminal_outcome(record, status, error=error, stop_reason=stop_reason)
             if terminal and self._has_terminal_ack(record, outcome):
                 return True
 
@@ -675,6 +685,11 @@ class RunManager:
                 record.run_id,
                 lambda: self._store.update_status(record.run_id, status.value, error=error, stop_reason=stop_reason, **({"goal_verdict": record.goal_verdict} if record.goal_verdict is not None else {})),
             )
+            # Preserve legacy single-worker recovery below, but only an explicit
+            # successful write can authorize a new terminal event. A matching row
+            # or a legacy None return does not attribute the commit to us.
+            if terminal and updated is True and not record.store_only:
+                self._acknowledge_terminal(record, outcome, staged_revision)
             if updated is False:
                 # ``update_status`` is now guarded by ``status IN ('pending','running')``.
                 # False can mean either:
@@ -1358,7 +1373,13 @@ class RunManager:
                 )
                 return
             proof = record.terminal_commit_proof
-            if self.heartbeat_enabled and proof is not None and proof.status != RunStatus.interrupted and self._terminal_outcome(record, status, error=error, stop_reason=stop_reason, goal_verdict=goal_verdict) != proof:
+            if (
+                self.heartbeat_enabled
+                and self._store is not None
+                and proof is not None
+                and proof.status != RunStatus.interrupted
+                and self._terminal_outcome(record, status, error=error, stop_reason=stop_reason, goal_verdict=goal_verdict) != proof
+            ):
                 logger.info("Skipped status change after acknowledged terminal outcome for run %s", run_id)
                 return
             record.status = status

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

@@ -2081,14 +2081,35 @@ async def run_agent(
                 if not record.ownership_lost and journal is not None:
                     evidence_failure = None
                     try:
-                        evidence_receipt = await journal.finish_evidence()
+
+                        async def confirm_terminal() -> None:
+                            nonlocal evidence_failure
+                            from deerflow.runtime.events.terminal import confirm_terminal_event
+
+                            status = run_manager.terminal_event_status(record)
+                            if status is None:
+                                evidence_failure = "Terminal event lacks an attributable final commit ACK"
+                                raise RuntimeError(evidence_failure)
+                            confirmation = await confirm_terminal_event(
+                                event_store,
+                                thread_id=thread_id,
+                                run_id=run_id,
+                                status=status,
+                                user_id=record.user_id,
+                                authorized=lambda: run_manager.terminal_event_status(record) == status,
+                            )
+                            if not confirmation.confirmed:
+                                evidence_failure = f"Terminal event confirmation: {confirmation.value}."
+                                raise RuntimeError(evidence_failure)
+
+                        evidence_receipt = await journal.finish_evidence(before_receipt=confirm_terminal)
                         if evidence_receipt is not None:
                             if not evidence_execution_started:
                                 evidence_failure = "Lead graph execution did not begin; producer coverage is partial."
                             elif not await run_manager.seal_completed_evidence(run_id, evidence_receipt):
                                 evidence_failure = "The owning worker could not persist the journal seal."
                     except BaseException as exc:
-                        evidence_failure = f"Journal seal failed ({type(exc).__name__})."
+                        evidence_failure = evidence_failure or f"Journal seal failed ({type(exc).__name__})."
                         if not isinstance(exc, Exception):
                             deferred_finalization_interrupt = _defer_finalization_interrupt(deferred_finalization_interrupt, exc)
                         logger.warning("Failed to seal completed-run evidence for %s (non-fatal)", run_id, exc_info=True)

backend/tests/AGENTS.md

@@ -18,6 +18,12 @@ ownership and capture cleanup. Retain the real XLSX conversion coverage.
 
 Backend tests must preserve the runtime invariants they exercise without changing production execution topology.
 
+`test_run_terminal_protocol.py` exercises authoritative `run.terminal`
+evidence through real Worker finalization. Keep final commit authority,
+producer drain, singleton confirmation, and Seal boundaries observable. Release all storage/producer gates before bounded cleanup,
+preserve healthy controls in refusal cases, and retain Journal/Seal compatibility
+coverage. `test_terminal_event_confirmation.py` covers supported stores; use
+`TEST_POSTGRES_URI` for independent PostgreSQL pools and real advisory contention. Protocol: [Run outcome](../docs/RUN_EVENT_STREAM.md#authoritative-run-outcome).
 `test_goal_loop_stop_probe.py` runs the standalone Goal-loop reproduction through the native graph, loop middleware, worker and checkpoints. Only model responses are scripted; do not replace the Goal decision or simulate the stop-reason write.
 
 InfoQuest fetch log assertions install the real `UrlRedactionFilter` on the

backend/tests/test_completed_run_evidence.py

@@ -893,3 +893,69 @@ async def test_commit_then_error_never_replays_or_issues_evidence_receipt(eviden
         assert (await reader.get_snapshot(thread_id="t-r", run_id="r")).seal_state != "sealed"
     finally:
         await journal.close(flush=False)
+
+
+@pytest.mark.asyncio
+async def test_force_close_during_receipt_cannot_return_confirmed_bounds(evidence, monkeypatch):
+    import asyncio
+
+    _sf, events, runs, _reader = evidence
+    await put_run(runs)
+    journal = RunJournal("r", "t-r", events, user_id="alice")
+    entered, release = asyncio.Event(), asyncio.Event()
+    original = events.evidence_receipt
+
+    async def held_receipt(*args):
+        receipt = await original(*args)
+        entered.set()
+        await release.wait()
+        return receipt
+
+    monkeypatch.setattr(events, "evidence_receipt", held_receipt)
+    finish = asyncio.create_task(journal.finish_evidence())
+    try:
+        await asyncio.wait_for(entered.wait(), timeout=5)
+        await journal.close(flush=False)
+    finally:
+        release.set()
+        await asyncio.wait_for(asyncio.gather(finish, return_exceptions=True), timeout=5)
+    with pytest.raises(RuntimeError, match="force-closed"):
+        finish.result()
+
+
+@pytest.mark.asyncio
+async def test_normal_close_waits_for_terminal_confirmation_before_detach(evidence):
+    import asyncio
+
+    _sf, events, runs, _reader = evidence
+    await put_run(runs)
+    journal = RunJournal("r", "t-r", events, user_id="alice")
+    journal.record_middleware("accepted", name="accepted", hook="end", action="done", changes={})
+    entered, release, closing = asyncio.Event(), asyncio.Event(), asyncio.Event()
+
+    async def confirm_tail():
+        entered.set()
+        await release.wait()
+        assert not journal._closed
+        await events.put_if_absent(thread_id="t-r", run_id="r", event_type="test.tail", category="trace", user_id="alice")
+
+    async def close():
+        closing.set()
+        await journal.close()
+
+    finish = asyncio.create_task(journal.finish_evidence(before_receipt=confirm_tail))
+    close_task = None
+    try:
+        await asyncio.wait_for(entered.wait(), timeout=5)
+        close_task = asyncio.create_task(close())
+        await asyncio.wait_for(closing.wait(), timeout=5)
+        assert not close_task.done()
+        assert not journal._closed
+    finally:
+        release.set()
+        await asyncio.wait_for(asyncio.gather(finish, *([close_task] if close_task else []), return_exceptions=True), timeout=5)
+    receipt = finish.result()
+    close_task.result()
+    assert receipt.event_count == 2
+    assert receipt.upper_event_seq == 2
+    assert journal._closed

backend/tests/test_run_event_stream_contract.py

@@ -577,3 +577,16 @@ def test_known_gaps_do_not_reclassify_current_events_as_missing():
 
     assert {"tool-call-intent", "terminal-run-status"}.issubset(gap_ids)
     assert all(gap.get("event_type") not in current_types for gap in contract["known_gaps"])
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("status", ["success", "error", "interrupted", "timeout"])
+async def test_authoritative_terminal_producer_matches_contract(status):
+    from deerflow.runtime.events.terminal import confirm_terminal_event
+
+    store = MemoryRunEventStore()
+    confirmation = await confirm_terminal_event(store, thread_id="thread", run_id="run", status=status, user_id=None, authorized=lambda: True)
+    assert confirmation.confirmed
+    records = await store.list_events("thread", "run")
+    assert len(records) == 1
+    _assert_fixed_event_valid(records[0], persisted=True)

backend/tests/test_run_terminal_protocol.py

@@ -0,0 +1,606 @@
+"""Behavioral regressions for authoritative Run terminal evidence.
+
+Pinned baseline: fee0cd0c180874325305e995197db99f03fa6a6f. These tests
+exercise owner-side run.terminal publication through actual finalization.
+See backend/docs/RUN_EVENT_STREAM.md for the protocol.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import threading
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+from typing import Annotated, TypedDict
+from uuid import uuid4
+
+import pytest
+import pytest_asyncio
+from langchain_core.messages import AnyMessage, HumanMessage
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.graph import END, START, StateGraph
+from langgraph.graph.message import add_messages
+from sqlalchemy import update
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+import deerflow.runtime.runs.worker as worker_module
+from deerflow.config.paths import Paths
+from deerflow.config.run_ownership_config import RunOwnershipConfig
+from deerflow.persistence.base import Base
+from deerflow.persistence.run.model import RunRow
+from deerflow.persistence.run.sql import RunRepository
+from deerflow.runtime.events.store.db import DbRunEventStore
+from deerflow.runtime.journal import RunJournal
+from deerflow.runtime.runs.manager import CancelOutcome, RunManager
+from deerflow.runtime.runs.schemas import RunStatus
+from deerflow.runtime.runs.worker import RunContext, run_agent
+from deerflow.runtime.stream_bridge import MemoryStreamBridge
+from deerflow.runtime.user_context import get_effective_user_id
+
+
+class RootAgent:
+    """Offline graph boundary; lifecycle callbacks and Worker remain real."""
+
+    def __init__(self, *, completions=1, after_graph=None):
+        self.completions = completions
+        self.after_graph = after_graph
+
+    async def astream(self, graph_input, config=None, **kwargs):
+        journal = config["context"]["__run_journal"]
+        for invocation in range(self.completions):
+            graph_id = uuid4()
+            journal.on_chain_start({"name": "offline-graph"}, {}, run_id=graph_id)
+            journal.on_chain_end({"invocation": invocation}, run_id=graph_id)
+        if self.after_graph is not None:
+            await self.after_graph(journal)
+        yield {"messages": []}
+
+
+class RollbackState(TypedDict):
+    messages: Annotated[list[AnyMessage], add_messages]
+
+
+async def settled(task, *, allow_cancelled=False):
+    # Shield the Worker: timeout must not cancel its owned finalizer before
+    # the test's finally block has released every storage/producer gate.
+    try:
+        return await asyncio.wait_for(asyncio.shield(task), timeout=5)
+    except asyncio.CancelledError:
+        if not allow_cancelled or not task.cancelled():
+            raise
+
+
+async def terminal_events(runtime, record):
+    return await runtime.events.list_events(record.thread_id, record.run_id, event_types=["run.terminal"], user_id=None)
+
+
+def confirmed_terminal(events, record, status):
+    assert len(events) == 1, f"expected one authoritative run.terminal({status}), got {events!r}"
+    terminal = events[0]
+    assert (terminal["thread_id"], terminal["run_id"], terminal["event_type"]) == (record.thread_id, record.run_id, "run.terminal")
+    assert terminal["category"] == "trace"
+    assert terminal["content"] == {"status": status}
+    assert terminal["metadata"]["protocol_version"] == 1
+    assert terminal["metadata"]["recovered"] is False
+    return terminal
+
+
+@pytest_asyncio.fixture
+async def runtime(tmp_path, monkeypatch):
+    """Real shared SQLite run/event stores, with bounded owned-task cleanup."""
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'terminal-protocol.db'}")
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    sf = async_sessionmaker(engine, expire_on_commit=False)
+    events, runs = DbRunEventStore(sf), RunRepository(sf)
+    paths = Paths(base_dir=tmp_path / "home")
+    monkeypatch.setattr("deerflow.workspace_changes.recorder.get_paths", lambda: paths)
+    workers, housekeeping = [], []
+    original_create = worker_module._create_contextless_task
+
+    def track_housekeeping(coro):
+        task = original_create(coro)
+        housekeeping.append(task)
+        return task
+
+    monkeypatch.setattr(worker_module, "_create_contextless_task", track_housekeeping)
+
+    async def start(*, agent=None, completion_hook=None, checkpointer=None, before_start=None, graph_input=None, heartbeat_enabled=True, user_id=None, event_store=None):
+        selected_events = event_store if event_store is not None else events
+        manager = RunManager(
+            store=runs,
+            event_store=selected_events,
+            worker_id=f"terminal-owner-{uuid4()}",
+            run_ownership_config=RunOwnershipConfig(heartbeat_enabled=heartbeat_enabled, lease_seconds=300, grace_seconds=0),
+        )
+        record = await manager.create_or_reject(f"terminal-thread-{uuid4()}", user_id=user_id)
+        if before_start is not None:
+            await before_start(record)
+        bridge = MemoryStreamBridge()
+        graph = agent if agent is not None else RootAgent()
+        task = asyncio.create_task(
+            run_agent(
+                bridge,
+                manager,
+                record,
+                ctx=RunContext(checkpointer=checkpointer, event_store=selected_events, on_run_completed=completion_hook),
+                agent_factory=lambda **kwargs: graph,
+                graph_input=graph_input or {},
+                config={"configurable": {"thread_id": record.thread_id}},
+            )
+        )
+        record.task = task
+        workers.append(task)
+        return SimpleNamespace(task=task, manager=manager, record=record, bridge=bridge)
+
+    try:
+        yield SimpleNamespace(sf=sf, events=events, runs=runs, paths=paths, start=start)
+    finally:
+        # Individual tests release their gates before this fallback cleanup.
+        for task in workers:
+            if not task.done():
+                task.cancel()
+        if workers:
+            await asyncio.wait_for(asyncio.gather(*workers, return_exceptions=True), timeout=5)
+        for task in housekeeping:
+            task.cancel()
+        await asyncio.gather(*housekeeping, return_exceptions=True)
+        await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_two_root_completions_keep_legacy_outputs_and_one_terminal(runtime):
+    """Catches graph completion being substituted for the final Run result."""
+    run = await runtime.start(agent=RootAgent(completions=2))
+    await settled(run.task)
+    row = await runtime.runs.get(run.record.run_id, user_id=run.record.user_id)
+    events = await runtime.events.list_events(run.record.thread_id, run.record.run_id)
+    assert row["status"] == "success"
+    assert run.manager._has_final_terminal_ack(run.record)
+    assert len([event for event in events if event["event_type"] == "run.start"]) == 2
+    legacy = [event for event in events if event["event_type"] == "run.end"]
+    assert [event["content"] for event in legacy] == [{"invocation": 0}, {"invocation": 1}]
+    assert all(event["category"] == "outputs" and event["metadata"]["status"] == "success" for event in legacy)
+    confirmed_terminal(await terminal_events(runtime, run.record), run.record, "success")
+
+
+@pytest.mark.asyncio
+async def test_graph_success_before_failed_artifact_delivery_has_error_terminal(runtime, monkeypatch):
+    """Catches terminal success published before real output verification."""
+    entered, release = asyncio.Event(), asyncio.Event()
+    original = worker_module._produced_output_paths
+    output_dir = None
+
+    async def prepare(record):
+        nonlocal output_dir
+        output_dir = runtime.paths.sandbox_outputs_dir(record.thread_id, user_id=get_effective_user_id())
+        output_dir.mkdir(parents=True)
+
+    async def produce(journal):
+        await journal.flush()
+        await asyncio.to_thread((output_dir / "report.md").write_text, "report\n", encoding="utf-8")
+
+    async def verify(*args, **kwargs):
+        result = await original(*args, **kwargs)
+        entered.set()
+        await release.wait()
+        return result
+
+    monkeypatch.setattr(worker_module, "_produced_output_paths", verify)
+    run = await runtime.start(agent=RootAgent(after_graph=produce), before_start=prepare)
+    try:
+        await asyncio.wait_for(entered.wait(), timeout=5)
+        legacy = await runtime.events.list_events(run.record.thread_id, run.record.run_id, event_types=["run.end"])
+        assert len(legacy) == 1 and legacy[0]["metadata"]["status"] == "success"
+        assert (await runtime.runs.get(run.record.run_id, user_id=run.record.user_id))["status"] == "running"
+        assert not await terminal_events(runtime, run.record)
+    finally:
+        release.set()
+        await settled(run.task)
+    row = await runtime.runs.get(run.record.run_id, user_id=run.record.user_id)
+    assert row["status"] == "error"
+    assert row["error"] == "Artifact delivery incomplete: no produced output artifact was presented"
+    delivery = await runtime.events.list_events(run.record.thread_id, run.record.run_id, event_types=["run.delivery"])
+    assert delivery[0]["content"]["produced_paths"] == ["/mnt/user-data/outputs/report.md"]
+    assert delivery[0]["content"]["satisfied"] is False
+    confirmed_terminal(await terminal_events(runtime, run.record), run.record, "error")
+
+
+@pytest.mark.asyncio
+async def test_peer_takeover_rejects_old_owner_and_healthy_owner_publishes(runtime, monkeypatch):
+    """Catches publication on rejected CAS; healthy control prevents vacuity."""
+    entered, release = asyncio.Event(), asyncio.Event()
+    original = runtime.runs.finalize_if_not_cancelled
+
+    async def gated_cas(*args, **kwargs):
+        entered.set()
+        await release.wait()
+        return await original(*args, **kwargs)
+
+    monkeypatch.setattr(runtime.runs, "finalize_if_not_cancelled", gated_cas)
+    run = await runtime.start()
+    try:
+        await asyncio.wait_for(entered.wait(), timeout=5)
+        assert (await runtime.runs.get(run.record.run_id, user_id=run.record.user_id))["status"] == "running"
+        assert run.record.terminal_commit_proof is None
+        expired = datetime.now(UTC) - timedelta(seconds=60)
+        async with runtime.sf() as session:
+            await session.execute(update(RunRow).where(RunRow.run_id == run.record.run_id).values(lease_expires_at=expired))
+            await session.commit()
+        run.record.lease_expires_at = expired.isoformat()
+        assert await runtime.runs.claim_for_takeover(run.record.run_id, grace_seconds=0, error="peer takeover", stop_reason="orphan_recovered")
+    finally:
+        release.set()
+        await settled(run.task, allow_cancelled=True)
+    assert (await runtime.runs.get(run.record.run_id, user_id=run.record.user_id))["error"] == "peer takeover"
+    assert run.record.ownership_lost
+    assert run.record.terminal_commit_proof is None
+    assert not await terminal_events(runtime, run.record)
+    healthy = await runtime.start()
+    await settled(healthy.task)
+    assert healthy.manager._has_final_terminal_ack(healthy.record)
+    confirmed_terminal(await terminal_events(runtime, healthy.record), healthy.record, "success")
+
+
+@pytest.mark.asyncio
+async def test_rollback_intermediate_ack_waits_for_restoration_and_final_error(runtime, monkeypatch):
+    """Catches interrupted ACK being published while rollback is unfinished."""
+    executing, restoring, release = asyncio.Event(), asyncio.Event(), asyncio.Event()
+    saver = InMemorySaver()
+    restored = []
+
+    async def work(_state):
+        executing.set()
+        await asyncio.Event().wait()
+        return {"messages": []}
+
+    builder = StateGraph(RollbackState)
+    builder.add_node("work", work)
+    builder.add_edge(START, "work")
+    builder.add_edge("work", END)
+    graph = builder.compile(checkpointer=saver)
+    original = worker_module._rollback_to_pre_run_checkpoint
+
+    async def gated_restore(**kwargs):
+        restoring.set()
+        await release.wait()
+        result = await original(**kwargs)
+        restored.append(result)
+        return result
+
+    async def seed(record):
+        await graph.aupdate_state({"configurable": {"thread_id": record.thread_id}}, {"messages": [HumanMessage(content="original", id="before")]}, as_node="work")
+
+    monkeypatch.setattr(worker_module, "_rollback_to_pre_run_checkpoint", gated_restore)
+    run = await runtime.start(agent=graph, checkpointer=saver, before_start=seed, graph_input={"messages": [HumanMessage(content="abandoned", id="during")]})
+    try:
+        await asyncio.wait_for(executing.wait(), timeout=5)
+        assert await run.manager.cancel(run.record.run_id, action="rollback") == CancelOutcome.cancelled
+        await asyncio.wait_for(restoring.wait(), timeout=5)
+        assert run.record.terminal_commit_proof.status == RunStatus.interrupted
+        assert run.record.rollback_finalization_pending
+        assert not run.manager._has_final_terminal_ack(run.record)
+        assert (await runtime.runs.get(run.record.run_id, user_id=run.record.user_id))["status"] == "interrupted"
+        assert not await terminal_events(runtime, run.record)
+    finally:
+        release.set()
+        await settled(run.task)
+    assert restored == [True]
+    snapshot = await graph.aget_state({"configurable": {"thread_id": run.record.thread_id}})
+    assert [(message.id, message.content) for message in snapshot.values["messages"]] == [("before", "original")]
+    assert run.manager._has_final_terminal_ack(run.record)
+    assert run.record.terminal_commit_proof.status == RunStatus.error
+    assert (await runtime.runs.get(run.record.run_id, user_id=run.record.user_id))["status"] == "error"
+    confirmed_terminal(await terminal_events(runtime, run.record), run.record, "error")
+
+
+@pytest.mark.asyncio
+async def test_unknown_journal_write_blocks_complete_tail_with_healthy_control(runtime, monkeypatch):
+    """Catches a terminal/complete seal overtaking an UNKNOWN predecessor."""
+    entered, release = asyncio.Event(), asyncio.Event()
+    original = runtime.events.put_batch
+    attempts = []
+
+    async def commit_then_lose_ack(batch):
+        result = await original(batch)
+        if any(event["event_type"] == "middleware:unknown" for event in batch):
+            attempts.append(batch)
+            entered.set()
+            await release.wait()
+            raise OSError("Journal commit acknowledgement lost")
+        return result
+
+    async def unknown_callback(journal):
+        journal.record_middleware("unknown", name="unknown", hook="test", action="accepted", changes={})
+
+    monkeypatch.setattr(runtime.events, "put_batch", commit_then_lose_ack)
+    run = await runtime.start(agent=RootAgent(after_graph=unknown_callback))
+    try:
+        await asyncio.wait_for(entered.wait(), timeout=5)
+        persisted = await runtime.events.list_events(run.record.thread_id, run.record.run_id)
+        assert any(event["event_type"] == "middleware:unknown" for event in persisted)
+        assert not await terminal_events(runtime, run.record)
+        assert (await runtime.runs.get(run.record.run_id, user_id=run.record.user_id))["evidence_seal_state"] != "sealed"
+    finally:
+        release.set()
+        await settled(run.task)
+    assert len(attempts) == 1, "an ambiguously committed batch must never be replayed"
+    row = await runtime.runs.get(run.record.run_id, user_id=run.record.user_id)
+    assert row["status"] == "success", "evidence failure must not rewrite business success"
+    assert row["evidence_seal_state"] == "partial"
+    assert not await terminal_events(runtime, run.record)
+    healthy = await runtime.start()
+    await settled(healthy.task)
+    confirmed_terminal(await terminal_events(runtime, healthy.record), healthy.record, "success")
+
+
+@pytest.mark.asyncio
+async def test_admitted_cross_thread_callback_precedes_terminal_after_freeze(runtime, monkeypatch):
+    """Catches terminal publication overtaking an admitted queued callback."""
+    freezing = asyncio.Event()
+    loop = asyncio.get_running_loop()
+    original_schedule = loop.call_soon_threadsafe
+    original_stop = RunJournal._stop_evidence_producers
+    held, journals, threads = [], [], []
+    holding = True
+    held_lock = threading.Lock()
+
+    def hold_callback(callback, *args, context=None):
+        method = getattr(callback, "func", None)
+        with held_lock:
+            if holding and journals and getattr(method, "__self__", None) is journals[0] and getattr(method, "__name__", None) == "_record_admitted_middleware":
+                held.append((callback, args, context))
+                return None
+        return original_schedule(callback, *args, context=context)
+
+    async def stop(journal):
+        if journals and journal is journals[0]:
+            freezing.set()
+        await original_stop(journal)
+
+    async def completion(_record):
+        # The hook itself is a legitimate evidence producer before freeze.
+        journal = journals[0]
+        thread = threading.Thread(target=lambda: journal.record_middleware("accepted", name="accepted", hook="stop", action="done", changes={}))
+        threads.append(thread)
+        thread.start()
+        await asyncio.to_thread(thread.join, 2)
+
+    async def capture(journal):
+        journals.append(journal)
+
+    def release_callback():
+        nonlocal holding
+        with held_lock:
+            holding = False
+            callbacks = held[:]
+            held.clear()
+        for callback, args, context in callbacks:
+            original_schedule(callback, *args, context=context)
+
+    monkeypatch.setattr(loop, "call_soon_threadsafe", hold_callback)
+    monkeypatch.setattr(RunJournal, "_stop_evidence_producers", stop)
+    run = await runtime.start(agent=RootAgent(after_graph=capture), completion_hook=completion)
+    try:
+        await asyncio.wait_for(freezing.wait(), timeout=5)
+        assert not threads[0].is_alive()
+        assert len(held) == 1
+        assert journals[0]._pending_producer_callbacks == 1
+        assert not journals[0]._accepting_evidence
+        assert not await terminal_events(runtime, run.record)
+        assert (await runtime.runs.get(run.record.run_id, user_id=run.record.user_id))["evidence_seal_state"] == "pending"
+    finally:
+        release_callback()
+        await settled(run.task)
+        for thread in threads:
+            await asyncio.to_thread(thread.join, 2)
+    events = await runtime.events.list_events(run.record.thread_id, run.record.run_id)
+    accepted = [event for event in events if event["event_type"] == "middleware:accepted"]
+    assert len(accepted) == 1
+    terminal = confirmed_terminal(await terminal_events(runtime, run.record), run.record, "success")
+    assert accepted[0]["seq"] < terminal["seq"]
+
+
+@pytest.mark.asyncio
+async def test_confirmed_terminal_is_inside_original_receipt_and_seal(runtime, monkeypatch):
+    """Catches Receipt before terminal ACK or terminal append after Receipt."""
+    entered, release = asyncio.Event(), asyncio.Event()
+    terminal_written, acknowledge = asyncio.Event(), asyncio.Event()
+    receipt_started = asyncio.Event()
+    original = runtime.events.evidence_receipt
+    original_put = runtime.events.put_if_absent
+    receipts, boundary_events, terminal_results = [], [], []
+
+    async def hold_terminal_ack(**kwargs):
+        result = await original_put(**kwargs)
+        if kwargs["event_type"] == "run.terminal":
+            terminal_results.append(result)
+            terminal_written.set()
+            await acknowledge.wait()
+        return result
+
+    async def capture_receipt(thread_id, run_id):
+        receipt_started.set()
+        receipt = await original(thread_id, run_id)
+        receipts.append(receipt)
+        boundary_events.extend(await runtime.events.list_events(thread_id, run_id))
+        entered.set()
+        await release.wait()
+        return receipt
+
+    monkeypatch.setattr(runtime.events, "evidence_receipt", capture_receipt)
+    monkeypatch.setattr(runtime.events, "put_if_absent", hold_terminal_ack)
+    run = await runtime.start()
+    ack_boundary = asyncio.create_task(terminal_written.wait())
+    receipt_boundary = asyncio.create_task(entered.wait())
+    try:
+        # Upstream has no publisher, so it reaches Receipt directly. Observe
+        # that boundary too instead of timing out waiting for a nonexistent API.
+        done, _ = await asyncio.wait({ack_boundary, receipt_boundary, run.task}, timeout=5, return_when=asyncio.FIRST_COMPLETED)
+        assert done, "Worker never reached either terminal write or Receipt"
+        if terminal_written.is_set():
+            assert not receipt_started.is_set(), "Receipt started while terminal write ACK was still withheld"
+            assert (await runtime.runs.get(run.record.run_id, user_id=run.record.user_id))["evidence_seal_state"] == "pending"
+            assert not receipt_started.is_set(), "Receipt started while terminal write ACK was still withheld"
+            acknowledge.set()
+        await asyncio.wait_for(asyncio.shield(receipt_boundary), timeout=5)
+        assert run.manager._has_final_terminal_ack(run.record)
+        assert (await runtime.runs.get(run.record.run_id, user_id=run.record.user_id))["evidence_seal_state"] == "pending"
+    finally:
+        acknowledge.set()
+        release.set()
+        try:
+            await settled(run.task)
+        finally:
+            ack_boundary.cancel()
+            receipt_boundary.cancel()
+            await asyncio.gather(ack_boundary, receipt_boundary, return_exceptions=True)
+    row = await runtime.runs.get(run.record.run_id, user_id=run.record.user_id)
+    assert row["evidence_seal_state"] == "sealed"
+    assert len(receipts) == 1
+    receipt = receipts[0]
+    assert row["evidence_upper_seq"] == receipt.upper_event_seq
+    assert row["evidence_event_count"] == receipt.event_count == len(boundary_events)
+    terminal = confirmed_terminal([event for event in boundary_events if event["event_type"] == "run.terminal"], run.record, "success")
+    assert len(terminal_results) == 1, "terminal evidence must use the run-scoped singleton confirmation"
+    assert terminal_results[0][0] == terminal
+    assert terminal["seq"] <= receipt.upper_event_seq
+    assert terminal == (await terminal_events(runtime, run.record))[0]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("backend", ["memory", "jsonl", "sqlite"])
+@pytest.mark.parametrize("heartbeat_enabled", [False, True])
+async def test_terminal_publication_supports_default_and_owned_workers(runtime, tmp_path, backend, heartbeat_enabled):
+    from deerflow.runtime.events.store.jsonl import JsonlRunEventStore
+    from deerflow.runtime.events.store.memory import MemoryRunEventStore
+
+    stores = {"memory": MemoryRunEventStore(), "jsonl": JsonlRunEventStore(base_dir=tmp_path / "events"), "sqlite": runtime.events}
+    store = stores[backend]
+    run = await runtime.start(event_store=store, heartbeat_enabled=heartbeat_enabled, user_id="alice")
+    await settled(run.task)
+    events = await store.list_events(run.record.thread_id, run.record.run_id, event_types=["run.terminal"], user_id="alice")
+    terminal = confirmed_terminal(events, run.record, "success")
+    if backend == "sqlite":
+        assert terminal["user_id"] == "alice"
+        assert not await store.list_events(run.record.thread_id, run.record.run_id, event_types=["run.terminal"], user_id="bob")
+
+
+@pytest.mark.asyncio
+async def test_terminal_ack_loss_confirms_existing_without_duplicate(runtime, monkeypatch):
+    original = runtime.events.put_if_absent
+    results = []
+
+    async def lose_first_ack(**kwargs):
+        result = await original(**kwargs)
+        if kwargs["event_type"] == "run.terminal":
+            results.append(result)
+            if len(results) == 1:
+                raise OSError("Terminal commit acknowledgement lost")
+        return result
+
+    monkeypatch.setattr(runtime.events, "put_if_absent", lose_first_ack)
+    run = await runtime.start()
+    await settled(run.task)
+    terminal = confirmed_terminal(await terminal_events(runtime, run.record), run.record, "success")
+    assert len(results) == 2
+    assert results[0] == (terminal, True)
+    assert results[1] == (terminal, False)
+    assert (await runtime.runs.get(run.record.run_id, user_id=run.record.user_id))["evidence_seal_state"] == "sealed"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("field,value", [("status", "error"), ("protocol_version", 2), ("protocol_version", True), ("category", "outputs"), ("recovered", "false"), ("user_id", "bob")])
+async def test_conflicting_terminal_preserves_record_and_business_status(runtime, field, value):
+    existing = []
+
+    async def seed(record):
+        event = dict(thread_id=record.thread_id, run_id=record.run_id, event_type="run.terminal", category="trace", content={"status": "success"}, metadata={"protocol_version": 1, "recovered": False}, user_id="alice")
+        if field == "status":
+            event["content"][field] = value
+        elif field in ("protocol_version", "recovered"):
+            event["metadata"][field] = value
+        else:
+            event[field] = value
+        existing.extend(await runtime.events.put_batch([event]))
+
+    run = await runtime.start(before_start=seed, user_id="alice")
+    await settled(run.task)
+    row = await runtime.runs.get(run.record.run_id, user_id=run.record.user_id)
+    assert row["status"] == "success"
+    assert row["evidence_seal_state"] == "partial"
+    assert "CONFLICT" in row["evidence_seal_error"]
+    assert await terminal_events(runtime, run.record) == existing
+    assert run.bridge._streams[run.record.run_id].ended
+    replacement = await run.manager.create_or_reject(run.record.thread_id, user_id="alice")
+    assert replacement.run_id != run.record.run_id
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("failure", ["unknown", "cancelled", "malformed"])
+async def test_unconfirmed_terminal_does_not_seal_or_block_stream_end(runtime, monkeypatch, failure):
+    original = runtime.events.put_if_absent
+    attempts = []
+
+    async def fail_terminal(**kwargs):
+        if kwargs["event_type"] == "run.terminal":
+            attempts.append(kwargs)
+            if failure == "unknown":
+                raise OSError("Terminal write unavailable")
+            if failure == "cancelled":
+                raise asyncio.CancelledError()
+            return None
+        return await original(**kwargs)
+
+    monkeypatch.setattr(runtime.events, "put_if_absent", fail_terminal)
+    run = await runtime.start()
+    try:
+        await settled(run.task, allow_cancelled=True)
+    finally:
+        assert run.task.done()
+    row = await runtime.runs.get(run.record.run_id, user_id=run.record.user_id)
+    assert row["status"] == "success"
+    assert row["evidence_seal_state"] == "partial"
+    assert row["evidence_seal_error"]
+    assert not await terminal_events(runtime, run.record)
+    assert run.bridge._streams[run.record.run_id].ended
+    assert len(attempts) == (2 if failure == "unknown" else 1)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("ack", [False, None])
+async def test_matching_terminal_row_without_commit_ack_cannot_authorize_event(runtime, monkeypatch, ack):
+    original = runtime.runs.update_status
+
+    async def lose_terminal_ack(run_id, status, **kwargs):
+        result = await original(run_id, status, **kwargs)
+        return ack if status in ("success", "error", "interrupted", "timeout") else result
+
+    monkeypatch.setattr(runtime.runs, "update_status", lose_terminal_ack)
+    run = await runtime.start(heartbeat_enabled=False)
+    await settled(run.task)
+    row = await runtime.runs.get(run.record.run_id, user_id=run.record.user_id)
+    assert row["status"] == "success"
+    assert not run.manager._has_final_terminal_ack(run.record)
+    assert not await terminal_events(runtime, run.record)
+    assert row["evidence_seal_state"] == "partial"
+    assert row["evidence_seal_error"] == "Terminal event lacks an attributable final commit ACK"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("heartbeat_enabled", [False, True])
+@pytest.mark.parametrize("status", [RunStatus.error, RunStatus.interrupted, RunStatus.timeout])
+async def test_process_local_terminal_proof_allows_later_staged_outcome(heartbeat_enabled, status):
+    manager = RunManager(run_ownership_config=RunOwnershipConfig(heartbeat_enabled=heartbeat_enabled))
+    record = await manager.create_or_reject("local-terminal")
+    await manager.set_status(record.run_id, RunStatus.success)
+    assert manager.terminal_event_status(record) == "success"
+
+    await manager.set_status(record.run_id, status, error="Later terminal outcome", persist=False)
+    assert record.status == status
+    assert record.error == "Later terminal outcome"
+    assert manager.terminal_event_status(record) is None
+
+    assert await manager.persist_current_status(record.run_id)
+    assert record.status == status
+    assert manager.terminal_event_status(record) == status.value

backend/tests/test_run_worker_delivery.py

@@ -790,9 +790,10 @@ async def test_delivery_write_retries_before_persisting_success():
             self.attempts = 0
 
         async def put_if_absent(self, **kwargs):
-            self.attempts += 1
+            if kwargs["event_type"] == "run.delivery":
-            if self.attempts == 1:
+                self.attempts += 1
-                raise RuntimeError("transient event store outage")
+                if self.attempts == 1:
+                    raise RuntimeError("transient event store outage")
             return await super().put_if_absent(**kwargs)
 
     event_store = FlakyReceiptStore()

backend/tests/test_run_worker_finalization_cancellation.py

@@ -84,7 +84,7 @@ async def reap(run):
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize("stage", ["receipt", "subagent_flush", "thread_status", "terminal_status", "journal_close", "end"])
+@pytest.mark.parametrize("stage", ["receipt", "terminal_event", "subagent_flush", "thread_status", "terminal_status", "journal_close", "end"])
 async def test_cancellation_preserves_terminal_lifecycle(monkeypatch, stage):
     entered, release, closed = asyncio.Event(), asyncio.Event(), asyncio.Event()
 
@@ -94,7 +94,7 @@ async def test_cancellation_preserves_terminal_lifecycle(monkeypatch, stage):
 
     class GatedEventStore(MemoryRunEventStore):
         async def put_if_absent(self, **kwargs):
-            if stage == "receipt" and kwargs.get("event_type") == "run.delivery":
+            if (stage == "receipt" and kwargs.get("event_type") == "run.delivery") or (stage == "terminal_event" and kwargs.get("event_type") == "run.terminal"):
                 await gate()
             return await super().put_if_absent(**kwargs)
 

backend/tests/test_scheduler_goal_runtime.py

@@ -172,6 +172,54 @@ async def test_scheduled_goal_is_installed_before_first_turn_and_removed_at_term
         assert persisted["status"] in {"error", "interrupted"}
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("heartbeat", [False, True])
+@pytest.mark.parametrize("with_events", [False, True])
+async def test_scheduled_cleanup_failure_updates_terminal_outcome_without_run_store(monkeypatch, heartbeat, with_events):
+    from deerflow.runtime.events.store.memory import MemoryRunEventStore
+
+    checkpointer = InMemorySaver()
+    events = MemoryRunEventStore() if with_events else None
+    manager = RunManager(event_store=events, run_ownership_config=RunOwnershipConfig(heartbeat_enabled=heartbeat))
+    metadata = {"scheduled_task_id": "task", "scheduled_task_run_id": "occurrence", "scheduled_goal_objective": "Make a verified report"}
+    record = await manager.create("fresh", user_id="alice", metadata=metadata)
+    graph = StateGraph(get_thread_state_schema("full"))
+    graph.add_node("report", lambda state: {"messages": [AIMessage(content="Verified report complete.")]})
+    graph.add_edge(START, "report")
+    graph.add_edge("report", END)
+    original_write = worker.write_thread_goal
+    cleanup_statuses = []
+
+    async def fail_cleanup(checkpointer, thread_id, value, **kwargs):
+        if value is None:
+            cleanup_statuses.append(record.status)
+            raise RuntimeError("deterministic scheduled cleanup failure")
+        return await original_write(checkpointer, thread_id, value, **kwargs)
+
+    monkeypatch.setattr(worker, "write_thread_goal", fail_cleanup)
+    monkeypatch.setattr(worker, "create_goal_evaluator_model", lambda **kwargs: object())
+    monkeypatch.setattr(worker, "evaluate_goal_completion", AsyncMock(return_value={"satisfied": True, "blocker": "none", "reason": "Verified", "relied_on_assumption": False}))
+    bridge = SimpleNamespace(publish=AsyncMock(), publish_end=AsyncMock(), cleanup=AsyncMock())
+    await worker.run_agent(
+        bridge,
+        manager,
+        record,
+        ctx=worker.RunContext(checkpointer=checkpointer, event_store=events, scheduled_task_runtime={"task_id": "task", "occurrence_id": "occurrence", "user_id": "alice", "goal_objective": metadata["scheduled_goal_objective"]}),
+        agent_factory=lambda config: graph.compile(checkpointer=checkpointer),
+        graph_input={"messages": [HumanMessage(content="Make the report")]},
+        config={"configurable": {"thread_id": "fresh"}, "context": {"non_interactive": True}},
+    )
+    assert cleanup_statuses == [RunStatus.success]
+    assert record.status == RunStatus.error
+    assert record.error == "Scheduled goal cleanup failed"
+    assert manager.terminal_event_status(record) == "error"
+    bridge.publish_end.assert_awaited_once_with(record.run_id)
+    if events is not None:
+        terminal = await events.list_events(record.thread_id, record.run_id, event_types=["run.terminal"])
+        assert len(terminal) == 1
+        assert terminal[0]["content"] == {"status": "error"}
+
+
 @pytest.mark.asyncio
 async def test_captured_verdict_counts_goal_continuations(monkeypatch):
     monkeypatch.setenv("LANGCHAIN_TRACING_V2", "false")

backend/tests/test_terminal_event_confirmation.py

@@ -0,0 +1,200 @@
+"""Terminal singleton confirmation across supported event stores."""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from copy import deepcopy
+from types import SimpleNamespace
+from uuid import uuid4
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import func, select, text
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from support.postgres import asyncpg_test_url
+
+from deerflow.persistence.models.run_event import RunEventRow, RunEventThreadSeqRow
+from deerflow.runtime.events.store.db import DbRunEventStore
+from deerflow.runtime.events.store.jsonl import JsonlRunEventStore
+from deerflow.runtime.events.store.memory import MemoryRunEventStore
+from deerflow.runtime.events.terminal import TerminalEventConfirmation, confirm_terminal_event
+
+
+@pytest_asyncio.fixture(params=["memory", "jsonl", "sqlite", "postgresql"])
+async def stores(request, tmp_path):
+    backend = request.param
+    if backend in ("memory", "jsonl"):
+        store = MemoryRunEventStore() if backend == "memory" else JsonlRunEventStore(base_dir=tmp_path)
+        yield SimpleNamespace(first=store, second=store, backend=backend)
+        return
+    schema = None
+    if backend == "postgresql":
+        uri = os.getenv("TEST_POSTGRES_URI")
+        if not uri:
+            pytest.skip("requires TEST_POSTGRES_URI for real PostgreSQL event confirmation")
+        schema = f"terminal_{uuid4().hex}"
+        url = asyncpg_test_url(uri)
+        options = {"connect_args": {"server_settings": {"search_path": schema}}}
+    else:
+        url = f"sqlite+aiosqlite:///{tmp_path / 'events.db'}"
+        options = {}
+    first_engine = create_async_engine(url, **options)
+    second_engine = create_async_engine(url, **options)
+    try:
+        async with first_engine.begin() as conn:
+            if schema:
+                await conn.execute(text(f'CREATE SCHEMA "{schema}"'))
+            await conn.run_sync(lambda c: RunEventRow.metadata.create_all(c, tables=[RunEventRow.__table__, RunEventThreadSeqRow.__table__]))
+        first = DbRunEventStore(async_sessionmaker(first_engine, expire_on_commit=False))
+        second = DbRunEventStore(async_sessionmaker(second_engine, expire_on_commit=False))
+        yield SimpleNamespace(first=first, second=second, backend=backend)
+    finally:
+        await second_engine.dispose()
+        if schema:
+            async with first_engine.begin() as conn:
+                await conn.execute(text(f'DROP SCHEMA "{schema}" CASCADE'))
+        await first_engine.dispose()
+
+
+async def confirm(store, status="success", *, authorized=lambda: True):
+    return await confirm_terminal_event(store, thread_id="thread", run_id="run", status=status, user_id="alice", authorized=authorized)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status", ["success", "error", "interrupted", "timeout"])
+async def test_confirmed_existing_retains_singleton_and_original_source(stores, status):
+    assert await confirm(stores.first, status) == TerminalEventConfirmation.CONFIRMED_NEW
+    first = await stores.first.list_events("thread", "run", user_id="alice")
+    assert len(first) == 1
+    assert first[0]["category"] == "trace"
+    assert first[0]["content"] == {"status": status}
+    assert first[0]["metadata"]["protocol_version"] == 1
+    assert first[0]["metadata"]["recovered"] is False
+    assert await confirm(stores.first, status) == TerminalEventConfirmation.CONFIRMED_EXISTING
+    assert await stores.first.list_events("thread", "run", user_id="alice") == first
+    assert not await stores.first.list_messages("thread", user_id="alice")
+
+
+@pytest.mark.asyncio
+async def test_changed_outcome_conflicts_instead_of_overwriting(stores):
+    assert (await confirm(stores.first)).confirmed
+    first = await stores.first.list_events("thread", "run", user_id="alice")
+    assert await confirm(stores.first, "error") == TerminalEventConfirmation.CONFLICT
+    assert await stores.first.list_events("thread", "run", user_id="alice") == first
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("field,value", [("run_id", "other"), ("thread_id", "other"), ("event_type", "run.end"), ("category", "outputs"), ("content", {"status": "success", "outputs": "private"}), ("seq", True), ("seq", 0)])
+async def test_invalid_returned_record_cannot_confirm(stores, monkeypatch, field, value):
+    original = stores.first.put_if_absent
+
+    async def corrupt_ack(**kwargs):
+        event, created = await original(**kwargs)
+        event = deepcopy(event)
+        event[field] = value
+        return event, created
+
+    monkeypatch.setattr(stores.first, "put_if_absent", corrupt_ack)
+    assert await confirm(stores.first) == TerminalEventConfirmation.CONFLICT
+    assert len(await stores.first.list_events("thread", "run", user_id="alice")) == 1
+
+
+@pytest.mark.asyncio
+async def test_ack_loss_uses_atomic_confirmation_without_duplicate(stores, monkeypatch):
+    original = stores.first.put_if_absent
+    attempts = []
+
+    async def lose_ack(**kwargs):
+        result = await original(**kwargs)
+        attempts.append(result)
+        if len(attempts) == 1:
+            raise OSError("ACK lost after commit")
+        return result
+
+    monkeypatch.setattr(stores.first, "put_if_absent", lose_ack)
+    assert await confirm(stores.first) == TerminalEventConfirmation.CONFIRMED_EXISTING
+    assert [created for _event, created in attempts] == [True, False]
+    assert attempts[0][0] == attempts[1][0]
+    assert len(await stores.first.list_events("thread", "run", user_id="alice")) == 1
+
+
+@pytest.mark.asyncio
+async def test_lost_authority_blocks_ack_loss_retry(stores, monkeypatch):
+    original = stores.first.put_if_absent
+    authorized = True
+    attempts = []
+
+    async def lose_ack(**kwargs):
+        nonlocal authorized
+        attempts.append(await original(**kwargs))
+        authorized = False
+        raise OSError("ACK lost after commit")
+
+    monkeypatch.setattr(stores.first, "put_if_absent", lose_ack)
+    with pytest.raises(RuntimeError, match="final commit ACK"):
+        await confirm(stores.first, authorized=lambda: authorized)
+    assert len(attempts) == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("second_status", ["success", "error"])
+async def test_postgres_independent_writers_contend_on_one_terminal(stores, monkeypatch, second_status):
+    if stores.backend != "postgresql":
+        pytest.skip("requires PostgreSQL independent connection pools and advisory locks")
+    first, second = stores.first, stores.second
+    held, attempted, release = asyncio.Event(), asyncio.Event(), asyncio.Event()
+    original_max = first._max_seq_for_thread
+    original_second_max = second._max_seq_for_thread
+    waiting_pid = None
+
+    async def hold_first(session, thread_id):
+        maximum = await original_max(session, thread_id)
+        held.set()
+        await release.wait()
+        return maximum
+
+    async def observe_second(session, thread_id):
+        nonlocal waiting_pid
+        waiting_pid = await session.scalar(select(func.pg_backend_pid()))
+        attempted.set()
+        return await original_second_max(session, thread_id)
+
+    monkeypatch.setattr(first, "_max_seq_for_thread", hold_first)
+    monkeypatch.setattr(second, "_max_seq_for_thread", observe_second)
+    first_task = asyncio.create_task(confirm(first))
+    second_task = None
+    try:
+        await asyncio.wait_for(held.wait(), timeout=5)
+        second_task = asyncio.create_task(confirm(second, second_status))
+        await asyncio.wait_for(attempted.wait(), timeout=5)
+        async with first._sf() as observer:
+            async with asyncio.timeout(5):
+                while not await observer.scalar(text("SELECT EXISTS (SELECT 1 FROM pg_locks WHERE pid=:pid AND locktype='advisory' AND NOT granted)"), {"pid": waiting_pid}):
+                    await asyncio.sleep(0.01)
+        assert not second_task.done()
+    finally:
+        release.set()
+        await asyncio.wait_for(asyncio.gather(first_task, *([second_task] if second_task else []), return_exceptions=True), timeout=5)
+    assert first_task.result() == TerminalEventConfirmation.CONFIRMED_NEW
+    assert second_task.result() == (TerminalEventConfirmation.CONFIRMED_EXISTING if second_status == "success" else TerminalEventConfirmation.CONFLICT)
+    terminal = await second.list_events("thread", "run", user_id="alice")
+    assert len(terminal) == 1
+    assert terminal[0]["content"] == {"status": "success"}
+    later = await second.put(thread_id="thread", run_id="run", event_type="test.later", category="trace")
+    assert later["seq"] > terminal[0]["seq"]
+
+
+@pytest.mark.asyncio
+async def test_database_ack_must_retain_owner_identity(stores, monkeypatch):
+    if not isinstance(stores.first, DbRunEventStore):
+        pytest.skip("only database records declare user ownership")
+    original = stores.first.put_if_absent
+
+    async def omit_owner(**kwargs):
+        event, created = await original(**kwargs)
+        event.pop("user_id")
+        return event, created
+
+    monkeypatch.setattr(stores.first, "put_if_absent", omit_owner)
+    assert await confirm(stores.first) == TerminalEventConfirmation.CONFLICT

contracts/run_event_stream_contract.json

@@ -106,6 +106,28 @@
         "additionalProperties": true
       }
     },
+    {
+      "event_type": "run.terminal",
+      "category": "trace",
+      "producer": "run_agent() finalization, after final commit ACK and Journal drain, before Evidence Receipt",
+      "identity": ["thread_id", "run_id", "event_type"],
+      "content_schema": {
+        "type": "object",
+        "required": ["status"],
+        "properties": {"status": {"enum": ["success", "error", "interrupted", "timeout"]}},
+        "additionalProperties": false
+      },
+      "metadata_schema": {
+        "type": "object",
+        "required": ["protocol_version", "recovered"],
+        "properties": {
+          "protocol_version": {"type": "integer", "const": 1},
+          "recovered": {"type": "boolean"}
+        },
+        "additionalProperties": true
+      },
+      "notes": "Owner-side terminal business outcome, confirmed as a run-scoped singleton. Current workers produce recovered=false. This does not redefine root-invocation run.end, prove complete producer coverage, or guarantee publication after a crash. Historical sealed records need not contain this additive event."
+    },
     {
       "event_type": "run.error",
       "category": "error",