diff --git a/README.md b/README.md index e14f644409..78e9b64c7e 100644 --- a/README.md +++ b/README.md @@ -1178,6 +1178,9 @@ keeps the Goal, while explicitly replacing the work can clear it. Unapplied messages survive database-backed Gateway restarts and require **Send again**; interrupted child side effects are not replayed automatically. Existing execution limits still apply. Busy input currently accepts text only. +If a run rejects an update because it has finished, retrying the retained draft +during the next run targets that run. Uncertain network failures retain the +original submission identity so retries can reconcile the same accepted input. Markdown and JSON conversation exports from the chat header or sidebar read all persisted history pages, including earlier turns outside the loaded view or diff --git a/frontend/src/AGENTS.md b/frontend/src/AGENTS.md index 0f39f3991d..2c5d08b76b 100644 --- a/frontend/src/AGENTS.md +++ b/frontend/src/AGENTS.md @@ -44,6 +44,9 @@ Tests: `html-assets.dom.test.ts`, `artifact-file-preview.dom.test.tsx`, exiting retains the draft and is disabled only during a pending save. 3. `useThreadHistory` loads persisted conversation pages from `GET /api/threads/{id}/messages/page`, preserving the backend's thread-global event `seq`; rendering overlays checkpoint/live copies by message ID, never provider tool-call ID (a summarized checkpoint may contain a protected early input plus a recent tail). Context-compaction rescue diffs every retained visible identity rather than slicing at the first anchor, and keeps a run-scoped ledger of committed visible messages so replacement updates and repeated rolling checkpoint windows cannot erase an already displayed step. A checkpoint/transient prefix whose canonical position is still behind an unloaded cursor page is woven in before the first shared anchor, not discarded: both the checkpoint and seq-sorted history place it earlier, so that position is known even when the pages between are not. Position authority is seq-first and lives in `core/threads/message-order.ts` (re-exported through `hooks.ts`): every normalized identity tracks latest visible content and trusted position separately; a valid `deerflow_seq` (positive safe integer, earliest value wins per identity, hidden control copies contribute a position only when no visible copy carries one) joins an ascending skeleton that outranks identity-anchor weaving, which remains the fallback for no-seq segments whose internal order is preserved. Live-only and rescued messages with trusted seq also serve as anchors for adjacent no-seq segments. A trailing segment follows its last live-only positioned anchor within the loaded window; after a shared anchor or a rescued prefix before the window, it stays at the tail. Optimistic messages remain last; the transient bridge inserts positioned rows before weaving so the insertion cannot reverse previously displayed steps. Rescued sequence anchors preserve preceding captured steps even before React has rendered them; only a leading prefix anchored to loaded history requires previously rendered ordering to cross an unloaded cursor gap. Content replacement never drops the known seq, `run_id`, or `turn_duration`, and the compaction transient bridge plus rendered ledger share the same position priority. `deerflow_seq` is server-owned display metadata and is never written back into a checkpoint. It must never be appended to the tail (#4065) — the tail is provably wrong — but suppressing it entirely is how a user's own question vanished from a long thread once the first 50-row history page no longer reached back to it (#4666). A collapsed unloaded gap is recoverable by paging; a dropped message is not. Weaving alone restores the message but not its exact position — after compaction the live window carries too few anchors — so both sides now carry the backend's thread-global `additional_kwargs.deerflow_seq`: `buildVisibleHistoryMessages` copies each row's `seq`, and the Gateway stamps it onto `values` frame messages it has already persisted. A live message whose seq is below the loaded window's lower bound is placed ahead of everything on screen instead of before the nearest anchor, which is what puts a compaction-rescued first user turn back at the head rather than mid-transcript. That split happens _before_ the anchor walk, not inside it: a compacted checkpoint can share no identity at all with the loaded page — it keeps only the current run's recent tail, while the page on screen was fetched turns earlier — and the anchor walk then never runs at all, which is precisely when a rescued turn most needs its seq. Doing the split inside the walk left that case appending the message after the whole window (#4666), the one arrangement #4065 proved wrong. A message without a seq (still streaming, so not in the feed yet) keeps the weaving path — the tail is already its correct position. Optimistic messages are then added without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded. Dynamic context re-keys each submitted user message from the client-generated `local-human-*` identity `X` to the visible server echo `X__user`; UI identity matching normalizes that reserved suffix only for human messages so the optimistic input and checkpoint replacement remain one visible turn. At dispatch, a local-turn anchor snapshots the checkpoint identity baseline, canonical history identities and maximum trusted seq, and any pre-existing transient-bridge identities. Render repair uses `confirmedHistoryIdentities` plus `preSubmitMaxSeq` to restore baseline or history-confirmed messages above that exact human anchor, while moving only speculative non-baseline AI/tool steps behind it; `currentTurnRunIds` keeps already-persisted steps of the active turn below their human. Keep the anchor scoped to its originating thread through finish, stop, and stream error because the SDK's settled frame can retain transient event order; replace it on the next local submit and clear it on thread switch or replay-gap recovery. 4. Stop invalidates chat/history/usage/sidebar caches immediately and once after SDK cancellation settles. Queue/Steer keeps owner-scoped IDs and drafts on uncertain sends; pending task receipts remain live until their same-Run final result. Both chat pages share this flow. + Release a rejected Queue/Steer attempt only after HTTP 409 and no recovered + receipt. Uncertain failures keep its run, request ID and context: the recent + input list can omit accepted inputs. 5. TanStack Query manages server state. `UserPreferencesBoundary` preserves workspace SSR and initializes the account cache before browser paint; only subsequent account switches gate consumers until the new cache is active. Four allowlisted diff --git a/frontend/src/core/threads/hooks.ts b/frontend/src/core/threads/hooks.ts index e5124e3ab3..74bba60888 100644 --- a/frontend/src/core/threads/hooks.ts +++ b/frontend/src/core/threads/hooks.ts @@ -23,6 +23,7 @@ import { toast } from "sonner"; import type { PromptInputMessage } from "@/components/ai-elements/prompt-input"; import { getAPIClient } from "../api"; +import { GatewayApiError } from "../api/errors"; import { fetch } from "../api/fetcher"; import { AuthQueryScopeContext } from "../auth/query-scope"; import { getBackendBaseURL } from "../config"; @@ -2472,6 +2473,17 @@ export function useThreadStream({ }); return accepted; } + // A conflict rejects this submission. Let the next user attempt + // target the current run, but retain the original identity for + // uncertain failures: the recent-input list may omit its receipt. + if ( + !controller.signal.aborted && + runInputAttemptRef.current === attempt && + error instanceof GatewayApiError && + error.status === 409 + ) { + runInputAttemptRef.current = null; + } toast.error(t.inputBox.runInputUnavailable); } throw error; diff --git a/frontend/src/core/threads/run-inputs.ts b/frontend/src/core/threads/run-inputs.ts index c9cab66cda..244ce71ec3 100644 --- a/frontend/src/core/threads/run-inputs.ts +++ b/frontend/src/core/threads/run-inputs.ts @@ -1,5 +1,6 @@ import { z } from "zod"; +import { throwGatewayApiError } from "@/core/api/errors"; import { fetch } from "@/core/api/fetcher"; import { getBackendBaseURL } from "@/core/config"; @@ -27,12 +28,7 @@ export const runInputsKey = (threadId: string) => async function readResponse(response: Response) { if (!response.ok) { - const body = (await response.json().catch(() => ({}))) as { - detail?: unknown; - }; - throw new Error( - typeof body.detail === "string" ? body.detail : `HTTP ${response.status}`, - ); + await throwGatewayApiError(response, `HTTP ${response.status}`); } return response.json() as Promise; } diff --git a/frontend/tests/e2e/run-input-retry.spec.ts b/frontend/tests/e2e/run-input-retry.spec.ts new file mode 100644 index 0000000000..e94a96256d --- /dev/null +++ b/frontend/tests/e2e/run-input-retry.spec.ts @@ -0,0 +1,167 @@ +import { expect, test, type Route } from "@playwright/test"; + +import { mockLangGraphAPI, MOCK_THREAD_ID } from "./utils/mock-api"; + +const RUN_A = "00000000-0000-0000-0000-000000000091"; +const RUN_B = "00000000-0000-0000-0000-000000000092"; +const INPUT_TEXT = "Check the updated requirement"; + +type SubmittedInput = { + request_id: string; + text: string; + mode: "queue" | "steer"; +}; + +for (const mode of ["queue", "steer"] as const) { + test(`${mode} retries a rejected draft against the new active run`, async ({ + page, + }) => { + let activeRunId = RUN_A; + let acceptedInput: Record | undefined; + const submissions: Array = []; + const streams = new Map(); + const run = (runId: string) => ({ + run_id: runId, + thread_id: MOCK_THREAD_ID, + assistant_id: "lead_agent", + status: runId === activeRunId ? "running" : "success", + 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: "Input retry regression", + messages: [ + { type: "human", id: "human-original", content: "Start working" }, + ], + }, + ], + }); + await page.route("**/api/features", (route) => + route.fulfill({ json: { run_inputs: true } }), + ); + await page.route(`**/api/threads/${MOCK_THREAD_ID}/inputs`, (route) => + route.fulfill({ + json: { + items: acceptedInput ? [acceptedInput] : [], + active_run_id: activeRunId, + }, + }), + ); + await page.route( + new RegExp(`/api/langgraph/threads/${MOCK_THREAD_ID}/runs(?:\\?|$)`), + (route) => route.fulfill({ json: [run(activeRunId)] }), + ); + await page.route( + new RegExp( + `/api/langgraph/threads/${MOCK_THREAD_ID}/runs/(${RUN_A}|${RUN_B})(?:\\?|$)`, + ), + (route) => { + const runId = new URL(route.request().url()).pathname + .split("/") + .at(-1)!; + return route.fulfill({ json: run(runId) }); + }, + ); + // Hold each observer stream until the test explicitly finishes that run. + // Receiving B's join request is the synchronization point proving the + // mounted page has adopted the new active run without resetting its draft. + await page.route( + `**/api/langgraph/threads/${MOCK_THREAD_ID}/runs/*/stream*`, + (route) => { + const runId = new URL(route.request().url()).pathname + .split("/") + .at(-2)!; + streams.set(runId, route); + }, + ); + await page.route( + `**/api/threads/${MOCK_THREAD_ID}/runs/*/inputs`, + (route) => { + const runId = new URL(route.request().url()).pathname + .split("/") + .at(-2)!; + const body = route.request().postDataJSON() as SubmittedInput; + submissions.push({ ...body, runId }); + if (runId === RUN_A) { + return route.fulfill({ + status: 409, + json: { + detail: + "Run is finishing or stopped; this input was not accepted", + }, + }); + } + acceptedInput = { + id: body.request_id, + target_run_id: runId, + message_id: "human-update", + text: body.text, + mode: body.mode, + status: "pending", + created_at: "2026-01-01T00:00:01Z", + }; + return route.fulfill({ json: acceptedInput }); + }, + ); + + async function finishStream(runId: string) { + const route = streams.get(runId); + if (!route) return; + streams.delete(runId); + await route.fulfill({ + contentType: "text/event-stream", + body: + `event: metadata\ndata: ${JSON.stringify({ run_id: runId, thread_id: MOCK_THREAD_ID })}\n\n` + + "event: end\ndata: {}\n\n", + }); + } + + try { + await page.goto(`/workspace/chats/${MOCK_THREAD_ID}`); + await expect.poll(() => streams.has(RUN_A)).toBe(true); + const textarea = page.getByPlaceholder(/how can i assist you/i); + const send = page.getByRole("button", { name: "Send update" }); + await textarea.fill(INPUT_TEXT); + if (mode === "steer") { + await page.getByRole("button", { name: "Send while working" }).click(); + await page.getByRole("menuitem", { name: /^Steer/ }).click(); + } + await send.click(); + await expect( + page.getByText( + "The message was not confirmed. Check the task status and try again; your draft is kept.", + ), + ).toBeVisible(); + await expect(textarea).toHaveValue(INPUT_TEXT); + expect(submissions).toHaveLength(1); + expect(submissions[0]).toMatchObject({ + runId: RUN_A, + text: INPUT_TEXT, + mode, + }); + + activeRunId = RUN_B; + await finishStream(RUN_A); + await expect.poll(() => streams.has(RUN_B)).toBe(true); + await send.click(); + await expect.poll(() => submissions.length).toBe(2); + expect(submissions[1]).toMatchObject({ + runId: RUN_B, + text: INPUT_TEXT, + mode, + }); + expect(submissions[1]!.request_id).not.toBe(submissions[0]!.request_id); + await expect(textarea).toHaveValue(""); + await expect( + page.getByTestId("run-inputs").getByText(INPUT_TEXT), + ).toBeVisible(); + } finally { + await Promise.all([...streams.keys()].map(finishStream)); + } + }); +} diff --git a/frontend/tests/unit/core/threads/run-input-retry.dom.test.tsx b/frontend/tests/unit/core/threads/run-input-retry.dom.test.tsx new file mode 100644 index 0000000000..36137e51fa --- /dev/null +++ b/frontend/tests/unit/core/threads/run-input-retry.dom.test.tsx @@ -0,0 +1,311 @@ +import type { Run } from "@langchain/langgraph-sdk"; +import { afterEach, beforeEach, expect, rs, test } from "@rstest/core"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { act, cleanup, renderHook } from "@testing-library/react"; +import { createElement, type ReactNode } from "react"; + +import { AuthQueryScopeContext } from "@/core/auth/query-scope"; +import { I18nContext } from "@/core/i18n/context"; +import { enUS } from "@/core/i18n/locales/en-US"; +import { DEFAULT_LOCAL_SETTINGS } from "@/core/settings/local"; +import { useThreadStream } from "@/core/threads/hooks"; +import { + runInputsKey, + type RunInput, + type RunInputMode, +} from "@/core/threads/run-inputs"; + +const streamMockState = rs.hoisted(() => ({ + joinStream: rs.fn(async () => undefined), +})); + +rs.mock("@/core/api", () => ({ + getAPIClient: () => ({ + runs: { list: async () => [] as Run[] }, + }), +})); + +rs.mock("@langchain/langgraph-sdk/react", () => ({ + useStream: () => ({ + isLoading: true, + joinStream: streamMockState.joinStream, + messages: [], + stop: async () => undefined, + submit: async () => undefined, + values: { artifacts: [], messages: [], title: "", todos: [] }, + }), +})); + +type SubmittedInput = { + runId: string; + body: { + request_id: string; + text: string; + mode: RunInputMode; + context: Record; + }; +}; + +const queryClients: QueryClient[] = []; +let activeRunId: string; +let receipts: RunInput[]; +let submittedInputs: SubmittedInput[]; +let firstPostFailure: + | "conflict" + | "network" + | "server" + | "malformed" + | "accepted"; + +function jsonResponse(body: unknown, status = 200) { + return new Response(JSON.stringify(body), { + status, + headers: { "Content-Type": "application/json" }, + }); +} + +function receiptFor({ runId, body }: SubmittedInput): RunInput { + return { + id: body.request_id, + target_run_id: runId, + message_id: `message-${body.request_id}`, + text: body.text, + mode: body.mode, + status: "pending", + created_at: "2026-11-02T00:00:00Z", + }; +} + +function createWrapper(queryClient: QueryClient) { + return function RunInputRetryTestWrapper({ + children, + }: { + children: ReactNode; + }) { + return createElement( + QueryClientProvider, + { client: queryClient }, + createElement( + AuthQueryScopeContext.Provider, + { value: "account-1" }, + createElement( + I18nContext.Provider, + { + value: { + locale: "en-US", + setLocale: () => undefined, + t: enUS, + }, + }, + children, + ), + ), + ); + }; +} + +async function flushFrames() { + for (let index = 0; index < 8; index += 1) { + await act(async () => { + await rs.advanceTimersByTimeAsync(1); + }); + } +} + +async function renderThread() { + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + queryClients.push(queryClient); + const rendered = renderHook( + ({ modelName }: { modelName: string }) => + useThreadStream({ + context: { + ...DEFAULT_LOCAL_SETTINGS.context, + model_name: modelName, + }, + threadId: "thread-1", + }), + { + initialProps: { modelName: "model-original" }, + wrapper: createWrapper(queryClient), + }, + ); + await flushFrames(); + expect(rendered.result.current.canSendInput).toBe(true); + expect(rendered.result.current.activeRunId).toBe("run-1"); + return { queryClient, ...rendered }; +} + +async function advanceToNextRun(queryClient: QueryClient) { + activeRunId = "run-2"; + await act(async () => { + await queryClient.invalidateQueries({ + queryKey: runInputsKey("thread-1"), + }); + }); + await flushFrames(); +} + +beforeEach(() => { + rs.useFakeTimers({ + toFake: ["setTimeout", "clearTimeout", "setInterval", "clearInterval"], + }); + window.sessionStorage.clear(); + activeRunId = "run-1"; + receipts = []; + submittedInputs = []; + firstPostFailure = "conflict"; + rs.stubGlobal( + "fetch", + rs.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = new URL( + typeof input === "string" + ? input + : input instanceof URL + ? input.href + : input.url, + "http://localhost", + ); + if (url.pathname === "/api/features") { + return jsonResponse({ run_inputs: true }); + } + if (url.pathname === "/api/threads/thread-1/messages/page") { + return jsonResponse({ + data: [], + has_more: false, + next_before_seq: null, + }); + } + if (url.pathname === "/api/threads/thread-1/inputs") { + return jsonResponse({ items: receipts, active_run_id: activeRunId }); + } + const match = /^\/api\/threads\/thread-1\/runs\/([^/]+)\/inputs$/.exec( + url.pathname, + ); + if (match && init?.method === "POST") { + expect(init.credentials).toBe("include"); + if (typeof init.body !== "string") { + throw new Error("Expected a JSON run-input request body"); + } + const submission: SubmittedInput = { + runId: match[1]!, + body: JSON.parse(init.body) as SubmittedInput["body"], + }; + submittedInputs.push(submission); + const receipt = receiptFor(submission); + if (submittedInputs.length === 1) { + if (firstPostFailure === "conflict") { + return jsonResponse( + { + detail: + "Run is finishing or stopped; this input was not accepted", + }, + 409, + ); + } + if (firstPostFailure === "server") { + return jsonResponse({ detail: "Gateway unavailable" }, 500); + } + if (firstPostFailure === "malformed") { + return new Response("{", { + status: 200, + headers: { "Content-Type": "application/json" }, + }); + } + if (firstPostFailure === "accepted") { + receipts.push(receipt); + } + throw new TypeError("Failed to fetch"); + } + receipts.push(receipt); + return jsonResponse(receipt); + } + throw new Error(`Unexpected request: ${init?.method ?? "GET"} ${url}`); + }), + ); +}); + +afterEach(() => { + cleanup(); + for (const queryClient of queryClients.splice(0)) { + queryClient.clear(); + } + rs.useRealTimers(); + rs.unstubAllGlobals(); +}); + +test.each(["queue", "steer"] as const)( + "sends a rejected %s input to the next active run with a new request ID", + async (mode) => { + const { result, queryClient } = await renderThread(); + + await act(async () => { + await expect(result.current.sendInput("Continue", mode)).rejects.toThrow( + "Run is finishing or stopped; this input was not accepted", + ); + }); + expect(submittedInputs).toHaveLength(1); + expect(receipts).toEqual([]); + + await advanceToNextRun(queryClient); + expect(result.current.activeRunId).toBe("run-2"); + await act(async () => { + await result.current.sendInput("Continue", mode); + }); + + expect(submittedInputs).toHaveLength(2); + expect(submittedInputs[1]?.runId).toBe("run-2"); + expect(submittedInputs[1]?.body.request_id).not.toBe( + submittedInputs[0]?.body.request_id, + ); + }, +); + +test.each(["network", "server", "malformed"] as const)( + "keeps the original run, request ID and context after an uncertain %s failure", + async (failure) => { + firstPostFailure = failure; + const { result, queryClient, rerender } = await renderThread(); + + await act(async () => { + await expect( + result.current.sendInput("Continue", "queue"), + ).rejects.toThrow(); + }); + expect(receipts).toEqual([]); + const first = submittedInputs[0]!; + expect(first.body.context.model_name).toBe("model-original"); + + await advanceToNextRun(queryClient); + rerender({ modelName: "model-new" }); + expect(result.current.activeRunId).toBe("run-2"); + await act(async () => { + await result.current.sendInput("Continue", "queue"); + }); + + expect(submittedInputs).toHaveLength(2); + expect(submittedInputs[1]).toEqual(first); + }, +); + +test("recovers an accepted input when the POST response was lost", async () => { + firstPostFailure = "accepted"; + const { result } = await renderThread(); + + await act(async () => { + await expect( + result.current.sendInput("Continue", "queue"), + ).resolves.toEqual( + expect.objectContaining({ + target_run_id: "run-1", + text: "Continue", + mode: "queue", + status: "pending", + }), + ); + }); + + expect(submittedInputs).toHaveLength(1); + expect(receipts[0]?.id).toBe(submittedInputs[0]?.body.request_id); +});