From 3dd9e8841ffa0290820cc015f65636edb9ad1b49 Mon Sep 17 00:00:00 2001 From: easT-T Date: Wed, 4 Nov 2026 20:16:09 +0800 Subject: [PATCH] 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 --- README.md | 4 + backend/tests/test_memory_updater.py | 12 +- frontend/src/AGENTS.md | 2 + .../workspace/messages/subtask-card.tsx | 2 +- frontend/src/core/tasks/context.tsx | 12 +- frontend/src/core/tasks/subtask-update.ts | 12 +- frontend/tests/e2e/subtask-backfill.spec.ts | 156 ++++++++++++++++++ .../messages/subtask-card.dom.test.tsx | 42 +++++ .../unit/core/tasks/subtask-update.test.ts | 44 +++++ 9 files changed, 278 insertions(+), 8 deletions(-) create mode 100644 frontend/tests/e2e/subtask-backfill.spec.ts diff --git a/README.md b/README.md index 63e2914cd7..0c9e15e187 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/backend/tests/test_memory_updater.py b/backend/tests/test_memory_updater.py index cda36abc79..72480ea413 100644 --- a/backend/tests/test_memory_updater.py +++ b/backend/tests/test_memory_updater.py @@ -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, }, ] ) diff --git a/frontend/src/AGENTS.md b/frontend/src/AGENTS.md index 2c5d08b76b..4d8b8284cd 100644 --- a/frontend/src/AGENTS.md +++ b/frontend/src/AGENTS.md @@ -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 diff --git a/frontend/src/components/workspace/messages/subtask-card.tsx b/frontend/src/components/workspace/messages/subtask-card.tsx index 0b65b032f4..723ec14869 100644 --- a/frontend/src/components/workspace/messages/subtask-card.tsx +++ b/frontend/src/components/workspace/messages/subtask-card.tsx @@ -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(() => { diff --git a/frontend/src/core/tasks/context.tsx b/frontend/src/core/tasks/context.tsx index 7479398b7c..5a62defafb 100644 --- a/frontend/src/core/tasks/context.tsx +++ b/frontend/src/core/tasks/context.tsx @@ -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 & { id: string }) => { + ( + task: Partial & { 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; diff --git a/frontend/src/core/tasks/subtask-update.ts b/frontend/src/core/tasks/subtask-update.ts index 8eae267c8c..35576805ce 100644 --- a/frontend/src/core/tasks/subtask-update.ts +++ b/frontend/src/core/tasks/subtask-update.ts @@ -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 & { 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 diff --git a/frontend/tests/e2e/subtask-backfill.spec.ts b/frontend/tests/e2e/subtask-backfill.spec.ts new file mode 100644 index 0000000000..6696f43ed9 --- /dev/null +++ b/frontend/tests/e2e/subtask-backfill.spec.ts @@ -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((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((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(); + } +}); diff --git a/frontend/tests/unit/components/workspace/messages/subtask-card.dom.test.tsx b/frontend/tests/unit/components/workspace/messages/subtask-card.dom.test.tsx index cca3888a42..87b2c3230f 100644 --- a/frontend/tests/unit/components/workspace/messages/subtask-card.dom.test.tsx +++ b/frontend/tests/unit/components/workspace/messages/subtask-card.dom.test.tsx @@ -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) => 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"); diff --git a/frontend/tests/unit/core/tasks/subtask-update.test.ts b/frontend/tests/unit/core/tasks/subtask-update.test.ts index 313ff75bd7..dc631aabd7 100644 --- a/frontend/tests/unit/core/tasks/subtask-update.test.ts +++ b/frontend/tests/unit/core/tasks/subtask-update.test.ts @@ -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)] });