fix(history,web): stop user messages rendering as duplicate bubbles

source/cydo/server/app.d

@@ -2308,12 +2308,18 @@ class App
 			if (jsonParse!TypeProbe(ts[0].translated).type != "item/started")
 				return; // tool_result etc. — keep awaiting the echo
 			auto ev = jsonParse!ItemStartedEvent(ts[0].translated);
-			// Prefer the echo's native uuid — it matches the bubble the live
-			// stdout echo created; the enqueue anchor is the fallback.
-			auto uuid = ev.uuid.length > 0 ? ev.uuid : td.queueTailAwaitingUuids[0];
-			emitUserMessageConsumed(tid, uuid,
+			// The confirmation's identity is the awaited anchor (or the nonce
+			// correlation), matching the provisional or optimistic bubble the
+			// client holds; the echo's own uuid travels as native_uuid so the
+			// client knows the canonical message follows and drops that bubble.
+			// Using the echo uuid as the identity matched nothing (that bubble
+			// does not exist yet), so the nonce correlation upgraded the
+			// optimistic placeholder in place instead, the echo could no longer
+			// displace it, and every live message rendered twice.
+			auto nativeUuid = ev.uuid.length > 0 ? ev.uuid : td.queueTailAwaitingUuids[0];
+			emitUserMessageConsumed(tid, td.queueTailAwaitingUuids[0],
 				ev.is_steering ? "steering" : "turn_start",
-				td.queueTailAwaitingNonces[0], uuid);
+				td.queueTailAwaitingNonces[0], nativeUuid);
 		}
 		else if (ta.isAssistantMessageLine(line))
 		{

web/src/sessionReducer.test.ts

@@ -987,3 +987,179 @@ describe("cydo/task_spawned reducer", () => {
     expect(s.pendingCydoTaskItemIds).toEqual([]);
   });
 });
