feat(extensions): expose incremental run evidence reader (#5405)

* feat(extensions): expose incremental run evidence reader

* fix(extensions): address run evidence review feedback

* docs(extensions): clarify run deletion reconciliation

* docs(migrations): align current head documentation

* fix(extensions): isolate run evidence event reads

* test: avoid pinning run change migration to latest head

---------

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

README.md

@@ -1133,7 +1133,22 @@ libraries they import.
 
 DeerFlow allocates a task-scoped extension store only for middleware, lifecycle, or
 system-model observation. Services receive app-scoped runtime dependencies after Gateway
-persistence is ready and stop in reverse order after active runs drain. Extension HTTP
+persistence is ready and stop in reverse order after active runs drain. The optional
+`ExtensionRuntimeDeps.run_evidence_reader` is a stable, read-only interface for audit,
+evaluation, synchronization, and observability services: it discovers changed runs with an
+opaque resumable cursor, pages a known run's persisted events with `after_seq`, and reads the
+authoritative run status separately from event evidence. A database-backed run store keeps
+the discovery cursor valid across Gateway restarts; the memory run store provides the same
+ordering only for the current process lifetime. Event metadata is secret-redacted at this
+boundary, but event content is returned unchanged. Both payloads are detached snapshots,
+so modifying nested values cannot change the host's stored evidence. The production
+Gateway supplies an
+app-scoped, cross-user reader to trusted operator extensions; an embedded host may bind the
+same adapter to one user. Changed-run pages contain creations and changes to retained rows,
+not deletion tombstones; consumers that reconcile deletions must poll status for known runs
+and treat a missing result as absent. Extensions still execute with Gateway privileges and
+retain the legacy `session_factory`, so the reader is an API-stability and
+least-accidental-privilege boundary, not a sandbox for untrusted Python packages. Extension HTTP
 routers are mounted after every host route; definite shadows and routes entering the
 host's authentication- or CSRF-exempt paths are rejected with attributed diagnostics,
 while unrelated routers continue to load. Because the host's public paths are a reserved

backend/app/gateway/deps.py

@@ -479,6 +479,23 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen
             # cannot be validated there and are rejected by the middleware.
             app.state.pat_repo = None
 
+        # Evidence readers are available to Gateway-lifetime extension services,
+        # so the configured event store must exist before those services start.
+        run_events_config = getattr(config, "run_events", None)
+        app.state.run_events_config = run_events_config
+        app.state.run_event_store = make_run_event_store(run_events_config)
+
+        from deerflow.extensions.run_evidence import StoreRunEvidenceReader
+
+        # Gateway-lifetime services are trusted operator extensions without a
+        # request principal. None deliberately binds this app-scoped reader to
+        # global, cross-user visibility; event content is not secret-redacted.
+        app.state.run_evidence_reader = StoreRunEvidenceReader(
+            app.state.run_store,
+            app.state.run_event_store,
+            user_id=None,
+        )
+
         # Services are app-scoped. Capture this app's immutable extension set
         # once and close over the same object for teardown; the process-wide
         # singleton may be replaced by another app/test before shutdown.
@@ -504,6 +521,7 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen
                 extensions,
                 config,
                 sf,
+                run_evidence_reader=app.state.run_evidence_reader,
                 attempted_services=attempted_services,
             )
         )
@@ -540,14 +558,6 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen
             app.state.scheduled_task_repo = None
             app.state.scheduled_task_run_repo = None
 
-        # Run event store. The store and the matching ``run_events_config`` are
-        # both frozen at startup so ``get_run_context`` does not combine a
-        # freshly-reloaded ``AppConfig.run_events`` with a store still bound to
-        # the previous backend.
-        run_events_config = getattr(config, "run_events", None)
-        app.state.run_events_config = run_events_config
-        app.state.run_event_store = make_run_event_store(run_events_config)
-
         # RunManager with store backing for persistence
         run_ownership_config = getattr(config, "run_ownership", None)
         sb_config = getattr(config, "stream_bridge", None)

backend/packages/extension-api/deerflow_extension_api/__init__.py

@@ -61,6 +61,14 @@ from deerflow_extension_api.release import (
     canonical_json,
     collect_release_policies,
 )
