From bbb2559160feda12741be080ed4585ad41e0ac41 Mon Sep 17 00:00:00 2001 From: qer Date: Thu, 20 Aug 2026 00:54:31 +0800 Subject: [PATCH] fix(transcript): attach mid-turn task notifications to the following step, cold and live --- .../src/services/transcript/coreEventMap.ts | 50 ++++++++---- .../test/services/transcript.test.ts | 76 +++++++++++++++++++ packages/transcript/src/history/groupTurns.ts | 41 ++++++---- packages/transcript/test/layers.test.ts | 48 +++++++++++- 4 files changed, 180 insertions(+), 35 deletions(-) diff --git a/packages/kap-server/src/services/transcript/coreEventMap.ts b/packages/kap-server/src/services/transcript/coreEventMap.ts index e6654822b..322e0e3e7 100644 --- a/packages/kap-server/src/services/transcript/coreEventMap.ts +++ b/packages/kap-server/src/services/transcript/coreEventMap.ts @@ -201,6 +201,7 @@ export class AgentTranscriptProjector { /** Latest header of the in-flight (or most recent) turn; kept whole so terminal upserts preserve `origin` / `startedAt` by reference. */ private currentTurn: TurnHeader | undefined; private currentStep: StepHeader | undefined; + private pendingTaskNotifications: { text: string; taskId: string | undefined }[] = []; /** turnId → highest step ordinal seen (engine-reported placement hint). */ private readonly stepOrdinals = new Map(); private frameOrdinal = 0; @@ -379,6 +380,7 @@ export class AgentTranscriptProjector { startedAt: nowIso(), }; this.currentStep = undefined; + this.pendingTaskNotifications = []; this.openText = undefined; this.openThinking = undefined; ops.push({ op: 'turn.upsert', turn: this.currentTurn }); @@ -420,6 +422,7 @@ export class AgentTranscriptProjector { }; ops.push({ op: 'turn.upsert', turn: this.currentTurn }); this.currentStep = undefined; + this.pendingTaskNotifications = []; if (event.reason === 'cancelled' && event.interruptReason === 'user_cancelled') { ops.push( this.markerOp('interruption', { turnId: event.turnId, reason: event.interruptReason }), @@ -473,7 +476,23 @@ export class AgentTranscriptProjector { this.frameOrdinal = 0; this.openText = undefined; this.openThinking = undefined; - return [{ op: 'step.upsert', turnId, step: this.currentStep }]; + const ops: TranscriptOperation[] = [{ op: 'step.upsert', turnId, step: this.currentStep }]; + for (const pending of this.pendingTaskNotifications) { + ops.push({ + op: 'frame.upsert', + turnId, + stepId, + frame: { + kind: 'text', + frameId: `${stepId}.f${++this.frameOrdinal}`, + role: 'user', + text: pending.text, + taskId: pending.taskId, + }, + }); + } + this.pendingTaskNotifications = []; + return ops; } private onStepCompleted(event: { @@ -865,20 +884,21 @@ export class AgentTranscriptProjector { }): TranscriptOperation[] { const step = this.currentStep; const turn = this.currentTurn; - const midTurn = - step !== undefined && - turn !== undefined && - step.state === 'running' && - turn.state === 'running'; - if (!midTurn) return []; - const frame: TextFrame = { - kind: 'text', - frameId: `${step.stepId}.f${++this.frameOrdinal}`, - role: 'user', - text: `${event.title}\n${event.body}`.trim(), - taskId: event.sourceId, - }; - return [{ op: 'frame.upsert', turnId: turn.turnId, stepId: step.stepId, frame }]; + if (turn === undefined || turn.state !== 'running') return []; + const text = `${event.title}\n${event.body}`.trim(); + if (step !== undefined && step.state === 'running') { + const frame: TextFrame = { + kind: 'text', + frameId: `${step.stepId}.f${++this.frameOrdinal}`, + role: 'user', + text, + taskId: event.sourceId, + }; + return [{ op: 'frame.upsert', turnId: turn.turnId, stepId: step.stepId, frame }]; + } + if (turn.origin?.kind === 'task') return []; + this.pendingTaskNotifications.push({ text, taskId: event.sourceId }); + return []; } private onTaskLifecycle(event: { diff --git a/packages/kap-server/test/services/transcript.test.ts b/packages/kap-server/test/services/transcript.test.ts index da74eadbf..230333dde 100644 --- a/packages/kap-server/test/services/transcript.test.ts +++ b/packages/kap-server/test/services/transcript.test.ts @@ -1464,6 +1464,82 @@ describe('AgentTranscriptProjector', () => { expect(frame?.kind === 'text' && frame.text).toContain('Background process completed'); }); + it('attaches a between-steps task notification to the following step', () => { + const projector = new AgentTranscriptProjector('main'); + const tx = new AgentTranscript('main'); + const feed = (event: ProjectorBusEvent): void => void tx.apply(projector.map(event)); + + const notified = (sourceId: string): ProjectorBusEvent => + ev({ + type: 'task.notified', + notificationType: 'task.completed', + title: 'Background agent completed', + body: 'inspect done.', + severity: 'info', + sourceKind: 'background_task', + sourceId, + }); + + feed(ev({ type: 'turn.started', turnId: 1, origin: { kind: 'user' } })); + feed(ev({ type: 'turn.step.started', turnId: 1, step: 1 })); + feed(ev({ type: 'turn.step.completed', turnId: 1, step: 1 })); + feed(notified('task_1')); + feed(notified('task_2')); + expect(turnOps('t1', tx.getItems()).steps[0]!.frames).toHaveLength(0); + + feed(ev({ type: 'turn.step.started', turnId: 1, step: 2 })); + const steps = turnOps('t1', tx.getItems()).steps; + expect(steps).toHaveLength(2); + expect(steps[1]!.frames.map((f) => f.kind === 'text' && f.taskId)).toEqual(['task_1', 'task_2']); + }); + + it('drops a task notification that is the turn prompt itself', () => { + const projector = new AgentTranscriptProjector('main'); + const tx = new AgentTranscript('main'); + const feed = (event: ProjectorBusEvent): void => void tx.apply(projector.map(event)); + + feed(ev({ type: 'turn.started', turnId: 1, origin: { kind: 'task', taskId: 'task_1' } })); + feed( + ev({ + type: 'task.notified', + notificationType: 'task.completed', + title: 'Background agent completed', + body: 'inspect done.', + severity: 'info', + sourceKind: 'background_task', + sourceId: 'task_1', + }), + ); + feed(ev({ type: 'turn.step.started', turnId: 1, step: 1 })); + expect(turnOps('t1', tx.getItems()).steps[0]!.frames).toHaveLength(0); + }); + + it('drops a buffered task notification when the turn ends before the next step', () => { + const projector = new AgentTranscriptProjector('main'); + const tx = new AgentTranscript('main'); + const feed = (event: ProjectorBusEvent): void => void tx.apply(projector.map(event)); + + feed(ev({ type: 'turn.started', turnId: 1, origin: { kind: 'user' } })); + feed(ev({ type: 'turn.step.started', turnId: 1, step: 1 })); + feed(ev({ type: 'turn.step.completed', turnId: 1, step: 1 })); + feed( + ev({ + type: 'task.notified', + notificationType: 'task.completed', + title: 'Background agent completed', + body: 'inspect done.', + severity: 'info', + sourceKind: 'background_task', + sourceId: 'task_1', + }), + ); + feed(ev({ type: 'turn.ended', turnId: 1, reason: 'completed' })); + + feed(ev({ type: 'turn.started', turnId: 2, origin: { kind: 'user' } })); + feed(ev({ type: 'turn.step.started', turnId: 2, step: 1 })); + expect(turnOps('t2', tx.getItems()).steps[0]!.frames).toHaveLength(0); + }); + it('replaces the global todo document on a confirmed TodoList write', () => { const projector = new AgentTranscriptProjector('main'); const tx = new AgentTranscript('main'); diff --git a/packages/transcript/src/history/groupTurns.ts b/packages/transcript/src/history/groupTurns.ts index 9e4e85960..e52dbf45b 100644 --- a/packages/transcript/src/history/groupTurns.ts +++ b/packages/transcript/src/history/groupTurns.ts @@ -72,6 +72,7 @@ export function groupMessagesIntoSnapshot( const items: TranscriptItem[] = []; const attachments: TranscriptAttachment[] = []; let turn: TurnDraft | undefined; + let pendingNotificationFrames: { text: string; taskId: string | undefined }[] = []; let nextOrdinal = 0; let markerCount = 0; @@ -143,6 +144,7 @@ export function groupMessagesIntoSnapshot( const startTurn = (origin: TurnOrigin, prompt?: string, attachmentIds?: string[]): TurnDraft => { const ordinal = nextOrdinal; nextOrdinal += 1; + pendingNotificationFrames = []; turn = { turnId: `t${ordinal}`, ordinal, origin, prompt, attachmentIds, steps: [] }; items.push(draftToTurnItem(turn)); return turn; @@ -192,16 +194,7 @@ export function groupMessagesIntoSnapshot( startTurn(mapOrigin(message), opening.text, opening.attachmentIds); continue; } - const step = turn?.steps.at(-1); - if (turn === undefined || step === undefined) continue; - step.frames.push({ - kind: 'text', - frameId: `${step.stepId}.f${step.frames.length + 1}`, - role: 'user', - text: notificationFrameText(textOf(message)), - taskId, - }); - syncTurnItem(items, turn); + pendingNotificationFrames.push({ text: notificationFrameText(textOf(message)), taskId }); continue; } const bundled = bundledSkillActivations(message); @@ -238,6 +231,16 @@ export function groupMessagesIntoSnapshot( frameCount += 1; return `${step.stepId}.f${frameCount}`; }; + for (const pending of pendingNotificationFrames) { + step.frames.push({ + kind: 'text', + frameId: nextFrameId(), + role: 'user', + text: pending.text, + taskId: pending.taskId, + }); + } + pendingNotificationFrames = []; for (const part of message.content ?? []) { if (part.type === 'text' && 'text' in part && typeof part.text === 'string' && part.text.length > 0) { step.frames.push({ kind: 'text', frameId: nextFrameId(), role: 'assistant', text: part.text }); @@ -285,18 +288,24 @@ function notificationFrameText(text: string): string { if (openingEnd === -1 || closingStart <= openingEnd) return text; const inner = text.slice(openingEnd + 1, closingStart); const lines = inner.split('\n'); + let headerEnd = 0; + while (headerEnd < lines.length && lines[headerEnd]!.trim() === '') headerEnd += 1; let title = ''; - let lastHeaderIndex = -1; - for (let i = 0; i < lines.length; i++) { + let bodyStart = headerEnd; + for (let i = headerEnd; i < lines.length; i++) { const line = lines[i]!; if (line.startsWith('Title: ')) { title = line.slice('Title: '.length); - lastHeaderIndex = i; - } else if (line.startsWith('Severity: ')) { - lastHeaderIndex = i; + bodyStart = i + 1; + continue; } + if (line.startsWith('Severity: ')) { + bodyStart = i + 1; + continue; + } + break; } - const bodyLines = lines.slice(lastHeaderIndex + 1); + const bodyLines = lines.slice(bodyStart); const childStart = bodyLines.findIndex((line) => { const trimmed = line.trimStart(); return trimmed.startsWith(' { .filter((f) => f.kind === 'text' && f.role === 'assistant') .map((f) => f.kind === 'text' && f.text); expect(assistantTexts).toEqual(['starting', 'continuing']); + expect(turn.steps).toHaveLength(2); + expect(turn.steps[1]?.frames.map((f) => f.kind === 'text' && f.role)).toEqual([ + 'user', + 'user', + 'assistant', + ]); }); it('stops folded notification text before child output blocks', () => { @@ -508,6 +514,7 @@ describe('groupMessagesIntoSnapshot (cold path)', () => { { role: 'user', content: [{ type: 'text', text: 'run' }], toolCalls: [], origin: { kind: 'user' } }, { role: 'assistant', content: [{ type: 'text', text: 'go' }], toolCalls: [] }, { role: 'user', content: [{ type: 'text', text: xml }], toolCalls: [], origin: { kind: 'task', taskId: 'task-9' } as { kind: string } }, + { role: 'assistant', content: [{ type: 'text', text: 'done' }], toolCalls: [] }, ], { taskOriginTurnTaskIds: new Set() }, ); @@ -517,7 +524,7 @@ describe('groupMessagesIntoSnapshot (cold path)', () => { expect(frame).toMatchObject({ text: 'Background agent completed\ninspect done.' }); }); - it('drops a folded notification that arrives before the turn has any step', () => { + it('buffers a folded notification that arrives before the first step into that step', () => { const xml = [ '', 'Title: Background agent completed', @@ -536,9 +543,14 @@ describe('groupMessagesIntoSnapshot (cold path)', () => { const turn = snapshot.items[0]; if (turn?.kind !== 'turn') throw new Error('expected turn'); expect(turn.steps).toHaveLength(1); - expect( - turn.steps.flatMap((step) => step.frames).filter((f) => f.kind === 'text' && f.role === 'user'), - ).toHaveLength(0); + expect(turn.steps[0]?.frames.map((f) => f.kind === 'text' && f.role)).toEqual([ + 'user', + 'assistant', + ]); + expect(turn.steps[0]?.frames[0]).toMatchObject({ + text: 'Background agent completed\nearly done.', + taskId: 'task-9', + }); }); it('drops a folded notification when no turn is open yet', () => { @@ -571,6 +583,7 @@ describe('groupMessagesIntoSnapshot (cold path)', () => { { role: 'user', content: [{ type: 'text', text: 'run' }], toolCalls: [], origin: { kind: 'user' } }, { role: 'assistant', content: [{ type: 'text', text: 'go' }], toolCalls: [] }, { role: 'user', content: [{ type: 'text', text: xml }], toolCalls: [], origin: { kind: 'task', taskId: 'task-9' } as { kind: string } }, + { role: 'assistant', content: [{ type: 'text', text: 'done' }], toolCalls: [] }, ], { taskOriginTurnTaskIds: new Set() }, ); @@ -582,6 +595,33 @@ describe('groupMessagesIntoSnapshot (cold path)', () => { }); }); + it('parses only the leading header lines, keeping later Title-like lines as body', () => { + const xml = [ + '', + 'Title: Background agent completed', + 'Severity: info', + 'first line.', + 'Title: details', + 'last line.', + '', + ].join('\n'); + const snapshot = groupMessagesIntoSnapshot( + [ + { role: 'user', content: [{ type: 'text', text: 'run' }], toolCalls: [], origin: { kind: 'user' } }, + { role: 'assistant', content: [{ type: 'text', text: 'go' }], toolCalls: [] }, + { role: 'user', content: [{ type: 'text', text: xml }], toolCalls: [], origin: { kind: 'task', taskId: 'task-9' } as { kind: string } }, + { role: 'assistant', content: [{ type: 'text', text: 'done' }], toolCalls: [] }, + ], + { taskOriginTurnTaskIds: new Set() }, + ); + const turn = snapshot.items[0]; + if (turn?.kind !== 'turn') throw new Error('expected turn'); + const frame = turn.steps.flatMap((step) => step.frames).find((f) => f.kind === 'text' && f.role === 'user'); + expect(frame).toMatchObject({ + text: 'Background agent completed\nfirst line.\nTitle: details\nlast line.', + }); + }); + it('expands a bundled prompt into per-skill markers and a caller-text turn', () => { const snapshot = groupMessagesIntoSnapshot([ {