diff --git a/backend/AGENTS.md b/backend/AGENTS.md index e58a6c984e7..58bcd878170 100644 --- a/backend/AGENTS.md +++ b/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 diff --git a/backend/app/gateway/routers/threads.py b/backend/app/gateway/routers/threads.py index 8fa7b120578..262c2d7039c 100644 --- a/backend/app/gateway/routers/threads.py +++ b/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 diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index 553221a32b0..13da6c61ba1 100644 --- a/backend/app/gateway/services.py +++ b/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, } ) diff --git a/backend/packages/harness/deerflow/agents/middlewares/dynamic_context_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/dynamic_context_middleware.py index 4f33daf9582..851fd3e877d 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/dynamic_context_middleware.py +++ b/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: diff --git a/backend/packages/harness/deerflow/runtime/events/message_identity.py b/backend/packages/harness/deerflow/runtime/events/message_identity.py new file mode 100644 index 00000000000..8f2fcab6687 --- /dev/null +++ b/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}" diff --git a/backend/packages/harness/deerflow/runtime/events/message_seq.py b/backend/packages/harness/deerflow/runtime/events/message_seq.py new file mode 100644 index 00000000000..0c63ff5ae52 --- /dev/null +++ b/backend/packages/harness/deerflow/runtime/events/message_seq.py @@ -0,0 +1,55 @@ +"""Attach the thread-global feed seq to serialized checkpoint messages. + +The checkpoint holds no seq of its own, so a client merging it with the +seq-ordered feed cannot place a message the loaded page window no longer +reaches back to. The streaming path solves this by stamping `values` frames as +they are published — but a client that merely *opens* a conversation never sees +a frame. It reads the checkpoint over REST, and without a seq there a +summarization-rescued early turn is placed by its nearest anchor instead, which +after compaction sits deep in the loaded page (#4666). + +This is the request-scoped counterpart of the worker's stamper: everything the +checkpoint still holds is already persisted, so one batched lookup resolves the +whole list and there is nothing to retry later. +""" + +from __future__ import annotations + +import logging +from collections.abc import Mapping, Sequence +from typing import Any + +from deerflow.runtime.events.message_identity import MESSAGE_SEQ_KEY, message_identity + +logger = logging.getLogger(__name__) + +__all__ = ["stamp_messages_with_seq"] + + +async def stamp_messages_with_seq(store: Any, thread_id: str, messages: Sequence[Any]) -> list[Any]: + """Return *messages* with ``MESSAGE_SEQ_KEY`` attached where the feed knows one. + + The input is never mutated: entries that gain a seq are shallow-copied, and + everything else is passed through as-is. A missing store, an entry that is + not a mapping, an identity the feed does not know, and a failing lookup all + degrade to "no seq" rather than raising — placement is an enhancement, and a + client without it falls back to its own ordering rule. + """ + if store is None or not messages: + return list(messages) + + identities = [message_identity(m) if isinstance(m, Mapping) else None for m in messages] + wanted = {identity for identity in identities if identity is not None} + if not wanted: + return list(messages) + + try: + found = await store.get_message_seqs(thread_id, sorted(wanted)) + except Exception: + logger.warning("Failed to resolve message seqs for thread %s", thread_id, exc_info=True) + return list(messages) + + return [ + {**message, "additional_kwargs": {**(message.get("additional_kwargs") or {}), MESSAGE_SEQ_KEY: seq}} if identity is not None and (seq := found.get(identity)) is not None and isinstance(message, Mapping) else message + for message, identity in zip(messages, identities, strict=True) + ] diff --git a/backend/packages/harness/deerflow/runtime/events/store/base.py b/backend/packages/harness/deerflow/runtime/events/store/base.py index df552e40d61..d7fe7ec88c1 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/base.py +++ b/backend/packages/harness/deerflow/runtime/events/store/base.py @@ -13,6 +13,7 @@ from __future__ import annotations import abc +from collections.abc import Sequence from deerflow.runtime.user_context import AUTO, _AutoSentinel @@ -148,6 +149,26 @@ async def get_last_visible_ai_seq_by_run( async def count_messages(self, thread_id: str) -> int: """Count displayable messages (category=message) in a thread.""" + @abc.abstractmethod + async def get_message_seqs(self, thread_id: str, identities: Sequence[str]) -> dict[str, int]: + """Return ``{identity: seq}`` for messages already persisted in this thread. + + A checkpoint carries no seq of its own and loses messages to + summarization, so a client merging a checkpoint frame with this + seq-ordered feed cannot place a surviving old message once the feed's + loaded page window no longer reaches back to it (#4666). The seq already + exists here; this exposes it without paging the whole feed. + + *identities* are the values produced by + ``deerflow.runtime.events.message_identity.message_identity`` — the same + rule the frontend applies — so both sides agree on what "same message" + means. Identities that are not persisted (or not `category="message"`) + are simply absent from the result: callers degrade to their own + placement rule rather than treating a miss as an error. When one + identity resolves to several rows, the earliest seq wins, so a message + re-persisted later keeps the position it first occupied. + """ + @abc.abstractmethod async def delete_by_thread(self, thread_id: str) -> int: """Delete all events for a thread. Return the number of deleted events.""" diff --git a/backend/packages/harness/deerflow/runtime/events/store/db.py b/backend/packages/harness/deerflow/runtime/events/store/db.py index 2227fd725fd..f98b72558a0 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/db.py +++ b/backend/packages/harness/deerflow/runtime/events/store/db.py @@ -16,6 +16,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from deerflow.persistence.models.run_event import RunEventRow +from deerflow.runtime.events.message_identity import message_identity from deerflow.runtime.events.store.base import RunEventStore from deerflow.runtime.user_context import AUTO, _AutoSentinel, get_current_user, resolve_user_id from deerflow.utils.time import coerce_iso @@ -388,6 +389,47 @@ async def count_messages( async with self._sf() as session: return await session.scalar(stmt) or 0 + async def get_message_seqs( + self, + thread_id, + identities, + *, + user_id: str | None | _AutoSentinel = AUTO, + ): + wanted = set(identities) + if not wanted: + return {} + resolved_user_id = resolve_user_id(user_id, method_name="DbRunEventStore.get_message_seqs") + # ``content`` is a TEXT column holding a JSON *string* (see + # ``_content_to_db``), not a JSON column, so the identity fields cannot + # be projected in SQL — the rows are decoded here instead. Only the two + # columns the lookup needs are selected, and the scan is bounded to this + # thread's message rows. + stmt = select(RunEventRow.seq, RunEventRow.content).where(RunEventRow.thread_id == thread_id, RunEventRow.category == "message").order_by(RunEventRow.seq) + if resolved_user_id is not None: + stmt = stmt.where(RunEventRow.user_id == resolved_user_id) + + found: dict[str, int] = {} + async with self._sf() as session: + result = await session.execute(stmt) + for seq, raw in result: + # Plain-text content (never a message dict) is skipped without + # paying for a failed JSON parse. + if not isinstance(raw, str) or not raw.startswith("{"): + continue + try: + content = json.loads(raw) + except (json.JSONDecodeError, ValueError): + continue + if not isinstance(content, dict): + continue + identity = message_identity(content) + # Earliest seq wins: a message re-persisted later keeps the + # position it first occupied in the feed. + if identity in wanted and identity not in found: + found[identity] = seq + return found + async def delete_by_thread( self, thread_id, diff --git a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py index 105ce894d28..374311b529a 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py +++ b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py @@ -30,6 +30,7 @@ from pathlib import Path from typing import Any +from deerflow.runtime.events.message_identity import message_identity from deerflow.runtime.events.store.base import RunEventStore from deerflow.runtime.user_context import AUTO, _AutoSentinel from deerflow.utils.thread_id import validate_thread_id @@ -290,6 +291,25 @@ async def count_messages(self, thread_id): all_events = await asyncio.to_thread(self._read_thread_events, thread_id) return sum(1 for e in all_events if e.get("category") == "message") + async def get_message_seqs(self, thread_id, identities): + wanted = set(identities) + if not wanted: + return {} + all_events = await asyncio.to_thread(self._read_thread_events, thread_id) + found: dict[str, int] = {} + for event in all_events: + if event.get("category") != "message": + continue + content = event.get("content") + if not isinstance(content, dict): + continue + identity = message_identity(content) + # Earliest seq wins: a message re-persisted later keeps the position + # it first occupied in the feed. + if identity in wanted and identity not in found: + found[identity] = event["seq"] + return found + async def delete_by_thread(self, thread_id): async with self._get_write_lock(thread_id): all_events = await asyncio.to_thread(self._read_thread_events, thread_id) diff --git a/backend/packages/harness/deerflow/runtime/events/store/memory.py b/backend/packages/harness/deerflow/runtime/events/store/memory.py index 113792560b7..53c1f5561a8 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/memory.py +++ b/backend/packages/harness/deerflow/runtime/events/store/memory.py @@ -9,6 +9,7 @@ import bisect from datetime import UTC, datetime +from deerflow.runtime.events.message_identity import message_identity from deerflow.runtime.events.store.base import RunEventStore from deerflow.runtime.user_context import AUTO, _AutoSentinel @@ -180,6 +181,22 @@ async def get_last_visible_ai_seq_by_run(self, thread_id, run_ids, *, user_id: s async def count_messages(self, thread_id): return len(self._messages.get(thread_id, [])) + async def get_message_seqs(self, thread_id, identities): + wanted = set(identities) + if not wanted: + return {} + found: dict[str, int] = {} + for record in self._messages.get(thread_id, []): + content = record.get("content") + if not isinstance(content, dict): + continue + identity = message_identity(content) + # Earliest seq wins: a message replaced later in the same thread + # keeps the position it first occupied in the feed. + if identity in wanted and identity not in found: + found[identity] = record["seq"] + return found + async def delete_by_thread(self, thread_id): events = self._events.pop(thread_id, []) self._messages.pop(thread_id, None) diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py index 6f3007fa1f2..537d41aa868 100644 --- a/backend/packages/harness/deerflow/runtime/journal.py +++ b/backend/packages/harness/deerflow/runtime/journal.py @@ -49,7 +49,6 @@ logger = logging.getLogger(__name__) _LEGACY_SUMMARY_MESSAGE_NAME = "summary" -_RECONCILED_TOOL_MESSAGE_NAMES = frozenset({"ask_clarification"}) _PERSISTED_HIDDEN_HUMAN_INPUT_RESPONSE_SOURCES = frozenset({"ask_clarification"}) @@ -613,16 +612,27 @@ def _final_output_messages(self, outputs: Any) -> list[Any]: return [] def _should_reconcile_tool_message(self, message: ToolMessage) -> bool: + """Whether a final-output ToolMessage still needs persisting. + + A middleware can answer a tool call itself and short-circuit execution, + so LangChain never emits ``on_tool_end`` and the result never reaches + the event store. The user saw that result during the run, and it + disappeared on reload (#4666). Any such result is reconciled here; the + scope is bounded by three independent conditions rather than a tool-name + allowlist: it 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 step feed), and it + must not already be persisted. + """ if message.additional_kwargs.get("hide_from_ui") is True: return False tool_call_id = getattr(message, "tool_call_id", None) if not isinstance(tool_call_id, str) or not tool_call_id: return False - tool_call_name = self._current_run_tool_call_names.get(tool_call_id) - if tool_call_name is None: - return False - message_name = getattr(message, "name", None) - if message_name not in _RECONCILED_TOOL_MESSAGE_NAMES and tool_call_name not in _RECONCILED_TOOL_MESSAGE_NAMES: + # The call must belong to this run: a retained ToolMessage from an + # earlier run is already persisted under its own run and must not be + # re-attributed here. + if self._current_run_tool_call_names.get(tool_call_id) is None: return False identity = self._message_identity(message) return identity is not None and identity not in self._persisted_tool_message_identities diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index 8d2a6f1b55e..31ff86391af 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -23,7 +23,7 @@ import sys import threading import weakref -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Mapping from contextlib import asynccontextmanager from dataclasses import dataclass, field from datetime import datetime @@ -48,6 +48,7 @@ graph_writable_channels, ) from deerflow.runtime.context_keys import CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY +from deerflow.runtime.events.message_identity import MESSAGE_SEQ_KEY, message_identity from deerflow.runtime.goal import ( DEFAULT_MAX_GOAL_CONTINUATIONS, DEFAULT_MAX_NO_PROGRESS_CONTINUATIONS, @@ -847,6 +848,10 @@ def _get_goal_evaluator_model() -> Any: ) return goal_evaluator_model + # Built once per run, not per _stream_once call: goal continuations + # re-enter the stream and would otherwise discard the resolved seqs. + seq_stamper = _MessageSeqStamper(event_store, thread_id) if "values" in requested_modes else None + async def _stream_once(input_payload: Any, stream_config: RunnableConfig) -> None: nonlocal llm_error_fallback_message file_tool_chunk_batcher = _LargeFileToolChunkBatcher() if "values" in requested_modes else None @@ -861,7 +866,10 @@ async def _stream_once(input_payload: Any, stream_config: RunnableConfig) -> Non break llm_error_fallback_message = llm_error_fallback_message or _extract_llm_error_fallback_message(chunk, pre_existing_message_ids) sse_event = _lg_mode_to_sse_event(single_mode) - await bridge.publish(run_id, sse_event, serialize(chunk, mode=single_mode)) + single_payload = serialize(chunk, mode=single_mode) + if single_mode == "values" and seq_stamper is not None: + single_payload = await seq_stamper.stamp(single_payload) + await bridge.publish(run_id, sse_event, single_payload) if single_mode == "custom": await subagent_events.add(chunk) return @@ -893,6 +901,7 @@ async def _stream_once(input_payload: Any, stream_config: RunnableConfig) -> Non namespace=namespace, file_tool_chunk_batcher=file_tool_chunk_batcher, subagent_events=subagent_events, + seq_stamper=seq_stamper, ) finally: stream_error = sys.exception() @@ -2258,6 +2267,59 @@ def _compose_sse_event(sse_event: str, namespace: tuple[str, ...]) -> str: return "|".join((sse_event, *namespace)) +class _MessageSeqStamper: + """Attach each already-persisted message's feed seq to a ``values`` frame. + + A checkpoint carries no seq of its own and loses messages to summarization, + so a client merging it with the seq-ordered thread feed cannot place a + surviving old message once the feed's loaded page window no longer reaches + back to it (#4666). The seq exists in the event store keyed by message + identity; this carries it to the client and writes nothing back to the + checkpoint. + + Cost is bounded to frames that introduce identities it has not resolved + yet. Messages produced by this run are not in the feed while streaming, so + they are looked up once, recorded as misses, and never retried — in a real + run the only frame that pays for a query is the one where compaction brings + older messages back into view. An unstamped message needs no seq: appending + it at the tail is already its correct position. + """ + + __slots__ = ("_store", "_thread_id", "_seqs", "_missing") + + def __init__(self, event_store: Any, thread_id: str) -> None: + self._store = event_store + self._thread_id = thread_id + self._seqs: dict[str, int] = {} + self._missing: set[str] = set() + + async def stamp(self, payload: Any) -> Any: + if self._store is None or not isinstance(payload, Mapping): + return payload + messages = payload.get("messages") + if not isinstance(messages, list) or not messages: + return payload + + identities = [message_identity(m) if isinstance(m, Mapping) else None for m in messages] + unresolved = {i for i in identities if i is not None and i not in self._seqs and i not in self._missing} + if unresolved: + try: + found = await self._store.get_message_seqs(self._thread_id, sorted(unresolved)) + except Exception: + # Placement is an enhancement: a client without seq falls back + # to its own ordering rule. Never fail the frame over it. + logger.warning("Failed to resolve message seqs for thread %s", self._thread_id, exc_info=True) + found = {} + self._seqs.update(found) + self._missing.update(unresolved - found.keys()) + + stamped = [ + {**message, "additional_kwargs": {**(message.get("additional_kwargs") or {}), MESSAGE_SEQ_KEY: seq}} if identity is not None and (seq := self._seqs.get(identity)) is not None and isinstance(message, Mapping) else message + for message, identity in zip(messages, identities, strict=True) + ] + return {**payload, "messages": stamped} + + async def _publish_stream_item( *, bridge: Any, @@ -2267,6 +2329,7 @@ async def _publish_stream_item( namespace: tuple[str, ...], file_tool_chunk_batcher: Any, subagent_events: Any, + seq_stamper: Any = None, ) -> None: """Publish one stream frame, preserving the subgraph namespace. @@ -2287,6 +2350,11 @@ async def _publish_stream_item( await bridge.publish(run_id, "messages", serialize(publish_chunk, mode="messages")) chunks_to_publish = file_tool_chunk_batcher.push(chunk) if mode == "messages" and file_tool_chunk_batcher is not None else [chunk] for publish_chunk in chunks_to_publish: - await bridge.publish(run_id, sse_event, serialize(publish_chunk, mode=mode)) + payload = serialize(publish_chunk, mode=mode) + if mode == "values" and seq_stamper is not None: + # Root frames only: a subagent's snapshot is not part of this + # thread's feed ordering (the namespaced branch returned above). + payload = await seq_stamper.stamp(payload) + await bridge.publish(run_id, sse_event, payload) if mode == "custom": await subagent_events.add(chunk) diff --git a/backend/packages/harness/deerflow/utils/messages.py b/backend/packages/harness/deerflow/utils/messages.py index 8e873c039aa..d9ce6a510f5 100644 --- a/backend/packages/harness/deerflow/utils/messages.py +++ b/backend/packages/harness/deerflow/utils/messages.py @@ -9,6 +9,28 @@ ORIGINAL_USER_CONTENT_KEY = "original_user_content" SUMMARY_MESSAGE_NAME = "summary" +#: Suffix ``DynamicContextMiddleware``'s ID-swap gives the real user message; the +#: reminder SystemMessage takes the original id so ``add_messages`` can replace it +#: in place. It lives here rather than beside the middleware because the message +#: identity rule in ``deerflow.runtime.events.message_identity`` needs it too, and +#: importing the middleware from there closes a cycle +#: (middleware -> deerflow.runtime -> worker -> events -> middleware). +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 + def message_content_to_text(content: Any) -> str: """Extract text from LangChain message content shapes.""" diff --git a/backend/tests/test_gateway_imports.py b/backend/tests/test_gateway_imports.py index 7272ea2f169..0e08ea31898 100644 --- a/backend/tests/test_gateway_imports.py +++ b/backend/tests/test_gateway_imports.py @@ -28,6 +28,36 @@ def test_gateway_app_imports_first_without_subagent_import_cycle() -> None: assert result.returncode == 0, result.stderr +def test_title_middleware_imports_without_message_identity_cycle() -> None: + """A middleware module must be importable as the process's first import. + + ``message_identity`` reaching back into ``agents.middlewares`` closed a cycle + (middleware -> deerflow.runtime -> worker -> events -> middleware) that only + stayed hidden while some earlier import happened to break it first. Running + ``tests/test_title_generation.py`` on its own was enough to hit it. + """ + result = subprocess.run( + [sys.executable, "-c", "from deerflow.agents.middlewares.title_middleware import TitleMiddleware; print(TitleMiddleware.__name__)"], + capture_output=True, + text=True, + env=_gateway_import_env(), + ) + assert result.returncode == 0, result.stderr + assert "TitleMiddleware" in result.stdout + + +def test_message_identity_imports_standalone() -> None: + """The seq-lookup identity helper must not require the agent package first.""" + result = subprocess.run( + [sys.executable, "-c", "from deerflow.runtime.events.message_identity import message_identity; print(message_identity({'id': 'x__user', 'type': 'human'}))"], + capture_output=True, + text=True, + env=_gateway_import_env(), + ) + assert result.returncode == 0, result.stderr + assert "message:x" in result.stdout + + def test_subagent_package_public_executor_exports_are_lazy_importable() -> None: """The package-level executor exports must not re-enter their own import.""" result = subprocess.run( diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py index 0d204b7dca1..778bc58b660 100644 --- a/backend/tests/test_gateway_services.py +++ b/backend/tests/test_gateway_services.py @@ -2762,3 +2762,30 @@ async def test_start_run_rejects_invalid_thread_id_before_resolving_dependencies assert exc_info.value.status_code == 422 assert "Invalid thread_id" in exc_info.value.detail + + +def test_normalize_input_strips_the_server_owned_message_seq(): + """`deerflow_seq` is display metadata the Gateway attaches on the way out. + + A client replaying messages (regenerate / edit-and-rerun) would otherwise + write it into the checkpoint, where it becomes wrong the moment the thread + is forked — a branch re-seeds its feed and reassigns seq (#4380). + """ + from app.gateway.services import normalize_input + from deerflow.runtime.events.message_identity import MESSAGE_SEQ_KEY + + result = normalize_input( + { + "messages": [ + { + "role": "human", + "content": "replayed turn", + "additional_kwargs": {MESSAGE_SEQ_KEY: 2, "keep_me": True}, + } + ] + } + ) + + kwargs = result["messages"][0].additional_kwargs + assert MESSAGE_SEQ_KEY not in kwargs + assert kwargs["keep_me"] is True diff --git a/backend/tests/test_run_event_store.py b/backend/tests/test_run_event_store.py index 00b543cb4d7..820367cf496 100644 --- a/backend/tests/test_run_event_store.py +++ b/backend/tests/test_run_event_store.py @@ -792,3 +792,255 @@ async def test_delete_by_run(self, tmp_path): assert c == 1 assert not (tmp_path / "jsonl" / "threads" / "t1" / "runs" / "r2.jsonl").exists() assert await s.count_messages("t1") == 1 + + +class TestGetMessageSeqs: + """Look up the thread-global seq of already-persisted messages by identity. + + A checkpoint carries no seq of its own and loses messages to summarization, + so a client merging it with the seq-ordered thread feed cannot place a + surviving old message (#4666). The seq already exists here, keyed by the + message's identity; this exposes it without paging the whole feed. + """ + + @pytest.mark.anyio + async def test_returns_seq_for_a_persisted_message(self, store): + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1", "content": "hello"}, + ) + + assert await store.get_message_seqs("t1", ["message:u1"]) == {"message:u1": 1} + + @pytest.mark.anyio + async def test_a_tool_message_is_identified_by_its_tool_call_id(self, store): + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.tool.result", + category="message", + content={"type": "tool", "id": "lc-abc", "tool_call_id": "call_1", "content": "OK"}, + ) + + assert await store.get_message_seqs("t1", ["tool:call_1"]) == {"tool:call_1": 1} + + @pytest.mark.anyio + async def test_the_injected_user_suffix_collapses_to_one_identity(self, store): + """DynamicContextMiddleware re-keys the submitted turn ``X`` to ``X__user``. + + The feed stores the ``__user`` copy while a caller may ask under either + spelling; both must resolve to the same row, or the very message this + feature exists to place would be the one it cannot find. + """ + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1__user", "content": "hello"}, + ) + + assert await store.get_message_seqs("t1", ["message:u1"]) == {"message:u1": 1} + + @pytest.mark.anyio + async def test_unknown_identities_are_absent_rather_than_an_error(self, store): + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1", "content": "hello"}, + ) + + result = await store.get_message_seqs("t1", ["message:u1", "message:never-persisted"]) + + assert result == {"message:u1": 1} + + @pytest.mark.anyio + async def test_non_message_events_are_not_looked_up(self, store): + await store.put( + thread_id="t1", + run_id="r1", + event_type="run.start", + category="trace", + content={"type": "human", "id": "u1"}, + ) + + assert await store.get_message_seqs("t1", ["message:u1"]) == {} + + @pytest.mark.anyio + async def test_lookup_is_scoped_to_the_thread(self, store): + await store.put( + thread_id="t2", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1", "content": "hello"}, + ) + + assert await store.get_message_seqs("t1", ["message:u1"]) == {} + + @pytest.mark.anyio + async def test_an_empty_request_does_not_scan(self, store): + assert await store.get_message_seqs("t1", []) == {} + + @pytest.mark.anyio + async def test_a_replaced_message_keeps_its_first_seq(self, store): + """A message re-persisted later must not jump to the tail of the feed.""" + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1", "content": "hello"}, + ) + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1", "content": "hello (edited)"}, + ) + + assert await store.get_message_seqs("t1", ["message:u1"]) == {"message:u1": 1} + + @pytest.mark.anyio + async def test_jsonl_store_resolves_identities(self, tmp_path): + from deerflow.runtime.events.store.jsonl import JsonlRunEventStore + + s = JsonlRunEventStore(base_dir=tmp_path / "jsonl") + await s.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1__user", "content": "hello"}, + ) + await s.put( + thread_id="t1", + run_id="r1", + event_type="llm.tool.result", + category="message", + content={"type": "tool", "tool_call_id": "call_1", "content": "OK"}, + ) + + assert await s.get_message_seqs("t1", ["message:u1", "tool:call_1"]) == { + "message:u1": 1, + "tool:call_1": 2, + } + + @pytest.mark.anyio + async def test_db_store_resolves_identities(self, tmp_path): + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + from deerflow.runtime.events.store.db import DbRunEventStore + + url = f"sqlite+aiosqlite:///{tmp_path / 'seqs.db'}" + await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path)) + try: + s = DbRunEventStore(get_session_factory()) + await s.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1__user", "content": "hello"}, + ) + await s.put( + thread_id="t1", + run_id="r1", + event_type="llm.tool.result", + category="message", + content={"type": "tool", "tool_call_id": "call_1", "content": "OK"}, + ) + await s.put( + thread_id="t1", + run_id="r1", + event_type="run.start", + category="trace", + content={"type": "human", "id": "ignored"}, + ) + + assert await s.get_message_seqs("t1", ["message:u1", "tool:call_1", "message:ignored"]) == { + "message:u1": 1, + "tool:call_1": 2, + } + finally: + await close_engine() + + +class TestStampMessagesWithSeq: + """Attach the feed seq to an arbitrary list of checkpoint messages. + + The streaming path stamps `values` frames as they are published, but a + client that merely opens a conversation never sees a frame: it reads the + checkpoint over REST. Without a seq there, a summarization-rescued early + turn has no absolute position and lands wherever the nearest anchor puts + it (#4666), which is behind the newest question rather than at the head. + """ + + @pytest.mark.anyio + async def test_stamps_a_persisted_message(self, store): + from deerflow.runtime.events.message_seq import stamp_messages_with_seq + + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1__user", "content": "MARK-FIRST"}, + ) + + stamped = await stamp_messages_with_seq(store, "t1", [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]) + + assert stamped[0]["additional_kwargs"]["deerflow_seq"] == 1 + + @pytest.mark.anyio + async def test_a_message_absent_from_the_feed_is_left_alone(self, store): + from deerflow.runtime.events.message_seq import stamp_messages_with_seq + + messages = [{"type": "ai", "id": "not-persisted", "content": "…"}] + + stamped = await stamp_messages_with_seq(store, "t1", messages) + + assert "deerflow_seq" not in (stamped[0].get("additional_kwargs") or {}) + + @pytest.mark.anyio + async def test_the_input_list_is_not_mutated(self, store): + from deerflow.runtime.events.message_seq import stamp_messages_with_seq + + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1", "content": "hi"}, + ) + original = [{"type": "human", "id": "u1", "content": "hi"}] + + await stamp_messages_with_seq(store, "t1", original) + + assert original[0].get("additional_kwargs") is None + + @pytest.mark.anyio + async def test_a_missing_store_returns_the_messages_unchanged(self): + from deerflow.runtime.events.message_seq import stamp_messages_with_seq + + messages = [{"type": "human", "id": "u1", "content": "hi"}] + + assert await stamp_messages_with_seq(None, "t1", messages) == messages + + @pytest.mark.anyio + async def test_a_failing_store_degrades_instead_of_raising(self, store): + """Placement is an enhancement; a broken lookup must not fail the read.""" + from deerflow.runtime.events.message_seq import stamp_messages_with_seq + + class _Broken: + async def get_message_seqs(self, *_args, **_kwargs): + raise RuntimeError("feed unavailable") + + messages = [{"type": "human", "id": "u1", "content": "hi"}] + + assert await stamp_messages_with_seq(_Broken(), "t1", messages) == messages diff --git a/backend/tests/test_run_journal.py b/backend/tests/test_run_journal.py index e531b9de0f5..ce045bcc499 100644 --- a/backend/tests/test_run_journal.py +++ b/backend/tests/test_run_journal.py @@ -330,7 +330,15 @@ async def test_root_chain_end_ignores_retained_old_tool_message_from_previous_ru assert not any(m["event_type"] == "llm.tool.result" for m in messages) @pytest.mark.anyio - async def test_root_chain_end_ignores_non_allowlisted_tool_message(self, journal_setup): + async def test_root_chain_end_ignores_subagent_tool_message(self, journal_setup): + """Reconciliation covers the lead agent's own calls only. + + A subagent's internal tool results belong to its own step feed + (``subagent.step``), not to the thread's message feed; + ``_remember_current_run_tool_calls`` records lead-agent calls only. + This is the boundary that keeps reconciliation safe now that it is no + longer narrowed to an ``ask_clarification`` allowlist. + """ from langchain_core.messages import ToolMessage j, store = journal_setup @@ -338,7 +346,7 @@ async def test_root_chain_end_ignores_non_allowlisted_tool_message(self, journal _make_llm_response("", tool_calls=[{"id": "call_search", "name": "web_search", "args": {"query": "deerflow"}}]), run_id=uuid4(), parent_run_id=None, - tags=["lead_agent"], + tags=["subagent:general-purpose"], ) tool_msg = ToolMessage(content="Search result", tool_call_id="call_search", name="web_search") @@ -372,6 +380,40 @@ async def test_root_chain_end_ignores_hidden_ask_clarification_tool_message(self messages = await store.list_messages("t1") assert not any(m["event_type"] == "llm.tool.result" for m in messages) + @pytest.mark.anyio + async def test_root_chain_end_reconciles_any_middleware_short_circuited_tool_message(self, journal_setup): + """A middleware that blocks a tool call still returns a user-visible result. + + ReadBeforeWriteMiddleware answers a blocked ``write_file`` with an error + ToolMessage instead of running the tool, so LangChain never emits + ``on_tool_end`` and the message never reached the event store. The user + saw it during the run and it vanished on reload (#4666). Reconciliation + is not specific to ``ask_clarification``: any visible tool result the + model asked for in this run belongs in the thread feed. + """ + from langchain_core.messages import ToolMessage + + j, store = journal_setup + j.on_llm_end( + _make_llm_response("", tool_calls=[{"id": "call_write", "name": "write_file", "args": {"path": "/mnt/user-data/outputs/a.txt"}}]), + run_id=uuid4(), + parent_run_id=None, + tags=["lead_agent"], + ) + blocked = ToolMessage( + content="Error: write_file blocked — read the file before writing to it", + tool_call_id="call_write", + name="write_file", + ) + + j.on_chain_end({"messages": [blocked]}, run_id=uuid4()) + await j.flush() + + messages = await store.list_messages("t1") + tool_results = [m for m in messages if m["event_type"] == "llm.tool.result"] + assert len(tool_results) == 1 + assert tool_results[0]["content"]["name"] == "write_file" + class TestCustomEvents: @pytest.mark.anyio diff --git a/backend/tests/test_threads_router.py b/backend/tests/test_threads_router.py index dd07023ef93..d7b8c6ca3ee 100644 --- a/backend/tests/test_threads_router.py +++ b/backend/tests/test_threads_router.py @@ -2874,3 +2874,90 @@ async def _collect(): assert all(cid is not None for cid in resp_ids), f"response missing checkpoint_id: {resp_ids}" assert set(resp_ids) <= set(ids), f"aput discarded endpoint-assigned id: returned {resp_ids}, stored {ids}" assert resp_ids[1] > resp_ids[0], f"endpoint-assigned uuid6 not preserved/ordered through aput: {resp_ids}" + + +class TestRestReadsCarryMessageSeq: + """Opening a conversation must expose the same feed seq the stream does. + + `_MessageSeqStamper` sits on the streaming publish path, so a client that + joins a live run gets placement information while one that merely opens the + thread does not — and opening is the common case. Without a seq the merge + falls back to the nearest shared anchor, which after summarization sits deep + inside the loaded page, so a rescued first user turn renders behind the + newest question instead of at the head (#4666). + """ + + @staticmethod + def _seed_thread(app, checkpointer, thread_id: str, *, with_feed: bool) -> None: + """Create the thread and its checkpoint without going through HTTP. + + ``POST /api/threads`` writes its own initial checkpoint, which would + overwrite the one under test. + """ + + async def _seed() -> None: + await app.state.thread_store.create(thread_id) + if with_feed: + await app.state.run_event_store.put( + thread_id=thread_id, + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1__user", "content": "MARK-FIRST"}, + ) + await checkpointer.aput( + {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}, + { + **empty_checkpoint(), + "id": str(uuid6(clock_seq=-2)), + "channel_values": {"messages": [HumanMessage(content="MARK-FIRST", id="u1__user")]}, + "channel_versions": {"messages": 1}, + }, + {"source": "loop", "step": 1, "writes": {}, "parents": {}}, + {"messages": 1}, + ) + + asyncio.run(_seed()) + + def _app_with_feed(self, thread_id: str): + from deerflow.runtime.events.store.memory import MemoryRunEventStore + + app, _store, checkpointer = _build_thread_app() + app.state.run_event_store = MemoryRunEventStore() + self._seed_thread(app, checkpointer, thread_id, with_feed=True) + return app + + def test_state_carries_the_seq_of_a_persisted_message(self) -> None: + app = self._app_with_feed("thread-seq-state") + + with TestClient(app) as client: + response = client.get("/api/threads/thread-seq-state/state") + + assert response.status_code == 200, response.text + messages = response.json()["values"]["messages"] + assert messages[0]["additional_kwargs"]["deerflow_seq"] == 1 + + def test_history_carries_the_seq_of_a_persisted_message(self) -> None: + app = self._app_with_feed("thread-seq-history") + + with TestClient(app) as client: + response = client.post("/api/threads/thread-seq-history/history", json={"limit": 1}) + + assert response.status_code == 200, response.text + messages = response.json()[0]["values"]["messages"] + assert messages[0]["additional_kwargs"]["deerflow_seq"] == 1 + + def test_a_message_the_feed_does_not_know_is_left_unstamped(self) -> None: + """Only persisted messages get a seq; the rest keep the weaving path.""" + from deerflow.runtime.events.store.memory import MemoryRunEventStore + + app, _store, checkpointer = _build_thread_app() + app.state.run_event_store = MemoryRunEventStore() + self._seed_thread(app, checkpointer, "thread-seq-unknown", with_feed=False) + + with TestClient(app) as client: + response = client.get("/api/threads/thread-seq-unknown/state") + + assert response.status_code == 200, response.text + messages = response.json()["values"]["messages"] + assert "deerflow_seq" not in (messages[0].get("additional_kwargs") or {}) diff --git a/backend/tests/test_worker_stream_subgraph_namespace.py b/backend/tests/test_worker_stream_subgraph_namespace.py index 7efef7e9892..aba9994d64f 100644 --- a/backend/tests/test_worker_stream_subgraph_namespace.py +++ b/backend/tests/test_worker_stream_subgraph_namespace.py @@ -562,3 +562,136 @@ async def test_without_stream_subgraphs_delegated_frames_stay_out_while_task_eve assert any(_PARENT_FINAL_ID in _collect_ids(payload) for payload in bare_values) custom_types = [payload.get("type") for event, payload in events if event == "custom" and isinstance(payload, dict)] assert "task_started" in custom_types and "task_completed" in custom_types + + +class TestMessageSeqStamping: + """A values frame carries the feed seq of messages already persisted. + + The checkpoint has no seq of its own and loses messages to summarization, + so a client merging it with the seq-ordered feed cannot place a surviving + old message once the feed's loaded page window no longer reaches back to it + (#4666). The worker already holds the event store, so it attaches the seq + that store assigned. Nothing is written back to the checkpoint. + """ + + @staticmethod + async def _seeded_store(): + from deerflow.runtime.events.store.memory import MemoryRunEventStore + + store = MemoryRunEventStore() + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1__user", "content": "MARK-FIRST"}, + ) + return store + + @pytest.mark.asyncio + async def test_root_values_frame_stamps_a_persisted_message(self): + from deerflow.runtime.runs.worker import _MessageSeqStamper + + bridge = _FakeBridge() + await _publish_stream_item( + bridge=bridge, + run_id="run-1", + mode="values", + chunk={"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]}, + namespace=(), + file_tool_chunk_batcher=None, + subagent_events=_FakeSubagentEvents(), + seq_stamper=_MessageSeqStamper(await self._seeded_store(), "t1"), + ) + + _run, _event, payload = bridge.published[0] + assert payload["messages"][0]["additional_kwargs"]["deerflow_seq"] == 1 + + @pytest.mark.asyncio + async def test_a_message_not_in_the_feed_is_left_unstamped(self): + """A message still streaming has no seq yet — and needs none: appending + it at the tail is already its correct position.""" + from deerflow.runtime.runs.worker import _MessageSeqStamper + + bridge = _FakeBridge() + await _publish_stream_item( + bridge=bridge, + run_id="run-1", + mode="values", + chunk={"messages": [{"type": "ai", "id": "not-persisted-yet", "content": "…"}]}, + namespace=(), + file_tool_chunk_batcher=None, + subagent_events=_FakeSubagentEvents(), + seq_stamper=_MessageSeqStamper(await self._seeded_store(), "t1"), + ) + + _run, _event, payload = bridge.published[0] + assert "deerflow_seq" not in (payload["messages"][0].get("additional_kwargs") or {}) + + @pytest.mark.asyncio + async def test_subgraph_frames_are_not_stamped(self): + """A subagent frame does not belong to the thread feed's ordering.""" + from deerflow.runtime.runs.worker import _MessageSeqStamper + + bridge = _FakeBridge() + await _publish_stream_item( + bridge=bridge, + run_id="run-1", + mode="values", + chunk={"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]}, + namespace=SUBAGENT_NS, + file_tool_chunk_batcher=None, + subagent_events=_FakeSubagentEvents(), + seq_stamper=_MessageSeqStamper(await self._seeded_store(), "t1"), + ) + + _run, _event, payload = bridge.published[0] + assert "deerflow_seq" not in (payload["messages"][0].get("additional_kwargs") or {}) + + @pytest.mark.asyncio + async def test_no_stamper_publishes_the_frame_unchanged(self): + bridge = _FakeBridge() + await _publish_stream_item( + bridge=bridge, + run_id="run-1", + mode="values", + chunk={"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]}, + namespace=(), + file_tool_chunk_batcher=None, + subagent_events=_FakeSubagentEvents(), + ) + + _run, _event, payload = bridge.published[0] + assert "deerflow_seq" not in (payload["messages"][0].get("additional_kwargs") or {}) + + @pytest.mark.asyncio + async def test_a_resolved_identity_is_not_looked_up_twice(self): + """Only a frame carrying messages it has not seen costs a query — in a + real run that is the compaction frame, not every frame.""" + from deerflow.runtime.runs.worker import _MessageSeqStamper + + store = await self._seeded_store() + calls: list[list[str]] = [] + original = store.get_message_seqs + + async def counting(thread_id, identities): + calls.append(list(identities)) + return await original(thread_id, identities) + + store.get_message_seqs = counting # type: ignore[method-assign] + stamper = _MessageSeqStamper(store, "t1") + frame = {"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]} + + for _ in range(3): + await _publish_stream_item( + bridge=_FakeBridge(), + run_id="run-1", + mode="values", + chunk=dict(frame), + namespace=(), + file_tool_chunk_batcher=None, + subagent_events=_FakeSubagentEvents(), + seq_stamper=stamper, + ) + + assert len(calls) == 1 diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index 5a168292f66..ee5aab9a972 100644 --- a/frontend/AGENTS.md +++ b/frontend/AGENTS.md @@ -76,7 +76,7 @@ The frontend is a stateful chat application. Users create **threads** (conversat full-file action; do not mount CodeMirror for that artifact until the user requests and receives the complete content. The Gateway retains range ownership and returns 206/416 through `FileResponse`. -3. `useThreadHistory` loads persisted conversation pages from `GET /api/threads/{id}/messages/page`, preserving the backend's thread-global event `seq`; rendering overlays checkpoint/live copies at their matching canonical identities (a summarized checkpoint may contain a protected early input plus a recent tail). Context-compaction rescue diffs every retained visible identity rather than slicing at the first anchor, and keeps a run-scoped ledger of committed visible messages so replacement updates and repeated rolling checkpoint windows cannot erase an already displayed step. The resolver suppresses checkpoint/transient prefixes whose canonical position is still behind an unloaded cursor page instead of collapsing that unknown gap before a recent anchor, then adds optimistic messages without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded. Dynamic context re-keys the submitted user message from `X` to `X__user`; UI identity matching normalizes that reserved suffix only for human messages so the submitted frame and checkpoint replacement remain one visible turn. A locally submitted turn also records its pre-submit identity baseline: if `messages-tuple` publishes new AI/tool steps before `values` publishes that turn's human message, render ordering moves only those non-baseline visible steps behind the new human while leaving history, hidden controls, and reconnected runs untouched. Keep that local order anchor through finish, stop, and stream error because the SDK's settled frame can retain transient event order; replace it on the next local submit and clear it on thread switch or replay-gap recovery. +3. `useThreadHistory` loads persisted conversation pages from `GET /api/threads/{id}/messages/page`, preserving the backend's thread-global event `seq`; rendering overlays checkpoint/live copies at their matching canonical identities (a summarized checkpoint may contain a protected early input plus a recent tail). Context-compaction rescue diffs every retained visible identity rather than slicing at the first anchor, and keeps a run-scoped ledger of committed visible messages so replacement updates and repeated rolling checkpoint windows cannot erase an already displayed step. A checkpoint/transient prefix whose canonical position is still behind an unloaded cursor page is woven in before the first shared anchor, not discarded: both the checkpoint and seq-sorted history place it earlier, so that position is known even when the pages between are not. It must never be appended to the tail (#4065) — the tail is provably wrong — but suppressing it entirely is how a user's own question vanished from a long thread once the first 50-row history page no longer reached back to it (#4666). A collapsed unloaded gap is recoverable by paging; a dropped message is not. Weaving alone restores the message but not its exact position — after compaction the live window carries too few anchors — so both sides now carry the backend's thread-global `additional_kwargs.deerflow_seq`: `buildVisibleHistoryMessages` copies each row's `seq`, and the Gateway stamps it onto `values` frame messages it has already persisted. A live message whose seq is below the loaded window's lower bound is placed ahead of everything on screen instead of before the nearest anchor, which is what puts a compaction-rescued first user turn back at the head rather than mid-transcript. That split happens _before_ the anchor walk, not inside it: a compacted checkpoint can share no identity at all with the loaded page — it keeps only the current run's recent tail, while the page on screen was fetched turns earlier — and the anchor walk then never runs at all, which is precisely when a rescued turn most needs its seq. Doing the split inside the walk left that case appending the message after the whole window (#4666), the one arrangement #4065 proved wrong. A message without a seq (still streaming, so not in the feed yet) keeps the weaving path — the tail is already its correct position. Optimistic messages are then added without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded. Dynamic context re-keys the submitted user message from `X` to `X__user`; UI identity matching normalizes that reserved suffix only for human messages so the submitted frame and checkpoint replacement remain one visible turn. A locally submitted turn also records its pre-submit identity baseline: if `messages-tuple` publishes new AI/tool steps before `values` publishes that turn's human message, render ordering moves only those non-baseline visible steps behind the new human while leaving history, hidden controls, and reconnected runs untouched. Keep that local order anchor through finish, stop, and stream error because the SDK's settled frame can retain transient event order; replace it on the next local submit and clear it on thread switch or replay-gap recovery. 4. Stop actions call the LangGraph SDK stream stop path; `core/threads/hooks.ts` invalidates current-thread, thread-history, token-usage, and sidebar/search caches immediately and schedules one follow-up refetch because SDK stop may finish via abort + fire-and-forget cancel before backend title finalization commits 5. TanStack Query manages server state; localStorage stores user settings. The Settings > Tools MCP switch calls the targeted `PATCH /api/mcp/config` diff --git a/frontend/src/core/threads/hooks.ts b/frontend/src/core/threads/hooks.ts index 4fefa2eb579..6ff0c45339a 100644 --- a/frontend/src/core/threads/hooks.ts +++ b/frontend/src/core/threads/hooks.ts @@ -171,6 +171,10 @@ const EMPTY_MESSAGES: Message[] = []; const EMPTY_RUN_MESSAGES: RunMessage[] = []; const EMPTY_MESSAGE_IDENTITIES: readonly string[] = []; const INJECTED_USER_MESSAGE_ID_SUFFIX = "__user"; +// Thread-global feed position, attached by the backend to history rows and to +// `values` frame messages it has already persisted. Mirrors MESSAGE_SEQ_KEY in +// `deerflow/runtime/events/message_identity.py`. +const MESSAGE_SEQ_KEY = "deerflow_seq"; const EMPTY_THREAD_VALUES: AgentThreadState = { title: "", @@ -188,6 +192,12 @@ const SUMMARIZATION_MIDDLEWARE_UPDATE_KEYS = new Set([ "DeerFlowSummarizationMiddleware.before_model", ]); +/** Thread-global feed position, when the backend has attached one. */ +function messageSeq(message: Message): number | undefined { + const seq = message.additional_kwargs?.[MESSAGE_SEQ_KEY]; + return typeof seq === "number" ? seq : undefined; +} + function messageIdentity(message: Message): string | undefined { if ( "tool_call_id" in message && @@ -309,9 +319,16 @@ export function buildVisibleHistoryMessages( // Carry the owning run_id onto the content message so historical subtask // cards can fetch their persisted step history on expand (#3779). run_id // lives on the RunMessage wrapper and would otherwise be dropped here. + // seq rides along for the same reason: it is the thread-global position + // this feed is ordered by, and merging needs it on the message itself to + // place a checkpoint copy that falls outside the loaded window (#4666). ...visibleRows.map((message) => ({ ...message.content, run_id: message.run_id, + additional_kwargs: { + ...message.content.additional_kwargs, + [MESSAGE_SEQ_KEY]: message.seq, + }, })), ]); } @@ -499,7 +516,41 @@ export function mergeMessages( const beforeAnchor = new Map(); let pending: Message[] = []; let lastAnchorIdentity: string | undefined; - let hasSharedAnchor = false; + + // Lower bound of the history page window that is currently loaded. A live + // message whose seq is below it belongs before everything on screen, which + // is knowledge the anchor weaving below cannot reach: the anchor only says + // "earlier than this row", and after compaction the nearest anchor can sit + // deep inside the window (#4666 — measured at row 25 of 50). + const canonicalMinSeq = canonical.reduce( + (min, message) => { + const seq = messageSeq(message); + return seq !== undefined && (min === undefined || seq < min) ? seq : min; + }, + undefined, + ); + const beforeWindow: Message[] = []; + + // Split off what the feed places before the loaded window BEFORE the anchor + // walk rather than inside it. A summarized checkpoint can share no identity + // at all with the loaded page — compaction keeps only this run's recent tail, + // while the page on screen was fetched turns earlier — and the anchor loop + // then never runs. That is exactly when a rescued early turn most needs its + // seq: user submits, waits without reloading, compaction fires, and the + // message is appended to the tail instead (#4666). + const liveInWindow: Message[] = []; + for (const message of live) { + const seq = messageSeq(message); + if ( + seq !== undefined && + canonicalMinSeq !== undefined && + seq < canonicalMinSeq + ) { + beforeWindow.push(message); + } else { + liveInWindow.push(message); + } + } // A summarized checkpoint is not necessarily a contiguous history suffix: // middleware may retain protected prompt/input messages at the front and a @@ -507,7 +558,7 @@ export function mergeMessages( // replacing the canonical copy in place. New live messages are woven before // the next shared anchor (or after the last one), so a protected early input // can never be moved to the tail by global last-copy deduplication. - for (const message of live) { + for (const message of liveInWindow) { const identity = messageIdentity(message); const canonicalMessage = identity ? canonicalByIdentity.get(identity) @@ -517,17 +568,22 @@ export function mergeMessages( continue; } - if (pending.length > 0 && hasSharedAnchor) { + // A summarized checkpoint may start with a protected message whose true + // canonical position is separated from this anchor by unloaded pages — + // rescued dynamic-context messages are the common case. Its position + // relative to this anchor is still known (both the checkpoint and + // seq-sorted history place it earlier), so it is woven in before the + // anchor like any other live-only segment. Dropping it instead was how a + // user's own question disappeared from a long thread once the first + // history page no longer reached back to it (#4666): a collapsed unloaded + // gap is recoverable by paging, a discarded message is not. + if (pending.length > 0) { beforeAnchor.set(identity, [ ...(beforeAnchor.get(identity) ?? []), ...pending, ]); } - // A summarized checkpoint may start with a protected message whose true - // canonical position is separated from this anchor by unloaded pages. - // Suppress that ambiguous prefix instead of visually collapsing the gap. pending = []; - hasSharedAnchor = true; lastAnchorIdentity = identity; // A hidden checkpoint control message must not replace a visible canonical @@ -543,7 +599,7 @@ export function mergeMessages( let canonicalAndLive: Message[]; if (!lastAnchorIdentity) { - canonicalAndLive = [...canonical, ...live]; + canonicalAndLive = [...canonical, ...liveInWindow]; } else { canonicalAndLive = []; for (const message of canonical) { @@ -564,6 +620,9 @@ export function mergeMessages( } const merged = dedupeMessagesByIdentity([ + ...[...beforeWindow].sort( + (left, right) => (messageSeq(left) ?? 0) - (messageSeq(right) ?? 0), + ), ...canonicalAndLive, ...optimisticMessages, ]); diff --git a/frontend/tests/unit/core/threads/message-merge.test.ts b/frontend/tests/unit/core/threads/message-merge.test.ts index 991b997e1f6..9a3618a7fc8 100644 --- a/frontend/tests/unit/core/threads/message-merge.test.ts +++ b/frontend/tests/unit/core/threads/message-merge.test.ts @@ -53,7 +53,17 @@ test("mergeMessages removes duplicate messages already present in history", () = expect(mergeMessages([human, ai, human, ai], [], [])).toEqual([human, ai]); }); -test("mergeMessages does not collapse an unloaded gap before the first shared anchor", () => { +test("mergeMessages keeps a protected early message before the first shared anchor instead of dropping it", () => { + // #4065 established that an early message rescued by summarization must not + // be appended to the tail: its canonical position is earlier, and the tail is + // provably wrong. Suppressing it entirely was the other half of that fix, and + // it is how a user's own question disappeared from a long thread once the + // first history page no longer reached back to it (#4666). + // + // Both concerns hold at once: the message stays before the first shared + // anchor (never the tail), which is the one position both the checkpoint and + // seq-sorted history agree on. The gap to the unloaded pages remains, but a + // gap is recoverable by paging — a dropped message is not. const protectedEarly = { id: "protected-early", type: "human", @@ -72,7 +82,7 @@ test("mergeMessages does not collapse an unloaded gap before the first shared an expect( mergeMessages([latestHuman, latestAi], [protectedEarly, latestHuman], []), - ).toEqual([latestHuman, latestAi]); + ).toEqual([protectedEarly, latestHuman, latestAi]); }); test("mergeMessages lets live thread messages replace overlapping history", () => { @@ -879,8 +889,8 @@ test("buildVisibleHistoryMessages filters superseded runs but keeps regenerated // run_id is carried onto each content message (#3779) so historical subtask // cards can fetch their persisted step history on expand. expect(buildVisibleHistoryMessages(rows, new Set(["run-old"]))).toEqual([ - { ...newHuman, run_id: "run-new" }, - { ...newAi, run_id: "run-new" }, + { ...newHuman, run_id: "run-new", additional_kwargs: { deerflow_seq: 3 } }, + { ...newAi, run_id: "run-new", additional_kwargs: { deerflow_seq: 4 } }, ]); }); @@ -2093,3 +2103,162 @@ test("refresh reconstructs the same 1-to-6 order from run events without a bridg ), ).toEqual(["1", "2", "3", "4", "5", "6"]); }); + +test("a compacted checkpoint's protected user message survives a history page window that misses it (#4666)", () => { + // Captured from a real two-round long run: once the thread passes the + // 50-row `/messages/page` window AND context compaction fires, the two + // sources stop overlapping at the head. History's first page starts + // mid-run, while the compacted checkpoint still carries the turn's first + // user message (summarization rescues the dynamic-context triplet). + // + // That message is the user's own question. Suppressing it because its + // canonical position sits in an unloaded page makes it vanish from the + // transcript entirely — the "用户消息消失" reports in #4666 / #4508 / #4363. + const reminder = { + id: "u1", + type: "system", + content: "…", + additional_kwargs: { hide_from_ui: true, dynamic_context_reminder: true }, + } as unknown as Message; + const firstUserMessage = { + id: "u1__user", + type: "human", + content: "MARK-FIRST-QUESTION", + } as Message; + const recentStep = { + id: "step-40", + type: "ai", + content: "step 40", + } as Message; + const laterUserMessage = { + id: "u2", + type: "human", + content: "SECOND-QUESTION", + } as Message; + + // First history page: starts mid-run, has_more=true — no first user message. + const canonicalWindow = [recentStep, laterUserMessage]; + // Compacted checkpoint: reminder + rescued first user message + recent tail. + const compactedCheckpoint = [reminder, firstUserMessage, recentStep]; + + const merged = mergeMessages(canonicalWindow, compactedCheckpoint, []); + + expect(merged.map((message) => message.content)).toContain( + "MARK-FIRST-QUESTION", + ); +}); + +test("a checkpoint message earlier than the loaded window is placed by its seq (#4666)", () => { + // With `deerflow_seq` on both sides, placement stops being a guess. Captured + // shape: after compaction the checkpoint still holds the turn's first user + // message (seq=2) while the first history page starts at seq=29, so the only + // anchor available to the old rule sat 25 rows into the window. + const withSeq = (message: Message, seq: number) => + ({ + ...message, + additional_kwargs: { ...message.additional_kwargs, deerflow_seq: seq }, + }) as Message; + + const firstUserMessage = withSeq( + { + id: "u1__user", + type: "human", + content: "MARK-FIRST-QUESTION", + } as Message, + 2, + ); + const windowStep = withSeq( + { id: "step-29", type: "ai", content: "…step 29" } as Message, + 29, + ); + const laterUserMessage = withSeq( + { id: "u2", type: "human", content: "SECOND-QUESTION" } as Message, + 41, + ); + + const anchorStep = withSeq( + { id: "step-58", type: "ai", content: "…step 58" } as Message, + 58, + ); + + // The anchor must sit INSIDE the window, as it does in the captured run + // (canonical #25 of 50): weaving before the anchor is what puts the message + // in the middle, and only seq can say it belongs at the head. + const canonicalWindow = [windowStep, laterUserMessage, anchorStep]; + const compactedCheckpoint = [firstUserMessage, anchorStep]; + + expect( + mergeMessages(canonicalWindow, compactedCheckpoint, []).map( + (m) => m.content, + ), + ).toEqual(["MARK-FIRST-QUESTION", "…step 29", "SECOND-QUESTION", "…step 58"]); +}); + +test("buildVisibleHistoryMessages carries each row's seq onto the message", () => { + const rows = [ + { + run_id: "run-1", + seq: 7, + content: { id: "m1", type: "human", content: "hi" } as Message, + metadata: { caller: "" }, + created_at: "2026-08-04T00:00:00Z", + }, + ] as RunMessage[]; + + expect( + buildVisibleHistoryMessages(rows, new Set())[0]!.additional_kwargs + ?.deerflow_seq, + ).toBe(7); +}); + +test("a checkpoint message earlier than the loaded window is placed by its seq even when the two sides share no anchor (#4666)", () => { + // What a user hits by opening an old, already-summarized conversation and + // sending a new message. The loaded page is the newest rows from BEFORE that + // turn; the compacted checkpoint holds only the rescued first user message + // plus steps of the new run, which are not in the feed yet. The two sides + // therefore share no identity at all and the anchor walk never runs — so the + // rescued turn was appended after the whole window (measured at row 50 of 50 + // on a reproducing run) even though its seq was known the entire time. + const withSeq = (message: Message, seq: number) => + ({ + ...message, + additional_kwargs: { ...message.additional_kwargs, deerflow_seq: seq }, + }) as Message; + + const rescuedFirstTurn = withSeq( + { + id: "u1__user", + type: "human", + content: "MARK-FIRST-QUESTION", + } as Message, + 2, + ); + const loadedWindow = [ + withSeq( + { id: "step-172", type: "ai", content: "…step 172" } as Message, + 172, + ), + withSeq( + { id: "step-174", type: "ai", content: "…step 174" } as Message, + 174, + ), + ]; + // Steps of the run the user just started: still streaming, so no seq yet, and + // no identity in common with the page on screen. + const newRunSteps = [ + { id: "step-new-1", type: "ai", content: "…new step 1" } as Message, + { id: "step-new-2", type: "ai", content: "…new step 2" } as Message, + ]; + + expect( + mergeMessages(loadedWindow, [rescuedFirstTurn, ...newRunSteps], []).map( + (m) => m.content, + ), + ).toEqual([ + "MARK-FIRST-QUESTION", + "…step 172", + "…step 174", + "…new step 1", + "…new step 2", + ]); +});