diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 69724fc99..6f2c054ea 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -19,8 +19,8 @@ DeerFlow is a LangGraph-based AI super agent system with a full-stack architectu - Background subagent identity is deliberately split: the provider `tool_call_id` remains the correlation key for `ToolMessage`, `task_*` SSE events, persisted lifecycle events, frontend cards, and the public `ExtensionData.scope_id` contract (stored as `SubagentResult.external_task_id`), while `SubagentExecutor.execute_async()` generates a full server-side `execution_id` for `SubagentResult.task_id`, the process-wide registry, polling, cancellation, timeout handling, and cleanup. Provider IDs are not globally unique across parent runs, so they must never become registry ownership keys; scheduler closures retain their own `SubagentResult` rather than resolving ownership again through the mutable registry. Terminal subagent token usage travels in the current run's `ToolMessage.additional_kwargs` and is attributed from message state, never through a process-global provider-ID cache. - Scheduled-task executions must reuse that same Gateway run lifecycle. The scheduler may decide *when* work runs, but it must dispatch through the existing run path rather than introducing a parallel execution stack. Scheduled launches pass `scheduler.recursion_limit` (default 1000, matching the web UI's `recursion_limit: 1000`, clamped by `max_recursion_limit`) via `launch_scheduled_thread_run`; the value is read from `get_app_config()` at dispatch. - The background scheduler is single-instance by default. `scheduler.multi_instance=true` opts into lease-aware recovery across Gateway instances and requires shared Postgres, `run_ownership.heartbeat_enabled=true`, and `run_events.backend=db`; otherwise startup rejects the configuration. Live scheduled runs are preserved when a peer starts; expired launch claims return to the durable queue, expired run leases are atomically taken over, stale launch writes are fenced by lease ownership, and the Postgres advisory-locked budget makes `max_concurrent_runs` a shared global cap for `launching`/`running` rows. -- Long-running MCP work uses a separate durable task runtime rather than keeping remote task IDs or status polling inside the Agent loop. Explicit `task_toolsets` bind raw submit/status/cancel names; only submit remains Agent-visible, and its wrapper persists the remote handle before returning a local ID. `McpTaskService` claims due rows with leases, resolves a protocol-specific `McpTaskDriver`, and writes normalized snapshots back to `mcp_tasks`; expired leases are the restart-recovery mechanism, and a result returned after expiry or after a cancel request must be discarded even when the owner token still matches. The first cancel request fences an in-flight poll lease, while repeats preserve an active cancellation lease so they cannot issue concurrent remote cancels; cancellation backoff starts when the remote attempt finishes, so a slow timeout cannot consume the retry delay. Cancellation, polling, and notification batches isolate per-task exceptions; an unexpected cancellation/poll failure leaves that record's lease to expire, while notification failures release only the affected lease for retry. Input-required and terminal event snapshots are delivered by idempotent Agent runs and marked delivered only after run success; the trusted notification instruction stays outside the input boundary while the serialized remote event is framed as untrusted data. A busy-thread conflict is normalized back to the service boundary so the queued snapshot coalesces to the latest task event. A missing dispatched run becomes a failed delivery attempt, while transient run-store hydration errors stay distinguishable and retry the same lookup. The database is the source of truth; `ThreadState` receives only a bounded current-thread projection, and display names are neutralized at that model-state boundary. The installed process-local submitter is the source of truth for management-tool exposure; hot `mcp_tasks` edits take effect only after restart, and active skills must explicitly declare the list/cancel business tools. -- MCP notification failures use a consecutive counter separate from the idempotency-key `dispatch_attempt`, capped exponential backoff, latest-event rebuilding before a run launches, and a five-attempt budget before `dead_letter`. A permanently missing/mismatched target thread is dead-lettered immediately instead of being recreated or reclaimed. HTTP and Agent cancellation requests return after the durable cancel fence; the background loop alone owns the potentially slow remote call and retry schedule. The HTTP cancel endpoint rejects requests with 503 when the loop is not running (`mcp_tasks_available` false, e.g. `mcp_tasks.enabled=false` with SQL persistence), so a cancellation is never acknowledged without a worker to perform it. The bounded notification error/count/status join poll and cancellation diagnostics in the task detail API and expanded card. +- Long-running MCP work uses a separate durable task runtime (`McpTaskService` + `mcp_tasks`, lease-based recovery) rather than keeping remote task IDs or status polling inside the Agent loop; only submit remains Agent-visible, the database is the source of truth, and `ThreadState` receives only a bounded current-thread projection. Full contract (leases, cancellation fencing, delivery idempotency, management-tool exposure): [packages/harness/deerflow/mcp/AGENTS.md](packages/harness/deerflow/mcp/AGENTS.md). +- MCP task notification retries, dead-lettering, and the cancel endpoint's worker-stopped 503 are part of that same contract — see [packages/harness/deerflow/mcp/AGENTS.md](packages/harness/deerflow/mcp/AGENTS.md). - Scheduled-task dispatch enforces at most one non-terminal occurrence per task through `uq_scheduled_task_run_active` (`task_id WHERE status IN ('queued','launching','running')`). `queued` is durable and survives restart; `launching` carries a short owner/expiry lease and is the only state that may call the normal Gateway launch path; `running` is associated with the durable run. Each occurrence also supplies a stable run-admission idempotency key, so a recovered launch retry reuses the same durable run. A reused-thread `ConflictError` moves `launching` back to `queued`, while non-conflict launch errors become terminal `failed`. Waiting rows do not consume `max_concurrent_runs`; the atomic queue claim enforces the budget. Repeated triggers coalesce on the one active row, and same-thread FIFO treats older `queued`, `launching`, and `running` rows as blockers. The task definition stays immutable for all three active states because queue admission, PATCH/resume, pause, and delete serialize on the parent task row before touching the occurrence row. Pause/delete atomically interrupt existing `queued` rows and reject `launching`/`running` rows; PATCH/resume reject every active state, and mutation errors advertise pause cancellation only for `queued` work. A manual trigger may queue and run while the parent schedule remains paused. Recovery and multi-instance reconciliation lock task/run pairs in deterministic task-id/run-id order and must reconstruct `run_id`, `started_at`, and the live error state before releasing the short launch claim. Launch/failure/timeout bookkeeping changes the occurrence and its parent task in one parent-first transaction so a peer cannot claim the released task between those writes. Queue timeout marks the occurrence failed and advances a scheduled occurrence so it cannot immediately requeue forever; repository write boundaries coerce serialized task timestamps before binding SQL `DateTime` fields. - `extensions_config.json` is written at runtime by the Gateway (`PUT`/`PATCH /api/mcp/config`, the MCP enable switch, skill updates), so the production compose mounts it read-write while `config.yaml` stays `:ro`; Helm copies its ConfigMap seed into a writable home-volume directory before Gateway starts. Every read-modify-write holds both `extensions_config_write_lock` and the sidecar advisory `extensions_config_file_lock`, because the process-local lock alone loses updates across workers. Docker mounts the compose file as its own mount point, and Linux refuses `rename()` over a mount point with `EBUSY` even when the mount is writable — so `atomic_write_extensions_config` keeps the temp-file-plus-rename path and falls back to an in-place overwrite only on `EBUSY`. That fallback is deliberately non-atomic (a crash mid-write truncates the file); it exists because the alternative is a write that can never succeed, and only its first occurrence per target is logged at warning level. Any other `errno` still propagates. Pinned by `tests/test_compose_extensions_config_writable.py`, `tests/test_extensions_config_atomic_write.py`, and `tests/test_helm_extensions_config_writable.py`. diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index 252e8f3f2..d9accda26 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -75,6 +75,15 @@ metadata and checkpoints but skips local filesystem cleanup, so the raw value is never interpolated into a host path. New runs, workspace/sandbox operations, and other state-producing mutations remain blocked. +**Message feed seq** (#4666): streaming `values` frames, `GET +/threads/{id}/state`, and `POST /threads/{id}/history` stamp serialized +messages with `additional_kwargs.deerflow_seq` so clients can place +checkpoint-kept messages against the paged feed; the REST reads resolve the +store via `threads.py::_optional_run_event_store` (a feed-less deployment +still reads threads), and `services.py::normalize_input` strips the +server-owned key from client input (#4380). Mechanism and identity rule: +`packages/harness/deerflow/runtime/AGENTS.md`. + **Workspace change review**: `packages/harness/deerflow/workspace_changes/` captures a pre-run and post-run snapshot of the thread-owned `workspace` and `outputs` directories. `runtime/runs/worker.py` performs the filesystem scan via @@ -89,56 +98,18 @@ source of truth for both writers and the scanner), and the worker threads the configured `tool_output.storage_subdir` through the snapshot capture as an extra excluded dir name so custom storage locations stay excluded too. -**Run delivery receipts**: `RunJournal` records each non-empty artifact update -once per tool `Command` for the terminal `run.delivery` event. When a command -contains multiple messages, a unique tool name resolved from matching -`ToolMessage` entries supplies attribution; additional command messages do not -duplicate artifact paths or counts. If multiple different tool names resolve -for one flat artifact update, the paths remain counted but unattributed because -the command does not carry a per-path mapping. `RunJournal` callbacks set -`run_inline=True`: they do only in-memory bookkeeping or schedule async writes, -and staying on the run's event-loop thread serializes parallel tool callbacks -before terminal delivery recording and flushing. Each worker creates a separate -journal per run before cancellable/fallible preflight work, so checkpoint -compatibility failures and cancellation while waiting for prior finalization -still emit a zero-delivery receipt. The worker flushes ordinary journal events, -idempotently persists the run-scoped receipt, and only then persists the staged -terminal run status. A receipt failure is retried on a short bounded schedule -while the owning worker still knows the real outcome and holds the lease. The -worker derives delivery requirements from the run's workspace snapshots rather -than a client request option: every regular file created or modified under -`/mnt/user-data/outputs` is a candidate produced artifact. Internal -process-feedback files are not candidates: the snapshot capture excludes the -scanner's `EXCLUDED_DIR_NAMES` (including the default tool-output -externalization subdir) plus the configured `tool_output.storage_subdir`, so a -run that only externalized oversized tool outputs does not fail delivery. At -least one candidate must be covered by a path attributed by the journal to -`present_files`; presenting only an unrelated pre-existing path does not -satisfy delivery. -Receipts for such runs add `produced_paths`, `presented_paths`, `matched_paths`, -`verification`, `stage`, and `satisfied` to the Slice 1 fact fields. Missing a -matching presentation becomes a run error; a successful presentation is also -downgraded to error if its receipt cannot be durably verified. Runs without -changed outputs preserve ordinary chat behavior and the original receipt shape. -Orphan recovery first -atomically claims an expired lease, then uses the same singleton write to -backfill a zero-delivery receipt. This ordering prevents a stale recovery scan -from overwriting a live run's later detailed receipt; an event-store outage -does not undo the terminal takeover. An existing detailed receipt is preserved -when a worker crashed after writing it. Event stores -serialize `put_if_absent` with ordinary thread writers: memory and JSONL provide -the documented single-process guarantee, while the DB store adds per-thread -in-process locks and PostgreSQL advisory locks for cross-process writers. -Moving journal construction ahead of preflight is receipt-only on early failure -paths: a separate boundary flag preserves the previous completion-data -semantics, so checkpoint incompatibility or cancellation while waiting for an -older finalizing run does not persist an empty completion snapshot. Worker tests -pin one accumulated receipt across multiple goal-continuation `_stream_once` -calls; journal tests drive LangChain's real async callback dispatcher to pin -serialized, deduplicated parallel tool callbacks. -Multi-worker deployments therefore require `run_events.backend: db` for shared, -ordered delivery events; the startup gate rejects process-local memory and -JSONL event stores when `GATEWAY_WORKERS > 1`. +**Run delivery receipts**: the worker derives delivery requirements from the +run's workspace snapshots rather than a client request option (files +created/modified under `/mnt/user-data/outputs`, minus internal +process-feedback exclusions) and idempotently persists a run-scoped +`run.delivery` receipt before the terminal run status; missing or +unverifiable `present_files` coverage downgrades the run to error, while runs +without changed outputs keep ordinary chat behavior. Journal mechanics +(callback attribution, receipt idempotency and retries, orphan recovery): +`packages/harness/deerflow/runtime/AGENTS.md`. Multi-worker deployments +require `run_events.backend: db` for shared, ordered delivery events; the +startup gate rejects process-local memory and JSONL event stores when +`GATEWAY_WORKERS > 1`. **RunManager / RunStore contract**: - LangGraph-compatible run requests validate their supported subset before creating a run. `runtime/stream_modes.py` is the shared backend contract for public stream modes and the worker's `graph.astream` mapping; the public `messages-tuple` mode maps to LangGraph's internal `messages` mode, while public `messages`, `events`, and other unsupported modes are rejected instead of being dropped or replaced with `values`. `app/gateway/run_models.py::RunCreateRequest` is shared by HTTP and internal scheduled launch paths, retains only truthful compatibility defaults for unimplemented options (`if_not_exists="create"` plus `None` placeholders), returns 422 for unsupported values including `on_completion="complete"`, `on_completion="continue"`, and `multitask_strategy="enqueue"`, and forbids undeclared SDK options so fields such as `checkpoint_during` and `durability` cannot be silently discarded. A placeholder must still accept the stock SDK's own default: `langgraph_sdk` drops only `None` from its run payload, so `stream_resumable=False` reaches every request and means "non-resumable", which is what DeerFlow serves — rejecting it 422'd every IM channel run (#4466). `tests/test_run_request_validation.py::test_gateway_accepts_langgraph_sdk_default_payload` pins the real SDK payload against this boundary; channel tests mock the SDK client and cannot catch this class of drift. diff --git a/backend/app/gateway/routers/threads.py b/backend/app/gateway/routers/threads.py index 6278bbd7b..146fa3217 100644 --- a/backend/app/gateway/routers/threads.py +++ b/backend/app/gateway/routers/threads.py @@ -56,6 +56,7 @@ from deerflow.runtime.context_compaction import ( ThreadCompactionResult, compact_thread_context, ) +from deerflow.runtime.events.message_seq import stamp_messages_with_seq from deerflow.runtime.goal import ( DEFAULT_MAX_GOAL_CONTINUATIONS, build_goal_state, @@ -79,6 +80,17 @@ router = APIRouter(prefix="/api/threads", tags=["threads"]) _CHECKPOINT_MODE_ERRORS = (CheckpointModeMismatchError, CheckpointModeReconfigurationError) +def _optional_run_event_store(request: Request) -> Any: + """Return the run event store, or ``None`` when the app has none wired. + + Reads must not start depending on the feed: seq is placement metadata, and a + response without it degrades to the client's own ordering rule rather than + failing. ``get_run_event_store`` raises instead, which is right for the + endpoints that cannot work without a feed. + """ + return getattr(request.app.state, "run_event_store", None) + + def _checkpoint_mode_http_error(exc: Exception, thread_id: str) -> HTTPException: """Map checkpoint-mode guard failures to precise HTTP statuses. @@ -1326,8 +1338,15 @@ async def get_thread_state(thread_id: ThreadId, request: Request) -> ThreadState tasks_raw = snapshot.tasks or () tasks = [{"id": getattr(task, "id", ""), "name": getattr(task, "name", "")} for task in tasks_raw] + values = serialize_channel_values_for_api(snapshot.values) + messages = values.get("messages") + if isinstance(messages, list) and messages: + # Same reason as the history endpoint: a client reading the checkpoint + # over REST needs the feed position the stream would have stamped. + values["messages"] = await stamp_messages_with_seq(_optional_run_event_store(request), thread_id, messages) + return ThreadStateResponse( - values=serialize_channel_values_for_api(snapshot.values), + values=values, next=list(snapshot.next or ()), metadata=metadata, checkpoint={"id": checkpoint_id, "ts": coerce_iso(created_at)}, @@ -1718,7 +1737,15 @@ async def get_thread_history( except Exception: logger.warning("Failed to inject turn_duration for thread %s", sanitize_log_param(thread_id), exc_info=True) - values["messages"] = serialized_msgs + # The stream stamps `values` frames as they are published, but a + # client that only opens a conversation never sees one — this is + # the read it does instead, and without a seq a rescued early turn + # has no absolute position to be placed at (#4666). + values["messages"] = await stamp_messages_with_seq( + _optional_run_event_store(request), + thread_id, + serialized_msgs, + ) is_latest_checkpoint = False diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index 3aa656722..f5c6e44d1 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -66,6 +66,7 @@ from deerflow.runtime.checkpoint_mode import ( inject_checkpoint_mode, ) from deerflow.runtime.checkpoint_state import graph_state_schema +from deerflow.runtime.events.message_identity import MESSAGE_SEQ_KEY from deerflow.runtime.goal import goal_thread_lock from deerflow.runtime.journal import build_checkpoint_history_seed_events from deerflow.runtime.runs.naming import resolve_root_run_name @@ -120,6 +121,10 @@ _SERVER_OWNED_MESSAGE_METADATA_KEYS = ( TOOL_RECEIPT_KEY, TOOL_RECEIPT_LEDGER_KEY, TOOL_TRANSFORMS_KEY, + # Attached when a values frame is serialized, for display ordering only. + # A replayed message carrying it back would write a thread-scoped seq + # into the checkpoint, which a fork then re-seeds and reassigns (#4380). + MESSAGE_SEQ_KEY, SUBAGENT_TOOL_RECEIPTS_KEY, SUBAGENT_RECEIPT_VERDICT_KEY, SUBAGENT_ACCEPTANCE_VERDICT_KEY, diff --git a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md index 650eac8ae..714d21c33 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md @@ -102,4 +102,4 @@ Before changing a later authorization phase, read the [authorization RFC](../../ 32. **TerminalResponseMiddleware** - When a provider returns an empty terminal `AIMessage` after tool execution, injects a hidden recovery prompt and retries the model once; a second empty response is replaced in checkpoint state by a visible error fallback marked for the run worker, so the run finishes as an error instead of a silent success 33. **ModelLengthFinishReasonMiddleware** - Records `stop_reason=model_length_capped` when provider-specific length detectors match a terminal `AIMessage` without tool-call intent (`finish_reason=length` / `MAX_TOKENS`, or `stop_reason=max_tokens`), preserving the original assistant content and never reparsing textual tool-call-like envelopes 34. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after terminal-response/custom/configured middlewares so LangChain's reverse-order `after_model` dispatch runs it first -35. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, writes a readable `ToolMessage.content` fallback plus structured `ToolMessage.artifact.human_input` request payload, and interrupts via `Command(goto=END)` (must be last). In `after_model` it drops sibling tool calls from the same turn so they cannot execute before the user answers; a malformed `ask_clarification` that LangChain parked on `invalid_tool_calls` is the same stop signal (the valid sibling would otherwise still run). `disable_clarification` runs keep the siblings. Payloads are versioned: legacy modes (`free_text` / `choice_with_other`) keep `version: 1` unchanged, while the v2 `form` mode (from `fields`) carries `version: 2` so older frontends reject the payload and degrade to the plain-text fallback. Field normalization is deterministic and lives in the middleware, not the tool schema — the middleware short-circuits before tool execution, so tool-arg typing alone provides no runtime validation. Validation is atomic: any structurally broken entry (non-dict, bad/duplicate name, a name colliding with a JS `Object.prototype` member like `__proto__`/`constructor`, exceeding the caps of 16 fields / 24 options per field / 200 chars per text, or the whole normalized definition exceeding `MAX_FORM_SERIALIZED_BYTES` = 16KB UTF-8 — the per-item caps alone admit forms whose IM text fallback would blow channel delivery limits and truncate away trailing fields) degrades the whole form to the legacy option/free-text modes, so a card can never render "complete" while silently missing a business field; benign issues keep local degradation (unknown types — including unhashable JSON like `type: []`, which must never raise from the membership probe — and option-less selects become `text`), and options are trimmed/deduped with blanks dropped (both form-level and top-level) because the frontend parser rejects blank option labels. Model-produced XML-to-dict option payloads are recursively flattened from dict/list containers in source order, scalar string/number leaves are retained, and residual XML tags are removed before the same trimming and deduplication. Checkbox fields are booleans that default to an explicit "no"; `required` on a checkbox means must-agree/consent semantics. The response protocol is deliberately unchanged (v1 `text`/`option` only): form cards submit a readable text summary as `response_kind: "text"`, so journal persistence and answered-card recovery need no new allowlist entries. Because this middleware can short-circuit tool execution before LangChain emits `on_tool_end`, `RunJournal` performs a root-run final reconciliation for allowlisted clarification `ToolMessage`s whose `tool_call_id` was produced by the current run, so human-input request cards remain recoverable from `run_events` after checkpoint compaction. Human Input Card replies are submitted as `hide_from_ui` `HumanMessage`s with `additional_kwargs.human_input_response`; `RunJournal` persists only allowlisted hidden response sources (currently `ask_clarification`) as `llm.human.input`, which preserves answered-card state after compaction without exposing generic internal hidden context. +35. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, writes a readable `ToolMessage.content` fallback plus structured `ToolMessage.artifact.human_input` request payload, and interrupts via `Command(goto=END)` (must be last). In `after_model` it drops sibling tool calls from the same turn so they cannot execute before the user answers; a malformed `ask_clarification` that LangChain parked on `invalid_tool_calls` is the same stop signal (the valid sibling would otherwise still run). `disable_clarification` runs keep the siblings. Payloads are versioned: legacy modes (`free_text` / `choice_with_other`) keep `version: 1` unchanged, while the v2 `form` mode (from `fields`) carries `version: 2` so older frontends reject the payload and degrade to the plain-text fallback. Field normalization is deterministic and lives in the middleware, not the tool schema — the middleware short-circuits before tool execution, so tool-arg typing alone provides no runtime validation. Validation is atomic: any structurally broken entry (non-dict, bad/duplicate name, a name colliding with a JS `Object.prototype` member like `__proto__`/`constructor`, exceeding the caps of 16 fields / 24 options per field / 200 chars per text, or the whole normalized definition exceeding `MAX_FORM_SERIALIZED_BYTES` = 16KB UTF-8 — the per-item caps alone admit forms whose IM text fallback would blow channel delivery limits and truncate away trailing fields) degrades the whole form to the legacy option/free-text modes, so a card can never render "complete" while silently missing a business field; benign issues keep local degradation (unknown types — including unhashable JSON like `type: []`, which must never raise from the membership probe — and option-less selects become `text`), and options are trimmed/deduped with blanks dropped (both form-level and top-level) because the frontend parser rejects blank option labels. Model-produced XML-to-dict option payloads are recursively flattened from dict/list containers in source order, scalar string/number leaves are retained, and residual XML tags are removed before the same trimming and deduplication. Checkbox fields are booleans that default to an explicit "no"; `required` on a checkbox means must-agree/consent semantics. The response protocol is deliberately unchanged (v1 `text`/`option` only): form cards submit a readable text summary as `response_kind: "text"`, so journal persistence and answered-card recovery need no new allowlist entries. Because this middleware can short-circuit tool execution before LangChain emits `on_tool_end`, `RunJournal` performs a root-run final reconciliation for `ToolMessage`s whose `tool_call_id` was produced by the current run, so human-input request cards remain recoverable from `run_events` after checkpoint compaction. That reconciliation is **not** limited to `ask_clarification`: every middleware that answers a tool call itself has the same gap, and a result the user saw during the run must not disappear on reload (#4666 — `ReadBeforeWriteMiddleware`'s blocked-write errors were reaching the UI but never the event store). Its scope is bounded by three independent conditions instead of a tool-name allowlist — the message must be user-visible, the call must belong to this run's **lead agent** (`_remember_current_run_tool_calls` records lead-agent calls only, so subagent results stay in their own `subagent.step` feed), and it must not already be persisted. Keep those three; they are what makes a name allowlist unnecessary. Human Input Card replies are submitted as `hide_from_ui` `HumanMessage`s with `additional_kwargs.human_input_response`; `RunJournal` persists only allowlisted hidden response sources (currently `ask_clarification`) as `llm.human.input`, which preserves answered-card state after compaction without exposing generic internal hidden context. diff --git a/backend/packages/harness/deerflow/agents/middlewares/dynamic_context_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/dynamic_context_middleware.py index a69d88396..9cc98a7c5 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/dynamic_context_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/dynamic_context_middleware.py @@ -43,6 +43,7 @@ from langgraph.runtime import Runtime from deerflow.runtime.context_keys import CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY from deerflow.runtime.user_context import resolve_runtime_user_id +from deerflow.utils.messages import INJECTED_USER_MESSAGE_ID_SUFFIX, strip_injected_user_message_id_suffix if TYPE_CHECKING: from deerflow.config.app_config import AppConfig @@ -62,9 +63,17 @@ _DYNAMIC_CONTEXT_REMINDER_KEY = "dynamic_context_reminder" # so it is never exposed to user-influenceable memory content. _REMINDER_DATE_KEY = "reminder_date" _SUMMARY_MESSAGE_NAME = "summary" -# Suffix the ID-swap gives the real user message; the reminder SystemMessage -# takes the original id so ``add_messages`` can replace it in place. -INJECTED_USER_MESSAGE_ID_SUFFIX = "__user" + +# ``INJECTED_USER_MESSAGE_ID_SUFFIX`` / ``strip_injected_user_message_id_suffix`` +# are defined in ``deerflow.utils.messages`` and re-exported here, where the +# ID-swap they describe actually happens. Existing importers keep working. +__all__ = [ + "INJECTED_USER_MESSAGE_ID_SUFFIX", + "DynamicContextMiddleware", + "SubagentDateContextMiddleware", + "is_dynamic_context_reminder", + "strip_injected_user_message_id_suffix", +] def _format_current_date() -> str: @@ -81,20 +90,6 @@ def _format_current_date_reminder(current_date: str) -> str: ) -def strip_injected_user_message_id_suffix(message_id: str | None) -> str | None: - """Return the id *message_id* had before the reminder ID-swap. - - Replaying a persisted user turn must feed the graph the id the client - originally sent: a ``{id}__user`` message is skipped as an injection target, - so replaying one into a state that has no reminder yet silently drops the - date and memory block for that turn. - """ - - if isinstance(message_id, str) and message_id.endswith(INJECTED_USER_MESSAGE_ID_SUFFIX): - return message_id[: -len(INJECTED_USER_MESSAGE_ID_SUFFIX)] or message_id - return message_id - - def _extract_date(content: str) -> str | None: """Return the first value found in *content*, or None.""" m = _DATE_RE.search(content) diff --git a/backend/packages/harness/deerflow/mcp/AGENTS.md b/backend/packages/harness/deerflow/mcp/AGENTS.md index d0e316606..2e8fd793f 100644 --- a/backend/packages/harness/deerflow/mcp/AGENTS.md +++ b/backend/packages/harness/deerflow/mcp/AGENTS.md @@ -33,3 +33,8 @@ Verdicts are pinned against the real launchers: for npx, every argument vector the validator rejects is one `npx` actually executes, and every vector it allows is one `npx` passes through to the server. `env` screening covers names that execute code **unconditionally** at process startup, e.g. `PYTHONPATH`/`PYTHONHOME`, which run a caller-controlled `sitecustomize.py` at interpreter startup under plain `uvx`. Caller-controlled **search paths** are a weaker, conditional class and are an accepted residual: `LD_LIBRARY_PATH`/`DYLD_LIBRARY_PATH` (conditional on the process loading a shadowable library, and legitimately set by native-dependency servers) and `NODE_PATH` (searched *after* the local `node_modules` chain, so it cannot shadow an installed dependency, and ignored entirely by ESM `import` — it can only supply a CJS module that would otherwise fail to resolve). Do not move a search path into the set: it would make the "unconditional" rule untrue, which is how a defense-in-depth list starts being mistaken for a boundary. Remote transports skip all three — they spawn nothing. **This is defense in depth, not a trust boundary.** `npx`/`uvx` exist to fetch and execute remote packages, so an admin can still point one at a package they published; the boundary is admin authentication plus network reachability. Do not add a check here on the assumption that it makes MCP registration safe for untrusted admins — it does not, and the fix for that is not a bigger denylist. + +**Durable MCP task runtime** (`mcp_tasks`, `McpTaskService`; summarized in +[backend/AGENTS.md](../../../../AGENTS.md)): Long-running MCP work uses a separate durable task runtime rather than keeping remote task IDs or status polling inside the Agent loop. Explicit `task_toolsets` bind raw submit/status/cancel names; only submit remains Agent-visible, and its wrapper persists the remote handle before returning a local ID. `McpTaskService` claims due rows with leases, resolves a protocol-specific `McpTaskDriver`, and writes normalized snapshots back to `mcp_tasks`; expired leases are the restart-recovery mechanism, and a result returned after expiry or after a cancel request must be discarded even when the owner token still matches. The first cancel request fences an in-flight poll lease, while repeats preserve an active cancellation lease so they cannot issue concurrent remote cancels; cancellation backoff starts when the remote attempt finishes, so a slow timeout cannot consume the retry delay. Cancellation, polling, and notification batches isolate per-task exceptions; an unexpected cancellation/poll failure leaves that record's lease to expire, while notification failures release only the affected lease for retry. Input-required and terminal event snapshots are delivered by idempotent Agent runs and marked delivered only after run success; the trusted notification instruction stays outside the input boundary while the serialized remote event is framed as untrusted data. A busy-thread conflict is normalized back to the service boundary so the queued snapshot coalesces to the latest task event. A missing dispatched run becomes a failed delivery attempt, while transient run-store hydration errors stay distinguishable and retry the same lookup. The database is the source of truth; `ThreadState` receives only a bounded current-thread projection, and display names are neutralized at that model-state boundary. The installed process-local submitter is the source of truth for management-tool exposure; hot `mcp_tasks` edits take effect only after restart, and active skills must explicitly declare the list/cancel business tools. + +**Task notification failure handling**: MCP notification failures use a consecutive counter separate from the idempotency-key `dispatch_attempt`, capped exponential backoff, latest-event rebuilding before a run launches, and a five-attempt budget before `dead_letter`. A permanently missing/mismatched target thread is dead-lettered immediately instead of being recreated or reclaimed. HTTP and Agent cancellation requests return after the durable cancel fence; the background loop alone owns the potentially slow remote call and retry schedule. The HTTP cancel endpoint rejects requests with 503 when the loop is not running (`mcp_tasks_available` false, e.g. `mcp_tasks.enabled=false` with SQL persistence), so a cancellation is never acknowledged without a worker to perform it. The bounded notification error/count/status join poll and cancellation diagnostics in the task detail API and expanded card. diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 09a905e58..9d820eabc 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -24,6 +24,87 @@ Checkpointer storage runs in one of two channel modes, selected by `checkpoint_c **Run rollback flow** (`runtime/runs/worker.py`): `_capture_rollback_point` materializes the complete pre-run state via the accessor and captures raw `pending_writes` via `aget_tuple` into an immutable `RollbackPoint` before the run starts — capture failure disables rollback (fail-closed), never restores partial state. In `full` mode, cancel-with-rollback forks from the pre-run checkpoint via the mutation graph and inherits non-message channels from that parent. In `delta` mode, forking is unsafe once the cancelled path has attached sibling writes to the pre-run checkpoint, so rollback replaces every captured channel on the current head, using `Overwrite` for reducers and schema defaults for current-head-only channels. Both modes reattach only the captured pre-run pending writes to the restored checkpoint. Edit replay runs (`metadata.replay_kind="edit"`) also restore the pre-run checkpoint on failed, timed-out, or interrupted completion and publish the restored `values` snapshot to the stream before `end`, so clients do not remain on a transient edited branch when the replay did not produce a successful replacement. +**Message feed seq stamping** (#4666): a checkpoint carries no position of its +own and loses messages to summarization, so a client merging a `values` frame +with the seq-ordered `run_events` feed cannot place a checkpoint-kept message +once the feed's loaded page window no longer reaches back to it. +`RunEventStore.get_message_seqs(thread_id, identities)` resolves the seq the +store already assigned, keyed by +`runtime/events/message_identity.py::message_identity` — the backend half of +the identity rule `frontend/src/core/threads/hooks.ts::messageIdentity` applies +(tool messages by `tool_call_id`; `X` / `X__user` human copies collapse to +one). The two halves must stay in sync: a mismatch is silent, degrading +placement rather than raising. `runs/worker.py::_MessageSeqStamper` attaches +the result as `additional_kwargs.deerflow_seq` on root `values` frames only — +subgraph frames are not part of the thread feed's ordering, and nothing is +written back to the checkpoint. The run-scoped cache makes the compaction frame +the only one that costs a lookup, and the stamper soft-resolves the user id +once at build time — like the worker's write paths beside it — so a launch path +that never inherits the auth contextvar (a null-owner scheduled task) still +stamps instead of the db store's strict `AUTO` default raising per frame. A +resolved seq is cached for the run (earliest-seq-wins makes it final), but a +miss is not: a message this run produces reaches a frame before `RunJournal` +flushes it, so it misses and is persisted moments later. Misses are re-asked +when `RunJournal.feed_generation` — bumped once per successful event-store +write, never while the buffer merely fills — shows the feed gained rows, which +keeps the retry bounded by writes rather than by frames and makes a failed +lookup cost one generation instead of the run. REST +reads (`GET /threads/{id}/state`, `POST /threads/{id}/history`) stamp through +`events/message_seq.py::stamp_messages_with_seq`, the request-scoped +counterpart: everything a checkpoint still holds is already persisted, so one +batched lookup resolves the whole list. The db store prefilters candidate rows +in SQL (a LIKE clause per wanted raw id, wildcards escaped; an id `json.dumps` +would escape falls the set back to the full scan) so a wanted identity absent +from the feed — a message still streaming — does not force a full +fetch-and-decode of every message row's tool outputs on long threads. +`deerflow_seq` is server-owned display metadata: the gateway strips it from +client input, because a welded-in seq goes stale when a fork re-seeds the feed +(#4380). + +**Run delivery receipts** (`runtime/journal.py` + `runs/worker.py`): +`RunJournal` records each non-empty artifact update once per tool `Command` for +the terminal `run.delivery` event. When a command contains multiple messages, a +unique tool name resolved from matching `ToolMessage` entries supplies +attribution; additional command messages do not duplicate artifact paths or +counts. If multiple different tool names resolve for one flat artifact update, +the paths remain counted but unattributed because the command does not carry a +per-path mapping. `RunJournal` callbacks set `run_inline=True`: they do only +in-memory bookkeeping or schedule async writes, and staying on the run's +event-loop thread serializes parallel tool callbacks before terminal delivery +recording and flushing. Each worker creates a separate journal per run before +cancellable/fallible preflight work, so checkpoint compatibility failures and +cancellation while waiting for prior finalization still emit a zero-delivery +receipt. The worker flushes ordinary journal events, idempotently persists the +run-scoped receipt, and only then persists the staged terminal run status. A +receipt failure is retried on a short bounded schedule while the owning worker +still knows the real outcome and holds the lease. Delivery candidates are every +regular file created or modified under `/mnt/user-data/outputs`; internal +process-feedback files are excluded (the scanner's `EXCLUDED_DIR_NAMES` plus +the configured `tool_output.storage_subdir`), so a run that only externalized +oversized tool outputs does not fail delivery. At least one candidate must be +covered by a path attributed by the journal to `present_files`; presenting only +an unrelated pre-existing path does not satisfy delivery. Receipts for such +runs add `produced_paths`, `presented_paths`, `matched_paths`, `verification`, +`stage`, and `satisfied` to the Slice 1 fact fields. Missing a matching +presentation becomes a run error; a successful presentation is also downgraded +to error if its receipt cannot be durably verified. Runs without changed +outputs preserve ordinary chat behavior and the original receipt shape. Orphan +recovery first atomically claims an expired lease, then uses the same singleton +write to backfill a zero-delivery receipt — a stale recovery scan cannot +overwrite a live run's later detailed receipt, an event-store outage does not +undo the terminal takeover, and an existing detailed receipt is preserved when +a worker crashed after writing it. Event stores serialize `put_if_absent` with +ordinary thread writers: memory and JSONL provide the documented +single-process guarantee, while the DB store adds per-thread in-process locks +and PostgreSQL advisory locks for cross-process writers. Moving journal +construction ahead of preflight is receipt-only on early failure paths: a +separate boundary flag preserves the previous completion-data semantics, so +checkpoint incompatibility or cancellation while waiting for an older +finalizing run does not persist an empty completion snapshot. Worker tests pin +one accumulated receipt across multiple goal-continuation `_stream_once` calls; +journal tests drive LangChain's real async callback dispatcher against a single +journal to pin serialized, deduplicated parallel tool callbacks. + **Targeted run-event attribution** (`runtime/events/store/`): `RunEventStore.find_latest_ai_message_run_ids()` has a complete-or-error contract. Its default implementation walks `list_messages()` backward in diff --git a/backend/packages/harness/deerflow/runtime/events/message_identity.py b/backend/packages/harness/deerflow/runtime/events/message_identity.py new file mode 100644 index 000000000..1a05d8beb --- /dev/null +++ b/backend/packages/harness/deerflow/runtime/events/message_identity.py @@ -0,0 +1,60 @@ +"""Stable UI identity of a persisted message. + +The thread feed (``run_events``) and the checkpoint hold the same message under +the same id, so a client can align them — but only if both sides agree on what +"same message" means. This is the backend half of that rule; the frontend half +is ``messageIdentity`` in ``frontend/src/core/threads/hooks.ts``. The two must +stay in sync: a mismatch is silent, degrading placement rather than raising. + +Two normalizations matter: + +* a ``ToolMessage`` is identified by its ``tool_call_id``, not its own id — + that is the id both sides can always resolve; +* ``DynamicContextMiddleware`` re-keys the submitted user turn from ``X`` to + ``X__user`` (giving ``X`` to the injected reminder), so the two human copies + must collapse to one identity. +""" + +from collections.abc import Mapping +from typing import Any + +from deerflow.utils.messages import strip_injected_user_message_id_suffix + +__all__ = ["MESSAGE_SEQ_KEY", "attach_message_seq", "message_identity"] + +#: ``additional_kwargs`` key carrying a message's thread-feed seq to clients. +#: Server-owned display metadata: it is attached when a frame is serialized and +#: must be stripped from anything a client sends back, or a replayed message +#: would write it into the checkpoint (where a fork re-seeds and reassigns seq). +MESSAGE_SEQ_KEY = "deerflow_seq" + + +def message_identity(message: Mapping[str, Any]) -> str | None: + """Return the stable identity of *message*, or ``None`` if it has none. + + *message* is a serialized message mapping (as stored in ``run_events`` + content or carried in a checkpoint ``values`` frame), not a ``BaseMessage``. + """ + tool_call_id = message.get("tool_call_id") + if isinstance(tool_call_id, str) and tool_call_id: + return f"tool:{tool_call_id}" + + message_id = message.get("id") + if not isinstance(message_id, str) or not message_id: + return None + + # Only human copies collapse: a hidden SystemMessage legitimately reuses the + # original id, and merging it with the visible turn would hide the turn. + if message.get("type") == "human": + message_id = strip_injected_user_message_id_suffix(message_id) or message_id + return f"message:{message_id}" + + +def attach_message_seq(message: Mapping[str, Any], seq: int) -> dict[str, Any]: + """Return a shallow copy of *message* with *seq* under ``MESSAGE_SEQ_KEY``. + + The one stamping expression shared by the worker's run-scoped stamper and + the request-scoped ``stamp_messages_with_seq``, so the two counterparts of + the same rule cannot silently diverge. The input is never mutated. + """ + return {**message, "additional_kwargs": {**(message.get("additional_kwargs") or {}), MESSAGE_SEQ_KEY: seq}} diff --git a/backend/packages/harness/deerflow/runtime/events/message_seq.py b/backend/packages/harness/deerflow/runtime/events/message_seq.py new file mode 100644 index 000000000..4dcd79d6f --- /dev/null +++ b/backend/packages/harness/deerflow/runtime/events/message_seq.py @@ -0,0 +1,52 @@ +"""Attach the thread-global feed seq to serialized checkpoint messages. + +The checkpoint holds no seq of its own, so a client merging it with the +seq-ordered feed cannot place a message the loaded page window no longer +reaches back to. The streaming path solves this by stamping `values` frames as +they are published — but a client that merely *opens* a conversation never sees +a frame. It reads the checkpoint over REST, and without a seq there a +summarization-rescued early turn is placed by its nearest anchor instead, which +after compaction sits deep in the loaded page (#4666). + +This is the request-scoped counterpart of the worker's stamper: everything the +checkpoint still holds is already persisted, so one batched lookup resolves the +whole list and there is nothing to retry later. +""" + +from __future__ import annotations + +import logging +from collections.abc import Mapping, Sequence +from typing import Any + +from deerflow.runtime.events.message_identity import attach_message_seq, message_identity + +logger = logging.getLogger(__name__) + +__all__ = ["stamp_messages_with_seq"] + + +async def stamp_messages_with_seq(store: Any, thread_id: str, messages: Sequence[Any]) -> list[Any]: + """Return *messages* with ``MESSAGE_SEQ_KEY`` attached where the feed knows one. + + The input is never mutated: entries that gain a seq are shallow-copied, and + everything else is passed through as-is. A missing store, an entry that is + not a mapping, an identity the feed does not know, and a failing lookup all + degrade to "no seq" rather than raising — placement is an enhancement, and a + client without it falls back to its own ordering rule. + """ + if store is None or not messages: + return list(messages) + + identities = [message_identity(m) if isinstance(m, Mapping) else None for m in messages] + wanted = {identity for identity in identities if identity is not None} + if not wanted: + return list(messages) + + try: + found = await store.get_message_seqs(thread_id, sorted(wanted)) + except Exception: + logger.warning("Failed to resolve message seqs for thread %s", thread_id, exc_info=True) + return list(messages) + + return [attach_message_seq(message, seq) if identity is not None and (seq := found.get(identity)) is not None else message for message, identity in zip(messages, identities, strict=True)] diff --git a/backend/packages/harness/deerflow/runtime/events/store/base.py b/backend/packages/harness/deerflow/runtime/events/store/base.py index 8cf4ee8bb..e5f9a895c 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/base.py +++ b/backend/packages/harness/deerflow/runtime/events/store/base.py @@ -13,6 +13,7 @@ Implementations: from __future__ import annotations import abc +from collections.abc import Sequence from deerflow.runtime.user_context import AUTO, _AutoSentinel @@ -239,6 +240,32 @@ class RunEventStore(abc.ABC): async def count_messages(self, thread_id: str) -> int: """Count displayable messages (category=message) in a thread.""" + @abc.abstractmethod + async def get_message_seqs( + self, + thread_id: str, + identities: Sequence[str], + *, + user_id: str | None | _AutoSentinel = AUTO, + ) -> dict[str, int]: + """Return ``{identity: seq}`` for messages already persisted in this thread. + + A checkpoint carries no seq of its own and loses messages to + summarization, so a client merging a checkpoint frame with this + seq-ordered feed cannot place a surviving old message once the feed's + loaded page window no longer reaches back to it (#4666). The seq already + exists here; this exposes it without paging the whole feed. + + *identities* are the values produced by + ``deerflow.runtime.events.message_identity.message_identity`` — the same + rule the frontend applies — so both sides agree on what "same message" + means. Identities that are not persisted (or not `category="message"`) + are simply absent from the result: callers degrade to their own + placement rule rather than treating a miss as an error. When one + identity resolves to several rows, the earliest seq wins, so a message + re-persisted later keeps the position it first occupied. + """ + @abc.abstractmethod async def delete_by_thread(self, thread_id: str) -> int: """Delete all events for a thread. Return the number of deleted events.""" diff --git a/backend/packages/harness/deerflow/runtime/events/store/db.py b/backend/packages/harness/deerflow/runtime/events/store/db.py index 2227fd725..6736d8bc3 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/db.py +++ b/backend/packages/harness/deerflow/runtime/events/store/db.py @@ -9,13 +9,15 @@ from __future__ import annotations import asyncio import json import logging +import re from datetime import UTC, datetime from typing import Any -from sqlalchemy import delete, func, select, text +from sqlalchemy import delete, func, or_, select, text from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from deerflow.persistence.models.run_event import RunEventRow +from deerflow.runtime.events.message_identity import message_identity from deerflow.runtime.events.store.base import RunEventStore from deerflow.runtime.user_context import AUTO, _AutoSentinel, get_current_user, resolve_user_id from deerflow.utils.time import coerce_iso @@ -104,6 +106,30 @@ class DbRunEventStore(RunEventStore): user = get_current_user() return str(user.id) if user is not None else None + #: Characters json.dumps escapes in the stored content (``ensure_ascii`` + #: is False, so non-ASCII survives verbatim and stays matchable). + _LIKE_UNSAFE_ID = re.compile(r'["\\\x00-\x1f]') + + @classmethod + def _prefilter_substrings(cls, wanted: set[str]) -> list[str] | None: + """Return the raw ids to LIKE-match in ``content``, or ``None`` to full-scan. + + An identity is ``kind:raw_id`` and the raw id appears verbatim in the + stored JSON string (``u1`` is a substring of a re-keyed ``u1__user`` + copy too), so a row not containing any wanted id cannot resolve any + wanted identity. An id json.dumps would escape breaks that verbatim + guarantee — one such id falls the whole set back to the full scan + rather than silently missing it. LIKE wildcards are escaped, not + rejected. + """ + ids = [] + for identity in wanted: + _kind, _sep, raw_id = identity.partition(":") + if not raw_id or cls._LIKE_UNSAFE_ID.search(raw_id): + return None + ids.append(raw_id) + return ids + @staticmethod async def _max_seq_for_thread(session: AsyncSession, thread_id: str) -> int | None: """Return the current max seq while serializing writers per thread. @@ -388,6 +414,62 @@ class DbRunEventStore(RunEventStore): async with self._sf() as session: return await session.scalar(stmt) or 0 + async def get_message_seqs( + self, + thread_id, + identities, + *, + user_id: str | None | _AutoSentinel = AUTO, + ): + wanted = set(identities) + if not wanted: + return {} + resolved_user_id = resolve_user_id(user_id, method_name="DbRunEventStore.get_message_seqs") + # ``content`` is a TEXT column holding a JSON *string* (see + # ``_content_to_db``), not a JSON column, so the identity fields cannot + # be projected in SQL — matching rows are decoded here instead. The + # ``content`` column carries full tool outputs, and a wanted identity + # absent from the feed (a message still streaming) defeats the early + # exit below — so without a prefilter a `/state`/`/history` read of a + # long thread pays a full fetch-and-decode of every message row. The + # LIKE prefilter keeps that cost in SQL: only rows containing a wanted + # id as a raw substring are fetched (false positives are re-checked by + # ``message_identity``; ids the prefilter cannot express fall back to + # the full scan). + stmt = select(RunEventRow.seq, RunEventRow.content).where(RunEventRow.thread_id == thread_id, RunEventRow.category == "message").order_by(RunEventRow.seq) + if resolved_user_id is not None: + stmt = stmt.where(RunEventRow.user_id == resolved_user_id) + prefilter_ids = self._prefilter_substrings(wanted) + if prefilter_ids is not None: + stmt = stmt.where(or_(*[RunEventRow.content.like(f"%{i.replace('%', '\\%').replace('_', '\\_')}%", escape="\\") for i in prefilter_ids])) + + found: dict[str, int] = {} + async with self._sf() as session: + result = await session.execute(stmt) + for seq, raw in result: + # Plain-text content (never a message dict) is skipped without + # paying for a failed JSON parse. + if not isinstance(raw, str) or not raw.startswith("{"): + continue + try: + content = json.loads(raw) + except (json.JSONDecodeError, ValueError): + continue + if not isinstance(content, dict): + continue + identity = message_identity(content) + # Earliest seq wins: a message re-persisted later keeps the + # position it first occupied in the feed. + if identity in wanted and identity not in found: + found[identity] = seq + # Later rows can only be re-persisted copies that already + # lose that tiebreak, so the scan (and its JSON decoding) + # ends with the last wanted seq instead of the thread's + # full message count. + if len(found) == len(wanted): + break + return found + async def delete_by_thread( self, thread_id, diff --git a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py index 9185adbfd..6d1a642c2 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py +++ b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py @@ -30,6 +30,7 @@ from datetime import UTC, datetime from pathlib import Path from typing import Any +from deerflow.runtime.events.message_identity import message_identity from deerflow.runtime.events.store.base import RunEventStore, match_ai_message_run_id, normalize_message_ids from deerflow.runtime.user_context import AUTO, _AutoSentinel from deerflow.utils.thread_id import validate_thread_id @@ -347,6 +348,29 @@ class JsonlRunEventStore(RunEventStore): all_events = await asyncio.to_thread(self._read_thread_events, thread_id) return sum(1 for e in all_events if e.get("category") == "message") + async def get_message_seqs(self, thread_id, identities, *, user_id: str | None | _AutoSentinel = AUTO): + wanted = set(identities) + if not wanted: + return {} + all_events = await asyncio.to_thread(self._read_thread_events, thread_id) + found: dict[str, int] = {} + for event in all_events: + if event.get("category") != "message": + continue + content = event.get("content") + if not isinstance(content, dict): + continue + identity = message_identity(content) + # Earliest seq wins: a message re-persisted later keeps the position + # it first occupied in the feed. + if identity in wanted and identity not in found: + found[identity] = event["seq"] + # Later events can only be re-persisted copies that already lose + # that tiebreak, so the scan ends with the last wanted seq. + if len(found) == len(wanted): + break + return found + async def delete_by_thread(self, thread_id): async with self._get_write_lock(thread_id): all_events = await asyncio.to_thread(self._read_thread_events, thread_id) diff --git a/backend/packages/harness/deerflow/runtime/events/store/memory.py b/backend/packages/harness/deerflow/runtime/events/store/memory.py index 113792560..d356e28f3 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/memory.py +++ b/backend/packages/harness/deerflow/runtime/events/store/memory.py @@ -9,6 +9,7 @@ from __future__ import annotations import bisect from datetime import UTC, datetime +from deerflow.runtime.events.message_identity import message_identity from deerflow.runtime.events.store.base import RunEventStore from deerflow.runtime.user_context import AUTO, _AutoSentinel @@ -180,6 +181,26 @@ class MemoryRunEventStore(RunEventStore): async def count_messages(self, thread_id): return len(self._messages.get(thread_id, [])) + async def get_message_seqs(self, thread_id, identities, *, user_id: str | None | _AutoSentinel = AUTO): + wanted = set(identities) + if not wanted: + return {} + found: dict[str, int] = {} + for record in self._messages.get(thread_id, []): + content = record.get("content") + if not isinstance(content, dict): + continue + identity = message_identity(content) + # Earliest seq wins: a message replaced later in the same thread + # keeps the position it first occupied in the feed. + if identity in wanted and identity not in found: + found[identity] = record["seq"] + # Later rows can only be re-persisted copies that already lose + # that tiebreak, so the scan ends with the last wanted seq. + if len(found) == len(wanted): + break + return found + async def delete_by_thread(self, thread_id): events = self._events.pop(thread_id, []) self._messages.pop(thread_id, None) diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py index 9bf7eae46..f4e92b701 100644 --- a/backend/packages/harness/deerflow/runtime/journal.py +++ b/backend/packages/harness/deerflow/runtime/journal.py @@ -49,7 +49,6 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) _LEGACY_SUMMARY_MESSAGE_NAME = "summary" -_RECONCILED_TOOL_MESSAGE_NAMES = frozenset({"ask_clarification"}) _PERSISTED_HIDDEN_HUMAN_INPUT_RESPONSE_SOURCES = frozenset({"ask_clarification"}) @@ -286,6 +285,12 @@ class RunJournal(BaseCallbackHandler): self._active_tool_names: dict[str, str] = {} self._persisted_tool_message_identities: set[str] = set() + # Bumped once per successful event-store write. A reader that cached a + # "the feed does not hold this message" answer compares this between + # reads to learn whether retrying could produce a different one, + # without polling the store (#4696 review). + self._feed_generation = 0 + # Artifact-production tracking for the terminal run.delivery event # (#4272 slice 1). Deduped by (path, tool_name); insertion order kept. self._produced_artifacts: list[tuple[str, str | None]] = [] @@ -621,16 +626,27 @@ class RunJournal(BaseCallbackHandler): return [] def _should_reconcile_tool_message(self, message: ToolMessage) -> bool: + """Whether a final-output ToolMessage still needs persisting. + + A middleware can answer a tool call itself and short-circuit execution, + so LangChain never emits ``on_tool_end`` and the result never reaches + the event store. The user saw that result during the run, and it + disappeared on reload (#4666). Any such result is reconciled here; the + scope is bounded by three independent conditions rather than a tool-name + allowlist: it must be user-visible, the call must belong to this run's + lead agent (``_remember_current_run_tool_calls`` records lead-agent + calls only, so subagent results stay in their own step feed), and it + must not already be persisted. + """ if message.additional_kwargs.get("hide_from_ui") is True: return False tool_call_id = getattr(message, "tool_call_id", None) if not isinstance(tool_call_id, str) or not tool_call_id: return False - tool_call_name = self._current_run_tool_call_names.get(tool_call_id) - if tool_call_name is None: - return False - message_name = getattr(message, "name", None) - if message_name not in _RECONCILED_TOOL_MESSAGE_NAMES and tool_call_name not in _RECONCILED_TOOL_MESSAGE_NAMES: + # The call must belong to this run: a retained ToolMessage from an + # earlier run is already persisted under its own run and must not be + # re-attributed here. + if self._current_run_tool_call_names.get(tool_call_id) is None: return False identity = self._message_identity(message) return identity is not None and identity not in self._persisted_tool_message_identities @@ -690,6 +706,7 @@ class RunJournal(BaseCallbackHandler): if store is None: return await store.put_batch(batch) + self._feed_generation += 1 except Exception: logger.warning( "Failed to flush %d events for run %s — returning to buffer", @@ -929,6 +946,7 @@ class RunJournal(BaseCallbackHandler): if store is None: return await store.put_batch(batch) + self._feed_generation += 1 except Exception: self._buffer = batch + self._buffer raise @@ -1056,6 +1074,18 @@ class RunJournal(BaseCallbackHandler): "first_human_message": self._first_human_msg, } + @property + def feed_generation(self) -> int: + """Monotonic count of successful writes to the thread feed. + + Buffered events are not in the feed yet, so a lookup for a message this + run just produced legitimately misses. This counter is what tells such + a reader that its cached miss is worth re-asking — it changes exactly + when the feed gained rows, and never while the buffer is merely + filling. A failed write leaves it alone: nothing became readable. + """ + return self._feed_generation + @property def had_llm_error_fallback(self) -> bool: return self._had_llm_error_fallback diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index 7fba8d692..d48e65439 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -25,7 +25,7 @@ import sys import threading import time import weakref -from collections.abc import AsyncIterator, Coroutine +from collections.abc import AsyncIterator, Callable, Coroutine, Mapping from contextlib import asynccontextmanager from contextvars import Context from dataclasses import dataclass, field @@ -53,6 +53,7 @@ from deerflow.runtime.checkpoint_state import ( graph_writable_channels, ) from deerflow.runtime.context_keys import CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY +from deerflow.runtime.events.message_identity import attach_message_seq, message_identity from deerflow.runtime.goal import ( DEFAULT_MAX_GOAL_CONTINUATIONS, DEFAULT_MAX_NO_PROGRESS_CONTINUATIONS, @@ -75,7 +76,7 @@ from deerflow.runtime.goal import ( from deerflow.runtime.serialization import serialize from deerflow.runtime.stream_bridge import StreamBridge from deerflow.runtime.stream_modes import normalize_stream_modes, to_langgraph_stream_modes -from deerflow.runtime.user_context import get_effective_user_id, resolve_runtime_user_id +from deerflow.runtime.user_context import get_current_user, get_effective_user_id, resolve_runtime_user_id from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, ensure_trace_id from deerflow.tracing import inject_langfuse_metadata from deerflow.utils.messages import message_to_text @@ -1158,6 +1159,10 @@ async def run_agent( ) return goal_evaluator_model + # Built once per run, not per _stream_once call: goal continuations + # re-enter the stream and would otherwise discard the resolved seqs. + seq_stamper = _build_seq_stamper(event_store, thread_id, journal) if "values" in requested_modes else None + async def _stream_once(input_payload: Any, stream_config: RunnableConfig) -> None: nonlocal llm_error_fallback_message file_tool_chunk_batcher = _LargeFileToolChunkBatcher() if "values" in requested_modes else None @@ -1176,7 +1181,10 @@ async def run_agent( break llm_error_fallback_message = llm_error_fallback_message or _extract_llm_error_fallback_message(chunk, pre_existing_message_ids) sse_event = _lg_mode_to_sse_event(single_mode) - await bridge.publish(run_id, sse_event, serialize(chunk, mode=single_mode)) + single_payload = serialize(chunk, mode=single_mode) + if single_mode == "values" and seq_stamper is not None: + single_payload = await seq_stamper.stamp(single_payload) + await bridge.publish(run_id, sse_event, single_payload) if single_mode == "custom": await subagent_events.add(chunk) finally: @@ -1224,6 +1232,7 @@ async def run_agent( namespace=namespace, file_tool_chunk_batcher=file_tool_chunk_batcher, subagent_events=subagent_events, + seq_stamper=seq_stamper, ) finally: close_error = sys.exception() @@ -2734,6 +2743,107 @@ def _compose_sse_event(sse_event: str, namespace: tuple[str, ...]) -> str: return "|".join((sse_event, *namespace)) +#: Sentinel generation for an identity no lookup has missed at yet. Below every +#: real generation, so it never reads as "already asked at this generation". +_NEVER_LOOKED_UP = -1 + + +def _NO_FEED_WRITES() -> int: # noqa: N802 — a callable constant, not a class + """Generation source for a path with no journal: the feed never moves.""" + return 0 + + +class _MessageSeqStamper: + """Attach each already-persisted message's feed seq to a ``values`` frame. + + A checkpoint carries no seq of its own and loses messages to summarization, + so a client merging it with the seq-ordered thread feed cannot place a + surviving old message once the feed's loaded page window no longer reaches + back to it (#4666). The seq exists in the event store keyed by message + identity; this carries it to the client and writes nothing back to the + checkpoint. + + Cost is bounded to frames that introduce identities it has not resolved + yet: a resolved seq is final (the feed's earliest-seq-wins rule), so it is + never looked up twice, and in a real run the only frame that pays for a + query is the one where compaction brings older messages back into view. + + A miss, unlike a hit, is only provisional. A message this run produces + reaches a frame before ``RunJournal`` flushes it, so it legitimately misses + and would stay unstamped for the whole run if that answer were kept — the + long run that then rolls past the history page and compacts is exactly the + case this stamper exists for. Misses are therefore re-asked, but only once + the feed has actually gained rows, which *feed_generation* reports; frames + alone never trigger a retry. A failed lookup is a miss under the same rule, + so a transient store error costs one generation, not the run. + + An unstamped message needs no seq while it is still streaming: appending it + at the tail is already its correct position. + """ + + __slots__ = ("_store", "_thread_id", "_user_id", "_seqs", "_missing", "_feed_generation") + + def __init__(self, event_store: Any, thread_id: str, *, feed_generation: Callable[[], int] | None = None) -> None: + self._store = event_store + self._thread_id = thread_id + # Without a generation source (no journal on this path) nothing can + # report a feed write, so a miss stays cached rather than being re-asked + # on every frame: a constant reads as "the feed has not moved". + self._feed_generation = feed_generation if feed_generation is not None else _NO_FEED_WRITES + # Soft-resolved once at build time, like the worker's write paths: a + # launch path that never inherits the auth contextvar (e.g. a + # null-owner scheduled task) writes rows with no user_id, so the + # lookup filters by the same id the writes stamped — or not at all — + # instead of the db store's strict AUTO default raising per frame. + user = get_current_user() + self._user_id: str | None = str(user.id) if user is not None else None + self._seqs: dict[str, int] = {} + # identity -> the feed generation its last lookup missed at. + self._missing: dict[str, int] = {} + + async def stamp(self, payload: Any) -> Any: + if self._store is None or not isinstance(payload, Mapping): + return payload + messages = payload.get("messages") + if not isinstance(messages, list) or not messages: + return payload + + identities = [message_identity(m) if isinstance(m, Mapping) else None for m in messages] + # Read before the lookup, never after: a write landing mid-query then + # advances past the generation the miss is recorded under, so the next + # frame re-asks. The reverse order could bury such a write. + generation = self._feed_generation() + unresolved = {i for i in identities if i is not None and i not in self._seqs and self._missing.get(i, _NEVER_LOOKED_UP) != generation} + if unresolved: + try: + found = await self._store.get_message_seqs(self._thread_id, sorted(unresolved), user_id=self._user_id) + except Exception: + # Placement is an enhancement: a client without seq falls back + # to its own ordering rule. Never fail the frame over it. + logger.warning("Failed to resolve message seqs for thread %s", self._thread_id, exc_info=True) + found = {} + self._seqs.update(found) + self._missing.update(dict.fromkeys(unresolved - found.keys(), generation)) + + stamped = [attach_message_seq(message, seq) if identity is not None and (seq := self._seqs.get(identity)) is not None else message for message, identity in zip(messages, identities, strict=True)] + return {**payload, "messages": stamped} + + +def _build_seq_stamper(event_store: Any, thread_id: str, journal: Any) -> _MessageSeqStamper: + """Build the run's stamper, reading feed writes from *journal*. + + The journal owns the writes that turn a lookup miss into a hit, so it is + also what can tell the stamper that a cached miss is worth re-asking. A run + without one has no writer to report, and the stamper falls back to keeping + its misses. + """ + return _MessageSeqStamper( + event_store, + thread_id, + feed_generation=(lambda: journal.feed_generation) if journal is not None else None, + ) + + async def _publish_stream_item( *, bridge: Any, @@ -2743,6 +2853,7 @@ async def _publish_stream_item( namespace: tuple[str, ...], file_tool_chunk_batcher: Any, subagent_events: Any, + seq_stamper: Any = None, ) -> None: """Publish one stream frame, preserving the subgraph namespace. @@ -2763,6 +2874,11 @@ async def _publish_stream_item( await bridge.publish(run_id, "messages", serialize(publish_chunk, mode="messages")) chunks_to_publish = file_tool_chunk_batcher.push(chunk) if mode == "messages" and file_tool_chunk_batcher is not None else [chunk] for publish_chunk in chunks_to_publish: - await bridge.publish(run_id, sse_event, serialize(publish_chunk, mode=mode)) + payload = serialize(publish_chunk, mode=mode) + if mode == "values" and seq_stamper is not None: + # Root frames only: a subagent's snapshot is not part of this + # thread's feed ordering (the namespaced branch returned above). + payload = await seq_stamper.stamp(payload) + await bridge.publish(run_id, sse_event, payload) if mode == "custom": await subagent_events.add(chunk) diff --git a/backend/packages/harness/deerflow/utils/messages.py b/backend/packages/harness/deerflow/utils/messages.py index 8e873c039..d9ce6a510 100644 --- a/backend/packages/harness/deerflow/utils/messages.py +++ b/backend/packages/harness/deerflow/utils/messages.py @@ -9,6 +9,28 @@ from langchain_core.messages import HumanMessage ORIGINAL_USER_CONTENT_KEY = "original_user_content" SUMMARY_MESSAGE_NAME = "summary" +#: Suffix ``DynamicContextMiddleware``'s ID-swap gives the real user message; the +#: reminder SystemMessage takes the original id so ``add_messages`` can replace it +#: in place. It lives here rather than beside the middleware because the message +#: identity rule in ``deerflow.runtime.events.message_identity`` needs it too, and +#: importing the middleware from there closes a cycle +#: (middleware -> deerflow.runtime -> worker -> events -> middleware). +INJECTED_USER_MESSAGE_ID_SUFFIX = "__user" + + +def strip_injected_user_message_id_suffix(message_id: str | None) -> str | None: + """Return the id *message_id* had before the reminder ID-swap. + + Replaying a persisted user turn must feed the graph the id the client + originally sent: a ``{id}__user`` message is skipped as an injection target, + so replaying one into a state that has no reminder yet silently drops the + date and memory block for that turn. + """ + + if isinstance(message_id, str) and message_id.endswith(INJECTED_USER_MESSAGE_ID_SUFFIX): + return message_id[: -len(INJECTED_USER_MESSAGE_ID_SUFFIX)] or message_id + return message_id + def message_content_to_text(content: Any) -> str: """Extract text from LangChain message content shapes.""" diff --git a/backend/tests/test_gateway_imports.py b/backend/tests/test_gateway_imports.py index 7272ea2f1..0e08ea318 100644 --- a/backend/tests/test_gateway_imports.py +++ b/backend/tests/test_gateway_imports.py @@ -28,6 +28,36 @@ def test_gateway_app_imports_first_without_subagent_import_cycle() -> None: assert result.returncode == 0, result.stderr +def test_title_middleware_imports_without_message_identity_cycle() -> None: + """A middleware module must be importable as the process's first import. + + ``message_identity`` reaching back into ``agents.middlewares`` closed a cycle + (middleware -> deerflow.runtime -> worker -> events -> middleware) that only + stayed hidden while some earlier import happened to break it first. Running + ``tests/test_title_generation.py`` on its own was enough to hit it. + """ + result = subprocess.run( + [sys.executable, "-c", "from deerflow.agents.middlewares.title_middleware import TitleMiddleware; print(TitleMiddleware.__name__)"], + capture_output=True, + text=True, + env=_gateway_import_env(), + ) + assert result.returncode == 0, result.stderr + assert "TitleMiddleware" in result.stdout + + +def test_message_identity_imports_standalone() -> None: + """The seq-lookup identity helper must not require the agent package first.""" + result = subprocess.run( + [sys.executable, "-c", "from deerflow.runtime.events.message_identity import message_identity; print(message_identity({'id': 'x__user', 'type': 'human'}))"], + capture_output=True, + text=True, + env=_gateway_import_env(), + ) + assert result.returncode == 0, result.stderr + assert "message:x" in result.stdout + + def test_subagent_package_public_executor_exports_are_lazy_importable() -> None: """The package-level executor exports must not re-enter their own import.""" result = subprocess.run( diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py index 9b2f8d47c..69aae6f0a 100644 --- a/backend/tests/test_gateway_services.py +++ b/backend/tests/test_gateway_services.py @@ -3319,6 +3319,33 @@ async def test_start_run_rejects_invalid_thread_id_before_resolving_dependencies assert "Invalid thread_id" in exc_info.value.detail +def test_normalize_input_strips_the_server_owned_message_seq(): + """`deerflow_seq` is display metadata the Gateway attaches on the way out. + + A client replaying messages (regenerate / edit-and-rerun) would otherwise + write it into the checkpoint, where it becomes wrong the moment the thread + is forked — a branch re-seeds its feed and reassigns seq (#4380). + """ + from app.gateway.services import normalize_input + from deerflow.runtime.events.message_identity import MESSAGE_SEQ_KEY + + result = normalize_input( + { + "messages": [ + { + "role": "human", + "content": "replayed turn", + "additional_kwargs": {MESSAGE_SEQ_KEY: 2, "keep_me": True}, + } + ] + } + ) + + kwargs = result["messages"][0].additional_kwargs + assert MESSAGE_SEQ_KEY not in kwargs + assert kwargs["keep_me"] is True + + def test_client_forged_user_id_is_scrubbed_for_external_callers(): """user_id now selects which credential user-scoped MCP auth injects, so a client-forged value must never survive merge + inject on any external path diff --git a/backend/tests/test_run_event_store.py b/backend/tests/test_run_event_store.py index 76ba803b8..5ba98c849 100644 --- a/backend/tests/test_run_event_store.py +++ b/backend/tests/test_run_event_store.py @@ -1162,3 +1162,383 @@ class TestJsonlRunEventStore: assert c == 1 assert not (tmp_path / "jsonl" / "threads" / "t1" / "runs" / "r2.jsonl").exists() assert await s.count_messages("t1") == 1 + + +class TestGetMessageSeqs: + """Look up the thread-global seq of already-persisted messages by identity. + + A checkpoint carries no seq of its own and loses messages to summarization, + so a client merging it with the seq-ordered thread feed cannot place a + surviving old message (#4666). The seq already exists here, keyed by the + message's identity; this exposes it without paging the whole feed. + """ + + @pytest.mark.anyio + async def test_returns_seq_for_a_persisted_message(self, store): + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1", "content": "hello"}, + ) + + assert await store.get_message_seqs("t1", ["message:u1"]) == {"message:u1": 1} + + @pytest.mark.anyio + async def test_a_tool_message_is_identified_by_its_tool_call_id(self, store): + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.tool.result", + category="message", + content={"type": "tool", "id": "lc-abc", "tool_call_id": "call_1", "content": "OK"}, + ) + + assert await store.get_message_seqs("t1", ["tool:call_1"]) == {"tool:call_1": 1} + + @pytest.mark.anyio + async def test_the_injected_user_suffix_collapses_to_one_identity(self, store): + """DynamicContextMiddleware re-keys the submitted turn ``X`` to ``X__user``. + + The feed stores the ``__user`` copy while a caller may ask under either + spelling; both must resolve to the same row, or the very message this + feature exists to place would be the one it cannot find. + """ + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1__user", "content": "hello"}, + ) + + assert await store.get_message_seqs("t1", ["message:u1"]) == {"message:u1": 1} + + @pytest.mark.anyio + async def test_unknown_identities_are_absent_rather_than_an_error(self, store): + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1", "content": "hello"}, + ) + + result = await store.get_message_seqs("t1", ["message:u1", "message:never-persisted"]) + + assert result == {"message:u1": 1} + + @pytest.mark.anyio + async def test_non_message_events_are_not_looked_up(self, store): + await store.put( + thread_id="t1", + run_id="r1", + event_type="run.start", + category="trace", + content={"type": "human", "id": "u1"}, + ) + + assert await store.get_message_seqs("t1", ["message:u1"]) == {} + + @pytest.mark.anyio + async def test_lookup_is_scoped_to_the_thread(self, store): + await store.put( + thread_id="t2", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1", "content": "hello"}, + ) + + assert await store.get_message_seqs("t1", ["message:u1"]) == {} + + @pytest.mark.anyio + async def test_an_empty_request_does_not_scan(self, store): + assert await store.get_message_seqs("t1", []) == {} + + @pytest.mark.anyio + async def test_a_replaced_message_keeps_its_first_seq(self, store): + """A message re-persisted later must not jump to the tail of the feed.""" + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1", "content": "hello"}, + ) + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1", "content": "hello (edited)"}, + ) + + assert await store.get_message_seqs("t1", ["message:u1"]) == {"message:u1": 1} + + @pytest.mark.anyio + async def test_the_scan_stops_once_every_wanted_identity_is_resolved(self, store): + """Rows past the last wanted seq can only lose the earliest-seq-wins + tiebreak, so scanning them is busy-work — on `/state`/`/history` reads + of long threads this lookup is the only one and the wanted set is + typically tiny.""" + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1", "content": "hello"}, + ) + + class _Tripwire(dict): + def get(self, *_args, **_kwargs): + raise AssertionError("scan continued past the row that resolved the last wanted identity") + + store._messages["t1"].append(_Tripwire()) + + assert await store.get_message_seqs("t1", ["message:u1"]) == {"message:u1": 1} + + @pytest.mark.anyio + async def test_jsonl_store_resolves_identities(self, tmp_path): + from deerflow.runtime.events.store.jsonl import JsonlRunEventStore + + s = JsonlRunEventStore(base_dir=tmp_path / "jsonl") + await s.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1__user", "content": "hello"}, + ) + await s.put( + thread_id="t1", + run_id="r1", + event_type="llm.tool.result", + category="message", + content={"type": "tool", "tool_call_id": "call_1", "content": "OK"}, + ) + + assert await s.get_message_seqs("t1", ["message:u1", "tool:call_1"]) == { + "message:u1": 1, + "tool:call_1": 2, + } + + @pytest.mark.anyio + async def test_db_store_resolves_identities(self, tmp_path): + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + from deerflow.runtime.events.store.db import DbRunEventStore + + url = f"sqlite+aiosqlite:///{tmp_path / 'seqs.db'}" + await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path)) + try: + s = DbRunEventStore(get_session_factory()) + await s.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1__user", "content": "hello"}, + ) + await s.put( + thread_id="t1", + run_id="r1", + event_type="llm.tool.result", + category="message", + content={"type": "tool", "tool_call_id": "call_1", "content": "OK"}, + ) + await s.put( + thread_id="t1", + run_id="r1", + event_type="run.start", + category="trace", + content={"type": "human", "id": "ignored"}, + ) + + assert await s.get_message_seqs("t1", ["message:u1", "tool:call_1", "message:ignored"]) == { + "message:u1": 1, + "tool:call_1": 2, + } + finally: + await close_engine() + + @pytest.mark.anyio + async def test_db_store_only_decodes_rows_that_can_match(self, tmp_path, monkeypatch): + """Rows that cannot hold a wanted identity must not be fetched and + JSON-decoded in Python. + + The ``content`` column carries full tool outputs, so on the long + threads this lookup exists for (a `/state` or `/history` read of a + compacted thread), decoding every message row is heavy I/O plus N + JSON parses — and a wanted identity absent from the feed (a message + still streaming) would defeat any early-exit and force exactly that + full scan. The candidate rows are prefiltered in SQL instead.""" + import json as real_json + from types import SimpleNamespace + + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + from deerflow.runtime.events.store import db as db_module + from deerflow.runtime.events.store.db import DbRunEventStore + + url = f"sqlite+aiosqlite:///{tmp_path / 'seqs.db'}" + await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path)) + try: + s = DbRunEventStore(get_session_factory()) + await s.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1__user", "content": "hello"}, + ) + for i in range(3): + await s.put( + thread_id="t1", + run_id="r1", + event_type="llm.ai.output", + category="message", + content={"type": "ai", "id": f"unrelated-{i}", "content": "big tool output " * 100}, + ) + + decoded: list[str] = [] + + def counting_loads(raw, *args, **kwargs): + decoded.append(raw) + return real_json.loads(raw, *args, **kwargs) + + monkeypatch.setattr(db_module, "json", SimpleNamespace(loads=counting_loads, dumps=real_json.dumps, JSONDecodeError=real_json.JSONDecodeError)) + + # "message:in-flight" is not in the feed: without the SQL + # prefilter it would defeat the early exit and decode all rows. + assert await s.get_message_seqs("t1", ["message:u1", "message:in-flight"]) == {"message:u1": 1} + assert len(decoded) == 1 + finally: + await close_engine() + + @pytest.mark.anyio + async def test_db_store_resolves_an_id_the_sql_prefilter_cannot_express(self, tmp_path): + """An id carrying LIKE wildcards or JSON-escaped characters cannot be + matched as a raw substring of the stored JSON — the lookup must fall + back to the full scan for the whole wanted set, not silently miss.""" + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + from deerflow.runtime.events.store.db import DbRunEventStore + + url = f"sqlite+aiosqlite:///{tmp_path / 'seqs.db'}" + await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path)) + try: + s = DbRunEventStore(get_session_factory()) + await s.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": 'odd%wild"card', "content": "hello"}, + ) + + assert await s.get_message_seqs("t1", ['message:odd%wild"card']) == {'message:odd%wild"card': 1} + finally: + await close_engine() + + +class TestAttachMessageSeq: + """The one stamping expression shared by the worker's `_MessageSeqStamper` + and the request-scoped `stamp_messages_with_seq` — a single helper so the + two counterparts cannot silently diverge.""" + + def test_attaches_the_seq_under_the_server_owned_key(self): + from deerflow.runtime.events.message_identity import attach_message_seq + + stamped = attach_message_seq({"type": "human", "id": "u1"}, 7) + + assert stamped["additional_kwargs"] == {"deerflow_seq": 7} + + def test_existing_additional_kwargs_are_preserved(self): + from deerflow.runtime.events.message_identity import attach_message_seq + + stamped = attach_message_seq({"type": "ai", "id": "a1", "additional_kwargs": {"run_id": "r1"}}, 3) + + assert stamped["additional_kwargs"] == {"run_id": "r1", "deerflow_seq": 3} + + def test_the_input_message_is_not_mutated(self): + from deerflow.runtime.events.message_identity import attach_message_seq + + message = {"type": "human", "id": "u1", "additional_kwargs": {"run_id": "r1"}} + + attach_message_seq(message, 5) + + assert message["additional_kwargs"] == {"run_id": "r1"} + + +class TestStampMessagesWithSeq: + """Attach the feed seq to an arbitrary list of checkpoint messages. + + The streaming path stamps `values` frames as they are published, but a + client that merely opens a conversation never sees a frame: it reads the + checkpoint over REST. Without a seq there, a summarization-rescued early + turn has no absolute position and lands wherever the nearest anchor puts + it (#4666), which is behind the newest question rather than at the head. + """ + + @pytest.mark.anyio + async def test_stamps_a_persisted_message(self, store): + from deerflow.runtime.events.message_seq import stamp_messages_with_seq + + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1__user", "content": "MARK-FIRST"}, + ) + + stamped = await stamp_messages_with_seq(store, "t1", [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]) + + assert stamped[0]["additional_kwargs"]["deerflow_seq"] == 1 + + @pytest.mark.anyio + async def test_a_message_absent_from_the_feed_is_left_alone(self, store): + from deerflow.runtime.events.message_seq import stamp_messages_with_seq + + messages = [{"type": "ai", "id": "not-persisted", "content": "…"}] + + stamped = await stamp_messages_with_seq(store, "t1", messages) + + assert "deerflow_seq" not in (stamped[0].get("additional_kwargs") or {}) + + @pytest.mark.anyio + async def test_the_input_list_is_not_mutated(self, store): + from deerflow.runtime.events.message_seq import stamp_messages_with_seq + + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1", "content": "hi"}, + ) + original = [{"type": "human", "id": "u1", "content": "hi"}] + + await stamp_messages_with_seq(store, "t1", original) + + assert original[0].get("additional_kwargs") is None + + @pytest.mark.anyio + async def test_a_missing_store_returns_the_messages_unchanged(self): + from deerflow.runtime.events.message_seq import stamp_messages_with_seq + + messages = [{"type": "human", "id": "u1", "content": "hi"}] + + assert await stamp_messages_with_seq(None, "t1", messages) == messages + + @pytest.mark.anyio + async def test_a_failing_store_degrades_instead_of_raising(self, store): + """Placement is an enhancement; a broken lookup must not fail the read.""" + from deerflow.runtime.events.message_seq import stamp_messages_with_seq + + class _Broken: + async def get_message_seqs(self, *_args, **_kwargs): + raise RuntimeError("feed unavailable") + + messages = [{"type": "human", "id": "u1", "content": "hi"}] + + assert await stamp_messages_with_seq(_Broken(), "t1", messages) == messages diff --git a/backend/tests/test_run_journal.py b/backend/tests/test_run_journal.py index 9adee7690..1b7aa5fa9 100644 --- a/backend/tests/test_run_journal.py +++ b/backend/tests/test_run_journal.py @@ -449,7 +449,15 @@ class TestFinalToolMessageReconciliation: assert not any(m["event_type"] == "llm.tool.result" for m in messages) @pytest.mark.anyio - async def test_root_chain_end_ignores_non_allowlisted_tool_message(self, journal_setup): + async def test_root_chain_end_ignores_subagent_tool_message(self, journal_setup): + """Reconciliation covers the lead agent's own calls only. + + A subagent's internal tool results belong to its own step feed + (``subagent.step``), not to the thread's message feed; + ``_remember_current_run_tool_calls`` records lead-agent calls only. + This is the boundary that keeps reconciliation safe now that it is no + longer narrowed to an ``ask_clarification`` allowlist. + """ from langchain_core.messages import ToolMessage j, store = journal_setup @@ -457,7 +465,7 @@ class TestFinalToolMessageReconciliation: _make_llm_response("", tool_calls=[{"id": "call_search", "name": "web_search", "args": {"query": "deerflow"}}]), run_id=uuid4(), parent_run_id=None, - tags=["lead_agent"], + tags=["subagent:general-purpose"], ) tool_msg = ToolMessage(content="Search result", tool_call_id="call_search", name="web_search") @@ -491,6 +499,40 @@ class TestFinalToolMessageReconciliation: messages = await store.list_messages("t1") assert not any(m["event_type"] == "llm.tool.result" for m in messages) + @pytest.mark.anyio + async def test_root_chain_end_reconciles_any_middleware_short_circuited_tool_message(self, journal_setup): + """A middleware that blocks a tool call still returns a user-visible result. + + ReadBeforeWriteMiddleware answers a blocked ``write_file`` with an error + ToolMessage instead of running the tool, so LangChain never emits + ``on_tool_end`` and the message never reached the event store. The user + saw it during the run and it vanished on reload (#4666). Reconciliation + is not specific to ``ask_clarification``: any visible tool result the + model asked for in this run belongs in the thread feed. + """ + from langchain_core.messages import ToolMessage + + j, store = journal_setup + j.on_llm_end( + _make_llm_response("", tool_calls=[{"id": "call_write", "name": "write_file", "args": {"path": "/mnt/user-data/outputs/a.txt"}}]), + run_id=uuid4(), + parent_run_id=None, + tags=["lead_agent"], + ) + blocked = ToolMessage( + content="Error: write_file blocked — read the file before writing to it", + tool_call_id="call_write", + name="write_file", + ) + + j.on_chain_end({"messages": [blocked]}, run_id=uuid4()) + await j.flush() + + messages = await store.list_messages("t1") + tool_results = [m for m in messages if m["event_type"] == "llm.tool.result"] + assert len(tool_results) == 1 + assert tool_results[0]["content"]["name"] == "write_file" + class TestCustomEvents: @pytest.mark.anyio @@ -542,6 +584,58 @@ class TestBufferFlush: assert any(e["event_type"] == "llm.ai.response" for e in events) +class TestFeedGeneration: + """The counter that tells a cached feed lookup when to re-ask. + + A message this run produces is not in the feed while it is only buffered, + so a reader looking it up legitimately misses. Bumping this on every write + lets that reader retry exactly when retrying could answer differently, + rather than either polling the store or caching the miss for the whole run + (#4696 review). + """ + + @pytest.mark.anyio + async def test_buffering_alone_does_not_advance_it(self, journal_setup): + j, _store = journal_setup + j.on_llm_end(_make_llm_response("A"), run_id=uuid4(), parent_run_id=None, tags=["lead_agent"]) + + assert len(j._buffer) == 1 + assert j.feed_generation == 0 + + @pytest.mark.anyio + async def test_a_threshold_flush_advances_it(self, journal_setup): + j, _store = journal_setup + j._flush_threshold = 1 + + j.on_llm_end(_make_llm_response("A"), run_id=uuid4(), parent_run_id=None, tags=["lead_agent"]) + await asyncio.sleep(0.1) + + assert j.feed_generation == 1 + + @pytest.mark.anyio + async def test_a_terminal_flush_advances_it(self, journal_setup): + j, _store = journal_setup + j.on_llm_end(_make_llm_response("A"), run_id=uuid4(), parent_run_id=None, tags=["lead_agent"]) + + await j.flush() + + assert j.feed_generation == 1 + + @pytest.mark.anyio + async def test_a_failed_write_leaves_it_alone(self): + """Nothing became readable, so a cached miss must not be re-asked.""" + + class FailingStore(MemoryRunEventStore): + async def put_batch(self, events): + raise RuntimeError("store unavailable") + + j = RunJournal("r-gen", "t-gen", FailingStore(), flush_threshold=1) + j.on_llm_end(_make_llm_response("A"), run_id=uuid4(), parent_run_id=None, tags=["lead_agent"]) + await asyncio.sleep(0.1) + + assert j.feed_generation == 0 + + class TestIdentifyCaller: def test_lead_agent_tag(self, journal_setup): j, _ = journal_setup diff --git a/backend/tests/test_threads_router.py b/backend/tests/test_threads_router.py index d4f3e764d..252eabf50 100644 --- a/backend/tests/test_threads_router.py +++ b/backend/tests/test_threads_router.py @@ -3780,3 +3780,90 @@ def test_update_thread_state_inserts_new_checkpoint_each_call() -> None: assert all(cid is not None for cid in resp_ids), f"response missing checkpoint_id: {resp_ids}" assert set(resp_ids) <= set(ids), f"aput discarded endpoint-assigned id: returned {resp_ids}, stored {ids}" assert resp_ids[1] > resp_ids[0], f"endpoint-assigned uuid6 not preserved/ordered through aput: {resp_ids}" + + +class TestRestReadsCarryMessageSeq: + """Opening a conversation must expose the same feed seq the stream does. + + `_MessageSeqStamper` sits on the streaming publish path, so a client that + joins a live run gets placement information while one that merely opens the + thread does not — and opening is the common case. Without a seq the merge + falls back to the nearest shared anchor, which after summarization sits deep + inside the loaded page, so a rescued first user turn renders behind the + newest question instead of at the head (#4666). + """ + + @staticmethod + def _seed_thread(app, checkpointer, thread_id: str, *, with_feed: bool) -> None: + """Create the thread and its checkpoint without going through HTTP. + + ``POST /api/threads`` writes its own initial checkpoint, which would + overwrite the one under test. + """ + + async def _seed() -> None: + await app.state.thread_store.create(thread_id) + if with_feed: + await app.state.run_event_store.put( + thread_id=thread_id, + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1__user", "content": "MARK-FIRST"}, + ) + await checkpointer.aput( + {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}, + { + **empty_checkpoint(), + "id": str(uuid6(clock_seq=-2)), + "channel_values": {"messages": [HumanMessage(content="MARK-FIRST", id="u1__user")]}, + "channel_versions": {"messages": 1}, + }, + {"source": "loop", "step": 1, "writes": {}, "parents": {}}, + {"messages": 1}, + ) + + asyncio.run(_seed()) + + def _app_with_feed(self, thread_id: str): + from deerflow.runtime.events.store.memory import MemoryRunEventStore + + app, _store, checkpointer = _build_thread_app() + app.state.run_event_store = MemoryRunEventStore() + self._seed_thread(app, checkpointer, thread_id, with_feed=True) + return app + + def test_state_carries_the_seq_of_a_persisted_message(self) -> None: + app = self._app_with_feed("thread-seq-state") + + with TestClient(app) as client: + response = client.get("/api/threads/thread-seq-state/state") + + assert response.status_code == 200, response.text + messages = response.json()["values"]["messages"] + assert messages[0]["additional_kwargs"]["deerflow_seq"] == 1 + + def test_history_carries_the_seq_of_a_persisted_message(self) -> None: + app = self._app_with_feed("thread-seq-history") + + with TestClient(app) as client: + response = client.post("/api/threads/thread-seq-history/history", json={"limit": 1}) + + assert response.status_code == 200, response.text + messages = response.json()[0]["values"]["messages"] + assert messages[0]["additional_kwargs"]["deerflow_seq"] == 1 + + def test_a_message_the_feed_does_not_know_is_left_unstamped(self) -> None: + """Only persisted messages get a seq; the rest keep the weaving path.""" + from deerflow.runtime.events.store.memory import MemoryRunEventStore + + app, _store, checkpointer = _build_thread_app() + app.state.run_event_store = MemoryRunEventStore() + self._seed_thread(app, checkpointer, "thread-seq-unknown", with_feed=False) + + with TestClient(app) as client: + response = client.get("/api/threads/thread-seq-unknown/state") + + assert response.status_code == 200, response.text + messages = response.json()["values"]["messages"] + assert "deerflow_seq" not in (messages[0].get("additional_kwargs") or {}) diff --git a/backend/tests/test_worker_stream_subgraph_namespace.py b/backend/tests/test_worker_stream_subgraph_namespace.py index 0db61c7a7..2010a3629 100644 --- a/backend/tests/test_worker_stream_subgraph_namespace.py +++ b/backend/tests/test_worker_stream_subgraph_namespace.py @@ -565,3 +565,343 @@ class TestWorkerSubgraphStreamIntegration: assert any(_PARENT_FINAL_ID in _collect_ids(payload) for payload in bare_values) custom_types = [payload.get("type") for event, payload in events if event == "custom" and isinstance(payload, dict)] assert "task_started" in custom_types and "task_completed" in custom_types + + +class TestMessageSeqStamping: + """A values frame carries the feed seq of messages already persisted. + + The checkpoint has no seq of its own and loses messages to summarization, + so a client merging it with the seq-ordered feed cannot place a surviving + old message once the feed's loaded page window no longer reaches back to it + (#4666). The worker already holds the event store, so it attaches the seq + that store assigned. Nothing is written back to the checkpoint. + """ + + @staticmethod + async def _seeded_store(): + from deerflow.runtime.events.store.memory import MemoryRunEventStore + + store = MemoryRunEventStore() + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1__user", "content": "MARK-FIRST"}, + ) + return store + + @pytest.mark.asyncio + async def test_root_values_frame_stamps_a_persisted_message(self): + from deerflow.runtime.runs.worker import _MessageSeqStamper + + bridge = _FakeBridge() + await _publish_stream_item( + bridge=bridge, + run_id="run-1", + mode="values", + chunk={"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]}, + namespace=(), + file_tool_chunk_batcher=None, + subagent_events=_FakeSubagentEvents(), + seq_stamper=_MessageSeqStamper(await self._seeded_store(), "t1"), + ) + + _run, _event, payload = bridge.published[0] + assert payload["messages"][0]["additional_kwargs"]["deerflow_seq"] == 1 + + @pytest.mark.asyncio + async def test_a_message_not_in_the_feed_is_left_unstamped(self): + """A message still streaming has no seq yet — and needs none: appending + it at the tail is already its correct position.""" + from deerflow.runtime.runs.worker import _MessageSeqStamper + + bridge = _FakeBridge() + await _publish_stream_item( + bridge=bridge, + run_id="run-1", + mode="values", + chunk={"messages": [{"type": "ai", "id": "not-persisted-yet", "content": "…"}]}, + namespace=(), + file_tool_chunk_batcher=None, + subagent_events=_FakeSubagentEvents(), + seq_stamper=_MessageSeqStamper(await self._seeded_store(), "t1"), + ) + + _run, _event, payload = bridge.published[0] + assert "deerflow_seq" not in (payload["messages"][0].get("additional_kwargs") or {}) + + @pytest.mark.asyncio + async def test_subgraph_frames_are_not_stamped(self): + """A subagent frame does not belong to the thread feed's ordering.""" + from deerflow.runtime.runs.worker import _MessageSeqStamper + + bridge = _FakeBridge() + await _publish_stream_item( + bridge=bridge, + run_id="run-1", + mode="values", + chunk={"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]}, + namespace=SUBAGENT_NS, + file_tool_chunk_batcher=None, + subagent_events=_FakeSubagentEvents(), + seq_stamper=_MessageSeqStamper(await self._seeded_store(), "t1"), + ) + + _run, _event, payload = bridge.published[0] + assert "deerflow_seq" not in (payload["messages"][0].get("additional_kwargs") or {}) + + @pytest.mark.asyncio + async def test_no_stamper_publishes_the_frame_unchanged(self): + bridge = _FakeBridge() + await _publish_stream_item( + bridge=bridge, + run_id="run-1", + mode="values", + chunk={"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]}, + namespace=(), + file_tool_chunk_batcher=None, + subagent_events=_FakeSubagentEvents(), + ) + + _run, _event, payload = bridge.published[0] + assert "deerflow_seq" not in (payload["messages"][0].get("additional_kwargs") or {}) + + @pytest.mark.asyncio + async def test_a_resolved_identity_is_not_looked_up_twice(self): + """Only a frame carrying messages it has not seen costs a query — in a + real run that is the compaction frame, not every frame.""" + from deerflow.runtime.runs.worker import _MessageSeqStamper + + store = await self._seeded_store() + calls: list[list[str]] = [] + original = store.get_message_seqs + + async def counting(thread_id, identities, **kwargs): + calls.append(list(identities)) + return await original(thread_id, identities, **kwargs) + + store.get_message_seqs = counting # type: ignore[method-assign] + stamper = _MessageSeqStamper(store, "t1") + frame = {"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]} + + for _ in range(3): + await _publish_stream_item( + bridge=_FakeBridge(), + run_id="run-1", + mode="values", + chunk=dict(frame), + namespace=(), + file_tool_chunk_batcher=None, + subagent_events=_FakeSubagentEvents(), + seq_stamper=stamper, + ) + + assert len(calls) == 1 + + @pytest.mark.asyncio + async def test_a_miss_is_retried_once_the_feed_advances(self): + """A miss is provisional: the same message can be persisted later in the run. + + A message this run produces reaches a ``values`` frame before + ``RunJournal`` flushes it, so its first lookup misses. The journal then + writes it and it does have a feed seq. Caching that miss permanently + would leave it unstamped for the rest of the run — and a long run that + afterwards rolls past the history page and compacts is exactly the + misplacement this stamper exists to prevent (#4696 review). + """ + from deerflow.runtime.runs.worker import _MessageSeqStamper + + store = await self._seeded_store() + generation = 0 + stamper = _MessageSeqStamper(store, "t1", feed_generation=lambda: generation) + frame = {"messages": [{"type": "ai", "id": "a1", "content": "…"}]} + + first = await stamper.stamp(dict(frame)) + assert "deerflow_seq" not in (first["messages"][0].get("additional_kwargs") or {}) + + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.ai.output", + category="message", + content={"type": "ai", "id": "a1", "content": "…"}, + ) + generation += 1 + + second = await stamper.stamp(dict(frame)) + assert second["messages"][0]["additional_kwargs"]["deerflow_seq"] == 2 + + @pytest.mark.asyncio + async def test_a_miss_is_not_retried_while_the_feed_is_unchanged(self): + """Retrying is bounded by feed writes, not by frames. + + Without the generation gate the retry would run on every frame that + carries a streaming message — a query per frame on exactly the long + threads this stamper is careful to cost one query in. + """ + from deerflow.runtime.runs.worker import _MessageSeqStamper + + store = await self._seeded_store() + calls: list[list[str]] = [] + original = store.get_message_seqs + + async def counting(thread_id, identities, **kwargs): + calls.append(list(identities)) + return await original(thread_id, identities, **kwargs) + + store.get_message_seqs = counting # type: ignore[method-assign] + stamper = _MessageSeqStamper(store, "t1", feed_generation=lambda: 7) + frame = {"messages": [{"type": "ai", "id": "never-persisted", "content": "…"}]} + + for _ in range(3): + await stamper.stamp(dict(frame)) + + assert len(calls) == 1 + + @pytest.mark.asyncio + async def test_a_failed_lookup_is_retried_once_the_feed_advances(self): + """A transient store error must not disable stamping for the whole run. + + The except clause degrades the frame to "no seq"; treating that answer + as final would make one failed query as permanent as a real miss. + """ + from deerflow.runtime.runs.worker import _MessageSeqStamper + + store = await self._seeded_store() + generation = 0 + original = store.get_message_seqs + failed = False + + async def failing_once(thread_id, identities, **kwargs): + nonlocal failed + if not failed: + failed = True + raise RuntimeError("transient store failure") + return await original(thread_id, identities, **kwargs) + + store.get_message_seqs = failing_once # type: ignore[method-assign] + stamper = _MessageSeqStamper(store, "t1", feed_generation=lambda: generation) + frame = {"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]} + + first = await stamper.stamp(dict(frame)) + assert "deerflow_seq" not in (first["messages"][0].get("additional_kwargs") or {}) + + generation += 1 + second = await stamper.stamp(dict(frame)) + assert second["messages"][0]["additional_kwargs"]["deerflow_seq"] == 1 + + @pytest.mark.asyncio + async def test_a_resolved_identity_survives_a_feed_advance(self): + """Only misses are provisional — a resolved seq is never looked up again. + + The feed's earliest-seq-wins rule makes a resolved answer final, so an + advancing feed must not turn the positive cache into a per-write query. + """ + from deerflow.runtime.runs.worker import _MessageSeqStamper + + store = await self._seeded_store() + calls: list[list[str]] = [] + original = store.get_message_seqs + + async def counting(thread_id, identities, **kwargs): + calls.append(list(identities)) + return await original(thread_id, identities, **kwargs) + + store.get_message_seqs = counting # type: ignore[method-assign] + generation = 0 + stamper = _MessageSeqStamper(store, "t1", feed_generation=lambda: generation) + frame = {"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]} + + await stamper.stamp(dict(frame)) + generation += 1 + stamped = await stamper.stamp(dict(frame)) + + assert len(calls) == 1 + assert stamped["messages"][0]["additional_kwargs"]["deerflow_seq"] == 1 + + @pytest.mark.asyncio + async def test_the_run_stamper_re_asks_after_the_journal_writes(self): + """The wiring itself: the run's stamper reads the journal's feed writes. + + Miss-retrying is only reachable if the stamper the worker builds is the + one connected to the journal, and a lambda pointing at the wrong object + fails silently — the stamper simply keeps every miss. + """ + from deerflow.runtime.journal import RunJournal + from deerflow.runtime.runs.worker import _build_seq_stamper + + store = await self._seeded_store() + journal = RunJournal("r1", "t1", store, flush_threshold=100) + stamper = _build_seq_stamper(store, "t1", journal) + frame = {"messages": [{"type": "ai", "id": "a1", "content": "…"}]} + + first = await stamper.stamp(dict(frame)) + assert "deerflow_seq" not in (first["messages"][0].get("additional_kwargs") or {}) + + journal._put( + event_type="llm.ai.response", + category="message", + content={"type": "ai", "id": "a1", "content": "…"}, + ) + await journal.flush() + + second = await stamper.stamp(dict(frame)) + assert second["messages"][0]["additional_kwargs"]["deerflow_seq"] == 2 + + @pytest.mark.asyncio + async def test_a_run_without_a_journal_still_builds_a_stamper(self): + """No writer to report feed growth is not a reason to stop stamping.""" + from deerflow.runtime.runs.worker import _build_seq_stamper + + stamper = _build_seq_stamper(await self._seeded_store(), "t1", None) + + stamped = await stamper.stamp({"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]}) + assert stamped["messages"][0]["additional_kwargs"]["deerflow_seq"] == 1 + + @pytest.mark.asyncio + @pytest.mark.no_auto_user + async def test_stamping_survives_a_launch_path_without_user_context(self, tmp_path): + """A launch path that never inherits the auth contextvar (e.g. a + null-owner scheduled task) must not silently disable stamping. + + The db store's ``user_id=AUTO`` default raises without a user in + context, and ``stamp``'s except clause would swallow that into a + per-frame warning — a graceful degrade that turns the fix off in + exactly the background runs that need it. The stamper therefore + soft-resolves the id once, when it is built, the same way the + worker's write paths beside it do (unset → no filter).""" + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + from deerflow.runtime.events.store.db import DbRunEventStore + from deerflow.runtime.runs.worker import _MessageSeqStamper + + url = f"sqlite+aiosqlite:///{tmp_path / 'seqs.db'}" + await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path)) + try: + store = DbRunEventStore(get_session_factory()) + # Written without a user in context, as the worker's own + # human_message put does on such a path: the row's user_id is NULL. + await store.put( + thread_id="t1", + run_id="r1", + event_type="llm.human.input", + category="message", + content={"type": "human", "id": "u1__user", "content": "MARK-FIRST"}, + ) + + bridge = _FakeBridge() + await _publish_stream_item( + bridge=bridge, + run_id="run-1", + mode="values", + chunk={"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]}, + namespace=(), + file_tool_chunk_batcher=None, + subagent_events=_FakeSubagentEvents(), + seq_stamper=_MessageSeqStamper(store, "t1"), + ) + + _run, _event, payload = bridge.published[0] + assert payload["messages"][0]["additional_kwargs"]["deerflow_seq"] == 1 + finally: + await close_engine() diff --git a/frontend/src/AGENTS.md b/frontend/src/AGENTS.md index 5c5c7d82b..fa599310d 100644 --- a/frontend/src/AGENTS.md +++ b/frontend/src/AGENTS.md @@ -11,7 +11,7 @@ full-file action; do not mount CodeMirror for that artifact until the user requests and receives the complete content. The Gateway retains range ownership and returns 206/416 through `FileResponse`. -3. `useThreadHistory` loads persisted conversation pages from `GET /api/threads/{id}/messages/page`, preserving the backend's thread-global event `seq`; rendering overlays checkpoint/live copies at their matching canonical identities (a summarized checkpoint may contain a protected early input plus a recent tail). Context-compaction rescue diffs every retained visible identity rather than slicing at the first anchor, and keeps a run-scoped ledger of committed visible messages so replacement updates and repeated rolling checkpoint windows cannot erase an already displayed step. The resolver suppresses checkpoint/transient prefixes whose canonical position is still behind an unloaded cursor page instead of collapsing that unknown gap before a recent anchor, then adds optimistic messages without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded. Dynamic context re-keys the submitted user message from `X` to `X__user`; UI identity matching normalizes that reserved suffix only for human messages so the submitted frame and checkpoint replacement remain one visible turn. A locally submitted turn also records its pre-submit identity baseline: if `messages-tuple` publishes new AI/tool steps before `values` publishes that turn's human message, render ordering moves only those non-baseline visible steps behind the new human while leaving history, hidden controls, and reconnected runs untouched. Keep that local order anchor through finish, stop, and stream error because the SDK's settled frame can retain transient event order; replace it on the next local submit and clear it on thread switch or replay-gap recovery. +3. `useThreadHistory` loads persisted conversation pages from `GET /api/threads/{id}/messages/page`, preserving the backend's thread-global event `seq`; rendering overlays checkpoint/live copies at their matching canonical identities (a summarized checkpoint may contain a protected early input plus a recent tail). Context-compaction rescue diffs every retained visible identity rather than slicing at the first anchor, and keeps a run-scoped ledger of committed visible messages so replacement updates and repeated rolling checkpoint windows cannot erase an already displayed step. A checkpoint/transient prefix whose canonical position is still behind an unloaded cursor page is woven in before the first shared anchor, not discarded: both the checkpoint and seq-sorted history place it earlier, so that position is known even when the pages between are not. It must never be appended to the tail (#4065) — the tail is provably wrong — but suppressing it entirely is how a user's own question vanished from a long thread once the first 50-row history page no longer reached back to it (#4666). A collapsed unloaded gap is recoverable by paging; a dropped message is not. Weaving alone restores the message but not its exact position — after compaction the live window carries too few anchors — so both sides now carry the backend's thread-global `additional_kwargs.deerflow_seq`: `buildVisibleHistoryMessages` copies each row's `seq`, and the Gateway stamps it onto `values` frame messages it has already persisted. A live message whose seq is below the loaded window's lower bound is placed ahead of everything on screen instead of before the nearest anchor, which is what puts a compaction-rescued first user turn back at the head rather than mid-transcript. That split happens _before_ the anchor walk, not inside it: a compacted checkpoint can share no identity at all with the loaded page — it keeps only the current run's recent tail, while the page on screen was fetched turns earlier — and the anchor walk then never runs at all, which is precisely when a rescued turn most needs its seq. Doing the split inside the walk left that case appending the message after the whole window (#4666), the one arrangement #4065 proved wrong. A message without a seq (still streaming, so not in the feed yet) keeps the weaving path — the tail is already its correct position. Optimistic messages are then added without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded. Dynamic context re-keys the submitted user message from `X` to `X__user`; UI identity matching normalizes that reserved suffix only for human messages so the submitted frame and checkpoint replacement remain one visible turn. A locally submitted turn also records its pre-submit identity baseline: if `messages-tuple` publishes new AI/tool steps before `values` publishes that turn's human message, render ordering moves only those non-baseline visible steps behind the new human while leaving history, hidden controls, and reconnected runs untouched. Keep that local order anchor through finish, stop, and stream error because the SDK's settled frame can retain transient event order; replace it on the next local submit and clear it on thread switch or replay-gap recovery. 4. Stop actions call the LangGraph SDK stream stop path; `core/threads/hooks.ts` invalidates current-thread, thread-history, token-usage, and sidebar/search caches immediately and schedules one follow-up refetch because SDK stop may finish via abort + fire-and-forget cancel before backend title finalization commits 5. TanStack Query manages server state; localStorage stores user settings. The Settings > Tools MCP switch calls the targeted `PATCH /api/mcp/config` diff --git a/frontend/src/core/threads/hooks.ts b/frontend/src/core/threads/hooks.ts index cbe6c6a96..8a00a98f8 100644 --- a/frontend/src/core/threads/hooks.ts +++ b/frontend/src/core/threads/hooks.ts @@ -171,6 +171,10 @@ const EMPTY_MESSAGES: Message[] = []; const EMPTY_RUN_MESSAGES: RunMessage[] = []; const EMPTY_MESSAGE_IDENTITIES: readonly string[] = []; const INJECTED_USER_MESSAGE_ID_SUFFIX = "__user"; +// Thread-global feed position, attached by the backend to history rows and to +// `values` frame messages it has already persisted. Mirrors MESSAGE_SEQ_KEY in +// `deerflow/runtime/events/message_identity.py`. +const MESSAGE_SEQ_KEY = "deerflow_seq"; const EMPTY_THREAD_VALUES: AgentThreadState = { title: "", @@ -188,6 +192,12 @@ const SUMMARIZATION_MIDDLEWARE_UPDATE_KEYS = new Set([ "DeerFlowSummarizationMiddleware.before_model", ]); +/** Thread-global feed position, when the backend has attached one. */ +function messageSeq(message: Message): number | undefined { + const seq = message.additional_kwargs?.[MESSAGE_SEQ_KEY]; + return typeof seq === "number" ? seq : undefined; +} + function messageIdentity(message: Message): string | undefined { if ( "tool_call_id" in message && @@ -309,9 +319,16 @@ export function buildVisibleHistoryMessages( // Carry the owning run_id onto the content message so historical subtask // cards can fetch their persisted step history on expand (#3779). run_id // lives on the RunMessage wrapper and would otherwise be dropped here. + // seq rides along for the same reason: it is the thread-global position + // this feed is ordered by, and merging needs it on the message itself to + // place a checkpoint copy that falls outside the loaded window (#4666). ...visibleRows.map((message) => ({ ...message.content, run_id: message.run_id, + additional_kwargs: { + ...message.content.additional_kwargs, + [MESSAGE_SEQ_KEY]: message.seq, + }, })), ]); } @@ -499,7 +516,41 @@ export function mergeMessages( const beforeAnchor = new Map(); let pending: Message[] = []; let lastAnchorIdentity: string | undefined; - let hasSharedAnchor = false; + + // Lower bound of the history page window that is currently loaded. A live + // message whose seq is below it belongs before everything on screen, which + // is knowledge the anchor weaving below cannot reach: the anchor only says + // "earlier than this row", and after compaction the nearest anchor can sit + // deep inside the window (#4666 — measured at row 25 of 50). + const canonicalMinSeq = canonical.reduce( + (min, message) => { + const seq = messageSeq(message); + return seq !== undefined && (min === undefined || seq < min) ? seq : min; + }, + undefined, + ); + const beforeWindow: Message[] = []; + + // Split off what the feed places before the loaded window BEFORE the anchor + // walk rather than inside it. A summarized checkpoint can share no identity + // at all with the loaded page — compaction keeps only this run's recent tail, + // while the page on screen was fetched turns earlier — and the anchor loop + // then never runs. That is exactly when a rescued early turn most needs its + // seq: user submits, waits without reloading, compaction fires, and the + // message is appended to the tail instead (#4666). + const liveInWindow: Message[] = []; + for (const message of live) { + const seq = messageSeq(message); + if ( + seq !== undefined && + canonicalMinSeq !== undefined && + seq < canonicalMinSeq + ) { + beforeWindow.push(message); + } else { + liveInWindow.push(message); + } + } // A summarized checkpoint is not necessarily a contiguous history suffix: // middleware may retain protected prompt/input messages at the front and a @@ -507,7 +558,7 @@ export function mergeMessages( // replacing the canonical copy in place. New live messages are woven before // the next shared anchor (or after the last one), so a protected early input // can never be moved to the tail by global last-copy deduplication. - for (const message of live) { + for (const message of liveInWindow) { const identity = messageIdentity(message); const canonicalMessage = identity ? canonicalByIdentity.get(identity) @@ -517,17 +568,22 @@ export function mergeMessages( continue; } - if (pending.length > 0 && hasSharedAnchor) { + // A summarized checkpoint may start with a protected message whose true + // canonical position is separated from this anchor by unloaded pages — + // rescued dynamic-context messages are the common case. Its position + // relative to this anchor is still known (both the checkpoint and + // seq-sorted history place it earlier), so it is woven in before the + // anchor like any other live-only segment. Dropping it instead was how a + // user's own question disappeared from a long thread once the first + // history page no longer reached back to it (#4666): a collapsed unloaded + // gap is recoverable by paging, a discarded message is not. + if (pending.length > 0) { beforeAnchor.set(identity, [ ...(beforeAnchor.get(identity) ?? []), ...pending, ]); } - // A summarized checkpoint may start with a protected message whose true - // canonical position is separated from this anchor by unloaded pages. - // Suppress that ambiguous prefix instead of visually collapsing the gap. pending = []; - hasSharedAnchor = true; lastAnchorIdentity = identity; // A hidden checkpoint control message must not replace a visible canonical @@ -543,7 +599,7 @@ export function mergeMessages( let canonicalAndLive: Message[]; if (!lastAnchorIdentity) { - canonicalAndLive = [...canonical, ...live]; + canonicalAndLive = [...canonical, ...liveInWindow]; } else { canonicalAndLive = []; for (const message of canonical) { @@ -564,6 +620,9 @@ export function mergeMessages( } const merged = dedupeMessagesByIdentity([ + ...[...beforeWindow].sort( + (left, right) => (messageSeq(left) ?? 0) - (messageSeq(right) ?? 0), + ), ...canonicalAndLive, ...optimisticMessages, ]); diff --git a/frontend/tests/unit/core/threads/message-merge.test.ts b/frontend/tests/unit/core/threads/message-merge.test.ts index 27615240c..9921561f6 100644 --- a/frontend/tests/unit/core/threads/message-merge.test.ts +++ b/frontend/tests/unit/core/threads/message-merge.test.ts @@ -53,7 +53,17 @@ test("mergeMessages removes duplicate messages already present in history", () = expect(mergeMessages([human, ai, human, ai], [], [])).toEqual([human, ai]); }); -test("mergeMessages does not collapse an unloaded gap before the first shared anchor", () => { +test("mergeMessages keeps a protected early message before the first shared anchor instead of dropping it", () => { + // #4065 established that an early message rescued by summarization must not + // be appended to the tail: its canonical position is earlier, and the tail is + // provably wrong. Suppressing it entirely was the other half of that fix, and + // it is how a user's own question disappeared from a long thread once the + // first history page no longer reached back to it (#4666). + // + // Both concerns hold at once: the message stays before the first shared + // anchor (never the tail), which is the one position both the checkpoint and + // seq-sorted history agree on. The gap to the unloaded pages remains, but a + // gap is recoverable by paging — a dropped message is not. const protectedEarly = { id: "protected-early", type: "human", @@ -72,7 +82,7 @@ test("mergeMessages does not collapse an unloaded gap before the first shared an expect( mergeMessages([latestHuman, latestAi], [protectedEarly, latestHuman], []), - ).toEqual([latestHuman, latestAi]); + ).toEqual([protectedEarly, latestHuman, latestAi]); }); test("mergeMessages lets live thread messages replace overlapping history", () => { @@ -879,8 +889,8 @@ test("buildVisibleHistoryMessages filters superseded runs but keeps regenerated // run_id is carried onto each content message (#3779) so historical subtask // cards can fetch their persisted step history on expand. expect(buildVisibleHistoryMessages(rows, new Set(["run-old"]))).toEqual([ - { ...newHuman, run_id: "run-new" }, - { ...newAi, run_id: "run-new" }, + { ...newHuman, run_id: "run-new", additional_kwargs: { deerflow_seq: 3 } }, + { ...newAi, run_id: "run-new", additional_kwargs: { deerflow_seq: 4 } }, ]); }); @@ -2133,3 +2143,162 @@ test("refresh reconstructs the same 1-to-6 order from run events without a bridg ), ).toEqual(["1", "2", "3", "4", "5", "6"]); }); + +test("a compacted checkpoint's protected user message survives a history page window that misses it (#4666)", () => { + // Captured from a real two-round long run: once the thread passes the + // 50-row `/messages/page` window AND context compaction fires, the two + // sources stop overlapping at the head. History's first page starts + // mid-run, while the compacted checkpoint still carries the turn's first + // user message (summarization rescues the dynamic-context triplet). + // + // That message is the user's own question. Suppressing it because its + // canonical position sits in an unloaded page makes it vanish from the + // transcript entirely — the "用户消息消失" reports in #4666 / #4508 / #4363. + const reminder = { + id: "u1", + type: "system", + content: "…", + additional_kwargs: { hide_from_ui: true, dynamic_context_reminder: true }, + } as unknown as Message; + const firstUserMessage = { + id: "u1__user", + type: "human", + content: "MARK-FIRST-QUESTION", + } as Message; + const recentStep = { + id: "step-40", + type: "ai", + content: "step 40", + } as Message; + const laterUserMessage = { + id: "u2", + type: "human", + content: "SECOND-QUESTION", + } as Message; + + // First history page: starts mid-run, has_more=true — no first user message. + const canonicalWindow = [recentStep, laterUserMessage]; + // Compacted checkpoint: reminder + rescued first user message + recent tail. + const compactedCheckpoint = [reminder, firstUserMessage, recentStep]; + + const merged = mergeMessages(canonicalWindow, compactedCheckpoint, []); + + expect(merged.map((message) => message.content)).toContain( + "MARK-FIRST-QUESTION", + ); +}); + +test("a checkpoint message earlier than the loaded window is placed by its seq (#4666)", () => { + // With `deerflow_seq` on both sides, placement stops being a guess. Captured + // shape: after compaction the checkpoint still holds the turn's first user + // message (seq=2) while the first history page starts at seq=29, so the only + // anchor available to the old rule sat 25 rows into the window. + const withSeq = (message: Message, seq: number) => + ({ + ...message, + additional_kwargs: { ...message.additional_kwargs, deerflow_seq: seq }, + }) as Message; + + const firstUserMessage = withSeq( + { + id: "u1__user", + type: "human", + content: "MARK-FIRST-QUESTION", + } as Message, + 2, + ); + const windowStep = withSeq( + { id: "step-29", type: "ai", content: "…step 29" } as Message, + 29, + ); + const laterUserMessage = withSeq( + { id: "u2", type: "human", content: "SECOND-QUESTION" } as Message, + 41, + ); + + const anchorStep = withSeq( + { id: "step-58", type: "ai", content: "…step 58" } as Message, + 58, + ); + + // The anchor must sit INSIDE the window, as it does in the captured run + // (canonical #25 of 50): weaving before the anchor is what puts the message + // in the middle, and only seq can say it belongs at the head. + const canonicalWindow = [windowStep, laterUserMessage, anchorStep]; + const compactedCheckpoint = [firstUserMessage, anchorStep]; + + expect( + mergeMessages(canonicalWindow, compactedCheckpoint, []).map( + (m) => m.content, + ), + ).toEqual(["MARK-FIRST-QUESTION", "…step 29", "SECOND-QUESTION", "…step 58"]); +}); + +test("buildVisibleHistoryMessages carries each row's seq onto the message", () => { + const rows = [ + { + run_id: "run-1", + seq: 7, + content: { id: "m1", type: "human", content: "hi" } as Message, + metadata: { caller: "" }, + created_at: "2026-08-04T00:00:00Z", + }, + ] as RunMessage[]; + + expect( + buildVisibleHistoryMessages(rows, new Set())[0]!.additional_kwargs + ?.deerflow_seq, + ).toBe(7); +}); + +test("a checkpoint message earlier than the loaded window is placed by its seq even when the two sides share no anchor (#4666)", () => { + // What a user hits by opening an old, already-summarized conversation and + // sending a new message. The loaded page is the newest rows from BEFORE that + // turn; the compacted checkpoint holds only the rescued first user message + // plus steps of the new run, which are not in the feed yet. The two sides + // therefore share no identity at all and the anchor walk never runs — so the + // rescued turn was appended after the whole window (measured at row 50 of 50 + // on a reproducing run) even though its seq was known the entire time. + const withSeq = (message: Message, seq: number) => + ({ + ...message, + additional_kwargs: { ...message.additional_kwargs, deerflow_seq: seq }, + }) as Message; + + const rescuedFirstTurn = withSeq( + { + id: "u1__user", + type: "human", + content: "MARK-FIRST-QUESTION", + } as Message, + 2, + ); + const loadedWindow = [ + withSeq( + { id: "step-172", type: "ai", content: "…step 172" } as Message, + 172, + ), + withSeq( + { id: "step-174", type: "ai", content: "…step 174" } as Message, + 174, + ), + ]; + // Steps of the run the user just started: still streaming, so no seq yet, and + // no identity in common with the page on screen. + const newRunSteps = [ + { id: "step-new-1", type: "ai", content: "…new step 1" } as Message, + { id: "step-new-2", type: "ai", content: "…new step 2" } as Message, + ]; + + expect( + mergeMessages(loadedWindow, [rescuedFirstTurn, ...newRunSteps], []).map( + (m) => m.content, + ), + ).toEqual([ + "MARK-FIRST-QUESTION", + "…step 172", + "…step 174", + "…new step 1", + "…new step 2", + ]); +});