fix(session): don't drop queued messages; guard selectSession races (#120)
Two real bugs found in an adversarial review of the core real-time path:
1. Queued message vanishes (high impact): the message.updated handler dropped
EVERY temp- optimistic message when any real message arrived, so sending a
second message while the first was still processing made the second
disappear from the chat until its own event landed ('did my message send?').
Extracted the merge into a tested pure helper (mergeIncomingMessage) that
resolves only the oldest pending temp of the same role.
2. selectSession race: rapidly switching sessions on a flaky network could let
a slow fetch for a previous session overwrite currentSession/messages of the
newer selection. Added a monotonic sequence token; a stale result is
discarded.
Also reviewed but intentionally NOT changed: the SSE-reconnect-on-connection-
switch path (already handled via the [client] effect cleanup + reconnect) and
abortSession leaving 'sending' set on failure (deliberate — the run may still
be live; per its own comment).
Claude-Session: https://claude.ai/code/session_01T12AhSnQVrSxNnvwfCx2z6
Co-authored-by: engineer <engineer@macbookpro.lan>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
48
src/lib/message-merge.test.ts
Normal file
48
src/lib/message-merge.test.ts
Normal file
@@ -0,0 +1,48 @@
|
||||
import { test } from "node:test"
|
||||
import assert from "node:assert/strict"
|
||||
import { mergeIncomingMessage } from "./message-merge.ts"
|
||||
import type { Message } from "./sdk.ts"
|
||||
|
||||
const msg = (id: string, role: "user" | "assistant" = "user"): Message => ({
|
||||
id,
|
||||
sessionID: "s1",
|
||||
role,
|
||||
time: { created: 1 },
|
||||
})
|
||||
|
||||
test("appends an assistant reply that has no optimistic placeholder", () => {
|
||||
const out = mergeIncomingMessage([msg("temp-1", "user")], msg("real-a", "assistant"))
|
||||
assert.deepEqual(out.map((m) => m.id), ["temp-1", "real-a"])
|
||||
})
|
||||
|
||||
test("replaces the single pending temp user message", () => {
|
||||
const out = mergeIncomingMessage([msg("temp-1", "user")], msg("real-1", "user"))
|
||||
assert.deepEqual(out.map((m) => m.id), ["real-1"])
|
||||
})
|
||||
|
||||
test("resolving the first message keeps a second queued temp message (the bug)", () => {
|
||||
// User sent msg1 (temp-1) then msg2 (temp-2) while msg1 was still processing.
|
||||
const list = [msg("temp-1", "user"), msg("temp-2", "user")]
|
||||
// The real event for msg1 arrives first.
|
||||
const out = mergeIncomingMessage(list, msg("real-1", "user"))
|
||||
// temp-1 is resolved; temp-2 must remain so it doesn't vanish from the UI.
|
||||
assert.deepEqual(out.map((m) => m.id), ["real-1", "temp-2"])
|
||||
})
|
||||
|
||||
test("updates an already-present message in place without reordering", () => {
|
||||
const list = [msg("real-1", "user"), msg("real-2", "assistant")]
|
||||
const updated: Message = { ...msg("real-2", "assistant"), time: { created: 1, completed: 9 } }
|
||||
const out = mergeIncomingMessage(list, updated)
|
||||
assert.deepEqual(out.map((m) => m.id), ["real-1", "real-2"])
|
||||
assert.equal(out[1].time.completed, 9)
|
||||
})
|
||||
|
||||
test("appends a real message when there is no matching temp and it is new", () => {
|
||||
const out = mergeIncomingMessage([msg("real-1", "user")], msg("real-2", "user"))
|
||||
assert.deepEqual(out.map((m) => m.id), ["real-1", "real-2"])
|
||||
})
|
||||
|
||||
test("does not remove a temp of a different role", () => {
|
||||
const out = mergeIncomingMessage([msg("temp-1", "user")], msg("real-a", "assistant"))
|
||||
assert.ok(out.some((m) => m.id === "temp-1"))
|
||||
})
|
||||
32
src/lib/message-merge.ts
Normal file
32
src/lib/message-merge.ts
Normal file
@@ -0,0 +1,32 @@
|
||||
import type { Message } from "./sdk"
|
||||
|
||||
/**
|
||||
* Merge a server `message.updated` event into the current message list.
|
||||
*
|
||||
* Optimistic sends add placeholder messages with `temp-` ids (see
|
||||
* sessions.ts sendMessage). When the real server message arrives we must
|
||||
* replace ONLY the oldest still-pending optimistic message of the same role —
|
||||
* not every `temp-` message.
|
||||
*
|
||||
* The bug this guards against: sending a second message while the first is
|
||||
* still processing leaves two `temp-` user messages in the list. A naive
|
||||
* "drop all temp messages when any real one arrives" would remove BOTH when
|
||||
* the first real message lands, so the second (already queued server-side but
|
||||
* not yet echoed back) vanishes from the chat until its own event arrives —
|
||||
* a "did my message send?" ghosting bug.
|
||||
*/
|
||||
export function mergeIncomingMessage(messages: Message[], message: Message): Message[] {
|
||||
// Already present (a later update to a message we've seen): replace in place.
|
||||
if (messages.some((m) => m.id === message.id)) {
|
||||
return messages.map((m) => (m.id === message.id ? message : m))
|
||||
}
|
||||
// First time we see this real message: resolve the oldest matching temp.
|
||||
const tempIdx = messages.findIndex((m) => m.id.startsWith("temp-") && m.role === message.role)
|
||||
if (tempIdx !== -1) {
|
||||
const next = messages.slice()
|
||||
next[tempIdx] = message
|
||||
return next
|
||||
}
|
||||
// No optimistic placeholder to resolve (e.g. an assistant reply): append.
|
||||
return [...messages, message]
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import { useSettings } from "./settings"
|
||||
import { addBreadcrumb } from "../lib/sentry"
|
||||
import { AnalyticsEvent, track } from "../lib/analytics"
|
||||
import { extractPromptFromParts, type PromptFromParts } from "../lib/prompt-from-parts"
|
||||
import { mergeIncomingMessage } from "../lib/message-merge"
|
||||
|
||||
// Helper to convert API response to our internal format
|
||||
function parseMessages(response: MessageWithParts[]): { messages: Message[]; parts: Record<string, Part[]> } {
|
||||
@@ -69,6 +70,12 @@ export type RevertResult = ({ ok: true } & PromptFromParts) | { ok: false; reaso
|
||||
// this module) clears entries on busy and checks them on busy -> idle.
|
||||
export const abortedSessions = new Set<string>()
|
||||
|
||||
// Monotonic token guarding selectSession against out-of-order resolution: a
|
||||
// slow fetch for a session the user has already navigated away from must not
|
||||
// overwrite the messages/currentSession of a newer selection. Each call takes
|
||||
// the next value and only commits its result if still the latest.
|
||||
let selectSeq = 0
|
||||
|
||||
// Get the right client for a session's directory
|
||||
function clientFor(directory?: string): Client | null {
|
||||
const connState = useConnections.getState()
|
||||
@@ -119,6 +126,7 @@ export const useSessions = create<SessionsState>((set, get) => ({
|
||||
return
|
||||
}
|
||||
|
||||
const seq = ++selectSeq
|
||||
addBreadcrumb({ category: "session", message: "select", data: { sessionID, hasDirectory: Boolean(directory) } })
|
||||
try {
|
||||
// Reset optimistic sending — SSE sessionStatus is the source of truth
|
||||
@@ -135,6 +143,10 @@ export const useSessions = create<SessionsState>((set, get) => ({
|
||||
client.session.messages(sessionID, { limit: pageSize() }),
|
||||
])
|
||||
|
||||
// A newer selectSession started while we were fetching — discard this
|
||||
// stale result so it can't clobber the newer selection.
|
||||
if (seq !== selectSeq) return
|
||||
|
||||
// Parse the API response format: array of { info, parts }
|
||||
const { messages, parts } = parseMessages(messagesResponse)
|
||||
|
||||
@@ -147,6 +159,7 @@ export const useSessions = create<SessionsState>((set, get) => ({
|
||||
hasMore: messagesResponse.length >= pageSize(),
|
||||
})
|
||||
} catch (err) {
|
||||
if (seq !== selectSeq) return
|
||||
console.error("Failed to load session:", err)
|
||||
set({ error: "Failed to load session", isLoading: false })
|
||||
}
|
||||
@@ -407,14 +420,7 @@ export const useSessions = create<SessionsState>((set, get) => ({
|
||||
const message = (props.info || props.message) as Message | undefined
|
||||
if (!message || message.sessionID !== currentSession.id) return
|
||||
|
||||
set((state) => {
|
||||
// Remove temp messages when real ones arrive
|
||||
const filtered = state.messages.filter((m) => !m.id.startsWith("temp-") || m.id === message.id)
|
||||
const exists = filtered.some((m) => m.id === message.id)
|
||||
return {
|
||||
messages: exists ? filtered.map((m) => (m.id === message.id ? message : m)) : [...filtered, message],
|
||||
}
|
||||
})
|
||||
set((state) => ({ messages: mergeIncomingMessage(state.messages, message) }))
|
||||
break
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user