diff --git a/source/cydo/server/app.d b/source/cydo/server/app.d index 48fe243d..2b756fd2 100644 --- a/source/cydo/server/app.d +++ b/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)) { diff --git a/web/src/sessionReducer.test.ts b/web/src/sessionReducer.test.ts index 91a10c13..49399476 100644 --- a/web/src/sessionReducer.test.ts +++ b/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"]); + }); +}); diff --git a/web/src/sessionReducer.ts b/web/src/sessionReducer.ts index 07092c4c..5329d704 100644 --- a/web/src/sessionReducer.ts +++ b/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 &&