mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-09 16:08:31 +00:00
4fcb4bc366
* 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.
45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
"""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"]
|