fix(gateway): preserve exact history attribution beyond event page limits

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

@@ -20,6 +20,39 @@ Checkpointer storage runs in one of two channel modes, selected by `checkpoint_c
 
 **Run rollback flow** (`runtime/runs/worker.py`): `_capture_rollback_point` materializes the complete pre-run state via the accessor and captures raw `pending_writes` via `aget_tuple` into an immutable `RollbackPoint` before the run starts — capture failure disables rollback (fail-closed), never restores partial state. In `full` mode, cancel-with-rollback forks from the pre-run checkpoint via the mutation graph and inherits non-message channels from that parent. In `delta` mode, forking is unsafe once the cancelled path has attached sibling writes to the pre-run checkpoint, so rollback replaces every captured channel on the current head, using `Overwrite` for reducers and schema defaults for current-head-only channels. Both modes reattach only the captured pre-run pending writes to the restored checkpoint. Edit replay runs (`metadata.replay_kind="edit"`) also restore the pre-run checkpoint on failed, timed-out, or interrupted completion and publish the restored `values` snapshot to the stream before `end`, so clients do not remain on a transient edited branch when the replay did not produce a successful replacement.
 
+**Targeted run-event attribution** (`runtime/events/store/`):
+`RunEventStore.find_latest_ai_message_run_ids()` has a complete-or-error
+contract. Its default implementation walks `list_messages()` backward in
+1000-row pages, preserves the first page's high-watermark through the exclusive
+`before_seq` cursor, and raises when a full page has no safe progressing `seq`.
+Memory and database stores use that bounded path; the JSONL store overrides it
+with one complete thread-log read because each JSONL page would otherwise
+rescan every run file. The default and JSONL paths share the public
+`normalize_message_ids()` and `match_ai_message_run_id()` helpers from
+`events/store/base.py`. Database owner filtering is inherited on every page.
+Callers may use a missing key as proof that no valid AI event exists only after
+an ordinary return, never after an exception. A caller that crosses a run or
+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.
+
+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
+write-on-read cache stores `run_message_ids` for every audited AI ID (including
+exhaustive misses) plus required `run_durations`; duration presence alone does
+not prove attribution. Historical `body.before` reads write the audit to the
+head, and the merge may retain IDs no longer in materialized history, which
+readers ignore. Migration must acquire the durable `checkpoint_write`
+reservation, then repeat the whole message audit and batch-reload required run
+rows before persisting. Post-admission exact hits replace foreground exact or
+boundary mappings, and recomputed final durations replace foreground snapshots.
+Successful workers keep their durable run row active through the final duration
+checkpoint write, so a peer migration cannot enter during terminalization.
+The first `RunManager.list_by_thread()` hydration page uses a 100-row floor or
+the number of required IDs, whichever is larger; missing exact runs use targeted
+`get()` calls.
+
 **Where things live**:
 - `runtime/checkpoint_mode.py` — mode + snapshot-frequency freeze, marker injection, delta detection, compatibility gate, both error types
 - `runtime/checkpoint_state.py` — `CheckpointStateAccessor`, `build_state_mutation_graph`, `RollbackPoint`

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

@@ -16,6 +16,31 @@
 
 from deerflow.runtime.user_context import AUTO, _AutoSentinel
 
+_AI_MESSAGE_RUN_LOOKUP_PAGE_SIZE = 1000
+
+
+class IncompleteMessageRunLookupError(RuntimeError):
+    """Raised when a store cannot prove that a targeted lookup is complete."""
+
+
+def normalize_message_ids(message_ids: set[str]) -> set[str]:
+    """Return the non-empty string IDs that can participate in a lookup."""
+    return {message_id for message_id in message_ids if isinstance(message_id, str) and message_id}
+
+
+def match_ai_message_run_id(event: object, message_ids: set[str]) -> tuple[str, str] | None:
+    """Return a target AI message ID and its valid run ID, if present."""
+    if not isinstance(event, dict) or event.get("category") != "message":
+        return None
+    content = event.get("content")
+    run_id = event.get("run_id")
+    if not isinstance(content, dict) or content.get("type") != "ai" or not isinstance(run_id, str) or not run_id:
+        return None
+    message_id = content.get("id")
+    if not isinstance(message_id, str) or message_id not in message_ids:
+        return None
+    return message_id, run_id
+
 
 class RunEventStore(abc.ABC):
     """Run event stream storage interface.
@@ -27,6 +52,8 @@ class RunEventStore(abc.ABC):
     4. list_events() returns all events for the specified run
     5. Returned dicts contain the required RunEvent envelope fields; backends
        may add documented fields such as DbRunEventStore.user_id
+    6. find_latest_ai_message_run_ids() returns the newest valid AI message
+       event for each requested ID and performs no storage work for empty input
     """
 
     @abc.abstractmethod
@@ -92,6 +119,70 @@ async def list_messages(
         user-scoped backends must apply it according to their isolation model.
         """
 
+    async def find_latest_ai_message_run_ids(
+        self,
+        thread_id: str,
+        message_ids: set[str],
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ) -> dict[str, str]:
+        """Map target message IDs to their newest valid AI event's run ID.
+
+        Only ``category="message"`` events whose structured content has
+        ``type="ai"`` and whose ``run_id`` is a non-empty string qualify. An
+        empty target set must return immediately without storage work. The
+        default implementation pages backward in bounded windows. It raises
+        :class:`IncompleteMessageRunLookupError` instead of returning a
+        partial result when a full page lacks a safe, progressing ``seq``
+        cursor; callers may only treat an ordinary return as an exhaustive
+        lookup for unresolved IDs.
+
+        ``user_id`` follows the same explicit-caller semantics as
+        :meth:`list_messages`.
+        """
+        pending = normalize_message_ids(message_ids)
+        if not pending:
+            return {}
+
+        result: dict[str, str] = {}
+        before_seq: int | None = None
+        while pending:
+            page = await self.list_messages(
+                thread_id,
+                limit=_AI_MESSAGE_RUN_LOOKUP_PAGE_SIZE,
+                before_seq=before_seq,
+                user_id=user_id,
+            )
+            if not page:
+                break
+
+            for event in reversed(page):
+                match = match_ai_message_run_id(event, pending)
+                if match is None:
+                    continue
+                message_id, run_id = match
+                result[message_id] = run_id
+                pending.remove(message_id)
+                if not pending:
+                    break
+
+            if not pending or len(page) < _AI_MESSAGE_RUN_LOOKUP_PAGE_SIZE:
+                break
+
+            seqs: list[int] = []
+            for event in page:
+                seq = event.get("seq") if isinstance(event, dict) else None
+                if not isinstance(seq, int) or isinstance(seq, bool):
+                    raise IncompleteMessageRunLookupError("Run event lookup could not form a safe backward cursor from a full page")
+                seqs.append(seq)
+
+            next_before_seq = min(seqs)
+            if before_seq is not None and next_before_seq >= before_seq:
+                raise IncompleteMessageRunLookupError("Run event lookup could not form a safe backward cursor because seq did not progress")
+            before_seq = next_before_seq
+
+        return result
+
     @abc.abstractmethod
     async def list_events(
         self,

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

@@ -30,7 +30,7 @@
 from pathlib import Path
 from typing import Any
 
-from deerflow.runtime.events.store.base import RunEventStore
+from deerflow.runtime.events.store.base import RunEventStore, match_ai_message_run_id, normalize_message_ids
 from deerflow.runtime.user_context import AUTO, _AutoSentinel
 from deerflow.utils.thread_id import validate_thread_id
 
@@ -280,6 +280,34 @@ async def list_messages(self, thread_id, *, limit=50, before_seq=None, after_seq
         else:
             return messages[-limit:]
 
+    async def find_latest_ai_message_run_ids(
+        self,
+        thread_id: str,
+        message_ids: set[str],
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ) -> dict[str, str]:
+        pending = normalize_message_ids(message_ids)
+        if not pending:
+            return {}
+
+        # Keep the one-pass view stable against this backend's supported
+        # single-process writers. Without the write lock, reading run files one
+        # by one can mix events from opposite sides of a concurrent append.
+        async with self._get_write_lock(thread_id):
+            events = await asyncio.to_thread(self._read_thread_events, thread_id)
+        result: dict[str, str] = {}
+        for event in reversed(events):
+            match = match_ai_message_run_id(event, pending)
+            if match is None:
+                continue
+            message_id, run_id = match
+            result[message_id] = run_id
+            pending.remove(message_id)
+            if not pending:
+                break
+        return result
+
     async def list_events(self, thread_id, run_id, *, event_types=None, task_id=None, limit=500, after_seq=None):
         events = await asyncio.to_thread(self._read_run_events, thread_id, run_id)
         if event_types is not None:

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

@@ -549,6 +549,26 @@ async def update_run_progress(self, run_id: str, **kwargs) -> None:
             except Exception:
                 logger.warning("Failed to persist run progress for %s", run_id, exc_info=True)
 
+    async def update_finalizing_progress(self, run_id: str, **kwargs) -> None:
+        """Persist final fields while the durable row is deliberately active."""
+        should_persist = False
+        async with self._lock:
+            record = self._runs.get(run_id)
+            if record is not None and not record.ownership_lost:
+                should_persist = record.status not in (RunStatus.pending, RunStatus.running)
+                if should_persist:
+                    for key, value in kwargs.items():
+                        if hasattr(record, key) and value is not None:
+                            setattr(record, key, value)
+                    record.updated_at = _now_iso()
+        if should_persist and self._store is not None:
+            try:
+                # The local status is already staged as terminal, but the store
+                # row intentionally remains running until checkpoint finalization.
+                await self._store.update_run_progress(run_id, **kwargs)
+            except Exception:
+                logger.warning("Failed to persist finalizing progress for %s", run_id, exc_info=True)
+
     async def create(
         self,
         thread_id: str,