Nan Gao 4fcb4bc366
feat(subagents): persist and display subagent step history (#3779) (#3845)
* 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.
2026-07-02 07:43:09 +08:00

99 lines
2.9 KiB
TypeScript

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();
});
});