+from deerflow_extension_api.run_evidence import (
+    InvalidRunEvidenceCursor,
+    RunEventPage,
+    RunEventView,
+    RunEvidenceReader,
+    RunPage,
+    RunStatusView,
+)
 from deerflow_extension_api.runtime_bridge import (
     EXTENSION_TASK_STORE_KEY,
     task_store_from_runtime,
@@ -69,7 +77,7 @@ from deerflow_extension_api.state import ExtensionData
 
 #: Contract version. Before 1.0, minors may break and patches are additive.
 #: From 1.0 on, bump the major for breaking changes.
-API_VERSION = "0.2.0"
+API_VERSION = "0.2.1"
 
 __all__ = [
     "API_VERSION",
@@ -93,12 +101,18 @@ __all__ = [
     "ExtensionRuntimeDeps",
     "ExtensionService",
     "HostPolicySnapshot",
+    "InvalidRunEvidenceCursor",
     "MessageProvenance",
     "MiddlewareContributor",
     "MiddlewareDescriptor",
     "MiddlewarePlacement",
     "Placement",
     "ReleasePolicyProvider",
+    "RunEvidenceReader",
+    "RunEventPage",
+    "RunEventView",
+    "RunPage",
+    "RunStatusView",
     "SystemModelCallObserver",
     "SystemModelRequest",
     "SystemModelResult",

backend/packages/extension-api/deerflow_extension_api/contracts.py

@@ -20,6 +20,7 @@ if TYPE_CHECKING:  # pragma: no cover - typing only
     from deerflow_extension_api.assembly import AgentAssemblyObserver
     from deerflow_extension_api.compaction import ContextCompactionObserver
     from deerflow_extension_api.placement import AgentBuildContext, MiddlewarePlacement
+    from deerflow_extension_api.run_evidence import RunEvidenceReader
 
 F = TypeVar("F", bound=Callable[..., Any])
 
@@ -164,6 +165,7 @@ class ExtensionRuntimeDeps:
     app_store: ExtensionData | None = None
     policy: HostPolicySnapshot = field(default_factory=HostPolicySnapshot)
     session_factory: Any | None = None
+    run_evidence_reader: RunEvidenceReader | None = None
 
 
 class ExtensionService(Protocol):

backend/packages/extension-api/deerflow_extension_api/run_evidence.py

@@ -0,0 +1,93 @@
+"""Stable, read-only run evidence contracts for extension services."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Protocol
+
+
+class InvalidRunEvidenceCursor(ValueError):
+    """The cursor is malformed, unsupported, or belongs to another scope."""
+
+
+@dataclass(frozen=True)
+class RunStatusView:
+    """Authoritative lifecycle projection for one visible run."""
+
+    thread_id: str = ""
+    run_id: str = ""
+    status: str = ""
+    created_at: str = ""
+    updated_at: str = ""
+    error: str | None = None
+    stop_reason: str | None = None
+
+
+@dataclass(frozen=True)
+class RunEventView:
+    """Persisted evidence envelope; ``seq`` is monotonic within its thread.
+
+    Reader-returned content and metadata are detached from host storage.
+    Fields are frozen, but nested payload containers may be modified locally.
+    """
+
+    thread_id: str = ""
+    run_id: str = ""
+    seq: int = 0
+    event_type: str = ""
+    category: str = ""
+    content: Any = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+    created_at: str = ""
+
+
+@dataclass(frozen=True)
+class RunPage:
+    """Changed runs plus the opaque cursor represented by this page."""
+
+    items: tuple[RunStatusView, ...] = ()
+    next_cursor: str | None = None
+    has_more: bool = False
+
+
+@dataclass(frozen=True)
+class RunEventPage:
+    """One forward page from a known run's persisted event stream."""
+
+    items: tuple[RunEventView, ...] = ()
+    next_after_seq: int | None = None
+    has_more: bool = False
+
+
+class RunEvidenceReader(Protocol):
+    """Host-bound evidence reader; implementations never expose writes."""
+
+    async def list_changed_runs(self, *, cursor: str | None, limit: int) -> RunPage:
+        """Return visible runs changed after ``cursor`` in stable order.
+
+        Callers persist ``next_cursor`` only after their own output is durable.
+        Reusing the input cursor is valid and may replay items. An empty page
+        means caught up; unsupported hosts omit the reader instead. Deletions
+        do not produce tombstones in this feed. Consumers reconciling a run
+        they already know must treat ``get_run_status(...) is None`` as absent.
+        """
+        raise NotImplementedError("the host does not provide changed-run discovery")
+
+    async def list_run_events(
+        self,
+        *,
+        thread_id: str,
+        run_id: str,
+        after_seq: int | None,
+        limit: int,
+    ) -> RunEventPage:
+        """Return events with thread-scoped ``seq > after_seq``.
+
+        A missing or invisible run returns an empty page, preventing identity
+        probing across the reader's host-bound scope.
+        """
+        raise NotImplementedError("the host does not provide run-event reading")
+
+    async def get_run_status(self, *, thread_id: str, run_id: str) -> RunStatusView | None:
+        """Return authoritative status, or ``None`` when not visible."""
+        raise NotImplementedError("the host does not provide run-status reading")

backend/packages/extension-api/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "deerflow-extension-api"
-version = "0.2.0"
+version = "0.2.1"
 description = "Public contracts for DeerFlow extensions"
 requires-python = ">=3.12"
 # Keep the contract package import-light and independent from the host. Public

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

@@ -277,7 +277,29 @@ supplies none.
 
 Gateway services start in registration order after the persistence engine and session
 factory are ready. Each receives the same `ExtensionRuntimeDeps` snapshot containing the
-app store, projected host policy, and session factory. Start failures are attributed and
+app store, projected host policy, session factory, and optional read-only
+`RunEvidenceReader`. The Gateway constructs the configured run and event stores before
+services so the reader is usable from `start()`. Changed-run discovery uses an opaque,
+scope-bound cursor over `(change_seq, run_id)`; a run that changes after it was returned may
+be replayed, but an unreturned run cannot be skipped. Legacy rows start at `change_seq=0`
+and sort by run id. Deletion is deliberately not represented by a tombstone, so the feed
+covers creations and changes to retained rows only; synchronization consumers must poll
+`get_run_status()` for known runs and treat `None` as absent when deletion reconciliation
+is required. A DB run store preserves positions across restarts, while memory only provides
+process-lifetime ordering. Per-run events retain the event store's thread-scoped
+`after_seq` semantics; metadata is secret-redacted, but event content is returned unchanged,
+and status comes from the authoritative run store. The reader passes its fixed scope to
+event reads explicitly, including global `None`, so ambient request identity cannot
+change its visibility. Content and redacted metadata are deep-copied snapshots: DTO
+fields are frozen, but nested containers remain locally mutable without touching host
+storage. The production Gateway injects one
+app-scoped reader with `user_id=None`, deliberately granting trusted operator extensions
+global cross-user visibility because services have no request principal. A host embedding
+the harness may instead bind a reader to one user. This is not a sandbox boundary: services
+already retain `session_factory` and execute with Gateway privileges. Empty pages mean
+caught up or not visible, never unsupported -- absence is represented
+by `ExtensionRuntimeDeps.run_evidence_reader is None`, and protocol defaults raise
+`NotImplementedError`. Start failures are attributed and
 fail open. The runtime captures `app.state.extensions` once, registers cleanup before the
 start batch, and stops the attempted service prefix in reverse order after run/subagent
 drain but before store, checkpointer, and engine teardown. Each stop has an independent

backend/packages/harness/deerflow/extensions/gateway.py

@@ -577,6 +577,7 @@ async def start_services(
     app_config: Any,
     session_factory: Any | None,
     *,
+    run_evidence_reader: Any | None = None,
     attempted_services: list[tuple[str, Any]] | None = None,
 ) -> list[Diagnostic]:
     """Start extension services in registration order, failing open per item."""
@@ -588,6 +589,7 @@ async def start_services(
         app_store=extensions.app_store,
         policy=project_host_policy(app_config),
         session_factory=session_factory,
+        run_evidence_reader=run_evidence_reader,
     )
     for entry in extensions.services:
         source, service = entry

backend/packages/harness/deerflow/extensions/run_evidence.py

@@ -0,0 +1,153 @@
+"""Host adapter for the public read-only run evidence contract."""
+
+from __future__ import annotations
+
+import base64
+import copy
+import hashlib
+import json
+from typing import Any
+
+from deerflow_extension_api import (
+    InvalidRunEvidenceCursor,
+    RunEventPage,
+    RunEventView,
+    RunPage,
+    RunStatusView,
+)
+
+from deerflow.runtime.secret_context import redact_metadata_secrets
+
+_CURSOR_VERSION = 1
+
+
+def _scope_fingerprint(user_id: str | None) -> str:
+    scope = user_id if user_id is not None else "*"
+    return hashlib.sha256(scope.encode("utf-8")).hexdigest()[:16]
+
+
+def _encode_cursor(change_seq: int, run_id: str, user_id: str | None) -> str:
+    payload = json.dumps(
+        {"q": change_seq, "r": run_id, "s": _scope_fingerprint(user_id), "v": _CURSOR_VERSION},
+        separators=(",", ":"),
+        sort_keys=True,
+    ).encode("utf-8")
+    return base64.urlsafe_b64encode(payload).decode("ascii").rstrip("=")
+
+
+def _decode_cursor(cursor: str | None, user_id: str | None) -> tuple[int, str]:
+    if cursor is None:
+        return -1, ""
+    try:
+        payload = json.loads(base64.urlsafe_b64decode(cursor + "=" * (-len(cursor) % 4)))
+    except (ValueError, TypeError, json.JSONDecodeError) as exc:
+        raise InvalidRunEvidenceCursor("invalid run evidence cursor") from exc
+    if not isinstance(payload, dict) or payload.get("v") != _CURSOR_VERSION:
+        raise InvalidRunEvidenceCursor("unsupported run evidence cursor")
+    if payload.get("s") != _scope_fingerprint(user_id):
+        raise InvalidRunEvidenceCursor("run evidence cursor scope does not match this reader")
+    change_seq, run_id = payload.get("q"), payload.get("r")
+    if not isinstance(change_seq, int) or isinstance(change_seq, bool) or not isinstance(run_id, str):
+        raise InvalidRunEvidenceCursor("invalid run evidence cursor")
+    return change_seq, run_id
+
+
+def _status_view(record: dict[str, Any]) -> RunStatusView:
+    return RunStatusView(
+        thread_id=str(record.get("thread_id") or ""),
+        run_id=str(record.get("run_id") or ""),
+        status=str(record.get("status") or ""),
+        created_at=str(record.get("created_at") or ""),
+        updated_at=str(record.get("updated_at") or ""),
+        error=record.get("error") if isinstance(record.get("error"), str) else None,
+        stop_reason=record.get("stop_reason") if isinstance(record.get("stop_reason"), str) else None,
+    )
+
+
+def _event_view(event: dict[str, Any]) -> RunEventView:
+    content = copy.deepcopy(event.get("content"))
+    metadata = copy.deepcopy(redact_metadata_secrets(event.get("metadata")))
+    return RunEventView(
+        thread_id=str(event.get("thread_id") or ""),
+        run_id=str(event.get("run_id") or ""),
+        seq=int(event.get("seq") or 0),
+        event_type=str(event.get("event_type") or ""),
+        category=str(event.get("category") or ""),
+        content=content,
+        metadata=metadata if isinstance(metadata, dict) else {},
+        created_at=str(event.get("created_at") or ""),
+    )
+
+
+class StoreRunEvidenceReader:
+    """Read configured stores through one immutable owner scope."""
+
+    def __init__(self, run_store: Any, event_store: Any, *, user_id: str | None = None) -> None:
+        self._run_store = run_store
+        self._event_store = event_store
+        self._user_id = user_id
+
+    @staticmethod
+    def _validate_limit(limit: int) -> None:
+        if not isinstance(limit, int) or isinstance(limit, bool) or not 1 <= limit <= 2000:
+            raise ValueError("limit must be an integer between 1 and 2000")
+
+    async def list_changed_runs(self, *, cursor: str | None, limit: int) -> RunPage:
+        self._validate_limit(limit)
+        after_seq, after_run_id = _decode_cursor(cursor, self._user_id)
+        records = await self._run_store.list_changed(
+            after_change_seq=after_seq,
+            after_run_id=after_run_id,
+            user_id=self._user_id,
+            limit=limit + 1,
+        )
+        page_records = records[:limit]
+        next_cursor = cursor
+        if page_records:
+            last = page_records[-1]
+            next_cursor = _encode_cursor(
+                int(last.get("change_seq") or 0),
+                str(last.get("run_id") or ""),
+                self._user_id,
+            )
+        return RunPage(
+            items=tuple(_status_view(record) for record in page_records),
+            next_cursor=next_cursor,
+            has_more=len(records) > limit,
+        )
+
+    async def get_run_status(self, *, thread_id: str, run_id: str) -> RunStatusView | None:
+        record = await self._run_store.get(run_id, user_id=self._user_id)
+        if record is None or record.get("thread_id") != thread_id or record.get("operation_kind", "run") != "run":
+            return None
+        return _status_view(record)
+
+    async def list_run_events(
+        self,
+        *,
+        thread_id: str,
+        run_id: str,
+        after_seq: int | None,
+        limit: int,
+    ) -> RunEventPage:
+        self._validate_limit(limit)
+        if after_seq is not None and (not isinstance(after_seq, int) or isinstance(after_seq, bool) or after_seq < 0):
+            raise ValueError("after_seq must be a non-negative integer or None")
+        if await self.get_run_status(thread_id=thread_id, run_id=run_id) is None:
+            return RunEventPage(next_after_seq=after_seq)
+        events = await self._event_store.list_events(
+            thread_id,
+            run_id,
+            limit=limit + 1,
+            after_seq=after_seq,
+            user_id=self._user_id,
+        )
+        page_events = events[:limit]
+        next_after_seq = after_seq
+        if page_events:
+            next_after_seq = int(page_events[-1]["seq"])
+        return RunEventPage(
+            items=tuple(_event_view(event) for event in page_events),
+            next_after_seq=next_after_seq,
+            has_more=len(events) > limit,
+        )

backend/packages/harness/deerflow/persistence/migrations/AGENTS.md

@@ -23,7 +23,8 @@ The empty-DB path keeps using `create_all` because `Base.metadata` is the only a
 `0018_oauth_identity_pg_partial` → `0019_projects` →
 `0020_threads_meta_project_id` → `0021_batch_acceptance` →
 `0019_thread_incarnations` → `0022_scheduled_occurrence_seq` →
-`0023_user_preferences` → `0024_project_documents` (current head). The preference
+`0023_run_change_seq` → `0023_user_preferences` →
+`0024_project_documents` (current head). The preference
 revision adds a separate owner/key table with a cascading users foreign key and
 does not alter users; the project-documents revision adds a new owner-scoped
 shelf table, so the bootstrap forward-compat floor is unchanged.
@@ -32,8 +33,8 @@ rollback-floor binary; Alembic orders revisions by `down_revision`, not by the
 numeric prefix.
 
 The deployed `0020_threads_meta_project_id` rollback-floor binary knows none of
-`0021_batch_acceptance`, `0019_thread_incarnations`, or
-`0022_scheduled_occurrence_seq`. It treats only the incarnation revision as
+`0021_batch_acceptance`, `0019_thread_incarnations`,
+`0022_scheduled_occurrence_seq`, or `0023_run_change_seq`. It treats only the incarnation revision as
 forward-compatible, after reflection confirms every
 table and column in its own ORM schema. The intervening acceptance columns and
 the incarnation columns are nullable and have no server default, so old
@@ -158,6 +159,7 @@ on installs that never enabled it. The convention is:
 - `migrations/versions/0021_batch_acceptance.py` — adds nullable per-item acceptance criteria and verdict JSON columns after `0020_threads_meta_project_id`; legacy rows remain unchecked
 - `migrations/versions/0019_thread_incarnations.py` — chains after `0021_batch_acceptance` while retaining the exact revision id audited by the rollback-floor binary. Adds nullable `threads_meta.incarnation` / `mcp_tasks.thread_incarnation` columns. New thread rows get a random 32-character incarnation. Memory mutations serialize per thread; an overwrite inherits the existing incarnation, while a delete/recreate gets a new one. SQLite MCP task INSERTs copy the owner-or-shared incarnation with a scalar subquery in the same statement. PostgreSQL task creation holds `FOR SHARE`, which conflicts with both current `FOR UPDATE` mutations and an older writer's plain owner update (`FOR NO KEY UPDATE`). Missing or differently owned threads store NULL, old writers may omit both columns, and current API/task serialization hides them. The migration preflights both tables before DDL and its SQLite downgrade cleans only safe remnants from its own interrupted batch-copy attempt
 - `migrations/versions/0022_scheduled_occurrence_seq.py` — adds the per-task `last_occurrence_seq` high-water mark, nullable occurrence `occurrence_seq` and `launch_accounted`, and a unique `(task_id, occurrence_seq)` index. New occurrences allocate their sequence under the existing parent lock; launch accounting is recorded atomically with the count so an older recovered occurrence cannot be counted twice. Legacy child columns remain NULL without guessed ordering or accounting backfill. All three fields are internal and omitted from repository responses. Both once-task recovery paths lock the parent, defer while any occurrence row is active (sequenced or not), and otherwise project only from the highest sequence (`can_project`), the same rule as the launch, completion, and queue-failure writes. Chains after `0019_thread_incarnations`.
+- `migrations/versions/0023_run_change_seq.py` — adds `runs.change_seq`, its global singleton allocation clock, and owner-aware cursor indexes. Legacy rows remain at zero and page by run id; lifecycle, cancellation, and model-name mutations allocate monotonically increasing positions in their own transaction. Atomic replacement uses one position for every affected row. Progress snapshots and lease heartbeats do not advance the clock. `0023_user_preferences` follows this revision. Its migration test verifies membership in the single-head chain and the expected predecessor rather than pinning the latest head, so later migrations can extend the chain.
 - `migrations/versions/0024_project_documents.py` — creates the `project_documents` shelf table (id/project_id/user_id/name/stored_relpath/sha256/size_bytes, nullable promotion provenance and trash fields, timestamps) with indexes on project_id, user_id, sha256 and trashed_at; no DB-level foreign key on project_id by design (project delete trashes the shelf inside its own transaction). New table, so the bootstrap forward-compat floor is unchanged; chains after `0023_user_preferences` (renumbered from 0023 after the rebase)
 - `persistence/bootstrap.py` — `bootstrap_schema(engine, backend=...)`, the three-branch provisioning decision, locked revision validation, and the narrow 0019 forward-compatibility exception
 - `extensions/loader.py::load_extensions` — registers each spec's `table_prefix` with `register_extension_table_prefix()`

backend/packages/harness/deerflow/persistence/migrations/versions/0023_run_change_seq.py

@@ -0,0 +1,52 @@
+"""Add stable changed-run discovery positions.
+
+Revision ID: 0023_run_change_seq
+Revises: 0022_scheduled_occurrence_seq
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0023_run_change_seq"
+down_revision: str | Sequence[str] | None = "0022_scheduled_occurrence_seq"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    from deerflow.persistence.migrations._helpers import safe_add_column
+
+    safe_add_column(
+        "runs",
+        sa.Column("change_seq", sa.BigInteger(), nullable=False, server_default=sa.text("0")),
+    )
+    inspector = sa.inspect(op.get_bind())
+    tables = set(inspector.get_table_names())
+    if "run_change_clock" not in tables:
+        op.create_table(
+            "run_change_clock",
+            sa.Column("id", sa.Integer(), primary_key=True),
+            sa.Column("value", sa.BigInteger(), nullable=False, server_default=sa.text("0")),
+        )
+    indexes = {index["name"] for index in sa.inspect(op.get_bind()).get_indexes("runs")}
+    if "ix_runs_change_seq" not in indexes:
+        op.create_index("ix_runs_change_seq", "runs", ["change_seq", "run_id"])
+    if "ix_runs_user_change_seq" not in indexes:
+        op.create_index("ix_runs_user_change_seq", "runs", ["user_id", "change_seq", "run_id"])
+
+
+def downgrade() -> None:
+    from deerflow.persistence.migrations._helpers import safe_drop_column
+
+    indexes = {index["name"] for index in sa.inspect(op.get_bind()).get_indexes("runs")}
+    if "ix_runs_user_change_seq" in indexes:
+        op.drop_index("ix_runs_user_change_seq", table_name="runs")
+    if "ix_runs_change_seq" in indexes:
+        op.drop_index("ix_runs_change_seq", table_name="runs")
+    safe_drop_column("runs", "change_seq")
+    if "run_change_clock" in set(sa.inspect(op.get_bind()).get_table_names()):
+        op.drop_table("run_change_clock")

backend/packages/harness/deerflow/persistence/migrations/versions/0023_user_preferences.py

@@ -4,7 +4,7 @@ import sqlalchemy as sa
 from alembic import op
 
 revision = "0023_user_preferences"
-down_revision = "0022_scheduled_occurrence_seq"
+down_revision = "0023_run_change_seq"
 branch_labels = None
 depends_on = None
 

backend/packages/harness/deerflow/persistence/run/model.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from datetime import UTC, datetime
 
-from sqlalchemy import JSON, DateTime, Index, String, Text, text
+from sqlalchemy import JSON, BigInteger, DateTime, Index, Integer, String, Text, text
 from sqlalchemy.orm import Mapped, mapped_column
 
 from deerflow.persistence.base import Base
@@ -57,11 +57,14 @@ class RunRow(Base):
 
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
     updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
+    change_seq: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0, server_default=text("0"))
 
     __table_args__ = (
         Index("ix_runs_thread_status", "thread_id", "status"),
         Index("ix_runs_lease", "lease_expires_at"),
         Index("uq_runs_idempotency_key", "idempotency_key", unique=True),
+        Index("ix_runs_change_seq", "change_seq", "run_id"),
+        Index("ix_runs_user_change_seq", "user_id", "change_seq", "run_id"),
         # Cross-process atomicity guarantee: at most one pending/running run per
         # thread. Must live in ORM ``__table_args__`` (not just the migration)
         # because the empty-DB bootstrap path runs ``create_all`` + ``stamp head``
@@ -74,3 +77,12 @@ class RunRow(Base):
             postgresql_where=text("status IN ('pending', 'running')"),
         ),
     )
+
+
+class RunChangeClockRow(Base):
+    """Singleton counter allocating backend-owned changed-run positions."""
+
+    __tablename__ = "run_change_clock"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    value: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0, server_default=text("0"))

backend/packages/harness/deerflow/persistence/run/sql.py

@@ -15,7 +15,7 @@ from sqlalchemy import and_, case, or_, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
-from deerflow.persistence.run.model import RunRow
+from deerflow.persistence.run.model import RunChangeClockRow, RunRow
 from deerflow.runtime.runs.store.base import (
     LeaseRenewal,
     RunIdempotencyConflict,
@@ -36,6 +36,21 @@ class RunRepository(RunStore):
     def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
         self._sf = session_factory
 
+    @staticmethod
+    async def _next_change_seq(session: AsyncSession) -> int:
+        dialect = session.bind.dialect.name
+        if dialect == "postgresql":
+            from sqlalchemy.dialects.postgresql import insert
+        elif dialect == "sqlite":
+            from sqlalchemy.dialects.sqlite import insert
+        else:  # pragma: no cover - configured databases are SQLite/Postgres
+            raise RuntimeError(f"unsupported run-change clock dialect: {dialect}")
+        await session.execute(insert(RunChangeClockRow).values(id=1, value=0).on_conflict_do_nothing(index_elements=[RunChangeClockRow.id]))
+        value = await session.scalar(update(RunChangeClockRow).where(RunChangeClockRow.id == 1).values(value=RunChangeClockRow.value + 1).returning(RunChangeClockRow.value))
+        if value is None:
+            raise RuntimeError("run-change clock did not return a position")
+        return int(value)
+
     @staticmethod
     def _normalize_model_name(model_name: str | None) -> str | None:
         """Normalize model_name for storage: strip whitespace, truncate to 128 chars."""
@@ -140,6 +155,7 @@ class RunRepository(RunStore):
             "updated_at": now,
         }
         async with self._sf() as session:
+            values["change_seq"] = await self._next_change_seq(session)
             row = await session.get(RunRow, run_id)
             if row is None:
                 session.add(RunRow(run_id=run_id, created_at=created, **values))
@@ -163,6 +179,29 @@ class RunRepository(RunStore):
                 return None
             return self._row_to_dict(row)
 
+    async def list_changed(
+        self,
+        *,
+        after_change_seq: int,
+        after_run_id: str,
+        user_id: str | None | _AutoSentinel = AUTO,
+        limit: int = 100,
+    ) -> list[dict[str, Any]]:
+        resolved_user_id = resolve_user_id(user_id, method_name="RunRepository.list_changed")
+        stmt = select(RunRow).where(
+            RunRow.operation_kind == "run",
+            or_(
+                RunRow.change_seq > after_change_seq,
+                and_(RunRow.change_seq == after_change_seq, RunRow.run_id > after_run_id),
+            ),
+        )
+        if resolved_user_id is not None:
+            stmt = stmt.where(RunRow.user_id == resolved_user_id)
+        stmt = stmt.order_by(RunRow.change_seq.asc(), RunRow.run_id.asc()).limit(limit)
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            return [self._row_to_dict(row) for row in result.scalars()]
+
     async def list_by_thread(
         self,
         thread_id,
@@ -269,6 +308,7 @@ class RunRepository(RunStore):
         # ``error`` and ``success`` remain locked so a peer's takeover (or a
         # completed run) cannot be overwritten by a late writer.
         async with self._sf() as session:
+            values["change_seq"] = await self._next_change_seq(session)
             result = await session.execute(update(RunRow).where(RunRow.run_id == run_id, RunRow.status.in_(("pending", "running", "interrupted"))).values(**values))
             await session.commit()
             return result.rowcount != 0
@@ -276,20 +316,30 @@ class RunRepository(RunStore):
     async def start_run(self, run_id: str) -> bool:
         """Start only a still-pending run; cancelled rows must not be resurrected."""
         async with self._sf() as session:
+            change_seq = await self._next_change_seq(session)
             result = await session.execute(
                 update(RunRow)
                 .where(
                     RunRow.run_id == run_id,
                     RunRow.status == "pending",
                 )
-                .values(status="running", updated_at=datetime.now(UTC))
+                .values(status="running", updated_at=datetime.now(UTC), change_seq=change_seq)
             )
             await session.commit()
             return result.rowcount != 0
 
     async def update_model_name(self, run_id, model_name):
         async with self._sf() as session:
-            await session.execute(update(RunRow).where(RunRow.run_id == run_id).values(model_name=self._normalize_model_name(model_name), updated_at=datetime.now(UTC)))
+            change_seq = await self._next_change_seq(session)
+            await session.execute(
+                update(RunRow)
+                .where(RunRow.run_id == run_id)
+                .values(
+                    model_name=self._normalize_model_name(model_name),
+                    updated_at=datetime.now(UTC),
+                    change_seq=change_seq,
+                )
+            )
             await session.commit()
 
     async def delete(
@@ -392,6 +442,7 @@ class RunRepository(RunStore):
         if status == "error" and "interrupted" not in allowed_sources:
             allowed_sources.append("interrupted")
         async with self._sf() as session:
+            values["change_seq"] = await self._next_change_seq(session)
             result = await session.execute(
                 update(RunRow)
                 .where(
@@ -574,6 +625,7 @@ class RunRepository(RunStore):
             raise ValueError(f"Unsupported cancellation action: {action}")
         now = datetime.now(UTC)
         async with self._sf() as session:
+            change_seq = await self._next_change_seq(session)
             result = await session.execute(
                 update(RunRow)
                 .where(
@@ -590,6 +642,7 @@ class RunRepository(RunStore):
                         else_=RunRow.cancel_requested_at,
                     ),
                     updated_at=now,
+                    change_seq=change_seq,
                 )
                 .returning(RunRow.cancel_action)
             )
@@ -616,6 +669,7 @@ class RunRepository(RunStore):
             values["stop_reason"] = stop_reason
 
         async with self._sf() as session:
+            values["change_seq"] = await self._next_change_seq(session)
             result = await session.execute(
                 update(RunRow)
                 .where(
@@ -655,6 +709,7 @@ class RunRepository(RunStore):
         if stop_reason is not None:
             values["stop_reason"] = stop_reason
         async with self._sf() as session:
+            values["change_seq"] = await self._next_change_seq(session)
             result = await session.execute(
                 update(RunRow)
                 .where(
@@ -750,6 +805,10 @@ class RunRepository(RunStore):
         }
 
         async with self._sf() as session:
+            # Keep the global clock -> run-row lock order used by every other
+            # mutator. One position covers this atomic set of changes; run_id
+            # provides deterministic ordering within the position.
+            change_seq = await self._next_change_seq(session)
             claimed: list[dict[str, Any]] = []
 
             if multitask_strategy in ("interrupt", "rollback"):
@@ -790,8 +849,10 @@ class RunRepository(RunStore):
                     row.error = "Cancelled by newer run"
                     row.owner_worker_id = owner_worker_id
                     row.updated_at = now
+                    row.change_seq = change_seq
                     claimed.append(self._row_to_dict(row))
 
+            values["change_seq"] = change_seq
             session.add(RunRow(run_id=run_id, **values))
             try:
                 await session.commit()

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

@@ -176,6 +176,32 @@ checkpoint-write admission boundary must repeat the complete audit after
 admission; a pre-admission exact hit can be superseded by a later event just as
 a pre-admission miss can become an exact hit.
 
+**Extension changed-run discovery** (`runtime/runs/store/` and
+`extensions/run_evidence.py`) orders public run-record changes by the
+backend-owned `(change_seq, run_id)` key rather than timestamps or per-thread
+event sequence numbers. The singleton SQL clock allocates positions in the same
+transaction as each public record mutation; memory uses a process-local counter.
+The clock row serializes position-bearing SQL mutations globally until their
+transactions commit, so every mutator must acquire it before any run-row lock.
+An atomic thread operation uses one position for its interrupted rows and new
+row, with `run_id` ordering ties. High-frequency progress snapshots and lease
+heartbeats do not advance this position: they are not lifecycle-discovery
+signals, progress evidence remains available from the event stream, and
+excluding them bounds clock contention. A later lifecycle change exposes the
+run row's latest `updated_at` and accumulated progress fields to internal readers.
+Rows from before the migration retain `change_seq=0` and page deterministically
+by run id. Because a later mutation only moves a row forward, concurrent paging
+may replay a run but cannot move an unseen run behind the committed cursor.
+Deletes produce no tombstone and therefore do not advance the cursor; consumers
+that require deletion reconciliation must poll authoritative status for known
+runs and treat a missing result as absent.
+The extension-facing cursor is versioned, opaque, and bound to the reader's
+fixed user scope. `RunEventStore.list_events()` accepts the same explicit
+`user_id` override as `list_messages()`; evidence readers must propagate their
+bound scope (including global `None`) rather than resolving an ambient request
+user. The adapter deep-copies event content and redacted metadata before
+exposing them, detaching nested mutable payloads from host storage.
+
 Gateway `POST /api/threads/{id}/history` uses that lookup to migrate legacy AI
 messages. An exhaustive miss preserves the human-boundary fallback; an
 incomplete lookup removes unproven synthesized IDs. Its metadata-only

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

@@ -194,6 +194,7 @@ class RunEventStore(abc.ABC):
         task_id: str | None = None,
         limit: int = 500,
         after_seq: int | None = None,
+        user_id: str | None | _AutoSentinel = AUTO,
     ) -> list[dict]:
         """Return the full event stream for a run, ordered by seq ascending.
 
@@ -201,7 +202,8 @@ class RunEventStore(abc.ABC):
         ``metadata["task_id"]``). ``after_seq`` is a forward cursor returning the
         first ``limit`` records with seq > after_seq, so callers can page through
         a single subagent task's events without the run-wide ``limit`` truncating
-        the tail (#3779).
+        the tail (#3779). ``user_id`` follows the same explicit-caller semantics
+        as :meth:`list_messages`.
         """
 
     @abc.abstractmethod

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

@@ -356,7 +356,17 @@ class JsonlRunEventStore(RunEventStore):
                 break
         return result
 
-    async def list_events(self, thread_id, run_id, *, event_types=None, task_id=None, limit=500, after_seq=None):
+    async def list_events(
+        self,
+        thread_id,
+        run_id,
+        *,
+        event_types=None,
+        task_id=None,
+        limit=500,
+        after_seq=None,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ):
         events = await asyncio.to_thread(self._read_run_events, thread_id, run_id)
         if event_types is not None:
             events = [e for e in events if e.get("event_type") in event_types]

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

@@ -140,7 +140,17 @@ class MemoryRunEventStore(RunEventStore):
             # Return the latest `limit` records, ascending.
             return messages[-limit:]
 
-    async def list_events(self, thread_id, run_id, *, event_types=None, task_id=None, limit=500, after_seq=None):
+    async def list_events(
+        self,
+        thread_id,
+        run_id,
+        *,
+        event_types=None,
+        task_id=None,
+        limit=500,
+        after_seq=None,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ):
         # ``_events_by_run`` is already scoped to this run and seq-ordered, so we
         # touch only this run's events instead of scanning the whole thread.
         run_events = self._events_by_run.get(thread_id, {}).get(run_id, [])

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

@@ -110,6 +110,17 @@ def run_is_before_cursor(
 
 
 class RunStore(abc.ABC):
+    async def list_changed(
+        self,
+        *,
+        after_change_seq: int,
+        after_run_id: str,
+        user_id: str | None = None,
+        limit: int = 100,
+    ) -> list[dict[str, Any]]:
+        """List public run-record changes in stable ascending cursor order."""
+        raise NotImplementedError
+
     @abc.abstractmethod
     async def put(
         self,

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

@@ -21,6 +21,7 @@ from deerflow.runtime.runs.store.base import (
 class MemoryRunStore(RunStore):
     def __init__(self) -> None:
         self._runs: dict[str, dict[str, Any]] = {}
+        self._change_seq = 0
         # Secondary index: thread_id -> insertion-ordered run_id set (a dict is
         # used as an ordered set), maintained in lockstep with ``_runs`` so
         # per-thread queries avoid O(total in-memory runs) full scans. Mirrors
@@ -39,6 +40,10 @@ class MemoryRunStore(RunStore):
             if not bucket:
                 self._runs_by_thread.pop(thread_id, None)
 
+    def _mark_changed(self, run: dict[str, Any]) -> None:
+        self._change_seq += 1
+        run["change_seq"] = self._change_seq
+
     async def put(
         self,
         run_id,
@@ -84,6 +89,7 @@ class MemoryRunStore(RunStore):
             "cancel_action": existing.get("cancel_action") if existing else None,
             "cancel_requested_at": existing.get("cancel_requested_at") if existing else None,
         }
+        self._mark_changed(self._runs[run_id])
         self._index_run(run_id, thread_id)
 
     async def get(self, run_id, *, user_id=None):
@@ -94,6 +100,19 @@ class MemoryRunStore(RunStore):
             return None
         return run
 
+    async def list_changed(
+        self,
+        *,
+        after_change_seq: int,
+        after_run_id: str,
+        user_id: str | None = None,
+        limit: int = 100,
+    ) -> list[dict[str, Any]]:
+        cursor = (after_change_seq, after_run_id)
+        results = [run for run in self._runs.values() if run.get("operation_kind", "run") == "run" and (user_id is None or run.get("user_id") == user_id) and (int(run.get("change_seq") or 0), run["run_id"]) > cursor]
+        results.sort(key=lambda run: (int(run.get("change_seq") or 0), run["run_id"]))
+        return results[:limit]
+
     async def list_by_thread(
         self,
         thread_id,
@@ -173,6 +192,7 @@ class MemoryRunStore(RunStore):
         if stop_reason is not None:
             run["stop_reason"] = stop_reason
         run["updated_at"] = datetime.now(UTC).isoformat()
+        self._mark_changed(run)
         return True
 
     async def start_run(self, run_id) -> bool:
@@ -181,12 +201,14 @@ class MemoryRunStore(RunStore):
             return False
         run["status"] = "running"
         run["updated_at"] = datetime.now(UTC).isoformat()
+        self._mark_changed(run)
         return True
 
     async def update_model_name(self, run_id, model_name):
         if run_id in self._runs:
             self._runs[run_id]["model_name"] = model_name
             self._runs[run_id]["updated_at"] = datetime.now(UTC).isoformat()
+            self._mark_changed(self._runs[run_id])
 
     async def delete(self, run_id, *, user_id=None):
         run = self._runs.pop(run_id, None)
@@ -208,6 +230,7 @@ class MemoryRunStore(RunStore):
             if value is not None:
                 run[key] = value
         run["updated_at"] = datetime.now(UTC).isoformat()
+        self._mark_changed(run)
         return True
 
     async def update_run_progress(self, run_id, **kwargs):
@@ -320,6 +343,7 @@ class MemoryRunStore(RunStore):
             run["cancel_action"] = action
             run["cancel_requested_at"] = datetime.now(UTC).isoformat()
         run["updated_at"] = datetime.now(UTC).isoformat()
+        self._mark_changed(run)
         return run["cancel_action"]
 
     async def finalize_if_not_cancelled(
@@ -346,6 +370,7 @@ class MemoryRunStore(RunStore):
         if stop_reason is not None:
             run["stop_reason"] = stop_reason
         run["updated_at"] = datetime.now(UTC).isoformat()
+        self._mark_changed(run)
         return StatusFinalization(finalized=True)
 
     async def claim_for_takeover(
@@ -371,6 +396,7 @@ class MemoryRunStore(RunStore):
         if stop_reason is not None:
             run["stop_reason"] = stop_reason
         run["updated_at"] = datetime.now(UTC).isoformat()
+        self._mark_changed(run)
         return True
 
     async def list_inflight_with_expired_lease(
@@ -493,6 +519,7 @@ class MemoryRunStore(RunStore):
                 r["error"] = "Cancelled by newer run"
                 r["owner_worker_id"] = owner_worker_id
                 r["updated_at"] = now
+                self._mark_changed(r)
                 claimed.append(r)
 
         new_row = {
@@ -516,5 +543,6 @@ class MemoryRunStore(RunStore):
             "updated_at": now,
         }
         self._runs[run_id] = new_row
+        self._mark_changed(new_row)
         self._index_run(run_id, thread_id)
         return new_row, claimed

backend/packages/harness/pyproject.toml

@@ -13,7 +13,7 @@ dependencies = [
     # the contract version it implements, extensions declare ranges. A range
     # here would let pip resolve a newer contract package than this harness
     # implements, making newer extensions look supported at runtime.
-    "deerflow-extension-api==0.2.0",
+    "deerflow-extension-api==0.2.1",
     "dotenv>=0.9.9",
     "exa-py>=1.0.0",
     "httpx>=0.28.0",

backend/tests/test_extension_api_contracts.py

@@ -24,9 +24,15 @@ from deerflow_extension_api import (
     ExtensionRuntimeDeps,
     ExtensionService,
     HostPolicySnapshot,
+    InvalidRunEvidenceCursor,
     MiddlewareContributor,
     MiddlewarePlacement,
     Placement,
+    RunEventPage,
+    RunEventView,
+    RunEvidenceReader,
+    RunPage,
+    RunStatusView,
     SystemModelCallObserver,
     SystemModelRequest,
     SystemModelResult,
@@ -71,6 +77,10 @@ def test_middleware_placement_defaults():
         SystemModelRequest,
         SystemModelResult,
         MiddlewarePlacement,
+        RunEventPage,
+        RunEventView,
+        RunPage,
+        RunStatusView,
     ],
 )
 def test_every_dataclass_is_frozen(cls):
@@ -119,6 +129,7 @@ def test_agent_build_context_optional_fields_keep_their_defaults():
         ExtensionRegistry,
         ExtensionService,
         MiddlewareContributor,
+        RunEvidenceReader,
         TaskLifecycleContributor,
         SystemModelCallObserver,
     ],
@@ -135,7 +146,10 @@ def test_every_protocol_method_has_a_default_implementation(protocol):
             continue
         checked += 1
         body = inspect.getsource(member).split("\n", 1)[1]
-        assert "return" in body, f"{protocol.__name__}.{name} has no default implementation. Adding a contract method is only additive when it returns a default; otherwise every already-released extension breaks on upgrade."
+        if protocol is RunEvidenceReader:
+            assert "raise NotImplementedError" in body
+        else:
+            assert "return" in body, f"{protocol.__name__}.{name} has no default implementation. Adding a contract method is only additive when it returns a default; otherwise every already-released extension breaks on upgrade."
     assert checked > 0, f"{protocol.__name__} declared no methods to check"
 
 
@@ -205,6 +219,10 @@ def test_gateway_contribution_points_are_part_of_the_public_surface():
     for name in (
         "ExtensionRuntimeDeps",
         "ExtensionService",
+        "InvalidRunEvidenceCursor",
+        "RunEvidenceReader",
+        "RunEventPage",
+        "RunPage",
     ):
         assert name in deerflow_extension_api.__all__
         assert hasattr(deerflow_extension_api, name)
@@ -213,6 +231,26 @@ def test_gateway_contribution_points_are_part_of_the_public_surface():
     assert not hasattr(deerflow_extension_api, "RouterContributor")
 
 
+def test_run_evidence_pages_are_immutable_and_empty_by_default():
+    assert issubclass(InvalidRunEvidenceCursor, ValueError)
+    assert RunPage().items == ()
+    assert RunPage().next_cursor is None
+    assert RunPage().has_more is False
+    assert RunEventPage().items == ()
+    assert RunEventPage().next_after_seq is None
+
+
+def test_bare_run_evidence_reader_fails_explicitly_instead_of_looking_caught_up():
+    class _Bare:
+        pass
+
+    async def invoke():
+        with pytest.raises(NotImplementedError):
+            await RunEvidenceReader.list_changed_runs(_Bare(), cursor=None, limit=10)
+
+    asyncio.run(invoke())
+
+
 def test_task_store_from_runtime_reads_the_host_key():
     class _Runtime:
         def __init__(self, context):
@@ -304,7 +342,7 @@ def test_runtime_api_version_matches_the_installed_contract_package():
     """Every additive contract slice bumps both gates together."""
     from importlib.metadata import version
 
-    assert API_VERSION == "0.2.0"
+    assert API_VERSION == "0.2.1"
     assert API_VERSION == version("deerflow-extension-api")
 
 
@@ -316,5 +354,6 @@ def test_extension_service_contract_is_public_and_defaults_to_noop():
 
     assert deps.app_store is None
     assert deps.session_factory is None
+    assert deps.run_evidence_reader is None
     assert asyncio.run(ExtensionService.start(_Bare(), deps)) is None
     assert asyncio.run(ExtensionService.stop(_Bare())) is None

backend/tests/test_extension_gateway_wiring.py

@@ -57,12 +57,19 @@ async def test_services_start_in_order_with_narrow_deps_and_fail_open():
         subagents=SimpleNamespace(max_total_per_run=4),
     )
 
-    diagnostics = await start_services(extensions, config, session_factory)
+    run_evidence_reader = object()
+    diagnostics = await start_services(
+        extensions,
+        config,
+        session_factory,
+        run_evidence_reader=run_evidence_reader,
+    )
 
     assert events == ["start:first", "start:second"]
     assert first.deps is second.deps
     assert second.deps.app_store is extensions.app_store
     assert second.deps.session_factory is session_factory
+    assert second.deps.run_evidence_reader is run_evidence_reader
     assert second.deps.policy.token_budget_enabled is False
     assert second.deps.policy.max_total_tokens is None
     assert second.deps.policy.max_subagents_per_run == 4

backend/tests/test_migration_0023_run_change_seq.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+import sqlalchemy as sa
+from alembic import command
+from alembic.script import ScriptDirectory
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from deerflow.persistence.bootstrap import _MIGRATIONS_DIR, _get_alembic_config
+from deerflow.persistence.run import RunRepository
+
+pytestmark = pytest.mark.asyncio
+
+REVISION = "0023_run_change_seq"
+PREVIOUS = "0022_scheduled_occurrence_seq"
+
+
+async def test_changed_run_revision_is_in_single_head_chain():
+    script = ScriptDirectory(str(_MIGRATIONS_DIR))
+    assert len(script.get_heads()) == 1
+    # Later migrations may advance the head without removing this revision.
+    assert REVISION in {revision.revision for revision in script.walk_revisions()}
+    assert script.get_revision(REVISION).down_revision == PREVIOUS
+
+
+async def test_upgrade_exposes_legacy_runs_and_allocates_new_positions(tmp_path):
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'migration.db'}")
+    cfg = _get_alembic_config(engine)
+    try:
+        await asyncio.to_thread(command.upgrade, cfg, PREVIOUS)
+        async with engine.begin() as connection:
+            for run_id in ("legacy-a", "legacy-b"):
+                await connection.execute(
+                    sa.text(
+                        "INSERT INTO runs "
+                        "(run_id, thread_id, user_id, status, operation_kind, metadata_json, "
+                        "kwargs_json, multitask_strategy, message_count, total_input_tokens, "
+                        "total_output_tokens, total_tokens, llm_call_count, lead_agent_tokens, "
+                        "subagent_tokens, middleware_tokens, created_at, updated_at) "
+                        "VALUES (:run_id, :thread_id, 'user-1', 'pending', 'run', '{}', '{}', "
+                        "'reject', 0, 0, 0, 0, 0, 0, 0, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
+                    ),
+                    {"run_id": run_id, "thread_id": f"thread-{run_id}"},
+                )
+
+        await asyncio.to_thread(command.upgrade, cfg, REVISION)
+        factory = async_sessionmaker(engine, expire_on_commit=False)
+        repo = RunRepository(factory)
+        legacy = await repo.list_changed(
+            after_change_seq=-1,
+            after_run_id="",
+            user_id="user-1",
+            limit=10,
+        )
+        assert [(row["change_seq"], row["run_id"]) for row in legacy] == [
+            (0, "legacy-a"),
+            (0, "legacy-b"),
+        ]
+
+        await repo.update_status("legacy-a", "error")
+        changed = await repo.list_changed(
+            after_change_seq=0,
+            after_run_id="legacy-b",
+            user_id="user-1",
+            limit=10,
+        )
+        assert [row["run_id"] for row in changed] == ["legacy-a"]
+        assert changed[0]["change_seq"] > 0
+    finally:
+        await engine.dispose()

backend/tests/test_run_evidence_reader.py

@@ -0,0 +1,343 @@
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+from deerflow_extension_api import InvalidRunEvidenceCursor
+
+from deerflow.extensions.run_evidence import StoreRunEvidenceReader
+from deerflow.runtime.events.store.memory import MemoryRunEventStore
+from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+
+async def _put_run(store: MemoryRunStore, run_id: str, thread_id: str, *, user_id: str = "user-1") -> None:
+    await store.put(run_id, thread_id=thread_id, user_id=user_id)
+
+
+@pytest.mark.asyncio
+async def test_changed_run_cursor_pages_without_loss_and_replays_updates():
+    runs = MemoryRunStore()
+    events = MemoryRunEventStore()
+    reader = StoreRunEvidenceReader(runs, events, user_id="user-1")
+    await _put_run(runs, "run-a", "thread-a")
+    await _put_run(runs, "run-b", "thread-b")
+
+    first = await reader.list_changed_runs(cursor=None, limit=1)
+    assert [item.run_id for item in first.items] == ["run-a"]
+    assert first.has_more is True
+    assert first.next_cursor
+
+    await runs.update_status("run-a", "running")
+    second = await reader.list_changed_runs(cursor=first.next_cursor, limit=10)
+    assert [item.run_id for item in second.items] == ["run-b", "run-a"]
+    assert second.has_more is False
+
+
+@pytest.mark.asyncio
+async def test_changed_run_reader_is_bound_to_one_user():
+    runs = MemoryRunStore()
+    events = MemoryRunEventStore()
+    await _put_run(runs, "mine", "thread-1", user_id="user-1")
+    await _put_run(runs, "theirs", "thread-2", user_id="user-2")
+
+    page = await StoreRunEvidenceReader(runs, events, user_id="user-1").list_changed_runs(cursor=None, limit=10)
+    assert [item.run_id for item in page.items] == ["mine"]
+
+
+@pytest.mark.asyncio
+async def test_event_pages_resume_and_status_is_authoritative():
+    runs = MemoryRunStore()
+    events = MemoryRunEventStore()
+    await _put_run(runs, "run-a", "thread-a")
+    await runs.update_status("run-a", "error", error="failed")
+    for index in range(3):
+        await events.put(
+            thread_id="thread-a",
+            run_id="run-a",
+            event_type="test.event",
+            category="trace",
+            content={"index": index},
+        )
+
+    reader = StoreRunEvidenceReader(runs, events, user_id="user-1")
+    first = await reader.list_run_events(thread_id="thread-a", run_id="run-a", after_seq=None, limit=2)
+    assert [item.content for item in first.items] == [{"index": 0}, {"index": 1}]
+    assert first.has_more is True
+    second = await reader.list_run_events(
+        thread_id="thread-a",
+        run_id="run-a",
+        after_seq=first.next_after_seq,
+        limit=2,
+    )
+    assert [item.content for item in second.items] == [{"index": 2}]
+    assert second.has_more is False
+
+    status = await reader.get_run_status(thread_id="thread-a", run_id="run-a")
+    assert status is not None
+    assert status.status == "error"
+    assert status.error == "failed"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("event_backend", ["memory", "jsonl", "db"])
+async def test_event_views_detach_nested_payloads_from_store(tmp_path, event_backend):
+    engine = None
+    runs = MemoryRunStore()
+    if event_backend == "memory":
+        events = MemoryRunEventStore()
+    elif event_backend == "jsonl":
+        from deerflow.runtime.events.store.jsonl import JsonlRunEventStore
+
+        events = JsonlRunEventStore(tmp_path / "events")
+    else:
+        from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+        from deerflow.persistence.base import Base
+        from deerflow.runtime.events.store.db import DbRunEventStore
+
+        engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'events.db'}")
+        async with engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+        events = DbRunEventStore(async_sessionmaker(engine, expire_on_commit=False))
+
+    await _put_run(runs, "run-a", "thread-a")
+    try:
+        await events.put(
+            thread_id="thread-a",
+            run_id="run-a",
+            event_type="test.event",
+            category="trace",
+            content={"text": ["original"]},
+            metadata={"custom": {"value": "original"}},
+        )
+
+        reader = StoreRunEvidenceReader(runs, events, user_id=None)
+        page = await reader.list_run_events(thread_id="thread-a", run_id="run-a", after_seq=None, limit=10)
+        page.items[0].content["text"][0] = "changed"
+        page.items[0].metadata["custom"]["value"] = "changed"
+
+        reread = await reader.list_run_events(thread_id="thread-a", run_id="run-a", after_seq=None, limit=10)
+        stored = await events.list_events("thread-a", "run-a", user_id=None)
+        assert reread.items[0].content == {"text": ["original"]}
+        assert reread.items[0].metadata["custom"] == {"value": "original"}
+        assert stored[0]["content"] == {"text": ["original"]}
+        assert stored[0]["metadata"]["custom"] == {"value": "original"}
+    finally:
+        if engine is not None:
+            await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_db_event_reads_use_reader_scope_not_ambient_user(tmp_path):
+    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+    from deerflow.persistence.base import Base
+    from deerflow.persistence.run import RunRepository
+    from deerflow.runtime import user_context
+    from deerflow.runtime.events.store.db import DbRunEventStore
+
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'scoped-events.db'}")
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+        factory = async_sessionmaker(engine, expire_on_commit=False)
+        runs = RunRepository(factory)
+        events = DbRunEventStore(factory)
+        await runs.put("run-a", thread_id="thread-a", user_id="alice")
+
+        alice_token = user_context.set_current_user(SimpleNamespace(id="alice"))
+        try:
+            await events.put(
+                thread_id="thread-a",
+                run_id="run-a",
+                event_type="test.event",
+                category="trace",
+                content={"owner": "alice"},
+            )
+        finally:
+            user_context.reset_current_user(alice_token)
+
+        for ambient_user in (None, SimpleNamespace(id="bob")):
+            ambient_token = user_context._current_user.set(ambient_user)
+            try:
+                for reader_user in (None, "alice"):
+                    reader = StoreRunEvidenceReader(runs, events, user_id=reader_user)
+                    page = await reader.list_run_events(
+                        thread_id="thread-a",
+                        run_id="run-a",
+                        after_seq=None,
+                        limit=10,
+                    )
+                    assert [item.content for item in page.items] == [{"owner": "alice"}]
+            finally:
+                user_context._current_user.reset(ambient_token)
+    finally:
+        await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_reader_hides_runs_outside_scope_and_rejects_cursor_from_another_scope():
+    runs = MemoryRunStore()
+    events = MemoryRunEventStore()
+    await _put_run(runs, "run-a", "thread-a", user_id="user-1")
+    first = await StoreRunEvidenceReader(runs, events, user_id="user-1").list_changed_runs(cursor=None, limit=1)
+
+    other = StoreRunEvidenceReader(runs, events, user_id="user-2")
+    with pytest.raises(InvalidRunEvidenceCursor, match="cursor scope"):
+        await other.list_changed_runs(cursor=first.next_cursor, limit=1)
+    assert await other.get_run_status(thread_id="thread-a", run_id="run-a") is None
+    assert (await other.list_run_events(thread_id="thread-a", run_id="run-a", after_seq=None, limit=10)).items == ()
+
+
+@pytest.mark.asyncio
+async def test_memory_progress_updates_do_not_advance_changed_run_cursor():
+    runs = MemoryRunStore()
+    reader = StoreRunEvidenceReader(runs, MemoryRunEventStore(), user_id="user-1")
+    await _put_run(runs, "run-a", "thread-a")
+    await runs.start_run("run-a")
+    before_progress = await reader.list_changed_runs(cursor=None, limit=10)
+
+    await runs.update_run_progress("run-a", total_tokens=10, last_ai_message="working")
+
+    after_progress = await reader.list_changed_runs(cursor=before_progress.next_cursor, limit=10)
+    assert after_progress.items == ()
+    assert after_progress.next_cursor == before_progress.next_cursor
+
+
+@pytest.mark.asyncio
+async def test_sql_changed_run_cursor_survives_repository_restart(tmp_path):
+    from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
+    from deerflow.persistence.run import RunRepository
+
+    url = f"sqlite+aiosqlite:///{tmp_path / 'evidence.db'}"
+    await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path))
+    try:
+        repo = RunRepository(get_session_factory())
+        await repo.put("run-a", thread_id="thread-a", user_id="user-1")
+        reader = StoreRunEvidenceReader(repo, MemoryRunEventStore(), user_id="user-1")
+        first = await reader.list_changed_runs(cursor=None, limit=1)
+        await repo.put("run-b", thread_id="thread-b", user_id="user-1")
+        cursor = first.next_cursor
+    finally:
+        await close_engine()
+
+    await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path))
+    try:
+        restarted = StoreRunEvidenceReader(
+            RunRepository(get_session_factory()),
+            MemoryRunEventStore(),
+            user_id="user-1",
+        )
+        page = await restarted.list_changed_runs(cursor=cursor, limit=10)
+        assert [item.run_id for item in page.items] == ["run-b"]
+    finally:
+        await close_engine()
+
+
+@pytest.mark.asyncio
+async def test_sql_progress_updates_do_not_advance_changed_run_cursor(tmp_path):
+    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+    from deerflow.persistence.base import Base
+    from deerflow.persistence.run import RunRepository
+
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'progress.db'}")
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+        repo = RunRepository(async_sessionmaker(engine, expire_on_commit=False))
+        await repo.put("run-a", thread_id="thread-a", user_id="user-1")
+        await repo.start_run("run-a")
+        reader = StoreRunEvidenceReader(repo, MemoryRunEventStore(), user_id="user-1")
+        before_progress = await reader.list_changed_runs(cursor=None, limit=10)
+
+        await repo.update_run_progress("run-a", total_tokens=10, last_ai_message="working")
+
+        after_progress = await reader.list_changed_runs(cursor=before_progress.next_cursor, limit=10)
+        assert after_progress.items == ()
+        assert after_progress.next_cursor == before_progress.next_cursor
+    finally:
+        await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_sql_atomic_interrupt_allocates_before_run_lock_and_shares_position(tmp_path):
+    from sqlalchemy import event
+    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+    from deerflow.persistence.base import Base
+    from deerflow.persistence.run import RunRepository
+
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'atomic.db'}")
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+        repo = RunRepository(async_sessionmaker(engine, expire_on_commit=False))
+        await repo.put(
+            "run-old",
+            thread_id="thread-a",
+            user_id="user-1",
+            status="running",
+            owner_worker_id="worker-1",
+        )
+        statements: list[str] = []
+
+        @event.listens_for(engine.sync_engine, "before_cursor_execute")
+        def capture_statement(_connection, _cursor, statement, _parameters, _context, _executemany):
+            statements.append(" ".join(statement.lower().split()))
+
+        new_run, claimed = await repo.create_thread_operation_atomic(
+            "run-new",
+            thread_id="thread-a",
+            user_id="user-1",
+            owner_worker_id="worker-1",
+            lease_expires_at=None,
+            multitask_strategy="interrupt",
+        )
+
+        assert [run["run_id"] for run in claimed] == ["run-old"]
+        assert claimed[0]["change_seq"] == new_run["change_seq"]
+        clock_update = next(index for index, statement in enumerate(statements) if statement.startswith("update run_change_clock"))
+        run_lock_query = next(index for index, statement in enumerate(statements) if statement.startswith("select runs."))
+        assert clock_update < run_lock_query
+    finally:
+        await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_sql_change_positions_are_unique_across_concurrent_threads(tmp_path):
+    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+    from deerflow.persistence.base import Base
+    from deerflow.persistence.run import RunRepository
+
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'concurrent.db'}")
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+        factory = async_sessionmaker(engine, expire_on_commit=False)
+        repositories = [RunRepository(factory) for _ in range(12)]
+        await asyncio.gather(
+            *(
+                repository.put(
+                    f"run-{index:02d}",
+                    thread_id=f"thread-{index:02d}",
+                    user_id="user-1",
+                )
+                for index, repository in enumerate(repositories)
+            )
+        )
+
+        rows = await repositories[0].list_changed(
+            after_change_seq=-1,
+            after_run_id="",
+            user_id="user-1",
+            limit=20,
+        )
+        positions = [row["change_seq"] for row in rows]
+        assert len(rows) == 12
+        assert positions == sorted(positions)
+        assert len(set(positions)) == len(positions)
+    finally:
+        await engine.dispose()

backend/tests/test_scheduler_occurrence_ordering.py

@@ -22,7 +22,7 @@ from sqlalchemy.schema import CreateSchema, DropSchema
 from app.scheduler.service import ScheduledTaskService
 from deerflow.persistence.base import Base
 from deerflow.persistence.run import RunRepository
-from deerflow.persistence.run.model import RunRow
+from deerflow.persistence.run.model import RunChangeClockRow, RunRow
 from deerflow.persistence.scheduled_task_runs import ScheduledTaskRunRepository
 from deerflow.persistence.scheduled_task_runs.model import ScheduledTaskRunRow
 from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository
@@ -72,7 +72,17 @@ async def occurrence_databases(request, tmp_path):
                 engine = create_async_engine(postgres_uri, connect_args={"server_settings": {"search_path": schema}})
             engines.append(engine)
         async with engines[0].begin() as connection:
-            await connection.run_sync(lambda sync: Base.metadata.create_all(sync, tables=[ScheduledTaskRow.__table__, ScheduledTaskRunRow.__table__, RunRow.__table__]))
+            await connection.run_sync(
+                lambda sync: Base.metadata.create_all(
+                    sync,
+                    tables=[
+                        ScheduledTaskRow.__table__,
+                        ScheduledTaskRunRow.__table__,
+                        RunRow.__table__,
+                        RunChangeClockRow.__table__,
+                    ],
+                )
+            )
         yield [async_sessionmaker(engine, expire_on_commit=False) for engine in engines], backend == "postgres-multi"
     finally:
         for engine in engines:

backend/uv.lock

@@ -918,7 +918,7 @@ extensions = []
 
 [[package]]
 name = "deerflow-extension-api"
-version = "0.2.0"
+version = "0.2.1"
 source = { editable = "packages/extension-api" }
 
 [[package]]