🔁 fix: Codex Round 4 — Resume Gap, Delta Flush, Agent-Scoped Intent, Token Counter

- Synthesize on_activity_label events for labels claimed or filled in the
  snapshot→subscribe window (the publish is fire-and-forget, so Redis-mode
  reconnects missed them). Feature-gated so the default path adds no
  content re-read; the client applier already ignores duplicates.
- Flush queued deltas before applying a label part, matching the pending-
  action and steer appliers — without it the handler read a stale message
  cache and syncStepMessage pushed a pre-delta copy back.
- Skip another agent's tail text when resolving intent, so parallel runs
  cannot seed a label prompt with a sibling agent's narration.
- Exclude activity_label parts from countFormattedMessageTokens (the
  agent-path counter), not just the legacy BaseClient one.

client/src/hooks/SSE/useResumableSSE.ts

@@ -794,6 +794,11 @@ export default function useResumableSSE(
             );
           }
         };
+        /** Same boundary as pending actions and steers: land queued deltas
+         * before the label part is placed and synced, or the later flush
+         * would clobber it (and `syncStepMessage` would sync a pre-delta
+         * copy back into the step handler's authoritative map). */
+        flushPendingDeltas();
         const messages = getMessages() ?? [];
         const index = findActivityLabelMessageIndex(messages, event);
         if (index < 0) {

packages/api/src/agents/activityLabels/__tests__/wiring.spec.ts

@@ -5,6 +5,7 @@ import {
   captureActivityBlockContext,
   createActivityLabelWiring,
   stripActivityLabelParts,
+  synthesizeActivityLabelGapEvents,
 } from '../wiring';
 
 async function flushDetached(): Promise<void> {
@@ -95,3 +96,48 @@ describe('stripActivityLabelParts', () => {
     expect(stripActivityLabelParts(clean)).toBe(clean);
   });
 });
+
+describe('captureActivityBlockContext intent filtering', () => {
+  it("skips another agent's tail text when resolving intent", () => {
+    const parts: LooseContentPart[] = [
+      { type: 'text', text: 'Agent A plan for this batch', agentId: 'agent-a' },
+      { type: 'text', text: 'Agent B unrelated narration', agentId: 'agent-b' },
+    ];
+    const context = captureActivityBlockContext(parts, 'agent-a');
+    expect(context.lastAssistantText).toBe('Agent A plan for this batch');
+  });
+});
+
+describe('synthesizeActivityLabelGapEvents', () => {
+  const meta = { conversationId: 'c1', responseMessageId: 'm1' };
+
+  it('re-emits a label filled during the snapshot gap', () => {
+    const snapshot: LooseContentPart[] = [
+      { type: 'tool_call', tool_call: { id: 't1' } },
+      { type: 'activity_label', activity_label: '', pending: true },
+    ];
+    const fresh: LooseContentPart[] = [
+      { type: 'tool_call', tool_call: { id: 't1' } },
+      { type: 'activity_label', activity_label: 'Searched release notes', pending: false },
+    ];
+    const events = synthesizeActivityLabelGapEvents(snapshot, fresh, meta);
+    expect(events).toHaveLength(1);
+    expect(events[0].event).toBe('on_activity_label');
+    expect(events[0].data).toMatchObject({ index: 1, conversationId: 'c1' });
+  });
+
+  it('re-emits a label claimed entirely within the gap', () => {
+    const fresh: LooseContentPart[] = [
+      { type: 'tool_call', tool_call: { id: 't1' } },
+      { type: 'activity_label', activity_label: '', pending: true },
+    ];
+    expect(synthesizeActivityLabelGapEvents([fresh[0]], fresh, meta)).toHaveLength(1);
+  });
+
+  it('emits nothing when the snapshot already matches', () => {
+    const parts: LooseContentPart[] = [
+      { type: 'activity_label', activity_label: 'Same label', pending: false },
+    ];
+    expect(synthesizeActivityLabelGapEvents(parts, parts, meta)).toEqual([]);
+  });
+});

packages/api/src/agents/activityLabels/index.ts

@@ -13,5 +13,6 @@ export {
   captureActivityBlockContext,
   createActivityLabelWiring,
   stripActivityLabelParts,
+  synthesizeActivityLabelGapEvents,
 } from './wiring';
 export type { ActivityLabelHostDeps, LooseContentPart } from './wiring';

packages/api/src/agents/activityLabels/wiring.ts

