fix(context): resolve context compress bug (#4065)

* fix(runtime): persist original human input outside model sanitization

* refactor(history): load thread messages by global event sequence

* fix(frontend): make summarization rescue a transient history bridge

* fix(frontend): old message not append tail

1. add identity anchor
2. add bridgeOrder

* fix(frontend): lint error fix

* fix: address review feedback and harden pagination coverage

- defer transient history ref writes until after render commit
- cover large middleware-only history scans
- verify infinite-query refetch recalculates page cursors
- document AI event types and anchor-weaving differences

* fix: harden message pagination and enrichment

- append unmatched live tails after canonical history
- warn and stop when pagination has_more lacks a cursor
- deep-copy restored UI messages to isolate model-facing content
- log invalid event sequence and non-advancing cursor errors
- pass user_id explicitly through event-store history queries
- cover middleware-only AI runs across memory, JSONL, and DB stores

* fix: address pagination review feedback

* fix(frontend): checkpoint has unknow redener content, optimize the anchor policy

* fix(frontend): unit test issue missed previously, remove the TanStack cache trimming

* fix(gateway): harden message history queries and provenance

- reject externally forged original_user_content metadata
- validate provenance metadata in upload and sanitization middleware
- make run lookups fail closed by default
- batch feedback queries by run ID
- align memory message filtering with persistent stores
This commit is contained in:
AnoobFeng 2026-07-14 10:43:13 +08:00 committed by GitHub
parent 81b3ed0188
commit 446fa03801
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
32 changed files with 2373 additions and 704 deletions

View File

@ -223,7 +223,7 @@ Lead-agent middlewares are assembled in strict order across three functions: the
**Shared runtime base** (`build_lead_runtime_middlewares`; subagents reuse most of this via `build_subagent_runtime_middlewares`):
1. **InputSanitizationMiddleware** - First, so it is the outermost `wrap_model_call` wrapper; every inner middleware (including LLM retries) sees sanitized messages
1. **InputSanitizationMiddleware** - First, so it is the outermost `wrap_model_call` wrapper; every inner middleware (including LLM retries) sees sanitized messages. `additional_kwargs.original_user_content` is server-owned provenance: Gateway strips caller-supplied values for non-internal run requests, trusted IM calls may carry the string they captured before adding transport/file context, and the middleware replaces any non-string value before wrapping. Uploads and sanitization retain first-writer-wins only for validated strings.
2. **ToolOutputBudgetMiddleware** - Caps tool output size (per app config) before it re-enters the model context
3. **ToolResultSanitizationMiddleware** - Neutralizes framework/injection tags (e.g. `<system-reminder>`) and boundary markers in *remote-content* tool results (`web_fetch`/`web_search`/`image_search`/`web_capture`) so attacker-controlled fetched pages cannot forge trusted framework context. Mirrors `InputSanitizationMiddleware`'s user-input guardrail for the other untrusted-content entry point; sits inner of `ToolOutputBudgetMiddleware` (neutralizes the raw output, then the budget truncates). Local tool output (bash/read_file) is left untouched. Scope is a name-based allowlist, so MCP remote-content tools registered under other names (e.g. `fetch_url`) are not yet covered — a metadata-tagging follow-up is tracked in the middleware source
4. **ThreadDataMiddleware** - Creates per-thread directories under the user's isolation scope (`backend/.deer-flow/users/{user_id}/threads/{thread_id}/user-data/{workspace,uploads,outputs}`); resolves `user_id` via `get_effective_user_id()` (falls back to `"default"` in no-auth mode)
@ -324,7 +324,7 @@ CORS is same-origin by default when requests enter through nginx on port 2026. S
| **Artifacts** (`/api/threads/{id}/artifacts`) | `GET /{path}` - serve artifacts; active content types (`text/html`, `application/xhtml+xml`, `image/svg+xml`) are always forced as download attachments to reduce XSS risk; `?download=true` still forces download for other file types |
| **Suggestions** (`/api/suggestions`) | `GET /config` - returns global suggestions config boolean; `POST /threads/{id}/suggestions` - generate follow-up questions; rich list/block model content is normalized and inline reasoning (`<think>...</think>`, including unclosed/truncated blocks from reasoning models like MiniMax-M3) is stripped before JSON parsing |
| **Input Polish** (`/api/input-polish`) | `POST /` - rewrite a composer draft before it is sent. This is a short authenticated `runs:create` LLM request using `input_polish` config; it does not create a LangGraph run, persist a message, or modify thread state. Shares the non-graph one-shot LLM path (`deerflow.utils.oneshot_llm.run_oneshot_llm`) with the suggestions route so model build + Langfuse metadata + invoke stay in one place; validates the same stripped view of the draft it sends to the model, and preserves literal `<think>` substrings in the rewrite (`strip_think_blocks(truncate_unclosed=False)`) |
| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block; `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest assistant answer; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/messages` - paginated messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET /../messages` - thread messages with feedback; `GET /../token-usage` - aggregate tokens |
| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block; `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest assistant answer; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/successful-regenerate filtering and page-run-scoped feedback enrichment; `GET /../token-usage` - aggregate tokens |
| **Feedback** (`/api/threads/{id}/runs/{rid}/feedback`) | `PUT /` - upsert feedback; `DELETE /` - delete user feedback; `POST /` - create feedback; `GET /` - list feedback; `GET /stats` - aggregate stats; `DELETE /{fid}` - delete specific |
| **Runs** (`/api/runs`) | `POST /stream` - stateless run + SSE; `POST /wait` - stateless run + block; `GET /{rid}/messages` - paginated messages by run_id `{data, has_more}` (cursor: `after_seq`/`before_seq`); `GET /{rid}/feedback` - list feedback by run_id |
| **GitHub Webhooks** (`/api/webhooks/github`) | `POST /` - receive GitHub App / repo webhook deliveries. Verifies `X-Hub-Signature-256` against `GITHUB_WEBHOOK_SECRET`; exempt from auth + CSRF because authenticity is enforced by HMAC. The route is fail-closed: mounted only when `GITHUB_WEBHOOK_SECRET` is set, or when explicit dev opt-in `DEER_FLOW_ALLOW_UNVERIFIED_GITHUB_WEBHOOKS=1` is set. Recognized events include `ping`, `issues`, `issue_comment`, `pull_request`, `pull_request_review`, and `pull_request_review_comment`; unknown events return 200 with `handled=false`. Fan-out runtime failures return 503 so GitHub retries; permanent/non-retryable conditions such as `channels.github.enabled: false`, unknown events, malformed payloads, or unavailable channel service return 200 with a skipped/handled response. |
@ -340,6 +340,7 @@ metadata only.
**RunManager / RunStore contract**:
- `RunManager.get()` is async; direct callers must `await` it.
- The history batch helpers `list_successful_regenerate_sources()` and `get_many_by_thread()` default to `user_id=AUTO`: they resolve the request user and fail closed when no user context exists. Migration/admin callers that intentionally need an unscoped read must pass `user_id=None` explicitly.
- When a persistent `RunStore` is configured, `get()` and `list_by_thread()` hydrate historical runs from the store. In-memory records win for the same `run_id` so task, abort, and stream-control state stays attached to active local runs.
- `cancel()` returns a :class:`~deerflow.runtime.CancelOutcome` enum: `cancelled` (local cancel), `taken_over` (non-owning worker claimed the run because the owner's lease expired — marks it as `error`), `lease_valid_elsewhere` (owner's lease is still alive — caller should return 409 + `Retry-After`), `not_active_locally` (heartbeat disabled, preserving the old 409 path), `not_cancellable` (terminal state), or `unknown` (not found in memory or store). `create_or_reject(..., multitask_strategy="interrupt"|"rollback")` persists interrupted status through `RunStore.update_status()`, matching normal `set_status()` transitions.
- Store-only hydrated runs are readable history. In multi-worker mode with heartbeat enabled, cancel on a store-only run can take over (mark `error`) when the owner's lease has expired past the grace window; otherwise it fails with 409 + `Retry-After`. In single-worker mode (heartbeat off), store-only runs still return 409.

View File

@ -13,6 +13,7 @@ from __future__ import annotations
import asyncio
import logging
from copy import deepcopy
from datetime import UTC, datetime
from typing import Any, Literal
@ -32,6 +33,7 @@ from deerflow.workspace_changes import get_workspace_changes_response
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/threads", tags=["runs"])
REGENERATE_HISTORY_SCAN_LIMIT = 200
THREAD_MESSAGE_PAGE_SCAN_BATCH = 201
def compute_run_durations(runs) -> dict[str, int]:
@ -91,6 +93,12 @@ class RegeneratePrepareResponse(BaseModel):
target_run_id: str
class ThreadMessagesPageResponse(BaseModel):
data: list[dict[str, Any]]
has_more: bool
next_before_seq: int | None = None
class RunResponse(BaseModel):
run_id: str
thread_id: str
@ -270,6 +278,10 @@ def _is_visible_ai_message(message: Any) -> bool:
return _message_type(message) == "ai" and not _is_hidden_or_control_message(message)
def _is_middleware_message_row(row: dict[str, Any]) -> bool:
return str((row.get("metadata") or {}).get("caller", "")).startswith("middleware:")
def _checkpoint_messages(checkpoint_tuple: Any) -> list[Any]:
checkpoint = getattr(checkpoint_tuple, "checkpoint", None) or {}
channel_values = checkpoint.get("channel_values", {}) if isinstance(checkpoint, dict) else {}
@ -746,6 +758,139 @@ async def list_thread_messages(
return messages
async def _scan_thread_message_page(
thread_id: str,
*,
limit: int,
before_seq: int | None,
request: Request,
user_id: str | None,
) -> tuple[list[dict[str, Any]], bool]:
"""Select the newest ``limit + 1`` page-eligible rows before a cursor."""
event_store = get_run_event_store(request)
run_mgr = get_run_manager(request)
superseded_run_ids = await run_mgr.list_successful_regenerate_sources(thread_id, user_id=user_id)
visible_desc: list[dict[str, Any]] = []
scan_before = before_seq
while len(visible_desc) < limit + 1:
raw = await event_store.list_messages(
thread_id,
limit=THREAD_MESSAGE_PAGE_SCAN_BATCH,
before_seq=scan_before,
user_id=user_id,
)
if not raw:
break
invalid_seq_rows = [row for row in raw if not isinstance(row.get("seq"), int)]
if invalid_seq_rows:
logger.error(
"Thread message scan found rows without sequence values: thread_id=%s scan_before=%s row_count=%d invalid_count=%d",
thread_id,
scan_before,
len(raw),
len(invalid_seq_rows),
)
raise RuntimeError("Run event message rows are missing sequence values")
for row in reversed(raw):
if _is_middleware_message_row(row) or row.get("run_id") in superseded_run_ids:
continue
visible_desc.append(row)
if len(visible_desc) == limit + 1:
break
raw_seqs = [row["seq"] for row in raw]
next_scan_before = min(raw_seqs)
if scan_before is not None and next_scan_before >= scan_before:
logger.error(
"Thread message scan cursor did not advance: thread_id=%s scan_before=%s next_scan_before=%s row_count=%d",
thread_id,
scan_before,
next_scan_before,
len(raw),
)
raise RuntimeError("Run event message scan did not advance its cursor")
scan_before = next_scan_before
if len(raw) < THREAD_MESSAGE_PAGE_SCAN_BATCH:
break
has_more = len(visible_desc) > limit
return list(reversed(visible_desc[:limit])), has_more
async def _enrich_thread_message_page(
thread_id: str,
rows: list[dict[str, Any]],
*,
request: Request,
user_id: str | None,
) -> list[dict[str, Any]]:
"""Attach run-scoped duration and feedback without mutating store rows."""
data = deepcopy(rows)
if not data:
return data
run_ids = {row["run_id"] for row in data if isinstance(row.get("run_id"), str)}
run_mgr = get_run_manager(request)
records = await run_mgr.get_many_by_thread(thread_id, run_ids, user_id=user_id)
run_durations = compute_run_durations(records.values())
event_store = get_run_event_store(request)
last_ai_seq_by_run = await event_store.get_last_visible_ai_seq_by_run(thread_id, run_ids, user_id=user_id)
feedback_map: dict[str, dict] = {}
feedback_run_ids = {run_id for row in data if isinstance((run_id := row.get("run_id")), str) and row.get("seq") == last_ai_seq_by_run.get(run_id)}
if feedback_run_ids:
feedback_repo = get_feedback_repo(request)
feedback_map = await feedback_repo.list_by_run_ids(thread_id, feedback_run_ids, user_id=user_id)
for row in data:
run_id = row.get("run_id")
row["feedback"] = None
if row.get("seq") == last_ai_seq_by_run.get(run_id):
feedback = feedback_map.get(run_id)
if feedback:
row["feedback"] = {
"feedback_id": feedback["feedback_id"],
"rating": feedback["rating"],
"comment": feedback.get("comment"),
}
content = row.get("content")
if isinstance(content, dict) and content.get("type") == "ai" and run_id in run_durations:
content.setdefault("additional_kwargs", {})["turn_duration"] = run_durations[run_id]
return data
@router.get("/{thread_id}/messages/page", response_model=ThreadMessagesPageResponse)
@require_permission("runs", "read", owner_check=True)
async def list_thread_messages_page(
thread_id: str,
request: Request,
limit: int = Query(default=50, ge=1, le=200),
before_seq: int | None = Query(default=None, ge=1),
) -> ThreadMessagesPageResponse:
"""Return a backward page ordered by the thread-global event sequence."""
if "after_seq" in request.query_params:
raise HTTPException(status_code=422, detail="after_seq is not supported by this backward-only endpoint")
user_id = await get_current_user(request)
rows, has_more = await _scan_thread_message_page(
thread_id,
limit=limit,
before_seq=before_seq,
request=request,
user_id=user_id,
)
data = await _enrich_thread_message_page(thread_id, rows, request=request, user_id=user_id)
return ThreadMessagesPageResponse(
data=data,
has_more=has_more,
next_before_seq=data[0]["seq"] if has_more else None,
)
@router.get("/{thread_id}/runs/{run_id}/messages")
@require_permission("runs", "read", owner_check=True)
async def list_run_messages(

View File

@ -46,6 +46,7 @@ from deerflow.runtime.goal import goal_thread_lock
from deerflow.runtime.runs.naming import resolve_root_run_name
from deerflow.runtime.secret_context import redact_config_secrets
from deerflow.runtime.user_context import reset_current_user, set_current_user
from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY
logger = logging.getLogger(__name__)
@ -117,7 +118,16 @@ def normalize_stream_modes(raw: list[str] | str | None) -> list[str]:
return raw if raw else ["values"]
def normalize_input(raw_input: dict[str, Any] | None) -> dict[str, Any]:
def _strip_external_message_metadata(message: Any) -> Any:
"""Remove server-owned metadata from an untrusted input message."""
if not isinstance(message, BaseMessage) or ORIGINAL_USER_CONTENT_KEY not in message.additional_kwargs:
return message
additional_kwargs = dict(message.additional_kwargs)
additional_kwargs.pop(ORIGINAL_USER_CONTENT_KEY, None)
return message.model_copy(update={"additional_kwargs": additional_kwargs})
def normalize_input(raw_input: dict[str, Any] | None, *, trusted_internal: bool = False) -> dict[str, Any]:
"""Convert LangGraph Platform input format to LangChain state dict.
Delegates dictmessage coercion to ``langchain_core.messages.utils.convert_to_messages``
@ -130,6 +140,11 @@ def normalize_input(raw_input: dict[str, Any] | None) -> dict[str, Any]:
role, etc.) raise ``HTTPException(400)`` with the offending index, instead
of bubbling up as a 500. The gateway is a system boundary, so per-entry
validation errors are the right shape for clients to retry against.
``original_user_content`` is server-owned provenance used to undo model-only
sanitization at persistence time. External callers cannot supply it; trusted
internal channel calls may preserve the value they captured before adding
transport or file context.
"""
if raw_input is None:
return {}
@ -149,6 +164,8 @@ def normalize_input(raw_input: dict[str, Any] | None) -> dict[str, Any]:
) from exc
else:
converted.append(msg)
if not trusted_internal:
converted = [_strip_external_message_metadata(message) for message in converted]
return {**raw_input, "messages": converted}
return raw_input
@ -657,11 +674,12 @@ async def start_run(
logger.warning("Failed to upsert thread_meta for %s (non-fatal)", sanitize_log_param(thread_id))
agent_factory = resolve_agent_factory(body.assistant_id)
is_internal_caller = getattr(getattr(request, "state", None), "auth_source", None) == AUTH_SOURCE_INTERNAL
command = getattr(body, "command", None)
if command and command.get("resume") is not None:
graph_input = Command(resume=command["resume"])
else:
graph_input = normalize_input(body.input)
graph_input = normalize_input(body.input, trusted_internal=is_internal_caller)
config = build_run_config(thread_id, body.config, body.metadata, assistant_id=body.assistant_id)
await apply_checkpoint_to_run_config(config, body=body, thread_id=thread_id, request=request)
@ -669,7 +687,6 @@ async def start_run(
# The ``context`` field is a custom extension for the langgraph-compat layer
# that carries agent configuration (model_name, thinking_enabled, etc.).
# Only agent-relevant keys are forwarded; unknown keys (e.g. thread_id) are ignored.
is_internal_caller = getattr(getattr(request, "state", None), "auth_source", None) == AUTH_SOURCE_INTERNAL
merge_run_context_overrides(config, getattr(body, "context", None), internal=is_internal_caller)
if not is_internal_caller:
# ``body.config`` is free-form and copied verbatim by

View File

@ -268,10 +268,19 @@ class InputSanitizationMiddleware(AgentMiddleware[AgentState]):
# Preserve the pre-sanitization user text so downstream consumers that
# must see the genuine input (slash skill activation, regenerate) can
# recover it after the BEGIN/END wrapping. setdefault keeps an existing
# value (e.g. set by UploadsMiddleware or an IM channel) authoritative.
# recover it after the BEGIN/END wrapping. Keep a valid value set by
# UploadsMiddleware or an IM channel, but repair malformed metadata so
# persistence never falls back to the wrapped model-facing content.
preserved_kwargs = dict(msg.additional_kwargs or {})
preserved_kwargs.setdefault(ORIGINAL_USER_CONTENT_KEY, message_content_to_text(content))
original_user_content = preserved_kwargs.get(ORIGINAL_USER_CONTENT_KEY)
if not isinstance(original_user_content, str):
if ORIGINAL_USER_CONTENT_KEY in preserved_kwargs:
logger.warning(
"InputSanitizationMiddleware replaced non-string %s metadata: type=%s",
ORIGINAL_USER_CONTENT_KEY,
type(original_user_content).__name__,
)
preserved_kwargs[ORIGINAL_USER_CONTENT_KEY] = message_content_to_text(content)
messages[i] = HumanMessage(
content=new_content,
id=msg.id,

View File

@ -396,7 +396,15 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
# Extract original content - handle both string and list formats
original_content = last_message.content
additional_kwargs = dict(last_message.additional_kwargs or {})
additional_kwargs.setdefault(ORIGINAL_USER_CONTENT_KEY, message_content_to_text(original_content))
original_user_content = additional_kwargs.get(ORIGINAL_USER_CONTENT_KEY)
if not isinstance(original_user_content, str):
if ORIGINAL_USER_CONTENT_KEY in additional_kwargs:
logger.warning(
"UploadsMiddleware replaced non-string %s metadata: type=%s",
ORIGINAL_USER_CONTENT_KEY,
type(original_user_content).__name__,
)
additional_kwargs[ORIGINAL_USER_CONTENT_KEY] = message_content_to_text(original_content)
if isinstance(original_content, str):
# Simple case: string content, just prepend files message
updated_content = f"{files_message}\n\n{original_content}"

View File

@ -202,6 +202,27 @@ class FeedbackRepository:
result = await session.execute(stmt)
return {row.run_id: self._row_to_dict(row) for row in result.scalars()}
async def list_by_run_ids(
self,
thread_id: str,
run_ids: set[str],
*,
user_id: str | None | _AutoSentinel = AUTO,
) -> dict[str, dict]:
"""Return feedback for only the selected runs in one thread."""
if not run_ids:
return {}
resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.list_by_run_ids")
stmt = select(FeedbackRow).where(
FeedbackRow.thread_id == thread_id,
FeedbackRow.run_id.in_(run_ids),
)
if resolved_user_id is not None:
stmt = stmt.where(FeedbackRow.user_id == resolved_user_id)
async with self._sf() as session:
result = await session.execute(stmt)
return {row.run_id: self._row_to_dict(row) for row in result.scalars()}
async def aggregate_by_run(self, thread_id: str, run_id: str) -> dict:
"""Aggregate feedback stats for a run using database-side counting."""
stmt = select(

View File

@ -166,6 +166,43 @@ class RunRepository(RunStore):
result = await session.execute(stmt)
return [self._row_to_dict(r) for r in result.scalars()]
async def list_successful_regenerate_sources(
self,
thread_id,
*,
user_id: str | None | _AutoSentinel = AUTO,
):
resolved_user_id = resolve_user_id(user_id, method_name="RunRepository.list_successful_regenerate_sources")
source = RunRow.metadata_json["regenerate_from_run_id"].as_string()
stmt = select(source).where(
RunRow.thread_id == thread_id,
RunRow.status == "success",
source.is_not(None),
source != "",
)
if resolved_user_id is not None:
stmt = stmt.where(RunRow.user_id == resolved_user_id)
async with self._sf() as session:
result = await session.execute(stmt)
return {value for value in result.scalars() if isinstance(value, str) and value}
async def get_many_by_thread(
self,
thread_id,
run_ids,
*,
user_id: str | None | _AutoSentinel = AUTO,
):
if not run_ids:
return {}
resolved_user_id = resolve_user_id(user_id, method_name="RunRepository.get_many_by_thread")
stmt = select(RunRow).where(RunRow.thread_id == thread_id, RunRow.run_id.in_(run_ids))
if resolved_user_id is not None:
stmt = stmt.where(RunRow.user_id == resolved_user_id)
async with self._sf() as session:
result = await session.execute(stmt)
return {row.run_id: self._row_to_dict(row) for row in result.scalars()}
async def update_status(self, run_id, status, *, error=None) -> bool:
values: dict[str, Any] = {"status": status, "updated_at": datetime.now(UTC)}
if error is not None:

View File

@ -13,6 +13,8 @@ from __future__ import annotations
import abc
from deerflow.runtime.user_context import AUTO, _AutoSentinel
class RunEventStore(abc.ABC):
"""Run event stream storage interface.
@ -55,6 +57,7 @@ class RunEventStore(abc.ABC):
limit: int = 50,
before_seq: int | None = None,
after_seq: int | None = None,
user_id: str | None | _AutoSentinel = AUTO,
) -> list[dict]:
"""Return displayable messages (category=message) for a thread, ordered by seq ascending.
@ -62,6 +65,9 @@ class RunEventStore(abc.ABC):
- before_seq: return the last ``limit`` records with seq < before_seq (ascending)
- after_seq: return the first ``limit`` records with seq > after_seq (ascending)
- neither: return the latest ``limit`` records (ascending)
``user_id`` may be passed explicitly by request-independent callers;
user-scoped backends must apply it according to their isolation model.
"""
@abc.abstractmethod
@ -102,6 +108,20 @@ class RunEventStore(abc.ABC):
- neither: return the latest ``limit`` records (ascending)
"""
@abc.abstractmethod
async def get_last_visible_ai_seq_by_run(
self,
thread_id: str,
run_ids: set[str],
*,
user_id: str | None | _AutoSentinel = AUTO,
) -> dict[str, int]:
"""Return each run's last non-middleware AI message sequence.
``user_id`` follows the same explicit-caller semantics as
:meth:`list_messages`.
"""
@abc.abstractmethod
async def count_messages(self, thread_id: str) -> int:
"""Count displayable messages (category=message) in a thread."""

View File

@ -292,6 +292,36 @@ class DbRunEventStore(RunEventStore):
rows = list(result.scalars())
return [self._row_to_dict(r) for r in reversed(rows)]
async def get_last_visible_ai_seq_by_run(
self,
thread_id,
run_ids,
*,
user_id: str | None | _AutoSentinel = AUTO,
):
if not run_ids:
return {}
resolved_user_id = resolve_user_id(user_id, method_name="DbRunEventStore.get_last_visible_ai_seq_by_run")
caller = RunEventRow.event_metadata["caller"].as_string()
# RunJournal canonically persists AI message rows as
# ``llm.ai.response``; ``ai_message`` remains for legacy compatibility.
stmt = (
select(RunEventRow.run_id, func.max(RunEventRow.seq))
.where(
RunEventRow.thread_id == thread_id,
RunEventRow.run_id.in_(run_ids),
RunEventRow.category == "message",
RunEventRow.event_type.in_(("llm.ai.response", "ai_message")),
~func.coalesce(caller, "").like("middleware:%"),
)
.group_by(RunEventRow.run_id)
)
if resolved_user_id is not None:
stmt = stmt.where(RunEventRow.user_id == resolved_user_id)
async with self._sf() as session:
result = await session.execute(stmt)
return {run_id: seq for run_id, seq in result if isinstance(seq, int)}
async def count_messages(
self,
thread_id,

View File

@ -31,6 +31,7 @@ from pathlib import Path
from typing import Any
from deerflow.runtime.events.store.base import RunEventStore
from deerflow.runtime.user_context import AUTO, _AutoSentinel
logger = logging.getLogger(__name__)
@ -206,7 +207,7 @@ class JsonlRunEventStore(RunEventStore):
with open(path, "a", encoding="utf-8") as f:
f.write(lines)
async def list_messages(self, thread_id, *, limit=50, before_seq=None, after_seq=None):
async def list_messages(self, thread_id, *, limit=50, before_seq=None, after_seq=None, user_id: str | None | _AutoSentinel = AUTO):
all_events = await asyncio.to_thread(self._read_thread_events, thread_id)
messages = [e for e in all_events if e.get("category") == "message"]
@ -241,6 +242,19 @@ class JsonlRunEventStore(RunEventStore):
else:
return filtered[-limit:] if len(filtered) > limit else filtered
async def get_last_visible_ai_seq_by_run(self, thread_id, run_ids, *, user_id: str | None | _AutoSentinel = AUTO):
def _scan() -> dict[str, int]:
result: dict[str, int] = {}
for run_id in run_ids:
for event in reversed(self._read_run_events(thread_id, run_id)):
caller = str((event.get("metadata") or {}).get("caller", ""))
if event.get("category") == "message" and event.get("event_type") in {"llm.ai.response", "ai_message"} and not caller.startswith("middleware:"):
result[run_id] = event["seq"]
break
return result
return await asyncio.to_thread(_scan)
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")

View File

@ -10,6 +10,7 @@ import bisect
from datetime import UTC, datetime
from deerflow.runtime.events.store.base import RunEventStore
from deerflow.runtime.user_context import AUTO, _AutoSentinel
class MemoryRunEventStore(RunEventStore):
@ -92,7 +93,7 @@ class MemoryRunEventStore(RunEventStore):
results.append(record)
return results
async def list_messages(self, thread_id, *, limit=50, before_seq=None, after_seq=None):
async def list_messages(self, thread_id, *, limit=50, before_seq=None, after_seq=None, user_id: str | None | _AutoSentinel = AUTO):
# ``messages`` is messages-only and seq-sorted, so the seq window is a
# contiguous slice located with bisect (O(log m)) rather than a full scan.
messages = self._messages.get(thread_id, [])
@ -136,6 +137,17 @@ class MemoryRunEventStore(RunEventStore):
return window[:limit]
return window[-limit:]
async def get_last_visible_ai_seq_by_run(self, thread_id, run_ids, *, user_id: str | None | _AutoSentinel = AUTO):
result: dict[str, int] = {}
messages_by_run = self._messages_by_run.get(thread_id, {})
for run_id in run_ids:
for event in reversed(messages_by_run.get(run_id, [])):
caller = str((event.get("metadata") or {}).get("caller", ""))
if event.get("category") == "message" and event.get("event_type") in {"llm.ai.response", "ai_message"} and not caller.startswith("middleware:"):
result[run_id] = event["seq"]
break
return result
async def count_messages(self, thread_id):
return len(self._messages.get(thread_id, []))

View File

@ -30,7 +30,7 @@ from langchain_core.messages import AIMessage, AnyMessage, BaseMessage, HumanMes
from langgraph.types import Command
from deerflow.agents.human_input import read_human_input_response
from deerflow.utils.messages import message_to_text
from deerflow.utils.messages import message_to_text, restore_original_human_message
if TYPE_CHECKING:
from deerflow.runtime.events.store.base import RunEventStore
@ -222,14 +222,15 @@ class RunJournal(BaseCallbackHandler):
for batch in reversed(messages):
for m in reversed(batch):
if _should_persist_human_input_message(m):
self.set_first_human_message(m.text)
persisted_message = restore_original_human_message(m)
self.set_first_human_message(self._message_text(persisted_message))
self._put(
event_type="llm.human.input",
category="message",
content=m.model_dump(),
content=persisted_message.model_dump(),
metadata={"caller": caller},
)
self._record_message_summary(m, caller=caller)
self._record_message_summary(persisted_message, caller=caller)
break
if self._first_human_msg:
break

View File

@ -15,6 +15,7 @@ from typing import TYPE_CHECKING, Any
from sqlalchemy.exc import IntegrityError as SAIntegrityError
from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id
from deerflow.utils.time import is_lease_expired
from deerflow.utils.time import now_iso as _now_iso
@ -590,6 +591,71 @@ class RunManager:
logger.warning("Failed to map store row for run %s", run_id, exc_info=True)
return sorted(records_by_id.values(), key=lambda record: record.created_at, reverse=True)[:limit]
async def list_successful_regenerate_sources(
self,
thread_id: str,
*,
user_id: str | None | _AutoSentinel = AUTO,
) -> set[str]:
"""Return all source runs superseded by successful regenerations.
Unlike :meth:`list_by_thread`, this query is intentionally unbounded.
Current-process records override matching persisted status: a latest
in-memory failure must not inherit an older successful store snapshot.
Store failures propagate because supersession filtering is required for
correct pagination.
"""
resolved_user_id = resolve_user_id(user_id, method_name="RunManager.list_successful_regenerate_sources")
async with self._lock:
memory_records = [record for record in self._thread_records_locked(thread_id) if resolved_user_id is None or record.user_id == resolved_user_id]
sources = set(await self._store.list_successful_regenerate_sources(thread_id, user_id=resolved_user_id)) if self._store is not None else set()
# _thread_records_locked preserves the insertion order of the thread
# index. Applying records oldest-to-newest makes the latest in-memory
# regeneration attempt authoritative when several attempts reference
# the same source run (for example, a failed retry after a success).
for record in memory_records:
source = record.metadata.get("regenerate_from_run_id")
if not isinstance(source, str) or not source:
continue
sources.discard(source)
if record.status == RunStatus.success:
sources.add(source)
return sources
async def get_many_by_thread(
self,
thread_id: str,
run_ids: set[str],
*,
user_id: str | None | _AutoSentinel = AUTO,
) -> dict[str, RunRecord]:
"""Batch-load selected thread runs with in-memory records preferred."""
if not run_ids:
return {}
resolved_user_id = resolve_user_id(user_id, method_name="RunManager.get_many_by_thread")
async with self._lock:
records_by_id = {record.run_id: record for record in self._thread_records_locked(thread_id) if record.run_id in run_ids and (resolved_user_id is None or record.user_id == resolved_user_id)}
if self._store is None:
return records_by_id
remaining = run_ids - records_by_id.keys()
if not remaining:
return records_by_id
try:
rows = await self._store.get_many_by_thread(thread_id, set(remaining), user_id=resolved_user_id)
except Exception:
logger.warning("Failed to batch-hydrate runs for thread %s", thread_id, exc_info=True)
return records_by_id
for run_id, row in rows.items():
if run_id in records_by_id:
continue
try:
records_by_id[run_id] = self._record_from_store(row)
except Exception:
logger.warning("Failed to map store row for run %s", run_id, exc_info=True)
return records_by_id
async def set_status(self, run_id: str, status: RunStatus, *, error: str | None = None) -> None:
"""Transition a run to a new status."""
async with self._lock:

View File

@ -54,6 +54,29 @@ class RunStore(abc.ABC):
) -> list[dict[str, Any]]:
pass
async def list_successful_regenerate_sources(
self,
thread_id: str,
*,
user_id: str | None = None,
) -> set[str]:
"""Return source run IDs superseded by successful regenerations.
Implementations must inspect the complete thread and must not apply the
normal bounded run-list limit.
"""
raise NotImplementedError
async def get_many_by_thread(
self,
thread_id: str,
run_ids: set[str],
*,
user_id: str | None = None,
) -> dict[str, dict[str, Any]]:
"""Batch-load selected runs belonging to one thread."""
raise NotImplementedError
@abc.abstractmethod
async def update_status(
self,

View File

@ -87,6 +87,24 @@ class MemoryRunStore(RunStore):
results.sort(key=lambda r: r["created_at"], reverse=True)
return results[:limit]
async def list_successful_regenerate_sources(self, thread_id, *, user_id=None):
run_ids = self._runs_by_thread.get(thread_id) or ()
sources: set[str] = set()
for run_id in run_ids:
run = self._runs.get(run_id)
if run is None or run.get("status") != "success":
continue
if user_id is not None and run.get("user_id") != user_id:
continue
source = (run.get("metadata") or {}).get("regenerate_from_run_id")
if isinstance(source, str) and source:
sources.add(source)
return sources
async def get_many_by_thread(self, thread_id, run_ids, *, user_id=None):
thread_run_ids = self._runs_by_thread.get(thread_id) or ()
return {run_id: run for run_id in thread_run_ids if run_id in run_ids and (run := self._runs.get(run_id)) is not None and (user_id is None or run.get("user_id") == user_id)}
async def update_status(self, run_id, status, *, error=None):
run = self._runs.get(run_id)
if run is None:

View File

@ -1,6 +1,7 @@
from __future__ import annotations
from collections.abc import Mapping
from copy import deepcopy
from typing import Any
from langchain_core.messages import HumanMessage
@ -77,6 +78,63 @@ def get_original_user_content_text(content: Any, additional_kwargs: Mapping[str,
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.

View File

@ -10,7 +10,8 @@ so any blocking IO here stalls the event loop on the hot path.
``tests/test_jsonl_event_store_async_io.py`` that covers ``put`` only. This
anchor complements it by driving the **full** async surface (``put``,
``put_batch``, ``list_messages``, ``list_events``, ``list_messages_by_run``,
``count_messages``, ``delete_by_run``, ``delete_by_thread``) under the strict
``get_last_visible_ai_seq_by_run``, ``count_messages``, ``delete_by_run``,
``delete_by_thread``) under the strict
Blockbuster runtime gate, so any blocking IO reintroduced on the event loop in
any of these methods not just removal of a specific ``to_thread`` call
fails CI.
@ -55,6 +56,7 @@ async def test_jsonl_run_event_store_async_api_does_not_block_event_loop(tmp_pat
assert isinstance(await store.list_events("t1", "r1"), list)
assert isinstance(await store.list_events("t1", "r1", event_types=["message"]), list)
assert isinstance(await store.list_messages_by_run("t1", "r2"), list)
assert isinstance(await store.get_last_visible_ai_seq_by_run("t1", {"r1", "r2"}, user_id="user-1"), dict)
assert await store.count_messages("t1") >= 1
# deletes: delete_by_run (single file) then delete_by_thread (remaining)

View File

@ -228,6 +228,29 @@ class TestFeedbackRepository:
assert grouped == {}
await _cleanup()
@pytest.mark.anyio
async def test_list_by_run_ids_is_thread_and_owner_scoped(self, tmp_path):
repo = await _make_feedback_repo(tmp_path)
await repo.upsert(run_id="r1", thread_id="t1", rating=1, user_id="u1")
await repo.upsert(run_id="r2", thread_id="t1", rating=-1, user_id="u1")
await repo.upsert(run_id="r3", thread_id="t1", rating=1, user_id="u1")
await repo.upsert(run_id="r1", thread_id="t1", rating=-1, user_id="u2")
await repo.upsert(run_id="r2", thread_id="t2", rating=1, user_id="u1")
grouped = await repo.list_by_run_ids("t1", {"r1", "r2"}, user_id="u1")
assert set(grouped) == {"r1", "r2"}
assert grouped["r1"]["rating"] == 1
assert grouped["r2"]["rating"] == -1
await _cleanup()
@pytest.mark.anyio
async def test_list_by_run_ids_empty_skips_query(self, tmp_path):
repo = await _make_feedback_repo(tmp_path)
assert await repo.list_by_run_ids("t1", set(), user_id="u1") == {}
await _cleanup()
# -- Follow-up association --

View File

@ -132,6 +132,54 @@ def test_normalize_input_preserves_additional_kwargs_and_id():
assert msg.additional_kwargs == {"files": files, "custom": "keep-me"}
@pytest.mark.parametrize(
"forged_original",
["spoofed audit text", [{"type": "text", "text": "spoofed audit text"}]],
)
def test_normalize_input_strips_external_original_user_content(forged_original):
from app.gateway.services import normalize_input
from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY
result = normalize_input(
{
"messages": [
{
"role": "user",
"content": "actual user input",
"additional_kwargs": {
ORIGINAL_USER_CONTENT_KEY: forged_original,
"custom": "keep-me",
},
}
]
}
)
assert result["messages"][0].additional_kwargs == {"custom": "keep-me"}
def test_normalize_input_preserves_trusted_internal_original_user_content():
from app.gateway.services import normalize_input
from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY
result = normalize_input(
{
"messages": [
{
"role": "user",
"content": "uploaded file context\n\nactual user input",
"additional_kwargs": {
ORIGINAL_USER_CONTENT_KEY: "actual user input",
},
}
]
},
trusted_internal=True,
)
assert result["messages"][0].additional_kwargs[ORIGINAL_USER_CONTENT_KEY] == "actual user input"
def test_normalize_input_preserves_human_input_response_metadata():
from langchain_core.messages import HumanMessage
@ -835,7 +883,7 @@ def test_inject_authenticated_user_context_strips_internal_spoofed_attribution()
assert "oauth_id" not in config["context"]
async def _capture_start_run_graph_input(body):
async def _capture_start_run_graph_input(body, *, auth_source=None):
from types import SimpleNamespace
from unittest.mock import patch
@ -859,7 +907,7 @@ async def _capture_start_run_graph_input(body):
)
request = SimpleNamespace(
headers={},
state=SimpleNamespace(),
state=SimpleNamespace(auth_source=auth_source),
app=SimpleNamespace(state=state),
)
captured: dict[str, object] = {}
@ -918,6 +966,60 @@ def test_start_run_uses_normalized_input_without_command(_stub_app_config):
assert graph_input["messages"][0].content == "hi"
def test_start_run_strips_external_original_user_content(_stub_app_config):
import asyncio
from app.gateway.routers.thread_runs import RunCreateRequest
from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY
graph_input = asyncio.run(
_capture_start_run_graph_input(
RunCreateRequest(
input={
"messages": [
{
"role": "human",
"content": "actual user input",
"additional_kwargs": {ORIGINAL_USER_CONTENT_KEY: "spoofed audit text"},
}
]
},
command=None,
)
)
)
assert ORIGINAL_USER_CONTENT_KEY not in graph_input["messages"][0].additional_kwargs
def test_start_run_preserves_internal_original_user_content(_stub_app_config):
import asyncio
from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL
from app.gateway.routers.thread_runs import RunCreateRequest
from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY
graph_input = asyncio.run(
_capture_start_run_graph_input(
RunCreateRequest(
input={
"messages": [
{
"role": "human",
"content": "uploaded file context\n\nactual user input",
"additional_kwargs": {ORIGINAL_USER_CONTENT_KEY: "actual user input"},
}
]
},
command=None,
),
auth_source=AUTH_SOURCE_INTERNAL,
)
)
assert graph_input["messages"][0].additional_kwargs[ORIGINAL_USER_CONTENT_KEY] == "actual user input"
def test_start_run_uses_internal_owner_header_for_persistence(_stub_app_config):
import asyncio
from types import SimpleNamespace

View File

@ -0,0 +1,290 @@
"""Cross-store contracts used by thread-global history pagination."""
from __future__ import annotations
import pytest
from deerflow.runtime import RunManager, RunStatus
from deerflow.runtime.events.store.memory import MemoryRunEventStore
from deerflow.runtime.runs.store.memory import MemoryRunStore
async def _seed_ai_messages(store):
await store.put(
thread_id="t1",
run_id="r1",
event_type="llm.ai.response",
category="message",
content={"type": "ai", "content": "first"},
metadata={"caller": "lead_agent"},
)
await store.put(
thread_id="t1",
run_id="r1",
event_type="llm.ai.response",
category="message",
content={"type": "ai", "content": "middleware"},
metadata={"caller": "middleware:title"},
)
last = await store.put(
thread_id="t1",
run_id="r1",
event_type="llm.ai.response",
category="message",
content={"type": "ai", "content": "last"},
metadata={"caller": "lead_agent"},
)
other = await store.put(
thread_id="t1",
run_id="r2",
event_type="llm.ai.response",
category="message",
content={"type": "ai", "content": "other"},
metadata={"caller": "lead_agent"},
)
await store.put(
thread_id="t1",
run_id="r_mw",
event_type="llm.ai.response",
category="message",
content={"type": "ai", "content": "middleware only"},
metadata={"caller": "middleware:title"},
)
return {"r1": last["seq"], "r2": other["seq"]}
@pytest.mark.anyio
async def test_memory_event_store_returns_global_last_non_middleware_ai_seq():
store = MemoryRunEventStore()
expected = await _seed_ai_messages(store)
result = await store.get_last_visible_ai_seq_by_run("t1", {"r1", "r2", "r_mw", "missing"})
assert result == expected
assert "r_mw" not in result
@pytest.mark.anyio
async def test_memory_event_store_defensively_rechecks_message_category():
store = MemoryRunEventStore()
expected = await store.put(
thread_id="t1",
run_id="r1",
event_type="llm.ai.response",
category="message",
content={"type": "ai", "content": "visible"},
metadata={"caller": "lead_agent"},
)
mutated = await store.put(
thread_id="t1",
run_id="r1",
event_type="llm.ai.response",
category="message",
content={"type": "ai", "content": "no longer a message"},
metadata={"caller": "lead_agent"},
)
# Memory projections intentionally share their row dictionaries. Recheck
# category at read time so an accidental mutation cannot violate the same
# contract that the DB and JSONL stores enforce explicitly.
mutated["category"] = "trace"
assert await store.get_last_visible_ai_seq_by_run("t1", {"r1"}) == {"r1": expected["seq"]}
@pytest.mark.anyio
async def test_jsonl_event_store_returns_global_last_non_middleware_ai_seq(tmp_path):
from deerflow.runtime.events.store.jsonl import JsonlRunEventStore
store = JsonlRunEventStore(base_dir=tmp_path)
expected = await _seed_ai_messages(store)
result = await store.get_last_visible_ai_seq_by_run("t1", {"r1", "r2", "r_mw", "missing"})
assert result == expected
assert "r_mw" not in result
@pytest.mark.anyio
async def test_db_event_store_returns_global_last_non_middleware_ai_seq(tmp_path):
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
from deerflow.runtime.events.store.db import DbRunEventStore
await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'events.db'}", sqlite_dir=str(tmp_path))
try:
store = DbRunEventStore(get_session_factory())
expected = await _seed_ai_messages(store)
result = await store.get_last_visible_ai_seq_by_run("t1", {"r1", "r2", "r_mw", "missing"})
assert result == expected
assert "r_mw" not in result
finally:
await close_engine()
@pytest.mark.anyio
async def test_memory_run_store_supersession_is_unbounded_and_owner_scoped():
store = MemoryRunStore()
for index in range(105):
await store.put(f"normal-{index}", thread_id="t1", user_id="alice", status="success")
await store.put(
"regen-success",
thread_id="t1",
user_id="alice",
status="success",
metadata={"regenerate_from_run_id": "source-success"},
)
await store.put(
"regen-failed",
thread_id="t1",
user_id="alice",
status="error",
metadata={"regenerate_from_run_id": "source-failed"},
)
await store.put(
"regen-bob",
thread_id="t1",
user_id="bob",
status="success",
metadata={"regenerate_from_run_id": "source-bob"},
)
assert await store.list_successful_regenerate_sources("t1", user_id="alice") == {"source-success"}
@pytest.mark.anyio
async def test_run_repository_batch_queries_are_unbounded_and_owner_scoped(tmp_path):
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
from deerflow.persistence.run import RunRepository
await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'runs.db'}", sqlite_dir=str(tmp_path))
try:
repo = RunRepository(get_session_factory())
for index in range(105):
await repo.put(f"normal-{index}", thread_id="t1", user_id="alice", status="success")
await repo.put(
"regen-a",
thread_id="t1",
user_id="alice",
status="success",
metadata={"regenerate_from_run_id": "source-a"},
)
await repo.put(
"regen-b",
thread_id="t1",
user_id="bob",
status="success",
metadata={"regenerate_from_run_id": "source-b"},
)
assert await repo.list_successful_regenerate_sources("t1", user_id="alice") == {"source-a"}
rows = await repo.get_many_by_thread("t1", {"normal-0", "regen-a", "regen-b"}, user_id="alice")
assert set(rows) == {"normal-0", "regen-a"}
finally:
await close_engine()
@pytest.mark.anyio
async def test_run_manager_prefers_latest_in_memory_regenerate_status():
store = MemoryRunStore()
await store.put(
"regen",
thread_id="t1",
status="success",
metadata={"regenerate_from_run_id": "source"},
)
manager = RunManager(store=store)
# Simulate the same logical run being newer in memory than its persisted
# successful snapshot.
persisted = await manager.get("regen")
assert persisted is not None
manager._runs["regen"] = persisted
manager._index_run_locked(persisted)
persisted.status = RunStatus.error
assert await manager.list_successful_regenerate_sources("t1", user_id=None) == set()
@pytest.mark.anyio
async def test_run_manager_uses_latest_attempt_for_shared_regenerate_source():
manager = RunManager()
older = await manager.create(
"t1",
metadata={"regenerate_from_run_id": "source"},
)
older.status = RunStatus.success
newer = await manager.create(
"t1",
metadata={"regenerate_from_run_id": "source"},
)
newer.status = RunStatus.error
assert await manager.list_successful_regenerate_sources("t1", user_id=None) == set()
@pytest.mark.anyio
async def test_run_manager_batch_history_methods_default_to_current_user():
from types import SimpleNamespace
from deerflow.runtime.user_context import reset_current_user, set_current_user
store = MemoryRunStore()
await store.put(
"regen-alice",
thread_id="shared-thread",
user_id="alice",
status="success",
metadata={"regenerate_from_run_id": "source-alice"},
)
await store.put(
"regen-bob",
thread_id="shared-thread",
user_id="bob",
status="success",
metadata={"regenerate_from_run_id": "source-bob"},
)
manager = RunManager(store=store)
token = set_current_user(SimpleNamespace(id="alice"))
try:
sources = await manager.list_successful_regenerate_sources("shared-thread")
records = await manager.get_many_by_thread("shared-thread", {"regen-alice", "regen-bob"})
finally:
reset_current_user(token)
assert sources == {"source-alice"}
assert set(records) == {"regen-alice"}
@pytest.mark.anyio
async def test_run_manager_batch_history_methods_fail_closed_without_user_context():
from deerflow.runtime import user_context
manager = RunManager(store=MemoryRunStore())
token = user_context._current_user.set(None)
try:
with pytest.raises(RuntimeError, match="user_id=AUTO"):
await manager.list_successful_regenerate_sources("t1")
with pytest.raises(RuntimeError, match="user_id=AUTO"):
await manager.get_many_by_thread("t1", {"run-1"})
finally:
user_context._current_user.reset(token)
@pytest.mark.anyio
async def test_run_manager_batch_history_methods_allow_explicit_unscoped_access():
store = MemoryRunStore()
await store.put(
"regen-alice",
thread_id="shared-thread",
user_id="alice",
status="success",
metadata={"regenerate_from_run_id": "source-alice"},
)
await store.put(
"regen-bob",
thread_id="shared-thread",
user_id="bob",
status="success",
metadata={"regenerate_from_run_id": "source-bob"},
)
manager = RunManager(store=store)
sources = await manager.list_successful_regenerate_sources("shared-thread", user_id=None)
records = await manager.get_many_by_thread("shared-thread", {"regen-alice", "regen-bob"}, user_id=None)
assert sources == {"source-alice", "source-bob"}
assert set(records) == {"regen-alice", "regen-bob"}

View File

@ -19,6 +19,7 @@ from deerflow.agents.middlewares.input_sanitization_middleware import (
_check_user_content,
_is_genuine_user_message,
)
from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY
def _make_middleware() -> InputSanitizationMiddleware:
@ -314,6 +315,40 @@ class TestWrapModelCallCleanInput:
assert _USER_INPUT_BEGIN in result_msgs[2].content
assert "Second" in result_msgs[2].content
def test_preserves_trusted_string_original_user_content(self):
mw = _make_middleware()
request = _make_request(
[
HumanMessage(
content="uploaded file context\n\nactual user input",
additional_kwargs={ORIGINAL_USER_CONTENT_KEY: "actual user input"},
)
]
)
captured = []
mw.wrap_model_call(request, lambda req: captured.append(req) or "ok")
assert captured[0].messages[0].additional_kwargs[ORIGINAL_USER_CONTENT_KEY] == "actual user input"
def test_replaces_non_string_original_user_content_before_wrapping(self):
mw = _make_middleware()
malformed_original = [{"type": "text", "text": "spoofed audit text"}]
request = _make_request(
[
HumanMessage(
content="actual user input",
additional_kwargs={ORIGINAL_USER_CONTENT_KEY: malformed_original},
)
]
)
captured = []
mw.wrap_model_call(request, lambda req: captured.append(req) or "ok")
assert captured[0].messages[0].additional_kwargs[ORIGINAL_USER_CONTENT_KEY] == "actual user input"
assert request.messages[0].additional_kwargs[ORIGINAL_USER_CONTENT_KEY] == malformed_original
# ---------------------------------------------------------------------------
# wrap_model_call — blocked input (escaped, not rejected)

View File

@ -8,9 +8,11 @@ from unittest.mock import MagicMock
from uuid import uuid4
import pytest
from langchain_core.messages import HumanMessage
from deerflow.runtime.events.store.memory import MemoryRunEventStore
from deerflow.runtime.journal import RunJournal
from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY
@pytest.fixture
@ -57,6 +59,28 @@ def _make_llm_response(content="Hello", usage=None, tool_calls=None, additional_
class TestLlmCallbacks:
@pytest.mark.anyio
async def test_on_chat_model_start_persists_original_user_input_without_mutating_model_message(self, journal_setup):
j, store = journal_setup
wrapped_content = "--- BEGIN USER INPUT ---\nShow revenue\n--- END USER INPUT ---"
model_message = HumanMessage(
content=wrapped_content,
id="human-1",
additional_kwargs={ORIGINAL_USER_CONTENT_KEY: "Show revenue", "channel": "web"},
)
j.on_chat_model_start({}, [[model_message]], run_id=uuid4(), tags=["lead_agent"])
await j.flush()
assert j._first_human_msg == "Show revenue"
events = await store.list_events("t1", "r1")
human_event = next(event for event in events if event["event_type"] == "llm.human.input")
assert human_event["content"]["content"] == "Show revenue"
assert human_event["content"]["id"] == "human-1"
assert human_event["content"]["additional_kwargs"] == {"channel": "web"}
assert model_message.content == wrapped_content
assert model_message.additional_kwargs[ORIGINAL_USER_CONTENT_KEY] == "Show revenue"
@pytest.mark.anyio
async def test_on_llm_end_produces_trace_event(self, journal_setup):
j, store = journal_setup

View File

@ -0,0 +1,322 @@
"""Tests for thread-global message history pagination."""
from __future__ import annotations
import asyncio
import logging
from unittest.mock import AsyncMock, MagicMock
import pytest
from _router_auth_helpers import make_authed_test_app
from fastapi.testclient import TestClient
from app.gateway.routers import thread_runs
from deerflow.runtime import RunRecord
from deerflow.runtime.events.store.memory import MemoryRunEventStore
def _make_app(event_store: MemoryRunEventStore, *, superseded: set[str] | None = None, records=None, feedback=None):
app = make_authed_test_app()
app.include_router(thread_runs.router)
app.state.run_event_store = event_store
run_manager = AsyncMock()
run_manager.list_successful_regenerate_sources.return_value = superseded or set()
run_manager.get_many_by_thread.return_value = records or {}
app.state.run_manager = run_manager
feedback_repo = AsyncMock()
feedback_repo.list_by_run_ids.return_value = feedback or {}
app.state.feedback_repo = feedback_repo
return app
async def _put_message(store, run_id, message_type, message_id, *, caller="lead_agent"):
return await store.put(
thread_id="thread-1",
run_id=run_id,
event_type="llm.ai.response" if message_type == "ai" else "llm.human.input",
category="message",
content={"type": message_type, "id": message_id, "content": message_id, "additional_kwargs": {}},
metadata={"caller": caller},
)
def test_thread_page_orders_across_runs_and_paginates_without_gaps():
store = MemoryRunEventStore()
async def seed():
for index in range(1, 7):
await _put_message(store, f"run-{(index + 1) // 2}", "human" if index % 2 else "ai", f"m-{index}")
asyncio.run(seed())
app = _make_app(store)
with TestClient(app) as client:
latest = client.get("/api/threads/thread-1/messages/page?limit=3")
older = client.get("/api/threads/thread-1/messages/page?limit=3&before_seq=4")
assert latest.status_code == 200
assert [row["seq"] for row in latest.json()["data"]] == [4, 5, 6]
assert latest.json()["has_more"] is True
assert latest.json()["next_before_seq"] == 4
assert [row["seq"] for row in older.json()["data"]] == [1, 2, 3]
assert older.json()["has_more"] is False
assert older.json()["next_before_seq"] is None
def test_thread_page_scans_past_middleware_chunks_to_fill_visible_page(monkeypatch):
monkeypatch.setattr(thread_runs, "THREAD_MESSAGE_PAGE_SCAN_BATCH", 3)
store = MemoryRunEventStore()
async def seed():
await _put_message(store, "run-1", "human", "visible-old")
for index in range(3):
await _put_message(store, "run-1", "ai", f"middleware-{index}", caller="middleware:title")
await _put_message(store, "run-2", "human", "visible-new-human")
await _put_message(store, "run-2", "ai", "visible-new-ai")
asyncio.run(seed())
app = _make_app(store)
with TestClient(app) as client:
response = client.get("/api/threads/thread-1/messages/page?limit=2")
body = response.json()
assert [row["seq"] for row in body["data"]] == [5, 6]
assert body["has_more"] is True
assert body["next_before_seq"] == 5
def test_thread_page_scans_large_middleware_only_region_with_production_batch_size():
store = MemoryRunEventStore()
async def seed():
await _put_message(store, "run-old", "human", "visible-old")
for index in range(thread_runs.THREAD_MESSAGE_PAGE_SCAN_BATCH * 2):
await _put_message(store, "run-middle", "ai", f"middleware-{index}", caller="middleware:title")
await _put_message(store, "run-new", "human", "visible-new-human")
await _put_message(store, "run-new", "ai", "visible-new-ai")
asyncio.run(seed())
original_list_messages = store.list_messages
store.list_messages = AsyncMock(wraps=original_list_messages)
app = _make_app(store)
with TestClient(app) as client:
response = client.get("/api/threads/thread-1/messages/page?limit=2")
body = response.json()
assert response.status_code == 200
assert [row["seq"] for row in body["data"]] == [404, 405]
assert body["has_more"] is True
assert body["next_before_seq"] == 404
assert store.list_messages.await_count == 3
def test_thread_page_filters_all_successfully_superseded_runs_before_filling():
store = MemoryRunEventStore()
async def seed():
await _put_message(store, "run-a", "ai", "answer-a")
await _put_message(store, "run-b", "ai", "answer-b")
await _put_message(store, "run-c", "ai", "answer-c")
asyncio.run(seed())
app = _make_app(store, superseded={"run-a", "run-b"})
with TestClient(app) as client:
response = client.get("/api/threads/thread-1/messages/page?limit=2")
body = response.json()
assert [row["run_id"] for row in body["data"]] == ["run-c"]
assert body["has_more"] is False
assert body["next_before_seq"] is None
def test_thread_page_logs_rows_missing_sequence_values(caplog):
store = AsyncMock()
store.list_messages.return_value = [{"run_id": "run-1", "content": {"type": "human"}}]
app = _make_app(store)
with caplog.at_level(logging.ERROR, logger="app.gateway.routers.thread_runs"):
with TestClient(app) as client, pytest.raises(RuntimeError, match="missing sequence values"):
client.get("/api/threads/thread-1/messages/page")
assert "Thread message scan found rows without sequence values" in caplog.text
assert "thread_id=thread-1" in caplog.text
assert "scan_before=None" in caplog.text
assert "row_count=1" in caplog.text
def test_thread_page_logs_when_scan_cursor_does_not_advance(caplog):
store = AsyncMock()
store.list_messages.return_value = [{"run_id": "run-1", "seq": 10, "content": {"type": "human"}}]
app = _make_app(store)
with caplog.at_level(logging.ERROR, logger="app.gateway.routers.thread_runs"):
with TestClient(app) as client, pytest.raises(RuntimeError, match="did not advance"):
client.get("/api/threads/thread-1/messages/page?before_seq=10")
assert "Thread message scan cursor did not advance" in caplog.text
assert "thread_id=thread-1" in caplog.text
assert "scan_before=10" in caplog.text
assert "next_scan_before=10" in caplog.text
assert "row_count=1" in caplog.text
def test_thread_page_feedback_only_attaches_to_global_last_ai_row():
store = MemoryRunEventStore()
async def seed():
await _put_message(store, "run-1", "ai", "draft")
await _put_message(store, "run-1", "human", "follow-up")
await _put_message(store, "run-1", "ai", "final")
asyncio.run(seed())
original_list_messages = store.list_messages
original_get_last_visible_ai_seq_by_run = store.get_last_visible_ai_seq_by_run
store.list_messages = AsyncMock(wraps=original_list_messages)
store.get_last_visible_ai_seq_by_run = AsyncMock(wraps=original_get_last_visible_ai_seq_by_run)
feedback = {"run-1": {"feedback_id": "fb-1", "rating": 1, "comment": "good"}}
app = _make_app(store, feedback=feedback)
with TestClient(app) as client:
response = client.get("/api/threads/thread-1/messages/page?limit=3")
data = response.json()["data"]
assert data[0]["feedback"] is None
assert data[1]["feedback"] is None
assert data[2]["feedback"] == {"feedback_id": "fb-1", "rating": 1, "comment": "good"}
scan_user_id = store.list_messages.await_args.kwargs["user_id"]
enrichment_user_id = store.get_last_visible_ai_seq_by_run.await_args.kwargs["user_id"]
assert enrichment_user_id == scan_user_id
feedback_repo = app.state.feedback_repo
feedback_repo.list_by_run_ids.assert_awaited_once_with("thread-1", {"run-1"}, user_id=scan_user_id)
feedback_repo.list_by_thread_grouped.assert_not_awaited()
def test_thread_page_helpers_forward_explicit_user_without_request_context():
event_store = AsyncMock()
event_store.list_messages.return_value = []
event_store.get_last_visible_ai_seq_by_run.return_value = {}
run_manager = AsyncMock()
run_manager.list_successful_regenerate_sources.return_value = set()
run_manager.get_many_by_thread.return_value = {}
request = MagicMock()
request.app.state.run_event_store = event_store
request.app.state.run_manager = run_manager
request.app.state.feedback_repo = AsyncMock()
async def exercise_helpers():
await thread_runs._scan_thread_message_page(
"thread-1",
limit=10,
before_seq=None,
request=request,
user_id="background-user",
)
await thread_runs._enrich_thread_message_page(
"thread-1",
[{"run_id": "run-1", "seq": 1, "content": {"type": "human"}}],
request=request,
user_id="background-user",
)
asyncio.run(exercise_helpers())
assert event_store.list_messages.await_args.kwargs["user_id"] == "background-user"
assert event_store.get_last_visible_ai_seq_by_run.await_args.kwargs["user_id"] == "background-user"
def test_thread_page_scan_rejects_any_row_without_sequence():
event_store = AsyncMock()
event_store.list_messages.return_value = [
{"run_id": "run-1", "seq": 1, "content": {"type": "human"}},
{"run_id": "run-1", "content": {"type": "ai"}},
]
run_manager = AsyncMock()
run_manager.list_successful_regenerate_sources.return_value = set()
request = MagicMock()
request.app.state.run_event_store = event_store
request.app.state.run_manager = run_manager
with pytest.raises(RuntimeError, match="missing sequence values"):
asyncio.run(
thread_runs._scan_thread_message_page(
"thread-1",
limit=1,
before_seq=None,
request=request,
user_id="user-1",
)
)
def test_thread_page_batch_hydrates_duration_for_old_runs():
store = MemoryRunEventStore()
asyncio.run(_put_message(store, "run-old", "ai", "answer"))
record = RunRecord(
run_id="run-old",
thread_id="thread-1",
assistant_id=None,
status="success",
on_disconnect="cancel",
created_at="2026-01-01T00:00:00Z",
updated_at="2026-01-01T00:00:07Z",
)
app = _make_app(store, records={"run-old": record})
with TestClient(app) as client:
response = client.get("/api/threads/thread-1/messages/page")
assert response.json()["data"][0]["content"]["additional_kwargs"]["turn_duration"] == 7
def test_thread_page_preserves_tool_and_subagent_wrapper_metadata():
store = MemoryRunEventStore()
asyncio.run(
store.put(
thread_id="thread-1",
run_id="run-tool",
event_type="tool.result",
category="message",
content={
"type": "tool",
"id": "tool-message-1",
"tool_call_id": "call-1",
"content": "result",
"artifact": {"kind": "subagent"},
},
metadata={"caller": "subagent:research", "task_id": "task-1", "message_index": 3},
)
)
app = _make_app(store)
with TestClient(app) as client:
response = client.get("/api/threads/thread-1/messages/page")
row = response.json()["data"][0]
assert row["run_id"] == "run-tool"
assert row["content"]["artifact"] == {"kind": "subagent"}
assert row["metadata"] == {"caller": "subagent:research", "task_id": "task-1", "message_index": 3}
def test_thread_page_empty_and_exact_limit_cursor_contract():
empty_store = MemoryRunEventStore()
with TestClient(_make_app(empty_store)) as client:
empty = client.get("/api/threads/thread-1/messages/page?limit=2")
assert empty.json() == {"data": [], "has_more": False, "next_before_seq": None}
store = MemoryRunEventStore()
async def seed():
await _put_message(store, "run-1", "human", "one")
await _put_message(store, "run-1", "ai", "two")
asyncio.run(seed())
with TestClient(_make_app(store)) as client:
exact = client.get("/api/threads/thread-1/messages/page?limit=2")
assert [row["seq"] for row in exact.json()["data"]] == [1, 2]
assert exact.json()["has_more"] is False
assert exact.json()["next_before_seq"] is None
def test_thread_page_rejects_forward_cursor_and_invalid_bounds():
app = _make_app(MemoryRunEventStore())
with TestClient(app) as client:
assert client.get("/api/threads/thread-1/messages/page?after_seq=1").status_code == 422
assert client.get("/api/threads/thread-1/messages/page?limit=0").status_code == 422
assert client.get("/api/threads/thread-1/messages/page?limit=201").status_code == 422
assert client.get("/api/threads/thread-1/messages/page?before_seq=0").status_code == 422

View File

@ -341,6 +341,23 @@ class TestBeforeAgent:
assert result is not None
assert result["messages"][-1].additional_kwargs[ORIGINAL_USER_CONTENT_KEY] == "/data-analysis run"
def test_replaces_non_string_original_user_content_before_upload_context(self, tmp_path):
mw = _middleware(tmp_path)
uploads_dir = _uploads_dir(tmp_path)
(uploads_dir / "report.pdf").write_bytes(b"pdf")
msg = _human(
"/data-analysis run",
files=[{"filename": "report.pdf", "size": 3, "path": "/mnt/user-data/uploads/report.pdf"}],
**{ORIGINAL_USER_CONTENT_KEY: [{"type": "text", "text": "spoofed audit text"}]},
)
result = mw.before_agent(self._state(msg), _runtime())
assert result is not None
updated_msg = result["messages"][-1]
assert updated_msg.additional_kwargs[ORIGINAL_USER_CONTENT_KEY] == "/data-analysis run"
assert updated_msg.content.startswith("<uploaded_files>")
def test_uploaded_files_returned_in_state_update(self, tmp_path):
mw = _middleware(tmp_path)
uploads_dir = _uploads_dir(tmp_path)

View File

@ -10,7 +10,9 @@ from __future__ import annotations
from types import SimpleNamespace
from deerflow.utils.messages import message_content_to_text, message_to_text
from langchain_core.messages import HumanMessage
from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, message_content_to_text, message_to_text, restore_original_human_message
# ---------- message_to_text: content shapes ----------
@ -70,3 +72,62 @@ def test_non_string_text_attribute_ignored():
def test_message_content_to_text_still_joins_with_newline():
assert message_content_to_text(["a", {"text": "b"}]) == "a\nb"
# ---------- restore_original_human_message ----------
def test_restore_original_human_message_restores_string_without_mutating_model_copy():
wrapped = HumanMessage(
content="--- BEGIN USER INPUT ---\nhello\n--- END USER INPUT ---",
id="human-1",
name="request",
additional_kwargs={ORIGINAL_USER_CONTENT_KEY: "hello", "hide_from_ui": False},
response_metadata={"source": "gateway"},
)
restored = restore_original_human_message(wrapped)
assert restored is not wrapped
assert restored.content == "hello"
assert restored.id == "human-1"
assert restored.name == "request"
assert restored.additional_kwargs == {"hide_from_ui": False}
assert restored.response_metadata == {"source": "gateway"}
assert wrapped.content.startswith("--- BEGIN USER INPUT ---")
assert wrapped.additional_kwargs[ORIGINAL_USER_CONTENT_KEY] == "hello"
def test_restore_original_human_message_preserves_mixed_non_text_blocks_in_order():
image = {"type": "image_url", "image_url": {"url": "data:image/png;base64,abc"}}
file_block = {"type": "file", "file_id": "file-1"}
wrapped = HumanMessage(
content=[
image,
{"type": "text", "text": "--- BEGIN USER INPUT ---\ncompare\n--- END USER INPUT ---"},
file_block,
],
additional_kwargs={ORIGINAL_USER_CONTENT_KEY: "compare", "metadata": {"source": "user"}},
)
restored = restore_original_human_message(wrapped)
assert restored.content == [image, {"type": "text", "text": "compare"}, file_block]
assert restored.additional_kwargs == {"metadata": {"source": "user"}}
assert wrapped.content[1]["text"].startswith("--- BEGIN USER INPUT ---")
assert restored.content[0] is not wrapped.content[0]
assert restored.content[0]["image_url"] is not wrapped.content[0]["image_url"]
assert restored.additional_kwargs["metadata"] is not wrapped.additional_kwargs["metadata"]
restored.content[0]["image_url"]["url"] = "data:image/png;base64,changed"
restored.additional_kwargs["metadata"]["source"] = "history"
assert wrapped.content[0]["image_url"]["url"] == "data:image/png;base64,abc"
assert wrapped.additional_kwargs["metadata"]["source"] == "user"
def test_restore_original_human_message_without_original_metadata_is_unchanged():
message = HumanMessage(content="already UI-facing", additional_kwargs={"source": "user"})
assert restore_original_human_message(message) is message

View File

@ -65,9 +65,10 @@ The frontend is a stateful chat application. Users create **threads** (conversat
1. Optional composer helpers such as `core/input-polish` can rewrite the local draft before submission, and `core/voice-input` can transcribe browser microphone input into that same local draft; confirmed user input then flows to thread hooks (`core/threads/hooks.ts`) → LangGraph SDK streaming
2. Stream events update thread state (messages, artifacts, todos, goal)
3. Stop actions call the LangGraph SDK stream stop path; `core/threads/hooks.ts` invalidates current-thread, 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
4. TanStack Query manages server state; localStorage stores user settings
5. Components subscribe to thread state and render updates
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), 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.
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
6. Components subscribe to thread state and render updates
`/goal` and `/compact` are built-in composer commands, not skill activations. `src/components/workspace/input-box.tsx` intercepts `/goal`, `/goal clear`, and `/goal <condition>` before normal chat submission, calling Gateway `GET/PUT/DELETE /api/threads/{thread_id}/goal`. Setting `/goal <condition>` also submits the condition text as the next user task so the agent starts running immediately; status and clear do not start a run. Goal and compact requests are tied to the current `threadId` with an `AbortController`, so switching threads or unmounting the composer aborts in-flight requests and stale responses cannot update the new thread's composer state. The chat pages render `GoalStatus` above the composer from `AgentThreadState.goal`, with local optimistic state until the next stream `values` update arrives. `/compact` calls `POST /api/threads/{thread_id}/compact` to summarize older active context while leaving the full visible chat history intact; it is skipped on new/empty threads and blocked server-side while a run is in flight.

View File

@ -229,26 +229,6 @@ function dedupeRunMessagesByIdentity(messages: RunMessage[]): RunMessage[] {
});
}
export function getSupersededRunIds(
runs: Run[] | undefined,
pendingSupersededRunIds?: ReadonlySet<string>,
) {
const ids = new Set(pendingSupersededRunIds ?? []);
for (const run of runs ?? []) {
if (run.status !== "success") {
continue;
}
const metadata = run.metadata;
if (metadata && typeof metadata === "object") {
const fromRunId = Reflect.get(metadata, "regenerate_from_run_id");
if (typeof fromRunId === "string" && fromRunId) {
ids.add(fromRunId);
}
}
}
return ids;
}
export function removeSetItems<T>(
values: ReadonlySet<T>,
itemsToRemove: Iterable<T>,
@ -263,7 +243,6 @@ export function removeSetItems<T>(
export function buildVisibleHistoryMessages(
messageRows: RunMessage[],
supersededRunIds: ReadonlySet<string>,
appendedMessages: Message[],
) {
const visibleRows = messageRows.filter(
(message) => !supersededRunIds.has(message.run_id),
@ -276,75 +255,40 @@ export function buildVisibleHistoryMessages(
...message.content,
run_id: message.run_id,
})),
...appendedMessages,
]);
}
export function findLatestUnloadedRunIndex(
runs: Run[],
loadedRunIds: ReadonlySet<string>,
): number {
for (let i = 0; i < runs.length; i++) {
const run = runs[i];
if (run && !loadedRunIds.has(run.run_id)) {
return i;
}
}
return -1;
}
export const MAX_CONSECUTIVE_EMPTY_RUN_LOADS = 5;
export function shouldAutoContinueOnEmptyRun(
fetchedMessageCount: number,
consecutiveEmptyLoads: number,
maxConsecutiveEmptyLoads: number = MAX_CONSECUTIVE_EMPTY_RUN_LOADS,
): boolean {
return (
fetchedMessageCount === 0 &&
consecutiveEmptyLoads < maxConsecutiveEmptyLoads
);
}
type RunMessagesPageResponse = {
export type ThreadMessagesPageResponse = {
data: RunMessage[];
has_more?: boolean;
hasMore?: boolean;
has_more: boolean;
next_before_seq: number | null;
};
export function runMessagesPageHasMore(result: RunMessagesPageResponse) {
return result.has_more ?? result.hasMore ?? false;
}
export function getOldestRunMessageSeq(messages: RunMessage[]) {
let oldestSeq: number | null = null;
for (const message of messages) {
if (typeof message.seq !== "number") {
continue;
}
oldestSeq =
oldestSeq === null ? message.seq : Math.min(oldestSeq, message.seq);
export function getThreadHistoryNextPageParam(
lastPage: ThreadMessagesPageResponse,
): number | undefined {
if (!lastPage.has_more) {
return undefined;
}
return oldestSeq;
}
export function getNextRunMessagesBeforeSeq(
result: RunMessagesPageResponse,
): number | null | undefined {
if (!runMessagesPageHasMore(result)) {
return null;
if (lastPage.next_before_seq === null) {
console.warn(
"Thread history returned has_more without next_before_seq; pagination cannot continue.",
);
return undefined;
}
return getOldestRunMessageSeq(result.data) ?? undefined;
return lastPage.next_before_seq;
}
export function buildRunMessagesUrl(
export const threadHistoryQueryKey = (threadId: string) =>
["thread-messages", threadId] as const;
export function buildThreadMessagesPageUrl(
baseUrl: string,
threadId: string,
runId: string,
beforeSeq?: number,
) {
const normalizedBaseUrl = baseUrl.replace(/\/$/, "");
const path = `/api/threads/${encodeURIComponent(threadId)}/runs/${encodeURIComponent(runId)}/messages`;
const path = `/api/threads/${encodeURIComponent(threadId)}/messages/page`;
const url = new URL(
`${normalizedBaseUrl}${path}`,
typeof window !== "undefined" ? window.location.origin : "http://localhost",
@ -355,16 +299,22 @@ export function buildRunMessagesUrl(
return normalizedBaseUrl ? url.toString() : `${url.pathname}${url.search}`;
}
export function flattenThreadHistoryPages(
pages: ThreadMessagesPageResponse[],
): RunMessage[] {
return dedupeRunMessagesByIdentity(
pages
.slice()
.reverse()
.flatMap((page) => page.data),
);
}
export function mergeMessages(
historyMessages: Message[],
threadMessages: Message[],
optimisticMessages: Message[],
): Message[] {
// Only visible live messages should trim overlapping history. Hidden messages
// are UI control messages in this path, not observability records; any hidden
// message that must survive as task/tracing data should use custom events or a
// separate state channel instead of participating in this overlap heuristic.
const savedTurnDurations = new Map<string, number>();
for (const msg of historyMessages) {
const identity = messageIdentity(msg);
@ -376,33 +326,87 @@ export function mergeMessages(
}
}
const threadMessageIds = new Set(
threadMessages
.filter((message) => !isHiddenFromUIMessage(message))
.map(messageIdentity)
.filter(isNonEmptyString),
const canonical = dedupeMessagesByIdentity(historyMessages);
const live = dedupeMessagesByIdentity(threadMessages);
const canonicalByIdentity = new Map(
canonical.flatMap((message) => {
const identity = messageIdentity(message);
return identity ? [[identity, message] as const] : [];
}),
);
const replacementByIdentity = new Map<string, Message>();
// This uses the same identity-anchor weaving shape as
// resolveTransientHistoryBridge, but intentionally remains separate: live
// messages may replace canonical copies and identity-less entries survive.
const beforeAnchor = new Map<string, Message[]>();
let pending: Message[] = [];
let lastAnchorIdentity: string | undefined;
let hasSharedAnchor = false;
// The overlap is a contiguous suffix of historyMessages (newest history == oldest thread).
// Scan from the end: shrink cutoff while messages are already in thread, stop as soon as
// we hit one that isn't — everything before that point is non-overlapping.
let cutoff = historyMessages.length;
for (let i = historyMessages.length - 1; i >= 0; i--) {
const msg = historyMessages[i];
if (!msg) {
// A summarized checkpoint is not necessarily a contiguous history suffix:
// middleware may retain protected prompt/input messages at the front and a
// recent tail at the back. Treat every shared identity as an ordering anchor,
// 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) {
const identity = messageIdentity(message);
const canonicalMessage = identity
? canonicalByIdentity.get(identity)
: undefined;
if (!identity || !canonicalMessage) {
pending.push(message);
continue;
}
const identity = messageIdentity(msg);
if (identity && threadMessageIds.has(identity)) {
cutoff = i;
} else {
break;
if (pending.length > 0 && hasSharedAnchor) {
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
// user turn that happens to reuse its identity. In every other case the
// live checkpoint copy is fresher and replaces history without moving it.
if (
!isHiddenFromUIMessage(message) ||
isHiddenFromUIMessage(canonicalMessage)
) {
replacementByIdentity.set(identity, message);
}
}
let canonicalAndLive: Message[];
if (!lastAnchorIdentity) {
canonicalAndLive = [...canonical, ...live];
} else {
canonicalAndLive = [];
for (const message of canonical) {
const identity = messageIdentity(message);
if (identity) {
canonicalAndLive.push(...(beforeAnchor.get(identity) ?? []));
}
const replacement = identity
? replacementByIdentity.get(identity)
: undefined;
canonicalAndLive.push(replacement ?? message);
}
// A trailing live-only segment is known to come after the last shared
// anchor, but that anchor may not be the end of canonical history (for
// example, another client may have persisted newer rows). Preserve the
// canonical source order before appending the live tail.
canonicalAndLive.push(...pending);
}
const merged = dedupeMessagesByIdentity([
...historyMessages.slice(0, cutoff),
...threadMessages,
...canonicalAndLive,
...optimisticMessages,
]);
@ -427,14 +431,14 @@ export function mergeMessages(
/**
* Derive the live turns that context summarization is about to drop and that
* therefore must be re-archived into history.
* therefore need a short-lived visual bridge until run-event history catches up.
*
* Summarization emits `RemoveMessage(ALL)` + a hidden summary + the retained
* tail. Everything in the current live thread before the first retained visible
* message is being removed; we keep those (minus the summary control messages
* already tracked) so the UI can still show the full conversation (#3825).
*/
export function computeSummarizationMovedMessages(
export function computeSummarizationTransientMessages(
currentMessages: Message[],
summarizationMessages: Message[],
summarizedMessageIds: ReadonlySet<string>,
@ -461,66 +465,211 @@ export function computeSummarizationMovedMessages(
}
/**
* Overlay the messages rescued from context summarization on top of the
* Overlay messages rescued from context summarization on top of the
* (possibly stale) visible history so the merged view never drops them.
*
* Background (#3825): after summarization the backend removes every live
* message (`RemoveMessage(ALL)`) and `onUpdateEvent` re-archives the removed
* messages into history through an async `setState`. The live thread messages
* are owned by the LangGraph SDK external store while the archived history is
* React state, so a render can observe the post-summary (shrunk) thread before
* the archive `setState` commits leaving the rescued messages in neither
* merge input. Reading them from a synchronous buffer here keeps the merge
* correct at every render regardless of how the two state channels interleave.
* message (`RemoveMessage(ALL)`) while canonical run events can still be
* waiting for the journal flush/refetch lifecycle. Reading the captured turns
* from a synchronous transient buffer keeps the merge correct during that gap.
*
* The rescued messages are the oldest live turns, so they follow whatever the
* already-loaded history holds. Only messages still missing from history are
* appended: once history absorbs a rescued message, its live copy stays
* authoritative (the buffered copy is an older snapshot and must never overwrite
* it), and ordering is preserved.
* Canonical history is cursor-paginated from newest to oldest. A rescued turn
* can therefore be older than the first row in the currently loaded page even
* though both came from the same pre-compression checkpoint. ``bridgeOrder``
* retains identities that canonical history has already confirmed so missing
* rescued turns can be inserted next to an overlapping anchor instead of being
* blindly appended after the newest page. Canonical copies always win.
*/
export function resolvePreservedHistory(
export function resolveTransientHistoryBridge(
visibleHistory: Message[],
pendingArchivedMessages: Message[],
transientMessages: Message[],
bridgeOrder: readonly string[] = transientMessages
.map(messageIdentity)
.filter(isNonEmptyString),
): Message[] {
if (pendingArchivedMessages.length === 0) {
if (transientMessages.length === 0) {
return visibleHistory;
}
const presentIdentities = new Set(
visibleHistory.map(messageIdentity).filter(isNonEmptyString),
);
const missing = pendingArchivedMessages.filter((message) => {
const missing = transientMessages.filter((message) => {
const identity = messageIdentity(message);
// Identity-less messages are intentionally skipped: without a stable
// identity they cannot be matched against history to drain or dedupe, so
// overlaying them would risk a permanent duplicate. They are still archived
// through appendMessages and surface via the normal history path instead.
// overlaying them would risk a permanent duplicate. Canonical history will
// surface them after the run journal is flushed and the page refetches.
return identity !== undefined && !presentIdentities.has(identity);
});
if (missing.length === 0) {
return visibleHistory;
}
return [...visibleHistory, ...missing];
const missingByIdentity = new Map(
missing.flatMap((message) => {
const identity = messageIdentity(message);
return identity ? [[identity, message] as const] : [];
}),
);
// This mirrors mergeMessages' identity-anchor weaving shape, but transient
// messages never replace canonical copies and identity-less entries are
// intentionally excluded to avoid permanent duplicates.
const beforeAnchor = new Map<string, Message[]>();
const emittedMissingIdentities = new Set<string>();
let pending: Message[] = [];
let lastAnchorIdentity: string | undefined;
let hasCanonicalAnchor = false;
for (const identity of bridgeOrder) {
if (presentIdentities.has(identity)) {
if (pending.length > 0 && hasCanonicalAnchor) {
beforeAnchor.set(identity, [
...(beforeAnchor.get(identity) ?? []),
...pending,
]);
}
// The prefix before the first loaded anchor has no trustworthy position:
// cursor pages containing its intervening history may not be loaded yet.
pending = [];
hasCanonicalAnchor = true;
lastAnchorIdentity = identity;
continue;
}
const message = missingByIdentity.get(identity);
if (message && !emittedMissingIdentities.has(identity)) {
pending.push(message);
emittedMissingIdentities.add(identity);
}
}
// No bridge identity overlaps canonical history. This is the original
// persistence-gap case: loaded history is older and the rescued live turns
// belong after it.
if (!lastAnchorIdentity) {
return [...visibleHistory, ...missing];
}
// A candidate added before its ordering snapshot (or carrying an identity
// absent from that snapshot) cannot be anchored. Keep it in capture order at
// the trailing edge of the anchored bridge rather than dropping it.
for (const message of missing) {
const identity = messageIdentity(message);
if (identity && !emittedMissingIdentities.has(identity)) {
pending.push(message);
emittedMissingIdentities.add(identity);
}
}
const resolved: Message[] = [];
for (const message of visibleHistory) {
const identity = messageIdentity(message);
if (identity) {
resolved.push(...(beforeAnchor.get(identity) ?? []));
}
resolved.push(message);
if (identity === lastAnchorIdentity) {
resolved.push(...pending);
}
}
return resolved;
}
export function mergeTransientHistoryBridge(
currentBridge: Message[],
capturedMessages: Message[],
): Message[] {
const merged = dedupeMessagesByIdentity(currentBridge);
const indexByIdentity = new Map<string, number>();
merged.forEach((message, index) => {
const identity = messageIdentity(message);
if (identity) {
indexByIdentity.set(identity, index);
}
});
for (const captured of dedupeMessagesByIdentity(capturedMessages)) {
const identity = messageIdentity(captured);
const existingIndex = identity ? indexByIdentity.get(identity) : undefined;
if (existingIndex === undefined) {
if (identity) {
indexByIdentity.set(identity, merged.length);
}
merged.push(captured);
continue;
}
const existing = merged[existingIndex];
if (
existing &&
(!isHiddenFromUIMessage(captured) || isHiddenFromUIMessage(existing))
) {
// Refresh the buffered snapshot without moving its first-known
// chronological position. Repeated compression can recapture protected
// prefix messages before a newer tail.
merged[existingIndex] = captured;
}
}
return merged;
}
/**
* Drop the archive-buffer entries that the canonical history state has already
* Preserve the complete checkpoint-relative identity order independently from
* bridge candidates. Confirmed candidates are pruned from the render buffer,
* but their identities remain useful as non-rendering pagination anchors.
*/
export function mergeTransientHistoryBridgeOrder(
currentOrder: readonly string[],
capturedMessages: Message[],
): string[] {
const capturedOrder = dedupeMessagesByIdentity(capturedMessages)
.map(messageIdentity)
.filter(isNonEmptyString);
const merged = [...currentOrder];
const seen = new Set(currentOrder);
for (const identity of capturedOrder) {
if (!seen.has(identity)) {
seen.add(identity);
merged.push(identity);
}
}
return merged;
}
export function resolveThreadTransientHistoryBridge(
visibleHistory: Message[],
transientMessages: Message[],
bridgeThreadId: string | null,
currentThreadId: string | null | undefined,
bridgeOrder?: readonly string[],
): Message[] {
if (!bridgeThreadId || bridgeThreadId !== currentThreadId) {
return visibleHistory;
}
return resolveTransientHistoryBridge(
visibleHistory,
transientMessages,
bridgeOrder,
);
}
/**
* Drop transient-buffer entries that canonical history has already
* absorbed. This keeps the buffer a transient bridge across the async gap
* rather than a second long-lived source of truth otherwise a stale copy
* could resurrect a message that history later filtered out (e.g. a superseded
* or regenerated run).
*/
export function pruneConfirmedArchivedMessages(
pendingArchivedMessages: Message[],
export function pruneConfirmedTransientMessages(
transientMessages: Message[],
visibleHistory: Message[],
): Message[] {
if (pendingArchivedMessages.length === 0) {
return pendingArchivedMessages;
if (transientMessages.length === 0) {
return transientMessages;
}
const confirmedIdentities = new Set(
visibleHistory.map(messageIdentity).filter(isNonEmptyString),
);
return pendingArchivedMessages.filter((message) => {
return transientMessages.filter((message) => {
const identity = messageIdentity(message);
return !identity || !confirmedIdentities.has(identity);
});
@ -681,6 +830,9 @@ export function invalidateStoppedThreadCaches(
}
void queryClient.invalidateQueries({ queryKey: ["thread", threadId] });
void queryClient.invalidateQueries({
queryKey: threadHistoryQueryKey(threadId),
});
void queryClient.invalidateQueries({
queryKey: ["thread", "metadata", threadId, isMock],
});
@ -831,7 +983,6 @@ export function useThreadStream({
hasMore: hasMoreHistory,
loadMore: loadMoreHistory,
loading: isHistoryLoading,
appendMessages,
} = useThreadHistory(onStreamThreadId ?? "", {
enabled: !isMock,
pendingSupersededRunIds,
@ -950,20 +1101,20 @@ export function useThreadStream({
summarizedRef.current?.add(m.id ?? "");
}
}
const _movedMessages = computeSummarizationMovedMessages(
const transientMessages = computeSummarizationTransientMessages(
messagesRef.current,
_messages,
summarizedRef.current ?? new Set<string>(),
);
// Buffer the rescued messages synchronously so the merge can keep
// displaying them immediately, even though appendMessages below only
// updates the archived-history state asynchronously (#3825).
pendingArchivedMessagesRef.current = dedupeMessagesByIdentity([
...pendingArchivedMessagesRef.current,
..._movedMessages,
]);
pendingArchiveThreadIdRef.current = threadIdRef.current;
appendMessages(_movedMessages);
transientHistoryOrderRef.current = mergeTransientHistoryBridgeOrder(
transientHistoryOrderRef.current,
transientMessages,
);
transientHistoryBridgeRef.current = mergeTransientHistoryBridge(
transientHistoryBridgeRef.current,
transientMessages,
);
transientHistoryThreadIdRef.current = threadIdRef.current;
messagesRef.current = [];
}
@ -1068,6 +1219,9 @@ export function useThreadStream({
.filter((id): id is string => Boolean(id)),
);
if (threadIdRef.current && !isMock) {
void queryClient.invalidateQueries({
queryKey: threadHistoryQueryKey(threadIdRef.current),
});
void queryClient.invalidateQueries({
queryKey: threadTokenUsageQueryKey(threadIdRef.current),
});
@ -1117,17 +1271,15 @@ export function useThreadStream({
const latestMessageCountsRef = useRef({ humanMessageCount });
const sendInFlightRef = useRef(false);
const messagesRef = useRef<Message[]>([]);
// Synchronous bridge for messages rescued from context summarization. The
// archived-history `setState` (via appendMessages) lands on a different
// schedule than the live thread external store, so the merge reads this buffer
// to avoid dropping rescued messages in the render window before history
// catches up (#3825).
const pendingArchivedMessagesRef = useRef<Message[]>([]);
// The thread the rescue buffer belongs to, captured when onUpdateEvent fills
// it. The merge only overlays the buffer when this matches the viewed
// `threadId`, so a previous thread's rescued messages can never flash into
// another thread or the new-chat screen (#3825).
const pendingArchiveThreadIdRef = useRef<string | null>(null);
// Current-stream lifecycle bridge for messages removed from the checkpoint
// tail before the canonical run-event page refetch observes the journal
// flush. It is never appended into useThreadHistory's persisted pages.
const transientHistoryBridgeRef = useRef<Message[]>([]);
// Full identity order of each captured checkpoint. Confirmed bridge entries
// are pruned from the message buffer, but remain here as non-rendering
// anchors so an older rescue can be placed before a newest-first page.
const transientHistoryOrderRef = useRef<string[]>([]);
const transientHistoryThreadIdRef = useRef<string | null>(null);
const summarizedRef = useRef<Set<string>>(null);
// Track human message count before sending to prevent clearing optimistic
// messages before the server's human message arrives (e.g. when AI messages
@ -1144,8 +1296,9 @@ export function useThreadStream({
startedRef.current = false;
sendInFlightRef.current = false;
messagesRef.current = [];
pendingArchivedMessagesRef.current = [];
pendingArchiveThreadIdRef.current = null;
transientHistoryBridgeRef.current = [];
transientHistoryOrderRef.current = [];
transientHistoryThreadIdRef.current = null;
summarizedRef.current = new Set<string>();
pendingUsageBaselineMessageIdsRef.current = new Set();
setPendingSupersededRunIds(new Set());
@ -1154,14 +1307,18 @@ export function useThreadStream({
latestMessageCountsRef.current.humanMessageCount;
}, [threadId]);
// Release archive-buffer entries once the canonical history state has absorbed
// them, so the synchronous bridge stays transient and never resurrects a
// message that history later filters out (e.g. a superseded run) (#3825).
// Release entries individually once canonical history confirms their stable
// identities. Keep unconfirmed entries across failure/refetch within this
// page lifecycle so a temporary persistence gap cannot hide a turn.
useEffect(() => {
pendingArchivedMessagesRef.current = pruneConfirmedArchivedMessages(
pendingArchivedMessagesRef.current,
transientHistoryBridgeRef.current = pruneConfirmedTransientMessages(
transientHistoryBridgeRef.current,
visibleHistory,
);
if (transientHistoryBridgeRef.current.length === 0) {
transientHistoryOrderRef.current = [];
transientHistoryThreadIdRef.current = null;
}
}, [visibleHistory]);
useEffect(() => {
@ -1532,16 +1689,37 @@ export function useThreadStream({
humanMessageCount,
);
// Overlay the summarization rescue buffer only onto the history of the thread
// it was captured from. visibleHistory is gated on `threadId`, so comparing the
// same prop keeps the buffer from flashing into another thread or the new-chat
// screen, and reading it here (instead of clearing a ref during render) is
// concurrent-mode safe (#3825).
const rescueBuffer = pendingArchivedMessagesRef.current;
const effectiveHistory =
rescueBuffer.length > 0 && pendingArchiveThreadIdRef.current === threadId
? resolvePreservedHistory(visibleHistory, rescueBuffer)
: visibleHistory;
const transientHistoryOrder =
transientHistoryBridgeRef.current.length > 0 &&
transientHistoryThreadIdRef.current === threadId
? mergeTransientHistoryBridgeOrder(
transientHistoryOrderRef.current,
persistedMessages,
)
: transientHistoryOrderRef.current;
// Commit the extended non-rendering order skeleton after React commits this
// render. The local value above keeps this render correctly anchored without
// mutating a ref during render.
useEffect(() => {
if (
transientHistoryBridgeRef.current.length > 0 &&
transientHistoryThreadIdRef.current === threadId
) {
transientHistoryOrderRef.current = mergeTransientHistoryBridgeOrder(
transientHistoryOrderRef.current,
persistedMessages,
);
}
}, [persistedMessages, threadId]);
const effectiveHistory = resolveThreadTransientHistoryBridge(
visibleHistory,
transientHistoryBridgeRef.current,
transientHistoryThreadIdRef.current,
threadId,
transientHistoryOrder,
);
const mergedMessages = mergeMessages(
effectiveHistory,
persistedMessages,
@ -1584,198 +1762,67 @@ export function useThreadHistory(
threadId: string,
{ enabled = true, pendingSupersededRunIds }: ThreadHistoryOptions = {},
) {
const runs = useThreadRuns(threadId, { enabled });
const threadIdRef = useRef(threadId);
const runsRef = useRef(runs.data ?? []);
const indexRef = useRef(-1);
const loadingRef = useRef(false);
const pendingLoadRef = useRef(false);
const loadingRunIdRef = useRef<string | null>(null);
const loadedRunIdsRef = useRef<Set<string>>(new Set());
const runBeforeSeqRef = useRef<Map<string, number>>(new Map());
const loadGenerationRef = useRef(0);
const [loading, setLoading] = useState(false);
const [messageRows, setMessageRows] = useState<RunMessage[]>([]);
const [appendedMessages, setAppendedMessages] = useState<Message[]>([]);
const historyQuery = useInfiniteQuery<
ThreadMessagesPageResponse,
Error,
InfiniteData<ThreadMessagesPageResponse>,
ReturnType<typeof threadHistoryQueryKey>,
number | null
>({
queryKey: threadHistoryQueryKey(threadId),
enabled: enabled && Boolean(threadId),
initialPageParam: null,
queryFn: async ({ pageParam, signal }) => {
const url = buildThreadMessagesPageUrl(
getBackendBaseURL(),
threadId,
pageParam ?? undefined,
);
const response = await fetch(url, {
method: "GET",
headers: {
"Content-Type": "application/json",
},
credentials: "include",
signal,
});
if (!response.ok) {
throw new Error(
await readResponseErrorMessage(
response,
"Failed to load thread history.",
),
);
}
return (await response.json()) as ThreadMessagesPageResponse;
},
getNextPageParam: getThreadHistoryNextPageParam,
});
const supersededRunIds = useMemo(() => {
return getSupersededRunIds(runs.data, pendingSupersededRunIds);
}, [pendingSupersededRunIds, runs.data]);
const messageRows = useMemo(
() => flattenThreadHistoryPages(historyQuery.data?.pages ?? []),
[historyQuery.data?.pages],
);
const messages = useMemo(() => {
return buildVisibleHistoryMessages(
messageRows,
supersededRunIds,
appendedMessages,
pendingSupersededRunIds ?? new Set<string>(),
);
}, [appendedMessages, messageRows, supersededRunIds]);
}, [messageRows, pendingSupersededRunIds]);
const loadMessages = useCallback(async () => {
if (!enabled) {
return;
}
const loadGeneration = loadGenerationRef.current;
if (loadingRef.current) {
const pendingRunIndex = findLatestUnloadedRunIndex(
runsRef.current,
loadedRunIdsRef.current,
);
const pendingRun = runsRef.current[pendingRunIndex];
if (pendingRun && pendingRun.run_id !== loadingRunIdRef.current) {
pendingLoadRef.current = true;
}
return;
}
if (runsRef.current.length === 0) {
return;
}
loadingRef.current = true;
setLoading(true);
try {
let consecutiveEmptyLoads = 0;
do {
pendingLoadRef.current = false;
const nextRunIndex = findLatestUnloadedRunIndex(
runsRef.current,
loadedRunIdsRef.current,
);
indexRef.current = nextRunIndex;
const run = runsRef.current[nextRunIndex];
if (!run) {
indexRef.current = -1;
return;
}
const requestThreadId = threadIdRef.current;
loadingRunIdRef.current = run.run_id;
const beforeSeq = runBeforeSeqRef.current.get(run.run_id);
const url = buildRunMessagesUrl(
getBackendBaseURL(),
requestThreadId,
run.run_id,
beforeSeq,
);
const result: RunMessagesPageResponse = await fetch(url, {
method: "GET",
headers: {
"Content-Type": "application/json",
},
credentials: "include",
}).then((res) => {
return res.json();
});
if (
loadGenerationRef.current !== loadGeneration ||
threadIdRef.current !== requestThreadId
) {
return;
}
const _messages = result.data.filter(
(m) => !m.metadata.caller?.startsWith("middleware:"),
);
setMessageRows((prev) =>
dedupeRunMessagesByIdentity([..._messages, ...prev]),
);
const nextBeforeSeq = getNextRunMessagesBeforeSeq(result);
if (typeof nextBeforeSeq === "number") {
runBeforeSeqRef.current.set(run.run_id, nextBeforeSeq);
pendingLoadRef.current = true;
} else if (nextBeforeSeq === undefined) {
console.warn(
`Run ${run.run_id} returned has_more without message seq values; leaving it pending for retry.`,
);
} else {
runBeforeSeqRef.current.delete(run.run_id);
loadedRunIdsRef.current.add(run.run_id);
if (
shouldAutoContinueOnEmptyRun(
_messages.length,
consecutiveEmptyLoads,
)
) {
consecutiveEmptyLoads += 1;
pendingLoadRef.current = true;
} else {
consecutiveEmptyLoads = 0;
}
}
indexRef.current = findLatestUnloadedRunIndex(
runsRef.current,
loadedRunIdsRef.current,
);
} while (pendingLoadRef.current);
} catch (err) {
console.error(err);
} finally {
if (loadGenerationRef.current === loadGeneration) {
loadingRef.current = false;
loadingRunIdRef.current = null;
setLoading(false);
}
}
}, [enabled]);
useEffect(() => {
const threadChanged = threadIdRef.current !== threadId;
threadIdRef.current = threadId;
if (!enabled || threadChanged) {
loadGenerationRef.current += 1;
runsRef.current = [];
indexRef.current = -1;
pendingLoadRef.current = false;
loadingRunIdRef.current = null;
loadedRunIdsRef.current = new Set();
runBeforeSeqRef.current = new Map();
loadingRef.current = false;
setLoading(false);
setMessageRows([]);
setAppendedMessages([]);
}
if (!enabled) {
return;
}
if (runs.data && runs.data.length > 0) {
runsRef.current = runs.data ?? [];
indexRef.current = findLatestUnloadedRunIndex(
runs.data,
loadedRunIdsRef.current,
);
}
loadMessages().catch(() => {
if (historyQuery.error) {
console.error(historyQuery.error);
toast.error("Failed to load thread history.");
});
}, [enabled, threadId, runs.data, loadMessages]);
}
}, [historyQuery.error]);
const appendMessages = useCallback((_messages: Message[]) => {
setAppendedMessages((prev) => {
return dedupeMessagesByIdentity([...prev, ..._messages]);
});
}, []);
const hasThreadId = Boolean(threadId);
const hasUnloadedRuns = Boolean(
runs.data?.some((run) => !loadedRunIdsRef.current.has(run.run_id)),
);
const isRunsLoading =
enabled &&
hasThreadId &&
(runs.isLoading || (runs.isFetching && !runs.data));
const isRunsUnresolved =
enabled && hasThreadId && !runs.data && !runs.isError;
const hasMore =
enabled && hasThreadId && (indexRef.current >= 0 || hasUnloadedRuns);
return {
runs: runs.data,
messages,
loading: loading || isRunsLoading || isRunsUnresolved,
appendMessages,
hasMore,
loadMore: loadMessages,
loading: historyQuery.isLoading || historyQuery.isFetchingNextPage,
hasMore: Boolean(historyQuery.hasNextPage),
loadMore: historyQuery.fetchNextPage,
};
}

View File

@ -573,7 +573,7 @@ test.describe("Side chat", () => {
},
);
await page.route(
new RegExp(`/api/threads/${MOCK_THREAD_ID}/runs/[^/]+/messages`),
new RegExp(`/api/threads/${MOCK_THREAD_ID}/messages/page`),
(route) => {
if (route.request().method() !== "GET") {
return route.fallback();
@ -584,11 +584,13 @@ test.describe("Side chat", () => {
body: JSON.stringify({
data: parentMessages.map((message, index) => ({
run_id: `run-${MOCK_THREAD_ID}`,
seq: index + 1,
content: message,
metadata: { caller: "lead_agent" },
created_at: `2025-01-01T00:00:${String(index).padStart(2, "0")}Z`,
})),
hasMore: false,
has_more: false,
next_before_seq: null,
}),
});
},
@ -671,7 +673,7 @@ test.describe("Side chat", () => {
},
);
await page.route(
new RegExp(`/api/threads/${MOCK_SIDECAR_THREAD_ID}/runs/[^/]+/messages`),
new RegExp(`/api/threads/${MOCK_SIDECAR_THREAD_ID}/messages/page`),
(route) => {
if (route.request().method() !== "GET") {
return route.fallback();
@ -682,11 +684,13 @@ test.describe("Side chat", () => {
body: JSON.stringify({
data: sidecarThreadMessages.map((message, index) => ({
run_id: `run-${MOCK_SIDECAR_THREAD_ID}`,
seq: index + 1,
content: message,
metadata: { caller: "lead_agent" },
created_at: `2025-01-01T00:00:${String(index).padStart(2, "0")}Z`,
})),
hasMore: false,
has_more: false,
next_before_seq: null,
}),
});
},

View File

@ -45,41 +45,40 @@ test("historical run messages preview labelled dotted Mermaid arrows", async ({
}),
);
await page.route(
`**/api/threads/${MOCK_THREAD_ID}/runs/${MOCK_RUN_ID}/messages`,
(route) =>
route.fulfill({
status: 200,
contentType: "application/json",
body: JSON.stringify({
data: [
{
thread_id: MOCK_THREAD_ID,
run_id: MOCK_RUN_ID,
event_type: "llm.ai.response",
category: "message",
content: {
content: mermaidContent,
additional_kwargs: {},
response_metadata: {},
type: "ai",
name: null,
id: "lc_run--issue-3193",
tool_calls: [],
invalid_tool_calls: [],
},
seq: 720,
created_at: "2026-05-24T04:47:01.123949+00:00",
metadata: {
caller: "lead_agent",
content_is_json: true,
content_is_dict: true,
},
await page.route(`**/api/threads/${MOCK_THREAD_ID}/messages/page`, (route) =>
route.fulfill({
status: 200,
contentType: "application/json",
body: JSON.stringify({
data: [
{
thread_id: MOCK_THREAD_ID,
run_id: MOCK_RUN_ID,
event_type: "llm.ai.response",
category: "message",
content: {
content: mermaidContent,
additional_kwargs: {},
response_metadata: {},
type: "ai",
name: null,
id: "lc_run--issue-3193",
tool_calls: [],
invalid_tool_calls: [],
},
],
has_more: false,
}),
seq: 720,
created_at: "2026-05-24T04:47:01.123949+00:00",
metadata: {
caller: "lead_agent",
content_is_json: true,
content_is_dict: true,
},
},
],
has_more: false,
next_before_seq: null,
}),
}),
);
await page.goto(`/workspace/chats/${MOCK_THREAD_ID}`);

View File

@ -951,31 +951,30 @@ export function mockLangGraphAPI(page: Page, options?: MockAPIOptions) {
return route.fallback();
});
void page.route(
/\/api\/threads\/([^/]+)\/runs\/([^/]+)\/messages/,
(route) => {
if (route.request().method() === "GET") {
const url = route.request().url();
const matchingThread = threads.find((t) =>
url.includes(`/api/threads/${t.thread_id}/runs/`),
);
return route.fulfill({
status: 200,
contentType: "application/json",
body: JSON.stringify({
data: (matchingThread?.messages ?? []).map((message, index) => ({
run_id: `run-${matchingThread?.thread_id ?? "unknown"}`,
content: message,
metadata: { caller: "lead_agent" },
created_at: `2025-01-01T00:00:${String(index).padStart(2, "0")}Z`,
})),
hasMore: false,
}),
});
}
return route.fallback();
},
);
void page.route(/\/api\/threads\/([^/]+)\/messages\/page/, (route) => {
if (route.request().method() === "GET") {
const url = route.request().url();
const matchingThread = threads.find((t) =>
url.includes(`/api/threads/${t.thread_id}/messages/page`),
);
return route.fulfill({
status: 200,
contentType: "application/json",
body: JSON.stringify({
data: (matchingThread?.messages ?? []).map((message, index) => ({
run_id: `run-${matchingThread?.thread_id ?? "unknown"}`,
seq: index + 1,
content: message,
metadata: { caller: "lead_agent" },
created_at: `2025-01-01T00:00:${String(index).padStart(2, "0")}Z`,
})),
has_more: false,
next_before_seq: null,
}),
});
}
return route.fallback();
});
// Run stream — returns a minimal SSE response with an AI message
const handleMockRunStream = (route: Route) => {

View File

@ -327,6 +327,24 @@ describe("invalidateStoppedThreadCaches", () => {
expect(queryKeys()).toContainEqual(["thread-token-usage", "thread-1"]);
});
test("preserves loaded history pages while invalidating", () => {
const client = new QueryClient();
const key = ["thread-messages", "thread-1"] as const;
const latest = { data: [], has_more: true, next_before_seq: 20 };
const older = { data: [], has_more: false, next_before_seq: null };
client.setQueryData(key, {
pages: [latest, older],
pageParams: [null, 20],
});
invalidateStoppedThreadCaches(client, "thread-1", false);
expect(client.getQueryData(key)).toEqual({
pages: [latest, older],
pageParams: [null, 20],
});
});
test("does not refresh per-thread API caches for mock threads", () => {
const client = new QueryClient();
const { queryKeys } = invalidatedQueryKeys(client);

View File

@ -1,23 +1,23 @@
import type { Message, Run } from "@langchain/langgraph-sdk";
import { expect, test } from "@rstest/core";
import type { Message } from "@langchain/langgraph-sdk";
import { expect, rs, test } from "@rstest/core";
import { InfiniteQueryObserver, QueryClient } from "@tanstack/react-query";
import {
buildRunMessagesUrl,
buildThreadMessagesPageUrl,
buildVisibleHistoryMessages,
computeSummarizationMovedMessages,
findLatestUnloadedRunIndex,
getNextRunMessagesBeforeSeq,
getOldestRunMessageSeq,
getSupersededRunIds,
computeSummarizationTransientMessages,
flattenThreadHistoryPages,
getSummarizationMiddlewareMessages,
getThreadHistoryNextPageParam,
getVisibleOptimisticMessages,
MAX_CONSECUTIVE_EMPTY_RUN_LOADS,
mergeTransientHistoryBridge,
mergeTransientHistoryBridgeOrder,
mergeMessages,
pruneConfirmedArchivedMessages,
pruneConfirmedTransientMessages,
removeSetItems,
resolvePreservedHistory,
runMessagesPageHasMore,
shouldAutoContinueOnEmptyRun,
resolveThreadTransientHistoryBridge,
resolveTransientHistoryBridge,
type ThreadMessagesPageResponse,
} from "@/core/threads/hooks";
import type { RunMessage } from "@/core/threads/types";
@ -46,6 +46,28 @@ 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", () => {
const protectedEarly = {
id: "protected-early",
type: "human",
content: "写一个算法PDF",
} as Message;
const latestHuman = {
id: "latest-human",
type: "human",
content: "写一本超级小说",
} as Message;
const latestAi = {
id: "latest-ai",
type: "ai",
content: "latest answer",
} as Message;
expect(
mergeMessages([latestHuman, latestAi], [protectedEarly, latestHuman], []),
).toEqual([latestHuman, latestAi]);
});
test("mergeMessages lets live thread messages replace overlapping history", () => {
const oldHuman = {
id: "human-1",
@ -74,6 +96,112 @@ test("mergeMessages lets live thread messages replace overlapping history", () =
]);
});
test("mergeMessages keeps a protected pre-compression input at its canonical position", () => {
const canonicalInput = {
id: "input-1",
type: "human",
content: "写一个算法PDF",
} as Message;
const checkpointInput = {
id: "input-1",
type: "human",
content: [{ type: "text", text: "写一个算法PDF" }],
} as Message;
const clarificationCard = {
id: "clarification-card",
type: "tool",
tool_call_id: "clarification-call",
content: "Create a new PDF",
} as Message;
const directionAnswer = {
id: "input-3",
type: "human",
content: "二叉树相关的即可",
} as Message;
const canonicalRetainedTail = {
id: "retained-ai",
type: "ai",
content: "persisted tail",
} as Message;
const checkpointRetainedTail = {
id: "retained-ai",
type: "ai",
content: "live tail",
} as Message;
expect(
mergeMessages(
[
canonicalInput,
clarificationCard,
directionAnswer,
canonicalRetainedTail,
],
[checkpointInput, checkpointRetainedTail],
[],
),
).toEqual([
checkpointInput,
clarificationCard,
directionAnswer,
checkpointRetainedTail,
]);
});
test("mergeMessages keeps source order when history and live tail do not overlap", () => {
const historyAi = {
id: "history-ai",
type: "ai",
content: "persisted",
} as Message;
const liveHuman = {
id: "live-human",
type: "human",
content: "live",
} as Message;
expect(mergeMessages([historyAi], [liveHuman], [])).toEqual([
historyAi,
liveHuman,
]);
});
test("mergeMessages appends a trailing live-only segment after newer canonical rows", () => {
const message = (id: string) =>
({ id, type: "human", content: id }) as Message;
const [a, b, c, d, y] = ["a", "b", "c", "d", "y"].map(message) as [
Message,
Message,
Message,
Message,
Message,
];
expect(mergeMessages([a, b, c, d], [b, y], [])).toEqual([a, b, c, d, y]);
});
test("mergeMessages keeps live-only messages between shared anchors in place", () => {
const message = (id: string) =>
({ id, type: "human", content: id }) as Message;
const [a, b, c, d, x, y] = ["a", "b", "c", "d", "x", "y"].map(message) as [
Message,
Message,
Message,
Message,
Message,
Message,
];
expect(mergeMessages([a, b, c, d], [b, x, d, y], [])).toEqual([
a,
b,
c,
x,
d,
y,
]);
});
test("mergeMessages deduplicates tool messages by tool_call_id", () => {
const oldTool = {
id: "tool-message-old",
@ -279,124 +407,153 @@ test("getVisibleOptimisticMessages hides optimistic user input after later serve
]);
});
test("runMessagesPageHasMore reads backend snake_case pagination field", () => {
expect(runMessagesPageHasMore({ data: [], has_more: true })).toBe(true);
expect(runMessagesPageHasMore({ data: [], has_more: false })).toBe(false);
});
test("runMessagesPageHasMore keeps compatibility with camelCase pagination field", () => {
expect(runMessagesPageHasMore({ data: [], hasMore: true })).toBe(true);
});
test("getOldestRunMessageSeq returns the cursor for the next older run page", () => {
test("buildThreadMessagesPageUrl encodes the thread and backward cursor", () => {
expect(
getOldestRunMessageSeq([runMessage(8), runMessage(9), runMessage(10)]),
).toBe(8);
});
test("getOldestRunMessageSeq ignores rows without seq", () => {
expect(getOldestRunMessageSeq([runMessage()])).toBeNull();
});
test("getNextRunMessagesBeforeSeq keeps runs pending when has_more lacks seq", () => {
expect(
getNextRunMessagesBeforeSeq({ data: [runMessage()], has_more: true }),
).toBeUndefined();
});
test("getNextRunMessagesBeforeSeq marks runs loaded when no more pages exist", () => {
expect(
getNextRunMessagesBeforeSeq({ data: [runMessage()], has_more: false }),
).toBeNull();
});
test("buildRunMessagesUrl encodes path segments and optional before_seq", () => {
expect(
buildRunMessagesUrl(
buildThreadMessagesPageUrl(
"https://api.example.test/",
"thread/with space",
"run?one",
18,
),
).toBe(
"https://api.example.test/api/threads/thread%2Fwith%20space/runs/run%3Fone/messages?before_seq=18",
"https://api.example.test/api/threads/thread%2Fwith%20space/messages/page?before_seq=18",
);
});
test("buildRunMessagesUrl omits before_seq when loading the latest page", () => {
test("buildThreadMessagesPageUrl omits before_seq for the latest page", () => {
expect(
buildRunMessagesUrl("https://api.example.test", "thread-1", "run-1"),
).toBe("https://api.example.test/api/threads/thread-1/runs/run-1/messages");
buildThreadMessagesPageUrl("https://api.example.test", "thread-1"),
).toBe("https://api.example.test/api/threads/thread-1/messages/page");
});
test("buildRunMessagesUrl returns a relative URL when using the nginx proxy", () => {
expect(buildRunMessagesUrl("", "thread-1", "run-1", 42)).toBe(
"/api/threads/thread-1/runs/run-1/messages?before_seq=42",
test("buildThreadMessagesPageUrl returns a relative URL behind nginx", () => {
expect(buildThreadMessagesPageUrl("", "thread-1", 42)).toBe(
"/api/threads/thread-1/messages/page?before_seq=42",
);
});
test("findLatestUnloadedRunIndex loads the newest run first from a newest-first list", () => {
const runs = [
{ run_id: "R6" },
{ run_id: "R5" },
{ run_id: "R4" },
{ run_id: "R3" },
{ run_id: "R2" },
{ run_id: "R1" },
] as unknown as Run[];
expect(findLatestUnloadedRunIndex(runs, new Set())).toBe(0);
test("flattenThreadHistoryPages prepends backward pages in global seq order", () => {
expect(
flattenThreadHistoryPages([
{
data: [runMessage(5), runMessage(6)],
has_more: true,
next_before_seq: 5,
},
{
data: [runMessage(3), runMessage(4)],
has_more: true,
next_before_seq: 3,
},
{
data: [runMessage(1), runMessage(2)],
has_more: false,
next_before_seq: null,
},
]).map((message) => message.seq),
).toEqual([1, 2, 3, 4, 5, 6]);
});
test("findLatestUnloadedRunIndex skips already-loaded runs and returns the next newest unloaded run", () => {
const runs = [
{ run_id: "R6" },
{ run_id: "R5" },
{ run_id: "R4" },
] as unknown as Run[];
expect(findLatestUnloadedRunIndex(runs, new Set(["R6"]))).toBe(1);
test("flattenThreadHistoryPages retains backward pages when the latest page refreshes", () => {
const olderPage = {
data: [runMessage(1), runMessage(2)],
has_more: false,
next_before_seq: null,
};
expect(
flattenThreadHistoryPages([
{
data: [runMessage(3), runMessage(4), runMessage(5)],
has_more: true,
next_before_seq: 3,
},
olderPage,
]).map((message) => message.seq),
).toEqual([1, 2, 3, 4, 5]);
});
test("findLatestUnloadedRunIndex returns -1 when every run is already loaded", () => {
const runs = [{ run_id: "R2" }, { run_id: "R1" }] as unknown as Run[];
expect(findLatestUnloadedRunIndex(runs, new Set(["R1", "R2"]))).toBe(-1);
test("infinite history refetch recalculates older-page cursors from the refreshed newest page", async () => {
const queryClient = new QueryClient({
defaultOptions: { queries: { retry: false } },
});
const queryKey = ["thread-messages", "thread-1"] as const;
const requestedCursors: Array<number | null> = [];
let availableSeqs = Array.from({ length: 9 }, (_, index) => index + 1);
const observer = new InfiniteQueryObserver(queryClient, {
queryKey,
initialPageParam: null as number | null,
queryFn: ({ pageParam }): ThreadMessagesPageResponse => {
requestedCursors.push(pageParam);
const eligible = availableSeqs.filter(
(seq) => pageParam === null || seq < pageParam,
);
const pageSeqs = eligible.slice(-3);
return {
data: pageSeqs.map(runMessage),
has_more: eligible.length > pageSeqs.length,
next_before_seq:
eligible.length > pageSeqs.length ? (pageSeqs[0] ?? null) : null,
};
},
getNextPageParam: getThreadHistoryNextPageParam,
});
const unsubscribe = observer.subscribe(() => undefined);
await observer.refetch();
await observer.fetchNextPage();
expect(requestedCursors).toEqual([null, 7]);
availableSeqs = Array.from({ length: 12 }, (_, index) => index + 1);
requestedCursors.length = 0;
await queryClient.invalidateQueries({ queryKey });
expect(requestedCursors).toEqual([null, 10]);
expect(
observer
.getCurrentResult()
.data?.pages.map((page) => page.data.map((message) => message.seq)),
).toEqual([
[10, 11, 12],
[7, 8, 9],
]);
expect(observer.getCurrentResult().data?.pageParams).toEqual([null, 10]);
unsubscribe();
queryClient.clear();
});
test("getSupersededRunIds combines completed regenerate metadata with pending ids", () => {
const runs = [
{
run_id: "run-new",
status: "success",
metadata: { regenerate_from_run_id: "run-old" },
test("infinite history stops and warns when has_more has no cursor", async () => {
const queryClient = new QueryClient({
defaultOptions: { queries: { retry: false } },
});
const requestedCursors: Array<number | null> = [];
const warnSpy = rs.spyOn(console, "warn").mockImplementation(() => ({}));
const observer = new InfiniteQueryObserver(queryClient, {
queryKey: ["thread-messages", "invalid-cursor"],
initialPageParam: null as number | null,
queryFn: ({ pageParam }): ThreadMessagesPageResponse => {
requestedCursors.push(pageParam);
return { data: [], has_more: true, next_before_seq: null };
},
{
run_id: "run-normal",
status: "success",
metadata: {},
},
] as unknown as Run[];
getNextPageParam: getThreadHistoryNextPageParam,
});
const unsubscribe = observer.subscribe(() => undefined);
expect(getSupersededRunIds(runs, new Set(["run-pending"]))).toEqual(
new Set(["run-old", "run-pending"]),
);
});
try {
await observer.refetch();
await observer.fetchNextPage();
test("getSupersededRunIds ignores failed regenerate runs but keeps pending ids", () => {
const runs = [
{
run_id: "run-error",
status: "error",
metadata: { regenerate_from_run_id: "run-old" },
},
{
run_id: "run-interrupted",
status: "interrupted",
metadata: { regenerate_from_run_id: "run-older" },
},
] as unknown as Run[];
expect(getSupersededRunIds(runs, new Set(["run-pending"]))).toEqual(
new Set(["run-pending"]),
);
expect(requestedCursors).toEqual([null]);
expect(observer.getCurrentResult().hasNextPage).toBe(false);
expect(warnSpy).toHaveBeenCalledWith(
"Thread history returned has_more without next_before_seq; pagination cannot continue.",
);
} finally {
unsubscribe();
warnSpy.mockRestore();
queryClient.clear();
}
});
test("removeSetItems removes pending superseded ids after submit failure", () => {
@ -455,7 +612,7 @@ 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([
expect(buildVisibleHistoryMessages(rows, new Set(["run-old"]))).toEqual([
{ ...newHuman, run_id: "run-new" },
{ ...newAi, run_id: "run-new" },
]);
@ -471,151 +628,15 @@ test("buildVisibleHistoryMessages attaches run_id to each content message (#3779
},
];
const result = buildVisibleHistoryMessages(rows, new Set(), []);
const result = buildVisibleHistoryMessages(rows, new Set());
expect((result[0] as { run_id?: string }).run_id).toBe("run-1");
});
test("loading runs in newest-first order and prepending pages yields chronological messages (regression for #3352)", () => {
// Simulate backend list_by_thread returning newest first.
const runs = [
{ run_id: "R6" },
{ run_id: "R5" },
{ run_id: "R4" },
{ run_id: "R3" },
{ run_id: "R2" },
{ run_id: "R1" },
] as unknown as Run[];
const runIdToContent: Record<string, string> = {
R1: "A",
R2: "B",
R3: "C",
R4: "D",
R5: "E",
R6: "F",
};
const loaded = new Set<string>();
let messages: Message[] = [];
while (true) {
const index = findLatestUnloadedRunIndex(runs, loaded);
if (index === -1) break;
const run = runs[index]!;
const pageMessages = [
{
id: run.run_id,
type: "human",
content: runIdToContent[run.run_id],
} as Message,
];
// Mirror loadMessages: prepend new page to existing messages.
messages = [...pageMessages, ...messages];
loaded.add(run.run_id);
}
expect(messages.map((m) => m.content)).toEqual([
"A",
"B",
"C",
"D",
"E",
"F",
]);
});
test("shouldAutoContinueOnEmptyRun does not continue when the run produced messages", () => {
expect(shouldAutoContinueOnEmptyRun(3, 0)).toBe(false);
expect(shouldAutoContinueOnEmptyRun(1, 4)).toBe(false);
});
test("shouldAutoContinueOnEmptyRun continues when an empty run is below the safety cap", () => {
expect(shouldAutoContinueOnEmptyRun(0, 0)).toBe(true);
expect(
shouldAutoContinueOnEmptyRun(0, MAX_CONSECUTIVE_EMPTY_RUN_LOADS - 1),
).toBe(true);
});
test("shouldAutoContinueOnEmptyRun stops once consecutive empty loads reach the cap", () => {
expect(shouldAutoContinueOnEmptyRun(0, MAX_CONSECUTIVE_EMPTY_RUN_LOADS)).toBe(
false,
);
expect(
shouldAutoContinueOnEmptyRun(0, MAX_CONSECUTIVE_EMPTY_RUN_LOADS + 1),
).toBe(false);
});
test("shouldAutoContinueOnEmptyRun honors a custom safety cap when provided", () => {
expect(shouldAutoContinueOnEmptyRun(0, 0, 1)).toBe(true);
expect(shouldAutoContinueOnEmptyRun(0, 1, 1)).toBe(false);
});
test("simulating auto-continue across empty runs skips empty contributions and lands on the next run with content (issue #3352 follow-up)", () => {
const runs = [
{ run_id: "R6" },
{ run_id: "R5" },
{ run_id: "R4" },
{ run_id: "R3" },
{ run_id: "R2" },
{ run_id: "R1" },
] as unknown as Run[];
const runIdToMessages: Record<string, Message[]> = {
R6: [{ id: "R6", type: "human", content: "F" } as Message],
R5: [{ id: "R5", type: "human", content: "E" } as Message],
R4: [],
R3: [],
R2: [],
R1: [{ id: "R1", type: "human", content: "A" } as Message],
};
const loaded = new Set<string>();
let messages: Message[] = [];
loaded.add("R6");
loaded.add("R5");
messages = [...runIdToMessages.R5!, ...runIdToMessages.R6!];
let consecutiveEmptyLoads = 0;
let visited = 0;
const visitedRunIds: string[] = [];
while (true) {
const index = findLatestUnloadedRunIndex(runs, loaded);
if (index === -1) break;
const run = runs[index]!;
visited += 1;
visitedRunIds.push(run.run_id);
const pageMessages = runIdToMessages[run.run_id] ?? [];
messages = [...pageMessages, ...messages];
loaded.add(run.run_id);
if (
!shouldAutoContinueOnEmptyRun(pageMessages.length, consecutiveEmptyLoads)
) {
consecutiveEmptyLoads = 0;
break;
}
consecutiveEmptyLoads += 1;
}
expect(visitedRunIds).toEqual(["R4", "R3", "R2", "R1"]);
expect(visited).toBe(4);
expect(messages.map((m) => m.content)).toEqual(["A", "E", "F"]);
});
test("shouldAutoContinueOnEmptyRun input must use the post-filter visible count, not the raw page size (middleware-only runs should still trigger auto-continue)", () => {
const filteredVisibleCount = 0;
const rawPageSize = 3; // pretend the raw page had 3 middleware-only entries
expect(shouldAutoContinueOnEmptyRun(filteredVisibleCount, 0)).toBe(true);
expect(shouldAutoContinueOnEmptyRun(rawPageSize, 0)).toBe(false);
});
// Regression coverage for #3825: after context summarization the backend emits
// RemoveMessage(ALL) + summary + retained, and onUpdateEvent rescues the removed
// messages into history via an async setState. The live thread.messages (an
// external store) and the archived history (React state) update through two
// independent scheduling channels, so a render can observe the post-summary
// (shrunk) thread while the rescued messages have NOT yet landed in
// visibleHistory. resolvePreservedHistory overlays a synchronous archive buffer
// so the merge never loses those messages regardless of the interleaving.
// messages into a current-stream transient bridge. The bridge fills only the
// journal flush/refetch gap and never mutates canonical history pages.
const summarizationHuman1 = {
id: "human-1",
@ -643,17 +664,15 @@ const summarizationMovedMessages = [
summarizationHuman2,
];
test("resolvePreservedHistory keeps rescued messages while history state is still stale (regression for #3825)", () => {
// visibleHistory has not yet absorbed the rescued messages (async setState
// from appendMessages is still pending in this render).
test("resolveTransientHistoryBridge keeps rescued messages while history state is stale", () => {
const staleHistory: Message[] = [];
expect(
resolvePreservedHistory(staleHistory, summarizationMovedMessages),
resolveTransientHistoryBridge(staleHistory, summarizationMovedMessages),
).toEqual(summarizationMovedMessages);
});
test("resolvePreservedHistory appends rescued messages after already-loaded history", () => {
test("resolveTransientHistoryBridge appends rescued messages after canonical history", () => {
const olderLoadedHuman = {
id: "older-human",
type: "human",
@ -661,24 +680,235 @@ test("resolvePreservedHistory appends rescued messages after already-loaded hist
} as Message;
expect(
resolvePreservedHistory([olderLoadedHuman], summarizationMovedMessages),
resolveTransientHistoryBridge(
[olderLoadedHuman],
summarizationMovedMessages,
),
).toEqual([olderLoadedHuman, ...summarizationMovedMessages]);
});
test("resolvePreservedHistory does not duplicate or reorder once history state catches up", () => {
// visibleHistory now contains the rescued messages (appendMessages committed),
// but the synchronous buffer still holds them this render.
test("resolveTransientHistoryBridge does not collapse an unloaded gap before its first canonical anchor", () => {
// Real regression shape from thread 4e81444d-c6ce-471e-93fd-b6ddb18dc938:
// the default history page starts at event seq=35, while the clarification
// conversation lives at seq=2..14. Context compression captured both the
// old turns and a later message that overlaps the canonical page. The old
// turns must stay suppressed until their canonical page loads; otherwise
// the unloaded seq=15..34 gap is visually collapsed before the page anchor.
const clarificationRequest = {
id: "clarification-request",
type: "ai",
content: "Which PDF should I create?",
} as Message;
const clarificationCard = {
id: "clarification-card",
tool_call_id: "clarification-call",
type: "tool",
content: "Create a new algorithm PDF",
} as Message;
const clarificationAnswer = {
id: "clarification-answer",
type: "human",
content: "Create a new algorithm PDF",
} as Message;
const directionQuestion = {
id: "direction-question",
type: "ai",
content: "Which topic?",
} as Message;
const directionAnswer = {
id: "direction-answer",
type: "human",
content: "Binary trees",
} as Message;
const pageAnchor = {
id: "event-seq-35",
type: "tool",
tool_call_id: "event-seq-35-call",
content: "first message on the latest history page",
} as Message;
const latestAnswer = {
id: "event-seq-88",
type: "ai",
content: "latest answer",
} as Message;
const captured = [
summarizationHuman1,
clarificationRequest,
clarificationCard,
clarificationAnswer,
directionQuestion,
directionAnswer,
pageAnchor,
];
const canonical = [pageAnchor, latestAnswer];
const missingAfterCanonicalRefetch = pruneConfirmedTransientMessages(
captured,
canonical,
);
const bridgeOrder = mergeTransientHistoryBridgeOrder([], captured);
expect(
resolvePreservedHistory(
resolveTransientHistoryBridge(
canonical,
missingAfterCanonicalRefetch,
bridgeOrder,
).map((message) => message.id),
).toEqual(["event-seq-35", "event-seq-88"]);
});
test("resolveTransientHistoryBridge does not duplicate once canonical history catches up", () => {
expect(
resolveTransientHistoryBridge(
summarizationMovedMessages,
summarizationMovedMessages,
),
).toEqual(summarizationMovedMessages);
});
test("resolvePreservedHistory returns history unchanged when nothing is pending archival", () => {
test("resolveTransientHistoryBridge returns history unchanged when the bridge is empty", () => {
const history = [summarizationHuman1, summarizationAi1];
expect(resolvePreservedHistory(history, [])).toBe(history);
expect(resolveTransientHistoryBridge(history, [])).toBe(history);
});
test("resolveThreadTransientHistoryBridge never leaks a bridge across threads", () => {
const canonical = [
{ id: "older-human", type: "human", content: "older" } as Message,
];
expect(
resolveThreadTransientHistoryBridge(
canonical,
summarizationMovedMessages,
"thread-a",
"thread-b",
),
).toBe(canonical);
expect(
resolveThreadTransientHistoryBridge(
canonical,
summarizationMovedMessages,
null,
null,
),
).toBe(canonical);
expect(
resolveThreadTransientHistoryBridge(
canonical,
summarizationMovedMessages,
"thread-a",
"thread-a",
),
).toEqual([canonical[0], ...summarizationMovedMessages]);
});
test("mergeTransientHistoryBridge preserves chronology across repeated compression", () => {
const human3 = {
id: "human-3",
type: "human",
content: "round 3 question",
} as Message;
const firstBridge = mergeTransientHistoryBridge(
[],
[summarizationHuman1, summarizationAi1],
);
const secondBridge = mergeTransientHistoryBridge(firstBridge, [
summarizationAi1,
summarizationHuman2,
human3,
]);
expect(secondBridge.map((message) => message.id)).toEqual([
"human-1",
"ai-1",
"human-2",
"human-3",
]);
});
test("mergeTransientHistoryBridge does not move a protected input recaptured by later compression", () => {
const protectedInput = {
id: "protected-input",
type: "human",
content: "写一个算法PDF",
} as Message;
const clarification = {
id: "clarification",
type: "ai",
content: "Which kind?",
} as Message;
const laterTail = {
id: "later-tail",
type: "ai",
content: "Working on the PDF",
} as Message;
const firstBridge = mergeTransientHistoryBridge(
[],
[protectedInput, clarification],
);
const secondBridge = mergeTransientHistoryBridge(firstBridge, [
{ ...protectedInput, content: [{ type: "text", text: "写一个算法PDF" }] },
laterTail,
]);
expect(secondBridge.map((message) => message.id)).toEqual([
"protected-input",
"clarification",
"later-tail",
]);
expect(secondBridge[0]?.content).toEqual([
{ type: "text", text: "写一个算法PDF" },
]);
});
test("mergeTransientHistoryBridgeOrder retains confirmed overlap as a non-rendering anchor", () => {
const firstOrder = mergeTransientHistoryBridgeOrder(
[],
[summarizationHuman1, summarizationAi1, summarizationHuman2],
);
const secondOrder = mergeTransientHistoryBridgeOrder(firstOrder, [
summarizationHuman2,
summarizationAi2,
]);
expect(secondOrder).toEqual([
"message:human-1",
"message:ai-1",
"message:human-2",
"message:ai-2",
]);
});
test("mergeTransientHistoryBridgeOrder keeps a recaptured protected prefix in place", () => {
const protectedInput = {
id: "protected-input",
type: "human",
content: "first",
} as Message;
const oldTail = {
id: "old-tail",
type: "ai",
content: "old",
} as Message;
const newTail = {
id: "new-tail",
type: "ai",
content: "new",
} as Message;
const firstOrder = mergeTransientHistoryBridgeOrder(
[],
[protectedInput, oldTail],
);
const secondOrder = mergeTransientHistoryBridgeOrder(firstOrder, [
protectedInput,
newTail,
]);
expect(secondOrder).toEqual([
"message:protected-input",
"message:old-tail",
"message:new-tail",
]);
});
test("merge keeps the full conversation across summarization even when visibleHistory lags (regression for #3825)", () => {
@ -694,7 +924,7 @@ test("merge keeps the full conversation across summarization even when visibleHi
// The bad render: visibleHistory is still empty, so without the buffer the
// rescued round-1/2 messages exist in neither merge input and are lost.
const effectiveHistory = resolvePreservedHistory(
const effectiveHistory = resolveTransientHistoryBridge(
[],
summarizationMovedMessages,
);
@ -709,23 +939,23 @@ test("merge keeps the full conversation across summarization even when visibleHi
]);
});
test("pruneConfirmedArchivedMessages drops messages history has absorbed but keeps the rest", () => {
test("pruneConfirmedTransientMessages drops canonical identities but keeps the rest", () => {
// History has caught up on the first two rescued messages only.
expect(
pruneConfirmedArchivedMessages(summarizationMovedMessages, [
pruneConfirmedTransientMessages(summarizationMovedMessages, [
summarizationHuman1,
summarizationAi1,
]),
).toEqual([summarizationHuman2]);
});
test("pruneConfirmedArchivedMessages keeps every pending message while history has not caught up", () => {
test("pruneConfirmedTransientMessages keeps entries while canonical history is stale", () => {
expect(
pruneConfirmedArchivedMessages(summarizationMovedMessages, []),
pruneConfirmedTransientMessages(summarizationMovedMessages, []),
).toEqual(summarizationMovedMessages);
});
test("resolvePreservedHistory prefers the live history copy over a stale buffered duplicate (#3825 review #3)", () => {
test("resolveTransientHistoryBridge prefers canonical copy over stale transient copy", () => {
// Same identity, but the buffered copy is an older snapshot. The live history
// copy (e.g. the finalized answer) must win — the buffer only fills gaps, it
// must never overwrite a message history already shows.
@ -740,12 +970,12 @@ test("resolvePreservedHistory prefers the live history copy over a stale buffere
content: "finalized answer",
} as Message;
expect(resolvePreservedHistory([liveFinal], [staleBuffered])).toEqual([
expect(resolveTransientHistoryBridge([liveFinal], [staleBuffered])).toEqual([
liveFinal,
]);
});
test("computeSummarizationMovedMessages returns the live turns dropped before the retained boundary (regression for #3825)", () => {
test("computeSummarizationTransientMessages captures live turns dropped before the retained boundary", () => {
const removeAll = {
id: "__remove_all__",
type: "remove",
@ -767,7 +997,7 @@ test("computeSummarizationMovedMessages returns the live turns dropped before th
const summarizationMessages = [removeAll, hiddenSummary, summarizationAi2];
expect(
computeSummarizationMovedMessages(
computeSummarizationTransientMessages(
liveThreadBeforeSummary,
summarizationMessages,
new Set([hiddenSummary.id!]),
@ -775,7 +1005,7 @@ test("computeSummarizationMovedMessages returns the live turns dropped before th
).toEqual([summarizationHuman1, summarizationAi1, summarizationHuman2]);
});
test("computeSummarizationMovedMessages excludes already-summarized control messages", () => {
test("computeSummarizationTransientMessages excludes already-summarized control messages", () => {
const priorSummary = {
id: "summary-0",
type: "human",
@ -799,9 +1029,9 @@ test("computeSummarizationMovedMessages excludes already-summarized control mess
summarizationAi2,
];
// priorSummary is in the summarized set, so it must not be re-archived.
// priorSummary is in the summarized set, so it must not enter the bridge.
expect(
computeSummarizationMovedMessages(
computeSummarizationTransientMessages(
liveThreadBeforeSummary,
summarizationMessages,
new Set([priorSummary.id!, "summary-1"]),
@ -812,7 +1042,7 @@ test("computeSummarizationMovedMessages excludes already-summarized control mess
test("full summarization rescue pipeline keeps the conversation when history state lags (regression for #3825)", () => {
// Exercises the whole rescue algorithm the hook runs: derive the moved
// messages, buffer them, then merge against the post-summary thread while the
// archived-history React state is still stale (empty).
// canonical run-event page is still stale (empty).
const removeAll = {
id: "__remove_all__",
type: "remove",
@ -832,7 +1062,7 @@ test("full summarization rescue pipeline keeps the conversation when history sta
];
const summarizationMessages = [removeAll, hiddenSummary, summarizationAi2];
const moved = computeSummarizationMovedMessages(
const moved = computeSummarizationTransientMessages(
liveThreadBeforeSummary,
summarizationMessages,
new Set([hiddenSummary.id!]),
@ -841,7 +1071,7 @@ test("full summarization rescue pipeline keeps the conversation when history sta
const postSummaryThread = [hiddenSummary, summarizationAi2];
const merged = mergeMessages(
resolvePreservedHistory(staleHistory, moved),
resolveTransientHistoryBridge(staleHistory, moved),
postSummaryThread,
[],
);
@ -854,3 +1084,18 @@ test("full summarization rescue pipeline keeps the conversation when history sta
"ai-2",
]);
});
test("refresh reconstructs the same 1-to-6 order from run events without a bridge", () => {
const canonical = Array.from({ length: 6 }, (_, index) => ({
id: `message-${index + 1}`,
type: index % 2 === 0 ? "human" : "ai",
content: String(index + 1),
})) as Message[];
const checkpointTail = canonical.slice(4);
expect(
mergeMessages(canonical, checkpointTail, []).map(
(message) => message.content,
),
).toEqual(["1", "2", "3", "4", "5", "6"]);
});