diff --git a/backend/AGENTS.md b/backend/AGENTS.md index e6945d25e..720c4e257 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -329,6 +329,7 @@ Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runti **Concurrency**: `MAX_CONCURRENT_SUBAGENTS = 3` enforced by `SubagentLimitMiddleware` (truncates excess tool calls in `after_model`); default subagent timeout `subagents.timeout_seconds=1800` (30 min) and built-in `general-purpose` `max_turns=150` (raised from 100/15-min so deep-research subtasks stop hitting `GraphRecursionError` out of the box) **Flow**: `task()` tool → `SubagentExecutor` → background thread → poll 5s → SSE events → result **Events**: `task_started`, `task_running`, `task_completed`/`task_failed`/`task_timed_out` +**Step capture & persistence (#3779)**: `executor.py` captures both assistant turns (`AIMessage`) **and** tool outputs (`ToolMessage`) via `subagents/step_events.py::capture_new_step_messages`, which walks the *newly-appended tail* of each `stream_mode="values"` chunk (not just `messages[-1]`) so a multi-tool-call turn — where LangGraph's `ToolNode` appends several `ToolMessage`s in one super-step — keeps every tool output instead of dropping all but the last. `runtime/runs/worker.py::_SubagentEventBuffer` additionally persists these `task_*` custom events to the `RunEventStore` as `subagent.start`/`subagent.step`/`subagent.end` (`category="subagent"`, `task_id` in `metadata`). It **batches** writes via `put_batch` (flushing on a terminal `subagent.end`, at `FLUSH_THRESHOLD` events, and in the worker's `finally`) rather than one `put()` per step, since `put()` is a documented low-frequency path (per-thread advisory lock per call) and a deep subagent (`max_turns=150`) emits hundreds of steps on the hot stream loop. `build_subagent_step` caps both the per-step `text` and each tool call's serialized `args` at `SUBAGENT_STEP_MAX_CHARS` (flagged `truncated` / `args_truncated`) so a large `write_file`/`bash` payload can't produce an unbounded row. The dedicated category keeps them out of `list_messages` (the thread feed) while `list_events` returns them for the frontend's fetch-on-expand backfill. `list_events` accepts `task_id` (filters on `metadata["task_id"]` — SQL-side in `DbRunEventStore` via `event_metadata["task_id"].as_string()`, in-memory in the JSONL/memory stores) plus an `after_seq` forward cursor, so the card pages through one subagent's steps without the run-wide `limit` truncating the tail (no schema migration: the filter rides the existing run-scoped index). `step_events.py` is a pure, unit-tested layer (`build_subagent_step` / `subagent_run_event`). **Deferred MCP tools** (if `tool_search.enabled`): `SubagentExecutor._build_initial_state` assembles deferral after policy filtering via the shared `assemble_deferred_tools` (fail-closed), appends the `tool_search` tool, injects the `` section into the subagent's `SystemMessage`, and threads the setup to `_create_agent`, which attaches `DeferredToolFilterMiddleware` through `build_subagent_runtime_middlewares(deferred_setup=...)`. Subagents thus withhold full MCP schemas until promotion, same as the lead agent; each task run gets a fresh `ThreadState` so promotion is isolated per run **Checkpointer isolation**: Subagent graphs are compiled with `checkpointer=False` to avoid inheriting the parent run's checkpointer, since subagents are one-shot and never resume. diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py index b6255c160..b1481291a 100644 --- a/backend/app/gateway/routers/thread_runs.py +++ b/backend/app/gateway/routers/thread_runs.py @@ -728,12 +728,18 @@ async def list_run_events( run_id: str, request: Request, event_types: str | None = Query(default=None), + task_id: str | None = Query(default=None), limit: int = Query(default=500, le=2000), + after_seq: int | None = Query(default=None), ) -> list[dict]: - """Return the full event stream for a run (debug/audit).""" + """Return the full event stream for a run (debug/audit). + + ``task_id`` + ``after_seq`` let the subtask card page through one subagent + task's persisted steps without the run-wide ``limit`` truncating the tail (#3779). + """ event_store = get_run_event_store(request) types = event_types.split(",") if event_types else None - return await event_store.list_events(thread_id, run_id, event_types=types, limit=limit) + return await event_store.list_events(thread_id, run_id, event_types=types, task_id=task_id, limit=limit, after_seq=after_seq) @router.get("/{thread_id}/token-usage", response_model=ThreadTokenUsageResponse) diff --git a/backend/packages/harness/deerflow/runtime/events/store/base.py b/backend/packages/harness/deerflow/runtime/events/store/base.py index df5136ba5..008a68e46 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/base.py +++ b/backend/packages/harness/deerflow/runtime/events/store/base.py @@ -71,11 +71,17 @@ class RunEventStore(abc.ABC): run_id: str, *, event_types: list[str] | None = None, + task_id: str | None = None, limit: int = 500, + after_seq: int | None = None, ) -> list[dict]: """Return the full event stream for a run, ordered by seq ascending. - Optionally filter by event_types. + Optionally filter by ``event_types`` and/or ``task_id`` (matched against + ``metadata["task_id"]``). ``after_seq`` is a forward cursor returning the + first ``limit`` records with seq > after_seq, so callers can page through + a single subagent task's events without the run-wide ``limit`` truncating + the tail (#3779). """ @abc.abstractmethod diff --git a/backend/packages/harness/deerflow/runtime/events/store/db.py b/backend/packages/harness/deerflow/runtime/events/store/db.py index 58a26896c..4190bf440 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/db.py +++ b/backend/packages/harness/deerflow/runtime/events/store/db.py @@ -215,7 +215,9 @@ class DbRunEventStore(RunEventStore): run_id, *, event_types=None, + task_id=None, limit=500, + after_seq=None, user_id: str | None | _AutoSentinel = AUTO, ): resolved_user_id = resolve_user_id(user_id, method_name="DbRunEventStore.list_events") @@ -224,6 +226,15 @@ class DbRunEventStore(RunEventStore): stmt = stmt.where(RunEventRow.user_id == resolved_user_id) if event_types: stmt = stmt.where(RunEventRow.event_type.in_(event_types)) + if task_id is not None: + # Filter on metadata["task_id"] in SQL (before LIMIT) so cursor + # pagination over a single subagent task stays correct (#3779). The + # query is already scoped to (thread_id, run_id), so the JSON probe + # only runs over this run's small candidate set; ``.as_string()`` + # renders to json_extract (SQLite) / ->> (Postgres). + stmt = stmt.where(RunEventRow.event_metadata["task_id"].as_string() == task_id) + if after_seq is not None: + stmt = stmt.where(RunEventRow.seq > after_seq) stmt = stmt.order_by(RunEventRow.seq.asc()).limit(limit) async with self._sf() as session: result = await session.execute(stmt) diff --git a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py index 8af047dcb..195086ea6 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py +++ b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py @@ -175,10 +175,14 @@ class JsonlRunEventStore(RunEventStore): else: return messages[-limit:] - async def list_events(self, thread_id, run_id, *, event_types=None, limit=500): + async def list_events(self, thread_id, run_id, *, event_types=None, task_id=None, limit=500, after_seq=None): events = await asyncio.to_thread(self._read_run_events, thread_id, run_id) if event_types is not None: events = [e for e in events if e.get("event_type") in event_types] + if task_id is not None: + events = [e for e in events if (e.get("metadata") or {}).get("task_id") == task_id] + if after_seq is not None: + events = [e for e in events if e.get("seq", 0) > after_seq] return events[:limit] async def list_messages_by_run(self, thread_id, run_id, *, limit=50, before_seq=None, after_seq=None): diff --git a/backend/packages/harness/deerflow/runtime/events/store/memory.py b/backend/packages/harness/deerflow/runtime/events/store/memory.py index c8e732d44..573e3f546 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/memory.py +++ b/backend/packages/harness/deerflow/runtime/events/store/memory.py @@ -109,12 +109,16 @@ class MemoryRunEventStore(RunEventStore): # Return the latest `limit` records, ascending. return messages[-limit:] - async def list_events(self, thread_id, run_id, *, event_types=None, limit=500): + async def list_events(self, thread_id, run_id, *, event_types=None, task_id=None, limit=500, after_seq=None): # ``_events_by_run`` is already scoped to this run and seq-ordered, so we # touch only this run's events instead of scanning the whole thread. run_events = self._events_by_run.get(thread_id, {}).get(run_id, []) if event_types is not None: run_events = [e for e in run_events if e["event_type"] in event_types] + if task_id is not None: + run_events = [e for e in run_events if (e.get("metadata") or {}).get("task_id") == task_id] + if after_seq is not None: + run_events = [e for e in run_events if e.get("seq", 0) > after_seq] return run_events[:limit] async def list_messages_by_run(self, thread_id, run_id, *, limit=50, before_seq=None, after_seq=None): diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index 2e90828bd..d51e16501 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -118,6 +118,65 @@ def _agent_factory_supports_app_config(agent_factory: Any) -> bool: return _compute_agent_factory_supports_app_config(agent_factory) +class _SubagentEventBuffer: + """Buffer subagent ``task_*`` step events and flush them in one locked batch (#3779). + + The live SSE bridge already forwards these events for real-time display; this + additionally writes them so the subtask card's step history survives a reload. + + ``RunEventStore.put`` is documented as a low-frequency path — on Postgres each + call opens its own transaction and takes a per-thread advisory lock. A deep + subagent (``general-purpose`` runs up to ``max_turns=150``) emits hundreds of + ``task_running`` steps on the hot stream loop, so persisting each with + ``put()`` would serialize against the run's own message-batch writer. This + accumulates recognized subagent events and writes them with ``put_batch``, + which acquires the lock once per batch, honoring the store's contract. + + Best-effort: a missing store (run_events not configured) or an unrecognized + chunk is a no-op, flush failures are logged but never propagate into the + stream loop, and terminal ``subagent.end`` events flush eagerly so a completed + subagent's step history is durable promptly rather than only at run end. + """ + + #: Flush once this many events are buffered, bounding memory and reload lag on + #: a single deep subagent without paying a per-step lock. + FLUSH_THRESHOLD = 25 + + def __init__(self, event_store: Any | None, thread_id: str, run_id: str) -> None: + self._event_store = event_store + self._thread_id = thread_id + self._run_id = run_id + self._pending: list[dict[str, Any]] = [] + + async def add(self, chunk: Any) -> None: + """Buffer one custom stream chunk; flush on a terminal event or threshold.""" + if self._event_store is None: + return + # Lazy import: importing deerflow.subagents at module load triggers its + # package __init__ (executor → agents → tools → task_tool), which imports + # back from deerflow.subagents and deadlocks at gateway startup. Deferring + # it to call time (after all modules are loaded) breaks that cycle. + from deerflow.subagents.step_events import subagent_run_event + + record = subagent_run_event(chunk) + if record is None: + return + self._pending.append({"thread_id": self._thread_id, "run_id": self._run_id, **record}) + if record["event_type"] == "subagent.end" or len(self._pending) >= self.FLUSH_THRESHOLD: + await self.flush() + + async def flush(self) -> None: + """Persist buffered events in one ``put_batch`` call; swallow store errors.""" + if self._event_store is None or not self._pending: + return + batch = self._pending + self._pending = [] + try: + await self._event_store.put_batch(batch) + except Exception: + logger.warning("Run %s: failed to persist %d subagent step event(s)", self._run_id, len(batch), exc_info=True) + + async def run_agent( bridge: StreamBridge, run_manager: RunManager, @@ -150,6 +209,10 @@ async def run_agent( llm_error_fallback_message: str | None = None journal = None + # Buffers subagent step events for batched persistence (#3779); assigned once + # streaming starts and flushed in the finally block. Pre-bound to None so the + # finally is safe even if an exception fires before streaming begins. + subagent_events: _SubagentEventBuffer | None = None # Track whether "events" was requested but skipped if "events" in requested_modes: @@ -304,6 +367,11 @@ async def run_agent( logger.info("Run %s: streaming with modes %s (requested: %s)", run_id, lg_modes, requested_modes) + # Buffer subagent step events and persist them in batches (#3779) instead + # of one low-frequency put() per step on the hot stream loop. Flushed in + # the finally block so buffered steps survive abort/exception paths too. + subagent_events = _SubagentEventBuffer(event_store, thread_id, run_id) + # 7. Stream using graph.astream if len(lg_modes) == 1 and not stream_subgraphs: # Single mode, no subgraphs: astream yields raw chunks @@ -315,6 +383,8 @@ async def run_agent( llm_error_fallback_message = llm_error_fallback_message or _extract_llm_error_fallback_message(chunk) sse_event = _lg_mode_to_sse_event(single_mode) await bridge.publish(run_id, sse_event, serialize(chunk, mode=single_mode)) + if single_mode == "custom": + await subagent_events.add(chunk) else: # Multiple modes or subgraphs: astream yields tuples async for item in agent.astream( @@ -334,6 +404,8 @@ async def run_agent( llm_error_fallback_message = llm_error_fallback_message or _extract_llm_error_fallback_message(chunk) sse_event = _lg_mode_to_sse_event(mode) await bridge.publish(run_id, sse_event, serialize(chunk, mode=mode)) + if mode == "custom": + await subagent_events.add(chunk) # 8. Final status if record.abort_event.is_set(): @@ -399,6 +471,11 @@ async def run_agent( ) finally: + # Persist any subagent step events still buffered (#3779) — including on + # abort/exception paths, where the stream loop broke before its own flush. + if subagent_events is not None: + await subagent_events.flush() + # Flush any buffered journal events and persist completion data if journal is not None: try: diff --git a/backend/packages/harness/deerflow/subagents/executor.py b/backend/packages/harness/deerflow/subagents/executor.py index cfdf21fa9..f47523860 100644 --- a/backend/packages/harness/deerflow/subagents/executor.py +++ b/backend/packages/harness/deerflow/subagents/executor.py @@ -27,6 +27,7 @@ from deerflow.models import create_chat_model from deerflow.skills.tool_policy import filter_tools_by_skill_allowed_tools from deerflow.skills.types import Skill from deerflow.subagents.config import SubagentConfig, resolve_subagent_model_name +from deerflow.subagents.step_events import capture_new_step_messages from deerflow.subagents.token_collector import SubagentTokenCollector from deerflow.tracing import build_tracing_callbacks, inject_langfuse_metadata @@ -533,6 +534,9 @@ class SubagentExecutor: # rescanning the append-only ``ai_messages`` list (O(n) per chunk -> O(n^2) # over a run, which reaches max_turns=150 for deep-research subagents). seen_message_ids: set[str] = {mid for msg in ai_messages if (mid := msg.get("id"))} + # Cursor into the append-only message history so each ``values``-mode + # chunk only re-scans the newly-appended tail (see capture_new_step_messages). + processed_message_count = 0 collector: SubagentTokenCollector | None = None try: @@ -628,28 +632,16 @@ class SubagentExecutor: final_state = chunk - # Extract AI messages from the current state + # Capture every step message (assistant turns AND tool outputs) + # appended since the last chunk. A single super-step can append + # several ToolMessages when the model emits multiple tool calls in + # one turn, so capturing only messages[-1] would drop all but the + # last output (#3779). Dedup/serialization live in capture_step_message. messages = chunk.get("messages", []) - if messages: - last_message = messages[-1] - # Check if this is a new AI message - if isinstance(last_message, AIMessage): - # Convert message to dict for serialization - message_dict = last_message.model_dump() - # Only add if it's not already in the list (avoid duplicates) - # Check by comparing message IDs if available, otherwise compare full dict - message_id = message_dict.get("id") - if message_id: - is_duplicate = message_id in seen_message_ids - else: - # id-less messages can't be keyed; fall back to a full-dict compare - is_duplicate = message_dict in ai_messages - - if not is_duplicate: - ai_messages.append(message_dict) - if message_id: - seen_message_ids.add(message_id) - logger.info(f"[trace={self.trace_id}] Subagent {self.config.name} captured AI message #{len(ai_messages)}") + previous_count = len(ai_messages) + processed_message_count = capture_new_step_messages(messages, ai_messages, seen_message_ids, processed_message_count) + if len(ai_messages) > previous_count: + logger.info(f"[trace={self.trace_id}] Subagent {self.config.name} captured {len(ai_messages) - previous_count} step message(s); total #{len(ai_messages)}") logger.info(f"[trace={self.trace_id}] Subagent {self.config.name} completed async execution") token_usage_records = collector.snapshot_records() diff --git a/backend/packages/harness/deerflow/subagents/step_events.py b/backend/packages/harness/deerflow/subagents/step_events.py new file mode 100644 index 000000000..f72e859d7 --- /dev/null +++ b/backend/packages/harness/deerflow/subagents/step_events.py @@ -0,0 +1,227 @@ +"""Build compact subagent step payloads for streaming + persistence. + +Issue #3779: subagent (subtask) execution steps were only visible as the +latest streamed frame and were never persisted, so users could not review +what tools a subagent ran or what each step produced after a reload. + +This module is the pure data-shaping layer. It converts a captured subagent +message dict — the ``model_dump()`` of an ``AIMessage`` (an assistant turn: +text + tool-call requests) or a ``ToolMessage`` (a tool's output) — into the +small, JSON-serializable ``step`` payload that is: + +- streamed live inside the ``task_running`` custom event (``task_tool.py``), and +- persisted as a ``subagent.step`` run event (``runtime/runs/worker.py``). + +Keeping it pure means it is unit-tested without spinning up a graph, and both +the streaming and persistence call sites share one definition of a "step". +""" + +from __future__ import annotations + +import json +from typing import Any + +from langchain_core.messages import AIMessage, BaseMessage, ToolMessage + +from deerflow.utils.messages import message_content_to_text + +#: Default per-step character cap for the ``text`` field. Tool outputs (web +#: search results, file contents) can be large; this cap bounds the persisted +#: run-event row and the streamed frame. It only affects display/storage — the +#: subagent's own LLM context is bounded separately by ToolOutputBudgetMiddleware. +SUBAGENT_STEP_MAX_CHARS = 8192 + +#: ``RunEvent.category`` for persisted subagent steps. A dedicated category (not +#: ``"message"``) keeps these events out of ``list_messages`` (the thread message +#: feed) while still being returned by ``list_events`` for fetch-on-expand (#3779). +SUBAGENT_EVENT_CATEGORY = "subagent" + +#: Map of ``task_*`` terminal custom-event types to their persisted status. +_TERMINAL_EVENT_STATUS: dict[str, str] = { + "task_completed": "completed", + "task_failed": "failed", + "task_cancelled": "cancelled", + "task_timed_out": "timed_out", +} + + +def capture_step_message( + message: BaseMessage, + captured: list[dict[str, Any]], + seen_ids: set[str], +) -> bool: + """Append ``message.model_dump()`` to ``captured`` if it is a new step. + + A "step" is an assistant turn (``AIMessage``) or a tool result + (``ToolMessage``) — issue #3779 added the latter so tool outputs survive. + Other message types (e.g. ``HumanMessage``) are ignored. Dedup is by id + when present, falling back to a full-dict compare for id-less messages so + ``stream_mode="values"`` re-yielding the same trailing message stays O(1). + + Returns ``True`` when a message was appended. + """ + if not isinstance(message, (AIMessage, ToolMessage)): + return False + + message_dict = message.model_dump() + message_id = message_dict.get("id") + if message_id: + if message_id in seen_ids: + return False + elif message_dict in captured: + return False + + captured.append(message_dict) + if message_id: + seen_ids.add(message_id) + return True + + +def capture_new_step_messages( + messages: list[BaseMessage], + captured: list[dict[str, Any]], + seen_ids: set[str], + processed_count: int, +) -> int: + """Capture every step message appended since ``processed_count`` (#3779). + + ``stream_mode="values"`` re-yields the full message history on each chunk, + and a single LangGraph super-step can append several messages at once — most + importantly one ``ToolMessage`` per tool call when the model emits multiple + tool calls in one turn. Capturing only ``messages[-1]`` (the previous + behaviour) silently dropped all but the last tool output. + + When the history grew, walk every newly-appended message. When it did not + grow, re-examine only the trailing message so an id-less in-place replacement + (same length, new content) is still captured — ``capture_step_message``'s + dedup makes an unchanged re-yield a no-op. Returns the new cursor. + """ + total = len(messages) + if total > processed_count: + for message in messages[processed_count:total]: + capture_step_message(message, captured, seen_ids) + return total + if messages: + capture_step_message(messages[-1], captured, seen_ids) + return max(processed_count, total) + + +def truncate_step_text(text: str, max_chars: int) -> tuple[str, bool]: + """Return ``(text, truncated)``, clipping to ``max_chars`` when longer.""" + if max_chars >= 0 and len(text) > max_chars: + return text[:max_chars], True + return text, False + + +def _bounded_tool_call(call: dict[str, Any], max_chars: int) -> dict[str, Any]: + """Return ``{name, args}`` for a captured tool call, capping large args (#3779). + + ``build_subagent_step`` caps the ``text`` field, but tool-call ``args`` were + copied verbatim, so a ``write_file``/``bash`` call carrying a big payload (full + file contents, a heredoc) produced an unbounded persisted ``subagent.step`` + row and streamed frame. When the JSON-serialized args exceed ``max_chars`` we + replace the structured value with a truncated serialized preview and flag it + with ``args_truncated`` — small args stay structured for the card to inspect. + """ + name = call.get("name") + args = call.get("args") + serialized = args if isinstance(args, str) else json.dumps(args, default=str, ensure_ascii=False) + if max_chars >= 0 and len(serialized) > max_chars: + return {"name": name, "args": serialized[:max_chars], "args_truncated": True} + return {"name": name, "args": args} + + +def build_subagent_step( + message: dict[str, Any], + *, + task_id: str, + message_index: int, + max_chars: int = SUBAGENT_STEP_MAX_CHARS, +) -> dict[str, Any]: + """Build the compact step payload from a captured subagent message dict. + + ``kind`` is ``"tool"`` for a ToolMessage (``type == "tool"``) and ``"ai"`` + otherwise. AI steps carry their ``tool_calls`` (name + args only, with large + args capped to ``max_chars`` — see ``_bounded_tool_call``); tool steps carry + the originating ``tool_name``. ``text`` is truncated to ``max_chars`` with the + ``truncated`` flag set accordingly. + """ + kind = "tool" if message.get("type") == "tool" else "ai" + # ``... or ""`` keeps a tool-call-only turn's content=None rendering as "" + # (message_content_to_text would otherwise str()-ify it to "None"). + text, truncated = truncate_step_text(message_content_to_text(message.get("content") or ""), max_chars) + + step: dict[str, Any] = { + "task_id": task_id, + "message_index": message_index, + "kind": kind, + "text": text, + "truncated": truncated, + } + + if kind == "tool": + step["tool_name"] = message.get("name") + else: + step["tool_calls"] = [_bounded_tool_call(call, max_chars) for call in (message.get("tool_calls") or [])] + + return step + + +def subagent_run_event(chunk: Any) -> dict[str, Any] | None: + """Map a ``task_*`` custom stream chunk to ``RunEventStore.put`` kwargs. + + Returns the ``event_type`` / ``category`` / ``content`` / ``metadata`` for a + persistable subagent lifecycle event, or ``None`` for any chunk that is not a + subagent event (so the worker only persists what it recognizes). ``thread_id`` + / ``run_id`` are filled in by the caller. + """ + if not isinstance(chunk, dict): + return None + + event = chunk.get("type") + if not isinstance(event, str) or not event.startswith("task_"): + return None + + task_id = chunk.get("task_id") + + if event == "task_started": + return { + "event_type": "subagent.start", + "category": SUBAGENT_EVENT_CATEGORY, + "content": {"task_id": task_id, "description": chunk.get("description")}, + "metadata": {"task_id": task_id}, + } + + if event == "task_running": + message_index = chunk.get("message_index") + return { + "event_type": "subagent.step", + "category": SUBAGENT_EVENT_CATEGORY, + "content": build_subagent_step(chunk.get("message") or {}, task_id=task_id, message_index=message_index), + "metadata": {"task_id": task_id, "message_index": message_index}, + } + + status = _TERMINAL_EVENT_STATUS.get(event) + if status is not None: + content: dict[str, Any] = {"task_id": task_id, "status": status} + # The final result/error can be a multi-page report; cap it so the + # persisted run-event row stays bounded (it is also kept verbatim on the + # terminal ToolMessage, which the card reads separately). + if chunk.get("result") is not None: + result, result_truncated = truncate_step_text(str(chunk["result"]), SUBAGENT_STEP_MAX_CHARS) + content["result"] = result + if result_truncated: + content["result_truncated"] = True + if chunk.get("error") is not None: + error, error_truncated = truncate_step_text(str(chunk["error"]), SUBAGENT_STEP_MAX_CHARS) + content["error"] = error + if error_truncated: + content["error_truncated"] = True + return { + "event_type": "subagent.end", + "category": SUBAGENT_EVENT_CATEGORY, + "content": content, + "metadata": {"task_id": task_id}, + } + + return None diff --git a/backend/tests/test_run_event_store_filter.py b/backend/tests/test_run_event_store_filter.py new file mode 100644 index 000000000..082fd9fff --- /dev/null +++ b/backend/tests/test_run_event_store_filter.py @@ -0,0 +1,156 @@ +"""Tests for list_events task_id filtering + after_seq cursor pagination (#3779). + +These power the subtask card's fetch-on-expand backfill, which must page through +ONE subagent task's persisted steps without the run-wide 500-event cap dropping +the tail (or an entire later subtask). The filter has to run in the store (before +the limit) so pagination stays correct. +""" + +import pytest + +from deerflow.runtime.events.store.memory import MemoryRunEventStore + + +async def _seed_two_tasks(store): + """Seed run r1 with task A (start + 3 steps) and task B (start + 2 steps).""" + await store.put(thread_id="t1", run_id="r1", event_type="subagent.start", category="subagent", content={"task_id": "A"}, metadata={"task_id": "A"}) + await store.put(thread_id="t1", run_id="r1", event_type="subagent.start", category="subagent", content={"task_id": "B"}, metadata={"task_id": "B"}) + for i in range(3): + await store.put(thread_id="t1", run_id="r1", event_type="subagent.step", category="subagent", content={"task_id": "A", "message_index": i}, metadata={"task_id": "A", "message_index": i}) + for i in range(2): + await store.put(thread_id="t1", run_id="r1", event_type="subagent.step", category="subagent", content={"task_id": "B", "message_index": i}, metadata={"task_id": "B", "message_index": i}) + + +def _task_ids(events): + return [e["metadata"].get("task_id") for e in events] + + +async def _check_task_id_filter(store): + await _seed_two_tasks(store) + a_events = await store.list_events("t1", "r1", task_id="A") + assert _task_ids(a_events) == ["A", "A", "A", "A"] # start + 3 steps + b_events = await store.list_events("t1", "r1", task_id="B") + assert _task_ids(b_events) == ["B", "B", "B"] # start + 2 steps + + +async def _check_task_id_with_event_types(store): + await _seed_two_tasks(store) + a_steps = await store.list_events("t1", "r1", task_id="A", event_types=["subagent.step"]) + assert [e["event_type"] for e in a_steps] == ["subagent.step"] * 3 + assert _task_ids(a_steps) == ["A", "A", "A"] + + +async def _check_after_seq_cursor(store): + await _seed_two_tasks(store) + everything = await store.list_events("t1", "r1") + cursor = everything[2]["seq"] + after = await store.list_events("t1", "r1", after_seq=cursor) + assert all(e["seq"] > cursor for e in after) + assert len(after) == len(everything) - 3 + + +async def _check_task_id_after_seq_paginate(store): + """task_id + after_seq + small limit pages through ONE task with no gaps/dupes.""" + await _seed_two_tasks(store) + collected = [] + after_seq = None + for _ in range(10): # safety bound + page = await store.list_events("t1", "r1", task_id="A", event_types=["subagent.step"], limit=2, after_seq=after_seq) + collected.extend(page) + if len(page) < 2: + break + after_seq = page[-1]["seq"] + assert [e["content"]["message_index"] for e in collected] == [0, 1, 2] + + +async def _check_no_task_id_returns_all(store): + await _seed_two_tasks(store) + everything = await store.list_events("t1", "r1") + assert len(everything) == 7 # 2 starts + 5 steps + + +# -- Memory backend -- + + +@pytest.mark.anyio +async def test_memory_task_id_filter(): + await _check_task_id_filter(MemoryRunEventStore()) + + +@pytest.mark.anyio +async def test_memory_task_id_with_event_types(): + await _check_task_id_with_event_types(MemoryRunEventStore()) + + +@pytest.mark.anyio +async def test_memory_after_seq_cursor(): + await _check_after_seq_cursor(MemoryRunEventStore()) + + +@pytest.mark.anyio +async def test_memory_task_id_after_seq_paginate(): + await _check_task_id_after_seq_paginate(MemoryRunEventStore()) + + +@pytest.mark.anyio +async def test_memory_no_task_id_returns_all(): + await _check_no_task_id_returns_all(MemoryRunEventStore()) + + +# -- DB backend (sqlite): exercises the JSON-field filter on a real dialect -- + + +@pytest.mark.anyio +async def test_db_task_id_filter(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 / 'test.db'}" + await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path)) + try: + await _check_task_id_filter(DbRunEventStore(get_session_factory())) + finally: + await close_engine() + + +@pytest.mark.anyio +async def test_db_task_id_after_seq_paginate(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 / 'test.db'}" + await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path)) + try: + await _check_task_id_after_seq_paginate(DbRunEventStore(get_session_factory())) + finally: + await close_engine() + + +@pytest.mark.anyio +async def test_db_no_task_id_returns_all(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 / 'test.db'}" + await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path)) + try: + await _check_no_task_id_returns_all(DbRunEventStore(get_session_factory())) + finally: + await close_engine() + + +# -- JSONL backend -- + + +@pytest.mark.anyio +async def test_jsonl_task_id_filter(tmp_path): + from deerflow.runtime.events.store.jsonl import JsonlRunEventStore + + await _check_task_id_filter(JsonlRunEventStore(base_dir=str(tmp_path))) + + +@pytest.mark.anyio +async def test_jsonl_task_id_after_seq_paginate(tmp_path): + from deerflow.runtime.events.store.jsonl import JsonlRunEventStore + + await _check_task_id_after_seq_paginate(JsonlRunEventStore(base_dir=str(tmp_path))) diff --git a/backend/tests/test_run_events_endpoint.py b/backend/tests/test_run_events_endpoint.py new file mode 100644 index 000000000..d3301033c --- /dev/null +++ b/backend/tests/test_run_events_endpoint.py @@ -0,0 +1,45 @@ +"""The /events route forwards task_id + after_seq to the store (#3779). + +The subtask card pages through one subagent task's persisted steps via these +query params; this locks the wiring so a rename/typo can't silently drop them +(which would make reload backfill fetch the whole run again, or nothing). +""" + +import pytest + + +@pytest.mark.anyio +async def test_list_run_events_forwards_task_id_and_after_seq(): + from app.gateway.routers.thread_runs import list_run_events + + calls: dict = {} + + class FakeStore: + async def list_events(self, thread_id, run_id, *, event_types=None, task_id=None, limit=500, after_seq=None): + calls.update(thread_id=thread_id, run_id=run_id, event_types=event_types, task_id=task_id, limit=limit, after_seq=after_seq) + return [{"seq": 1, "event_type": "subagent.step"}] + + class FakeState: + run_event_store = FakeStore() + + class FakeApp: + state = FakeState() + + class FakeRequest: + app = FakeApp() + _deerflow_test_bypass_auth = True + + result = await list_run_events( + thread_id="t1", + run_id="r1", + request=FakeRequest(), + event_types="subagent.start,subagent.step,subagent.end", + task_id="task-A", + limit=500, + after_seq=7, + ) + + assert result == [{"seq": 1, "event_type": "subagent.step"}] + assert calls["task_id"] == "task-A" + assert calls["after_seq"] == 7 + assert calls["event_types"] == ["subagent.start", "subagent.step", "subagent.end"] diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py index 87f80cc52..2c01d0abe 100644 --- a/backend/tests/test_subagent_executor.py +++ b/backend/tests/test_subagent_executor.py @@ -79,7 +79,7 @@ def _setup_executor_classes(): sys.modules["deerflow.skills.storage"] = storage_module # Import real classes inside fixture - from langchain_core.messages import AIMessage, HumanMessage + from langchain_core.messages import AIMessage, HumanMessage, ToolMessage from deerflow.subagents.config import SubagentConfig from deerflow.subagents.executor import ( @@ -100,6 +100,7 @@ def _setup_executor_classes(): classes = { "AIMessage": AIMessage, "HumanMessage": HumanMessage, + "ToolMessage": ToolMessage, "SubagentConfig": SubagentConfig, "SubagentExecutor": SubagentExecutor, "SubagentResult": SubagentResult, @@ -229,6 +230,12 @@ class _MsgHelper: msg.id = msg_id return msg + def tool(self, content, tool_call_id, name=None, msg_id=None): + msg = self.classes["ToolMessage"](content=content, tool_call_id=tool_call_id, name=name) + if msg_id: + msg.id = msg_id + return msg + @pytest.fixture def msg(classes): @@ -805,6 +812,34 @@ class TestAsyncExecutionPath: assert [m["content"] for m in result.ai_messages] == ["same", "different"] + @pytest.mark.anyio + async def test_aexecute_captures_all_tool_outputs_from_one_super_step(self, classes, base_config, mock_agent, msg): + """Regression for #3779: when the model emits several tool calls in one + turn, LangGraph's ToolNode appends all their ToolMessages in a single + ``values`` super-step. Capturing only ``messages[-1]`` dropped every tool + output but the last; all three must now survive in ``ai_messages``.""" + SubagentExecutor = classes["SubagentExecutor"] + + human = msg.human("Task") + ai_turn = msg.ai("running three tools", "ai-1") + t1 = msg.tool("result 1", "call_1", name="web_search", msg_id="tool-1") + t2 = msg.tool("result 2", "call_2", name="read_file", msg_id="tool-2") + t3 = msg.tool("result 3", "call_3", name="web_search", msg_id="tool-3") + final = msg.ai("done", "ai-2") + chunks = [ + {"messages": [human, ai_turn]}, + # One super-step appends all three ToolMessages at once. + {"messages": [human, ai_turn, t1, t2, t3]}, + {"messages": [human, ai_turn, t1, t2, t3, final]}, + ] + mock_agent.astream = lambda *args, **kwargs: async_iterator(chunks) + + executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread") + with patch.object(executor, "_create_agent", return_value=mock_agent): + result = await executor._aexecute("Task") + + assert [m["id"] for m in result.ai_messages] == ["ai-1", "tool-1", "tool-2", "tool-3", "ai-2"] + @pytest.mark.anyio async def test_aexecute_handles_list_content(self, classes, base_config, mock_agent, msg): """Test handling of list-type content in AIMessage.""" diff --git a/backend/tests/test_subagent_step_events.py b/backend/tests/test_subagent_step_events.py new file mode 100644 index 000000000..4193d4598 --- /dev/null +++ b/backend/tests/test_subagent_step_events.py @@ -0,0 +1,294 @@ +"""Tests for the pure subagent step-payload builder (issue #3779). + +``build_subagent_step`` turns a captured subagent message dict (the +``model_dump()`` of an AIMessage or ToolMessage) into the compact, +serializable step payload that is both streamed (``task_running``) and +persisted (``subagent.step`` run events). It is a pure function so it can +be unit-tested without the executor/graph. +""" + +from __future__ import annotations + +from langchain_core.messages import AIMessage, HumanMessage, ToolMessage + +from deerflow.subagents.step_events import ( + SUBAGENT_EVENT_CATEGORY, + SUBAGENT_STEP_MAX_CHARS, + build_subagent_step, + capture_new_step_messages, + capture_step_message, + subagent_run_event, + truncate_step_text, +) + + +def test_ai_message_becomes_ai_step_with_tool_calls(): + message = { + "type": "ai", + "id": "ai-1", + "content": "Let me search the web.", + "tool_calls": [ + {"name": "web_search", "args": {"query": "deerflow"}, "id": "call_1", "type": "tool_call"}, + ], + } + + step = build_subagent_step(message, task_id="call_task", message_index=1) + + assert step["task_id"] == "call_task" + assert step["message_index"] == 1 + assert step["kind"] == "ai" + assert step["text"] == "Let me search the web." + assert step["truncated"] is False + assert step["tool_calls"] == [{"name": "web_search", "args": {"query": "deerflow"}}] + assert "tool_name" not in step + + +def test_tool_message_becomes_tool_step_with_output(): + message = { + "type": "tool", + "id": "tool-1", + "name": "web_search", + "tool_call_id": "call_1", + "content": "Result: DeerFlow is a LangGraph super-agent.", + } + + step = build_subagent_step(message, task_id="call_task", message_index=2) + + assert step["kind"] == "tool" + assert step["tool_name"] == "web_search" + assert step["text"] == "Result: DeerFlow is a LangGraph super-agent." + assert step["truncated"] is False + assert "tool_calls" not in step + + +def test_long_tool_output_is_truncated_and_flagged(): + big = "x" * (SUBAGENT_STEP_MAX_CHARS + 500) + message = {"type": "tool", "name": "read_file", "content": big} + + step = build_subagent_step(message, task_id="t", message_index=3, max_chars=SUBAGENT_STEP_MAX_CHARS) + + assert step["truncated"] is True + assert len(step["text"]) == SUBAGENT_STEP_MAX_CHARS + + +def test_list_content_blocks_are_flattened_to_text(): + message = { + "type": "ai", + "content": [ + {"type": "text", "text": "first"}, + {"type": "text", "text": "second"}, + ], + "tool_calls": [], + } + + step = build_subagent_step(message, task_id="t", message_index=1) + + assert "first" in step["text"] + assert "second" in step["text"] + assert step["tool_calls"] == [] + + +def test_ai_text_is_also_truncated(): + big = "y" * (SUBAGENT_STEP_MAX_CHARS + 10) + message = {"type": "ai", "content": big, "tool_calls": []} + + step = build_subagent_step(message, task_id="t", message_index=1, max_chars=SUBAGENT_STEP_MAX_CHARS) + + assert step["truncated"] is True + assert len(step["text"]) == SUBAGENT_STEP_MAX_CHARS + + +def test_truncate_step_text_helper(): + assert truncate_step_text("abc", 10) == ("abc", False) + assert truncate_step_text("abcdef", 3) == ("abc", True) + + +def test_capture_ai_message_appends_dict(): + captured: list[dict] = [] + seen: set[str] = set() + + appended = capture_step_message(AIMessage(content="hi", id="ai-1"), captured, seen) + + assert appended is True + assert len(captured) == 1 + assert captured[0]["type"] == "ai" + + +def test_capture_tool_message_is_now_captured(): + # Regression for #3779: tool outputs (ToolMessage) used to be dropped, + # so "what each step produced" never reached the UI/store. + captured: list[dict] = [] + seen: set[str] = set() + + appended = capture_step_message( + ToolMessage(content="search results", tool_call_id="call_1", name="web_search", id="tool-1"), + captured, + seen, + ) + + assert appended is True + assert captured[0]["type"] == "tool" + assert captured[0]["name"] == "web_search" + + +def test_capture_dedupes_by_id(): + captured: list[dict] = [] + seen: set[str] = set() + msg = AIMessage(content="hi", id="ai-1") + + assert capture_step_message(msg, captured, seen) is True + assert capture_step_message(msg, captured, seen) is False + assert len(captured) == 1 + + +def test_capture_ignores_human_message(): + captured: list[dict] = [] + seen: set[str] = set() + + appended = capture_step_message(HumanMessage(content="user input", id="h-1"), captured, seen) + + assert appended is False + assert captured == [] + + +def test_none_content_flattens_to_empty_string(): + # A tool-call-only AI turn can carry content=None; it must render as "" (not + # the literal "None"), matching the shared message_content_to_text guard. + message = {"type": "ai", "content": None, "tool_calls": []} + + step = build_subagent_step(message, task_id="t", message_index=1) + + assert step["text"] == "" + + +def test_ai_step_caps_large_tool_call_args(): + # Regression for #3779: build_subagent_step capped `text` but copied + # `tool_calls[].args` verbatim, so a write_file/bash call carrying a big + # payload produced an unbounded persisted row. Args must now be capped too. + big_payload = "F" * (SUBAGENT_STEP_MAX_CHARS + 4096) + message = { + "type": "ai", + "content": "writing the file", + "tool_calls": [ + {"name": "write_file", "args": {"path": "/mnt/out.txt", "content": big_payload}}, + ], + } + + step = build_subagent_step(message, task_id="t", message_index=1, max_chars=SUBAGENT_STEP_MAX_CHARS) + + call = step["tool_calls"][0] + assert call["name"] == "write_file" + assert call["args_truncated"] is True + # The serialized args are bounded by the same cap the text field uses. + assert isinstance(call["args"], str) + assert len(call["args"]) == SUBAGENT_STEP_MAX_CHARS + + +def test_ai_step_keeps_small_tool_call_args_structured(): + message = { + "type": "ai", + "content": "searching", + "tool_calls": [{"name": "web_search", "args": {"query": "deerflow"}}], + } + + step = build_subagent_step(message, task_id="t", message_index=1) + + call = step["tool_calls"][0] + assert call["args"] == {"query": "deerflow"} + assert "args_truncated" not in call + + +def test_capture_new_step_messages_captures_full_multi_tool_tail(): + # Regression for #3779: a single super-step can append several ToolMessages + # (one per tool call in a multi-tool turn). Capturing only messages[-1] + # dropped all but the last; the tail walk must capture every new message. + captured: list[dict] = [] + seen: set[str] = set() + + # Chunk 1: human + one AIMessage requesting 3 tool calls. + chunk1 = [ + HumanMessage(content="do work", id="h-1"), + AIMessage(content="running tools", id="ai-1"), + ] + processed = capture_new_step_messages(chunk1, captured, seen, 0) + assert processed == 2 + assert [c["id"] for c in captured] == ["ai-1"] + + # Chunk 2: values-mode re-yields the whole history plus 3 new ToolMessages + # appended in one super-step. + chunk2 = chunk1 + [ + ToolMessage(content="r1", tool_call_id="c1", name="web_search", id="tool-1"), + ToolMessage(content="r2", tool_call_id="c2", name="read_file", id="tool-2"), + ToolMessage(content="r3", tool_call_id="c3", name="web_search", id="tool-3"), + ] + processed = capture_new_step_messages(chunk2, captured, seen, processed) + + assert processed == 5 + # All three tool outputs survive, not just the last. + assert [c["id"] for c in captured] == ["ai-1", "tool-1", "tool-2", "tool-3"] + + +def test_capture_new_step_messages_is_noop_on_values_reyield(): + # stream_mode="values" re-yields the same trailing message with unchanged + # length; re-processing must not duplicate captures. + captured: list[dict] = [] + seen: set[str] = set() + messages = [AIMessage(content="hi", id="ai-1")] + + processed = capture_new_step_messages(messages, captured, seen, 0) + assert processed == 1 + # Same list handed back (no growth) — cursor already at the end. + processed = capture_new_step_messages(messages, captured, seen, processed) + assert processed == 1 + assert len(captured) == 1 + + +def test_run_event_for_task_started(): + record = subagent_run_event({"type": "task_started", "task_id": "call_1", "description": "research X"}) + + assert record["event_type"] == "subagent.start" + assert record["category"] == SUBAGENT_EVENT_CATEGORY + assert record["metadata"]["task_id"] == "call_1" + assert record["content"]["description"] == "research X" + + +def test_run_event_for_task_running_carries_step_payload(): + chunk = { + "type": "task_running", + "task_id": "call_1", + "message": {"type": "tool", "name": "web_search", "content": "results"}, + "message_index": 2, + } + + record = subagent_run_event(chunk) + + assert record["event_type"] == "subagent.step" + assert record["category"] == SUBAGENT_EVENT_CATEGORY + assert record["metadata"] == {"task_id": "call_1", "message_index": 2} + assert record["content"] == build_subagent_step(chunk["message"], task_id="call_1", message_index=2) + + +def test_run_event_for_terminal_status(): + record = subagent_run_event({"type": "task_completed", "task_id": "call_1", "result": "done"}) + + assert record["event_type"] == "subagent.end" + assert record["content"]["status"] == "completed" + assert record["content"]["result"] == "done" + + failed = subagent_run_event({"type": "task_failed", "task_id": "call_1", "error": "boom"}) + assert failed["content"]["status"] == "failed" + assert failed["content"]["error"] == "boom" + + +def test_run_event_terminal_result_is_truncated(): + big = "z" * (SUBAGENT_STEP_MAX_CHARS + 100) + record = subagent_run_event({"type": "task_completed", "task_id": "c1", "result": big}) + + assert len(record["content"]["result"]) == SUBAGENT_STEP_MAX_CHARS + assert record["content"]["result_truncated"] is True + + +def test_run_event_ignores_non_task_chunks(): + assert subagent_run_event({"type": "something_else"}) is None + assert subagent_run_event({"no_type": True}) is None + assert subagent_run_event("not-a-dict") is None diff --git a/backend/tests/test_worker_subagent_persistence.py b/backend/tests/test_worker_subagent_persistence.py new file mode 100644 index 000000000..fbd77b1c0 --- /dev/null +++ b/backend/tests/test_worker_subagent_persistence.py @@ -0,0 +1,175 @@ +"""Worker-side persistence of subagent step events (issue #3779). + +The worker streams ``task_*`` custom events to the SSE bridge for live display. +``_SubagentEventBuffer`` additionally writes them to the RunEventStore so the +subtask card's full step history survives a reload. This module tests that glue: +recognized events are buffered and flushed via ``put_batch`` (not per-event +``put``, which the store documents as a low-frequency path), unknown chunks are +skipped, a missing store is a no-op, terminal events flush eagerly, and store +failures never bubble into the stream loop. +""" + +from __future__ import annotations + +import os +import subprocess +import sys + +import pytest + +from deerflow.runtime.runs.worker import _SubagentEventBuffer + + +def test_worker_imports_first_without_circular_import(): + """Gateway startup imports worker early; importing it first must not trigger + a circular import through deerflow.subagents (regression for the #3779 fix). + + pytest preloads many modules, so the cycle only reproduces when worker is the + first deerflow import — hence a clean subprocess. + """ + repo_backend = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + env = {**os.environ, "PYTHONPATH": repo_backend} + result = subprocess.run( + [sys.executable, "-c", "import deerflow.runtime.runs.worker"], + capture_output=True, + text=True, + env=env, + ) + assert result.returncode == 0, result.stderr + + +class _FakeStore: + def __init__(self): + self.puts: list[dict] = [] + self.batches: list[list[dict]] = [] + + async def put(self, **kwargs): + self.puts.append(kwargs) + return kwargs + + async def put_batch(self, events): + # Copy so later buffer reuse can't mutate what we recorded. + self.batches.append([dict(e) for e in events]) + return list(events) + + +class _BoomStore: + async def put_batch(self, events): + raise RuntimeError("db down") + + +def _running_step(task_id="call_1", message_index=1): + return { + "type": "task_running", + "task_id": task_id, + "message": {"type": "tool", "name": "web_search", "content": "results"}, + "message_index": message_index, + } + + +@pytest.mark.asyncio +async def test_steps_are_buffered_not_put_per_event(): + # Steps must not hit the low-frequency put() path; they accumulate until flush. + store = _FakeStore() + buffer = _SubagentEventBuffer(store, "thread_1", "run_1") + + await buffer.add(_running_step(message_index=1)) + await buffer.add(_running_step(message_index=2)) + + assert store.puts == [] # never uses the per-event put path + assert store.batches == [] # nothing flushed yet + + await buffer.flush() + + assert len(store.batches) == 1 + batch = store.batches[0] + assert [e["metadata"]["message_index"] for e in batch] == [1, 2] + assert all(e["thread_id"] == "thread_1" and e["run_id"] == "run_1" for e in batch) + assert all(e["event_type"] == "subagent.step" and e["category"] == "subagent" for e in batch) + + +@pytest.mark.asyncio +async def test_flush_is_idempotent_when_empty(): + store = _FakeStore() + buffer = _SubagentEventBuffer(store, "t", "r") + + await buffer.flush() # nothing buffered + await buffer.flush() + + assert store.batches == [] + + +@pytest.mark.asyncio +async def test_terminal_event_flushes_eagerly(): + # A completed subagent's steps should be durable promptly, not stuck in the + # buffer until the whole run ends. + store = _FakeStore() + buffer = _SubagentEventBuffer(store, "thread_1", "run_1") + + await buffer.add(_running_step(message_index=1)) + await buffer.add({"type": "task_completed", "task_id": "call_1", "result": "done"}) + + assert len(store.batches) == 1 + batch = store.batches[0] + assert [e["event_type"] for e in batch] == ["subagent.step", "subagent.end"] + + +@pytest.mark.asyncio +async def test_size_threshold_triggers_flush(): + store = _FakeStore() + buffer = _SubagentEventBuffer(store, "thread_1", "run_1") + + for i in range(_SubagentEventBuffer.FLUSH_THRESHOLD): + await buffer.add(_running_step(message_index=i + 1)) + + # Reaching the threshold flushes without waiting for the run to end. + assert len(store.batches) == 1 + assert len(store.batches[0]) == _SubagentEventBuffer.FLUSH_THRESHOLD + + +@pytest.mark.asyncio +async def test_skips_non_task_chunk(): + store = _FakeStore() + buffer = _SubagentEventBuffer(store, "t", "r") + + await buffer.add({"type": "messages"}) + await buffer.flush() + + assert store.batches == [] + + +@pytest.mark.asyncio +async def test_missing_store_is_noop(): + # Must not raise when run_events is not configured. + buffer = _SubagentEventBuffer(None, "t", "r") + await buffer.add({"type": "task_started", "task_id": "c1"}) + await buffer.flush() + + +@pytest.mark.asyncio +async def test_store_errors_do_not_propagate(): + # A persistence failure must never break the live stream loop. + buffer = _SubagentEventBuffer(_BoomStore(), "t", "r") + await buffer.add(_running_step()) + await buffer.flush() # BoomStore raises inside; must be swallowed + + +@pytest.mark.asyncio +async def test_roundtrip_step_is_listable_but_not_in_message_feed(): + # End-to-end against the real in-memory store: a persisted subagent step is + # retrievable via list_events (fetch-on-expand) yet never leaks into the + # thread message feed (list_messages), which filters category == "message". + from deerflow.runtime.events.store.memory import MemoryRunEventStore + + store = MemoryRunEventStore() + buffer = _SubagentEventBuffer(store, "thread_1", "run_1") + + await buffer.add(_running_step(message_index=1)) + await buffer.flush() + + events = await store.list_events("thread_1", "run_1", event_types=["subagent.step"]) + assert len(events) == 1 + assert events[0]["metadata"]["task_id"] == "call_1" + + messages = await store.list_messages("thread_1") + assert messages == [] diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index 85a1d86f8..bd0a200a3 100644 --- a/frontend/AGENTS.md +++ b/frontend/AGENTS.md @@ -75,6 +75,7 @@ The frontend is a stateful chat application. Users create **threads** (conversat - **Thread hooks** (`useThreadStream`, `useSubmitThread`, `useThreads`) are the primary API interface - **LangGraph client** is a singleton obtained via `getAPIClient()` in `core/api/` - **Environment validation** uses `@t3-oss/env-nextjs` with Zod schemas (`src/env.js`). Skip with `SKIP_ENV_VALIDATION=1` +- **Subtask step history** (`core/tasks/`) — the subtask card shows a subagent's full step timeline (#3779): its assistant reasoning turns interleaved with the tools it ran. `Subtask.steps[]` is accumulated live from `task_running` events (appended via `mergeSteps`, not overwritten) and backfilled on expand for historical runs by `fetchSubtaskSteps`, which pages the events endpoint scoped to one task (GET `/runs/{runId}/events?event_types=subagent.step&task_id=…&after_seq=…`) until a short page, so the run-wide limit can't truncate the timeline. `core/tasks/steps.ts` is the pure model: `messageToStep` (live), `eventsToSteps` (reload), `mergeSteps` (dedup by `message_index`), and `stepsForDisplay` (what the card renders — keeps tool steps + AI steps with text, drops the trailing final-answer AI step when completed since it's shown as `result`). `core/tasks/subtask-update.ts::computeNextSubtask` is the pure per-subtask state transition (merge step deltas, keep terminal status stable); `core/tasks/context.tsx`'s `useUpdateSubtask` applies it against a `tasksRef` mirroring the latest state (not a closure snapshot), so a late-resolving `fetchSubtaskSteps` backfill merges into current state instead of clobbering SSE steps or sibling subtasks that arrived meanwhile. The owning `run_id` is carried onto history content messages in `buildVisibleHistoryMessages` so the card can resolve the events endpoint. ### Interaction Ownership diff --git a/frontend/src/components/workspace/messages/message-list.tsx b/frontend/src/components/workspace/messages/message-list.tsx index 2c99640ac..2d979352e 100644 --- a/frontend/src/components/workspace/messages/message-list.tsx +++ b/frontend/src/components/workspace/messages/message-list.tsx @@ -551,6 +551,8 @@ export function MessageList({ , ); diff --git a/frontend/src/components/workspace/messages/subtask-card.tsx b/frontend/src/components/workspace/messages/subtask-card.tsx index ea82336e3..3e22f913d 100644 --- a/frontend/src/components/workspace/messages/subtask-card.tsx +++ b/frontend/src/components/workspace/messages/subtask-card.tsx @@ -3,9 +3,11 @@ import { ChevronUp, ClipboardListIcon, Loader2Icon, + SparklesIcon, + WrenchIcon, XCircleIcon, } from "lucide-react"; -import { useMemo, useState } from "react"; +import { useEffect, useMemo, useRef, useState } from "react"; import { ChainOfThought, @@ -20,7 +22,9 @@ import { hasToolCalls } from "@/core/messages/utils"; import { useRehypeSplitWordsIntoSpans } from "@/core/rehype"; import { streamdownPluginsWithWordAnimation } from "@/core/streamdown"; import { SafeStreamdown } from "@/core/streamdown/components"; -import { useSubtask } from "@/core/tasks/context"; +import { fetchSubtaskSteps } from "@/core/tasks/api"; +import { useSubtask, useUpdateSubtask } from "@/core/tasks/context"; +import { stepsForDisplay } from "@/core/tasks/steps"; import { explainLastToolCall } from "@/core/tools/utils"; import { cn } from "@/lib/utils"; @@ -32,16 +36,50 @@ import { MarkdownContent } from "./markdown-content"; export function SubtaskCard({ className, taskId, + threadId, + runId, isLoading, }: { className?: string; taskId: string; + threadId?: string; + runId?: string; isLoading: boolean; }) { const { t } = useI18n(); const [collapsed, setCollapsed] = useState(true); const rehypePlugins = useRehypeSplitWordsIntoSpans(isLoading); const task = useSubtask(taskId)!; + const updateSubtask = useUpdateSubtask(); + + // The card shows the subagent's step timeline (#3779): its reasoning turns + // (AI text) interleaved with the tools it ran (by name). See stepsForDisplay + // for what is kept/dropped. + const displaySteps = stepsForDisplay(task.steps, task.status); + + // Backfill step history on expand for historical runs (#3779). Live runs + // already have steps from SSE, so the `steps.length` guard skips the fetch. + const stepsCount = task.steps?.length ?? 0; + const backfilledRef = useRef(false); + useEffect(() => { + if (collapsed || backfilledRef.current || stepsCount > 0) { + return; + } + if (!threadId || !runId) { + return; + } + backfilledRef.current = true; + fetchSubtaskSteps(threadId, runId, taskId) + .then((steps) => { + if (steps.length > 0) { + updateSubtask({ id: taskId, steps }); + } + }) + .catch(() => { + // Allow a retry on the next expand if the fetch failed. + backfilledRef.current = false; + }); + }, [collapsed, stepsCount, threadId, runId, taskId, updateSubtask]); const icon = useMemo(() => { if (task.status === "completed") { return ; @@ -135,16 +173,36 @@ export function SubtaskCard({ } > )} - {task.status === "in_progress" && - task.latestMessage && - hasToolCalls(task.latestMessage) && ( + {displaySteps.map((step, i) => { + const isLastWhileRunning = + task.status === "in_progress" && i === displaySteps.length - 1; + const icon = isLastWhileRunning ? ( + + ) : step.kind === "tool" ? ( + + ) : ( + + ); + return ( } - > - {explainLastToolCall(task.latestMessage, t)} - - )} + key={`${step.message_index}-${i}`} + label={ + step.kind === "tool" ? ( + (step.tool_name ?? t.subtasks[task.status]) + ) : ( +
+ +
+ ) + } + icon={icon} + /> + ); + })} {task.status === "completed" && ( <> [0][number] & { + seq?: number; +}; + +/** + * Fetch a subtask's persisted step history for a historical run (#3779). + * + * Scoped server-side to this `taskId` (and to `subagent.step` events) and paged + * forward with an `after_seq` cursor until a short page, so the run-wide event + * limit can never truncate a subagent's step timeline — even for long runs or + * runs with several subagents. Used by the subtask card to backfill steps on + * expand when the live SSE steps are gone (e.g. after a page reload). + */ +export async function fetchSubtaskSteps( + threadId: string, + runId: string, + taskId: string, + pageSize: number = SUBTASK_STEPS_PAGE_SIZE, +): Promise { + const base = `${getBackendBaseURL()}/api/threads/${encodeURIComponent( + threadId, + )}/runs/${encodeURIComponent(runId)}/events`; + + const events: FetchedEvent[] = []; + let afterSeq: number | undefined; + + for (let page = 0; page < SUBTASK_STEPS_MAX_PAGES; page++) { + const params = new URLSearchParams({ + event_types: "subagent.step", + task_id: taskId, + limit: String(pageSize), + }); + if (afterSeq !== undefined) { + params.set("after_seq", String(afterSeq)); + } + + const res = await fetch(`${base}?${params.toString()}`); + if (!res.ok) { + throw new Error(`Failed to fetch subtask steps: ${res.status}`); + } + const batch = (await res.json()) as FetchedEvent[]; + events.push(...batch); + + if (batch.length < pageSize) { + break; + } + const lastSeq = batch[batch.length - 1]?.seq; + if (lastSeq === undefined) { + break; // can't advance the cursor; stop rather than refetch page 0 forever + } + afterSeq = lastSeq; + } + + return eventsToSteps(events, taskId); +} diff --git a/frontend/src/core/tasks/context.tsx b/frontend/src/core/tasks/context.tsx index 52bfc7d4e..71116a326 100644 --- a/frontend/src/core/tasks/context.tsx +++ b/frontend/src/core/tasks/context.tsx @@ -7,19 +7,21 @@ import { useState, } from "react"; +import { computeNextSubtask } from "./subtask-update"; import type { Subtask } from "./types"; -function isTerminalSubtaskStatus(status: Subtask["status"] | undefined) { - return status === "completed" || status === "failed"; -} - export interface SubtaskContextValue { tasks: Record; + // Always mirrors the latest `tasks` (updated during render). `updateSubtask` + // reads/writes through this instead of a closure snapshot so async callers + // (e.g. a late-resolving backfill) merge into current state, not stale state. + tasksRef: React.RefObject>; setTasks: (tasks: Record) => void; } export const SubtaskContext = createContext({ tasks: {}, + tasksRef: { current: {} }, setTasks: () => { /* noop */ }, @@ -27,8 +29,12 @@ export const SubtaskContext = createContext({ export function SubtasksProvider({ children }: { children: React.ReactNode }) { const [tasks, setTasks] = useState>({}); + const tasksRef = useRef(tasks); + // Keep the ref pointing at the freshest state on every render so reads in + // async callbacks (backfill `.then`) never see a stale map. + tasksRef.current = tasks; return ( - + {children} ); @@ -50,7 +56,7 @@ export function useSubtask(id: string) { } export function useUpdateSubtask() { - const { tasks, setTasks } = useSubtaskContext(); + const { tasksRef, setTasks } = useSubtaskContext(); const shouldNotifyAfterRenderRef = useRef(false); // No deps: must run after every render to check the ref set during render. useEffect(() => { @@ -58,37 +64,32 @@ export function useUpdateSubtask() { return; } shouldNotifyAfterRenderRef.current = false; - setTasks({ ...tasks }); + setTasks({ ...tasksRef.current }); }); const updateSubtask = useCallback( (task: Partial & { id: string }) => { - const previous = tasks[task.id]; - const previousStatus = previous?.status; - // MessageList writes the pending task tool-call state before parsing the - // matching ToolMessage in the same render. Keep terminal results stable - // across the next render so the refresh notification does not loop. - const next = { - ...previous, - ...task, - ...(task.status === "in_progress" && - isTerminalSubtaskStatus(previousStatus) - ? { status: previousStatus } - : {}), - } as Subtask; + // Read the *latest* state via the ref, never a `tasks` snapshot captured in + // this callback's closure. Without this, an in-flight + // fetchSubtaskSteps().then(updateSubtask) resolving late would write a stale + // map, clobbering SSE steps/status and sibling subtasks added meanwhile (#3779). + const current = tasksRef.current; + const { next, becameTerminal } = computeNextSubtask( + current[task.id], + task, + ); - const becameTerminal = - isTerminalSubtaskStatus(next.status) && previousStatus !== next.status; + current[task.id] = next; - tasks[task.id] = next; - - if (task.latestMessage) { - setTasks({ ...tasks }); + if (task.latestMessage || task.steps) { + setTasks({ ...current }); } else if (becameTerminal) { + // Defer the render to the after-render effect so a terminal-only update + // does not loop with MessageList's same-render pending write. shouldNotifyAfterRenderRef.current = true; } }, - [tasks, setTasks], + [tasksRef, setTasks], ); return updateSubtask; diff --git a/frontend/src/core/tasks/steps.ts b/frontend/src/core/tasks/steps.ts new file mode 100644 index 000000000..f3760f2f6 --- /dev/null +++ b/frontend/src/core/tasks/steps.ts @@ -0,0 +1,169 @@ +/** + * Subtask step model shared by the live (SSE) and reload (fetched) paths. + * + * Issue #3779: the subtask card used to keep only the latest subagent message, + * so earlier steps flashed by and nothing survived a reload. A `SubtaskStep` is + * the normalized, renderable unit of subagent progress — one assistant turn + * (`kind: "ai"`, carrying its tool-call requests) or one tool result + * (`kind: "tool"`, carrying the tool's output). The backend persists the same + * shape as `subagent.step` run-event content; `messageToStep` mirrors that + * shaping for the live `task_running` event, which still carries the raw message. + */ + +export interface SubtaskStepToolCall { + name?: string; + args?: unknown; +} + +export interface SubtaskStep { + message_index: number; + kind: "ai" | "tool"; + text: string; + truncated?: boolean; + tool_calls?: SubtaskStepToolCall[]; + tool_name?: string; +} + +type RawMessage = { + type?: string; + content?: unknown; + name?: string; + tool_calls?: { name?: string; args?: unknown; [key: string]: unknown }[]; + [key: string]: unknown; +}; + +function contentToText(content: unknown): string { + if (typeof content === "string") { + return content; + } + if (Array.isArray(content)) { + return content + .map((block) => { + if (typeof block === "string") { + return block; + } + if (block && typeof block === "object" && "text" in block) { + const text = (block as { text?: unknown }).text; + return typeof text === "string" ? text : ""; + } + return ""; + }) + .filter(Boolean) + .join("\n"); + } + return ""; +} + +/** Normalize a raw subagent message (live `task_running` payload) into a step. */ +export function messageToStep( + message: RawMessage, + messageIndex: number, +): SubtaskStep { + const kind = message.type === "tool" ? "tool" : "ai"; + const step: SubtaskStep = { + message_index: messageIndex, + kind, + text: contentToText(message.content), + }; + + if (kind === "tool") { + step.tool_name = message.name; + } else { + step.tool_calls = (message.tool_calls ?? []).map((call) => ({ + name: call.name, + args: call.args, + })); + } + + return step; +} + +/** + * Steps to render in the subtask card timeline (#3779). Interleaves the + * subagent's assistant turns and tool steps, ordered by `message_index`: + * + * - tool steps are always kept (one "the subagent ran " row each); + * - AI steps are kept only when they carry visible reasoning text — a turn that + * only requests tools (blank text) adds no information beyond the tool rows + * that follow it, so it is dropped; + * - when the task is `completed`, a trailing AI step with no tool_calls is the + * subagent's final answer, which the card already renders as `task.result`, + * so it is dropped here to avoid showing the answer twice. + */ +export function stepsForDisplay( + steps: SubtaskStep[] | undefined, + status: "in_progress" | "completed" | "failed", +): SubtaskStep[] { + const visible = (steps ?? []) + .filter((step) => step.kind === "tool" || step.text.trim() !== "") + .sort((a, b) => a.message_index - b.message_index); + + if (status === "completed") { + const last = visible[visible.length - 1]; + if (last?.kind === "ai" && !last?.tool_calls?.length) { + return visible.slice(0, -1); + } + } + return visible; +} + +type RunEvent = { + event_type?: string; + content?: unknown; + metadata?: { task_id?: string } & Record; +}; + +/** + * Map persisted run events (from `GET /{rid}/events`) into the subtask's steps, + * keeping only `subagent.step` events for `taskId` and ordering by message_index. + * The persisted `content` already matches the step shape (it is what the backend + * `build_subagent_step` produced), so this filters, projects, and sorts (#3779). + */ +export function eventsToSteps( + events: RunEvent[], + taskId: string, +): SubtaskStep[] { + const steps: SubtaskStep[] = []; + for (const event of events) { + if (event.event_type !== "subagent.step") { + continue; + } + const content = event.content as + | (SubtaskStep & { task_id?: string }) + | undefined; + const eventTaskId = content?.task_id ?? event.metadata?.task_id; + if (!content || eventTaskId !== taskId) { + continue; + } + steps.push({ + message_index: content.message_index, + kind: content.kind, + text: content.text ?? "", + truncated: content.truncated, + tool_calls: content.tool_calls, + tool_name: content.tool_name, + }); + } + return steps.sort((a, b) => a.message_index - b.message_index); +} + +/** + * Merge `incoming` steps into `existing`, deduping by `message_index` (incoming + * wins) and keeping the result ordered. Used to reconcile live SSE steps with + * steps fetched on expand without double-rendering shared indices. + */ +export function mergeSteps( + existing: SubtaskStep[], + incoming: SubtaskStep[], +): SubtaskStep[] { + const byIndex = new Map(); + for (const step of existing) { + byIndex.set(step.message_index, step); + } + for (const step of incoming) { + byIndex.set(step.message_index, step); + } + return [...byIndex.values()].sort( + (a, b) => a.message_index - b.message_index, + ); +} diff --git a/frontend/src/core/tasks/subtask-update.ts b/frontend/src/core/tasks/subtask-update.ts new file mode 100644 index 000000000..994ca6aea --- /dev/null +++ b/frontend/src/core/tasks/subtask-update.ts @@ -0,0 +1,47 @@ +import { mergeSteps } from "./steps"; +import type { Subtask } from "./types"; + +export function isTerminalSubtaskStatus(status: Subtask["status"] | undefined) { + return status === "completed" || status === "failed"; +} + +/** + * Pure state transition for a single subtask update (#3779). + * + * Kept separate from the React hook so it can be unit-tested and, crucially, so + * the hook can compute `next` from the *latest* `previous` handed to a + * functional `setTasks` updater — not a stale `tasks` snapshot captured in a + * closure. Deriving `next` from whatever `previous` the caller passes is what + * lets an in-flight `fetchSubtaskSteps().then(...)` merge into current state + * instead of clobbering SSE steps / sibling subtasks that arrived meanwhile. + * + * `steps` are treated as deltas: they are merged into `previous.steps` + * (deduped/ordered by message_index) rather than replacing them, so live SSE + * steps and fetched-on-expand backfill build one timeline. + */ +export function computeNextSubtask( + previous: Subtask | undefined, + task: Partial & { id: string }, +): { next: Subtask; becameTerminal: boolean } { + const previousStatus = previous?.status; + + // MessageList writes the pending task tool-call state before parsing the + // matching ToolMessage in the same render. Keep terminal results stable + // across the next render so the refresh notification does not loop. + const next = { + ...previous, + ...task, + ...(task.status === "in_progress" && isTerminalSubtaskStatus(previousStatus) + ? { status: previousStatus } + : {}), + } as Subtask; + + if (task.steps) { + next.steps = mergeSteps(previous?.steps ?? [], task.steps); + } + + const becameTerminal = + isTerminalSubtaskStatus(next.status) && previousStatus !== next.status; + + return { next, becameTerminal }; +} diff --git a/frontend/src/core/tasks/types.ts b/frontend/src/core/tasks/types.ts index 98f9490d2..a61e6e1fa 100644 --- a/frontend/src/core/tasks/types.ts +++ b/frontend/src/core/tasks/types.ts @@ -1,11 +1,19 @@ import type { AIMessage } from "@langchain/langgraph-sdk"; +import type { SubtaskStep } from "./steps"; + export interface Subtask { id: string; status: "in_progress" | "completed" | "failed"; subagent_type: string; description: string; latestMessage?: AIMessage; + /** + * Full ordered step history (assistant turns + tool outputs) of the subagent. + * Accumulated live from `task_running` events and backfilled on expand for + * historical runs (#3779). Replaces the old "only latestMessage" behavior. + */ + steps?: SubtaskStep[]; prompt: string; result?: string; error?: string; diff --git a/frontend/src/core/threads/hooks.ts b/frontend/src/core/threads/hooks.ts index 2ef614ab8..8cc99d8c7 100644 --- a/frontend/src/core/threads/hooks.ts +++ b/frontend/src/core/threads/hooks.ts @@ -22,6 +22,7 @@ import { isHiddenFromUIMessage } from "../messages/utils"; import type { FileInMessage } from "../messages/utils"; import type { LocalSettings } from "../settings"; import { useUpdateSubtask } from "../tasks/context"; +import { messageToStep } from "../tasks/steps"; import type { UploadedFileInfo } from "../uploads"; import { promptInputFilePartToFile, uploadFiles } from "../uploads"; @@ -215,7 +216,13 @@ export function buildVisibleHistoryMessages( (message) => !supersededRunIds.has(message.run_id), ); return dedupeMessagesByIdentity([ - ...visibleRows.map((message) => message.content), + // 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. + ...visibleRows.map((message) => ({ + ...message.content, + run_id: message.run_id, + })), ...appendedMessages, ]); } @@ -967,8 +974,16 @@ export function useThreadStream({ type: "task_running"; task_id: string; message: AIMessage; + message_index?: number; }; - updateSubtask({ id: e.task_id, latestMessage: e.message }); + // Accumulate the full step history instead of overwriting (#3779): keep + // latestMessage for the collapsed-header tool-call hint, and append the + // normalized step (assistant turn or tool output) to the timeline. + updateSubtask({ + id: e.task_id, + latestMessage: e.message, + steps: [messageToStep(e.message, e.message_index ?? 0)], + }); return; } diff --git a/frontend/tests/unit/core/tasks/api.test.ts b/frontend/tests/unit/core/tasks/api.test.ts new file mode 100644 index 000000000..f69178498 --- /dev/null +++ b/frontend/tests/unit/core/tasks/api.test.ts @@ -0,0 +1,98 @@ +import { beforeEach, describe, expect, rs, test } from "@rstest/core"; + +rs.mock("@/core/api/fetcher", () => ({ + fetch: rs.fn(), +})); + +rs.mock("@/core/config", () => ({ + getBackendBaseURL: () => "/backend", +})); + +import { fetch as fetcher } from "@/core/api/fetcher"; +import { fetchSubtaskSteps } from "@/core/tasks/api"; + +const mockedFetch = rs.mocked(fetcher); + +function jsonResponse(status: number, body: unknown): Response { + return new Response(JSON.stringify(body), { + status, + statusText: status >= 400 ? "Error" : "OK", + headers: { "Content-Type": "application/json" }, + }); +} + +function stepEvent(seq: number, messageIndex: number, toolName: string) { + return { + event_type: "subagent.step", + seq, + content: { + task_id: "A", + message_index: messageIndex, + kind: "tool", + text: "", + tool_name: toolName, + }, + metadata: { task_id: "A", message_index: messageIndex }, + }; +} + +beforeEach(() => { + mockedFetch.mockReset(); +}); + +describe("fetchSubtaskSteps", () => { + test("scopes the request to the task and only fetches subagent.step", async () => { + mockedFetch.mockResolvedValueOnce(jsonResponse(200, [])); + + await fetchSubtaskSteps("thread 1", "run/1", "task-A"); + + const url = mockedFetch.mock.calls[0]![0] as string; + expect(url).toContain( + "/backend/api/threads/thread%201/runs/run%2F1/events", + ); + expect(url).toContain("task_id=task-A"); + expect(url).toContain("event_types=subagent.step"); + expect(url).toContain("limit="); + expect(url).not.toContain("after_seq"); + }); + + test("pages forward with after_seq until a short page, accumulating in order", async () => { + mockedFetch + .mockResolvedValueOnce( + jsonResponse(200, [ + stepEvent(10, 0, "web_search"), + stepEvent(11, 1, "read_file"), + ]), + ) + .mockResolvedValueOnce(jsonResponse(200, [stepEvent(12, 2, "bash")])); + + const steps = await fetchSubtaskSteps("t", "r", "A", 2); + + expect(steps.map((s) => s.message_index)).toEqual([0, 1, 2]); + expect(steps.map((s) => s.tool_name)).toEqual([ + "web_search", + "read_file", + "bash", + ]); + expect(mockedFetch).toHaveBeenCalledTimes(2); + expect(mockedFetch.mock.calls[0]![0] as string).not.toContain("after_seq"); + expect(mockedFetch.mock.calls[1]![0] as string).toContain("after_seq=11"); + }); + + test("stops after a single page when it is shorter than the page size", async () => { + mockedFetch.mockResolvedValueOnce( + jsonResponse(200, [stepEvent(10, 0, "web_search")]), + ); + + const steps = await fetchSubtaskSteps("t", "r", "A", 500); + + expect(steps).toHaveLength(1); + expect(mockedFetch).toHaveBeenCalledTimes(1); + }); + + test("throws when a page request fails", async () => { + mockedFetch.mockResolvedValueOnce(jsonResponse(500, { detail: "boom" })); + + await expect(fetchSubtaskSteps("t", "r", "A")).rejects.toThrow(); + }); +}); diff --git a/frontend/tests/unit/core/tasks/steps.test.ts b/frontend/tests/unit/core/tasks/steps.test.ts new file mode 100644 index 000000000..d0501f82f --- /dev/null +++ b/frontend/tests/unit/core/tasks/steps.test.ts @@ -0,0 +1,220 @@ +import { describe, expect, it } from "@rstest/core"; + +import { + eventsToSteps, + mergeSteps, + messageToStep, + stepsForDisplay, +} from "@/core/tasks/steps"; + +describe("messageToStep", () => { + it("normalizes an AI message into an ai step with tool calls", () => { + const step = messageToStep( + { + type: "ai", + id: "ai-1", + content: "Let me search.", + tool_calls: [{ name: "web_search", args: { query: "x" }, id: "c1" }], + }, + 1, + ); + + expect(step.kind).toBe("ai"); + expect(step.message_index).toBe(1); + expect(step.text).toBe("Let me search."); + expect(step.tool_calls).toEqual([ + { name: "web_search", args: { query: "x" } }, + ]); + expect(step.tool_name).toBeUndefined(); + }); + + it("normalizes a tool message into a tool step with its output", () => { + const step = messageToStep( + { type: "tool", id: "t-1", name: "web_search", content: "results" }, + 2, + ); + + expect(step.kind).toBe("tool"); + expect(step.tool_name).toBe("web_search"); + expect(step.text).toBe("results"); + expect(step.tool_calls).toBeUndefined(); + }); + + it("flattens list-of-blocks content to text", () => { + const step = messageToStep( + { + type: "ai", + content: [ + { type: "text", text: "first" }, + { type: "text", text: "second" }, + ], + }, + 1, + ); + + expect(step.text).toContain("first"); + expect(step.text).toContain("second"); + }); +}); + +describe("mergeSteps", () => { + it("appends a new step", () => { + const a = messageToStep({ type: "ai", content: "a" }, 1); + const b = messageToStep({ type: "tool", name: "x", content: "b" }, 2); + + expect(mergeSteps([a], [b])).toEqual([a, b]); + }); + + it("dedupes by message_index, preferring the incoming step", () => { + const old = messageToStep({ type: "ai", content: "old" }, 1); + const fresh = messageToStep({ type: "ai", content: "fresh" }, 1); + + const merged = mergeSteps([old], [fresh]); + + expect(merged).toHaveLength(1); + expect(merged[0]!.text).toBe("fresh"); + }); + + it("keeps steps ordered by message_index", () => { + const s1 = messageToStep({ type: "ai", content: "1" }, 1); + const s2 = messageToStep({ type: "ai", content: "2" }, 2); + const s3 = messageToStep({ type: "ai", content: "3" }, 3); + + const merged = mergeSteps([s3], [s1, s2]); + + expect(merged.map((s) => s.message_index)).toEqual([1, 2, 3]); + }); +}); + +describe("stepsForDisplay", () => { + it("keeps tool steps and AI steps that have text, ordered by message_index", () => { + const steps = [ + messageToStep( + { type: "tool", name: "web_search", content: "big result body" }, + 2, + ), + messageToStep( + { + type: "ai", + content: "Let me search", + tool_calls: [{ name: "web_search", args: {} }], + }, + 1, + ), + ]; + + const display = stepsForDisplay(steps, "in_progress"); + + expect(display.map((s) => s.message_index)).toEqual([1, 2]); + expect(display.map((s) => s.kind)).toEqual(["ai", "tool"]); + }); + + it("drops AI steps with blank text even if they have tool_calls", () => { + const steps = [ + messageToStep( + { + type: "ai", + content: " ", + tool_calls: [{ name: "web_search", args: {} }], + }, + 1, + ), + messageToStep({ type: "tool", name: "read_file", content: "x" }, 2), + ]; + + expect( + stepsForDisplay(steps, "in_progress").map((s) => s.message_index), + ).toEqual([2]); + }); + + it("drops the trailing final AI answer when completed (already shown as result)", () => { + const steps = [ + messageToStep({ type: "tool", name: "web_search", content: "x" }, 1), + messageToStep({ type: "ai", content: "The final answer is 42." }, 2), + ]; + + expect(stepsForDisplay(steps, "completed").map((s) => s.kind)).toEqual([ + "tool", + ]); + }); + + it("keeps the trailing AI step while still in progress", () => { + const steps = [ + messageToStep({ type: "tool", name: "web_search", content: "x" }, 1), + messageToStep({ type: "ai", content: "Thinking about the answer..." }, 2), + ]; + + expect(stepsForDisplay(steps, "in_progress").map((s) => s.kind)).toEqual([ + "tool", + "ai", + ]); + }); + + it("returns empty for undefined", () => { + expect(stepsForDisplay(undefined, "in_progress")).toEqual([]); + }); +}); + +describe("eventsToSteps", () => { + const events = [ + { + event_type: "subagent.start", + content: { task_id: "call_1", description: "research" }, + metadata: { task_id: "call_1" }, + }, + { + event_type: "subagent.step", + content: { + task_id: "call_1", + message_index: 2, + kind: "tool", + tool_name: "web_search", + text: "results", + truncated: false, + }, + metadata: { task_id: "call_1", message_index: 2 }, + }, + { + event_type: "subagent.step", + content: { + task_id: "call_1", + message_index: 1, + kind: "ai", + text: "searching", + tool_calls: [{ name: "web_search", args: {} }], + }, + metadata: { task_id: "call_1", message_index: 1 }, + }, + { + event_type: "subagent.step", + content: { task_id: "other", message_index: 1, kind: "ai", text: "nope" }, + metadata: { task_id: "other", message_index: 1 }, + }, + { + event_type: "subagent.end", + content: { task_id: "call_1", status: "completed", result: "done" }, + metadata: { task_id: "call_1" }, + }, + ]; + + it("maps subagent.step events for the task into ordered steps", () => { + const steps = eventsToSteps(events, "call_1"); + + expect(steps.map((s) => s.message_index)).toEqual([1, 2]); + expect(steps[0]!.kind).toBe("ai"); + expect(steps[1]!.kind).toBe("tool"); + expect(steps[1]!.tool_name).toBe("web_search"); + }); + + it("ignores steps belonging to other tasks and non-step events", () => { + const steps = eventsToSteps(events, "call_1"); + + expect(steps.every((s) => s.text !== "nope")).toBe(true); + expect(steps).toHaveLength(2); + }); + + it("returns empty array when no events match", () => { + expect(eventsToSteps(events, "missing")).toEqual([]); + expect(eventsToSteps([], "call_1")).toEqual([]); + }); +}); diff --git a/frontend/tests/unit/core/tasks/subtask-update.test.ts b/frontend/tests/unit/core/tasks/subtask-update.test.ts new file mode 100644 index 000000000..cd9ce3bed --- /dev/null +++ b/frontend/tests/unit/core/tasks/subtask-update.test.ts @@ -0,0 +1,108 @@ +import { describe, expect, it } from "@rstest/core"; + +import type { SubtaskStep } from "@/core/tasks/steps"; +import { + computeNextSubtask, + isTerminalSubtaskStatus, +} from "@/core/tasks/subtask-update"; +import type { Subtask } from "@/core/tasks/types"; + +function baseTask(overrides: Partial = {}): Subtask { + return { + id: "t1", + status: "in_progress", + subagent_type: "general-purpose", + description: "research", + prompt: "do it", + ...overrides, + }; +} + +function step(message_index: number): SubtaskStep { + return { + kind: "tool", + message_index, + text: `step ${message_index}`, + truncated: false, + }; +} + +describe("computeNextSubtask", () => { + it("merges step deltas into the provided previous, preserving both", () => { + const previous = baseTask({ steps: [step(1), step(2)] }); + + const { next } = computeNextSubtask(previous, { + id: "t1", + steps: [step(3)], + }); + + // Regression for #3779 stale-closure race: next is derived from whatever + // `previous` is passed (the functional-update latest state), so concurrently + // arrived steps are kept, not clobbered by a backfill resolving late. + expect(next.steps?.map((s) => s.message_index)).toEqual([1, 2, 3]); + }); + + it("does not drop steps present only on the latest previous", () => { + // Simulates a backfill that fired when previous had 0 steps, but by the time + // it resolves the latest previous already has SSE steps 1..2. The backfill + // brings historical steps 1..3; the merge must retain all, not reset to []. + const latestPrevious = baseTask({ steps: [step(1), step(2)] }); + + const { next } = computeNextSubtask(latestPrevious, { + id: "t1", + steps: [step(1), step(2), step(3)], + }); + + expect(next.steps?.map((s) => s.message_index)).toEqual([1, 2, 3]); + }); + + it("keeps a terminal status stable against a late in_progress write", () => { + const previous = baseTask({ status: "completed" }); + + const { next, becameTerminal } = computeNextSubtask(previous, { + id: "t1", + status: "in_progress", + }); + + expect(next.status).toBe("completed"); + expect(becameTerminal).toBe(false); + }); + + it("flags becameTerminal on the first transition to a terminal status", () => { + const previous = baseTask({ status: "in_progress" }); + + const { next, becameTerminal } = computeNextSubtask(previous, { + id: "t1", + status: "completed", + result: "done", + }); + + expect(next.status).toBe("completed"); + expect(next.result).toBe("done"); + expect(becameTerminal).toBe(true); + }); + + it("handles an undefined previous (first write for a task)", () => { + const { next, becameTerminal } = computeNextSubtask(undefined, { + id: "t1", + status: "in_progress", + subagent_type: "bash", + description: "run", + prompt: "p", + steps: [step(1)], + }); + + expect(next.id).toBe("t1"); + expect(next.steps?.map((s) => s.message_index)).toEqual([1]); + expect(becameTerminal).toBe(false); + }); +}); + +describe("isTerminalSubtaskStatus", () => { + it("recognizes terminal statuses only", () => { + expect(isTerminalSubtaskStatus("completed")).toBe(true); + expect(isTerminalSubtaskStatus("failed")).toBe(true); + expect(isTerminalSubtaskStatus("in_progress")).toBe(false); + expect(isTerminalSubtaskStatus(undefined)).toBe(false); + }); +}); diff --git a/frontend/tests/unit/core/threads/message-merge.test.ts b/frontend/tests/unit/core/threads/message-merge.test.ts index 264e346d9..dcc05f2e5 100644 --- a/frontend/tests/unit/core/threads/message-merge.test.ts +++ b/frontend/tests/unit/core/threads/message-merge.test.ts @@ -453,12 +453,29 @@ 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, - newAi, + { ...newHuman, run_id: "run-new" }, + { ...newAi, run_id: "run-new" }, ]); }); +test("buildVisibleHistoryMessages attaches run_id to each content message (#3779)", () => { + const rows: RunMessage[] = [ + { + run_id: "run-1", + content: { id: "ai-1", type: "ai", content: "answer" } as Message, + metadata: { caller: "lead_agent" }, + created_at: "2026-06-26T00:00:00Z", + }, + ]; + + const result = buildVisibleHistoryMessages(rows, new Set(), []); + + expect((result[0] as { run_id?: string }).run_id).toBe("run-1"); +}); + test("loading runs in newest-first order and prepending pages yields chronological messages (regression for #3352)", () => { // Simulate backend list_by_thread returning newest first. const runs = [