+
+describe("user message identity dedup", () => {
+  const canonical = (uuid: string, text: string) =>
+    asEvent({
+      type: "item/started",
+      item_id: "cc-user-msg",
+      item_type: "user_message",
+      uuid,
+      content: [{ type: "text", text }],
+      is_replay: true,
+    });
+
+  it("updates in place when the same user record arrives twice", () => {
+    // overlapping history replays merge without a reset, so the same record
+    // can be delivered again; two full bubbles for one prompt is never right
+    let state = reduceMessage(makeState(), canonical("u-1", "same prompt"), 4);
+    state = reduceMessage(state, canonical("u-1", "same prompt"), 9);
+
+    const users = state.messages.filter((m) => m.type === "user");
+    expect(users).toHaveLength(1);
+    expect(users[0]?.seq).toBe(9);
+  });
+
+  it("keeps distinct records as distinct bubbles", () => {
+    let state = reduceMessage(makeState(), canonical("u-1", "first"));
+    state = reduceMessage(state, canonical("u-2", "second"));
+    expect(state.messages.filter((m) => m.type === "user")).toHaveLength(2);
+  });
+
+  it("does not regress an upgraded bubble to pending on re-delivery", () => {
+    let state = reduceMessage(makeState(), canonical("u-1", "prompt"));
+    const again = asEvent({
+      type: "item/started",
+      item_id: "cc-user-msg",
+      item_type: "user_message",
+      uuid: "u-1",
+      content: [{ type: "text", text: "prompt" }],
+      pending: true,
+      is_replay: true,
+    });
+    state = reduceMessage(state, again);
+    const users = state.messages.filter((m) => m.type === "user");
+    expect(users).toHaveLength(1);
+    expect(users[0]?.pending).toBeUndefined();
+  });
+});
+
+describe("user_message/consumed canonical-follows handling", () => {
+  const provisional = (uuid: string, text: string) =>
+    asEvent({
+      type: "item/started",
+      item_id: "cc-user-msg",
+      item_type: "user_message",
+      uuid,
+      pending: true,
+      content: [{ type: "text", text }],
+      is_replay: true,
+    });
+  const consumed = (fields: object) =>
+    asEvent({ type: "user_message/consumed", ...fields });
+
+  it("upgrades a uuid-less optimistic placeholder in place", () => {
+    // live flow: the optimistic placeholder is the message's display (no
+    // separate echo bubble renders live), so the nonce-correlated
+    // confirmation must upgrade it, never drop it
+    const state0 = {
+      ...makeState(),
+      messages: [
+        {
+          id: "opt-1",
+          type: "user" as const,
+          content: [{ type: "text" as const, text: "hi" }],
+          ackState: 3 as const,
+          nonce: "n-1",
+          pending: true,
+        },
+      ],
+    };
+    const state = reduceMessage(
+      state0,
+      consumed({
+        uuid: "enqueue-10",
+        native_uuid: "native-1",
+        correlation_id: "n-1",
+        consumed_as: "turn_start",
+      }),
+    );
+    const users = state.messages.filter((m) => m.type === "user");
+    expect(users).toHaveLength(1);
+    expect(users[0]?.pending).toBeUndefined();
+  });
+
+  it("heals a stored pair whose confirmation names the echo identity", () => {
+    // histories recorded while the live tail misnamed the confirmation carry
+    // uuid === native_uuid; the FIFO front of the pending enqueue bubbles is
+    // the message it described
+    let state = reduceMessage(makeState(), provisional("enqueue-5", "prompt"));
+    state = reduceMessage(
+      state,
+      consumed({
+        uuid: "native-9",
+        native_uuid: "native-9",
+        consumed_as: "turn_start",
+      }),
+    );
+    expect(state.messages.filter((m) => m.uuid === "enqueue-5")).toHaveLength(
+      0,
+    );
+  });
+
+  it("replays a stored unconfirmed+consumed+echo sequence as one bubble", () => {
+    // the exact sequence a replayed history delivers for a message sent live:
+    // the unconfirmed placeholder, the consumed confirmation that upgrades it
+    // in place (no echo bubble renders live), then the canonical echo, which
+    // must displace the upgraded placeholder rather than append beside it
+    const state0 = {
+      ...makeState(),
+      messages: [
+        {
+          id: "unconfirmed-1",
+          type: "user" as const,
+          content: [{ type: "text" as const, text: "the prompt" }],
+          ackState: 3 as const,
+          nonce: "corr-1",
+          pending: true,
+        },
+      ],
+    };
+    let state = reduceMessage(
+      state0,
+      consumed({
+        uuid: "enqueue-1",
+        native_uuid: "native-7",
+        correlation_id: "corr-1",
+        consumed_as: "turn_start",
+      }),
+    );
+    // upgraded in place, still one bubble, no longer pending
+    expect(state.messages).toHaveLength(1);
+    expect(state.messages[0]?.pending).toBeUndefined();
+
+    state = reduceMessage(
+      state,
+      asEvent({
+        type: "item/started",
+        item_id: "cc-user-msg",
+        item_type: "user_message",
+        uuid: "native-7",
+        content: [{ type: "text", text: "the prompt" }],
+        is_replay: true,
+      }),
+    );
+    const users = state.messages.filter((m) => m.type === "user");
+    expect(users).toHaveLength(1);
+    expect(users[0]?.uuid).toBe("native-7");
+  });
+
+  it("leaves a genuinely queued later bubble alone", () => {
+    // a correctly named confirmation resolves its own bubble directly and
+    // must not fall back onto some other still-queued message
+    let state = reduceMessage(makeState(), provisional("enqueue-5", "same"));
+    state = reduceMessage(state, provisional("enqueue-6", "same"));
+    state = reduceMessage(
+      state,
+      consumed({
+        uuid: "enqueue-5",
+        native_uuid: "native-9",
+        consumed_as: "turn_start",
+      }),
+    );
+    const remaining = state.messages.filter((m) =>
+      m.uuid?.startsWith("enqueue-"),
+    );
+    expect(remaining.map((m) => m.uuid)).toEqual(["enqueue-6"]);
+  });
+});

web/src/sessionReducer.ts

