fix(frontend): stop streamed deltas rendering twice and fragmenting in V1 chat (#15108)

2026-10-07 16:19:05 +08:00 · 2026-07-03 14:01:43 +02:00
parent 760d6e9b03
commit 531b2c195d
2 changed files with 286 additions and 9 deletions
@@ -385,3 +385,150 @@ describe("handleEventForUI - streaming deltas", () => {
    expect(isStreamingDeltaEvent(ui[1])).toBe(false);
  });
 });
+
+describe("handleEventForUI - durable events supersede streaming deltas", () => {
+  const userMessage: MessageEvent = {
+    id: "sup-user-1",
+    timestamp: "2026-07-03T00:00:00.000Z",
+    source: "user",
+    llm_message: { role: "user", content: [{ type: "text", text: "go" }] },
+    activated_microagents: [],
+    extended_content: [],
+  };
+
+  const mkDelta = (
+    id: string,
+    timestamp: string,
+    content: string | null,
+    reasoning_content: string | null = null,
+  ): StreamingDeltaEvent => ({
+    id,
+    timestamp,
+    source: "agent",
+    kind: "StreamingDeltaEvent",
+    content,
+    reasoning_content,
+  });
+
+  const mkAssistant = (timestamp: string, text: string): MessageEvent => ({
+    id: "sup-assistant-1",
+    timestamp,
+    source: "agent",
+    llm_message: { role: "assistant", content: [{ type: "text", text }] },
+    activated_microagents: [],
+    extended_content: [],
+  });
+
+  const mkAction = (timestamp: string, thoughtText: string): ActionEvent => ({
+    id: "sup-action-1",
+    timestamp,
+    source: "agent",
+    thought: [{ type: "text", text: thoughtText }],
+    thinking_blocks: [],
+    action: {
+      kind: "ExecuteBashAction",
+      command: "echo hello",
+      is_input: false,
+      timeout: null,
+      reset: false,
+    },
+    tool_name: "execute_bash",
+    tool_call_id: "sup-call-1",
+    tool_call: {
+      id: "sup-call-1",
+      type: "function",
+      function: {
+        name: "execute_bash",
+        arguments: '{"command": "echo hello"}',
+      },
+    },
+    llm_response_id: "sup-response-1",
+    security_risk: SecurityRisk.UNKNOWN,
+  });
+
+  it("drops a stale delta that arrives after a newer durable event (history replay race)", () => {
+    // On a page reload the WebSocket replay delivers durable events WITHOUT
+    // deltas, while the REST history includes them; the REST deltas can
+    // therefore arrive after the turn's final message is already rendered.
+    // Rendering them would fragment and duplicate the finished message.
+    const final = mkAssistant("2026-07-03T00:00:09.000Z", "Hello, world");
+    let ui: OpenHandsEvent[] = [userMessage];
+    ui = handleEventForUI(final, ui);
+    ui = handleEventForUI(
+      mkDelta("sup-late-1", "2026-07-03T00:00:01.000Z", "Hello"),
+      ui,
+    );
+
+    expect(ui).toEqual([userMessage, final]);
+  });
+
+  it("keeps a live delta that arrives after an earlier durable event", () => {
+    // The next step's narration streams after the previous step's action; it
+    // is newer than every durable event and must still render.
+    const action = mkAction("2026-07-03T00:00:02.000Z", "I'll run a command");
+    let ui: OpenHandsEvent[] = [userMessage, action];
+    const live = mkDelta("sup-live-1", "2026-07-03T00:00:03.000Z", "Next, ");
+    ui = handleEventForUI(live, ui);
+
+    expect(ui).toHaveLength(3);
+    expect(ui[2]).toBe(live);
+  });
+
+  it("removes the turn's preview deltas when a tool-call action arrives", () => {
+    // The action card renders its own thought (the same text that just
+    // streamed), so leaving the preview deltas in place displays every
+    // step's reasoning twice.
+    let ui: OpenHandsEvent[] = [userMessage];
+    ui = handleEventForUI(
+      mkDelta("sup-d1", "2026-07-03T00:00:01.000Z", "I'll run a command"),
+      ui,
+    );
+    const action = mkAction("2026-07-03T00:00:02.000Z", "I'll run a command");
+    ui = handleEventForUI(action, ui);
+
+    expect(ui).toEqual([userMessage, action]);
+  });
+
+  it("keeps a reasoning-only bubble when the superseding action lacks reasoning", () => {
+    // For many models the delta is the sole carrier of reasoning_content, so
+    // stripping the streamed text must not lose the reasoning display.
+    let ui: OpenHandsEvent[] = [userMessage];
+    ui = handleEventForUI(
+      mkDelta(
+        "sup-d5",
+        "2026-07-03T00:00:01.000Z",
+        "I'll run a command",
+        "considering the request",
+      ),
+      ui,
+    );
+    const action = mkAction("2026-07-03T00:00:02.000Z", "I'll run a command");
+    ui = handleEventForUI(action, ui);
+
+    expect(ui).toHaveLength(3);
+    const kept = ui[1] as StreamingDeltaEvent;
+    expect(isStreamingDeltaEvent(kept)).toBe(true);
+    expect(kept.content).toBeNull();
+    expect(kept.reasoning_content).toBe("considering the request");
+    expect(ui[2]).toBe(action);
+  });
+
+  it("replaces non-reconciling preview deltas with the final message instead of duplicating", () => {
+    // In a multi-step turn the deltas accumulated since the last user message
+    // include earlier steps' text, so the final summary never reconciles
+    // against them; previously both the stale bubble and the final message
+    // rendered. The durable final message is canonical.
+    let ui: OpenHandsEvent[] = [userMessage];
+    ui = handleEventForUI(
+      mkDelta("sup-d2", "2026-07-03T00:00:01.000Z", "Fin"),
+      ui,
+    );
+    const final = mkAssistant(
+      "2026-07-03T00:00:09.000Z",
+      "All three commands ran fine.",
+    );
+    ui = handleEventForUI(final, ui);
+
+    expect(ui).toEqual([userMessage, final]);
+  });
+});
@@ -43,6 +43,32 @@ const findLastUserMessageIndex = (events: OpenHandsEvent[]): number => {
  return -1;
 };

