From 446fa03801dc4138ae5c74c5694cc5fc84482a84 Mon Sep 17 00:00:00 2001 From: AnoobFeng <109097565+AnoobFeng@users.noreply.github.com> Date: Tue, 14 Jul 2026 10:43:13 +0800 Subject: [PATCH] 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 --- backend/AGENTS.md | 5 +- backend/app/gateway/routers/thread_runs.py | 145 ++++ backend/app/gateway/services.py | 23 +- .../input_sanitization_middleware.py | 15 +- .../agents/middlewares/uploads_middleware.py | 10 +- .../deerflow/persistence/feedback/sql.py | 21 + .../harness/deerflow/persistence/run/sql.py | 37 + .../deerflow/runtime/events/store/base.py | 20 + .../deerflow/runtime/events/store/db.py | 30 + .../deerflow/runtime/events/store/jsonl.py | 16 +- .../deerflow/runtime/events/store/memory.py | 14 +- .../harness/deerflow/runtime/journal.py | 9 +- .../harness/deerflow/runtime/runs/manager.py | 66 ++ .../deerflow/runtime/runs/store/base.py | 23 + .../deerflow/runtime/runs/store/memory.py | 18 + .../harness/deerflow/utils/messages.py | 58 ++ .../blocking_io/test_jsonl_run_event_store.py | 4 +- backend/tests/test_feedback.py | 23 + backend/tests/test_gateway_services.py | 106 ++- backend/tests/test_history_batch_queries.py | 290 +++++++ .../test_input_sanitization_middleware.py | 35 + backend/tests/test_run_journal.py | 24 + backend/tests/test_thread_messages_page.py | 322 +++++++ .../test_uploads_middleware_core_logic.py | 17 + backend/tests/test_utils_messages.py | 63 +- frontend/AGENTS.md | 7 +- frontend/src/core/threads/hooks.ts | 739 ++++++++-------- frontend/tests/e2e/sidecar-chat.spec.ts | 12 +- .../tests/e2e/thread-history-mermaid.spec.ts | 65 +- frontend/tests/e2e/utils/mock-api.ts | 49 +- .../tests/unit/core/threads/infinite.test.ts | 18 + .../unit/core/threads/message-merge.test.ts | 793 ++++++++++++------ 32 files changed, 2373 insertions(+), 704 deletions(-) create mode 100644 backend/tests/test_history_batch_queries.py create mode 100644 backend/tests/test_thread_messages_page.py diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 818376d27..910c15647 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -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. ``) 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 (`...`, 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 `` 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. diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py index 4cbdb17d3..3439b05f0 100644 --- a/backend/app/gateway/routers/thread_runs.py +++ b/backend/app/gateway/routers/thread_runs.py @@ -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( diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index beae03cd8..27c677b02 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -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 dict→message 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 diff --git a/backend/packages/harness/deerflow/agents/middlewares/input_sanitization_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/input_sanitization_middleware.py index 1aa1072b1..ba3d544fc 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/input_sanitization_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/input_sanitization_middleware.py @@ -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, diff --git a/backend/packages/harness/deerflow/agents/middlewares/uploads_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/uploads_middleware.py index c5eb193ff..030f6eefa 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/uploads_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/uploads_middleware.py @@ -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}" diff --git a/backend/packages/harness/deerflow/persistence/feedback/sql.py b/backend/packages/harness/deerflow/persistence/feedback/sql.py index cdb5db89b..5cd03d0e9 100644 --- a/backend/packages/harness/deerflow/persistence/feedback/sql.py +++ b/backend/packages/harness/deerflow/persistence/feedback/sql.py @@ -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( diff --git a/backend/packages/harness/deerflow/persistence/run/sql.py b/backend/packages/harness/deerflow/persistence/run/sql.py index 3cb41aef1..11d44af1c 100644 --- a/backend/packages/harness/deerflow/persistence/run/sql.py +++ b/backend/packages/harness/deerflow/persistence/run/sql.py @@ -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: diff --git a/backend/packages/harness/deerflow/runtime/events/store/base.py b/backend/packages/harness/deerflow/runtime/events/store/base.py index 008a68e46..72757d33a 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/base.py +++ b/backend/packages/harness/deerflow/runtime/events/store/base.py @@ -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.""" diff --git a/backend/packages/harness/deerflow/runtime/events/store/db.py b/backend/packages/harness/deerflow/runtime/events/store/db.py index 719c45ddb..e365a976b 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/db.py +++ b/backend/packages/harness/deerflow/runtime/events/store/db.py @@ -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, diff --git a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py index 7116d318d..cbca8c8f5 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py +++ b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py @@ -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") diff --git a/backend/packages/harness/deerflow/runtime/events/store/memory.py b/backend/packages/harness/deerflow/runtime/events/store/memory.py index 573e3f546..da5ce7ba2 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/memory.py +++ b/backend/packages/harness/deerflow/runtime/events/store/memory.py @@ -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, [])) diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py index 8680f7653..7c2e357c4 100644 --- a/backend/packages/harness/deerflow/runtime/journal.py +++ b/backend/packages/harness/deerflow/runtime/journal.py @@ -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 diff --git a/backend/packages/harness/deerflow/runtime/runs/manager.py b/backend/packages/harness/deerflow/runtime/runs/manager.py index 5571001ed..0248b5024 100644 --- a/backend/packages/harness/deerflow/runtime/runs/manager.py +++ b/backend/packages/harness/deerflow/runtime/runs/manager.py @@ -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: diff --git a/backend/packages/harness/deerflow/runtime/runs/store/base.py b/backend/packages/harness/deerflow/runtime/runs/store/base.py index 0089640b2..f8420f8e3 100644 --- a/backend/packages/harness/deerflow/runtime/runs/store/base.py +++ b/backend/packages/harness/deerflow/runtime/runs/store/base.py @@ -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, diff --git a/backend/packages/harness/deerflow/runtime/runs/store/memory.py b/backend/packages/harness/deerflow/runtime/runs/store/memory.py index d597db922..97fc49e9f 100644 --- a/backend/packages/harness/deerflow/runtime/runs/store/memory.py +++ b/backend/packages/harness/deerflow/runtime/runs/store/memory.py @@ -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: diff --git a/backend/packages/harness/deerflow/utils/messages.py b/backend/packages/harness/deerflow/utils/messages.py index e54807467..8e873c039 100644 --- a/backend/packages/harness/deerflow/utils/messages.py +++ b/backend/packages/harness/deerflow/utils/messages.py @@ -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. diff --git a/backend/tests/blocking_io/test_jsonl_run_event_store.py b/backend/tests/blocking_io/test_jsonl_run_event_store.py index a7590de1a..15a78f054 100644 --- a/backend/tests/blocking_io/test_jsonl_run_event_store.py +++ b/backend/tests/blocking_io/test_jsonl_run_event_store.py @@ -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) diff --git a/backend/tests/test_feedback.py b/backend/tests/test_feedback.py index a592bdd22..d9a34fbf1 100644 --- a/backend/tests/test_feedback.py +++ b/backend/tests/test_feedback.py @@ -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 -- diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py index b5eab8aa7..8f90babab 100644 --- a/backend/tests/test_gateway_services.py +++ b/backend/tests/test_gateway_services.py @@ -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 diff --git a/backend/tests/test_history_batch_queries.py b/backend/tests/test_history_batch_queries.py new file mode 100644 index 000000000..0ea6696e7 --- /dev/null +++ b/backend/tests/test_history_batch_queries.py @@ -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"} diff --git a/backend/tests/test_input_sanitization_middleware.py b/backend/tests/test_input_sanitization_middleware.py index 25837c2b7..fefe55bc7 100644 --- a/backend/tests/test_input_sanitization_middleware.py +++ b/backend/tests/test_input_sanitization_middleware.py @@ -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) diff --git a/backend/tests/test_run_journal.py b/backend/tests/test_run_journal.py index 495341fd7..90d6615ab 100644 --- a/backend/tests/test_run_journal.py +++ b/backend/tests/test_run_journal.py @@ -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 diff --git a/backend/tests/test_thread_messages_page.py b/backend/tests/test_thread_messages_page.py new file mode 100644 index 000000000..89b1f9f17 --- /dev/null +++ b/backend/tests/test_thread_messages_page.py @@ -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 diff --git a/backend/tests/test_uploads_middleware_core_logic.py b/backend/tests/test_uploads_middleware_core_logic.py index 2f85f287b..01ceecf03 100644 --- a/backend/tests/test_uploads_middleware_core_logic.py +++ b/backend/tests/test_uploads_middleware_core_logic.py @@ -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("") + def test_uploaded_files_returned_in_state_update(self, tmp_path): mw = _middleware(tmp_path) uploads_dir = _uploads_dir(tmp_path) diff --git a/backend/tests/test_utils_messages.py b/backend/tests/test_utils_messages.py index 5251d5c07..2856a5cdc 100644 --- a/backend/tests/test_utils_messages.py +++ b/backend/tests/test_utils_messages.py @@ -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 diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index 62ac4f1c7..02e32d1b8 100644 --- a/frontend/AGENTS.md +++ b/frontend/AGENTS.md @@ -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 ` before normal chat submission, calling Gateway `GET/PUT/DELETE /api/threads/{thread_id}/goal`. Setting `/goal ` 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. diff --git a/frontend/src/core/threads/hooks.ts b/frontend/src/core/threads/hooks.ts index 28f789557..a0f96c0fd 100644 --- a/frontend/src/core/threads/hooks.ts +++ b/frontend/src/core/threads/hooks.ts @@ -229,26 +229,6 @@ function dedupeRunMessagesByIdentity(messages: RunMessage[]): RunMessage[] { }); } -export function getSupersededRunIds( - runs: Run[] | undefined, - pendingSupersededRunIds?: ReadonlySet, -) { - 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( values: ReadonlySet, itemsToRemove: Iterable, @@ -263,7 +243,6 @@ export function removeSetItems( export function buildVisibleHistoryMessages( messageRows: RunMessage[], supersededRunIds: ReadonlySet, - 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, -): 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(); 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(); + // 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(); + 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, @@ -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(); + const emittedMissingIdentities = new Set(); + 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(); + 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(), ); - // 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([]); - // 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([]); - // 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(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([]); + // 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([]); + const transientHistoryThreadIdRef = useRef(null); const summarizedRef = useRef>(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(); 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(null); - const loadedRunIdsRef = useRef>(new Set()); - const runBeforeSeqRef = useRef>(new Map()); - const loadGenerationRef = useRef(0); - const [loading, setLoading] = useState(false); - const [messageRows, setMessageRows] = useState([]); - const [appendedMessages, setAppendedMessages] = useState([]); + const historyQuery = useInfiniteQuery< + ThreadMessagesPageResponse, + Error, + InfiniteData, + ReturnType, + 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(), ); - }, [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, }; } diff --git a/frontend/tests/e2e/sidecar-chat.spec.ts b/frontend/tests/e2e/sidecar-chat.spec.ts index b7c44e943..c12b9263e 100644 --- a/frontend/tests/e2e/sidecar-chat.spec.ts +++ b/frontend/tests/e2e/sidecar-chat.spec.ts @@ -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, }), }); }, diff --git a/frontend/tests/e2e/thread-history-mermaid.spec.ts b/frontend/tests/e2e/thread-history-mermaid.spec.ts index 990b11991..254b90f6c 100644 --- a/frontend/tests/e2e/thread-history-mermaid.spec.ts +++ b/frontend/tests/e2e/thread-history-mermaid.spec.ts @@ -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}`); diff --git a/frontend/tests/e2e/utils/mock-api.ts b/frontend/tests/e2e/utils/mock-api.ts index ed1795aed..0e7ec8c39 100644 --- a/frontend/tests/e2e/utils/mock-api.ts +++ b/frontend/tests/e2e/utils/mock-api.ts @@ -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) => { diff --git a/frontend/tests/unit/core/threads/infinite.test.ts b/frontend/tests/unit/core/threads/infinite.test.ts index 43be4cc35..c06306473 100644 --- a/frontend/tests/unit/core/threads/infinite.test.ts +++ b/frontend/tests/unit/core/threads/infinite.test.ts @@ -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); diff --git a/frontend/tests/unit/core/threads/message-merge.test.ts b/frontend/tests/unit/core/threads/message-merge.test.ts index dcc05f2e5..470aa2c42 100644 --- a/frontend/tests/unit/core/threads/message-merge.test.ts +++ b/frontend/tests/unit/core/threads/message-merge.test.ts @@ -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 = []; + 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 = []; + 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 = { - R1: "A", - R2: "B", - R3: "C", - R4: "D", - R5: "E", - R6: "F", - }; - - const loaded = new Set(); - 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 = { - 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(); - 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"]); +});