mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-21 13:38:05 +00:00
* feat(subagents): persist and display subagent step history (#3779) Capture both assistant turns and tool outputs during subagent execution, stream them in task_running events, and persist them as subagent.* run events so the subtask card's step timeline survives a reload. Backend: - step_events.py: pure layer (capture_step_message, build_subagent_step, subagent_run_event) shared by streaming and persistence - executor.py: capture ToolMessage outputs, not just AIMessage turns - worker.py: persist task_* custom events to RunEventStore (category "subagent" keeps them out of the thread feed; list_events backfills) Frontend: - core/tasks/steps.ts + api.ts: SubtaskStep model, messageToStep, eventsToSteps, mergeSteps, fetchSubtaskSteps - subtask card accumulates live steps and backfills on expand - carry run_id onto history content messages for the events endpoint * fix(subagents): show AI turns in subtask card + paginate step backfill (#3779) Two follow-ups to the subagent step-history feature: Problem 1 — reload backfill could silently truncate the step timeline because list_events capped at 500 events (seq-ASC) across the whole run. Add task_id filtering + an after_seq forward cursor to list_events (all three stores + abstract base + the /events route), and make fetchSubtaskSteps page through one task's subagent.step events until a short page. No schema migration: the DB filter rides the existing run-scoped index via event_metadata["task_id"]. Problem 2 — the card only rendered tool steps, so persisted AI turns were never shown. Replace toolStepsForDisplay with stepsForDisplay: interleave AI reasoning turns (with text) and tool steps by message_index, drop blank-text AI turns, and drop the trailing final-answer AI turn when completed (already shown as result). Card renders AI steps as muted clamped markdown with a sparkles icon. Tests: store task_id/after_seq filtering + pagination across memory/db/jsonl, the /events route forwarding, stepsForDisplay rules, and fetchSubtaskSteps pagination. Docs updated in both AGENTS.md. * make format * fix(subagents): capture full multi-tool step tail, batch step persistence, cap tool-call args (#3779) Address PR review findings on the subagent step-history feature: 1. executor.py streamed on stream_mode="values" and captured only messages[-1] per chunk, so a multi-tool-call turn (ToolNode appends one ToolMessage per call in a single super-step) lost all but the last tool output in both the live task_running stream and the persisted history. Replace with capture_new_step_messages, which walks the newly-appended tail (and still re-checks the trailing message on no-growth chunks so id-less in-place replacements survive). 2. worker.py persisted each step with the store's low-frequency put() (a per-thread advisory lock per call); a deep subagent (max_turns=150) emits hundreds of steps on the hot stream loop. Replace with _SubagentEventBuffer, which batches via put_batch (flush on terminal subagent.end, at FLUSH_THRESHOLD, and in the worker finally). 3. build_subagent_step capped only text; tool_calls[].args were copied verbatim, so a large write_file/bash payload produced an unbounded subagent.step row. Cap each call's serialized args at SUBAGENT_STEP_MAX_CHARS, flagged args_truncated. Tests updated/added for all three; AGENTS.md refreshed. * fix(subagents): merge backfill into latest subtask state; reuse message_content_to_text (#3779) Address the remaining two PR review findings: 4. subtask-card's fetchSubtaskSteps().then(updateSubtask) closed over a stale tasks snapshot: a late-resolving backfill wrote setTasks({...stale}), clobbering SSE steps/status and sibling subtasks that arrived during the fetch. useUpdateSubtask now reads/writes through a tasksRef mirroring the latest state (ref-to-latest), and the pure per-subtask transition is extracted to core/tasks/subtask-update.ts::computeNextSubtask (unit-tested). 5. step_events._content_to_text duplicated deerflow.utils.messages. message_content_to_text; call the shared helper instead (guarding None content with 'or ""' so a tool-call-only turn still renders as ""). Tests added for computeNextSubtask and the None-content case; AGENTS.md docs updated.
This commit is contained in:
Nan Gao
GitHub
parent
2e15e3fe0d
commit
4fcb4bc366
@ -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 `<available-deferred-tools>` 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.
|
||||
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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):
|
||||
|
||||
@ -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):
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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()
|
||||
|
||||
227
backend/packages/harness/deerflow/subagents/step_events.py
Normal file
227
backend/packages/harness/deerflow/subagents/step_events.py
Normal file
@ -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
|
||||
156
backend/tests/test_run_event_store_filter.py
Normal file
156
backend/tests/test_run_event_store_filter.py
Normal file
@ -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)))
|
||||
45
backend/tests/test_run_events_endpoint.py
Normal file
45
backend/tests/test_run_events_endpoint.py
Normal file
@ -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"]
|
||||
@ -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."""
|
||||
|
||||
294
backend/tests/test_subagent_step_events.py
Normal file
294
backend/tests/test_subagent_step_events.py
Normal file
@ -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
|
||||
175
backend/tests/test_worker_subagent_persistence.py
Normal file
175
backend/tests/test_worker_subagent_persistence.py
Normal file
@ -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 == []
|
||||
@ -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
|
||||
|
||||
|
||||
@ -551,6 +551,8 @@ export function MessageList({
|
||||
<SubtaskCard
|
||||
key={"task-group-" + taskId}
|
||||
taskId={taskId}
|
||||
threadId={threadId}
|
||||
runId={(message as { run_id?: string }).run_id}
|
||||
isLoading={groupIsLoading}
|
||||
/>,
|
||||
);
|
||||
|
||||
@ -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 <CheckCircleIcon className="size-3" />;
|
||||
@ -135,16 +173,36 @@ export function SubtaskCard({
|
||||
}
|
||||
></ChainOfThoughtStep>
|
||||
)}
|
||||
{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 ? (
|
||||
<Loader2Icon className="size-4 animate-spin" />
|
||||
) : step.kind === "tool" ? (
|
||||
<WrenchIcon className="size-4" />
|
||||
) : (
|
||||
<SparklesIcon className="size-4" />
|
||||
);
|
||||
return (
|
||||
<ChainOfThoughtStep
|
||||
label={t.subtasks.in_progress}
|
||||
icon={<Loader2Icon className="size-4 animate-spin" />}
|
||||
>
|
||||
{explainLastToolCall(task.latestMessage, t)}
|
||||
</ChainOfThoughtStep>
|
||||
)}
|
||||
key={`${step.message_index}-${i}`}
|
||||
label={
|
||||
step.kind === "tool" ? (
|
||||
(step.tool_name ?? t.subtasks[task.status])
|
||||
) : (
|
||||
<div className="text-muted-foreground line-clamp-3 text-sm">
|
||||
<MarkdownContent
|
||||
content={step.text}
|
||||
isLoading={false}
|
||||
rehypePlugins={rehypePlugins}
|
||||
/>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
icon={icon}
|
||||
/>
|
||||
);
|
||||
})}
|
||||
{task.status === "completed" && (
|
||||
<>
|
||||
<ChainOfThoughtStep
|
||||
|
||||
65
frontend/src/core/tasks/api.ts
Normal file
65
frontend/src/core/tasks/api.ts
Normal file
@ -0,0 +1,65 @@
|
||||
import { fetch } from "../api/fetcher";
|
||||
import { getBackendBaseURL } from "../config";
|
||||
|
||||
import { eventsToSteps, type SubtaskStep } from "./steps";
|
||||
|
||||
/** Default per-request page size; matches the events endpoint's default. */
|
||||
const SUBTASK_STEPS_PAGE_SIZE = 500;
|
||||
/** Safety bound on pagination so a misbehaving cursor can't loop forever. */
|
||||
const SUBTASK_STEPS_MAX_PAGES = 100;
|
||||
|
||||
type FetchedEvent = Parameters<typeof eventsToSteps>[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<SubtaskStep[]> {
|
||||
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);
|
||||
}
|
||||
@ -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<string, Subtask>;
|
||||
// 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<Record<string, Subtask>>;
|
||||
setTasks: (tasks: Record<string, Subtask>) => void;
|
||||
}
|
||||
|
||||
export const SubtaskContext = createContext<SubtaskContextValue>({
|
||||
tasks: {},
|
||||
tasksRef: { current: {} },
|
||||
setTasks: () => {
|
||||
/* noop */
|
||||
},
|
||||
@ -27,8 +29,12 @@ export const SubtaskContext = createContext<SubtaskContextValue>({
|
||||
|
||||
export function SubtasksProvider({ children }: { children: React.ReactNode }) {
|
||||
const [tasks, setTasks] = useState<Record<string, Subtask>>({});
|
||||
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 (
|
||||
<SubtaskContext.Provider value={{ tasks, setTasks }}>
|
||||
<SubtaskContext.Provider value={{ tasks, tasksRef, setTasks }}>
|
||||
{children}
|
||||
</SubtaskContext.Provider>
|
||||
);
|
||||
@ -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<Subtask> & { 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;
|
||||
|
||||
169
frontend/src/core/tasks/steps.ts
Normal file
169
frontend/src/core/tasks/steps.ts
Normal file
@ -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 <tool>" 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<string, unknown>;
|
||||
};
|
||||
|
||||
/**
|
||||
* 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<number, SubtaskStep>();
|
||||
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,
|
||||
);
|
||||
}
|
||||
47
frontend/src/core/tasks/subtask-update.ts
Normal file
47
frontend/src/core/tasks/subtask-update.ts
Normal file
@ -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<Subtask> & { 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 };
|
||||
}
|
||||
@ -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;
|
||||
|
||||
@ -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;
|
||||
}
|
||||
|
||||
|
||||
98
frontend/tests/unit/core/tasks/api.test.ts
Normal file
98
frontend/tests/unit/core/tasks/api.test.ts
Normal file
@ -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();
|
||||
});
|
||||
});
|
||||
220
frontend/tests/unit/core/tasks/steps.test.ts
Normal file
220
frontend/tests/unit/core/tasks/steps.test.ts
Normal file
@ -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([]);
|
||||
});
|
||||
});
|
||||
108
frontend/tests/unit/core/tasks/subtask-update.test.ts
Normal file
108
frontend/tests/unit/core/tasks/subtask-update.test.ts
Normal file
@ -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> = {}): 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);
|
||||
});
|
||||
});
|
||||
@ -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 = [
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user