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(); });