fix(web): late callback dedup across invocation boundaries

packages/web/src/hooks/reply-target-compat.ts

@@ -0,0 +1,3 @@
+export function isReplyTargetCompatible(existingReplyTo?: string, incomingReplyTo?: string): boolean {
+  return (existingReplyTo ?? null) === (incomingReplyTo ?? null);
+}

packages/web/src/hooks/useAgentMessages.ts

@@ -4,6 +4,7 @@ import { useCallback, useEffect, useRef } from 'react';
 import { recordDebugEvent } from '@/debug/invocationEventDebug';
 import { useChatStore } from '@/stores/chatStore';
 import { compactToolResultDetail } from '@/utils/toolPreview';
+import { isReplyTargetCompatible } from './reply-target-compat';
 
 /** Timeout for done(isFinal) - 5 minutes */
 const DONE_TIMEOUT_MS = 5 * 60 * 1000;
@@ -106,8 +107,12 @@ export function useAgentMessages() {
 
   /** #586 follow-up: Track just-finalized stream bubble per cat. Set on done when
    *  activeRefs entry existed, consumed by callback replacement or next invocation start.
-   *  Prevents the greedy scan from matching arbitrary historical messages. */
-  const finalizedStreamRef = useRef<Map<string, string>>(new Map());
+   *  Prevents the greedy scan from matching arbitrary historical messages.
+   *  fencedAt: set by invocation_created to mark the boundary — allows late callbacks
+   *  within a grace period but blocks new-invocation callbacks (P1 guard). */
+  const finalizedStreamRef = useRef<Map<string, { messageId: string; invocationId?: string; fencedAt?: number }>>(
+    new Map(),
+  );
 
   /** Bug C P2: Track whether stream data was received per cat (avoids false catch-up on callback-only flows) */
   const sawStreamDataRef = useRef<Set<string>>(new Set());
@@ -256,69 +261,96 @@ export function useAgentMessages() {
     [getCurrentInvocationIdForCat],
   );
 
-  const findCallbackReplacementTarget = useCallback((catId: string, invocationId: string): { id: string } | null => {
-    const currentMessages = useChatStore.getState().messages;
-    for (let i = currentMessages.length - 1; i >= 0; i -= 1) {
-      const msg = currentMessages[i];
-      if (
-        msg?.type === 'assistant' &&
-        msg.catId === catId &&
-        msg.origin === 'stream' &&
-        msg.extra?.stream?.invocationId === invocationId
-      ) {
-        return { id: msg.id };
+  const findCallbackReplacementTarget = useCallback(
+    (catId: string, invocationId: string): { id: string; matchKind: 'exact' } | null => {
+      const currentMessages = useChatStore.getState().messages;
+      for (let i = currentMessages.length - 1; i >= 0; i -= 1) {
+        const msg = currentMessages[i];
+        if (
+          msg?.type === 'assistant' &&
+          msg.catId === catId &&
+          msg.origin === 'stream' &&
+          msg.extra?.stream?.invocationId === invocationId
+        ) {
+          return { id: msg.id, matchKind: 'exact' };
+        }
       }
-    }
-    return null;
-  }, []);
+      return null;
+    },
+    [],
+  );
 
-  const findInvocationlessStreamPlaceholder = useCallback((catId: string): { id: string } | null => {
-    const currentMessages = useChatStore.getState().messages;
-    const activeId = activeRefs.current.get(catId)?.id;
+  const findInvocationlessStreamPlaceholder = useCallback(
+    (
+      catId: string,
+      callbackContent?: string,
+      callbackReplyTo?: string,
+    ):
+      | { id: string; matchKind: 'active_invocationless' }
+      | { id: string; matchKind: 'finalized_fallback'; replacementInvocationId?: string }
+      | null => {
+      const currentMessages = useChatStore.getState().messages;
+      const activeId = activeRefs.current.get(catId)?.id;
 
-    if (activeId) {
-      const activeMessage = currentMessages.find(
-        (msg) =>
-          msg.id === activeId &&
-          msg.type === 'assistant' &&
-          msg.catId === catId &&
-          msg.origin === 'stream' &&
-          !msg.extra?.stream?.invocationId,
-      );
-      if (activeMessage) {
-        return { id: activeMessage.id };
+      if (activeId) {
+        const activeMessage = currentMessages.find(
+          (msg) =>
+            msg.id === activeId &&
+            msg.type === 'assistant' &&
+            msg.catId === catId &&
+            msg.origin === 'stream' &&
+            !msg.extra?.stream?.invocationId,
+        );
+        if (activeMessage) {
+          return { id: activeMessage.id, matchKind: 'active_invocationless' };
+        }
       }
-    }
 
-    // First pass: find actively-streaming invocationless bubble
-    for (let i = currentMessages.length - 1; i >= 0; i -= 1) {
-      const msg = currentMessages[i];
-      if (
-        msg?.type === 'assistant' &&
-        msg.catId === catId &&
-        msg.origin === 'stream' &&
-        msg.isStreaming &&
-        !msg.extra?.stream?.invocationId
-      ) {
-        return { id: msg.id };
+      // First pass: find actively-streaming invocationless bubble
+      for (let i = currentMessages.length - 1; i >= 0; i -= 1) {
+        const msg = currentMessages[i];
+        if (
+          msg?.type === 'assistant' &&
+          msg.catId === catId &&
+          msg.origin === 'stream' &&
+          msg.isStreaming &&
+          !msg.extra?.stream?.invocationId
+        ) {
+          return { id: msg.id, matchKind: 'active_invocationless' };
+        }
       }
-    }
 
-    // #586 follow-up: Check finalizedStreamRef — the done handler records the
-    // exact message ID of the just-finalized stream bubble. This avoids the
-    // greedy scan that could match arbitrary historical messages (P1 from review).
-    const finalizedId = finalizedStreamRef.current.get(catId);
-    if (finalizedId) {
-      const finalized = currentMessages.find(
-        (m) => m.id === finalizedId && m.type === 'assistant' && m.catId === catId && m.origin === 'stream',
-      );
-      if (finalized) {
-        return { id: finalized.id };
+      // #586 follow-up: Check finalizedStreamRef — the done handler records the
+      // exact message ID of the just-finalized stream bubble. This avoids the
+      // greedy scan that could match arbitrary historical messages (P1 from review).
+      const finalizedEntry = finalizedStreamRef.current.get(catId);
+      if (finalizedEntry) {
+        const finalized = currentMessages.find(
+          (m) =>
+            m.id === finalizedEntry.messageId && m.type === 'assistant' && m.catId === catId && m.origin === 'stream',
+        );
+        if (finalized) {
+          // #586 P1 guard: if invocation_created has fired since finalization
+          // (fencedAt is set), apply time + content fence to prevent a callback-only
+          // inv-2 from silently overwriting inv-1's finalized bubble.
+          if (finalizedEntry.fencedAt) {
+            const LATE_CALLBACK_GRACE_MS = 5_000;
+            if (Date.now() - finalizedEntry.fencedAt > LATE_CALLBACK_GRACE_MS) return null;
+            if (callbackContent !== undefined && finalized.content !== callbackContent) return null;
+          }
+          if (!isReplyTargetCompatible(finalized.replyTo, callbackReplyTo)) return null;
+          return {
+            id: finalized.id,
+            matchKind: 'finalized_fallback',
+            replacementInvocationId: finalizedEntry.invocationId ?? finalized.extra?.stream?.invocationId,
+          };
+        }
       }
-    }
 
-    return null;
-  }, []);
+      return null;
+    },
+    [],
+  );
 
   const getOrRecoverActiveAssistantMessageId = useCallback(
     (catId: string, metadata?: AgentMsg['metadata'], options?: { ensureStreaming?: boolean }): string | null => {
@@ -426,12 +458,15 @@ export function useAgentMessages() {
         }
 
         if (msg.origin === 'callback') {
-          const invocationId = msg.invocationId ?? getCurrentInvocationIdForCat(msg.catId);
-          const hasExplicitInvocationId = !!msg.invocationId;
-          const replacementTarget = invocationId
-            ? (findCallbackReplacementTarget(msg.catId, invocationId) ??
-              (hasExplicitInvocationId ? null : findInvocationlessStreamPlaceholder(msg.catId)))
-            : findInvocationlessStreamPlaceholder(msg.catId);
+          const explicitInvocationId = msg.invocationId;
+          const inferredInvocationId = getCurrentInvocationIdForCat(msg.catId);
+          const invocationId = explicitInvocationId ?? inferredInvocationId;
+          const hasExplicitInvocationId = !!explicitInvocationId;
+          const exactReplacementTarget = invocationId ? findCallbackReplacementTarget(msg.catId, invocationId) : null;
+          const fallbackReplacementTarget = hasExplicitInvocationId
+            ? null
+            : findInvocationlessStreamPlaceholder(msg.catId, msg.content, msg.replyTo);
+          const replacementTarget = exactReplacementTarget ?? fallbackReplacementTarget;
 
           if (replacementTarget) {
             const finalId = msg.messageId ?? replacementTarget.id;
@@ -451,15 +486,29 @@ export function useAgentMessages() {
             activeRefs.current.delete(msg.catId);
             // Consume the finalized ref — callback successfully replaced the bubble
             finalizedStreamRef.current.delete(msg.catId);
-            if (invocationId) {
-              replacedInvocationsRef.current.set(msg.catId, invocationId);
+            if (explicitInvocationId) {
+              replacedInvocationsRef.current.set(msg.catId, explicitInvocationId);
+            } else if (replacementTarget?.matchKind === 'finalized_fallback') {
+              const suppressionInvocationId = replacementTarget.replacementInvocationId ?? inferredInvocationId;
+              if (suppressionInvocationId) {
+                // finalized_fallback prefers the replaced bubble's own invocationId.
+                // If that bubble finalized before invocation binding, fall back to
+                // the current late-bound invocationId so reordered stream chunks for
+                // the same invocation are still suppressed.
+                replacedInvocationsRef.current.set(msg.catId, suppressionInvocationId);
+              }
+            } else if (inferredInvocationId) {
+              // Exact or active invocationless placeholder matches still belong to
+              // the current invocation, so later stream chunks should be suppressed.
+              replacedInvocationsRef.current.set(msg.catId, inferredInvocationId);
             }
           } else {
             // Use backend messageId when available for rich_block correlation (#83 P2)
             const id = msg.messageId ?? `msg-${Date.now()}-${msg.catId}-cb-${++cbSeq}`;
             const extraForAdd = {
               ...(msg.extra?.crossPost ? { crossPost: msg.extra.crossPost } : {}),
-              ...(hasExplicitInvocationId && msg.invocationId ? { stream: { invocationId: msg.invocationId } } : {}),
+              ...(explicitInvocationId ? { stream: { invocationId: explicitInvocationId } } : {}),
+              ...(!explicitInvocationId ? { callbackBridge: { skipDedup: true } } : {}),
             };
             addMessage({
               id,
@@ -477,12 +526,19 @@ export function useAgentMessages() {
             // #586 Bug 1 (TD112): Callback created a new bubble because no stream
             // placeholder existed yet. Mark the invocation as replaced so that
             // late-arriving stream chunks for the same invocation are suppressed
-            // instead of spawning a second bubble.
-            const shouldLockReplacement =
-              invocationId &&
-              (!hasExplicitInvocationId || getCurrentInvocationIdForCat(msg.catId) === msg.invocationId);
-            if (shouldLockReplacement) {
-              replacedInvocationsRef.current.set(msg.catId, invocationId);
+            // instead of spawning a second bubble. Explicit callback invocationIds
+            // are only safe when the current invocation is already bound to the
+            // same ID; otherwise a stale callback can suppress live stream chunks
+            // before invocation_created/catInvocations catches up.
+            if (explicitInvocationId) {
+              if (inferredInvocationId && inferredInvocationId === explicitInvocationId) {
+                replacedInvocationsRef.current.set(msg.catId, explicitInvocationId);
+              }
+            } else if (inferredInvocationId) {
+              // Invocationless callback-only bubbles still need the inferred lock
+              // even when an older finalized ref was not matchable; otherwise the
+              // current invocation can still spawn a duplicate late stream bubble.
+              replacedInvocationsRef.current.set(msg.catId, inferredInvocationId);
             }
           }
         } else {
@@ -588,7 +644,11 @@ export function useAgentMessages() {
           // #586 follow-up: Record the finalized bubble so callback can find it
           // even after isStreaming=false + activeRefs cleared. Unlike a greedy
           // scan, this is scoped to the exact just-finalized message only.
-          finalizedStreamRef.current.set(msg.catId, messageId);
+          const finalizedMsg = useChatStore.getState().messages.find((m) => m.id === messageId);
+          finalizedStreamRef.current.set(msg.catId, {
+            messageId,
+            invocationId: finalizedMsg?.extra?.stream?.invocationId,
+          });
           activeRefs.current.delete(msg.catId);
         }
         // Bugfix: clear stale invocationId so findRecoverableAssistantMessage
@@ -685,11 +745,19 @@ export function useAgentMessages() {
             sysContent = mentions.map((m) => `${m.mentionedBy} @了 ${m.catId}`).join('、');
             sysVariant = 'a2a_followup';
           } else if (parsed?.type === 'invocation_created') {
-            // New invocation boundary: clear stale task snapshot + finalized ref for this cat.
-            // #586: Without clearing finalizedStreamRef here, a stale ref from the
-            // previous invocation could cause the next callback to overwrite the old message.
+            // #586 P3+P1: First invocation boundary fences finalizedStreamRef so
+            // immediate late callbacks from the previous invocation can still merge.
+            // If a second boundary arrives before the ref was consumed, the ref is
+            // stale across multiple invocations and must be invalidated.
             const targetCatId = parsed.catId ?? msg.catId;
-            finalizedStreamRef.current.delete(targetCatId);
+            const existingEntry = finalizedStreamRef.current.get(targetCatId);
+            if (existingEntry) {
+              if (existingEntry.fencedAt) {
+                finalizedStreamRef.current.delete(targetCatId);
+              } else {
+                existingEntry.fencedAt = Date.now();
+              }
+            }
             const invocationId = typeof parsed.invocationId === 'string' ? parsed.invocationId : undefined;
             if (targetCatId && invocationId) {
               setCatInvocation(targetCatId, {

packages/web/src/hooks/useSocket-background-system-info.ts

@@ -44,10 +44,18 @@ export function consumeBackgroundSystemInfo(
     if (parsed?.type === 'invocation_created') {
       const targetCatId = parsed.catId ?? msg.catId;
       const invocationId = typeof parsed.invocationId === 'string' ? parsed.invocationId : undefined;
-      // #586: Clear stale finalizedBgRef so previous invocation's finalized bubble
-      // can't be overwritten by the next invocation's callback.
+      // #586 P1: First boundary fences finalizedBgRefs so immediate late callbacks
+      // can still merge; a second boundary means the ref is stale and must be
+      // invalidated before another callback-only invocation reuses it.
       const bgStreamKey = `${msg.threadId}::${targetCatId}`;
-      options.finalizedBgRefs.delete(bgStreamKey);
+      const existingBgEntry = options.finalizedBgRefs.get(bgStreamKey);
+      if (existingBgEntry) {
+        if (existingBgEntry.fencedAt) {
+          options.finalizedBgRefs.delete(bgStreamKey);
+        } else {
+          existingBgEntry.fencedAt = Date.now();
+        }
+      }
       if (targetCatId && invocationId) {
         options.store.setThreadCatInvocation(msg.threadId, targetCatId, {
           invocationId,