fix(frontend): preserve live subtask steps during history backfill (#7864)

* fix(frontend): preserve live subtask steps during history backfill

* test(deermem): keep updater eviction fixtures fresh

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
easT-T 2026-11-04 20:16:09 +08:00 committed by GitHub
parent e5969fd8b9
commit 3dd9e8841f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 278 additions and 8 deletions

View File

@ -2802,6 +2802,10 @@ losing that turn's per-run delegation count.
The lead agent can spawn sub-agents on the fly — each with its own scoped context, tools, and termination conditions — when delegation has clear net benefit from real parallel latency, specialist capability, or context isolation. It keeps interdependent scopes and overlapping side effects out of parallel dispatch; a bounded sequential chain can still run in one sub-agent when specialist or context-isolation benefit clearly wins. The lead uses the fewest useful sub-agents and re-evaluates later batches instead of fanning out solely because a task is large or multi-step. Sub-agents report back structured results, and the lead agent verifies and synthesizes them into a coherent output. Deterministic tool receipts cover both direct tool messages and state-updating `Command` results such as delegated `task` responses; when the receipt ledger reaches its context budget, it retains the newest actions and their original receipt IDs. Operators can disable this provenance layer with `verification.receipts_enabled: false`. Their configured skills are resolved from the same user-scoped catalog as the lead agent, so user-owned custom skills remain available without exposing another user's version. Their internal AI and tool messages stay scoped to the delegated graph instead of entering the parent chat stream. Reloaded thread history enforces the same boundary: callback-captured sub-agent AI responses remain available in run-event diagnostics but are excluded from the parent transcript, while the parent `task` result remains attached to its subtask card. Long-running sub-agents compact older history when summarization is enabled and re-inject the summary as guarded, hidden durable context before continuing, so recent assistant/tool activity remains grounded in the task. Their system instructions, including the role and report contract, survive compaction; if only those instructions and the current request would be summarized, compaction is skipped. Provider/model request failures are reported as failed sub-agent tasks rather than successful results, so the lead agent and Web UI can react to them correctly. Concurrent parent runs also receive independent server-side sub-agent execution IDs, so a provider that reuses a tool-call ID cannot make one run poll, cancel, or clean up another run's background task. Collapsed sub-agent cards show the effective model and, when the provider returns usage metadata, a cumulative token total that updates after each completed sub-agent LLM call and persists after a reload. When token usage tracking is enabled, completed sub-agent usage is attributed back to the dispatching step from that run's terminal tool-message metadata rather than a process-global provider-ID cache.
When expanded sub-agent cards load persisted steps, delayed history responses
fill missing timeline entries without replacing complete steps already received
from the live stream with truncated previews.
Sub-agent cards keep unrecognized structured task statuses in progress. Accompanying
result text does not turn an unknown status into success or failure.

View File

@ -3,6 +3,7 @@ import copy
import json
import logging
import threading
from datetime import UTC, datetime, timedelta
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@ -270,6 +271,9 @@ def test_automatic_update_uses_hybrid_capacity_policy() -> None:
fact_confidence_threshold=0.7,
)
)
now = datetime.now(UTC)
recent_timestamp = now.isoformat().replace("+00:00", "Z")
stale_timestamp = (now - timedelta(days=365)).isoformat().replace("+00:00", "Z")
current_memory = _make_memory(
facts=[
*[
@ -278,7 +282,7 @@ def test_automatic_update_uses_hybrid_capacity_policy() -> None:
"content": f"high {index}",
"category": "preference",
"confidence": 0.99,
"createdAt": "2026-08-12T00:00:00Z",
"createdAt": recent_timestamp,
}
for index in range(8)
],
@ -287,15 +291,15 @@ def test_automatic_update_uses_hybrid_capacity_policy() -> None:
"content": "stale high confidence",
"category": "preference",
"confidence": 0.9,
"createdAt": "2025-01-01T00:00:00Z",
"createdAt": stale_timestamp,
},
{
"id": "confirmed_recently",
"content": "recently confirmed preference",
"category": "preference",
"confidence": 0.7,
"createdAt": "2025-01-01T00:00:00Z",
"lastConfirmedAt": "2026-08-12T00:00:00Z",
"createdAt": stale_timestamp,
"lastConfirmedAt": recent_timestamp,
},
]
)

