mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
fix(chat): per-event SSE ids for safe reconnect dedup + queue race recovery
This commit is contained in:
matevip
parent
dfe186f928
commit
7d02841d1d
@ -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<MessageContentPart> 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;
|
||||
}
|
||||
|
||||
@ -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<SseEvent> 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).
|
||||
*
|
||||
* <p>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<MessageContentPart> 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.
|
||||
*
|
||||
* <p>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.
|
||||
*
|
||||
* <p>Strategy (must be called under {@code state.lock}):
|
||||
* <ol>
|
||||
* <li>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.</li>
|
||||
* <li>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.</li>
|
||||
* </ol>
|
||||
*/
|
||||
private static void trimBuffer(List<SseEvent> buffer) {
|
||||
if (buffer.size() <= MAX_BUFFER_SIZE) return;
|
||||
int target = buffer.size() - MAX_BUFFER_SIZE;
|
||||
|
||||
// 第一步：合并连续的同类型 delta 事件，拼接 delta 文本而非丢弃
|
||||
List<SseEvent> 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<SseEvent> 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<SseEvent> 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 清理 ====================
|
||||
|
||||
@ -713,7 +713,13 @@ const segments = computed<MessageSegment[]>(() => {
|
||||
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 })
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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<boolean>
|
||||
/** 当前错误 */
|
||||
error: import('vue').Ref<Error | null>
|
||||
/**
|
||||
* 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<string | null>
|
||||
/** 连接流 */
|
||||
connect: (body?: any) => Promise<void>
|
||||
/** 断开连接 */
|
||||
@ -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<SSEEventType, Set<(data: any) => 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<string | null>(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<string>()
|
||||
|
||||
// 触发事件
|
||||
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,
|
||||
|
||||
Loading…
Reference in New Issue
Block a user