+/**
+ * Newest timestamp among the durable (non-delta, non-metrics) agent/environment
+ * events already rendered. Streaming deltas older than this are stale: their
+ * content is already represented by durable events (the WebSocket history
+ * replay does not include deltas, while the REST history does, so on a reload
+ * the REST deltas can arrive AFTER the turn's final message was rendered).
+ */
+const findNewestDurableTimestamp = (
+  events: OpenHandsEvent[],
+): string | null => {
+  let newest: string | null = null;
+  for (const uiEvent of events) {
+    const isPreviewOrMetrics =
+      isStreamingDeltaEvent(uiEvent) || isConversationStateUpdateEvent(uiEvent);
+    const isDurableSource =
+      uiEvent.source === "agent" || uiEvent.source === "environment";
+    if (!isPreviewOrMetrics && isDurableSource) {
+      const ts = uiEvent.timestamp;
+      if (typeof ts === "string" && (newest === null || ts > newest)) {
+        newest = ts;
+      }
+    }
+  }
+  return newest;
+};
+
 // Join text blocks WITHOUT a separator: streaming deltas concatenate content
 // tokens directly with no separator between LLM content blocks, so using "\n"
 // here would cause startsWith/findTextSegmentsInOrder to miss when reconciling
@@ -137,17 +163,38 @@ const finalizeStreamingDeltasInPlace = (

  const nextUiEvents = [...uiEvents];
  const streamedText = streamingSegments.join("");
-  let unstreamedSuffix = "";
+  let matched = false;
+  let lastMatchEnd = 0;

-  if (finalText.startsWith(streamedText)) {
-    unstreamedSuffix = finalText.slice(streamedText.length);
-  } else {
-    const match = findTextSegmentsInOrder(finalText, streamingSegments);
-    if (!match.matched) {
-      return null;
+  // The SDK strips the finalized text, so it may lack trailing whitespace the
+  // model streamed - tolerate that by also trying the trailing-trimmed
+  // streamed text (mirrors agent-canvas #1552).
+  for (const candidate of [streamedText, streamedText.trimEnd()]) {
+    if (candidate && finalText.startsWith(candidate)) {
+      matched = true;
+      lastMatchEnd = candidate.length;
+      break;
    }
-    unstreamedSuffix = finalText.slice(match.lastMatchEnd);
  }
+  if (!matched) {
+    const lastIndex = streamingSegments.length - 1;
+    const searchSegments = streamingSegments.map((segment, index) =>
+      index === lastIndex ? segment.trimEnd() : segment,
+    );
+    const match = findTextSegmentsInOrder(finalText, searchSegments);
+    matched = match.matched;
+    lastMatchEnd = match.lastMatchEnd;
+  }
+  if (!matched) {
+    // The streamed preview never reconciles (e.g. it contains earlier
+    // steps' text, or chunks were reordered in delivery). The durable final
+    // message is canonical: drop the preview deltas and render it once.
+    const removeSet = new Set(contentStreamingDeltas.map(({ index }) => index));
+    const cleaned = uiEvents.filter((_, index) => !removeSet.has(index));
+    cleaned.push(finalEvent);
+    return cleaned;
+  }
+  const unstreamedSuffix = finalText.slice(lastMatchEnd);

  const lastDeltaIndex = contentStreamingDeltas.at(-1)?.index;
  const lastDelta =
@@ -171,6 +218,62 @@ const finalizeStreamingDeltasInPlace = (
  return nextUiEvents;
 };

+/**
+ * A tool-call ``ActionEvent`` carries the agent's ``thought`` - the same text
+ * that was just streamed as deltas. The action card renders that thought
+ * itself, so once the action arrives, the streamed preview deltas for the turn
+ * are redundant: leaving them in place displays every step's reasoning twice.
+ *
+ * When the current turn's content-bearing deltas reconcile (in order) against
+ * the action's thought text, return uiEvents with those deltas removed so the
+ * caller can append the action as the single rendered copy. Returns ``null``
+ * (no change) when there is nothing to reconcile or the text doesn't match.
+ */
+const supersedeStreamingDeltasForAction = (
+  actionEvent: OpenHandsEvent,
+  uiEvents: OpenHandsEvent[],
+): OpenHandsEvent[] | null => {
+  if (!isActionEvent(actionEvent)) {
+    return null;
+  }
+  const lastUserMessageIndex = findLastUserMessageIndex(uiEvents);
+  const currentTurnDeltaIndexes = uiEvents
+    .map((uiEvent, index) => ({ uiEvent, index }))
+    .filter(
+      ({ uiEvent, index }) =>
+        index > lastUserMessageIndex && isStreamingDeltaEvent(uiEvent),
+    )
+    .map(({ index }) => index);
+
+  if (currentTurnDeltaIndexes.length === 0) {
+    return null;
+  }
+
+  // The action renders its own thought, so the streamed preview text is
+  // redundant once the durable action arrives. Text-matching is deliberately
+  // NOT required: chunk reordering between the delta stream and event
+  // persistence would otherwise leak stray preview bubbles that later merge
+  // with the next stream. For many models the delta is the sole carrier of
+  // reasoning_content, though - when the action does not render reasoning of
+  // its own, keep the delta as a reasoning-only bubble (content cleared)
+  // instead of dropping it (mirrors agent-canvas #1552).
+  const actionRendersReasoning =
+    Boolean(actionEvent.reasoning_content?.trim()) ||
+    (actionEvent.thinking_blocks?.length ?? 0) > 0;
+  const stripSet = new Set(currentTurnDeltaIndexes);
+  const nextUiEvents: OpenHandsEvent[] = [];
+  uiEvents.forEach((uiEvent, index) => {
+    if (!stripSet.has(index) || !isStreamingDeltaEvent(uiEvent)) {
+      nextUiEvents.push(uiEvent);
+      return;
+    }
+    if (!actionRendersReasoning && uiEvent.reasoning_content) {
+      nextUiEvents.push({ ...uiEvent, content: null });
+    }
+  });
+  return nextUiEvents;
+};
+
 /**
 * Handles adding an event to the UI events array
 * Replaces actions with observations when they arrive (so UI shows observation instead of action)
@@ -178,7 +281,10 @@ const finalizeStreamingDeltasInPlace = (
 *
 * StreamingDeltaEvent: consecutive deltas merge in place into a single growing
 * bubble; when the turn's final agent message arrives it is reconciled into that
- * bubble (see `finalizeStreamingDeltasInPlace`) rather than appended.
+ * bubble (see `finalizeStreamingDeltasInPlace`) rather than appended, and a
+ * tool-call action removes the deltas that streamed its thought (see
+ * `supersedeStreamingDeltasForAction`). Deltas that arrive out of order (older
+ * than an already-rendered durable event) are dropped as stale.
 *
 * ACPToolCallEvent dedup: multiple events share a ``tool_call_id`` as an ACP
 * tool call progresses (in_progress → completed / failed). Collapse them to
@@ -196,6 +302,20 @@ export const handleEventForUI = (
      return newUiEvents;
    }

+    // Drop stale deltas. Deltas are a live-preview mechanism; one that is
+    // older than the newest durable event arrived late (e.g. the REST history
+    // response racing the WebSocket replay, which omits deltas). Its content
+    // is already represented by durable events, so rendering it would fragment
+    // and duplicate the finished message.
+    const newestDurableTimestamp = findNewestDurableTimestamp(newUiEvents);
+    if (
+      newestDurableTimestamp !== null &&
+      typeof event.timestamp === "string" &&
+      event.timestamp < newestDurableTimestamp
+    ) {
+      return newUiEvents;
+    }
+
    // Merge into the most recent streaming delta, treating interleaved non-chat
    // events as transparent: the agent server emits periodic
    // ConversationStateUpdateEvents (metrics) mid-stream, and those land in
@@ -243,6 +363,16 @@ export const handleEventForUI = (
    }
  }

+  // A tool-call action renders its own thought; remove the preview deltas that
+  // streamed it, then fall through so the action is appended normally below.
+  if (isActionEvent(event) && event.action.kind !== "FinishAction") {
+    const superseded = supersedeStreamingDeltasForAction(event, newUiEvents);
+    if (superseded) {
+      superseded.push(event);
+      return superseded;
+    }
+  }
+
  if (isACPToolCallEvent(event)) {
    const existingIndex = newUiEvents.findIndex(
      (uiEvent) =>