View File

@ -220,6 +220,8 @@ export on a page error or non-advancing cursor; demos keep their fixture path.
- Tool-step links in `message-group.tsx` (`web_fetch` args, `web_search` / `image_search` result URLs) are model- or provider-controlled, so they pass the markdown `isSafeHref` allowlist; every surface renders a rejected href through the shared `UnsafeLink` marker.
- **Environment validation** uses `@t3-oss/env-nextjs` with Zod schemas (`src/env.js`). Skip with `SKIP_ENV_VALIDATION=1`
- **Subtask step history and runtime metadata** (`core/tasks/`) — cards interleave subagent reasoning and tool steps (#3779). Task `description` is optional progress text; `MessageList` titles use the required `prompt`, then a localized generic label, so valid calls never show a blank title. `Subtask.steps[]` appends live `task_running` events via `mergeSteps`. On historical-card expansion, `fetchSubtaskSteps` backfills task-scoped events (GET `/runs/{runId}/events?event_types=subagent.step&task_id=…&after_seq=…`) until a short page or a non-advancing full page; run-wide limits must not truncate the timeline. `task_started` carries the effective `model_name`; `task_running` carries a cumulative usage snapshot after each completed LLM call. `core/tasks/lifecycle.ts` normalizes these additive events, and `computeNextSubtask` keeps the largest cumulative total so replayed or late SSE frames cannot double-count or roll the folded card backward. Terminal ToolMessage metadata (`subagent_model_name` / `subagent_token_usage`) restores the same values from normal history after reload; no per-card event fetch is needed. `core/tasks/steps.ts` is the pure step 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/context.tsx`'s `useUpdateSubtask` applies updates 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.
- Subtask backfills pass `stepSource: "history"`: fill missing indices but keep
current steps over bounded previews. Live updates still replace matching indices.
- Structured subtask status parsing in `core/tasks/subtask-result.ts` accepts only
own keys of the status mapping. Prototype names such as `constructor`,
`toString`, and `__proto__` use the same `in_progress` fallback as other unknown

View File

@ -117,7 +117,7 @@ export function SubtaskCard({
fetchSubtaskSteps(threadId, runId, taskId)
.then((steps) => {
if (steps.length > 0) {
updateSubtask({ id: taskId, steps });
updateSubtask({ id: taskId, steps }, { stepSource: "history" });
}
})
.catch(() => {

View File

@ -7,7 +7,11 @@ import {
useState,
} from "react";
import { computeNextSubtask, subtaskNotification } from "./subtask-update";
import {
computeNextSubtask,
subtaskNotification,
type SubtaskUpdateOptions,
} from "./subtask-update";
import type { Subtask } from "./types";
export interface SubtaskContextValue {
@ -68,7 +72,10 @@ export function useUpdateSubtask() {
});
const updateSubtask = useCallback(
(task: Partial<Subtask> & { id: string }) => {
(
task: Partial<Subtask> & { id: string },
options?: SubtaskUpdateOptions,
) => {
// 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
@ -77,6 +84,7 @@ export function useUpdateSubtask() {
const { next, becameTerminal, changed } = computeNextSubtask(
current[task.id],
task,
options,
);
current[task.id] = next;

View File

@ -1,6 +1,10 @@
import { mergeSteps } from "./steps";
import type { Subtask } from "./types";
export interface SubtaskUpdateOptions {
stepSource?: "live" | "history";
}
export function isTerminalSubtaskStatus(status: Subtask["status"] | undefined) {
return status === "completed" || status === "failed";
}
@ -22,6 +26,7 @@ export function isTerminalSubtaskStatus(status: Subtask["status"] | undefined) {
export function computeNextSubtask(
previous: Subtask | undefined,
task: Partial<Subtask> & { id: string },
{ stepSource = "live" }: SubtaskUpdateOptions = {},
): { next: Subtask; becameTerminal: boolean; changed: boolean } {
const previousStatus = previous?.status;
@ -37,7 +42,12 @@ export function computeNextSubtask(
} as Subtask;
if (task.steps) {
next.steps = mergeSteps(previous?.steps ?? [], task.steps);
// Persisted steps are bounded previews. A delayed backfill must not
// replace a complete live step that arrived while the read was pending.
next.steps =
stepSource === "history"
? mergeSteps(task.steps, previous?.steps ?? [])
: mergeSteps(previous?.steps ?? [], task.steps);
}
// Usage events are cumulative snapshots. A delayed older frame must never

View File

@ -0,0 +1,156 @@
import { createServer, type ServerResponse } from "node:http";
import type { AddressInfo } from "node:net";
import { expect, test, type Route } from "@playwright/test";
import {
mockLangGraphAPI,
MOCK_RUN_ID,
MOCK_THREAD_ID,
} from "./utils/mock-api";
const TASK_ID = "call-backfill-subtask";
const TASK_PROMPT = "Research the recovered subtask timeline";
const LIVE_SUFFIX = "Complete live reasoning survives the history response.";
const LIVE_TEXT = `${"Live reasoning. ".repeat(600)}${LIVE_SUFFIX}`;
const EARLIER_TEXT = "Earlier reasoning recovered from persisted history.";
const messages = [
{ type: "human", id: "human-backfill", content: "Continue the research" },
{
type: "ai",
id: "ai-backfill",
run_id: MOCK_RUN_ID,
content: "",
tool_calls: [
{
id: TASK_ID,
name: "task",
args: { subagent_type: "general-purpose", prompt: TASK_PROMPT },
type: "tool_call",
},
],
},
];
async function startSubtaskStream() {
let response: ServerResponse | undefined;
const send = (event: string, data: unknown) => {
if (!response) throw new Error("The observer stream has not joined");
response.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
};
const server = createServer((_request, stream) => {
response = stream;
stream.writeHead(200, {
"Access-Control-Allow-Origin": "*",
"Cache-Control": "no-cache",
"Content-Type": "text/event-stream",
});
send("metadata", { run_id: MOCK_RUN_ID, thread_id: MOCK_THREAD_ID });
send("values", { messages });
});
await new Promise<void>((resolve, reject) => {
server.once("error", reject);
server.listen(0, "127.0.0.1", resolve);
});
return {
url: `http://127.0.0.1:${(server.address() as AddressInfo).port}/stream`,
joined: () => response !== undefined,
send,
async close() {
server.closeAllConnections();
await new Promise<void>((resolve, reject) => {
server.close((error) => (error ? reject(error) : resolve()));
});
},
};
}
test("late subtask history preserves the full live step at the same index", async ({
page,
}) => {
const stream = await startSubtaskStream();
let historyRequest: Route | undefined;
const run = {
run_id: MOCK_RUN_ID,
thread_id: MOCK_THREAD_ID,
assistant_id: "lead_agent",
status: "running",
metadata: {},
kwargs: {},
created_at: "2026-01-01T00:00:00Z",
updated_at: "2026-01-01T00:00:00Z",
};
mockLangGraphAPI(page, {
threads: [
{ thread_id: MOCK_THREAD_ID, title: "Subtask backfill", messages },
],
});
await page.route(
new RegExp(`/api/langgraph/threads/${MOCK_THREAD_ID}/runs(?:\\?|$)`),
(route) => route.fulfill({ json: [run] }),
);
await page.route(
new RegExp(
`/api/langgraph/threads/${MOCK_THREAD_ID}/runs/${MOCK_RUN_ID}(?:\\?|$)`,
),
(route) => route.fulfill({ json: run }),
);
await page.route(
`**/api/langgraph/threads/${MOCK_THREAD_ID}/runs/${MOCK_RUN_ID}/stream*`,
(route) => route.continue({ url: stream.url }),
);
await page.route(
`**/api/threads/${MOCK_THREAD_ID}/runs/${MOCK_RUN_ID}/events*`,
(route) => {
historyRequest = route;
},
);
try {
await page.goto(`/workspace/chats/${MOCK_THREAD_ID}`);
await expect.poll(stream.joined).toBe(true);
await page.getByRole("button", { name: new RegExp(TASK_PROMPT) }).click();
await expect.poll(() => historyRequest !== undefined).toBe(true);
const query = new URL(historyRequest!.request().url()).searchParams;
expect(query.get("task_id")).toBe(TASK_ID);
expect(query.get("event_types")).toBe("subagent.step");
stream.send("custom", {
type: "task_running",
task_id: TASK_ID,
message_index: 2,
message: { type: "ai", id: "subtask-live", content: LIVE_TEXT },
});
// The card clamps long reasoning visually. Check its rendered text so
// truncation by the history payload cannot hide behind the layout clamp.
const fullStep = page.getByText(LIVE_TEXT, { exact: true });
await expect(fullStep).toHaveCount(1);
await historyRequest!.fulfill({
json: [
{ seq: 1, text: EARLIER_TEXT, truncated: false },
{ seq: 2, text: LIVE_TEXT.slice(0, 8192), truncated: true },
].map(({ seq, text, truncated }) => ({
seq,
event_type: "subagent.step",
content: {
task_id: TASK_ID,
message_index: seq,
kind: "ai",
text,
truncated,
},
})),
});
historyRequest = undefined;
// The newly recovered earlier step proves the response has been applied.
await expect(page.getByText(EARLIER_TEXT, { exact: true })).toHaveCount(1);
await expect(fullStep).toHaveCount(1);
await expect(
page.getByText(LIVE_TEXT.slice(0, 8192), { exact: true }),
).toHaveCount(0);
} finally {
await historyRequest?.abort();
await stream.close();
}
});

View File

@ -270,6 +270,48 @@ describe("SubtaskCard message synchronization", () => {
);
});
it("keeps full live steps when a delayed backfill contains truncated copies", async () => {
let finishBackfill!: (steps: NonNullable<Subtask["steps"]>) => void;
rs.mocked(fetchSubtaskSteps).mockImplementation(
() =>
new Promise((resolve) => {
finishBackfill = resolve;
}),
);
const view = setup(true);
fireEvent.click(screen.getByRole("button"));
expect(fetchSubtaskSteps).toHaveBeenCalledWith(
"thread-1",
"run-1",
"task-1",
);
const historical = {
kind: "tool" as const,
message_index: 1,
text: "history",
tool_name: "read_file",
};
const live = {
kind: "ai" as const,
message_index: 2,
text: `${"x".repeat(8192)} live reasoning tail`,
tool_calls: [{ name: "web_search", args: { query: "sources" } }],
};
await act(async () => view.publish({ id: "task-1", steps: [live] }));
expect(view.container.textContent).toContain("live reasoning tail");
// Persisted steps are bounded; a late response must only fill missing indices.
await act(async () =>
finishBackfill([
historical,
{ ...live, text: live.text.slice(0, 8192), truncated: true },
]),
);
expect(view.read().steps).toEqual([historical, live]);
expect(screen.getAllByText("read_file")).toHaveLength(1);
expect(view.container.textContent).toContain("live reasoning tail");
});
it("updates the folded title when final arguments and task_started arrive together", async () => {
const view = setup();
expect(screen.getByRole("button").textContent).toContain("Res");

View File

@ -29,6 +29,50 @@ function step(message_index: number): SubtaskStep {
}
describe("computeNextSubtask", () => {
it("backfills missing indices without replacing live text or structured arguments", () => {
const live = {
...step(2),
kind: "ai" as const,
text: "Complete reasoning",
tool_calls: [{ name: "write_file", args: { content: "Complete file" } }],
};
const history = [
step(1),
{
...live,
text: "Complete",
truncated: true,
tool_calls: [{ name: "write_file", args: '{"content":"Complete' }],
},
];
const previous = baseTask({ steps: [live, step(3)] });
const { next } = computeNextSubtask(
previous,
{ id: "t1", steps: history },
{ stepSource: "history" },
);
expect(next.steps).toEqual([step(1), live, step(3)]);
expect(previous.steps).toEqual([live, step(3)]);
});
it("accepts history without existing steps and lets subsequent live updates replace it", () => {
const historical = { ...step(1), text: "prefix", truncated: true };
const { next: restored } = computeNextSubtask(
baseTask(),
{ id: "t1", steps: [historical] },
{ stepSource: "history" },
);
expect(restored.steps).toEqual([historical]);
const { next } = computeNextSubtask(restored, {
id: "t1",
steps: [step(1)],
});
expect(next.steps).toEqual([step(1)]);
});
it("merges step deltas into the provided previous, preserving both", () => {
const previous = baseTask({ steps: [step(1), step(2)] });