fix(frontend): preserve submitted input during initial stream gap recovery (#6800)

* fix(frontend): preserve submitted input during gap recovery

* docs(frontend): clarify replay recovery input precedence

* docs(frontend): keep replay gap guidance within budget

---------

Co-authored-by: TYY-2005 <TYY-2005@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
Co-authored-by: TYY-2005 <334029581+TYY-2005@users.noreply.github.com>

frontend/src/AGENTS.md

@@ -192,13 +192,20 @@ export on a page error or non-advancing cursor; demos keep their fixture path.
 - **Run creation retries** are disabled only for the initial `POST /runs/stream` in `core/api/api-client.ts`, because an ambiguous failure can otherwise create duplicate runs. Automatic SSE recovery keeps the SDK's normal HTTP retry budget for `GET` requests to the server-provided `Location`, including transient 5xx responses, and forwards `Last-Event-ID`. Reads and explicit SSE joins keep their normal retry behavior. Regression tests cover both a failed creation POST and POST success → stream interruption → GET 504 → GET success.
 - **Run stream options** are sanitized by `core/api/stream-mode.ts`: the Gateway-supported set is `values`, `messages-tuple`, `updates`, `debug`, `tasks`, `checkpoints`, and `custom`; any request containing an unsupported mode throws before HTTP instead of being partially forwarded or silently defaulting to `values`. `streamResumable` is retained by thread hooks only for SDK-side reconnect bookkeeping but stripped before the HTTP request because the Gateway does not accept that request option; actual replay uses the SSE `Last-Event-ID` cursor. The main chat client's initial and rejoined streams are forced to incremental `messages-tuple`, `updates`, and `custom` modes so SDK lazy tracking cannot add repeated full-state `values` snapshots; explicitly requested non-snapshot modes such as `debug`, `tasks`, and `checkpoints` are preserved. `values` remains supported outside this chat wrapper for explicit state inspection and by the replay-gap durable-state reload below. The backend file-tool chunk batcher is keyed to `messages-tuple`, not `values`, so omitting snapshots must not regress `write_file` / `str_replace` streaming into one SSE frame per model token. `core/threads/stream-state.ts` folds the user-visible non-message fields from `updates` with the matching DeerFlow reducer semantics, including `tool_artifacts` capture/consumption updates (same-handle replacement preserves order, retention directives trim the oldest entries, and the absolute ceiling is 1000) and rejects irrelevant frames before calling the SDK mutator, while the SDK's `messages-tuple` manager remains the sole owner of live message chunk assembly and deduplication. Keep this boundary aligned with the backend request schema; `messages` and `events` are not supported and must not be forwarded.
 - **Active-run recovery** in `core/threads/hooks.ts` discovers pending/running runs when a reopened chat has no tab-local reconnect pointer. Remember SDK-completed run IDs for the mounted hook, including submitted and natively reconnected runs: stale or delayed runs reads must not rejoin a completed run and interrupt follow-up suggestions. Keep recovery retries bounded and continue allowing a different active run to reconnect. The SDK's same-tab reconnect keeps its pointer on error and never retries, so a failed reconnect (not a failed submit) releases the pointer and recovery takes over with its own bounded retries.
-- **SSE replay gaps** are handled in `core/api/api-client.ts`, which wraps both initial and joined run streams because the upstream SDK ignores unknown event names. An id-less backend `gap` control frame clears stale reconnect metadata, emits an internal `stream_replay_gap` custom event, reloads durable thread values, and resumes after the server-provided retained tail when one exists (or rejoins without a cursor if the buffer is empty), with up to five recovery rejoins after the original stream (six total stream calls on an all-gap exhaustion path). The wrapper remains a lazy async iterable because the SDK consumes it with `for await`. `core/threads/hooks.ts` clears optimistic/transient/subtask state, invalidates durable history caches, and shows the localized recovery warning; never let a gap fall through as a normal stream finish or cancel the still-running backend run.
-  Replay-gap recovery restores its tab-local run pointer only if the slot is empty
-  or still belongs to that run; never overwrite another run's pointer after an
-  asynchronous durable-state read. A stale foreign pointer may cost an extra
-  run lookup/rejoin on reload; recovery adds no server probe. The same-named
-  `rememberReconnectRun` in `core/threads/hooks.ts` stays unconditional:
-  active-run rejoin makes a fresh claim that must win over stored metadata.
+- **SSE replay gaps** (`core/api/api-client.ts`) wrap initial/joined streams as a lazy
+  async iterable: the SDK ignores unknown events. An id-less `gap` clears stale
+  reconnect metadata, emits a `stream_replay_gap` custom event, reloads durable values,
+  then rejoins after the server-provided retained tail (without a cursor for an empty
+  buffer). Allow five recovery rejoins after the original stream (six calls total).
+  `core/threads/hooks.ts` clears optimistic/transient/subtask state, invalidates durable
+  history caches and shows the localized warning; never finish normally on a gap or
+  cancel the running backend run. Merge submitted initial-stream input or saved
+  joined-stream input into each recovered values frame: the first checkpoint may not
+  yet contain the human message.
+  Restore the tab-local run pointer only if empty or still owned by this run, never
+  over another run's pointer after a durable-state read. A foreign pointer can cost
+  an extra lookup/rejoin on reload; add no server probe. The hooks' same-named
+  `rememberReconnectRun` remains unconditional: active-run rejoin claims ownership.
 - **Streaming Markdown rendering** is owned by `core/streamdown`: Streamdown's `animated` / `isAnimating` API handles incremental word animation, while the shared `streamdownRenderingPlugins` config registers the named code-highlighting and Mermaid plugins required by Streamdown 2.5. Keep wrappers and derived configs wired to that shared object; do not reintroduce a rehype plugin that wraps every word, because reparsing a growing block remounts old words and replays their animation.
 - Citation links in message and artifact Markdown must derive their `citation:` label from the full `ReactNode` children tree, since Streamdown may provide element or array children during streaming rather than a plain string.
 - Web-search steps accept either a provider result array or a Webz-style `{ query, returned_results, results }` envelope. Unwrap only in the web-search renderer; skip entries without string titles/URLs, preserve other tool payloads and apply the same link allowlist to both formats. UI regressions pin source titles/links and browser reloads, including malformed neighboring entries.

frontend/src/core/api/api-client.ts

@@ -111,10 +111,12 @@ function streamOptionSignal(options: unknown): AbortSignal | undefined {
  * Recover the submitted input before replaying an active run. The incremental
  * chat stream intentionally omits `values`, so a page reload can otherwise
  * receive the run's AI/tool chunks before its human message has reached the
- * durable history feed. `runs.get` retains the original graph input in
- * `kwargs.input`; merge it into the latest durable values for one synthetic
- * snapshot. Any read failure is deliberately ignored so reconnect semantics
- * remain unchanged for deployments without run metadata.
+ * durable history feed. Initial streams supply the sanitized `submittedInput`
+ * directly, avoiding a run-metadata read. Otherwise, use `kwargs.input` from
+ * the supplied run record, falling back to `runs.get` when needed. Merge the
+ * input into the latest durable values for one synthetic snapshot. Any read
+ * failure is deliberately ignored so reconnect semantics remain unchanged for
+ * deployments without run metadata.
  */
 async function loadReconnectInputSnapshot(
   client: LangGraphClient,
@@ -123,15 +125,19 @@ async function loadReconnectInputSnapshot(
   run?: Awaited<ReturnType<LangGraphClient["runs"]["get"]>>,
   durableValues?: unknown,
   signal?: AbortSignal,
+  submittedInput?: unknown,
 ): Promise<ReconnectInputSnapshot | undefined> {
   try {
-    const resolvedRun =
-      run ?? (await client.runs.get(threadId, runId, { signal }));
-    const runKwargs = Reflect.get(resolvedRun, "kwargs");
-    const input =
-      typeof runKwargs === "object" && runKwargs !== null
-        ? Reflect.get(runKwargs, "input")
-        : undefined;
+    let input = submittedInput;
+    if (input === undefined) {
+      const resolvedRun =
+        run ?? (await client.runs.get(threadId, runId, { signal }));
+      const runKwargs = Reflect.get(resolvedRun, "kwargs");
+      input =
+        typeof runKwargs === "object" && runKwargs !== null
+          ? Reflect.get(runKwargs, "input")
+          : undefined;
+    }
     const inputMessages =
       typeof input === "object" && input !== null
         ? Reflect.get(input, "messages")
@@ -347,6 +353,7 @@ async function* recoverStreamReplayGaps({
   resume,
   signal,
   reconnectRun,
+  submittedInput,
 }: {
   client: LangGraphClient;
   threadId: string | null | undefined;
@@ -355,6 +362,7 @@ async function* recoverStreamReplayGaps({
   resume: (runId: string, lastEventId?: string) => AsyncIterable<StreamPart>;
   signal?: AbortSignal;
   reconnectRun?: Awaited<ReturnType<LangGraphClient["runs"]["get"]>>;
+  submittedInput?: unknown;
 }): AsyncGenerator<StreamPart> {
   let stream = initialStream;
   let recoveryAttempts = 0;
@@ -404,18 +412,20 @@ async function* recoverStreamReplayGaps({
       });
     if (durableState.values != null) {
       // A gap can arrive after the initial hydration frame but before the
-      // input reaches the checkpoint. Rebuild the snapshot from run metadata
-      // so this recovery path cannot overwrite the rescued human message.
-      const recoveredSnapshot = reconnectRun
-        ? await loadReconnectInputSnapshot(
-            client,
-            threadId,
-            runId,
-            reconnectRun,
-            durableState.values,
-            signal,
-          )
-        : undefined;
+      // input reaches the checkpoint. Rebuild the snapshot from submitted or
+      // saved run input so recovery cannot overwrite the human message.
+      const recoveredSnapshot =
+        reconnectRun || submittedInput !== undefined
+          ? await loadReconnectInputSnapshot(
+              client,
+              threadId,
+              runId,
+              reconnectRun,
+              durableState.values,
+              signal,
+              submittedInput,
+            )
+          : undefined;
       yield {
         event: "values",
         data: recoveredSnapshot ?? durableState.values,
@@ -507,6 +517,7 @@ function createCompatibleClient(isMock?: boolean): LangGraphClient {
       initialStream,
       signal: streamOptionSignal(sanitizedPayload),
       reconnectRun: undefined,
+      submittedInput: sanitizedPayload?.input,
       resume: (resolvedRunId, lastEventId) => {
         // Keep the recovery run id available to the shared inactive-stream
         // handler even if the SDK omitted its onRunCreated callback.

frontend/tests/unit/core/api/api-client.test.ts

@@ -1053,7 +1053,18 @@ test("recovers a gap emitted by the initial run stream", async () => {
   for await (const entry of getAPIClient(true).runs.stream(
     "thread-2",
     "lead_agent",
-    { streamResumable: true },
+    {
+      input: {
+        messages: [
+          {
+            id: "human-2",
+            type: "human",
+            content: "Question awaiting checkpoint",
+          },
+        ],
+      },
+      streamResumable: true,
+    },
   )) {
     received.push(entry);
   }
@@ -1065,7 +1076,16 @@ test("recovers a gap emitted by the initial run stream", async () => {
     },
     {
       event: "values",
-      data: { messages: [{ type: "ai", content: "checkpoint" }] },
+      data: {
+        messages: [
+          { type: "ai", content: "checkpoint" },
+          {
+            id: "human-2",
+            type: "human",
+            content: "Question awaiting checkpoint",
+          },
+        ],
+      },
     },
     { event: "end", data: null },
   ]);