rayhpeng cd35363a05
fix(history): early user messages vanish or jump mid-run when pagination and context compaction overlap (#4696)
* fix(history): stop dropping user messages that fall outside the loaded page window

Two independent paths made a user's own message disappear from a long thread
(#4666, #4508, #4363). Both are reproduced by a real two-round run: once the
thread passes the 50-row `/messages/page` window AND context compaction fires,
the two sources of truth stop overlapping at the head.

1. Middleware-answered tool results never reached the event store. A middleware
   that short-circuits a tool call (e.g. ReadBeforeWriteMiddleware's blocked
   write) returns a user-visible ToolMessage, but LangChain never emits
   `on_tool_end`, so RunJournal never persisted it — the user saw it during the
   run and it vanished on reload. RunJournal already reconciles final-output
   tool messages, but only for an `ask_clarification` allowlist. The allowlist
   is removed; scope stays bounded by the three conditions that actually matter
   (visible, this run's lead agent, not already persisted), so subagent results
   still stay in their own step feed.

2. mergeMessages discarded the checkpoint prefix before the first shared anchor.
   #4065 correctly established that a summarization-rescued early message must
   not be appended to the tail, and suppressed it instead. That suppression is
   what deletes the message when the first history page no longer reaches back
   to it. It is now woven in before the first shared anchor — the one position
   both the checkpoint and seq-sorted history agree on — so #4065's invariant
   (never the tail) still holds. A collapsed unloaded gap is recoverable by
   paging; a dropped message is not.

Verified against real captured payloads from the reproducing run: the first user
message returns to the transcript. Its exact position is still approximate —
after compaction the live window carries too few anchors to place it precisely,
which only seq-based ordering can close.

Backend: 10809 passed (baseline 10808; same 15 pre-existing failures in
browser/crawler community tools). Frontend: 986 passed, typecheck + eslint clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* feat(events): look up a persisted message's seq by identity

Groundwork for placing checkpoint messages in the seq-ordered thread feed
(#4666). A checkpoint carries no seq of its own and loses messages to
summarization, so once the feed's 50-row page window no longer reaches back to a
surviving old message, a client has nothing to place it by. The seq already
exists in run_events keyed by the message id — this exposes it without paging
the whole feed.

`message_identity` is the backend half of the identity rule the frontend applies
in `hooks.ts::messageIdentity`: a ToolMessage is keyed by `tool_call_id`, and
DynamicContextMiddleware's `X` / `X__user` human copies collapse to one identity.
The two halves must stay in sync — a mismatch is silent, degrading placement
rather than raising.

`get_message_seqs` is implemented for all three stores. Misses are absent from
the result rather than an error, so callers degrade to their own placement rule;
the earliest seq wins when one identity resolves to several rows, so a
re-persisted message keeps the position it first occupied. The DB store decodes
rows in Python because `content` is a TEXT column holding a JSON string, not a
JSON column — the identity fields cannot be projected in SQL.

Nothing consumes this yet; no behavior change.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* feat(runtime): carry each persisted message's feed seq on values frames

Attaches `additional_kwargs.deerflow_seq` to messages in a root `values` frame
that the thread feed already holds, so a client can place a message the
checkpoint kept but its loaded history page window no longer reaches (#4666).
Nothing is written back to the checkpoint: the seq is added when the frame is
serialized and belongs to that frame only.

Cost is bounded to frames introducing identities the run has not resolved yet.
Messages this run produces 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. Measured on a reproducing two-round run: 1 lookup across 25
values frames.

The stamper is built once per run rather than per `_stream_once`, or a goal
continuation would discard the resolved seqs. Subgraph frames are not stamped:
a subagent's snapshot is not part of this thread's feed ordering. A lookup
failure logs and leaves the frame unstamped rather than failing it — placement
is an enhancement and clients fall back to their own ordering rule.

Frontend does not read the field yet; no behavior change.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(gateway): strip the server-owned message seq from untrusted input

`deerflow_seq` is display metadata the Gateway attaches when it serializes a
values frame. 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).

Joins the existing server-owned key set, so it follows the same trusted-internal
rule as the dynamic-context and view-image markers.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(frontend): place a checkpoint message by its feed seq, not its nearest anchor

Completes #4666. Weaving a compaction-rescued message before the first shared
anchor keeps it in the transcript, but not in the right place: after compaction
the live window carries too few anchors, and the nearest one can sit deep inside
the loaded page window — measured at row 25 of 50 on a reproducing run, which is
why the first user turn rendered mid-transcript instead of at the head.

Both sides now carry the backend's thread-global seq. `buildVisibleHistoryMessages`
copies each row's `seq` onto the message (same shape as the existing `run_id`),
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 rather than before the nearest anchor. A message with no
seq — still streaming, so not in the feed yet — keeps the weaving path, since the
tail is already its correct position.

Verified against the captured payloads of the reproducing run: the first user
message goes from absent, to #13 (behind the second question), to #0.

Frontend: 988 passed, typecheck + eslint clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(frontend): place a pre-window checkpoint message even when no anchor is shared

Also #4666. Placing a compaction-rescued message by its feed seq was gated on
reaching a shared anchor, because the split ran inside the anchor walk. When the
loaded page and the live checkpoint share no identity at all, that walk never
runs and the message fell through to `[...canonical, ...live]` — appended after
the entire window, the one arrangement #4065 proved wrong, with its seq known
the whole time.

That is not a corner case. Open an old, already-summarized conversation and send
a message: the page on screen is the newest rows from before that turn, while
the checkpoint holds the rescued first user turn plus steps of the new run that
are not in the feed yet. On a reproducing run the two sides shared zero anchors
and the user's own first question rendered at row 50 of 50 — the reported
"first message jumps to the bottom".

Split `beforeWindow` out of `live` before walking anchors, walk `liveInWindow`,
and use it for the no-anchor branch as well, so a message routed ahead of the
window is not re-appended at the tail by dedup.

Measured on captured payloads of a reproducing run (real gateway, real
compaction), first user message position:

  no shared anchor:  row 50 -> row 0, seq order monotonic again
  shared anchors:    row 0 -> row 0 (unchanged)
  paged to the top:  row 0 -> row 0 (unchanged)

Regression test verified red-green: reverting the fix fails it with the message
rendered after the window.

Frontend: 989 passed, eslint + tsc clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(gateway): stamp the message feed seq on checkpoint reads, not only on stream frames

Completes #4666. `_MessageSeqStamper` sits on the streaming publish path, so a
client that joins a live run learns where a summarization-rescued turn belongs
while a client that merely opens the conversation does not — and opening is the
common case. `GET /threads/{id}/state` and `POST /threads/{id}/history` returned
the checkpoint with no seq at all, so the merge fell back to the nearest shared
anchor, which after summarization sits deep inside the loaded page.

Reproduced in a browser against a real gateway, on a thread that had already
compacted: the user's first question rendered at row 320 of 389, behind the
newest question instead of at the head. Both reads showed 0 of 13 messages
carrying a seq. That is the reported symptom, still present after the streaming
fix.

Add `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. Resolve the store
through `_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.

After the fix, on the same thread in the same browser: 13 of 13 messages carry a
seq and the first question renders at the head, ahead of the newest one.

Backend: ruff clean, 326 passed across the touched suites.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* refactor(harness): move the injected-user-id suffix helpers to utils.messages to break an import cycle

message_identity imported strip_injected_user_message_id_suffix from the
dynamic-context middleware, closing a cycle (middleware -> deerflow.runtime
-> worker -> events -> middleware) that only stayed hidden while an earlier
import happened to break it. Define INJECTED_USER_MESSAGE_ID_SUFFIX and the
strip helper in deerflow.utils.messages and re-export them from the
middleware so existing importers keep working.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(docs): improve formatting and clarity in AGENTS.md and message-merge.test.ts

* perf(events): stop the seq scan once every wanted identity is resolved

Rows past the last wanted seq can only be re-persisted copies that
already lose the earliest-seq-wins tiebreak, so all three stores now
break out of the scan (and the db store out of its per-row JSON
decoding) once found covers wanted. Matters most for /state and
/history reads of long threads, where this lookup runs with no run
cache and a typically tiny wanted set.

Raised by review on #4696.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* refactor(events): share the seq-stamping expression between the two stampers

The walrus-plus-merge expression was duplicated verbatim between
stamp_messages_with_seq and _MessageSeqStamper.stamp — two counterparts
of one rule where silent divergence is the likely failure mode if only
one side is edited. Both now call attach_message_seq next to
MESSAGE_SEQ_KEY in message_identity.py. The trailing
isinstance(message, Mapping) guard was unreachable (a non-Mapping entry
already got identity = None) and is gone with the extraction.

Raised by review on #4696.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(events): seq stamping survives launch paths without user context

The db store's get_message_seqs defaults to user_id=AUTO, which raises
when no user is in the contextvar — the first strict-AUTO read ever
called from the worker context. On a launch path that never inherits
the auth context (e.g. a null-owner scheduled task), stamp()'s except
clause swallowed that into a per-frame warning and silently disabled
seq stamping for exactly the background runs that need it.

The stamper now soft-resolves the user id once at build time — the
same rule as the worker's write paths beside it (unset -> no filter)
— and passes it explicitly. jsonl/memory stores gain the same
user_id kwarg the base list_messages contract already carries.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* perf(events): SQL-prefilter the message seq lookup's candidate rows

get_message_seqs scanned and JSON-decoded every message row of the
thread: the early exit never fires when a wanted identity is absent
from the feed (a message still streaming, or checkpoint-only), and
/state / /history reads want the newest messages, so the ascending
scan traversed essentially the whole feed — with the content column
carrying full tool outputs, that is heavy I/O plus N JSON parses on
exactly the long threads this lookup exists for.

A LIKE prefilter now keeps that cost in SQL: only rows containing a
wanted raw id as a substring are fetched and decoded. False positives
are re-checked by message_identity; LIKE wildcards are escaped; an id
json.dumps would escape (breaking the verbatim-substring guarantee)
falls the whole set back to the full scan rather than silently
missing.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* docs(agents): sink runtime mechanism docs below the gateway guidance budget

Merging main pushed backend/app/gateway/AGENTS.md past its 40KB soft
budget (main had left 81 bytes of headroom). Per the nearest-file rule,
move the mechanism detail of the message-seq stamping and run-delivery
receipt sections — both owned by runtime/ code — into
packages/harness/deerflow/runtime/AGENTS.md, leaving the gateway file
the REST-surface summary and a pointer. The seq section also documents
the stamper's build-time soft user-id resolution and the db store's SQL
prefilter from the review follow-ups.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* docs(agents): sink durable-MCP task detail below the backend guidance budget

Merging main pushed backend/AGENTS.md past its 24KB module soft budget
(main itself is at 24762 after #4848 — this branch adds zero net bytes
to the file). Per the nearest-file rule, move the two durable-MCP task
runtime bullets' mechanism detail into
packages/harness/deerflow/mcp/AGENTS.md, leaving summaries and
pointers; this also restores ~2KB of headroom so the next merge does
not trip the same wire.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(events): re-ask a message-seq miss once the feed advances

The run-scoped stamper cached lookup misses for the whole run. A message
this run produces reaches a values frame before RunJournal flushes it, so
its first lookup legitimately misses — and the journal persists it moments
later, giving it a feed seq the stamper never asks for again. A long run
that afterwards rolls past the history page and compacts then carries that
message unstamped, back to the approximate anchor placement this stamper
exists to replace (#4666). A transient store error had the same permanent
effect, since the except clause degrades to an empty result.

A miss is now provisional while a hit stays final: RunJournal counts its
successful event-store writes as `feed_generation`, and the stamper re-asks
a missed identity only once that counter moves. Retrying is therefore
bounded by feed writes rather than by frames — the per-frame query the
run-scoped cache was built to avoid — and a failed lookup costs one
generation instead of the run.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
2026-09-01 22:04:17 +08:00

173 lines
7.1 KiB
Python

from __future__ import annotations
from collections.abc import Mapping
from copy import deepcopy
from typing import Any
from langchain_core.messages import HumanMessage
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."""
if isinstance(content, str):
return content
if isinstance(content, list):
parts: list[str] = []
for item in content:
if isinstance(item, str):
parts.append(item)
elif isinstance(item, dict):
text = item.get("text")
if isinstance(text, str):
parts.append(text)
return "\n".join(part for part in parts if part)
return str(content)
def message_to_text(message: Any, *, text_attribute_fallback: bool = False) -> str:
"""Extract display text from a whole message (``BaseMessage`` or dict-shaped).
Reads ``content`` from either an attribute (``BaseMessage``) or a mapping key
(``run_events`` rows are dicts), then walks the mixed ``content`` shapes:
plain string; a list of string / ``{"text": ...}`` / nested ``{"content": ...}``
blocks joined without a separator; or a mapping with a ``text``/``content`` key.
Set ``text_attribute_fallback=True`` to fall back to ``message.text`` when
content yields nothing (matches ``RunJournal._message_text``).
Unlike :func:`message_content_to_text` (which takes raw ``content`` and joins
list blocks with newlines), this keeps the no-separator join and the broader
shape handling that several call sites had each reimplemented.
"""
content = message.get("content") if isinstance(message, Mapping) else getattr(message, "content", None)
if isinstance(content, str):
return content
if isinstance(content, list):
parts: list[str] = []
for block in content:
if isinstance(block, str):
parts.append(block)
elif isinstance(block, Mapping):
text = block.get("text")
if isinstance(text, str):
parts.append(text)
else:
nested = block.get("content")
if isinstance(nested, str):
parts.append(nested)
return "".join(parts)
if isinstance(content, Mapping):
for key in ("text", "content"):
value = content.get(key)
if isinstance(value, str):
return value
if text_attribute_fallback:
text = getattr(message, "text", None)
if isinstance(text, str):
return text
return ""
def get_original_user_content_text(content: Any, additional_kwargs: Mapping[str, Any] | None) -> str:
"""Return pre-middleware user text when available, otherwise content text."""
original_content = (additional_kwargs or {}).get(ORIGINAL_USER_CONTENT_KEY)
if isinstance(original_content, str):
return original_content
return message_content_to_text(content)
def restore_original_human_message(message: HumanMessage) -> HumanMessage:
"""Build the UI-facing copy of a model-sanitized human message.
Input middleware intentionally keeps the original user text in
``additional_kwargs`` while replacing the model-facing text with transport
wrappers and other context. Run-event history must persist the original
text without mutating the message that is actually sent to the model.
Mixed content is already normalized by the sanitization middleware to a
single text block. For defensive compatibility, multiple current text
blocks are collapsed at the first text position while every non-text block
retains its value and relative order.
"""
original_content = message.additional_kwargs.get(ORIGINAL_USER_CONTENT_KEY)
if not isinstance(original_content, str):
return message
additional_kwargs = dict(message.additional_kwargs)
additional_kwargs.pop(ORIGINAL_USER_CONTENT_KEY, None)
content = message.content
if isinstance(content, str):
restored_content: str | list = original_content
elif isinstance(content, list):
restored_content = []
restored_text = False
for block in content:
is_string_text = isinstance(block, str)
is_mapping_text = isinstance(block, Mapping) and block.get("type") == "text" and isinstance(block.get("text"), str)
if not is_string_text and not is_mapping_text:
restored_content.append(block)
continue
if restored_text:
continue
if is_mapping_text:
restored_content.append({**block, "text": original_content})
else:
restored_content.append(original_content)
restored_text = True
if not restored_text:
restored_content.insert(0, {"type": "text", "text": original_content})
else:
restored_content = original_content
return message.model_copy(
update={
# Pydantic deep-copies the original model for ``deep=True``, but
# applies values supplied through ``update`` without copying them.
# Keep the persisted/UI copy fully isolated from the model-facing
# message, including nested image/file blocks and metadata.
"content": deepcopy(restored_content),
"additional_kwargs": deepcopy(additional_kwargs),
},
deep=True,
)
def is_real_user_message(message: object) -> bool:
"""Return whether ``message`` is a real user-authored HumanMessage.
Middleware-injected hidden HumanMessages and summarization markers should not
drive user-intent features such as slash-skill activation or MCP routing.
"""
if not isinstance(message, HumanMessage):
return False
if message.name == SUMMARY_MESSAGE_NAME:
return False
if message.additional_kwargs.get("hide_from_ui"):
return False
return True