From 7d02841d1d9d5286eb760ca1d8783de2a01727e8 Mon Sep 17 00:00:00 2001 From: matevip Date: Sun, 3 May 2026 19:57:06 +0800 Subject: [PATCH] fix(chat): per-event SSE ids for safe reconnect dedup + queue race recovery --- .../vip/mate/channel/web/ChatController.java | 11 +- .../mate/channel/web/ChatStreamTracker.java | 210 +++++++++--------- .../src/components/chat/MessageBubble.vue | 8 +- mateclaw-ui/src/composables/chat/useChat.ts | 70 ++++-- mateclaw-ui/src/composables/chat/useStream.ts | 101 ++++++++- 5 files changed, 272 insertions(+), 128 deletions(-) diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java index 2aa7652f..a16eb3a9 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java @@ -107,7 +107,8 @@ public class ChatController { // without sticky session)". They look identical from attach()'s // boolean return, but the user-facing remediation is different. boolean existsLocally = streamTracker.streamExistsOnThisNode(conversationId); - boolean attached = streamTracker.attach(conversationId, emitter); + long lastEventId = request.getLastEventId() == null ? 0L : request.getLastEventId(); + boolean attached = streamTracker.attach(conversationId, emitter, lastEventId); if (!attached) { try { if (existsLocally) { @@ -1024,6 +1025,14 @@ public class ChatController { private List contentParts; /** true 表示断线重连，不发送新消息，只附着到已有的流 */ private Boolean reconnect; + /** + * Last SSE event id the client has already processed. Only meaningful + * when {@link #reconnect} is true — the server skips events with + * id ≤ this value during buffer replay so the client doesn't + * see them twice. 0 (or null) means "replay everything", matching + * the legacy attach behavior for backwards compatibility. + */ + private Long lastEventId; /** 思考深度：off / low / medium / high / max，null 表示跟随 Agent 默认 */ private String thinkingLevel; } diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java index 68dd9745..d0830d50 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java @@ -124,7 +124,13 @@ public class ChatStreamTracker { return iterationEventsEnabled; } - record SseEvent(String name, String json) {} + /** + * One buffered SSE event. The {@code id} is a per-conversation monotonic + * sequence — the SSE protocol's standard {@code id:} line carries this + * value so the client can echo it back via {@code lastEventId} when + * reconnecting, allowing us to skip already-delivered events on replay. + */ + record SseEvent(long id, String name, String json) {} /** * 中断类型：区分用户主动停止和用户在运行中追加新消息 @@ -142,6 +148,15 @@ public class ChatStreamTracker { final List buffer = new ArrayList<>(); final Object lock = new Object(); volatile boolean done; + /** + * Monotonic sequence used as the SSE protocol {@code id:} field. + * Incremented inside {@code state.lock} as each event is buffered, + * so the buffer is always in (id-asc) order. On reconnect, the + * client echoes its last-seen id back via {@code lastEventId} and + * we skip events whose id is ≤ that value during replay — + * eliminating the duplicate-delivery class of bugs. + */ + long nextEventId = 0L; /** Flux 订阅的 Disposable，用于取消 LLM 流 */ volatile Disposable disposable; /** 停止标志：requestStop() 设为 true，各图节点和 LLM 调用检查此标志以提前退出 */ @@ -535,8 +550,9 @@ public class ChatStreamTracker { if (isDone || isAsyncTask) { if (state == null) return; - SseEvent ev = new SseEvent(eventName, jsonData); synchronized (state.lock) { + long id = ++state.nextEventId; + SseEvent ev = new SseEvent(id, eventName, jsonData); state.buffer.add(ev); if (state.buffer.size() > MAX_BUFFER_SIZE) { trimBuffer(state.buffer); @@ -545,7 +561,7 @@ public class ChatStreamTracker { while (it.hasNext()) { SseEmitter emitter = it.next(); try { - emitter.send(SseEmitter.event().name(eventName).data(jsonData)); + emitter.send(SseEmitter.event().id(String.valueOf(id)).name(eventName).data(jsonData)); if (isDone) { log.debug("Sent final 'done' event to subscriber for {}", conversationId); } @@ -588,8 +604,9 @@ public class ChatStreamTracker { return; } - SseEvent event = new SseEvent(eventName, jsonData); synchronized (state.lock) { + long id = ++state.nextEventId; + SseEvent event = new SseEvent(id, eventName, jsonData); state.buffer.add(event); // buffer 容量保护：超出上限时优先丢弃 thinking_delta（占比最大且非关键） if (state.buffer.size() > MAX_BUFFER_SIZE) { @@ -599,7 +616,7 @@ public class ChatStreamTracker { while (it.hasNext()) { SseEmitter emitter = it.next(); try { - emitter.send(SseEmitter.event().name(eventName).data(jsonData)); + emitter.send(SseEmitter.event().id(String.valueOf(id)).name(eventName).data(jsonData)); } catch (IOException | IllegalStateException e) { log.debug("Removing dead subscriber for {}: {}", conversationId, e.getMessage()); it.remove(); @@ -821,21 +838,53 @@ public class ChatStreamTracker { * @return true 如果成功附着或重放（订阅者已加入或事件已重放完毕），false 如果没有任何状态可恢复 */ public boolean attach(String conversationId, SseEmitter emitter) { + return attach(conversationId, emitter, 0L); + } + + /** + * Reconnect-aware attach: replays only events whose id > + * {@code lastEventId}. Pass 0 to replay everything (fresh attach + * behavior — same as the no-arg overload). + * + *

The id is the per-conversation monotonic sequence stamped on + * each {@link SseEvent} when it was first emitted. Frontend tracks + * the last id it processed and echoes it back via the request + * body's {@code lastEventId} field, eliminating the duplicate- + * delivery class of bugs (the symptom: thinking segments rendered + * with the wrong iterationIndex because frontend processed the + * same {@code iteration_start} twice). + */ + public boolean attach(String conversationId, SseEmitter emitter, long lastEventId) { RunState state = runs.get(conversationId); if (state == null) { return false; } synchronized (state.lock) { - // 回放全部缓冲事件（包含 done 事件本身——见 broadcast 的 done 分支） + // Replay buffer with id-based dedup. Each buffered event keeps its + // original (1:1) id, so the skip condition is the simple + // `id <= lastEventId`. trimBuffer no longer merges delta events, + // so a single id always corresponds to a single contiguous run of + // text — there's no straddling-range edge case. + int replayed = 0; + int skipped = 0; for (SseEvent event : state.buffer) { + if (event.id() <= lastEventId) { + skipped++; + continue; + } try { - emitter.send(SseEmitter.event().name(event.name()).data(event.json())); + emitter.send(SseEmitter.event().id(String.valueOf(event.id())).name(event.name()).data(event.json())); + replayed++; } catch (IOException | IllegalStateException e) { log.warn("Failed to replay buffer to reconnecting client for {}: {}", conversationId, e.getMessage()); return false; } } + if (lastEventId > 0 && skipped > 0) { + log.info("[SSE] Reconnect dedup for {}: skipped {} already-seen events, replayed {} new", + conversationId, skipped, replayed); + } // Stream complete: buffer replayed (including the `done` event itself). // We DO NOT auto-complete the emitter here — keep it subscribed so any // late-arriving async_task_* events (image/video/music generation that @@ -1215,6 +1264,17 @@ public class ChatStreamTracker { public boolean enqueueMessage(String conversationId, String message, Long agentId, boolean persisted, List contentParts) { RunState state = runs.get(conversationId); + // Reject when there's no live producer to drain the queue: + // - state == null: conversation truly gone (cleanup completed) + // - state.done: stream's doOnComplete has already fired and + // called completeAndConsumeIfLast — no later + // consumer is guaranteed to invoke + // startQueuedMessage. Accepting an enqueue here + // would silently park the message in memory + // until the 5-minute retention sweep deletes it. + // Frontend treats `queued: false` as the cue to fall back to a fresh + // send (after the stale isGenerating settles), eliminating the + // race that previously merged messages into the prior turn. if (state == null || state.done) { return false; } @@ -1323,116 +1383,56 @@ public class ChatStreamTracker { } /** - * 将 buffer 裁剪到 MAX_BUFFER_SIZE 以内。 - * 策略：将连续的同类型 delta 事件合并为一条（拼接 delta 文本，保留完整内容但减少条目数）。 - * 如果合并后仍超限，丢弃最早的 thinking_delta（thinking 对重连恢复不是关键内容）。 - * 必须在 state.lock 内调用。 + * Trim the replay buffer to {@link #MAX_BUFFER_SIZE} entries while + * preserving SSE-id semantics required by reconnect dedup. + * + *

We deliberately do NOT merge delta events even though it would + * reduce entry count more aggressively. Merging concatenates a range + * of original event ids into a single record; on reconnect a client + * whose {@code lastEventId} falls inside the merged range would + * either re-receive the head text (replay = duplicate) or lose the + * tail text (skip = data loss). Both are correctness bugs, and the + * dropping strategy below avoids them entirely — events kept in the + * buffer always correspond 1:1 to the ids the client originally saw. + * + *

Strategy (must be called under {@code state.lock}): + *

    + *
  1. Drop earliest {@code thinking_delta} entries — thinking text + * is not part of the canonical answer; losing the head of a + * very long reasoning trace on reconnect is acceptable.
    2. + *
  2. If still over the cap, drop earliest {@code content_delta} + * entries. This loses visible answer text, but only after we've + * buffered > {@link #MAX_BUFFER_SIZE} events — >1 MB of + * output. Rare enough that we accept the trade-off rather + * than mangle reconnect semantics.
    3. + *
*/ private static void trimBuffer(List buffer) { if (buffer.size() <= MAX_BUFFER_SIZE) return; + int target = buffer.size() - MAX_BUFFER_SIZE; - // 第一步：合并连续的同类型 delta 事件，拼接 delta 文本而非丢弃 - List compacted = new ArrayList<>(buffer.size()); - int i = 0; - while (i < buffer.size()) { - SseEvent current = buffer.get(i); - if ("thinking_delta".equals(current.name()) || "content_delta".equals(current.name())) { - // 收集连续同类型 delta 的文本 - StringBuilder merged = new StringBuilder(); - merged.append(extractDelta(current.json())); - int j = i + 1; - while (j < buffer.size() && current.name().equals(buffer.get(j).name())) { - merged.append(extractDelta(buffer.get(j).json())); - j++; - } - // 合并为一条事件 - compacted.add(new SseEvent(current.name(), buildDeltaJson(merged.toString()))); - i = j; - } else { - compacted.add(current); - i++; + // Pass 1: drop earliest thinking_delta entries. + Iterator it = buffer.iterator(); + while (it.hasNext() && target > 0) { + SseEvent e = it.next(); + if ("thinking_delta".equals(e.name())) { + it.remove(); + target--; } } - // 第二步：如果仍超限，丢弃最早的 thinking_delta（对重连恢复不是关键） - if (compacted.size() > MAX_BUFFER_SIZE) { - Iterator it = compacted.iterator(); - int removed = 0; - int target = compacted.size() - MAX_BUFFER_SIZE; - while (it.hasNext() && removed < target) { + // Pass 2: if still over the cap, drop earliest content_delta entries. + if (target > 0) { + it = buffer.iterator(); + while (it.hasNext() && target > 0) { SseEvent e = it.next(); - if ("thinking_delta".equals(e.name())) { + if ("content_delta".equals(e.name())) { it.remove(); - removed++; + target--; } } } - - buffer.clear(); - buffer.addAll(compacted); - log.debug("Buffer trimmed: {} events", buffer.size()); - } - - /** - * 从 delta JSON（如 {"delta":"text"}）中提取 delta 值 - */ - private static String extractDelta(String json) { - // 快速解析 {"delta":"..."} — 避免引入完整 JSON 解析器依赖 - int idx = json.indexOf("\"delta\""); - if (idx < 0) return ""; - int colonIdx = json.indexOf(':', idx); - if (colonIdx < 0) return ""; - int startQuote = json.indexOf('"', colonIdx + 1); - if (startQuote < 0) return ""; - StringBuilder sb = new StringBuilder(); - for (int k = startQuote + 1; k < json.length(); k++) { - char c = json.charAt(k); - if (c == '\\' && k + 1 < json.length()) { - char next = json.charAt(k + 1); - if (next == '"') { sb.append('"'); k++; } - else if (next == '\\') { sb.append('\\'); k++; } - else if (next == 'n') { sb.append('\n'); k++; } - else if (next == 't') { sb.append('\t'); k++; } - else if (next == 'r') { sb.append('\r'); k++; } - else if (next == '/') { sb.append('/'); k++; } - else if (next == 'b') { sb.append('\b'); k++; } - else if (next == 'f') { sb.append('\f'); k++; } - else if (next == 'u' && k + 5 < json.length()) { - // Unicode escape: backslash-u followed by 4 hex digits - String hex = json.substring(k + 2, k + 6); - try { - sb.append((char) Integer.parseInt(hex, 16)); - k += 5; - } catch (NumberFormatException e) { - sb.append(c); // 无法解析，保留原样 - } - } - else { sb.append(c); } - } else if (c == '"') { - break; - } else { - sb.append(c); - } - } - return sb.toString(); - } - - /** - * 构建 delta JSON 字符串 - */ - private static String buildDeltaJson(String delta) { - StringBuilder sb = new StringBuilder("{\"delta\":\""); - for (int k = 0; k < delta.length(); k++) { - char c = delta.charAt(k); - if (c == '"') sb.append("\\\""); - else if (c == '\\') sb.append("\\\\"); - else if (c == '\n') sb.append("\\n"); - else if (c == '\t') sb.append("\\t"); - else if (c == '\r') sb.append("\\r"); - else sb.append(c); - } - sb.append("\"}"); - return sb.toString(); + log.debug("Buffer trimmed: {} events remain", buffer.size()); } // ==================== Stale RunState 清理 ==================== diff --git a/mateclaw-ui/src/components/chat/MessageBubble.vue b/mateclaw-ui/src/components/chat/MessageBubble.vue index ac0a61be..3167d8d8 100644 --- a/mateclaw-ui/src/components/chat/MessageBubble.vue +++ b/mateclaw-ui/src/components/chat/MessageBubble.vue @@ -713,7 +713,13 @@ const segments = computed(() => { if (!hasThinking) { const thinkingPart = props.message.contentParts?.find(p => p.type === 'thinking') if (thinkingPart?.text) { - segs.unshift({ id: 'th-fb', type: 'thinking', status: 'completed', thinkingText: thinkingPart.text }) + // Tag with iterationIndex=0 so groupedIterations puts it in the FIRST + // iteration's thinking bucket instead of the default-zero bucket + // colliding with later iteration content. Without this, the fallback + // thinking renders below the answer for any conversation that has + // multi-iteration RFC-22 segments tagged elsewhere. + const firstIter = segs.find(s => typeof s.iterationIndex === 'number')?.iterationIndex ?? 0 + segs.unshift({ id: 'th-fb', type: 'thinking', status: 'completed', thinkingText: thinkingPart.text, iterationIndex: firstIter }) } } diff --git a/mateclaw-ui/src/composables/chat/useChat.ts b/mateclaw-ui/src/composables/chat/useChat.ts index 515538ee..86553d0a 100644 --- a/mateclaw-ui/src/composables/chat/useChat.ts +++ b/mateclaw-ui/src/composables/chat/useChat.ts @@ -540,8 +540,14 @@ export function useChat(options: UseChatOptions): UseChatReturn { reconnectingForAsyncTasks = true // Defer one tick so the current 'done' handler chain finishes before // disconnect() fires inside connect(). + // lastEventId is NOT passed here — connect() owns the per-conversation + // dedup state and injects its own lastEventId only when the target + // conversation matches what the dedup state was tracking. setTimeout(() => { - stream.connect({ conversationId: targetConv, reconnect: true }) + stream.connect({ + conversationId: targetConv, + reconnect: true, + }) .catch(() => { /* swallow — handled by stream.error event */ }) .finally(() => { reconnectingForAsyncTasks = false }) }, 50) @@ -1444,7 +1450,13 @@ export function useChat(options: UseChatOptions): UseChatReturn { const isApprovalCommand = /^\/(approve|deny)$/i.test(content.trim()) // ===== Sending while generating: route to interrupt / queue path ===== - if (isGenerating.value && !isApprovalCommand) { + // _skipQueueRoute is set by the stale-state recovery path (when /interrupt + // returned queued=false because the backend stream is gone). It forces a + // single direct fresh-send attempt regardless of isGenerating, with a hard + // cap on recursive entries to prevent infinite loops if something is + // genuinely stuck. + const skipQueueRoute = (options as any)._skipQueueRoute === true + if (isGenerating.value && !isApprovalCommand && !skipQueueRoute) { return await handleInterruptOrQueue(content, options) } @@ -1525,24 +1537,42 @@ export function useChat(options: UseChatOptions): UseChatReturn { streamPhase.value = 'interrupting' messageQueue.markSending() } else if (result.data?.queued) { - // Non-interruptible but queued: will auto-resume when the current step ends + // Non-interruptible but queued: will auto-resume when the current step ends. + // (Backend now also returns queued=true while state.done is still within + // its retention window — see ChatStreamTracker.enqueueMessage. This + // closes the race that previously caused the new submit to bypass the + // queue, race the previous turn's `done` handler, and merge into the + // previous user message bubble.) streamPhase.value = 'queued' } else { - // No active stream — send directly - messageQueue.clear() - createUserMessage(content, options.contentParts, conversationId) - resetCurrentTurnState() - const assistantMessage = createAssistantMessage('', conversationId) - ;(assistantMessage as any)._turnId = activeTurnId - currentAssistantId.value = assistantMessage.id as string - streamPhase.value = thinkingLevelRef?.value === 'off' ? 'streaming' : 'thinking' - phaseInfo.value = null - await stream.connect({ - agentId, - message: content, - conversationId, - contentParts: options.contentParts || [], - }) + // queued=false now means there's no producer to drain into: + // - state == null → conversation cleaned up post-retention + // - state.done → stream's doOnComplete already fired and + // no consumer remains to call startQueuedMessage + // Either way, accepting another queue entry would silently park + // the message in memory until cleanup; instead, drop the local + // queue entry and restart as a fresh send. Pass _skipQueueRoute + // so the recursive sendMessage takes the normal path even if the + // frontend's isGenerating still reads true (e.g. the previous + // turn's `done` event hasn't landed yet) — without this flag the + // recursion loops back into handleInterruptOrQueue and gets the + // same queued=false response. + console.warn('[useChat] interrupt returned queued=false (RunState gone or done); restarting as fresh send') + const stale = messageQueue.dequeue() + const restartParts = stale?.contentParts ?? options.contentParts + // setTimeout(0) yields to any in-flight `done` handler that's + // about to flip isGenerating itself; the _skipQueueRoute is the + // belt-and-braces guard for the case where it never lands. + setTimeout(() => { + sendMessage(content, { + ...options, + contentParts: restartParts, + _skipQueueRoute: true, + } as SendMessageOptions & { _skipQueueRoute: boolean }).catch(err => { + console.error('[useChat] restart-after-stale-interrupt failed:', err) + error.value = err instanceof Error ? err : new Error(String(err)) + }) + }, 0) } } catch (e) { console.error('[useChat] Interrupt request failed:', e) @@ -1673,6 +1703,10 @@ export function useChat(options: UseChatOptions): UseChatReturn { currentAssistantId.value = assistantMessage.id as string try { + // connect() owns lastEventId injection — it knows whether the dedup + // state still applies to this conversation. Passing it from out here + // would race the per-conv reset that connect does and could leak a + // different conv's id into this reconnect. await stream.connect({ conversationId, reconnect: true, diff --git a/mateclaw-ui/src/composables/chat/useStream.ts b/mateclaw-ui/src/composables/chat/useStream.ts index 35afdf47..d95a0243 100644 --- a/mateclaw-ui/src/composables/chat/useStream.ts +++ b/mateclaw-ui/src/composables/chat/useStream.ts @@ -62,6 +62,13 @@ export type SSEEventType = export interface SSEEvent { type: SSEEventType data: any + /** + * Server-assigned per-conversation monotonic id (the SSE protocol's + * native `id:` line). Frontend de-dupes by this value so a reconnect + * replay doesn't double-process events the client already saw. + * Absent on legacy events from servers that don't stamp ids. + */ + id?: string } export interface UseStreamOptions { @@ -84,6 +91,12 @@ export interface UseStreamReturn { isReceiving: import('vue').Ref /** 当前错误 */ error: import('vue').Ref + /** + * Highest SSE event id seen so far this connection. Pass back to the + * server as {@code lastEventId} on reconnect to skip already-delivered + * events and avoid duplicate handler dispatch. + */ + lastEventId: import('vue').Ref /** 连接流 */ connect: (body?: any) => Promise /** 断开连接 */ @@ -134,10 +147,11 @@ class SSEParser { let eventType: SSEEventType = 'content_delta' let data: any = {} let hasData = false + let eventId: string | undefined for (const line of lines) { if (!line.trim()) continue - + const colonIndex = line.indexOf(':') if (colonIndex === -1) continue @@ -146,6 +160,8 @@ class SSEParser { if (key === 'event') { eventType = value as SSEEventType + } else if (key === 'id') { + eventId = value } else if (key === 'data') { hasData = true try { @@ -156,7 +172,10 @@ class SSEParser { } } - return hasData ? { type: eventType, data } : null + if (!hasData) return null + return eventId !== undefined + ? { type: eventType, data, id: eventId } + : { type: eventType, data } } } @@ -177,8 +196,40 @@ export function useStream(options: UseStreamOptions): UseStreamReturn { const eventHandlers = new Map void>>() const globalHandlers = new Set<(event: SSEEvent) => void>() + /** + * Largest server event id seen on this stream. Echoed back in the + * `lastEventId` request body field on reconnect so the server can + * skip events the client has already processed. + */ + const lastEventId = ref(null) + + /** + * Set of event ids already dispatched to handlers this connection. + * Reconnect replays the same id'd events; without this set, handlers + * fire twice and `iteration_start` / `thinking_delta` end up with + * the wrong segment ordering. Cleared on connect() / disconnect(). + */ + const seenEventIds = new Set() + // 触发事件 const emit = (event: SSEEvent) => { + // De-dup by server-assigned id. Events without an id (legacy / heartbeat) + // bypass — they're either idempotent or carry their own dedup logic. + if (event.id) { + if (seenEventIds.has(event.id)) { + return + } + seenEventIds.add(event.id) + // Track highest id for reconnect Last-Event-ID echo. String compare is + // fine because ids are zero-padded server-side... actually they're plain + // numbers, so coerce to BigInt-safe numeric compare. + const incoming = Number(event.id) + const current = lastEventId.value === null ? -1 : Number(lastEventId.value) + if (!Number.isNaN(incoming) && incoming > current) { + lastEventId.value = event.id + } + } + // 全局处理器 globalHandlers.forEach(handler => { try { @@ -208,12 +259,55 @@ export function useStream(options: UseStreamOptions): UseStreamReturn { } // 连接流 + // Conversation last seen by the dedup state. SSE ids are per-conversation + // on the server, so a lastEventId carried over from a different conv would + // mass-skip valid events on the new conv's reconnect (e.g. user processed + // events 1..1000 in conv A, then reconnects to conv B which has events + // 1..50 — the server filter `id <= 1000` would drop EVERY event in B). + let lastDedupConversationId: string | null = null + const connect = async (body?: any) => { // 断开已有连接 disconnect() - + parser = new SSEParser() error.value = null + + const incomingConv = body?.conversationId ?? null + const isReconnect = !!body?.reconnect + // Reset dedup state when: + // - Fresh stream (not a reconnect) — always clears, matches the + // pre-fix behavior for normal sends. + // - Reconnect targeting a DIFFERENT conversation than the one whose + // ids are in our state — the per-conversation server semantics + // make a cross-conv lastEventId actively harmful. + // Reconnect to the SAME conversation preserves dedup state so the + // server can skip already-seen events on replay. + const sameConv = isReconnect && lastDedupConversationId === incomingConv + if (!sameConv) { + seenEventIds.clear() + lastEventId.value = null + } + lastDedupConversationId = incomingConv + + // Own the lastEventId injection here. Callers must NOT put their own + // lastEventId in the body — connect() is the only layer that knows + // whether the dedup state still applies to this conversation. Reading + // a stale `stream.lastEventId.value` from outside (and embedding it + // in the body before this point) would fail to clear after a conv + // switch. We only inject when the dedup state is still relevant + // (sameConv) AND we actually have an id to echo. + if (sameConv && lastEventId.value !== null && body && body.reconnect) { + const numericId = Number(lastEventId.value) + if (!Number.isNaN(numericId)) { + body = { ...body, lastEventId: numericId } + } + } else if (body && 'lastEventId' in body) { + // Defensive: strip any caller-provided lastEventId so a refactor + // can't reintroduce the cross-conv bug. + const { lastEventId: _, ...rest } = body + body = rest + } try { abortController = new AbortController() @@ -394,6 +488,7 @@ export function useStream(options: UseStreamOptions): UseStreamReturn { isConnected, isReceiving, error, + lastEventId, connect, disconnect, abort,