@@ -939,6 +939,18 @@ function reduceItemStartedUserMessage(
     isPendingUserMsg(m) &&
     !m.uuid &&
     (eventNonce ? m.nonce === eventNonce : !m.nonce || hasSameContent(m));
+  // A replayed canonical echo also supersedes an already-upgraded placeholder:
+  // a consumed confirmation upgrades the placeholder in place (correct live,
+  // where no echo bubble renders), so when the echo does follow on replay the
+  // bubble is no longer pending yet is still the same message. Canonical
+  // messages always carry uuids, so they are never displaced; this widening
+  // applies only to replay echoes, keeping two genuine identical live sends
+  // as two bubbles.
+  const isReplayDisposableUserMsg = (m: DisplayMessage) =>
+    (m.type === "user" &&
+      !m.uuid &&
+      (eventNonce ? m.nonce === eventNonce : !m.nonce || hasSameContent(m))) ||
+    isReplayDisposablePendingUserMsg(m);
 
   // Extract cydoMeta from the pending placeholder BEFORE the is_replay filter
   // removes it from the message list.
@@ -955,7 +967,7 @@ function reduceItemStartedUserMessage(
     // One replay echo accounts for exactly one sent message: displace at
     // most one placeholder. Same-content placeholders from other sends
     // (distinct nonces) must keep their own bubbles.
-    const dropIdx = state.messages.findIndex(isReplayDisposablePendingUserMsg);
+    const dropIdx = state.messages.findIndex(isReplayDisposableUserMsg);
     if (dropIdx >= 0) {
       displacedPlaceholder = true;
       state = {
@@ -965,6 +977,34 @@ function reduceItemStartedUserMessage(
     }
   }
 
+  // A user record's uuid is its identity: the same record can reach the
+  // reducer more than once (overlapping history replays merge without a
+  // reset, and a live echo can precede the replayed copy). Assistant content
+  // dedups through its item ids; user messages appended twice became two
+  // full bubbles. Update the existing bubble in place instead.
+  if (event.uuid) {
+    const dupIdx = state.messages.findIndex(
+      (m) => m.type === "user" && m.uuid === event.uuid,
+    );
+    if (dupIdx >= 0) {
+      return {
+        ...state,
+        messages: state.messages.map((m, i) =>
+          i === dupIdx
+            ? {
+                ...m,
+                content: blocks,
+                seq: seq ?? m.seq,
+                ts: ts ?? m.ts,
+                rawSource: event,
+                cydoMeta: m.cydoMeta ?? eventCydoMeta,
+              }
+            : m,
+        ),
+      };
+    }
+  }
+
   if (event.pending) {
     const id = `user-echo-${++state.msgIdCounter}`;
     const echoMsg: DisplayMessage = {
@@ -1069,15 +1109,45 @@ export function reduceUserMessageConsumed(
   s: SessionState,
   event: UserMessageConsumedEvent,
 ): SessionState {
-  const idx = s.messages.findIndex(
+  let idx = s.messages.findIndex(
     (m) =>
       m.type === "user" &&
       ((event.uuid && m.uuid === event.uuid) ||
         (event.correlation_id && m.nonce === event.correlation_id)),
   );
+  // Histories recorded while the live tail misnamed the confirmation carry
+  // uuid === native_uuid (the echo's identity, which no bubble has yet) and
+  // nonces that replay strips, so the lookup misses and the provisional
+  // survived beside its echo. The queue is FIFO, so such a confirmation can
+  // only describe the oldest still-pending enqueue-emitted bubble; a
+  // correctly named confirmation never reaches this fallback.
+  if (
+    idx < 0 &&
+    event.native_uuid &&
+    event.native_uuid === event.uuid &&
+    event.consumed_as !== "removed"
+  ) {
+    idx = s.messages.findIndex(
+      (m) =>
+        m.type === "user" &&
+        m.pending === true &&
+        !!m.uuid?.startsWith("enqueue-"),
+    );
+    if (idx >= 0) {
+      // the canonical echo follows under its own identity: drop, not upgrade
+      return {
+        ...s,
+        messages: s.messages.filter((_, i) => i !== idx),
+      };
+    }
+  }
   if (idx < 0) return s;
 
   const target = s.messages[idx]!;
+  // Only an enqueue-emitted provisional is superseded by a canonical echo
+  // that follows in the same stream. A uuid-less optimistic placeholder is
+  // the live display of the message itself (no separate echo bubble renders
+  // live), so it upgrades in place below rather than being dropped.
   const canonicalFollows =
     event.native_uuid &&
     event.native_uuid !== event.uuid &&