fix(frontend): release rejected run-input attempts before retrying (#7790)

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

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;

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<unknown>;
 }

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<string, unknown> | undefined;
+    const submissions: Array<SubmittedInput & { runId: string }> = [];
+    const streams = new Map<string, Route>();
+    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));
+    }
+  });
+}

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<string, unknown>;
+  };
+};
+
+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);
+});