diff --git a/src/main/claude/claude-structured-journal-translation.test.ts b/src/main/claude/claude-structured-journal-translation.test.ts index faf79a7b15d..65be68bb05f 100644 --- a/src/main/claude/claude-structured-journal-translation.test.ts +++ b/src/main/claude/claude-structured-journal-translation.test.ts @@ -220,10 +220,15 @@ describe('Claude structured journal translation', () => { for (const event of turn.start) { translator.handle(event) } + expect(lifecycleAppends(state.items)).toEqual([ + ['turn-lifecycle:msg_01-message-start', 'running'] + ]) + expect(assistantMessages(state.items)).toEqual([]) + for (const delta of turn.deltas) { translator.handle(delta) } - expect(state.items).toEqual([]) + expect(assistantMessages(state.items)).toEqual([]) const run = scheduled as (() => void) | null run?.() diff --git a/src/main/claude/claude-structured-journal-translation.ts b/src/main/claude/claude-structured-journal-translation.ts index c1dc33bf7f1..702ff5f6b26 100644 --- a/src/main/claude/claude-structured-journal-translation.ts +++ b/src/main/claude/claude-structured-journal-translation.ts @@ -25,7 +25,7 @@ import { journalClaudePrompt } from './claude-prompt-journaling' import type { ClaudePromptRegistry } from './claude-structured-prompt-replies' import { claudeProviderFrameActivity } from '../native-chat/agent-session-wire/provider-frame-activity' import { - appendUnmodeledClaudeContent, + appendUnmodeledContent, claudeProviderFrameKind, claudeResultFailure, createClaudeProviderFrameFallback, @@ -35,6 +35,7 @@ import { ClaudeSubagentRoster } from './claude-subagent-roster' import { createClaudeStreamedBlockRegistry } from './claude-streamed-block-identity' import { createClaudeStreamedTextCheckpoints } from './claude-streamed-text-checkpoints' import { + claudeStreamTurnStartSource, claudeStreamTurnSource, claudeTurnOpenedBySendEcho, createClaudeTurnOpener, @@ -150,16 +151,13 @@ export function createClaudeJournalTranslator( const handleStream = (message: Record, observedAt: number): boolean => { const delta = streamedBlocks.observe(message) + // `message_start` is the provider's turn boundary. Keep the first text + // delta as a compatibility fallback for streams that omit it. + const source = delta ? claudeStreamTurnSource(message) : claudeStreamTurnStartSource(message) + ensureTurnOpen(message, source, observedAt) if (!delta) { return false } - // Streamed text is the common first output of a resumed turn, and it is - // journaled here; a turn that opened only on the block's final frame would - // leave visible partial text reading idle. - const source = claudeStreamTurnSource(message) - if (source) { - ensureTurnOpen(message, source, observedAt) - } streamedText.append(delta.identity, delta.text) return true } @@ -190,6 +188,7 @@ export function createClaudeJournalTranslator( uuid: envelope.uuid, assistant: envelope.role === 'assistant' } + const openOutputTurn = (): void => ensureTurnOpen(message, source, observedAt) if (body) { // Opening before the append is what brackets a turn around its own first // output; a reader that scans back to the turn record and stops would @@ -234,7 +233,8 @@ export function createClaudeJournalTranslator( }) changed = true } - changed = appendUnmodeledClaudeContent(providerFallback, outputEnvelope, message) || changed + changed = + appendUnmodeledContent(providerFallback, outputEnvelope, message, openOutputTurn) || changed // The send's turn is anchored to the user row journaled just above it. const sendEchoTurn = claudeTurnOpenedBySendEcho({ envelope, diff --git a/src/main/claude/claude-structured-provider-fallback.ts b/src/main/claude/claude-structured-provider-fallback.ts index 68aec07976b..167ab37ce98 100644 --- a/src/main/claude/claude-structured-provider-fallback.ts +++ b/src/main/claude/claude-structured-provider-fallback.ts @@ -106,16 +106,22 @@ export function createClaudeProviderFrameFallback( acquisitionId: string ): { /** `displayText` leads the row when Claude knows the sentence the frame itself does not name. */ - append: (kind: string, payload: unknown, displayText?: string | null) => void + append: ( + kind: string, + payload: unknown, + displayText?: string | null, + beforeAppend?: () => void + ) => boolean } { let sequence = 0 return { - append: (kind, payload, displayText) => { + append: (kind, payload, displayText, beforeAppend) => { sequence += 1 const translated = unhandledProviderFrameJournalItem('claude', kind, payload) if (!translated) { - return + return false } + beforeAppend?.() const bounded = displayText ? boundInlineText(displayText, DEFAULT_JOURNAL_PAYLOAD_LIMITS).text : null @@ -127,6 +133,7 @@ export function createClaudeProviderFrameFallback( bounded ? { ...translated.body, text: bounded } : translated.body ) sink.publish() + return true } } } @@ -136,24 +143,27 @@ export type ClaudeProviderFrameFallback = ReturnType + message: Record, + beforeAppend: () => void ): boolean { let changed = false for (const part of envelope.content.filter((part) => !isModeledClaudeContent(part))) { const partType = claudeText(claudeRecord(part)?.type) ?? 'unknown' - fallback.append( - `message:${envelope.role}:content:${partType}`, - part, - readableProviderFrameText(part) ?? CLAUDE_UNRENDERABLE_CONTENT_TEXT - ) - changed = true + changed = + fallback.append( + `message:${envelope.role}:content:${partType}`, + part, + readableProviderFrameText(part) ?? CLAUDE_UNRENDERABLE_CONTENT_TEXT, + beforeAppend + ) || changed } if (envelope.content.length === 0 && envelope.role === 'assistant') { - fallback.append(`message:${envelope.role}:empty`, message) - changed = true + // Empty provider placeholders do not prove work began, and may have no + // later result capable of closing a turn. + changed = fallback.append(`message:${envelope.role}:empty`, message) || changed } return changed } diff --git a/src/main/claude/claude-turn-lifecycle-item.ts b/src/main/claude/claude-turn-lifecycle-item.ts index ac679a5b47f..664a06f0898 100644 --- a/src/main/claude/claude-turn-lifecycle-item.ts +++ b/src/main/claude/claude-turn-lifecycle-item.ts @@ -2,6 +2,7 @@ import type { AgentJournalItemIdentity, AgentJournalTurnItem } from '../../shared/agent-session-journal-types' +import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' import { agentJournalTurnBody } from '../../shared/agent-session-turn-record' import type { StructuredAgentSessionAppendOptions } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' import { claudeText } from './claude-structured-item-translation' @@ -10,9 +11,9 @@ export type ClaudeCurrentTurn = { sessionId: string turnId: string startedAt: number - /** Provider key of the user echo that opened the turn. Absent when the - * provider resumed the work itself and there is no user row to anchor to. */ - userItemId?: string + /** Provider key of the user echo, or the lifecycle row itself when provider + * output opened a turn with no user row to receive its timing. */ + userItemId: string } export type ClaudeTurnEnd = { @@ -51,6 +52,12 @@ export function claudeTurnLifecycleIdentity( } } +/** Keep provider-resumed timing off the preceding prompt on clients that treat + * a missing user key as an older-host lifecycle row. */ +export function claudeProviderResumedTurnTimingAnchor(sessionId: string, turnId: string): string { + return agentJournalItemKey(claudeTurnLifecycleIdentity(sessionId, turnId)) +} + /** The lifecycle row is revised to its terminal state, never tombstoned, so the * turn's host-clock endpoints outlive the turn. */ export function claudeTurnLifecycleItem( @@ -72,15 +79,10 @@ export function claudeTurnLifecycleItem( state: end.state, startedAt, completedAt: end.completedAt, - ...(userItemId === undefined ? {} : { userItemId }), + userItemId, ...(end.durationMs === undefined ? {} : { durationMs: end.durationMs }) } - : { - turnId, - state: 'running', - startedAt, - ...(userItemId === undefined ? {} : { userItemId }) - } + : { turnId, state: 'running', startedAt, userItemId } ), // The running row's ts is the turn start itself, so clients read no append lag. options: end ? {} : { observedAt: startedAt }, diff --git a/src/main/claude/claude-turn-opening.ts b/src/main/claude/claude-turn-opening.ts index 4cc56aa849e..9b4be37eb8e 100644 --- a/src/main/claude/claude-turn-opening.ts +++ b/src/main/claude/claude-turn-opening.ts @@ -9,10 +9,14 @@ import { claudeHasReplayContent, + claudeRecord, claudeText, type ClaudeMessageEnvelope } from './claude-structured-item-translation' -import type { ClaudeCurrentTurn } from './claude-turn-lifecycle-item' +import { + claudeProviderResumedTurnTimingAnchor, + type ClaudeCurrentTurn +} from './claude-turn-lifecycle-item' export type ClaudeSendEchoTurnInput = { envelope: ClaudeMessageEnvelope @@ -62,6 +66,16 @@ export function claudeStreamTurnSource(frame: Record): ClaudeTu return sessionId && uuid ? { sessionId, uuid, assistant: true } : null } +/** A streamed assistant message has begun, before its first content delta. */ +export function claudeStreamTurnStartSource( + frame: Record +): ClaudeTurnSource | null { + const event = claudeRecord(frame.event) + return frame.type === 'stream_event' && event?.type === 'message_start' + ? claudeStreamTurnSource(frame) + : null +} + /** The provider produced, so a turn is running. Root-ness first, then the * suppression latch, then idempotency — every frame of one reply stays inside * the turn its first frame opened. */ @@ -69,16 +83,21 @@ export function createClaudeTurnOpener(deps: { isTurnOpen: () => boolean isSuppressed: () => boolean open: (turn: ClaudeCurrentTurn, observedAt: number) => void -}): (frame: Record, source: ClaudeTurnSource, observedAt: number) => void { +}): (frame: Record, source: ClaudeTurnSource | null, observedAt: number) => void { return (frame, source, observedAt) => { - if (!source.assistant || !isRootClaudeFrame(frame)) { + if (!source?.assistant || !isRootClaudeFrame(frame)) { return } if (deps.isSuppressed() || deps.isTurnOpen()) { return } deps.open( - { sessionId: source.sessionId, turnId: source.uuid, startedAt: observedAt }, + { + sessionId: source.sessionId, + turnId: source.uuid, + startedAt: observedAt, + userItemId: claudeProviderResumedTurnTimingAnchor(source.sessionId, source.uuid) + }, observedAt ) } diff --git a/src/main/claude/claude-turn-resumption.test.ts b/src/main/claude/claude-turn-resumption.test.ts index cdcc5a3ba26..96d364f91a5 100644 --- a/src/main/claude/claude-turn-resumption.test.ts +++ b/src/main/claude/claude-turn-resumption.test.ts @@ -12,7 +12,11 @@ import type { AgentJournalRenderItem } from '../../shared/agent-session-journal-types' import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' -import { readAgentJournalTurn } from '../../shared/agent-session-turn-record' +import { + legacyAgentJournalTurnStatusBody, + readAgentJournalTurn +} from '../../shared/agent-session-turn-record' +import { selectStructuredAgentSettledTurns } from '../../shared/structured-agent-session-turn-timing' import { hasUnansweredStructuredAgentSessionDispatch, projectStructuredAgentSessionStatus, @@ -108,7 +112,21 @@ function textDelta(uuid: string, messageId: string, text: string) { } } -function result(uuid: string, parentToolUseId: string | null = null) { +function streamMessageStart(uuid: string, parentToolUseId: string | null = null) { + return { + type: 'message' as const, + sessionId: 'orca-session', + message: { + type: 'stream_event', + uuid, + session_id: SESSION, + parent_tool_use_id: parentToolUseId, + event: { type: 'message_start', message: { id: `msg-${uuid}`, role: 'assistant' } } + } + } +} + +function result(uuid: string, parentToolUseId: string | null = null, durationMs = 322_937) { return { type: 'message' as const, sessionId: 'orca-session', @@ -118,7 +136,7 @@ function result(uuid: string, parentToolUseId: string | null = null) { uuid, session_id: SESSION, parent_tool_use_id: parentToolUseId, - duration_ms: 322_937 + duration_ms: durationMs } } } @@ -158,7 +176,7 @@ describe('a Claude turn the provider resumed on its own', () => { expect(projected(items())).toBe('idle') }) - it('gives the resumed turn its own record, with no user row to anchor to', () => { + it('gives the resumed turn its own record, anchored away from the preceding user row', () => { const { translator, items } = harness() translator.handle(frame('user', 'u1', [{ type: 'text', text: 'go' }])) translator.handle(result('r1')) @@ -170,7 +188,45 @@ describe('a Claude turn the provider resumed on its own', () => { }) expect(turns.map((turn) => turn.state)).toEqual(['completed', 'running']) expect(turns[1]?.turnId).toBe('a1') - expect(turns[1]?.userItemId).toBeUndefined() + expect(turns[1]?.userItemId).toBe('legacy:claude:claude-session:turn-lifecycle%3Aa1') + }) + + it('does not replace the preceding prompt timing with provider-resumed work', () => { + const { translator, items } = harness() + translator.handle(frame('user', 'u1', [{ type: 'text', text: 'go' }])) + translator.handle(result('r1', null, 1_000)) + translator.handle(frame('assistant', 'a1', [{ type: 'text', text: 'Back on it.' }])) + translator.handle(result('r2', null, 9_000)) + + const translatedItems = items() + const originalTurn = translatedItems + .map((item) => readAgentJournalTurn(item.body)) + .find((turn) => turn?.turnId === 'u1') + expect(originalTurn?.userItemId).toBeDefined() + if (!originalTurn?.userItemId) { + throw new Error('expected the original turn to name its user row') + } + const userItem: AgentJournalRenderItem = { + itemId: originalTurn.userItemId, + revision: 1, + sequence: -1, + observedAt: 0, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'go' }] } + } + const currentItems = [userItem, ...translatedItems] + expect(selectStructuredAgentSettledTurns(currentItems).get(userItem.itemId)).toMatchObject({ + workedSeconds: 1 + }) + + const legacyItems = currentItems.map((item) => { + const turn = readAgentJournalTurn(item.body) + return item.body.kind === 'turn' && turn + ? { ...item, body: legacyAgentJournalTurnStatusBody(turn, item.itemId) } + : item + }) + expect(selectStructuredAgentSettledTurns(legacyItems).get(userItem.itemId)).toMatchObject({ + workedSeconds: 1 + }) }) it('leaves a settled turn settled when only a subagent is still producing', () => { @@ -178,10 +234,8 @@ describe('a Claude turn the provider resumed on its own', () => { translator.handle(frame('user', 'u1', [{ type: 'text', text: 'go' }])) translator.handle(result('r1')) - // Children outlive the turn that spawned them; their frames are not a turn. - translator.handle( - frame('assistant', 'a1', [{ type: 'text', text: 'child work' }], 'toolu_parent') - ) + // Children outlive the turn that spawned them; their streams are not a turn. + translator.handle(streamMessageStart('child-start', 'toolu_parent')) expect(projected(items())).toBe('idle') }) @@ -241,7 +295,7 @@ describe('a Claude turn the provider resumed on its own', () => { expect(projected(items())).toBe('idle') // Nothing can close a turn opened now, so nothing may open one. - translator.handle(frame('assistant', 'a1', [{ type: 'text', text: 'late frame' }])) + translator.handle(streamMessageStart('late-start')) expect(projected(items())).toBe('idle') }) @@ -284,6 +338,50 @@ describe('a Claude turn the provider resumed on its own', () => { expect(projected(items())).toBe('working') }) + it('opens before a resumed stream produces its first content delta', () => { + const { translator, items } = harness() + translator.handle(frame('user', 'u1', [{ type: 'text', text: 'go' }])) + translator.handle(result('r1')) + + translator.handle(streamMessageStart('message-start-1')) + + expect(projected(items())).toBe('working') + expect(readAgentJournalTurn(items().at(-1)?.body)?.turnId).toBe('message-start-1') + expect(items().some((item) => item.body.kind === 'status')).toBe(false) + }) + + it('opens before journaling substantive fallback output', () => { + const { translator, items } = harness() + translator.handle(frame('user', 'u1', [{ type: 'text', text: 'go' }])) + translator.handle(result('r1')) + + translator.handle( + frame('assistant', 'a1', [{ type: 'future_content', message: 'new provider output' }]) + ) + + const resumed = items().slice(-2) + expect(readAgentJournalTurn(resumed[0]?.body)?.state).toBe('running') + expect(resumed[1]?.body).toMatchObject({ + kind: 'status', + providerFrame: { kind: 'message:assistant:content:future_content' } + }) + expect(projected(items())).toBe('working') + }) + + it('does not open a turn for an empty assistant placeholder', () => { + const { translator, items } = harness() + translator.handle(frame('user', 'u1', [{ type: 'text', text: 'go' }])) + translator.handle(result('r1')) + + translator.handle(frame('assistant', 'empty-1', [])) + + expect(projected(items())).toBe('idle') + expect(items().at(-1)?.body).toMatchObject({ + kind: 'status', + providerFrame: { kind: 'message:assistant:empty' } + }) + }) + it('still reports a nested result failure even though it settles no turn', () => { const { translator, items, appended } = harness() translator.handle(frame('user', 'u1', [{ type: 'text', text: 'go' }])) diff --git a/src/shared/agent-session-journal-types.ts b/src/shared/agent-session-journal-types.ts index 7ba5b277e6d..fc603fdca08 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -170,7 +170,8 @@ export type AgentJournalTurnLifecycle = { turnId: string state: AgentJournalTurnLifecycleState /** Provider key of the user item that opened the turn; clients resolve a - * submission alias through it. Absent on rows from older hosts. */ + * submission alias through it. A lifecycle row may key itself when provider + * output opened a turn with no user item; absent means an older host. */ userItemId?: string startedAt?: number completedAt?: number