@@ -59,6 +59,12 @@ export function captureActivityBlockContext(
       continue;
     }
     if (part.type === ContentTypes.TEXT) {
+      /** Parallel/added-agent runs interleave text parts from several
+       *  agents; another agent's text at the tail is not this batch's
+       *  intent, so skip it rather than stopping the scan there. */
+      if (executingAgentId != null && part.agentId != null && part.agentId !== executingAgentId) {
+        continue;
+      }
       const text = textValue(part.text).trim();
       if (text.length > 0) {
         lastAssistantText = text.slice(-INTENT_CHARS);
@@ -110,6 +116,54 @@ export function stripActivityLabelParts<T extends { content?: unknown }>(payload
   return changed ? result : payload;
 }
 
+/** Minimal SSE shape for synthesized gap events. */
+interface ActivityLabelGapEvent {
+  event: string;
+  data: Record<string, unknown>;
+}
+
+/**
+ * Synthesizes `on_activity_label` events for labels that appeared OR were
+ * filled between a resume snapshot and subscriber attach. In Redis mode the
+ * label publish is fire-and-forget and the sync payload carries only the
+ * snapshot, so a label claimed or resolved in that window would otherwise
+ * never reach the reconnecting client. Compares by index: a fresh label part
+ * whose text or pending state differs from the snapshot's (or that has no
+ * snapshot counterpart) is re-emitted. Idempotent - the client applier
+ * ignores duplicates and refuses stale pending placeholders.
+ */
+export function synthesizeActivityLabelGapEvents(
+  snapshotContent: ReadonlyArray<LooseContentPart | null | undefined>,
+  freshContent: ReadonlyArray<LooseContentPart | null | undefined>,
+  meta: { conversationId: string; responseMessageId?: string },
+): ActivityLabelGapEvent[] {
+  const events: ActivityLabelGapEvent[] = [];
+  for (let i = 0; i < freshContent.length; i++) {
+    const part = freshContent[i];
+    if (part?.type !== ContentTypes.ACTIVITY_LABEL) {
+      continue;
+    }
+    const snapshot = snapshotContent[i];
+    const isSameLabel =
+      snapshot?.type === ContentTypes.ACTIVITY_LABEL &&
+      snapshot[ContentTypes.ACTIVITY_LABEL] === part[ContentTypes.ACTIVITY_LABEL] &&
+      snapshot.pending === part.pending;
+    if (isSameLabel) {
+      continue;
+    }
+    events.push({
+      event: 'on_activity_label',
+      data: {
+        index: i,
+        part,
+        conversationId: meta.conversationId,
+        ...(meta.responseMessageId != null && { responseMessageId: meta.responseMessageId }),
+      },
+    });
+  }
+  return events;
+}
+
 /** Host closures the wiring needs; each is a thin bridge into the caller. */
 export interface ActivityLabelHostDeps {
   abortSignal?: AbortSignal;

packages/api/src/agents/client.ts

@@ -285,7 +285,12 @@ export function countFormattedMessageTokens(
           continue;
         }
 
-        if (type === ContentTypes.THINK || type === ContentTypes.ERROR) {
+        if (
+          type === ContentTypes.THINK ||
+          type === ContentTypes.ERROR ||
+          // UI-only progress headers — never model input, never billed output
+          type === ContentTypes.ACTIVITY_LABEL
+        ) {
           continue;
         }
 

packages/api/src/stream/GenerationJobManager.ts

@@ -37,6 +37,10 @@ import {
   toPendingSteer,
   synthesizeAppliedSteerEvents,
 } from './SteeringLifecycle';
+import {
+  isActivityLabelPocEnabled,
+  synthesizeActivityLabelGapEvents,
+} from '~/agents/activityLabels';
 import { isPendingActionStale, isPendingActionExpired } from './interfaces/IJobStore';
 import { InMemoryEventTransport } from './implementations/InMemoryEventTransport';
 import { InMemoryJobStore } from './implementations/InMemoryJobStore';
@@ -1256,6 +1260,29 @@ class GenerationJobManagerClass {
       }
     }
 
+    // Same snapshot->subscribe race for activity labels: the label publish
+    // is fire-and-forget, so a slot claimed (or filled) in the window is in
+    // neither the snapshot nor the chunk replay the client already applied.
+    // Gated on the feature so the default path adds no content re-read.
+    // Compare the snapshot content view against a fresh read and re-emit any
+    // label whose text/pending state moved; the client applier is idempotent
+    // and refuses stale pending placeholders.
+    if (resumeState != null && jobActive && isActivityLabelPocEnabled()) {
+      const labelContent = await this.jobStore.getContentParts(streamId);
+      if (labelContent?.content != null) {
+        const labelGapEvents = synthesizeActivityLabelGapEvents(
+          (resumeState.aggregatedContent ?? []) as Parameters<
+            typeof synthesizeActivityLabelGapEvents
+          >[0],
+          labelContent.content as Parameters<typeof synthesizeActivityLabelGapEvents>[1],
+          { conversationId: streamId, responseMessageId: resumeState.responseMessageId },
+        );
+        if (labelGapEvents.length > 0) {
+          pendingEvents = [...pendingEvents, ...(labelGapEvents as t.ServerSentEvent[])];
+        }
+      }
+    }
+
     return { subscription, resumeState, pendingEvents };
   }
 

packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts

@@ -2348,12 +2348,16 @@ describe('RedisJobStore Integration Tests', () => {
         activity_label?: string;
         pending?: boolean;
       }>;
-      expect(parts[1]).toMatchObject({
-        type: 'activity_label',
+      /** Position-independent: the placeholder and the filled event share a
+       *  slot, so exactly ONE label part must survive and it must carry the
+       *  resolved text (last write wins). */
+      const labels = parts.filter((part) => part?.type === 'activity_label');
+      expect(labels).toHaveLength(1);
+      expect(labels[0]).toMatchObject({
         activity_label: 'Searched runtime release notes',
         pending: false,
       });
-      expect(parts[2]?.type).toBe('text');
+      expect(parts.some((part) => part?.type === 'text')).toBe(true);
 
       await store.destroy();
     });