Replace assistant-prose heuristic with explicit turn lifecycle markers

Extract provider-authored turn boundaries (completion, interruption) directly
from Claude/Codex transcripts so the chat view knows when work ends without
guessing from message presence. Reconciles live hook state with transcript
lifecycle: when a terminal boundary lands, it settles a dropped Stop hook
instead of letting prose mislead the UI into showing 'working' after done.
This commit is contained in:
Jinjing
2026-07-16 20:51:20 -07:00
parent cc1ad064d7
commit c717bcaec8
32 changed files with 1551 additions and 206 deletions
@@ -1,4 +1,5 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
import type { NativeChatTurnLifecycle } from '../../shared/native-chat-types'
const { handlers, listeners, subscribeTranscript } = vi.hoisted(() => ({
handlers: new Map<string, (_event: unknown, args?: unknown) => unknown>(),
@@ -111,7 +112,8 @@ type InitialSnapshotCallback = (
messages: unknown[],
hasMore: boolean,
beforeOffset: number,
error?: string
error?: string,
lifecycle?: NativeChatTurnLifecycle
) => void
// The onInitialSnapshot callback the handler passed into the Nth subscribeTranscript
@@ -260,6 +262,7 @@ describe('nativeChat subscribe lifecycle', () => {
type: 'snapshot',
messages: [],
hasMore: false,
turnLifecycleCapable: true,
error: 'Transcript unavailable'
}
})
@@ -275,7 +278,33 @@ describe('nativeChat subscribe lifecycle', () => {
expect(renderer.sender.send).toHaveBeenCalledWith('nativeChat:appended', {
subscriptionId: 'drain-clean',
frame: { type: 'snapshot', messages: [], hasMore: false }
frame: {
type: 'snapshot',
messages: [],
hasMore: false,
turnLifecycleCapable: true
}
})
})
it('forwards replayable lifecycle on the initial snapshot', () => {
const pending = deferredSubscription()
subscribeTranscript.mockReturnValueOnce(pending.promise)
const renderer = createSender(7)
const lifecycle = { state: 'completed', turnId: 'turn-1', timestamp: 42 } as const
subscribe(renderer.sender, 'lifecycle')
initialSnapshot(0)([], false, 0, undefined, lifecycle)
expect(renderer.sender.send).toHaveBeenCalledWith('nativeChat:appended', {
subscriptionId: 'lifecycle',
frame: {
type: 'snapshot',
messages: [],
hasMore: false,
turnLifecycleCapable: true,
lifecycle
}
})
})
})
+60 -11
View File
@@ -1,5 +1,9 @@
import { ipcMain, type IpcMainEvent, type WebContents } from 'electron'
import type { AgentType, NativeChatMessage } from '../../shared/native-chat-types'
import type {
AgentType,
NativeChatMessage,
NativeChatTurnLifecycle
} from '../../shared/native-chat-types'
import { clearNativeChatTranscriptCache } from '../native-chat/transcript-read-cache'
import type { ReadTranscriptResult } from '../native-chat/transcript-reader'
import {
@@ -7,6 +11,7 @@ import {
readNativeChatTranscriptTail,
type NativeChatTranscriptSubscription
} from '../native-chat/transcript-watch'
import { supportsNativeChatTranscriptTurnLifecycle } from '../native-chat/transcript-turn-lifecycle'
// Re-export so existing test imports of `clearNativeChatTranscriptCache` from
// this module keep working after the cache moved to transcript-read-cache.ts.
@@ -31,12 +36,18 @@ async function readSession(args: NativeChatReadSessionArgs): Promise<ReadTranscr
const { agent, sessionId } = args
// Clamp to a positive window; default to the desktop window for the first page.
const limit = args.limit && args.limit > 0 ? Math.floor(args.limit) : DESKTOP_READ_WINDOW
return readNativeChatTranscriptTail({
const result = await readNativeChatTranscriptTail({
agent,
sessionId,
transcriptPath: args.transcriptPath,
limit
})
return 'messages' in result
? {
...result,
turnLifecycleCapable: supportsNativeChatTranscriptTurnLifecycle(agent)
}
: result
}
export type NativeChatSubscribeArgs = {
@@ -53,9 +64,27 @@ export type NativeChatSubscribeArgs = {
export type NativeChatAppendedPayload = {
subscriptionId: string
frame:
| { type: 'snapshot'; messages: NativeChatMessage[]; hasMore: boolean; error?: string }
| { type: 'replacement'; messages: NativeChatMessage[]; hasMore: boolean }
| { type: 'appended'; messages: NativeChatMessage[] }
| {
type: 'snapshot'
messages: NativeChatMessage[]
hasMore: boolean
error?: string
lifecycle?: NativeChatTurnLifecycle
turnLifecycleCapable: boolean
}
| {
type: 'replacement'
messages: NativeChatMessage[]
hasMore: boolean
lifecycle?: NativeChatTurnLifecycle
turnLifecycleCapable: boolean
}
| {
type: 'appended'
messages: NativeChatMessage[]
lifecycle?: NativeChatTurnLifecycle
turnLifecycleCapable: boolean
}
}
type LiveSubscription = {
@@ -139,6 +168,7 @@ async function handleSubscribe(event: IpcMainEvent, args: NativeChatSubscribeArg
return
}
const { subscriptionId, agent, sessionId, transcriptPath } = args
const turnLifecycleCapable = supportsNativeChatTranscriptTurnLifecycle(agent)
const limit = args.limit && args.limit > 0 ? Math.floor(args.limit) : DESKTOP_READ_WINDOW
// Replace any prior subscription under the same id (session change/resubscribe).
const pendingToken = beginPendingSubscription(sender.id, subscriptionId)
@@ -151,7 +181,7 @@ async function handleSubscribe(event: IpcMainEvent, args: NativeChatSubscribeArg
sessionId,
transcriptPath,
initialLimit: limit,
onInitialSnapshot: (messages, hasMore, _beforeOffset, error) => {
onInitialSnapshot: (messages, hasMore, _beforeOffset, error, lifecycle) => {
if (sender.isDestroyed()) {
return
}
@@ -159,26 +189,44 @@ async function handleSubscribe(event: IpcMainEvent, args: NativeChatSubscribeArg
// instead of stranding the view at 'loading' when the read keeps throwing.
const payload: NativeChatAppendedPayload = {
subscriptionId,
frame: { type: 'snapshot', messages, hasMore, ...(error ? { error } : {}) }
frame: {
type: 'snapshot',
messages,
hasMore,
turnLifecycleCapable,
...(error ? { error } : {}),
...(lifecycle ? { lifecycle } : {})
}
}
sender.send('nativeChat:appended', payload)
},
onReplace: (messages, hasMore) => {
onReplace: (messages, hasMore, _beforeOffset, lifecycle) => {
if (sender.isDestroyed()) {
return
}
sender.send('nativeChat:appended', {
subscriptionId,
frame: { type: 'replacement', messages, hasMore }
frame: {
type: 'replacement',
messages,
hasMore,
turnLifecycleCapable,
...(lifecycle ? { lifecycle } : {})
}
} satisfies NativeChatAppendedPayload)
},
onAppend: (messages) => {
onAppend: (messages, lifecycle) => {
if (sender.isDestroyed()) {
return
}
const payload: NativeChatAppendedPayload = {
subscriptionId,
frame: { type: 'appended', messages }
frame: {
type: 'appended',
messages,
turnLifecycleCapable,
...(lifecycle ? { lifecycle } : {})
}
}
sender.send('nativeChat:appended', payload)
}
@@ -210,6 +258,7 @@ async function handleSubscribe(event: IpcMainEvent, args: NativeChatSubscribeArg
type: 'snapshot',
messages: [],
hasMore: false,
turnLifecycleCapable,
error: 'Transcript unavailable'
}
}
@@ -1,5 +1,5 @@
import { open, stat } from 'node:fs/promises'
import type { NativeChatMessage } from '../../shared/native-chat-types'
import type { NativeChatMessage, NativeChatTurnLifecycle } from '../../shared/native-chat-types'
import { transcriptFallbackId } from './transcript-fallback-id'
import {
MAX_NATIVE_CHAT_TRANSCRIPT_RECORD_BYTES,
@@ -28,7 +28,9 @@ export async function readIncrementalTranscriptMessages(
filePath: string,
state: IncrementalTranscriptState,
decode: NativeChatLineDecoder,
onBatch?: (messages: NativeChatMessage[]) => void
onBatch?: (messages: NativeChatMessage[]) => void,
decodeLifecycle?: (line: string, fallbackId: string) => NativeChatTurnLifecycle | null,
onLifecycle?: (lifecycle: NativeChatTurnLifecycle) => void
): Promise<NativeChatMessage[]> {
const end = (await stat(filePath)).size
if (end <= state.offset) {
@@ -91,7 +93,12 @@ export async function readIncrementalTranscriptMessages(
if (!line) {
return
}
const message = decode(line, transcriptFallbackId(filePath, state.pendingStart))
const fallbackId = transcriptFallbackId(filePath, state.pendingStart)
const lifecycle = decodeLifecycle?.(line, fallbackId)
if (lifecycle) {
onLifecycle?.(lifecycle)
}
const message = decode(line, fallbackId)
if (!message) {
return
}
@@ -0,0 +1,52 @@
import { describe, expect, it } from 'vitest'
import {
NATIVE_CHAT_INTERRUPTED_STATUS_TEXT,
type NativeChatMessage
} from '../../shared/native-chat-types'
import { stripNoiseMessages } from '../../shared/native-chat-noise'
import { decodeClaudeTranscriptLine } from './transcript-line-decoders-claude'
import { decodeCodexTranscriptLine } from './transcript-line-decoders-codex'
function expectNormalizedInterruption(message: NativeChatMessage | null): void {
expect(message).toMatchObject({
role: 'system',
blocks: [{ type: 'text', text: NATIVE_CHAT_INTERRUPTED_STATUS_TEXT }],
source: 'transcript'
})
expect(stripNoiseMessages(message ? [message] : [])).toEqual([message])
}
describe('native chat transcript interruption messages', () => {
it('normalizes Claude interruption boilerplate into one visible status row', () => {
const message = decodeClaudeTranscriptLine(
JSON.stringify({
type: 'user',
uuid: 'interrupt-row',
interruptedMessageId: 'assistant-request-1',
timestamp: '2026-07-16T23:46:01.000Z',
message: {
role: 'user',
content: [{ type: 'text', text: '[Request interrupted by user]' }]
}
}),
'fallback'
)
expectNormalizedInterruption(message)
expect(message?.id).toBe('interrupt-row')
})
it('normalizes Codex turn_aborted into one visible status row', () => {
const message = decodeCodexTranscriptLine(
JSON.stringify({
type: 'event_msg',
timestamp: '2026-07-16T23:46:01.000Z',
payload: { type: 'turn_aborted', reason: 'interrupted', turn_id: 'turn-2' }
}),
'fallback'
)
expectNormalizedInterruption(message)
expect(message?.id).toBe('fallback')
})
})
@@ -1,6 +1,10 @@
// Claude JSONL line → NativeChatMessage decoder.
import type { NativeChatBlock, NativeChatMessage } from '../../shared/native-chat-types'
import {
NATIVE_CHAT_INTERRUPTED_STATUS_TEXT,
type NativeChatBlock,
type NativeChatMessage
} from '../../shared/native-chat-types'
import {
asRecord,
extractString,
@@ -21,6 +25,19 @@ export function decodeClaudeTranscriptLine(
if (role !== 'user' && role !== 'assistant') {
return null
}
const timestamp = parseTimestamp(record.timestamp)
const recordMessageId = extractString(record.uuid) ?? fallbackId
if (role === 'user' && extractString(record.interruptedMessageId)) {
// Why: keep Claude's injected boilerplate out of the user-bubble path while
// preserving the interruption as a quiet, replayable conversation status.
return {
id: recordMessageId,
role: 'system',
blocks: [{ type: 'text', text: NATIVE_CHAT_INTERRUPTED_STATUS_TEXT }],
timestamp,
source: 'transcript'
}
}
const message = asRecord(record.message)
const decodedBlocks = claudeContentBlocks(message?.content)
if (decodedBlocks.length === 0) {
@@ -42,7 +59,7 @@ export function decodeClaudeTranscriptLine(
id: messageId ?? fallbackId,
role: claudeMessageRole(role, blocks),
blocks,
timestamp: parseTimestamp(record.timestamp),
timestamp,
source: 'transcript'
}
}
@@ -1,6 +1,10 @@
// Codex JSONL line → NativeChatMessage decoder.
import type { NativeChatBlock, NativeChatMessage } from '../../shared/native-chat-types'
import {
NATIVE_CHAT_INTERRUPTED_STATUS_TEXT,
type NativeChatBlock,
type NativeChatMessage
} from '../../shared/native-chat-types'
import {
asRecord,
extractString,
@@ -87,6 +91,15 @@ function codexEventMessage(
id: string,
timestamp: number | null
): NativeChatMessage | null {
if (payload.type === 'turn_aborted') {
return {
id,
role: 'system',
blocks: [{ type: 'text', text: NATIVE_CHAT_INTERRUPTED_STATUS_TEXT }],
timestamp,
source: 'transcript'
}
}
if (payload.type === 'user_message') {
const text = extractString(payload.message)
return text
+10 -2
View File
@@ -1,5 +1,9 @@
import { createReadStream } from 'node:fs'
import type { AgentType, NativeChatMessage } from '../../shared/native-chat-types'
import type {
AgentType,
NativeChatMessage,
NativeChatTurnLifecycle
} from '../../shared/native-chat-types'
import { resolveNativeChatTranscriptAgent } from '../../shared/native-chat-agent-support'
import { errorMessage } from '../ai-vault/session-scanner-values'
import { resolveSessionFilePath, type ResolveSessionFileOptions } from './session-file-resolver'
@@ -11,7 +15,11 @@ import {
import { decodeTranscriptStream } from './transcript-stream-lines'
export type ReadTranscriptResult =
| { messages: NativeChatMessage[] }
| {
messages: NativeChatMessage[]
lifecycle?: NativeChatTurnLifecycle
turnLifecycleCapable?: boolean
}
// notFound marks a retry-worthy miss (transcript not flushed to disk yet,
// #8401) as opposed to a real parse/IO error callers surface immediately.
| { error: string; notFound?: true }
+34 -5
View File
@@ -1,5 +1,9 @@
import { open, stat } from 'node:fs/promises'
import type { AgentType, NativeChatMessage } from '../../shared/native-chat-types'
import type {
AgentType,
NativeChatMessage,
NativeChatTurnLifecycle
} from '../../shared/native-chat-types'
import { resolveNativeChatTranscriptAgent } from '../../shared/native-chat-agent-support'
import { resolveSessionFilePath, type ResolveSessionFileOptions } from './session-file-resolver'
import {
@@ -8,6 +12,10 @@ import {
decodeGrokTranscriptLine
} from './transcript-line-decoders'
import { transcriptFallbackId } from './transcript-fallback-id'
import {
nativeChatTurnLifecycleDecoderForAgent,
type NativeChatTurnLifecycleDecoder
} from './transcript-turn-lifecycle'
export const MAX_NATIVE_CHAT_TRANSCRIPT_RECORD_BYTES = 2 * 1024 * 1024
const TAIL_CHUNK_BYTES = 64 * 1024
@@ -33,9 +41,11 @@ export async function readNativeChatTranscriptTailFile(
limit: number,
decode: NativeChatLineDecoder,
includeTrailingLine = false,
endOffset?: number
endOffset?: number,
decodeLifecycle?: NativeChatTurnLifecycleDecoder | null
): Promise<{
messages: NativeChatMessage[]
lifecycle?: NativeChatTurnLifecycle
consumedTo: number
hasMore: boolean
beforeOffset: number
@@ -48,6 +58,7 @@ export async function readNativeChatTranscriptTailFile(
const lineParts: Buffer[] = []
let lineBytes = 0
let lineOversized = false
let lifecycle: NativeChatTurnLifecycle | undefined
try {
const consumedTo = includeTrailingLine ? end : await findLastCompleteLineEnd(handle, end)
if (consumedTo === 0) {
@@ -87,6 +98,7 @@ export async function readNativeChatTranscriptTailFile(
const selected = limit > 0 ? chronological.slice(Math.max(0, chronological.length - limit)) : []
return {
messages: selected.map((entry) => entry.message),
...(lifecycle ? { lifecycle } : {}),
consumedTo,
hasMore: limit > 0 && chronological.length > limit,
beforeOffset: selected[0]?.offset ?? end
@@ -125,7 +137,12 @@ export async function readNativeChatTranscriptTailFile(
if (!line) {
return
}
const message = decode(line, transcriptFallbackId(filePath, lineOffset))
const fallbackId = transcriptFallbackId(filePath, lineOffset)
// Why: scan the same bounded JSONL window for provider-authored lifecycle
// records so reconnect snapshots can replay completion without guessing
// from the last visible assistant message.
lifecycle ??= decodeLifecycle?.(line, fallbackId) ?? undefined
const message = decode(line, fallbackId)
if (message) {
messages.push({ message, offset: lineOffset })
}
@@ -165,10 +182,16 @@ export async function readNativeChatTranscriptTail(
beforeOffset?: number
}
): Promise<
| { messages: NativeChatMessage[]; hasMore: boolean; beforeOffset: number }
| {
messages: NativeChatMessage[]
lifecycle?: NativeChatTurnLifecycle
hasMore: boolean
beforeOffset: number
}
| { error: string; notFound?: true }
> {
const decode = nativeChatLineDecoderForAgent(args.agent)
const decodeLifecycle = nativeChatTurnLifecycleDecoderForAgent(args.agent)
const filePath = args.filePath ?? (await resolveSessionFilePath(args.agent, args.sessionId, args))
if (!decode) {
return { error: 'Transcript unavailable' }
@@ -184,10 +207,16 @@ export async function readNativeChatTranscriptTail(
args.limit,
decode,
true,
args.beforeOffset
args.beforeOffset,
decodeLifecycle
)
return {
messages: result.messages,
// Why: an older pagination page must not rewind the live lifecycle; only
// the current transcript tail can authoritatively describe turn state.
...(args.beforeOffset === undefined && result.lifecycle
? { lifecycle: result.lifecycle }
: {}),
hasMore: result.hasMore,
beforeOffset: result.beforeOffset
}
@@ -0,0 +1,168 @@
import { describe, expect, it } from 'vitest'
import {
decodeClaudeTurnLifecycle,
decodeCodexTurnLifecycle,
supportsNativeChatTranscriptTurnLifecycle
} from './transcript-turn-lifecycle'
describe('native chat transcript turn lifecycle', () => {
it('reports which transcript formats have explicit boundaries', () => {
expect(supportsNativeChatTranscriptTurnLifecycle('claude')).toBe(true)
expect(supportsNativeChatTranscriptTurnLifecycle('openclaude')).toBe(true)
expect(supportsNativeChatTranscriptTurnLifecycle('codex')).toBe(true)
expect(supportsNativeChatTranscriptTurnLifecycle('grok')).toBe(false)
})
it('decodes Codex task boundaries with the provider turn id', () => {
expect(
decodeCodexTurnLifecycle(
JSON.stringify({
timestamp: '2026-07-16T23:40:14.001Z',
type: 'event_msg',
payload: { type: 'task_started', turn_id: 'turn-1' }
}),
'fallback'
)
).toEqual({
state: 'working',
turnId: 'turn-1',
timestamp: Date.parse('2026-07-16T23:40:14.001Z')
})
expect(
decodeCodexTurnLifecycle(
JSON.stringify({
timestamp: '2026-07-16T23:45:37.608Z',
type: 'event_msg',
payload: { type: 'task_complete', turn_id: 'turn-1' }
}),
'fallback'
)
).toEqual({
state: 'completed',
turnId: 'turn-1',
timestamp: Date.parse('2026-07-16T23:45:37.608Z')
})
expect(
decodeCodexTurnLifecycle(
JSON.stringify({
timestamp: '2026-07-16T23:46:01.000Z',
type: 'event_msg',
payload: { type: 'turn_aborted', reason: 'interrupted', turn_id: 'turn-2' }
}),
'fallback'
)
).toEqual({
state: 'interrupted',
turnId: 'turn-2',
timestamp: Date.parse('2026-07-16T23:46:01.000Z')
})
})
it('does not mistake a Codex assistant message for completion', () => {
expect(
decodeCodexTurnLifecycle(
JSON.stringify({
timestamp: '2026-07-16T23:45:37.472Z',
type: 'event_msg',
payload: { type: 'agent_message', message: 'final-looking prose' }
}),
'fallback'
)
).toBeNull()
})
it('uses Claude end_turn and excludes tool-result user rows', () => {
expect(
decodeClaudeTurnLifecycle(
JSON.stringify({
type: 'assistant',
uuid: 'assistant-1',
timestamp: '2026-07-16T23:45:37.608Z',
message: { role: 'assistant', stop_reason: 'end_turn', content: [] }
}),
'fallback'
)
).toEqual({
state: 'completed',
turnId: 'assistant-1',
timestamp: Date.parse('2026-07-16T23:45:37.608Z')
})
expect(
decodeClaudeTurnLifecycle(
JSON.stringify({
type: 'user',
uuid: 'tool-result-1',
timestamp: '2026-07-16T23:45:38.000Z',
message: {
role: 'user',
content: [{ type: 'tool_result', tool_use_id: 'tool-1', content: 'ok' }]
}
}),
'fallback'
)
).toBeNull()
})
it('excludes Claude tool-result rows that also carry text sidecars', () => {
expect(
decodeClaudeTurnLifecycle(
JSON.stringify({
type: 'user',
uuid: 'mixed-tool-result',
timestamp: '2026-07-16T23:45:38.000Z',
message: {
role: 'user',
content: [
{ type: 'tool_result', tool_use_id: 'tool-1', content: 'ok' },
{ type: 'text', text: '<system-reminder>continue</system-reminder>' }
]
}
}),
'fallback'
)
).toBeNull()
})
it('treats a real Claude user row as the next working generation', () => {
expect(
decodeClaudeTurnLifecycle(
JSON.stringify({
type: 'user',
uuid: 'user-2',
timestamp: '2026-07-16T23:46:00.000Z',
message: { role: 'user', content: 'next task' }
}),
'fallback'
)
).toEqual({
state: 'working',
turnId: 'user-2',
timestamp: Date.parse('2026-07-16T23:46:00.000Z')
})
})
it('treats Claude interruptedMessageId as terminal instead of a user generation', () => {
expect(
decodeClaudeTurnLifecycle(
JSON.stringify({
type: 'user',
uuid: 'interrupt-row',
interruptedMessageId: 'assistant-request-1',
timestamp: '2026-07-16T23:46:01.000Z',
message: {
role: 'user',
content: [{ type: 'text', text: '[Request interrupted by user]' }]
}
}),
'fallback'
)
).toEqual({
state: 'interrupted',
turnId: 'assistant-request-1',
timestamp: Date.parse('2026-07-16T23:46:01.000Z')
})
})
})
@@ -0,0 +1,100 @@
import type { AgentType, NativeChatTurnLifecycle } from '../../shared/native-chat-types'
import { resolveNativeChatTranscriptAgent } from '../../shared/native-chat-agent-support'
import {
asRecord,
extractString,
parseJsonObject,
timestampMs
} from '../ai-vault/session-scanner-values'
import { decodeClaudeTranscriptLine } from './transcript-line-decoders-claude'
export type NativeChatTurnLifecycleDecoder = (
line: string,
fallbackId: string
) => NativeChatTurnLifecycle | null
export function nativeChatTurnLifecycleDecoderForAgent(
agent: AgentType
): NativeChatTurnLifecycleDecoder | null {
const transcriptAgent = resolveNativeChatTranscriptAgent(agent)
if (transcriptAgent === 'codex') {
return decodeCodexTurnLifecycle
}
if (transcriptAgent === 'claude') {
return decodeClaudeTurnLifecycle
}
return null
}
export function supportsNativeChatTranscriptTurnLifecycle(agent: AgentType): boolean {
return nativeChatTurnLifecycleDecoderForAgent(agent) !== null
}
export function decodeCodexTurnLifecycle(
line: string,
fallbackId: string
): NativeChatTurnLifecycle | null {
const record = parseJsonObject(line)
const payload = asRecord(record?.payload)
if (record?.type !== 'event_msg' || !payload) {
return null
}
if (
payload.type !== 'task_started' &&
payload.type !== 'task_complete' &&
payload.type !== 'turn_aborted'
) {
return null
}
const state =
payload.type === 'task_started'
? 'working'
: payload.type === 'turn_aborted'
? 'interrupted'
: 'completed'
return {
state,
turnId: extractString(payload.turn_id) ?? fallbackId,
timestamp: lifecycleTimestamp(record.timestamp)
}
}
export function decodeClaudeTurnLifecycle(
line: string,
fallbackId: string
): NativeChatTurnLifecycle | null {
const record = parseJsonObject(line)
if (!record) {
return null
}
const message = asRecord(record.message)
const timestamp = lifecycleTimestamp(record.timestamp)
const interruptedMessageId = extractString(record.interruptedMessageId)
if (record.type === 'user' && interruptedMessageId) {
// Why: Claude stores its interrupt notice as an injected user row; it ends
// the active generation and must not be mistaken for the next user prompt.
return { state: 'interrupted', turnId: interruptedMessageId, timestamp }
}
if (record.type === 'assistant' && message?.stop_reason === 'end_turn') {
return {
state: 'completed',
turnId: extractString(record.uuid) ?? extractString(message.id) ?? fallbackId,
timestamp
}
}
if (record.type !== 'user') {
return null
}
const decoded = decodeClaudeTranscriptLine(line, fallbackId)
if (decoded?.role !== 'user' || decoded.blocks.some((block) => block.type === 'tool-result')) {
// Why: Claude can attach text sidecars to tool-result user rows; those are
// continuations of the active turn, not a new user-authored generation.
return null
}
return { state: 'working', turnId: decoded.id, timestamp }
}
function lifecycleTimestamp(value: unknown): number | null {
const parsed = timestampMs(value)
return Number.isFinite(parsed) ? parsed : null
}
@@ -0,0 +1,36 @@
import type {
AgentType,
NativeChatMessage,
NativeChatTurnLifecycle
} from '../../shared/native-chat-types'
import type { ResolveSessionFileOptions } from './session-file-resolver'
export type SubscribeNativeChatTranscriptArgs = ResolveSessionFileOptions & {
agent: AgentType
sessionId: string
onAppend: (messages: NativeChatMessage[], lifecycle?: NativeChatTurnLifecycle) => void
onInitialSnapshot?: (
messages: NativeChatMessage[],
hasMore: boolean,
beforeOffset: number,
/** Set when the initial drain could not deliver a transcript. */
error?: string,
lifecycle?: NativeChatTurnLifecycle
) => void
onReplace?: (
messages: NativeChatMessage[],
hasMore: boolean,
beforeOffset: number,
lifecycle?: NativeChatTurnLifecycle
) => void
initialLimit?: number
filePath?: string
debounceMs?: number
/** Test-only override for the production resolve-poll backoff. */
resolvePollIntervalMs?: number
}
export type NativeChatTranscriptSubscription = {
unsubscribe: () => void
watching: boolean
}
+47 -38
View File
@@ -1,42 +1,18 @@
import { watch, type FSWatcher } from 'node:fs'
import { open, stat } from 'node:fs/promises'
import { basename, dirname } from 'node:path'
import type { AgentType, NativeChatMessage } from '../../shared/native-chat-types'
import type { ResolveSessionFileOptions } from './session-file-resolver'
import type { NativeChatMessage, NativeChatTurnLifecycle } from '../../shared/native-chat-types'
import {
readIncrementalTranscriptMessages,
resetIncrementalTranscriptState,
type IncrementalTranscriptState
} from './transcript-incremental-reader'
import { readNativeChatTranscriptTailFile } from './transcript-tail-reader'
export type SubscribeNativeChatTranscriptArgs = ResolveSessionFileOptions & {
agent: AgentType
sessionId: string
onAppend: (messages: NativeChatMessage[]) => void
onInitialSnapshot?: (
messages: NativeChatMessage[],
hasMore: boolean,
beforeOffset: number,
/** Set when the initial drain could not deliver a transcript; the subscriber
* surfaces it as an error snapshot so a watching client never sticks on
* 'loading'. Empty messages accompany it. */
error?: string
) => void
onReplace?: (messages: NativeChatMessage[], hasMore: boolean, beforeOffset: number) => void
initialLimit?: number
filePath?: string
debounceMs?: number
/** Overrides the resolve-poll interval (see subscribeViaResolvePoll) so tests
* don't wait out the production backoff. Production ignores this and backs
* off from 500ms to a 5s cap. */
resolvePollIntervalMs?: number
}
export type NativeChatTranscriptSubscription = {
unsubscribe: () => void
watching: boolean
}
import { nativeChatTurnLifecycleDecoderForAgent } from './transcript-turn-lifecycle'
import type {
NativeChatTranscriptSubscription,
SubscribeNativeChatTranscriptArgs
} from './transcript-watch-contract'
const DEFAULT_DEBOUNCE_MS = 40
const ROTATION_RETRY_MS = 25
@@ -87,6 +63,7 @@ export async function installTranscriptWatcher(
return null
}
const { onAppend, onInitialSnapshot, onReplace, initialLimit, debounceMs } = args
const decodeLifecycle = nativeChatTurnLifecycleDecoderForAgent(args.agent)
const state: IncrementalTranscriptState = {
offset: 0,
@@ -138,6 +115,7 @@ export async function installTranscriptWatcher(
}
async function readAndEmitAppends(): Promise<void> {
let lifecycle: NativeChatTurnLifecycle | undefined
const remaining = await readIncrementalTranscriptMessages(
filePath,
state,
@@ -146,10 +124,14 @@ export async function installTranscriptWatcher(
if (!closed) {
onAppend(messages)
}
},
decodeLifecycle ?? undefined,
(nextLifecycle) => {
lifecycle = nextLifecycle
}
)
if (!closed && remaining.length > 0) {
onAppend(remaining)
if (!closed && (remaining.length > 0 || lifecycle)) {
onAppend(remaining, lifecycle)
}
}
@@ -176,7 +158,14 @@ export async function installTranscriptWatcher(
// Why: 0 is a valid window — an explicit undefined check keeps an empty
// snapshot empty instead of falling back to an unbounded incremental read.
contentReplaced && !initialDrain && onReplace && initialLimit !== undefined
? await readNativeChatTranscriptTailFile(filePath, initialLimit, decode)
? await readNativeChatTranscriptTailFile(
filePath,
initialLimit,
decode,
false,
undefined,
decodeLifecycle
)
: null
if (closed) {
return
@@ -187,7 +176,8 @@ export async function installTranscriptWatcher(
onReplace(
replacementSnapshot.messages,
replacementSnapshot.hasMore,
replacementSnapshot.beforeOffset
replacementSnapshot.beforeOffset,
replacementSnapshot.lifecycle
)
await readAndEmitAppends()
watchedBoundary = await boundaryFingerprint(filePath, state.offset)
@@ -197,7 +187,14 @@ export async function installTranscriptWatcher(
const initialSnapshot =
initialDrain && onInitialSnapshot && initialLimit !== undefined
? await readNativeChatTranscriptTailFile(filePath, initialLimit, decode)
? await readNativeChatTranscriptTailFile(
filePath,
initialLimit,
decode,
false,
undefined,
decodeLifecycle
)
: null
if (closed) {
return
@@ -210,12 +207,24 @@ export async function installTranscriptWatcher(
onInitialSnapshot(
initialSnapshot.messages,
initialSnapshot.hasMore,
initialSnapshot.beforeOffset
initialSnapshot.beforeOffset,
undefined,
initialSnapshot.lifecycle
)
await readAndEmitAppends()
} else {
const messages = await readIncrementalTranscriptMessages(filePath, state, decode)
onInitialSnapshot(messages, false, 0)
let lifecycle: NativeChatTurnLifecycle | undefined
const messages = await readIncrementalTranscriptMessages(
filePath,
state,
decode,
undefined,
decodeLifecycle ?? undefined,
(nextLifecycle) => {
lifecycle = nextLifecycle
}
)
onInitialSnapshot(messages, false, 0, undefined, lifecycle)
}
} else {
initialDrain = false
+130 -1
View File
@@ -2,7 +2,7 @@ import { appendFile, mkdtemp, rename, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
import type { NativeChatMessage } from '../../shared/native-chat-types'
import type { NativeChatMessage, NativeChatTurnLifecycle } from '../../shared/native-chat-types'
import {
getActiveNativeChatWatcherCount,
readNativeChatTranscriptTail,
@@ -45,6 +45,30 @@ function claudeLine(uuid: string, role: 'user' | 'assistant', text: string): str
})}\n`
}
function claudeEndTurnLine(uuid: string, text: string): string {
return `${JSON.stringify({
type: 'assistant',
uuid,
timestamp: '2026-06-01T10:00:01.000Z',
message: {
role: 'assistant',
stop_reason: 'end_turn',
content: [{ type: 'text', text }]
}
})}\n`
}
function codexLifecycleLine(
state: 'task_started' | 'task_complete' | 'turn_aborted',
turnId = 'turn-1'
): string {
return `${JSON.stringify({
type: 'event_msg',
timestamp: state === 'task_started' ? '2026-06-01T10:00:00.000Z' : '2026-06-01T10:00:01.000Z',
payload: { type: state, turn_id: turnId }
})}\n`
}
async function waitFor(predicate: () => boolean, timeoutMs = 2000): Promise<void> {
const start = Date.now()
while (!predicate()) {
@@ -99,6 +123,111 @@ describe('subscribeNativeChatTranscript', () => {
expect(snapshots).toEqual([[]])
})
it('replays and appends provider-authored turn lifecycle markers', async () => {
const filePath = await tempFile(claudeLine('u-1', 'user', 'first'))
const lifecycles: NativeChatTurnLifecycle[] = []
const sub = await subscribeNativeChatTranscript({
agent: 'claude',
sessionId: 'ignored',
filePath,
onInitialSnapshot: (_messages, _hasMore, _beforeOffset, _error, lifecycle) => {
if (lifecycle) {
lifecycles.push(lifecycle)
}
},
onAppend: (_messages, lifecycle) => {
if (lifecycle) {
lifecycles.push(lifecycle)
}
},
debounceMs: 5
})
await waitFor(() => lifecycles.length === 1)
await appendFile(filePath, claudeEndTurnLine('a-1', 'done'))
await waitFor(() => lifecycles.length === 2)
sub.unsubscribe()
expect(lifecycles.map((lifecycle) => lifecycle.state)).toEqual(['working', 'completed'])
expect(lifecycles.map((lifecycle) => lifecycle.turnId)).toEqual(['u-1', 'a-1'])
})
it('emits Codex task_complete even when the frame has no visible messages', async () => {
const filePath = await tempFile(codexLifecycleLine('task_started'))
const lifecycles: NativeChatTurnLifecycle[] = []
const sub = await subscribeNativeChatTranscript({
agent: 'codex',
sessionId: 'ignored',
filePath,
onInitialSnapshot: (_messages, _hasMore, _beforeOffset, _error, lifecycle) => {
if (lifecycle) {
lifecycles.push(lifecycle)
}
},
onAppend: (messages, lifecycle) => {
expect(messages).toEqual([])
if (lifecycle) {
lifecycles.push(lifecycle)
}
},
debounceMs: 5
})
await waitFor(() => lifecycles.length === 1)
await appendFile(filePath, codexLifecycleLine('task_complete'))
await waitFor(() => lifecycles.length === 2)
sub.unsubscribe()
expect(lifecycles).toMatchObject([
{ state: 'working', turnId: 'turn-1' },
{ state: 'completed', turnId: 'turn-1' }
])
})
it('replays Codex interruption as a terminal lifecycle and visible status row', async () => {
const filePath = await tempFile(
codexLifecycleLine('task_started') + codexLifecycleLine('turn_aborted')
)
let snapshot:
| { messages: NativeChatMessage[]; lifecycle: NativeChatTurnLifecycle | undefined }
| undefined
const sub = await subscribeNativeChatTranscript({
agent: 'codex',
sessionId: 'ignored',
filePath,
initialLimit: 40,
onInitialSnapshot: (messages, _hasMore, _beforeOffset, _error, lifecycle) => {
snapshot = { messages, lifecycle }
},
onAppend: () => {},
debounceMs: 5
})
await waitFor(() => snapshot !== undefined)
sub.unsubscribe()
expect(snapshot?.lifecycle).toMatchObject({ state: 'interrupted', turnId: 'turn-1' })
expect(snapshot?.messages).toMatchObject([
{ role: 'system', blocks: [{ type: 'text', text: 'Conversation interrupted' }] }
])
})
it('does not replay an older interruption over a newer working turn', async () => {
const filePath = await tempFile(
codexLifecycleLine('task_started', 'turn-1') +
codexLifecycleLine('turn_aborted', 'turn-1') +
codexLifecycleLine('task_started', 'turn-2')
)
const result = await readNativeChatTranscriptTail({
agent: 'codex',
sessionId: 'ignored',
filePath,
limit: 40
})
expect(result).toMatchObject({ lifecycle: { state: 'working', turnId: 'turn-2' } })
})
it('emits a bulk append in bounded ordered batches', async () => {
const filePath = await tempFile('')
const batches: NativeChatMessage[][] = []
+10 -10
View File
@@ -1,19 +1,19 @@
import { extname } from 'node:path'
import type { NativeChatMessage } from '../../shared/native-chat-types'
import { resolveSessionFilePath } from './session-file-resolver'
import {
installTranscriptWatcher,
type NativeChatTranscriptSubscription,
type SubscribeNativeChatTranscriptArgs
} from './transcript-watch-engine'
import { installTranscriptWatcher } from './transcript-watch-engine'
import type {
NativeChatTranscriptSubscription,
SubscribeNativeChatTranscriptArgs
} from './transcript-watch-contract'
import { nativeChatLineDecoderForAgent } from './transcript-tail-reader'
export { readNativeChatTranscriptTail } from './transcript-tail-reader'
export {
getActiveNativeChatWatcherCount,
type NativeChatTranscriptSubscription,
type SubscribeNativeChatTranscriptArgs
} from './transcript-watch-engine'
export { getActiveNativeChatWatcherCount } from './transcript-watch-engine'
export type {
NativeChatTranscriptSubscription,
SubscribeNativeChatTranscriptArgs
} from './transcript-watch-contract'
/** One resolve+install attempt. Returns null when the file isn't resolvable
* yet, or vanished between resolve and `watch()` — either case is retried by
@@ -331,7 +331,13 @@ describe('nativeChat.subscribe initial snapshot', () => {
)
expect(emitted).toEqual([
{ type: 'snapshot', messages: [], hasMore: false, error: 'Transcript unavailable' }
{
type: 'snapshot',
messages: [],
hasMore: false,
turnLifecycleCapable: true,
error: 'Transcript unavailable'
}
])
})
@@ -353,6 +359,7 @@ describe('nativeChat.subscribe initial snapshot', () => {
messages: [],
hasMore: false,
beforeOffset: 0,
turnLifecycleCapable: true,
error: 'Transcript unavailable'
}
])
@@ -375,7 +382,8 @@ describe('nativeChat.subscribe initial snapshot', () => {
type: 'snapshot',
messages: [expect.objectContaining({ id: 'a-1' })],
hasMore: false,
beforeOffset: 7
beforeOffset: 7,
turnLifecycleCapable: true
}
])
})
+28 -8
View File
@@ -5,6 +5,7 @@ import {
readNativeChatTranscriptTail,
subscribeNativeChatTranscript
} from '../../../native-chat/transcript-watch'
import { supportsNativeChatTranscriptTurnLifecycle } from '../../../native-chat/transcript-turn-lifecycle'
import { defineMethod, defineStreamingMethod, type RpcAnyMethod, type RpcContext } from '../core'
// Why: native chat renders an agent's own transcript (Claude/Codex JSONL). The
@@ -183,6 +184,7 @@ export const NATIVE_CHAT_METHODS: readonly RpcAnyMethod[] = [
params: NativeChatSession,
handler: async (params, { clientKind }) => {
const limit = params.limit ?? MOBILE_NATIVE_CHAT_DEFAULT_WINDOW
const turnLifecycleCapable = supportsNativeChatTranscriptTurnLifecycle(params.agent)
const result = await readNativeChatTranscriptTail({
agent: params.agent,
sessionId: params.sessionId,
@@ -194,7 +196,9 @@ export const NATIVE_CHAT_METHODS: readonly RpcAnyMethod[] = [
? {
messages: windowForClient(result.messages, clientKind, limit),
hasMore: result.hasMore,
beforeOffset: result.beforeOffset
beforeOffset: result.beforeOffset,
turnLifecycleCapable,
...(result.lifecycle ? { lifecycle: result.lifecycle } : {})
}
: result
}
@@ -215,6 +219,7 @@ export const NATIVE_CHAT_METHODS: readonly RpcAnyMethod[] = [
const cleanupToken = params.subscriptionId ?? `${params.agent}:${params.sessionId}`
const subscriptionId = `nativeChat:${connectionId ?? 'local'}:${cleanupToken}`
const limit = params.limit ?? MOBILE_NATIVE_CHAT_DEFAULT_WINDOW
const turnLifecycleCapable = supportsNativeChatTranscriptTurnLifecycle(params.agent)
runtime.registerSubscriptionCleanup(
subscriptionId,
() => {
@@ -232,7 +237,7 @@ export const NATIVE_CHAT_METHODS: readonly RpcAnyMethod[] = [
sessionId: params.sessionId,
transcriptPath: params.transcriptPath,
initialLimit: limit,
onInitialSnapshot: (messages, hasMore, beforeOffset, error) => {
onInitialSnapshot: (messages, hasMore, beforeOffset, error, lifecycle) => {
if (closed) {
return
}
@@ -243,10 +248,12 @@ export const NATIVE_CHAT_METHODS: readonly RpcAnyMethod[] = [
messages: windowForClient(messages, clientKind, limit),
hasMore,
beforeOffset,
...(error ? { error } : {})
turnLifecycleCapable,
...(error ? { error } : {}),
...(lifecycle ? { lifecycle } : {})
})
},
onReplace: (messages, hasMore, beforeOffset) => {
onReplace: (messages, hasMore, beforeOffset, lifecycle) => {
if (closed) {
return
}
@@ -254,14 +261,21 @@ export const NATIVE_CHAT_METHODS: readonly RpcAnyMethod[] = [
type: 'replacement',
messages: windowForClient(messages, clientKind, limit),
hasMore,
beforeOffset
beforeOffset,
turnLifecycleCapable,
...(lifecycle ? { lifecycle } : {})
})
},
onAppend: (messages) => {
onAppend: (messages, lifecycle) => {
if (closed) {
return
}
emit({ type: 'appended', messages: sanitizeAppendForClient(messages, clientKind) })
emit({
type: 'appended',
messages: sanitizeAppendForClient(messages, clientKind),
turnLifecycleCapable,
...(lifecycle ? { lifecycle } : {})
})
}
})
// The connection may have closed while the file was being resolved.
@@ -270,7 +284,13 @@ export const NATIVE_CHAT_METHODS: readonly RpcAnyMethod[] = [
return
}
if (!subscription.watching) {
emit({ type: 'snapshot', messages: [], hasMore: false, error: 'Transcript unavailable' })
emit({
type: 'snapshot',
messages: [],
hasMore: false,
turnLifecycleCapable,
error: 'Transcript unavailable'
})
}
unsubscribe = subscription.unsubscribe
}
+32 -5
View File
@@ -427,7 +427,11 @@ import type {
AiVaultSubagentListArgs,
AiVaultSubagentListResult
} from '../shared/ai-vault-types'
import type { AgentType, NativeChatMessage } from '../shared/native-chat-types'
import type {
AgentType,
NativeChatMessage,
NativeChatTurnLifecycle
} from '../shared/native-chat-types'
import type { TelemetryConsentState } from '../shared/telemetry-consent-types'
import type { AgentKind, LaunchSource, RequestKind } from '../shared/telemetry-events'
import type { AppStarSource } from '../shared/gh-star-source'
@@ -836,16 +840,39 @@ export type AiVaultApi = {
// notFound marks a miss caused by the transcript not existing on disk yet
// (retry-worthy), as opposed to a real read/parse error (#8401).
export type NativeChatReadSessionResult =
| { messages: NativeChatMessage[] }
| {
messages: NativeChatMessage[]
lifecycle?: NativeChatTurnLifecycle
/** True when this host can decode authoritative transcript turn boundaries. */
turnLifecycleCapable?: boolean
}
| { error: string; notFound?: true }
/** Messages appended to a live-tailed transcript since the previous emit. */
export type NativeChatAppendedMessages = NativeChatMessage[]
export type NativeChatSubscriptionFrame =
| { type: 'snapshot'; messages: NativeChatMessage[]; hasMore: boolean; error?: string }
| { type: 'replacement'; messages: NativeChatMessage[]; hasMore: boolean }
| { type: 'appended'; messages: NativeChatMessage[] }
| {
type: 'snapshot'
messages: NativeChatMessage[]
hasMore: boolean
error?: string
lifecycle?: NativeChatTurnLifecycle
turnLifecycleCapable?: boolean
}
| {
type: 'replacement'
messages: NativeChatMessage[]
hasMore: boolean
lifecycle?: NativeChatTurnLifecycle
turnLifecycleCapable?: boolean
}
| {
type: 'appended'
messages: NativeChatMessage[]
lifecycle?: NativeChatTurnLifecycle
turnLifecycleCapable?: boolean
}
/** Wire payload for the `nativeChat:appended` push channel. */
export type NativeChatAppendedPayload = {
@@ -156,10 +156,9 @@ function NativeChatResolvedView({
const launchPrompt = useAppStore((s) => s.nativeChatLaunchPromptByTabId[terminalTabId] ?? null)
const clearNativeChatLaunchPrompt = useAppStore((s) => s.clearNativeChatLaunchPrompt)
const paneLaunchPrompt = launchPrompt?.agent === agent ? launchPrompt : null
// Live hook state for this pane, selected directly so the working indicator
// flips the instant the agent reports 'working' — even when switching to chat
// mid-turn before the transcript merge has caught up.
const hookWorking = useAppStore((s) => s.agentStatusByPaneKey[paneKey]?.state === 'working')
// The live-session merge reconciles hooks with replayable transcript turn
// boundaries; all working consumers must use that one lifecycle decision.
const liveWorking = session.status === 'working'
// The agent's in-progress reply preview (hook), shown as a live streaming
// bubble while it works — before the completed turn flushes to the transcript.
const hookPreview = useAppStore((s) => s.agentStatusByPaneKey[paneKey]?.lastAssistantMessage)
@@ -308,9 +307,9 @@ function NativeChatResolvedView({
? [...sessionAfterCommandBoundaries.messages, ...pendingMessages]
: sessionAfterCommandBoundaries.messages,
previewText: hookPreview,
working: hookWorking
working: liveWorking
})
}, [sessionAfterCommandBoundaries.messages, pendingMessages, hookPreview, hookWorking])
}, [sessionAfterCommandBoundaries.messages, pendingMessages, hookPreview, liveWorking])
const sessionWithPending = useMemo<typeof session>(() => {
if (pending.length === 0 && commandMarkers.length === 0 && !streamingText) {
return sessionAfterCommandBoundaries
@@ -331,19 +330,14 @@ function NativeChatResolvedView({
const viewState = selectNativeChatViewState(sessionWithPending)
const isConversation = viewState.kind === 'ready'
// Drive "working" from the live hook state too: when toggling to chat while the
// agent is mid-turn, the merged transcript may not yet reflect the in-flight
// turn, but the hook already says 'working' — show the indicator immediately.
const viewWorking = viewState.kind === 'ready' && viewState.isWorking
useEffect(() => {
if (shouldClearNativeChatWorkingSuppression({ viewWorking, hookWorking })) {
if (shouldClearNativeChatWorkingSuppression({ working: liveWorking })) {
setWorkingInterrupted(false)
}
}, [viewWorking, hookWorking])
}, [liveWorking])
const isWorking = shouldShowNativeChatWorking({
isConversation,
viewWorking,
hookWorking,
working: liveWorking,
interrupted: workingInterrupted
})
@@ -1,14 +1,15 @@
// Pure merge of live hook turn-state into a NativeChatSession status override.
// Kept separate from the React hook so the precedence rule (live 'working'
// surfaces before the transcript flushes the final assistant message, then is
// superseded once it lands) is unit-testable without IPC or the store.
// surfaces before the transcript flushes its explicit terminal record, then is
// reconciled once that boundary lands) is unit-testable without IPC.
import type { AgentStatusState } from '../../../../shared/agent-status-types'
import { assembleNativeChatSession, type NativeChatSources } from './native-chat-session-assembler'
import type {
AgentType,
NativeChatSession,
NativeChatSessionStatus
NativeChatSessionStatus,
NativeChatTurnLifecycle
} from '../../../../shared/native-chat-types'
export type NativeChatLiveMergeInput = {
@@ -17,9 +18,14 @@ export type NativeChatLiveMergeInput = {
agent: AgentType
/** Live hook state for the pane, or null when no hook entry exists. */
hookState: AgentStatusState | null
/** Epoch ms when the current hook state began, or null when unknown. Lets a
* stale 'working' self-heal once this turn's own assistant reply has landed. */
/** Epoch ms when the current hook state began, or null when unknown. */
stateStartedAt?: number | null
/** Latest provider-authored turn boundary recovered from the transcript. */
transcriptLifecycle?: NativeChatTurnLifecycle
/** Whether the serving host can decode explicit transcript turn boundaries. */
turnLifecycleCapable?: boolean
/** Claude can finish its lead turn while background children remain active. */
hookHasWorkingSubagents?: boolean
/** True before the initial snapshot resolves; forces 'loading'. */
loading?: boolean
/** Set when the initial snapshot failed; forces 'error'. */
@@ -28,25 +34,42 @@ export type NativeChatLiveMergeInput = {
/**
* Decide the session status given the merged transcript/append messages and the
* live hook state. The transcript is the source of truth for content; the hook
* only fills the gap while the agent is mid-turn.
* live hook state. The transcript is the source of truth for content; explicit
* provider lifecycle records reconcile a dropped final hook.
*
* Precedence:
* - error / loading overrides win outright.
* - hook 'working' stays authoritative until the hook exits that state OR this
* turn's own assistant reply lands (a trailing reply newer than
* stateStartedAt); a trailing reply from an EARLIER turn does not suppress it.
* - errors win outright; live work wins over transcript loading.
* - hook 'working' stays authoritative until the hook exits that state OR an
* explicit terminal marker for this turn lands.
*/
export function mergeNativeChatLiveSession(input: NativeChatLiveMergeInput): NativeChatSession {
const { sources, sessionId, agent, hookState, stateStartedAt, loading, error } = input
const {
sources,
sessionId,
agent,
hookState,
stateStartedAt,
transcriptLifecycle,
turnLifecycleCapable,
hookHasWorkingSubagents,
loading,
error
} = input
if (error) {
return assembleNativeChatSession({ sources, sessionId, agent, status: 'error', error })
}
if (loading) {
const status = liveStatusOverride(
hookState,
sources,
stateStartedAt,
transcriptLifecycle,
turnLifecycleCapable ?? false,
hookHasWorkingSubagents ?? false
)
if (loading && status !== 'working') {
return assembleNativeChatSession({ sources, sessionId, agent, status: 'loading' })
}
const status = liveStatusOverride(hookState, sources, stateStartedAt)
return assembleNativeChatSession({
sources,
sessionId,
@@ -58,26 +81,51 @@ export function mergeNativeChatLiveSession(input: NativeChatLiveMergeInput): Nat
function liveStatusOverride(
hookState: AgentStatusState | null,
sources: NativeChatSources,
stateStartedAt: number | null | undefined
stateStartedAt: number | null | undefined,
transcriptLifecycle: NativeChatTurnLifecycle | undefined,
turnLifecycleCapable: boolean,
hookHasWorkingSubagents: boolean
): NativeChatSessionStatus | undefined {
// Only 'working' drives a live override; blocked/waiting/done leave the
// derived (ready/empty) status alone so completed turns render normally.
if (hookState !== 'working') {
return undefined
}
// Self-heal a stale 'working' (dropped/late Stop hook): if this turn's own
// assistant reply has already landed, the turn is effectively visible — stop
// asserting 'working'. A trailing reply from a PRIOR turn (older than the
// working turn's start) must not suppress it: the agent is working again.
if (trailingAssistantPostDates(sources, stateStartedAt)) {
// Why: a lead completion does not end Claude's aggregate turn while a
// background child still runs; the hook roster owns that extra lifetime.
if (hookHasWorkingSubagents) {
return 'working'
}
if (lifecycleTerminatesCurrentTurn(transcriptLifecycle, stateStartedAt)) {
return undefined
}
// Why: Grok and older remote hosts have no explicit boundary decoder. Keep
// their degraded recovery without letting prose override capable providers.
if (
!turnLifecycleCapable &&
transcriptLifecycle === undefined &&
trailingAssistantPostDates(sources, stateStartedAt)
) {
return undefined
}
return 'working'
}
/** True when the transcript's last message is an assistant reply that landed at
* or after `stateStartedAt`. Unknown timings (no start, no message timestamp)
* return false so the caller keeps 'working' — the safe, non-regressing default. */
function lifecycleTerminatesCurrentTurn(
lifecycle: NativeChatTurnLifecycle | undefined,
stateStartedAt: number | null | undefined
): boolean {
if (lifecycle?.state !== 'completed' && lifecycle?.state !== 'interrupted') {
return false
}
if (stateStartedAt == null || lifecycle.timestamp == null) {
// Why: without comparable timing, a replayed terminal marker could belong to
// the prior turn and must not hide a newer live working signal.
return false
}
return lifecycle.timestamp >= stateStartedAt
}
function trailingAssistantPostDates(
sources: NativeChatSources,
stateStartedAt: number | null | undefined
@@ -86,8 +134,5 @@ function trailingAssistantPostDates(
return false
}
const last = (sources.transcript ?? []).at(-1)
if (last?.role !== 'assistant' || last.timestamp == null) {
return false
}
return last.timestamp >= stateStartedAt
return last?.role === 'assistant' && last.timestamp != null && last.timestamp >= stateStartedAt
}
@@ -38,6 +38,7 @@ describe('isNoiseMessage', () => {
it('keeps assistant and tool turns', () => {
expect(isNoiseMessage(msg('assistant', '<system-reminder> in prose'))).toBe(false)
expect(isNoiseMessage(msg('system', 'Conversation interrupted'))).toBe(false)
})
it('keeps a user turn that carries tool results', () => {
@@ -0,0 +1,58 @@
import type {
NativeChatAppendedMessages,
NativeChatReadSessionResult
} from '../../../../preload/api-types'
import type { NativeChatTurnLifecycle } from '../../../../shared/native-chat-types'
export const RUNTIME_NATIVE_CHAT_READ_ERROR = "Couldn't read agent chat from the remote runtime."
export function parseRuntimeNativeChatTurnLifecycle(
value: unknown
): NativeChatTurnLifecycle | undefined {
if (typeof value !== 'object' || value === null) {
return undefined
}
const record = value as Record<string, unknown>
if (
(record.state !== 'working' &&
record.state !== 'completed' &&
record.state !== 'interrupted') ||
typeof record.turnId !== 'string' ||
record.turnId.trim().length === 0 ||
(record.timestamp !== null &&
(typeof record.timestamp !== 'number' ||
!Number.isFinite(record.timestamp) ||
record.timestamp <= 0))
) {
return undefined
}
return {
state: record.state,
turnId: record.turnId.trim(),
timestamp: record.timestamp
}
}
export function parseRuntimeNativeChatReadSessionResult(
value: unknown
): NativeChatReadSessionResult {
if (typeof value !== 'object' || value === null) {
return { error: RUNTIME_NATIVE_CHAT_READ_ERROR }
}
const record = value as Record<string, unknown>
if (Array.isArray(record.messages)) {
const lifecycle = parseRuntimeNativeChatTurnLifecycle(record.lifecycle)
return {
messages: record.messages as NativeChatAppendedMessages,
turnLifecycleCapable: record.turnLifecycleCapable === true,
...(lifecycle ? { lifecycle } : {})
}
}
if (typeof record.error === 'string') {
return {
error: record.error,
...(record.notFound === true ? { notFound: true } : {})
}
}
return { error: RUNTIME_NATIVE_CHAT_READ_ERROR }
}
@@ -68,7 +68,9 @@ describe('getNativeChatSessionTransport — selection', () => {
it('returns the runtime adapter for an owner on the desktop client', async () => {
markRuntimeEnvironmentCompatible(ENV)
runtimeEnvironmentsCall.mockResolvedValue(okEnvelope({ messages: [] }))
runtimeEnvironmentsCall.mockResolvedValue(
okEnvelope({ messages: [], turnLifecycleCapable: true })
)
const transport = getNativeChatSessionTransport(ENV)
await transport.readSession('claude', 'sess-1', 40, '/t/path')
@@ -82,6 +84,47 @@ describe('getNativeChatSessionTransport — selection', () => {
expect(nativeChatReadSession).not.toHaveBeenCalled()
})
it('validates lifecycle metadata on runtime read responses', async () => {
markRuntimeEnvironmentCompatible(ENV)
const lifecycle = { state: 'completed', turnId: 'turn-1', timestamp: 42 } as const
runtimeEnvironmentsCall
.mockResolvedValueOnce(
okEnvelope({ messages: [message('valid')], turnLifecycleCapable: true, lifecycle })
)
.mockResolvedValueOnce(
okEnvelope({
messages: [message('invalid')],
lifecycle: { state: 'completed', turnId: '', timestamp: undefined }
})
)
const transport = getNativeChatSessionTransport(ENV)
await expect(transport.readSession('claude', 'sess-1')).resolves.toEqual({
messages: [message('valid')],
turnLifecycleCapable: true,
lifecycle
})
await expect(transport.readSession('claude', 'sess-1')).resolves.toEqual({
messages: [message('invalid')],
turnLifecycleCapable: false
})
})
it('accepts interrupted lifecycle metadata from a remote runtime', async () => {
markRuntimeEnvironmentCompatible(ENV)
const lifecycle = { state: 'interrupted', turnId: 'turn-2', timestamp: 43 } as const
runtimeEnvironmentsCall.mockResolvedValueOnce(
okEnvelope({ messages: [message('interrupted')], turnLifecycleCapable: true, lifecycle })
)
const transport = getNativeChatSessionTransport(ENV)
await expect(transport.readSession('codex', 'sess-1')).resolves.toEqual({
messages: [message('interrupted')],
turnLifecycleCapable: true,
lifecycle
})
})
it('returns the local adapter on the web client even with an owner (R3 guard)', async () => {
;(window as unknown as { __ORCA_WEB_CLIENT__?: boolean }).__ORCA_WEB_CLIENT__ = true
nativeChatReadSession.mockResolvedValue({ messages: [] })
@@ -139,17 +182,55 @@ describe('runtime subscribe', () => {
expect(onFrame).toHaveBeenNthCalledWith(1, {
type: 'snapshot',
messages: [message('m-1')],
hasMore: false
hasMore: false,
turnLifecycleCapable: false
})
expect(onFrame).toHaveBeenNthCalledWith(2, {
type: 'snapshot',
messages: [message('m-snapshot')],
hasMore: false
hasMore: false,
turnLifecycleCapable: false
})
expect(onFrame).toHaveBeenNthCalledWith(3, {
type: 'replacement',
messages: [message('m-replacement')],
hasMore: true
hasMore: true,
turnLifecycleCapable: false
})
})
it('validates lifecycle metadata on runtime stream frames', async () => {
markRuntimeEnvironmentCompatible(ENV)
const { deliver } = stubSubscribe()
const onFrame = vi.fn()
const transport = getNativeChatSessionTransport(ENV)
const lifecycle = { state: 'completed', turnId: 'turn-1', timestamp: 42 } as const
transport.subscribe({ subscriptionId: 's-1', agent: 'claude', sessionId: 'sess-1' }, onFrame)
await Promise.resolve()
deliver({
type: 'snapshot',
messages: [message('valid')],
turnLifecycleCapable: true,
lifecycle
})
deliver({
type: 'appended',
messages: [message('invalid')],
lifecycle: { state: 'completed', turnId: '', timestamp: undefined }
})
expect(onFrame).toHaveBeenNthCalledWith(1, {
type: 'snapshot',
messages: [message('valid')],
hasMore: false,
turnLifecycleCapable: true,
lifecycle
})
expect(onFrame).toHaveBeenNthCalledWith(2, {
type: 'appended',
messages: [message('invalid')],
turnLifecycleCapable: false
})
})
@@ -192,6 +273,7 @@ describe('runtime subscribe', () => {
type: 'snapshot',
messages: [],
hasMore: false,
turnLifecycleCapable: false,
error: 'runtime too old'
})
})
@@ -1,8 +1,4 @@
import type {
NativeChatApi,
NativeChatAppendedMessages,
NativeChatReadSessionResult
} from '../../../../preload/api-types'
import type { NativeChatApi, NativeChatAppendedMessages } from '../../../../preload/api-types'
import { isWebClientLocation } from '@/lib/web-client-location'
import {
callRuntimeRpc,
@@ -10,6 +6,11 @@ import {
type RuntimeClientTarget
} from '@/runtime/runtime-rpc-client'
import { isRuntimeCompatBlockError } from '@/runtime/runtime-protocol-compat'
import {
parseRuntimeNativeChatReadSessionResult,
parseRuntimeNativeChatTurnLifecycle,
RUNTIME_NATIVE_CHAT_READ_ERROR
} from './native-chat-runtime-contract'
/** The read/subscribe surface the live-session hook needs, decoupled from where
* the transcript actually lives. Same shape as `window.api.nativeChat`, so the
@@ -34,7 +35,7 @@ export function toRuntimeNativeChatErrorMessage(err: unknown): string {
if (isRuntimeCompatBlockError(err)) {
return RUNTIME_TOO_OLD
}
return "Couldn't read agent chat from the remote runtime."
return RUNTIME_NATIVE_CHAT_READ_ERROR
}
/** Delegates straight to the local Electron IPC bridge. On the web client
@@ -53,12 +54,13 @@ function createRuntimeNativeChatTransport(environmentId: string): NativeChatSess
return {
readSession: async (agent, sessionId, limit, transcriptPath) => {
try {
return await callRuntimeRpc<NativeChatReadSessionResult>(
const result = await callRuntimeRpc<unknown>(
target,
'nativeChat.readSession',
{ agent, sessionId, limit, transcriptPath },
{ timeoutMs: 15_000 }
)
return parseRuntimeNativeChatReadSessionResult(result)
} catch (err) {
return { error: toRuntimeNativeChatErrorMessage(err) }
}
@@ -131,7 +133,11 @@ function createRuntimeNativeChatTransport(environmentId: string): NativeChatSess
messages?: NativeChatAppendedMessages
hasMore?: boolean
error?: string
lifecycle?: unknown
turnLifecycleCapable?: unknown
}
const lifecycle = parseRuntimeNativeChatTurnLifecycle(frame?.lifecycle)
const turnLifecycleCapable = frame?.turnLifecycleCapable === true
if (
(frame?.type === 'appended' ||
frame?.type === 'snapshot' ||
@@ -144,14 +150,18 @@ function createRuntimeNativeChatTransport(environmentId: string): NativeChatSess
type: 'snapshot',
messages: frame.messages,
hasMore: frame.hasMore ?? frame.messages.length >= (limit ?? 300),
...(frame.error ? { error: frame.error } : {})
turnLifecycleCapable,
...(frame.error ? { error: frame.error } : {}),
...(lifecycle ? { lifecycle } : {})
})
} else if (frame.type === 'snapshot') {
onFrame({
type: 'snapshot',
messages: frame.messages,
hasMore: frame.hasMore ?? false,
...(frame.error ? { error: frame.error } : {})
turnLifecycleCapable,
...(frame.error ? { error: frame.error } : {}),
...(lifecycle ? { lifecycle } : {})
})
} else {
onFrame(
@@ -159,9 +169,16 @@ function createRuntimeNativeChatTransport(environmentId: string): NativeChatSess
? {
type: 'replacement',
messages: frame.messages,
hasMore: frame.hasMore ?? false
hasMore: frame.hasMore ?? false,
turnLifecycleCapable,
...(lifecycle ? { lifecycle } : {})
}
: {
type: 'appended',
messages: frame.messages,
turnLifecycleCapable,
...(lifecycle ? { lifecycle } : {})
}
: { type: 'appended', messages: frame.messages }
)
}
} else if (!receivedInitial) {
@@ -9,8 +9,7 @@ describe('native chat working suppression', () => {
expect(
shouldShowNativeChatWorking({
isConversation: true,
viewWorking: true,
hookWorking: true,
working: true,
interrupted: true
})
).toBe(false)
@@ -20,19 +19,14 @@ describe('native chat working suppression', () => {
expect(
shouldShowNativeChatWorking({
isConversation: true,
viewWorking: false,
hookWorking: true,
working: true,
interrupted: false
})
).toBe(true)
})
it('clears suppression only after all working signals clear', () => {
expect(shouldClearNativeChatWorkingSuppression({ viewWorking: true, hookWorking: false })).toBe(
false
)
expect(
shouldClearNativeChatWorkingSuppression({ viewWorking: false, hookWorking: false })
).toBe(true)
it('clears suppression after reconciled working clears', () => {
expect(shouldClearNativeChatWorkingSuppression({ working: true })).toBe(false)
expect(shouldClearNativeChatWorkingSuppression({ working: false })).toBe(true)
})
})
@@ -1,16 +1,11 @@
export function shouldShowNativeChatWorking(args: {
isConversation: boolean
viewWorking: boolean
hookWorking: boolean
working: boolean
interrupted: boolean
}): boolean {
const rawWorking = args.isConversation && (args.viewWorking || args.hookWorking)
return rawWorking && !args.interrupted
return args.isConversation && args.working && !args.interrupted
}
export function shouldClearNativeChatWorkingSuppression(args: {
viewWorking: boolean
hookWorking: boolean
}): boolean {
return !args.viewWorking && !args.hookWorking
export function shouldClearNativeChatWorkingSuppression(args: { working: boolean }): boolean {
return !args.working
}
@@ -0,0 +1,20 @@
import { useAppStore } from '../../store'
import type { AgentStatusState } from '../../../../shared/agent-status-types'
export function useNativeChatHookStatus(
paneKey: string
): readonly [AgentStatusState | null, number | null, boolean] {
// Why: primitive selectors keep unrelated pane/status updates from rerendering
// native chat while still exposing the three fields used for reconciliation.
const state = useAppStore((store) => store.agentStatusByPaneKey[paneKey]?.state ?? null)
const stateStartedAt = useAppStore(
(store) => store.agentStatusByPaneKey[paneKey]?.stateStartedAt ?? null
)
const hasWorkingSubagents = useAppStore(
(store) =>
store.agentStatusByPaneKey[paneKey]?.subagents?.some(
(subagent) => subagent.state === 'working'
) ?? false
)
return [state, stateStartedAt, hasWorkingSubagents]
}
@@ -99,28 +99,103 @@ describe('mergeNativeChatLiveSession', () => {
expect(session.status).toBe('working')
})
it("self-heals a stale 'working' once this turn's assistant reply lands", () => {
// Trailing assistant reply (ts 2) post-dates the working turn's start (ts 1),
// so a dropped/late Stop hook must not strand 'working' on a visible reply.
it('does not treat assistant prose as turn completion', () => {
const session = mergeNativeChatLiveSession({
sources: { transcript: [user('u-1', 'go'), assistant('a-1', 'done')] },
sessionId: 'sess',
agent: 'claude',
hookState: 'working',
stateStartedAt: 1
stateStartedAt: 1,
turnLifecycleCapable: true
})
expect(session.status).toBe('working')
})
it('settles a dropped working hook from an explicit completion marker', () => {
const session = mergeNativeChatLiveSession({
sources: { transcript: [user('u-1', 'go'), assistant('a-1', 'done')] },
sessionId: 'sess',
agent: 'claude',
hookState: 'working',
stateStartedAt: 1,
turnLifecycleCapable: true,
transcriptLifecycle: { state: 'completed', turnId: 'turn-1', timestamp: 2 }
})
expect(session.status).toBe('ready')
})
it("keeps 'working' when the trailing assistant reply predates the working turn", () => {
// Trailing reply (ts 2) is older than the new working turn (started at ts 5),
// so the agent is working again — the stale reply must not suppress 'working'.
it('settles a dropped working hook from an explicit interruption marker', () => {
const session = mergeNativeChatLiveSession({
sources: { transcript: [user('u-1', 'go'), assistant('a-1', 'prior')] },
sources: { transcript: [user('u-1', 'go')] },
sessionId: 'sess',
agent: 'claude',
hookState: 'working',
stateStartedAt: 5
stateStartedAt: 1,
transcriptLifecycle: { state: 'interrupted', turnId: 'turn-1', timestamp: 2 }
})
expect(session.status).toBe('ready')
})
it('does not apply an older completion marker to a newer working turn', () => {
const session = mergeNativeChatLiveSession({
sources: { transcript: [assistant('a-1', 'prior')] },
sessionId: 'sess',
agent: 'claude',
hookState: 'working',
stateStartedAt: 5,
turnLifecycleCapable: true,
transcriptLifecycle: { state: 'completed', turnId: 'turn-1', timestamp: 2 }
})
expect(session.status).toBe('working')
})
it('does not apply an older interruption marker to a newer working turn', () => {
const session = mergeNativeChatLiveSession({
sources: { transcript: [assistant('a-1', 'prior')] },
sessionId: 'sess',
agent: 'claude',
hookState: 'working',
stateStartedAt: 5,
transcriptLifecycle: { state: 'interrupted', turnId: 'turn-1', timestamp: 2 }
})
expect(session.status).toBe('working')
})
it('does not apply an unorderable completion marker to live work', () => {
const session = mergeNativeChatLiveSession({
sources: { transcript: [assistant('a-1', 'prior')] },
sessionId: 'sess',
agent: 'claude',
hookState: 'working',
stateStartedAt: 5,
turnLifecycleCapable: true,
transcriptLifecycle: { state: 'completed', turnId: 'turn-1', timestamp: null }
})
expect(session.status).toBe('working')
})
it('preserves the assistant fallback when the serving host lacks explicit boundaries', () => {
const session = mergeNativeChatLiveSession({
sources: { transcript: [assistant('a-1', 'done')] },
sessionId: 'sess',
agent: 'grok',
hookState: 'working',
stateStartedAt: 1,
turnLifecycleCapable: false
})
expect(session.status).toBe('ready')
})
it('keeps working while the hook reports a live background child', () => {
const session = mergeNativeChatLiveSession({
sources: { transcript: [assistant('a-1', 'lead done')] },
sessionId: 'sess',
agent: 'claude',
hookState: 'working',
stateStartedAt: 1,
turnLifecycleCapable: true,
transcriptLifecycle: { state: 'completed', turnId: 'turn-1', timestamp: 2 },
hookHasWorkingSubagents: true
})
expect(session.status).toBe('working')
})
@@ -135,7 +210,7 @@ describe('mergeNativeChatLiveSession', () => {
expect(session.status).toBe('ready')
})
it('honors loading and error overrides outright', () => {
it('surfaces live work while the transcript loads and honors errors outright', () => {
expect(
mergeNativeChatLiveSession({
sources: { transcript: [] },
@@ -144,7 +219,7 @@ describe('mergeNativeChatLiveSession', () => {
hookState: 'working',
loading: true
}).status
).toBe('loading')
).toBe('working')
const errored = mergeNativeChatLiveSession({
sources: { transcript: [] },
@@ -475,9 +550,7 @@ describe('useNativeChatLiveSession — transport routing', () => {
expect(latest?.messages.map((message) => message.id)).toEqual(['u-live'])
})
it("self-heals a stale 'working' hook once this turn's reply has landed", async () => {
// The hook must read stateStartedAt from the store and thread it into the
// merge, so a dropped Stop hook doesn't strand 'Agent is working'.
it("self-heals a stale 'working' hook once the turn-complete marker lands", async () => {
useAppStore.setState({
agentStatusByPaneKey: { [PANE]: { state: 'working', stateStartedAt: 1 } as never }
})
@@ -487,11 +560,169 @@ describe('useNativeChatLiveSession — transport routing', () => {
transport.emit({
type: 'snapshot',
messages: [user('u-1', 'go'), assistant('a-1', 'done')],
hasMore: false
hasMore: false,
lifecycle: { state: 'completed', turnId: 'turn-1', timestamp: 2 }
})
)
expect(latest?.status).toBe('ready')
})
it('applies a lifecycle-only append after the final message frame', async () => {
useAppStore.setState({
agentStatusByPaneKey: { [PANE]: { state: 'working', stateStartedAt: 1 } as never }
})
const transport = getMockTransport('env-1')
await render({ paneKey: PANE, agent: AGENT, sessionId: SESSION, runtimeEnvironmentId: 'env-1' })
await act(async () =>
transport.emit({
type: 'snapshot',
messages: [user('u-1', 'go'), assistant('a-1', 'done')],
hasMore: false,
lifecycle: { state: 'working', turnId: 'turn-1', timestamp: 1 }
})
)
expect(latest?.status).toBe('working')
await act(async () =>
transport.emit({
type: 'appended',
messages: [],
lifecycle: { state: 'completed', turnId: 'turn-1', timestamp: 2 }
})
)
expect(latest?.status).toBe('ready')
})
it('applies a terminal-side interruption frame without a local Stop action', async () => {
useAppStore.setState({
agentStatusByPaneKey: { [PANE]: { state: 'working', stateStartedAt: 1 } as never }
})
const transport = getMockTransport('env-1')
await render({ paneKey: PANE, agent: AGENT, sessionId: SESSION, runtimeEnvironmentId: 'env-1' })
await act(async () =>
transport.emit({
type: 'snapshot',
messages: [user('u-1', 'go')],
hasMore: false,
lifecycle: { state: 'working', turnId: 'turn-1', timestamp: 1 }
})
)
expect(latest?.status).toBe('working')
await act(async () =>
transport.emit({
type: 'appended',
messages: [],
lifecycle: { state: 'interrupted', turnId: 'turn-1', timestamp: 2 }
})
)
expect(latest?.status).toBe('ready')
})
it('does not let an older pagination read rewind a live completion', async () => {
useAppStore.setState({
agentStatusByPaneKey: { [PANE]: { state: 'working', stateStartedAt: 1 } as never }
})
const transport = getMockTransport('env-1')
const many = Array.from({ length: NATIVE_CHAT_INITIAL_LIMIT }, (_unused, index) =>
assistant(`m-${index}`, 'working')
)
await render({ paneKey: PANE, agent: AGENT, sessionId: SESSION, runtimeEnvironmentId: 'env-1' })
await act(async () =>
transport.emit({
type: 'snapshot',
messages: many,
hasMore: true,
lifecycle: { state: 'working', turnId: 'turn-1', timestamp: 1 }
})
)
let resolveEarlier: (result: {
messages: NativeChatMessage[]
lifecycle: { state: 'working'; turnId: string; timestamp: number }
}) => void = () => {}
transport.readSession.mockImplementationOnce(
() => new Promise((resolve) => (resolveEarlier = resolve))
)
await act(async () => latest?.loadEarlier())
await act(async () =>
transport.emit({
type: 'appended',
messages: [],
lifecycle: { state: 'completed', turnId: 'turn-1', timestamp: 2 }
})
)
expect(latest?.status).toBe('ready')
await act(async () => {
resolveEarlier({
messages: many,
lifecycle: { state: 'working', turnId: 'turn-1', timestamp: 1 }
})
await Promise.resolve()
})
expect(latest?.status).toBe('ready')
})
it('reconciles completion from a reconnect snapshot', async () => {
useAppStore.setState({
agentStatusByPaneKey: { [PANE]: { state: 'working', stateStartedAt: 10 } as never }
})
const transport = getMockTransport('env-1')
await render({ paneKey: PANE, agent: AGENT, sessionId: SESSION, runtimeEnvironmentId: 'env-1' })
await act(async () =>
transport.emit({
type: 'snapshot',
messages: [user('u-1', 'go')],
hasMore: false,
lifecycle: { state: 'working', turnId: 'turn-1', timestamp: 10 }
})
)
expect(latest?.status).toBe('working')
await act(async () =>
transport.emit({
type: 'snapshot',
messages: [user('u-1', 'go'), assistant('a-1', 'done')],
hasMore: false,
lifecycle: { state: 'completed', turnId: 'turn-1', timestamp: 20 }
})
)
expect(latest?.status).toBe('ready')
})
it('reconciles interruption from a reconnect snapshot', async () => {
useAppStore.setState({
agentStatusByPaneKey: { [PANE]: { state: 'working', stateStartedAt: 10 } as never }
})
const transport = getMockTransport('env-1')
await render({ paneKey: PANE, agent: AGENT, sessionId: SESSION, runtimeEnvironmentId: 'env-1' })
await act(async () =>
transport.emit({
type: 'snapshot',
messages: [user('u-1', 'go')],
hasMore: false,
turnLifecycleCapable: true,
lifecycle: { state: 'working', turnId: 'turn-1', timestamp: 10 }
})
)
expect(latest?.status).toBe('working')
await act(async () =>
transport.emit({
type: 'snapshot',
messages: [user('u-1', 'go')],
hasMore: false,
turnLifecycleCapable: true,
lifecycle: { state: 'interrupted', turnId: 'turn-1', timestamp: 20 }
})
)
// The reply (ts 2) post-dates stateStartedAt (1): the stuck 'working' heals.
expect(latest?.status).toBe('ready')
})
})
@@ -1,5 +1,4 @@
import { useCallback, useEffect, useMemo, useRef, useState } from 'react'
import { useAppStore } from '../../store'
import {
NATIVE_CHAT_SOURCE_PRIORITY,
type AgentType,
@@ -23,6 +22,8 @@ import {
nextNativeChatLimit
} from './native-chat-pagination'
import { getNativeChatSessionTransport } from './native-chat-session-transport'
import { useNativeChatTranscriptLifecycle } from './use-native-chat-transcript-lifecycle'
import { useNativeChatHookStatus } from './use-native-chat-hook-status'
export type UseNativeChatLiveSessionArgs = {
/** Composite `${tabId}:${leafId}` key — selects the live hook entry. */
@@ -126,6 +127,8 @@ export function useNativeChatLiveSession(
const [read, setRead] = useState<ReadState>({ phase: 'loading' })
const [hasMore, setHasMore] = useState(false)
const [loadingEarlier, setLoadingEarlier] = useState(false)
const [transcriptLifecycle, turnLifecycleCapable, transcriptLifecycleControl] =
useNativeChatTranscriptLifecycle()
// The active read window; raised by loadEarlier to page in older history.
const limitRef = useRef(NATIVE_CHAT_INITIAL_LIMIT)
@@ -139,14 +142,7 @@ export function useNativeChatLiveSession(
// live frame costs O(incoming), not O(existing) (#18 parity for desktop).
const appendMergerRef = useRef(createNativeChatMerger(NATIVE_CHAT_SOURCE_PRIORITY))
// Live hook state for this pane, selected narrowly so unrelated status churn
// doesn't re-render the chat view.
const hookState = useAppStore((s) => s.agentStatusByPaneKey[paneKey]?.state ?? null)
// When that state began (epoch ms). A separate primitive selector so it doesn't
// churn renders; lets a stale 'working' self-heal once this turn's reply lands.
const hookStateStartedAt = useAppStore(
(s) => s.agentStatusByPaneKey[paneKey]?.stateStartedAt ?? null
)
const [hookState, hookStateStartedAt, hookHasWorkingSubagents] = useNativeChatHookStatus(paneKey)
const latestSessionId = useRef<string | null>(sessionId)
latestSessionId.current = sessionId
@@ -170,6 +166,7 @@ export function useNativeChatLiveSession(
// every source generation must invalidate pagination captured before it.
transcriptEpochRef.current += 1
setLoadingEarlier(false)
transcriptLifecycleControl.reset()
if (!sessionId) {
// No session id yet: nothing to read or tail. Surface live hook state on
// an empty transcript; backfills once the id arrives (effect re-runs).
@@ -225,6 +222,10 @@ export function useNativeChatLiveSession(
return
}
const messages = result?.messages ?? []
transcriptLifecycleControl.replace(
result?.lifecycle,
result?.turnLifecycleCapable === true
)
setRead({ phase: 'ready', messages })
setHasMore(hasMoreNativeChatHistory(messages.length, limitRef.current))
})
@@ -258,12 +259,14 @@ export function useNativeChatLiveSession(
setRead({ phase: 'error', error: frame.error })
return
}
transcriptLifecycleControl.replace(frame.lifecycle, frame.turnLifecycleCapable === true)
replaceList(appendMergerRef.current, frame.messages)
setAppended([])
setRead({ phase: 'ready', messages: appendMergerRef.current.list })
setHasMore(frame.hasMore)
return
}
transcriptLifecycleControl.append(frame.lifecycle, frame.turnLifecycleCapable === true)
// Merge by id (re-emits replace in place) then bound to the window so
// the bucket can't grow without limit. The base read still holds older
// turns, and the assembler re-dedups the concat, so trimming the recent
@@ -296,7 +299,7 @@ export function useNativeChatLiveSession(
}
// `transport` identity changes on an owner flip, re-running this effect to
// tear down the old host's subscription and open one against the new host.
}, [agent, sessionId, transcriptPath, transport])
}, [agent, sessionId, transcriptPath, transport, transcriptLifecycleControl])
const loadEarlier = useCallback(() => {
if (!sessionId || loadingEarlier || !hasMore || read.phase !== 'ready') {
@@ -304,6 +307,7 @@ export function useNativeChatLiveSession(
}
const nextLimit = nextNativeChatLimit(limitRef.current)
const requestEpoch = transcriptEpochRef.current
const lifecycleRevision = transcriptLifecycleControl.revision()
setLoadingEarlier(true)
void transport
.readSession(agent, sessionId, nextLimit, transcriptPath ?? undefined)
@@ -324,6 +328,7 @@ export function useNativeChatLiveSession(
// Read results are an ordered tail — replace the base list so the older
// page prepends in order; live appends stay in their separate bucket.
setRead({ phase: 'ready', messages: result.messages })
transcriptLifecycleControl.replaceFromPagination(result.lifecycle, lifecycleRevision)
setHasMore(hasMoreNativeChatHistory(result.messages.length, nextLimit))
})
.catch(() => {
@@ -339,7 +344,16 @@ export function useNativeChatLiveSession(
setLoadingEarlier(false)
}
})
}, [agent, sessionId, transcriptPath, transport, hasMore, loadingEarlier, read.phase])
}, [
agent,
sessionId,
transcriptPath,
transport,
hasMore,
loadingEarlier,
read.phase,
transcriptLifecycleControl
])
// Assembled messages reuse the incremental assembler across appends. Computed
// outside the status memo: hookState changes only the status override, not the
@@ -383,6 +397,9 @@ export function useNativeChatLiveSession(
agent,
hookState,
stateStartedAt: hookStateStartedAt,
transcriptLifecycle,
turnLifecycleCapable,
hookHasWorkingSubagents,
// Why: a watcher append (fix for #8401) can land content while the read is
// still retrying ('loading') or after it settled into 'error' — in both
// cases showing the live content beats a spinner or a stale error, so each
@@ -398,6 +415,9 @@ export function useNativeChatLiveSession(
agent,
hookState,
hookStateStartedAt,
transcriptLifecycle,
turnLifecycleCapable,
hookHasWorkingSubagents,
hasMore,
loadingEarlier,
loadEarlier,
@@ -0,0 +1,66 @@
import { useCallback, useMemo, useRef, useState } from 'react'
import type { NativeChatTurnLifecycle } from '../../../../shared/native-chat-types'
type TranscriptLifecycleState = {
lifecycle?: NativeChatTurnLifecycle
capable: boolean
}
type TranscriptLifecycleControl = {
reset: () => void
replace: (lifecycle: NativeChatTurnLifecycle | undefined, capable: boolean) => void
append: (lifecycle: NativeChatTurnLifecycle | undefined, capable: boolean) => void
revision: () => number
replaceFromPagination: (lifecycle: NativeChatTurnLifecycle | undefined, revision: number) => void
}
export function useNativeChatTranscriptLifecycle(): readonly [
NativeChatTurnLifecycle | undefined,
boolean,
TranscriptLifecycleControl
] {
const [state, setState] = useState<TranscriptLifecycleState>({ capable: false })
// Why: pagination may resolve after a live completion; its older boundary
// can update history only when no live lifecycle write won the race.
const revisionRef = useRef(0)
const replace = useCallback(
(lifecycle: NativeChatTurnLifecycle | undefined, capable: boolean): void => {
revisionRef.current += 1
setState({ lifecycle, capable })
},
[]
)
const reset = useCallback((): void => replace(undefined, false), [replace])
const append = useCallback(
(lifecycle: NativeChatTurnLifecycle | undefined, capable: boolean): void => {
if (lifecycle) {
revisionRef.current += 1
}
setState((current) => {
if (!lifecycle && current.capable === capable) {
return current
}
return { lifecycle: lifecycle ?? current.lifecycle, capable }
})
},
[]
)
const revision = useCallback((): number => revisionRef.current, [])
const replaceFromPagination = useCallback(
(lifecycle: NativeChatTurnLifecycle | undefined, expectedRevision: number): void => {
if (!lifecycle || revisionRef.current !== expectedRevision) {
return
}
revisionRef.current += 1
setState((current) => ({ ...current, lifecycle }))
},
[]
)
const control = useMemo<TranscriptLifecycleControl>(
() => ({ reset, replace, append, revision, replaceFromPagination }),
[append, replace, replaceFromPagination, reset, revision]
)
return [state.lifecycle, state.capable, control]
}
@@ -702,6 +702,91 @@ describe('web settings preload API', () => {
})
})
describe('web native chat preload API', () => {
beforeEach(() => {
vi.resetModules()
})
afterEach(() => {
vi.unstubAllGlobals()
vi.doUnmock('./web-runtime-client')
})
it('forwards validated lifecycle metadata from reads and stream frames', async () => {
const lifecycle = { state: 'completed', turnId: 'turn-1', timestamp: 42 } as const
const message = {
id: 'a-1',
role: 'assistant' as const,
blocks: [{ type: 'text' as const, text: 'done' }],
timestamp: 42,
source: 'transcript' as const
}
vi.doMock('./web-runtime-client', () => ({
WebRuntimeClient: class {
call(): Promise<RuntimeRpcResponse<unknown>> {
return Promise.resolve({
id: 'read-1',
ok: true,
result: { messages: [message], turnLifecycleCapable: true, lifecycle },
_meta: { runtimeId: 'runtime-1' }
})
}
subscribe(
_method: string,
_params: unknown,
callbacks: { onResponse: (response: RuntimeRpcResponse<unknown>) => void }
): Promise<{ unsubscribe: () => void }> {
callbacks.onResponse({
id: 'stream-1',
ok: true,
result: {
type: 'snapshot',
messages: [message],
hasMore: false,
turnLifecycleCapable: true,
lifecycle
},
_meta: { runtimeId: 'runtime-1' }
})
return Promise.resolve({ unsubscribe: vi.fn() })
}
close(): void {}
}
}))
const globals = installBrowserGlobals('Linux')
writeStoredRuntimeEnvironment(globals.storage)
const { installWebPreloadApi } = await import('./web-preload-api')
installWebPreloadApi()
await expect(globals.window.api.nativeChat.readSession('claude', 'session-1')).resolves.toEqual(
{
messages: [message],
turnLifecycleCapable: true,
lifecycle
}
)
const frames: unknown[] = []
globals.window.api.nativeChat.subscribe(
{ subscriptionId: 'sub-1', agent: 'claude', sessionId: 'session-1' },
(frame) => frames.push(frame)
)
await Promise.resolve()
expect(frames).toEqual([
{
type: 'snapshot',
messages: [message],
hasMore: false,
turnLifecycleCapable: true,
lifecycle
}
])
})
})
describe('web MiniMax preload API', () => {
beforeEach(() => {
vi.resetModules()
+32 -12
View File
@@ -6,7 +6,6 @@ import type {
PreflightStatus,
RefreshAgentsResult,
NativeChatApi,
NativeChatReadSessionResult,
NativeChatAppendedMessages
} from '../../../preload/api-types'
import type { RuntimeRpcResponse } from '../../../shared/runtime-rpc-envelope'
@@ -127,6 +126,10 @@ import {
import { normalizeContextualTourIds, type ContextualTourId } from '../../../shared/contextual-tours'
import { translate } from '@/i18n/i18n'
import { getDefaultCreateProjectParent } from '@/components/sidebar/create-project-defaults'
import {
parseRuntimeNativeChatReadSessionResult,
parseRuntimeNativeChatTurnLifecycle
} from '@/components/native-chat/native-chat-runtime-contract'
const SETTINGS_STORAGE_KEY = 'orca.web.settings.v1'
const UI_STORAGE_KEY = 'orca.web.ui.v1'
@@ -1084,13 +1087,15 @@ function createWebKeybindingsApi(): WebKeybindingsApi {
// undefined on web and the chat view showed no messages.
function createNativeChatApi(): NativeChatApi {
return {
readSession: (agent, sessionId, limit, transcriptPath) =>
callRuntimeResult<NativeChatReadSessionResult>('nativeChat.readSession', {
agent,
sessionId,
limit,
transcriptPath
}),
readSession: async (agent, sessionId, limit, transcriptPath) =>
parseRuntimeNativeChatReadSessionResult(
await callRuntimeResult<unknown>('nativeChat.readSession', {
agent,
sessionId,
limit,
transcriptPath
})
),
subscribe: (args, onFrame) => {
// No paired runtime yet: nothing to subscribe to, and
// requireActiveEnvironment() would throw. Return a no-op teardown so the
@@ -1144,7 +1149,11 @@ function createNativeChatApi(): NativeChatApi {
messages?: NativeChatAppendedMessages
hasMore?: boolean
error?: string
lifecycle?: unknown
turnLifecycleCapable?: unknown
}
const lifecycle = parseRuntimeNativeChatTurnLifecycle(result?.lifecycle)
const turnLifecycleCapable = result?.turnLifecycleCapable === true
if (
(result?.type === 'appended' ||
result?.type === 'snapshot' ||
@@ -1157,14 +1166,18 @@ function createNativeChatApi(): NativeChatApi {
type: 'snapshot',
messages: result.messages,
hasMore: result.hasMore ?? result.messages.length >= (args.limit ?? 300),
...(result.error ? { error: result.error } : {})
turnLifecycleCapable,
...(result.error ? { error: result.error } : {}),
...(lifecycle ? { lifecycle } : {})
})
} else if (result.type === 'snapshot') {
onFrame({
type: 'snapshot',
messages: result.messages,
hasMore: result.hasMore ?? false,
...(result.error ? { error: result.error } : {})
turnLifecycleCapable,
...(result.error ? { error: result.error } : {}),
...(lifecycle ? { lifecycle } : {})
})
} else {
onFrame(
@@ -1172,9 +1185,16 @@ function createNativeChatApi(): NativeChatApi {
? {
type: 'replacement',
messages: result.messages,
hasMore: result.hasMore ?? false
hasMore: result.hasMore ?? false,
turnLifecycleCapable,
...(lifecycle ? { lifecycle } : {})
}
: {
type: 'appended',
messages: result.messages,
turnLifecycleCapable,
...(lifecycle ? { lifecycle } : {})
}
: { type: 'appended', messages: result.messages }
)
}
} else if (!receivedInitial) {
+16
View File
@@ -79,6 +79,22 @@ export type NativeChatMessage = {
turnId?: string
}
export const NATIVE_CHAT_TURN_LIFECYCLE_STATES = ['working', 'completed', 'interrupted'] as const
export type NativeChatTurnLifecycleState = (typeof NATIVE_CHAT_TURN_LIFECYCLE_STATES)[number]
export const NATIVE_CHAT_INTERRUPTED_STATUS_TEXT = 'Conversation interrupted'
/** A provider-authored turn boundary recovered from the transcript itself.
* Unlike assistant prose, this is explicit lifecycle evidence (completion or
* interruption records) and is safe to replay. */
export type NativeChatTurnLifecycle = {
state: NativeChatTurnLifecycleState
/** Stable provider id when available, otherwise the JSONL record position. */
turnId: string
/** Provider timestamp; null only when the transcript omitted one. */
timestamp: number | null
}
export const NATIVE_CHAT_SESSION_STATUSES = [
'loading',
'ready',