From d5583dfd7f99027b5e85ef1db2bc6fb772dd0dbb Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Wed, 22 Jul 2026 11:06:04 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=94=81=20fix:=20Codex=20Round=204=20?= =?UTF-8?q?=E2=80=94=20Resume=20Gap,=20Delta=20Flush,=20Agent-Scoped=20Int?= =?UTF-8?q?ent,=20Token=20Counter?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 | 5 ++ .../activityLabels/__tests__/wiring.spec.ts | 46 ++++++++++++++++ .../api/src/agents/activityLabels/index.ts | 1 + .../api/src/agents/activityLabels/wiring.ts | 54 +++++++++++++++++++ packages/api/src/agents/client.ts | 7 ++- .../api/src/stream/GenerationJobManager.ts | 27 ++++++++++ .../RedisJobStore.stream_integration.spec.ts | 10 ++-- 7 files changed, 146 insertions(+), 4 deletions(-) diff --git a/client/src/hooks/SSE/useResumableSSE.ts b/client/src/hooks/SSE/useResumableSSE.ts index d25d824983..38081785b6 100644 --- a/client/src/hooks/SSE/useResumableSSE.ts +++ b/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) { diff --git a/packages/api/src/agents/activityLabels/__tests__/wiring.spec.ts b/packages/api/src/agents/activityLabels/__tests__/wiring.spec.ts index 0cd41a038e..d1d9f6c6f3 100644 --- a/packages/api/src/agents/activityLabels/__tests__/wiring.spec.ts +++ b/packages/api/src/agents/activityLabels/__tests__/wiring.spec.ts @@ -5,6 +5,7 @@ import { captureActivityBlockContext, createActivityLabelWiring, stripActivityLabelParts, + synthesizeActivityLabelGapEvents, } from '../wiring'; async function flushDetached(): Promise { @@ -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([]); + }); +}); diff --git a/packages/api/src/agents/activityLabels/index.ts b/packages/api/src/agents/activityLabels/index.ts index 3b0d0db640..8ca609819c 100644 --- a/packages/api/src/agents/activityLabels/index.ts +++ b/packages/api/src/agents/activityLabels/index.ts @@ -13,5 +13,6 @@ export { captureActivityBlockContext, createActivityLabelWiring, stripActivityLabelParts, + synthesizeActivityLabelGapEvents, } from './wiring'; export type { ActivityLabelHostDeps, LooseContentPart } from './wiring'; diff --git a/packages/api/src/agents/activityLabels/wiring.ts b/packages/api/src/agents/activityLabels/wiring.ts index 658b8b4561..26de7e7db7 100644 --- a/packages/api/src/agents/activityLabels/wiring.ts +++ b/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(payload return changed ? result : payload; } +/** Minimal SSE shape for synthesized gap events. */ +interface ActivityLabelGapEvent { + event: string; + data: Record; +} + +/** + * 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, + freshContent: ReadonlyArray, + 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; diff --git a/packages/api/src/agents/client.ts b/packages/api/src/agents/client.ts index e3cb6ff280..df846bf8eb 100644 --- a/packages/api/src/agents/client.ts +++ b/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; } diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index 53d0c1d20e..8f61731ba7 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/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[1], + { conversationId: streamId, responseMessageId: resumeState.responseMessageId }, + ); + if (labelGapEvents.length > 0) { + pendingEvents = [...pendingEvents, ...(labelGapEvents as t.ServerSentEvent[])]; + } + } + } + return { subscription, resumeState, pendingEvents }; } diff --git a/packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts b/packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts index 19714e4ba6..b31f91d922 100644 --- a/packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts +++ b/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(); });