fix(history): early user messages vanish or jump mid-run when pagination and context compaction overlap

backend/AGENTS.md

@@ -433,7 +433,7 @@ Before changing a later authorization phase, read the [authorization RFC](../doc
 32. **TerminalResponseMiddleware** - When a provider returns an empty terminal `AIMessage` after tool execution, injects a hidden recovery prompt and retries the model once; a second empty response is replaced in checkpoint state by a visible error fallback marked for the run worker, so the run finishes as an error instead of a silent success
 33. **ModelLengthFinishReasonMiddleware** - Records `stop_reason=model_length_capped` when provider-specific length detectors match a terminal `AIMessage` without tool-call intent (`finish_reason=length` / `MAX_TOKENS`, or `stop_reason=max_tokens`), preserving the original assistant content and never reparsing textual tool-call-like envelopes
 34. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after terminal-response/custom/configured middlewares so LangChain's reverse-order `after_model` dispatch runs it first
-35. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, writes a readable `ToolMessage.content` fallback plus structured `ToolMessage.artifact.human_input` request payload, and interrupts via `Command(goto=END)` (must be last). Payloads are versioned: legacy modes (`free_text` / `choice_with_other`) keep `version: 1` unchanged, while the v2 `form` mode (from `fields`) carries `version: 2` so older frontends reject the payload and degrade to the plain-text fallback. Field normalization is deterministic and lives in the middleware, not the tool schema — the middleware short-circuits before tool execution, so tool-arg typing alone provides no runtime validation. Validation is atomic: any structurally broken entry (non-dict, bad/duplicate name, a name colliding with a JS `Object.prototype` member like `__proto__`/`constructor`, exceeding the caps of 16 fields / 24 options per field / 200 chars per text, or the whole normalized definition exceeding `MAX_FORM_SERIALIZED_BYTES` = 16KB UTF-8 — the per-item caps alone admit forms whose IM text fallback would blow channel delivery limits and truncate away trailing fields) degrades the whole form to the legacy option/free-text modes, so a card can never render "complete" while silently missing a business field; benign issues keep local degradation (unknown types — including unhashable JSON like `type: []`, which must never raise from the membership probe — and option-less selects become `text`), and options are trimmed/deduped with blanks dropped (both form-level and top-level) because the frontend parser rejects blank option labels. Model-produced XML-to-dict option payloads are recursively flattened from dict/list containers in source order, scalar string/number leaves are retained, and residual XML tags are removed before the same trimming and deduplication. Checkbox fields are booleans that default to an explicit "no"; `required` on a checkbox means must-agree/consent semantics. The response protocol is deliberately unchanged (v1 `text`/`option` only): form cards submit a readable text summary as `response_kind: "text"`, so journal persistence and answered-card recovery need no new allowlist entries. Because this middleware can short-circuit tool execution before LangChain emits `on_tool_end`, `RunJournal` performs a root-run final reconciliation for allowlisted clarification `ToolMessage`s whose `tool_call_id` was produced by the current run, so human-input request cards remain recoverable from `run_events` after checkpoint compaction. Human Input Card replies are submitted as `hide_from_ui` `HumanMessage`s with `additional_kwargs.human_input_response`; `RunJournal` persists only allowlisted hidden response sources (currently `ask_clarification`) as `llm.human.input`, which preserves answered-card state after compaction without exposing generic internal hidden context.
+35. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, writes a readable `ToolMessage.content` fallback plus structured `ToolMessage.artifact.human_input` request payload, and interrupts via `Command(goto=END)` (must be last). Payloads are versioned: legacy modes (`free_text` / `choice_with_other`) keep `version: 1` unchanged, while the v2 `form` mode (from `fields`) carries `version: 2` so older frontends reject the payload and degrade to the plain-text fallback. Field normalization is deterministic and lives in the middleware, not the tool schema — the middleware short-circuits before tool execution, so tool-arg typing alone provides no runtime validation. Validation is atomic: any structurally broken entry (non-dict, bad/duplicate name, a name colliding with a JS `Object.prototype` member like `__proto__`/`constructor`, exceeding the caps of 16 fields / 24 options per field / 200 chars per text, or the whole normalized definition exceeding `MAX_FORM_SERIALIZED_BYTES` = 16KB UTF-8 — the per-item caps alone admit forms whose IM text fallback would blow channel delivery limits and truncate away trailing fields) degrades the whole form to the legacy option/free-text modes, so a card can never render "complete" while silently missing a business field; benign issues keep local degradation (unknown types — including unhashable JSON like `type: []`, which must never raise from the membership probe — and option-less selects become `text`), and options are trimmed/deduped with blanks dropped (both form-level and top-level) because the frontend parser rejects blank option labels. Model-produced XML-to-dict option payloads are recursively flattened from dict/list containers in source order, scalar string/number leaves are retained, and residual XML tags are removed before the same trimming and deduplication. Checkbox fields are booleans that default to an explicit "no"; `required` on a checkbox means must-agree/consent semantics. The response protocol is deliberately unchanged (v1 `text`/`option` only): form cards submit a readable text summary as `response_kind: "text"`, so journal persistence and answered-card recovery need no new allowlist entries. Because this middleware can short-circuit tool execution before LangChain emits `on_tool_end`, `RunJournal` performs a root-run final reconciliation for `ToolMessage`s whose `tool_call_id` was produced by the current run, so human-input request cards remain recoverable from `run_events` after checkpoint compaction. That reconciliation is **not** limited to `ask_clarification`: every middleware that answers a tool call itself has the same gap, and a result the user saw during the run must not disappear on reload (#4666 — `ReadBeforeWriteMiddleware`'s blocked-write errors were reaching the UI but never the event store). Its scope is bounded by three independent conditions instead of a tool-name allowlist — the message must be user-visible, the call must belong to this run's **lead agent** (`_remember_current_run_tool_calls` records lead-agent calls only, so subagent results stay in their own `subagent.step` feed), and it must not already be persisted. Keep those three; they are what makes a name allowlist unnecessary. Human Input Card replies are submitted as `hide_from_ui` `HumanMessage`s with `additional_kwargs.human_input_response`; `RunJournal` persists only allowlisted hidden response sources (currently `ask_clarification`) as `llm.human.input`, which preserves answered-card state after compaction without exposing generic internal hidden context.
 
 ### Configuration System
 
@@ -526,6 +526,43 @@ filesystem cleanup, so the raw value is never interpolated into a host path;
 new runs, workspace/sandbox operations, and other state-producing mutations
 remain blocked.
 
+**Message feed seq for client ordering** (#4666): a checkpoint carries no seq of
+its own and loses messages to summarization, so a client merging a `values` frame
+with the seq-ordered thread feed cannot place a message the checkpoint kept once
+the feed's loaded page window (`GET /messages/page`, `limit=50`) no longer reaches
+back to it. `RunEventStore.get_message_seqs(thread_id, identities)` resolves the
+seq the store already assigned, keyed by
+`runtime/events/message_identity.py::message_identity` — the backend half of the
+identity rule `frontend/src/core/threads/hooks.ts::messageIdentity` applies (tool
+messages by `tool_call_id`; `X` / `X__user` human copies collapse to one). The two
+halves must stay in sync: a mismatch is silent, degrading placement rather than
+raising. `worker.py::_MessageSeqStamper` attaches the result as
+`additional_kwargs.deerflow_seq` when a root `values` frame is serialized —
+subgraph frames are not stamped, and nothing is written back to the checkpoint.
+The stamper is built once per run so goal continuations reuse its cache; only a
+frame introducing unresolved identities costs a query, which in practice is the
+compaction frame alone (measured: 1 lookup across 25 frames).
+
+Streaming is not the only way a client obtains the checkpoint, and it is not even
+the common one: opening a conversation reads it over REST, and a client that never
+joins a live run would otherwise get no placement information at all. `GET
+/threads/{id}/state` and `POST /threads/{id}/history` therefore stamp their
+serialized messages through `runtime/events/message_seq.py::stamp_messages_with_seq`,
+the request-scoped counterpart of the stamper — everything a checkpoint still holds
+is already persisted, so one batched lookup resolves the whole list and there is
+nothing to retry later. Both reads resolve the store through
+`threads.py::_optional_run_event_store` rather than `get_run_event_store`, because
+seq is placement metadata: a deployment without a feed must still be able to read a
+thread. Leaving these two endpoints unstamped is what kept #4666 reproducing after
+the streaming fix — the merge fell back to the nearest shared anchor, which after
+summarization sits deep inside the loaded page, and the rescued first user turn
+rendered behind the newest question (measured: row 320 of 389 on a reproducing
+thread). `deerflow_seq` is
+server-owned and joins `_SERVER_OWNED_MESSAGE_METADATA_KEYS` in
+`services.py::normalize_input`, because a replayed message carrying it back would
+write a thread-scoped seq into the checkpoint that a fork re-seeds and reassigns
+(#4380).
+
 **Workspace change review**: `packages/harness/deerflow/workspace_changes/`
 captures a pre-run and post-run snapshot of the thread-owned `workspace` and
 `outputs` directories. `runtime/runs/worker.py` performs the filesystem scan via

backend/app/gateway/routers/threads.py

@@ -55,6 +55,7 @@
     ThreadCompactionResult,
     compact_thread_context,
 )
+from deerflow.runtime.events.message_seq import stamp_messages_with_seq
 from deerflow.runtime.goal import (
     DEFAULT_MAX_GOAL_CONTINUATIONS,
     build_goal_state,
@@ -77,6 +78,17 @@
 _CHECKPOINT_MODE_ERRORS = (CheckpointModeMismatchError, CheckpointModeReconfigurationError)
 
 
+def _optional_run_event_store(request: Request) -> Any:
+    """Return the run event store, or ``None`` when the app has none wired.
+
+    Reads must not start depending on the feed: seq is placement metadata, and a
+    response without it degrades to the client's own ordering rule rather than
+    failing. ``get_run_event_store`` raises instead, which is right for the
+    endpoints that cannot work without a feed.
+    """
+    return getattr(request.app.state, "run_event_store", None)
+
+
 def _checkpoint_mode_http_error(exc: Exception, thread_id: str) -> HTTPException:
     """Map checkpoint-mode guard failures to precise HTTP statuses.
 
@@ -1219,8 +1231,15 @@ async def get_thread_state(thread_id: ThreadId, request: Request) -> ThreadState
     tasks_raw = snapshot.tasks or ()
     tasks = [{"id": getattr(task, "id", ""), "name": getattr(task, "name", "")} for task in tasks_raw]
 
+    values = serialize_channel_values_for_api(snapshot.values)
+    messages = values.get("messages")
+    if isinstance(messages, list) and messages:
+        # Same reason as the history endpoint: a client reading the checkpoint
+        # over REST needs the feed position the stream would have stamped.
+        values["messages"] = await stamp_messages_with_seq(_optional_run_event_store(request), thread_id, messages)
+
     return ThreadStateResponse(
-        values=serialize_channel_values_for_api(snapshot.values),
+        values=values,
         next=list(snapshot.next or ()),
         metadata=metadata,
         checkpoint={"id": checkpoint_id, "ts": coerce_iso(created_at)},
@@ -1464,7 +1483,15 @@ async def get_thread_history(
                     except Exception:
                         logger.warning("Failed to inject turn_duration for thread %s", thread_id, exc_info=True)
 
-                    values["messages"] = serialized_msgs
+                    # The stream stamps `values` frames as they are published, but a
+                    # client that only opens a conversation never sees one — this is
+                    # the read it does instead, and without a seq a rescued early turn
+                    # has no absolute position to be placed at (#4666).
+                    values["messages"] = await stamp_messages_with_seq(
+                        _optional_run_event_store(request),
+                        thread_id,
+                        serialized_msgs,
+                    )
 
             is_latest_checkpoint = False
 

backend/app/gateway/services.py

@@ -60,6 +60,7 @@
     inject_checkpoint_mode,
 )
 from deerflow.runtime.checkpoint_state import graph_state_schema
+from deerflow.runtime.events.message_identity import MESSAGE_SEQ_KEY
 from deerflow.runtime.goal import goal_thread_lock
 from deerflow.runtime.journal import build_checkpoint_history_seed_events
 from deerflow.runtime.runs.naming import resolve_root_run_name
@@ -108,6 +109,10 @@ async def reserve_checkpoint_write(
         _DYNAMIC_CONTEXT_REMINDER_KEY,
         _REMINDER_DATE_KEY,
         _IMAGE_CONTEXT_MESSAGE_MARKER_KEY,
+        # Attached when a values frame is serialized, for display ordering only.
+        # A replayed message carrying it back would write a thread-scoped seq
+        # into the checkpoint, which a fork then re-seeds and reassigns (#4380).
+        MESSAGE_SEQ_KEY,
     }
 )
 

backend/packages/harness/deerflow/agents/middlewares/dynamic_context_middleware.py

@@ -42,6 +42,7 @@
 
 from deerflow.runtime.context_keys import CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY
 from deerflow.runtime.user_context import resolve_runtime_user_id
+from deerflow.utils.messages import INJECTED_USER_MESSAGE_ID_SUFFIX, strip_injected_user_message_id_suffix
 
 if TYPE_CHECKING:
     from deerflow.config.app_config import AppConfig
@@ -61,23 +62,16 @@
 # so it is never exposed to user-influenceable memory content.
 _REMINDER_DATE_KEY = "reminder_date"
 _SUMMARY_MESSAGE_NAME = "summary"
-# Suffix the ID-swap gives the real user message; the reminder SystemMessage
-# takes the original id so ``add_messages`` can replace it in place.
-INJECTED_USER_MESSAGE_ID_SUFFIX = "__user"
 
-
-def strip_injected_user_message_id_suffix(message_id: str | None) -> str | None:
-    """Return the id *message_id* had before the reminder ID-swap.
-
-    Replaying a persisted user turn must feed the graph the id the client
-    originally sent: a ``{id}__user`` message is skipped as an injection target,
-    so replaying one into a state that has no reminder yet silently drops the
-    date and memory block for that turn.
-    """
-
-    if isinstance(message_id, str) and message_id.endswith(INJECTED_USER_MESSAGE_ID_SUFFIX):
-        return message_id[: -len(INJECTED_USER_MESSAGE_ID_SUFFIX)] or message_id
-    return message_id
+# ``INJECTED_USER_MESSAGE_ID_SUFFIX`` / ``strip_injected_user_message_id_suffix``
+# are defined in ``deerflow.utils.messages`` and re-exported here, where the
+# ID-swap they describe actually happens. Existing importers keep working.
+__all__ = [
+    "INJECTED_USER_MESSAGE_ID_SUFFIX",
+    "DynamicContextMiddleware",
+    "is_dynamic_context_reminder",
+    "strip_injected_user_message_id_suffix",
+]
 
 
 def _extract_date(content: str) -> str | None:

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

@@ -0,0 +1,50 @@
+"""Stable UI identity of a persisted message.
+
+The thread feed (``run_events``) and the checkpoint hold the same message under
+the same id, so a client can align them — but only if both sides agree on what
+"same message" means. This is the backend half of that rule; the frontend half
+is ``messageIdentity`` in ``frontend/src/core/threads/hooks.ts``. The two must
+stay in sync: a mismatch is silent, degrading placement rather than raising.
+
+Two normalizations matter:
+
+* a ``ToolMessage`` is identified by its ``tool_call_id``, not its own id —
+  that is the id both sides can always resolve;
+* ``DynamicContextMiddleware`` re-keys the submitted user turn from ``X`` to
+  ``X__user`` (giving ``X`` to the injected reminder), so the two human copies
+  must collapse to one identity.
+"""
+
+from collections.abc import Mapping
+from typing import Any
+
+from deerflow.utils.messages import strip_injected_user_message_id_suffix
+
+__all__ = ["MESSAGE_SEQ_KEY", "message_identity"]
+
+#: ``additional_kwargs`` key carrying a message's thread-feed seq to clients.
+#: Server-owned display metadata: it is attached when a frame is serialized and
+#: must be stripped from anything a client sends back, or a replayed message
+#: would write it into the checkpoint (where a fork re-seeds and reassigns seq).
+MESSAGE_SEQ_KEY = "deerflow_seq"
+
+
+def message_identity(message: Mapping[str, Any]) -> str | None:
+    """Return the stable identity of *message*, or ``None`` if it has none.
+
+    *message* is a serialized message mapping (as stored in ``run_events``
+    content or carried in a checkpoint ``values`` frame), not a ``BaseMessage``.
+    """
+    tool_call_id = message.get("tool_call_id")
+    if isinstance(tool_call_id, str) and tool_call_id:
+        return f"tool:{tool_call_id}"
+
+    message_id = message.get("id")
+    if not isinstance(message_id, str) or not message_id:
+        return None
+
+    # Only human copies collapse: a hidden SystemMessage legitimately reuses the
+    # original id, and merging it with the visible turn would hide the turn.
+    if message.get("type") == "human":
+        message_id = strip_injected_user_message_id_suffix(message_id) or message_id
+    return f"message:{message_id}"