fix(frontend): dedupe injected user messages during long runs (#4620)

frontend/AGENTS.md

@@ -70,7 +70,7 @@ The frontend is a stateful chat application. Users create **threads** (conversat
    File-tool artifact auto-open work must run in an effect with timer cleanup; never schedule timers while rendering streamed `write_file` or `str_replace` updates.
    `ThreadState.artifacts` remains the authoritative artifact list. The artifacts provider persists only thread-scoped panel UI state (`open`, selected path, and a refresh bootstrap cache) in session storage; an initial empty stream value must not overwrite that restored state before history finishes loading.
    Formal artifact content is refreshed once when the run finishes; transient `write-file:` previews remain message-driven.
-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 at their matching canonical identities (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. The resolver suppresses checkpoint/transient prefixes whose canonical position is still behind an unloaded cursor page instead of collapsing that unknown gap before a recent anchor, then adds optimistic messages without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded.
+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 at their matching canonical identities (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. The resolver suppresses checkpoint/transient prefixes whose canonical position is still behind an unloaded cursor page instead of collapsing that unknown gap before a recent anchor, then adds optimistic messages without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded. Dynamic context re-keys the submitted user message from `X` to `X__user`; UI identity matching normalizes that reserved suffix only for human messages so the submitted frame and checkpoint replacement remain one visible turn. A locally submitted turn also records its pre-submit identity baseline: if `messages-tuple` publishes new AI/tool steps before `values` publishes that turn's human message, render ordering moves only those non-baseline visible steps behind the new human while leaving history, hidden controls, and reconnected runs untouched. Keep that local order anchor 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 actions call the LangGraph SDK stream stop path; `core/threads/hooks.ts` invalidates current-thread, thread-history, token-usage, and sidebar/search caches immediately and schedules one follow-up refetch because SDK stop may finish via abort + fire-and-forget cancel before backend title finalization commits
 5. TanStack Query manages server state; localStorage stores user settings. The
    Settings > Tools MCP switch calls the targeted `PATCH /api/mcp/config`

frontend/src/core/threads/hooks.ts

@@ -166,6 +166,7 @@ export function buildThreadSubmitMessages({
 const EMPTY_MESSAGES: Message[] = [];
 const EMPTY_RUN_MESSAGES: RunMessage[] = [];
 const EMPTY_MESSAGE_IDENTITIES: readonly string[] = [];
+const INJECTED_USER_MESSAGE_ID_SUFFIX = "__user";
 
 const EMPTY_THREAD_VALUES: AgentThreadState = {
   title: "",
@@ -192,7 +193,17 @@ function messageIdentity(message: Message): string | undefined {
     return `tool:${message.tool_call_id}`;
   }
   if (typeof message.id === "string" && message.id.length > 0) {
-    return `message:${message.id}`;
+    // DynamicContextMiddleware replaces the submitted HumanMessage(id=X) with
+    // a hidden SystemMessage(id=X) and the real HumanMessage(id=X__user).
+    // Treat those human copies as one UI message so a committed render ledger
+    // cannot retain X beside the later checkpoint copy X__user.
+    const messageId =
+      message.type === "human" &&
+      message.id.endsWith(INJECTED_USER_MESSAGE_ID_SUFFIX)
+        ? message.id.slice(0, -INJECTED_USER_MESSAGE_ID_SUFFIX.length) ||
+          message.id
+        : message.id;
+    return `message:${messageId}`;
   }
   return undefined;
 }
@@ -581,6 +592,59 @@ export function mergeMessages(
   });
 }
 
+/**
+ * Keep messages from a locally submitted turn behind that turn's user input.
+ * LangGraph `messages-tuple` events can publish the first AI/tool steps before
+ * the `values` event containing the user message. Those steps are not part of
+ * the pre-submit baseline, so move only that visible pending segment behind the
+ * first new human message without disturbing established history or hidden
+ * checkpoint controls. The caller keeps the baseline after stream completion
+ * because the SDK may retain its transient event order until the next submit.
+ */
+export function restoreLocalTurnMessageOrder(
+  messages: Message[],
+  baselineMessageIdentities: ReadonlySet<string>,
+): Message[] {
+  const pendingHumanIndex = messages.findIndex((message) => {
+    const identity = messageIdentity(message);
+    return (
+      message.type === "human" &&
+      !isHiddenFromUIMessage(message) &&
+      identity !== undefined &&
+      !baselineMessageIdentities.has(identity)
+    );
+  });
+  if (pendingHumanIndex <= 0) {
+    return messages;
+  }
+
+  const stablePrefix: Message[] = [];
+  const earlyPendingSteps: Message[] = [];
+  for (const message of messages.slice(0, pendingHumanIndex)) {
+    const identity = messageIdentity(message);
+    const isVisiblePendingStep =
+      (message.type === "ai" || message.type === "tool") &&
+      !isHiddenFromUIMessage(message) &&
+      identity !== undefined &&
+      !baselineMessageIdentities.has(identity);
+    if (isVisiblePendingStep) {
+      earlyPendingSteps.push(message);
+    } else {
+      stablePrefix.push(message);
+    }
+  }
+  if (earlyPendingSteps.length === 0) {
+    return messages;
+  }
+
+  return [
+    ...stablePrefix,
+    messages[pendingHumanIndex]!,
+    ...earlyPendingSteps,
+    ...messages.slice(pendingHumanIndex + 1),
+  ];
+}
+
 /**
  * Keep a run-scoped ledger of every visible message that reached a committed
  * UI frame. Live checkpoint windows can roll forward between two
@@ -1615,6 +1679,7 @@ export function useThreadStream({
         transientHistoryThreadIdRef.current = null;
         summarizedRef.current = new Set<string>();
         pendingUsageBaselineMessageIdsRef.current = new Set();
+        localTurnOrderBaselineIdentitiesRef.current = null;
         tasksRef.current = {};
         setTasks({});
         invalidateStoppedThreadCaches(queryClient, threadIdRef.current, isMock);
@@ -1734,6 +1799,12 @@ export function useThreadStream({
   const latestMessageCountsRef = useRef({ humanMessageCount });
   const sendInFlightRef = useRef(false);
   const messagesRef = useRef<Message[]>([]);
+  // Non-null only after a turn submitted by this mounted client. Keep it after
+  // finish/stop/error because the SDK can retain its transient event order in
+  // the settled frame. The next local submit replaces it and a thread switch or
+  // replay gap clears it. An empty set is meaningful for a new thread and must
+  // not be confused with a reconnect that has no local turn anchor.
+  const localTurnOrderBaselineIdentitiesRef = useRef<Set<string> | null>(null);
   // Current-stream lifecycle bridge for messages removed from the checkpoint
   // tail before the canonical run-event page refetch observes the journal
   // flush. It is never appended into useThreadHistory's persisted pages.
@@ -1782,6 +1853,7 @@ export function useThreadStream({
     };
     summarizedRef.current = new Set<string>();
     pendingUsageBaselineMessageIdsRef.current = new Set();
+    localTurnOrderBaselineIdentitiesRef.current = null;
     pendingPreparedReplayRef.current = null;
     setPendingSupersededRunIds(new Set());
     setPendingSupersededMessageIds(new Set());
@@ -1883,6 +1955,9 @@ export function useThreadStream({
           .map(messageIdentity)
           .filter((id): id is string => Boolean(id)),
       );
+      localTurnOrderBaselineIdentitiesRef.current = new Set(
+        pendingUsageBaselineMessageIdsRef.current,
+      );
 
       // Build optimistic files list with uploading status
       const optimisticFiles: FileInMessage[] = (message.files ?? []).map(
@@ -2049,6 +2124,7 @@ export function useThreadStream({
         setOptimisticThreadId(null);
         setLiveMessagesThreadId(null);
         setIsUploading(false);
+        localTurnOrderBaselineIdentitiesRef.current = null;
         throw error;
       } finally {
         sendInFlightRef.current = false;
@@ -2086,6 +2162,9 @@ export function useThreadStream({
           .map(messageIdentity)
           .filter((id): id is string => Boolean(id)),
       );
+      localTurnOrderBaselineIdentitiesRef.current = new Set(
+        pendingUsageBaselineMessageIdsRef.current,
+      );
       setLiveMessagesThreadId(threadId);
       listeners.current.onSend?.(threadId);
       let preparedSupersededRunId: string | null = null;
@@ -2172,6 +2251,7 @@ export function useThreadStream({
         setOptimisticMessages([]);
         setOptimisticThreadId(null);
         setLiveMessagesThreadId(null);
+        localTurnOrderBaselineIdentitiesRef.current = null;
         if (preparedSupersededRunId) {
           const supersededRunId = preparedSupersededRunId;
           pendingPreparedReplayRef.current = null;
@@ -2332,11 +2412,15 @@ export function useThreadStream({
       transientHistoryOrder,
       previouslyRenderedOrder,
     );
-    return mergeMessages(
+    const merged = mergeMessages(
       effectiveHistory,
       renderMessages,
       visibleOptimisticMessages,
     );
+    const localTurnOrderBaseline = localTurnOrderBaselineIdentitiesRef.current;
+    return localTurnOrderBaseline === null
+      ? merged
+      : restoreLocalTurnMessageOrder(merged, localTurnOrderBaseline);
   }, [
     previouslyRenderedOrder,
     renderMessages,

frontend/tests/unit/core/threads/local-turn-order.dom.test.tsx

@@ -0,0 +1,108 @@
+import type { Message } from "@langchain/langgraph-sdk";
+import { expect, rs, test } from "@rstest/core";
+import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
+import { act, renderHook } from "@testing-library/react";
+import { createElement, type ReactNode } from "react";
+
+import { I18nContext } from "@/core/i18n/context";
+import { DEFAULT_LOCAL_SETTINGS } from "@/core/settings/local";
+
+const streamMockState = rs.hoisted(() => ({
+  isLoading: false,
+  messages: [] as Message[],
+  onFinish: undefined as
+    | ((state: { values: { messages: Message[] } }) => void)
+    | undefined,
+  stop: rs.fn(async () => undefined),
+  submit: rs.fn(async () => undefined),
+}));
+
+rs.mock("@langchain/langgraph-sdk/react", () => ({
+  useStream: (options: {
+    onFinish?: (state: { values: { messages: Message[] } }) => void;
+  }) => {
+    streamMockState.onFinish = options.onFinish;
+    return {
+      isLoading: streamMockState.isLoading,
+      messages: streamMockState.messages,
+      stop: streamMockState.stop,
+      submit: streamMockState.submit,
+      values: {
+        artifacts: [],
+        messages: streamMockState.messages,
+        title: "",
+        todos: [],
+      },
+    };
+  },
+}));
+
+test("keeps early streamed steps behind a local user message after finish", async () => {
+  const { useThreadStream } = await import("@/core/threads/hooks");
+  const queryClient = new QueryClient({
+    defaultOptions: { queries: { retry: false } },
+  });
+  const wrapper = ({ children }: { children: ReactNode }) =>
+    createElement(
+      QueryClientProvider,
+      { client: queryClient },
+      createElement(
+        I18nContext.Provider,
+        {
+          value: {
+            locale: "en-US",
+            setLocale: () => undefined,
+          },
+        },
+        children,
+      ),
+    );
+  const { rerender, result } = renderHook(
+    () =>
+      useThreadStream({
+        context: DEFAULT_LOCAL_SETTINGS.context,
+        isMock: true,
+        threadId: "thread-1",
+      }),
+    { wrapper },
+  );
+
+  await act(async () => {
+    await result.current.sendMessage("thread-1", {
+      files: [],
+      text: "Build a presentation",
+    });
+  });
+
+  const earlyAssistantStep = {
+    id: "early-assistant-step",
+    type: "ai",
+    content: "Reading the presentation skill",
+  } as Message;
+  const injectedHuman = {
+    id: "current-request__user",
+    type: "human",
+    content: "Build a presentation",
+  } as Message;
+  streamMockState.messages = [earlyAssistantStep, injectedHuman];
+  streamMockState.isLoading = true;
+  rerender();
+
+  expect(result.current.thread.messages).toEqual([
+    injectedHuman,
+    earlyAssistantStep,
+  ]);
+
+  act(() => {
+    streamMockState.onFinish?.({
+      values: { messages: streamMockState.messages },
+    });
+    streamMockState.isLoading = false;
+    rerender();
+  });
+
+  expect(result.current.thread.messages).toEqual([
+    injectedHuman,
+    earlyAssistantStep,
+  ]);
+});

frontend/tests/unit/core/threads/message-merge.test.ts

@@ -22,6 +22,7 @@ import {
   removeSetItems,
   resolveThreadTransientHistoryBridge,
   resolveTransientHistoryBridge,
+  restoreLocalTurnMessageOrder,
   type ThreadMessagesPageResponse,
 } from "@/core/threads/hooks";
 import type { RunMessage } from "@/core/threads/types";
@@ -1502,6 +1503,126 @@ test("rendered message ledger survives rolling live windows before repeated comp
   );
 });
 
+test("rendered message ledger replaces a submitted user message with its injected server copy", () => {
+  const submittedHuman = {
+    id: "request-1",
+    type: "human",
+    content: "Build a presentation",
+  } as Message;
+  const injectedSystemReminder = {
+    id: "request-1",
+    type: "system",
+    content: "<system-reminder>today</system-reminder>",
+    additional_kwargs: { hide_from_ui: true },
+  } as Message;
+  const injectedMemory = {
+    id: "request-1__memory",
+    type: "human",
+    content: "<memory>context</memory>",
+    additional_kwargs: { hide_from_ui: true },
+  } as Message;
+  const injectedHuman = {
+    id: "request-1__user",
+    type: "human",
+    content: "Build a presentation",
+    name: "user-input",
+  } as Message;
+  const assistantStep = {
+    id: "assistant-step-1",
+    type: "ai",
+    content: "Reading the presentation skill",
+  } as Message;
+
+  const firstLedger = mergeRenderedMessageLedger([], [submittedHuman]);
+  const nextFrame = mergeMessages(
+    [submittedHuman],
+    [injectedSystemReminder, injectedMemory, injectedHuman, assistantStep],
+    [],
+  ).filter((message) => message.additional_kwargs?.hide_from_ui !== true);
+  const nextLedger = mergeRenderedMessageLedger(firstLedger, nextFrame);
+
+  expect(nextLedger).toEqual([injectedHuman, assistantStep]);
+  expect(nextLedger.filter((message) => message.type === "human")).toHaveLength(
+    1,
+  );
+});
+
+test("local turn order keeps early streamed steps behind the user message", () => {
+  const previousHuman = {
+    id: "previous-human",
+    type: "human",
+    content: "Previous request",
+  } as Message;
+  const previousAssistant = {
+    id: "previous-assistant",
+    type: "ai",
+    content: "Previous answer",
+  } as Message;
+  const earlyAssistantStep = {
+    id: "early-assistant-step",
+    type: "ai",
+    content: "Reading the presentation skill",
+  } as Message;
+  const optimisticHuman = {
+    id: "opt-human-current",
+    type: "human",
+    content: "Build a presentation",
+  } as Message;
+  const injectedHuman = {
+    id: "current-request__user",
+    type: "human",
+    content: "Build a presentation",
+  } as Message;
+  const injectedMemory = {
+    id: "current-request__memory",
+    type: "human",
+    content: "<memory>context</memory>",
+    additional_kwargs: { hide_from_ui: true },
+  } as Message;
+  const laterAssistantStep = {
+    id: "later-assistant-step",
+    type: "ai",
+    content: "Writing the presentation plan",
+  } as Message;
+  const baselineIdentities = new Set([
+    "message:previous-human",
+    "message:previous-assistant",
+  ]);
+
+  expect(
+    restoreLocalTurnMessageOrder(
+      [previousHuman, previousAssistant, earlyAssistantStep, optimisticHuman],
+      baselineIdentities,
+    ),
+  ).toEqual([
+    previousHuman,
+    previousAssistant,
+    optimisticHuman,
+    earlyAssistantStep,
+  ]);
+
+  expect(
+    restoreLocalTurnMessageOrder(
+      [
+        previousHuman,
+        previousAssistant,
+        earlyAssistantStep,
+        injectedMemory,
+        injectedHuman,
+        laterAssistantStep,
+      ],
+      baselineIdentities,
+    ),
+  ).toEqual([
+    previousHuman,
+    previousAssistant,
+    injectedMemory,
+    injectedHuman,
+    earlyAssistantStep,
+    laterAssistantStep,
+  ]);
+});
+
 test("rendered message ledger does not retain explicitly superseded messages", () => {
   const retained = {
     id: "retained-answer",