fix(chat): keep streaming replies after their prompts (#37773)

* fix(chat): keep streaming replies after their prompts

* fix(chat): retain queued placement through turn transitions

* fix(chat): preserve queued sends through read errors

* test(chat): keep act callbacks void

* test(chat): align transcript tiers with prompt order
This commit is contained in:
Kelvin Amoaba
2026-11-03 10:15:37 +00:00
committed by GitHub
parent ee532e571c
commit debbb5b496
22 changed files with 448 additions and 86 deletions
@@ -27,6 +27,7 @@ export type MobileNativeChatSendOrigin = {
'afterMessageId' | 'afterMessageTimestamp' | 'afterEmptyTranscriptSessionId'
>
sentAt: number
queuedDuringWorkingEpoch?: number | null
}
type PendingByKey = Record<string, MobileNativeChatPendingMessage[]>
@@ -55,6 +56,7 @@ export function appendMobileNativeChatPending(
id,
text,
sentAt: origin.sentAt,
queuedDuringWorkingEpoch: origin.queuedDuringWorkingEpoch,
...origin.boundary,
...(images?.length ? { images } : {})
})
@@ -78,6 +78,13 @@ function build(
}
describe('buildMobileNativeChatTransientData', () => {
it('places an idle send before the streaming reply it caused', () => {
const data = build([assistant('a1', 'Earlier answer')], 'Starting the tests', [
{ id: 'p1', text: 'Run the tests', sentAt: 100 }
])
expect(data.map((message) => message.id)).toEqual(['a1', 'p1', 'streaming'])
})
it('appends pending optimistic messages at the tail as user turns', () => {
const data = build([assistant('a1', 'hello')], null, [{ id: 'p1', text: 'queued' }])
const last = data[data.length - 1]
@@ -240,8 +247,10 @@ describe('buildMobileNativeChatTransientData', () => {
expect(result.data[0]?.blocks).toEqual([{ type: 'image-ref', url: 'file:///phone-photo.jpg' }])
})
it('appends a synthetic bubble for gated streaming text, between transcript and pending', () => {
const data = build([user('u1', 'hi')], 'thinking out loud', [{ id: 'p1', text: 'queued' }])
it('keeps a queued follow-up after the current streaming reply', () => {
const data = build([user('u1', 'hi')], 'thinking out loud', [
{ id: 'p1', text: 'queued', pendingSend: 'queued' }
])
expect(data.map((message) => message.id)).toEqual(['u1', 'streaming', 'p1'])
expect(data[1].blocks).toEqual([{ type: 'text', text: 'thinking out loud' }])
})
@@ -18,6 +18,7 @@ import {
} from '../../../../src/shared/native-chat/native-chat-types'
import type { AgentType } from '../../../../src/shared/agent-status-types'
import type { MobileNativeChatStatus } from './use-mobile-native-chat-session'
import { orderNativeChatDisplayMessages } from '../../../../src/shared/native-chat/transcript/native-chat-message-order'
import { translate } from '../../i18n/mobile-i18n'
function mobileNativeChatEmptyStateCopy(
@@ -91,6 +92,7 @@ export type MobileNativeChatPendingItem = {
text: string
images?: string[]
sentAt?: number
pendingSend?: NativeChatMessage['pendingSend']
}
export function foldMobileNativeChatMessages(
@@ -146,13 +148,12 @@ export function buildMobileNativeChatTransientData({
}
return { ...message, blocks }
})
// Desktop order: transcript rows, then the streaming preview, then unmatched echoes in send order.
const data: NativeChatMessage[] = [...renderedFolded]
const transient: NativeChatMessage[] = []
if (streaming) {
data.push(nativeChatStreamingMessage(streaming))
transient.push(nativeChatStreamingMessage(streaming))
}
for (const item of pending) {
data.push({
transient.push({
id: item.id,
role: 'user',
blocks: [
@@ -160,8 +161,13 @@ export function buildMobileNativeChatTransientData({
...(item.images ?? []).map((uri) => ({ type: 'image-ref' as const, url: uri }))
],
timestamp: item.sentAt ?? null,
pendingSend: item.pendingSend ?? 'active',
source: 'transcript'
})
}
return { folded: renderedFolded, streaming, data }
return {
folded: renderedFolded,
streaming,
data: [...renderedFolded, ...orderNativeChatDisplayMessages(transient)]
}
}
@@ -14,7 +14,12 @@ const restoreRejectedDraft = vi.fn()
const holdUnconfirmedSend = vi.fn()
const viewMode = { isTabChatView: (_tabId: string) => true }
let draftPending: { id: string }[] = []
let draftPending: {
id: string
text?: string
sentAt?: number
queuedDuringWorkingEpoch?: number | null
}[] = []
const sessionState = { messages: [] as unknown[], status: 'ready', transcriptLoading: false }
const structuredSendWithOutcome = vi.fn()
const structuredCancel = vi.fn()
@@ -639,6 +644,8 @@ describe('useMobileNativeChatController terminal turn', () => {
viewMode.isTabChatView = () => true
workingTab.agentStatus.workingMode = undefined
workingTab.agentStatus.stateStartedAt = turnStartedAt
workingTab.agentStatus.turnStartedAt = turnStartedAt
workingTab.agentStatus.lastAssistantMessage = 'Partial reply'
act(() => {
renderer = create(createElement(Harness))
})
@@ -653,6 +660,74 @@ describe('useMobileNativeChatController terminal turn', () => {
vi.useRealTimers()
})
it.each([turnStartedAt, undefined])(
'keeps queued echoes behind the answer through state changes and Stop, with turn epoch %s',
(epoch) => {
if (epoch === undefined) {
Reflect.deleteProperty(workingTab.agentStatus, 'turnStartedAt')
}
const originalMessages = sessionState.messages
sessionState.messages = [
{
id: 'active-prompt',
role: 'user',
blocks: [{ type: 'text', text: 'Run tests' }],
source: 'transcript',
timestamp: null
}
]
draftPending = [
{ id: 'follow-up', text: 'Then docs', sentAt: 1, queuedDuringWorkingEpoch: epoch ?? null }
]
try {
rerender()
expect(controller?.chatPending[0]).toMatchObject({ pendingSend: 'queued' })
workingTab.agentStatus.stateStartedAt += 1000
rerender()
expect(controller?.chatPending[0]).toMatchObject({ pendingSend: 'queued' })
act(() => controller?.handleNativeChatStop())
expect(controller?.nativeChatCanStop).toBe(false)
expect(controller?.nativeChatStreamingText).toBe('Partial reply')
expect(controller?.chatPending[0]).toMatchObject({ pendingSend: 'queued' })
} finally {
draftPending = []
sessionState.messages = originalMessages
}
}
)
it('captures an unknown active turn while an older host transcript is still empty', async () => {
Reflect.deleteProperty(workingTab.agentStatus, 'turnStartedAt')
workingTab.agentStatus.lastAssistantMessage = ''
rerender()
expect(controller?.nativeChatTurnActive).toBe(false)
captureSendOrigin.mockReturnValue(ORIGIN)
clientStub.sendRequest.mockResolvedValue({ ok: true, result: { send: { accepted: true } } })
sendWithOutcome.mockResolvedValueOnce('accepted')
await act(async () => {
await controller?.handleNativeChatSend('Then docs')
})
expect(acceptSend).toHaveBeenLastCalledWith(
expect.objectContaining({ queuedDuringWorkingEpoch: null }),
'Then docs',
undefined
)
})
it('captures the turn epoch when sending a follow-up before its acknowledgement', async () => {
captureSendOrigin.mockReturnValue(ORIGIN)
clientStub.sendRequest.mockResolvedValue({ ok: true, result: { send: { accepted: true } } })
sendWithOutcome.mockResolvedValueOnce('accepted')
await act(async () => {
await controller?.handleNativeChatSend('Then docs')
})
expect(acceptSend).toHaveBeenLastCalledWith(
expect.objectContaining({ queuedDuringWorkingEpoch: turnStartedAt }),
'Then docs',
undefined
)
})
it('treats passive monitoring as outside the foreground turn', () => {
workingTab.agentStatus.workingMode = 'monitoring'
rerender()
@@ -685,7 +760,7 @@ describe('useMobileNativeChatController terminal turn', () => {
act(() => controller?.handleNativeChatStop())
expect(controller?.nativeChatCanStop).toBe(false)
draftPending = [{ id: 'pending-1' }]
draftPending = [{ id: 'pending-1', text: 'New prompt', sentAt: turnStartedAt }]
try {
rerender()
expect(controller?.nativeChatCanStop).toBe(true)
@@ -1,5 +1,5 @@
import { asyncQuestionSourceScope } from '../../../../src/shared/native-chat/input/native-chat-ask'
import { useLayoutEffect, useRef, type MutableRefObject } from 'react'
import { useCallback, useLayoutEffect, useMemo, useRef, type MutableRefObject } from 'react'
import type { RpcClient } from '../../transport/rpc-client'
import type { ConnectionState } from '../../transport/types'
import type { MobileNativeChatTab } from './mobile-native-chat-eligibility'
@@ -15,6 +15,7 @@ import type { MobileNativeChatController } from './mobile-native-chat-controller
import { useMobileBridgeChatPromptWrites } from '../use-mobile-bridge-chat-prompt-writes'
import { useMobileNativeChatActiveResolution } from './use-mobile-native-chat-active-resolution'
import { useMobileNativeChatPromptController } from '../use-mobile-native-chat-prompt-controller'
import { nativeChatPendingSendPlacement } from '../../../../src/shared/native-chat/transcript/native-chat-pending-sends'
import { mobileNativeChatScopeKey } from './mobile-native-chat-scope-key'
export function useMobileNativeChatController(args: {
@@ -153,12 +154,29 @@ export function useMobileNativeChatController(args: {
resolution: activeChatResolution,
session: activeChatStructured ? null : nativeChatSession,
latestPendingId: chatPending.at(-1)?.id ?? null,
pending: chatPending,
hasPromptCard,
scopeKey: streamIdentity,
stop: handleNativeChatStop
})
const activeTurn = activeChatStructured ? structuredNativeChat : terminalTurn
const nativeChatAgentWorking = activeTurn.isWorking
const workingEpoch = terminalTurn.workingEpoch
const placedChatPending = useMemo(
() =>
chatPending.map((entry) => ({
...entry,
pendingSend: nativeChatPendingSendPlacement(entry, workingEpoch)
})),
[chatPending, workingEpoch]
)
const captureTerminalSendOrigin = useCallback(
(text: string, sentAt?: number) => {
const origin = captureSendOrigin(text, sentAt)
return origin ? { ...origin, queuedDuringWorkingEpoch: workingEpoch } : null
},
[captureSendOrigin, workingEpoch]
)
const workspaceCatalog = useMobileNativeChatWorkspaceCatalog(args, activeChatAgent)
@@ -176,7 +194,7 @@ export function useMobileNativeChatController(args: {
deviceTokenRef,
agentRef: activeChatAgentRef,
commandSendRef: recordSessionOptionCommandRef,
captureSendOrigin,
captureSendOrigin: captureTerminalSendOrigin,
readSeededLaunchDraftSeed,
clearDraftForSend,
restoreRejectedDraft,
@@ -239,7 +257,7 @@ export function useMobileNativeChatController(args: {
chatDraftKey,
setChatComposerText,
getChatComposerEditGeneration,
chatPending,
chatPending: placedChatPending,
chatImagePreviewsByMessageId: activeChatStructured
? structuredNativeChat.imagePreviewsByMessageId
: chatImagePreviewsByMessageId,
@@ -22,11 +22,17 @@ import { useThrottledLatestValue } from './use-throttled-latest-value'
import type { MobileNativeChatSession } from './native-chat/use-mobile-native-chat-session'
import type { MobileNativeChatResolution } from './native-chat/mobile-native-chat-eligibility'
import {
pendingSendsAsMessages,
type NativeChatPendingSend
} from '../../../src/shared/native-chat/transcript/native-chat-pending-sends'
const NO_MESSAGES: NativeChatMessage[] = []
export type MobileNativeChatTerminalTurn = NativeChatTerminalTurn & {
streamingText: string | undefined
workingStartedAt: number | null
workingEpoch: number | null | undefined
settledTurns: NativeChatSettledTurns
/** Stop hides working until the turn ends or a newer working epoch starts. */
stop: () => void
@@ -41,6 +47,7 @@ export function useMobileNativeChatTerminalTurn(args: {
session: Pick<MobileNativeChatSession, 'messages' | 'transcriptLifecycle'> | null
/** The newest pending echo; a new one is a new send, which ends Stop's suppression like desktop. */
latestPendingId: string | null
pending?: NativeChatPendingSend[]
hasPromptCard: boolean
scopeKey: string
stop: () => Promise<boolean>
@@ -61,11 +68,17 @@ export function useMobileNativeChatTerminalTurn(args: {
hookHasWorkingSubagents: nativeChatHookHasWorkingSubagents(status)
})
const liveWorking = live.status === 'working'
const turnStartedAt = nativeChatHookTurnStartedAt(status ?? undefined)
const workingEpoch =
liveWorking || live.hookAwaitingInput ? (status?.turnStartedAt ?? null) : undefined
const messagesWithPending = useMemo(
() => [...messages, ...pendingSendsAsMessages(args.pending ?? [], messages, workingEpoch)],
[messages, args.pending, workingEpoch]
)
// Throttle status-part previews to avoid reparsing the whole accumulated markdown per part.
const streamingText = useThrottledLatestValue(
deriveNativeChatStreamingText({
// A pending echo is the transcript's tail on desktop, so no earlier reply can hide the preview.
messages: hasPending ? NO_MESSAGES : messages,
messages: messagesWithPending,
previewText: status?.lastAssistantMessage,
working: liveWorking,
previewIsToolOutput: status?.lastAssistantMessageIsToolOutput === true
@@ -114,7 +127,6 @@ export function useMobileNativeChatTerminalTurn(args: {
hasPromptCard
})
const turnStartedAt = nativeChatHookTurnStartedAt(status ?? undefined)
const latestWorkedSeconds = nativeChatHookLatestTurnWorkedSeconds(
status ?? undefined,
hookState === null
@@ -136,6 +148,7 @@ export function useMobileNativeChatTerminalTurn(args: {
...turn,
streamingText,
workingStartedAt: turn.turnActive ? turnStartedAt : null,
workingEpoch,
settledTurns,
stop: stopTurn
}
@@ -12,6 +12,8 @@ import type {
NativeChatComposerProps
} from './native-chat-composer-types'
import type * as RuntimeTerminalInspectionModule from '@/runtime/runtime-terminal-inspection'
import { mergeNativeChatLiveSession } from '../../../../shared/native-chat/native-chat-live-status'
import { clearPendingSendCacheForTests } from './native-chat-pending'
type RuntimeTerminalInspection = typeof RuntimeTerminalInspectionModule
@@ -177,6 +179,45 @@ afterEach(() => {
}
})
describe('queued follow-ups through transcript recovery', () => {
it.each(['error', 'ready'] as const)(
'keeps a send after the current reply when a read error is presented as %s',
async (status) => {
vi.useFakeTimers({ toFake: ['Date', 'setTimeout', 'clearTimeout'] })
clearPendingSendCacheForTests()
setStatus({ state: 'working', lastAssistantMessage: 'Current reply' })
const input = {
messages: [userTurn],
sessionId: 'session-hidden',
agent: 'claude' as const,
hookState: 'working' as const,
stateStartedAt: Date.now()
}
retained.session = {
...retained.session!,
...mergeNativeChatLiveSession({ ...input, error: 'Read failed' }),
status,
readPhase: 'error'
}
renderPane()
await act(async () => {
fireEvent.click(screen.getByTestId('composer-send'))
})
act(() => {
retained.session = {
...retained.session!,
...mergeNativeChatLiveSession(input),
readPhase: 'ready'
}
setStatus({ state: 'working', lastAssistantMessage: 'Current reply continues' })
})
const text = document.body.textContent!
expect(text).toContain('Current')
expect(text.indexOf(MESSAGE)).toBeGreaterThan(text.indexOf('Current'))
}
)
})
describe('a prompt card hides the composer without unmounting it', () => {
it('keeps an in-flight message visible and never submits it into the approval', async () => {
vi.useFakeTimers({ toFake: ['Date', 'setTimeout', 'clearTimeout'] })
@@ -102,6 +102,9 @@ export function NativeChatResolvedView({
)
// A coalesced interrupt/restart can stay working; the newer epoch must still clear Stop suppression.
const workingEpoch = useAppStore((s) => s.agentStatusByPaneKey[paneKey]?.stateStartedAt ?? null)
const turnEpoch = useAppStore((s) => s.agentStatusByPaneKey[paneKey]?.turnStartedAt ?? null)
const sendWorkingEpoch =
(session.hookWorking ?? liveWorking) || session.hookAwaitingInput ? turnEpoch : undefined
const canSend = useNativeChatCanSend(targetPtyId)
// Share verified composer delivery with prompt answers and ESC-based Stop.
const send = useNativeChatInteractiveSend(paneKey, targetPtyId, agent)
@@ -121,7 +124,12 @@ export function NativeChatResolvedView({
() => ({ paneKey, agent, sessionId }),
[paneKey, agent, sessionId]
)
const delivery = useNativeChatPendingDelivery({ paneKey, agent, session })
const delivery = useNativeChatPendingDelivery({
paneKey,
agent,
session,
workingEpoch: sendWorkingEpoch
})
const { pending, record, clear } = delivery
// Slash commands are local session-scoped feedback, not user turns; cap and cache their markers.
const [commandMarkers, setCommandMarkers] = useState<NativeChatCommandMarker[]>(() =>
@@ -210,17 +218,13 @@ export function NativeChatResolvedView({
composerReady
})
// Match mobile ordering: hook preview before optimistic user echoes.
const pendingMessages = useMemo(
() => pendingSendsAsMessages(pending, sessionAfterCommandBoundaries.messages),
[pending, sessionAfterCommandBoundaries.messages]
() => pendingSendsAsMessages(pending, sessionAfterCommandBoundaries.messages, sendWorkingEpoch),
[pending, sessionAfterCommandBoundaries.messages, sendWorkingEpoch]
)
const streamingText = useMemo(() => {
return deriveNativeChatStreamingText({
messages:
pendingMessages.length > 0
? [...sessionAfterCommandBoundaries.messages, ...pendingMessages]
: sessionAfterCommandBoundaries.messages,
messages: [...sessionAfterCommandBoundaries.messages, ...pendingMessages],
previewText: hookPreview,
working: liveWorking,
previewIsToolOutput: hookPreviewIsToolOutput
@@ -22,6 +22,8 @@ export type NativeChatLiveSession = NativeChatSession & {
transcriptLifecycle?: NativeChatTurnLifecycle
/** Hook-reported human wait not yet ended by the transcript; absent on hookless lanes. */
hookAwaitingInput?: boolean
/** Reconciled host work before read errors or retained-history presentation override status. */
hookWorking?: boolean
/** True when an older page may still exist (the last read filled the window). */
hasMore: boolean
loadingEarlier: boolean
@@ -2,6 +2,7 @@ import { describe, it, expect } from 'vitest'
import type { NativeChatMessage } from '../../../../shared/native-chat/native-chat-types'
import { orderNativeChatMessages } from './native-chat-message-grouping'
import { NATIVE_CHAT_STREAMING_ID } from '../../../../shared/native-chat/native-chat-streaming'
import { sortForImageNormalization } from './native-chat-session-assembler'
function msg(
overrides: Partial<NativeChatMessage> & Pick<NativeChatMessage, 'id'>
@@ -16,12 +17,21 @@ function msg(
}
describe('orderNativeChatMessages', () => {
it('sorts the streaming preview after real content but before optimistic echoes', () => {
const ordered = orderNativeChatMessages([
it('sorts the streaming preview after the prompt that caused it', () => {
const normalized = sortForImageNormalization([
msg({ id: 'pending:abc', role: 'user', timestamp: 20, source: 'scrape' }),
msg({ id: NATIVE_CHAT_STREAMING_ID, timestamp: null }),
msg({ id: 'real-user', role: 'user', timestamp: 10 })
msg({ id: 'real-user', role: 'user', timestamp: 10, transcriptOrder: 1 }),
msg({ id: 'pending:first-queued', role: 'user', timestamp: 30, pendingSend: 'queued' }),
msg({ id: 'pending:next-queued', role: 'user', timestamp: 2, pendingSend: 'queued' })
])
const ordered = orderNativeChatMessages([...normalized])
expect(ordered.map((m) => m.id)).toEqual([
'real-user',
'pending:abc',
'streaming',
'pending:first-queued',
'pending:next-queued'
])
expect(ordered.map((m) => m.id)).toEqual(['real-user', 'streaming', 'pending:abc'])
})
})
@@ -105,6 +105,4 @@ export function nextNativeChatPendingSendId(now = Date.now()): string {
return `${now}-${pendingSendCounter}`
}
export function isLaunchPromptMessageId(id: string): boolean {
return id.startsWith('launch-pending:')
}
export { isLaunchPromptMessageId } from '../../../../shared/native-chat/transcript/native-chat-pending-sends'
@@ -6,18 +6,18 @@ import {
type NativeChatSessionStatus
} from '../../../../shared/native-chat/native-chat-types'
import type { AgentType } from '../../../../shared/agent-status-types'
import { NATIVE_CHAT_STREAMING_ID } from '../../../../shared/native-chat/native-chat-streaming'
import {
compareNativeChatTranscriptMessages,
orderNativeChatTranscriptEntries
} from '../../../../shared/native-chat/transcript/native-chat-transcript-projection'
compareNativeChatDisplayMessages as compareMessages,
orderNativeChatDisplayEntries as orderEntries,
orderNativeChatDisplayMessages as orderMessages
} from '../../../../shared/native-chat/transcript/native-chat-message-order'
import {
hasImagePromptMarker,
isImageSourceUserTurn,
normalizeImageTranscriptMessages
} from '../../../../shared/native-chat/transcript/native-chat-image-transcript-markers'
import { isLaunchPromptMessageId } from './native-chat-pending'
import { isPendingMessageId } from '../../../../shared/native-chat/transcript/native-chat-pending-sends'
export { compareMessages, orderMessages }
/** For the same turn, prefer transcript over hook over scrape. */
export type NativeChatSources = {
@@ -79,39 +79,6 @@ function supersedes(candidate: NativeChatMessage, existing: NativeChatMessage):
return candidateRank > existingRank
}
// Sort real content before streaming preview before optimistic/queued sends, regardless of timestamps.
function messageSortRank(message: NativeChatMessage): number {
if (message.id === NATIVE_CHAT_STREAMING_ID) {
return 1
}
if (message.queued || isPendingMessageId(message.id) || isLaunchPromptMessageId(message.id)) {
return 2
}
return 0
}
export function compareMessages(a: NativeChatMessage, b: NativeChatMessage): number {
const ar = messageSortRank(a)
const br = messageSortRank(b)
if (ar !== br) {
return ar - br
}
return compareNativeChatTranscriptMessages(a, b)
}
function orderEntries<T>(entries: readonly T[], messageOf: (entry: T) => NativeChatMessage): T[] {
return [0, 1, 2].flatMap((rank) => {
const tier = entries.filter((entry) => messageSortRank(messageOf(entry)) === rank)
return rank === 0
? orderNativeChatTranscriptEntries(tier, messageOf)
: tier.sort((a, b) => compareMessages(messageOf(a), messageOf(b)))
})
}
export function orderMessages(messages: readonly NativeChatMessage[]): NativeChatMessage[] {
return orderEntries(messages, (message) => message)
}
export function assembleNativeChatSession(
input: AssembleNativeChatSessionInput
): NativeChatSession {
@@ -42,6 +42,18 @@ afterEach(() => {
})
describe('terminal Chat pending delivery', () => {
it('records the host working epoch before delivery can change the live status', () => {
const { result } = renderHook(() =>
useNativeChatPendingDelivery({
...args,
session: { messages: [boundary] },
workingEpoch: 500
})
)
act(() => result.current.record('follow up'))
expect(result.current.pending[0]?.queuedDuringWorkingEpoch).toBe(500)
})
it('records a send into a read empty transcript as after every row of that session', () => {
const { result } = renderHook(() =>
useNativeChatPendingDelivery({
@@ -27,6 +27,7 @@ const NO_NOTICES: ReadonlyMap<string, NativeChatDeliveryNotice> = new Map()
export function useNativeChatPendingDelivery(args: {
paneKey: string
agent: AgentType
workingEpoch?: number | null
/** `readPhase` 'ready' with a `sessionId` means an empty `messages` was a real read. */
session: {
messages: NativeChatMessage[]
@@ -35,7 +36,7 @@ export function useNativeChatPendingDelivery(args: {
}
}) {
const languageRevision = useUiLanguageRevision()
const { paneKey, agent } = args
const { paneKey, agent, workingEpoch } = args
const { messages } = args.session
const sessionId = args.session.sessionId ?? null
// Without a sessionId, ready is an unread placeholder.
@@ -70,6 +71,7 @@ export function useNativeChatPendingDelivery(args: {
id: nextNativeChatPendingSendId(sentAt),
text,
sentAt,
...(workingEpoch !== undefined ? { queuedDuringWorkingEpoch: workingEpoch } : {}),
afterMessageId: boundary?.id ?? null,
afterMessageTimestamp: boundary?.timestamp ?? null,
...(readEmptySessionId ? { afterEmptyTranscriptSessionId: readEmptySessionId } : {}),
@@ -78,7 +80,7 @@ export function useNativeChatPendingDelivery(args: {
setPending(appendPendingSendCache(scope, entry))
return entry.id
},
[messages, readEmptySessionId, scope]
[messages, readEmptySessionId, scope, workingEpoch]
)
const cancel = useCallback(
(id: string) => save((entries) => entries.filter((entry) => entry.id !== id)),
@@ -51,6 +51,7 @@ describe('mergeNativeChatLiveSession', () => {
transcriptLifecycle: { state: 'completed', turnId: 'turn-1', timestamp: 2 }
})
expect(session.status).toBe('ready')
expect(session.hookWorking).toBe(false)
})
it('does not apply an older completion marker to a newer working turn', () => {
@@ -186,11 +187,12 @@ describe('mergeNativeChatLiveSession', () => {
messages: [],
sessionId: 'sess',
agent: 'claude',
hookState: null,
hookState: 'working',
error: 'unreadable'
})
expect(errored.status).toBe('error')
expect(errored.error).toBe('unreadable')
expect(errored.hookWorking).toBe(true)
})
})
@@ -31,7 +31,10 @@ export type NativeChatLiveMergeInput = {
error?: string
}
export type NativeChatLiveMergedSession = NativeChatSession & { hookAwaitingInput: boolean }
export type NativeChatLiveMergedSession = NativeChatSession & {
hookAwaitingInput: boolean
hookWorking: boolean
}
/** Errors win; working hooks override loading until their turn ends, with provider lifecycle settling dropped Stop hooks. */
export function mergeNativeChatLiveSession(
@@ -54,10 +57,6 @@ export function mergeNativeChatLiveSession(
stateStartedAt,
transcriptLifecycle
)
if (error) {
return { messages, status: 'error', sessionId, agent, error, hookAwaitingInput }
}
const status = liveStatusOverride(
agent,
hookState,
@@ -66,16 +65,20 @@ export function mergeNativeChatLiveSession(
transcriptLifecycle,
hookHasWorkingSubagents ?? false
)
const activity = { hookAwaitingInput, hookWorking: status === 'working' }
if (error) {
return { messages, status: 'error', sessionId, agent, error, ...activity }
}
// Keep Stop/activity live while loading; view-state selection separately holds empty transcripts on their loading surface.
if (loading && status !== 'working') {
return { messages, status: 'loading', sessionId, agent, hookAwaitingInput }
return { messages, status: 'loading', sessionId, agent, ...activity }
}
return {
messages,
status: status ?? (messages.length === 0 ? 'empty' : 'ready'),
sessionId,
agent,
hookAwaitingInput
...activity
}
}
@@ -23,6 +23,9 @@ export function latestNativeChatReplyText(
): string {
for (let index = messages.length - 1; index > after; index -= 1) {
const message = messages[index]!
if (message.queued || message.pendingSend === 'queued') {
continue
}
if (message.role === 'user') {
return ''
}
@@ -41,7 +44,7 @@ export function deriveNativeChatStreamingText(args: {
working: boolean
previewIsToolOutput?: boolean
}): string | null {
const { messages, previewText, working, previewIsToolOutput } = args
const { previewText, working, previewIsToolOutput } = args
if (!working || previewIsToolOutput) {
return null
}
@@ -49,7 +52,10 @@ export function deriveNativeChatStreamingText(args: {
if (!text) {
return null
}
const lastText = assistantText(messages.at(-1))
const messages = args.messages
const lastText = assistantText(
messages.findLast((message) => !message.queued && message.pendingSend !== 'queued')
)
if (lastText.includes(text) || text.length <= lastText.length) {
return null
}
@@ -238,6 +238,8 @@ export type NativeChatMessage = AgentJournalProducerLinkage & {
agentHandoff?: true
/** Accepted but not yet handed to the agent: drawn after everything the agent has done. */
queued?: true
/** Client-only terminal echo placement relative to the current streaming reply. */
pendingSend?: 'active' | 'queued'
/** Unsent rows belong to no turn. Draw at journal position, or after the conversation when unplaced. */
unsent?: true
/** This client's send, made while the chat read Stopping, that the host has not recorded yet. */
@@ -0,0 +1,119 @@
import { describe, expect, it } from 'vitest'
import type { NativeChatMessage } from '../native-chat-types'
import { deriveNativeChatStreamingText, nativeChatStreamingMessage } from '../native-chat-streaming'
import { orderNativeChatDisplayMessages } from './native-chat-message-order'
import { pendingSendsAsMessages, type NativeChatPendingSend } from './native-chat-pending-sends'
const answer: NativeChatMessage = {
id: 'old-answer',
role: 'assistant',
blocks: [{ type: 'text', text: 'Earlier answer that is longer than the new reply' }],
timestamp: 9_000_000,
source: 'transcript',
transcriptOrder: 1
}
const active: NativeChatPendingSend = {
id: 'active',
text: 'Run the tests',
sentAt: 10,
afterMessageId: answer.id
}
const queued: NativeChatPendingSend = {
id: 'follow-up',
text: 'Then update the docs',
sentAt: 1,
afterMessageId: answer.id,
queuedDuringWorkingEpoch: 500
}
const preview = nativeChatStreamingMessage('Starting the tests')
function tail(
messages: NativeChatMessage[],
pending: NativeChatPendingSend[],
epoch: number | null
) {
const echoes = pendingSendsAsMessages(pending, messages, epoch)
return {
messages: orderNativeChatDisplayMessages([...messages, preview, ...echoes]),
preview: deriveNativeChatStreamingText({
messages: [...messages, ...echoes],
previewText: 'Starting the tests',
working: true
})
}
}
describe('terminal chat causal order', () => {
it('keeps idle sends and launch prompts before their answer and queued sends after it despite clock skew', () => {
const echoes = pendingSendsAsMessages([active, queued], [answer], 500)
const rows = orderNativeChatDisplayMessages([
answer,
preview,
...echoes,
{ ...answer, id: 'launch-pending:tab', role: 'user', timestamp: 99_000_000 },
{ ...answer, id: 'host-queue', role: 'user', queued: true }
])
expect(rows.map((row) => row.id)).toEqual([
'old-answer',
'pending:active',
'launch-pending:tab',
'streaming',
'pending:follow-up',
'host-queue'
])
})
it('does not promote follow-ups when the active prompt lands, and suppresses the caught-up preview', () => {
expect(tail([answer], [active, queued], 500).preview).toBe('Starting the tests')
const landed: NativeChatMessage = {
...answer,
id: 'prompt',
transcriptOrder: 2,
role: 'user',
blocks: [{ type: 'text', text: active.text }]
}
const caughtUp: NativeChatMessage = {
...preview,
id: 'answer',
source: 'transcript',
transcriptOrder: 3
}
const result = tail(
[answer, landed, caughtUp],
[active, queued, { ...queued, id: 'last', text: 'Then summarize', sentAt: 0 }],
500
)
expect(result.messages.map((row) => row.id)).toEqual([
'old-answer',
'prompt',
'answer',
'streaming',
'pending:follow-up',
'pending:last'
])
expect(result.preview).toBeNull()
})
it('makes a follow-up the active prompt when a later host epoch starts', () => {
const result = tail([answer], [queued], 501)
expect(result.messages.map((row) => row.id)).toEqual([
'old-answer',
'pending:follow-up',
'streaming'
])
expect(result.preview).toBe('Starting the tests')
})
it.each([
[null, null],
[500, null],
[null, 500]
])('keeps queued placement without durable evidence of a new turn (%s → %s)', (sent, current) => {
const result = tail([answer], [{ ...queued, queuedDuringWorkingEpoch: sent }], current)
expect(result.messages.map((row) => row.id)).toEqual([
'old-answer',
'streaming',
'pending:follow-up'
])
})
})
@@ -0,0 +1,50 @@
import type { NativeChatMessage } from '../native-chat-types'
import { NATIVE_CHAT_STREAMING_ID } from '../native-chat-streaming'
import { isLaunchPromptMessageId, isPendingMessageId } from './native-chat-pending-sends'
import {
compareNativeChatTranscriptMessages,
orderNativeChatTranscriptEntries
} from './native-chat-transcript-projection'
// Clocks cannot order a null-timestamp preview against a client send or a remote transcript.
function messageSortRank(message: NativeChatMessage): number {
if (message.queued || message.pendingSend === 'queued') {
return 3
}
if (message.id === NATIVE_CHAT_STREAMING_ID) {
return 2
}
if (
message.pendingSend ||
isPendingMessageId(message.id) ||
isLaunchPromptMessageId(message.id)
) {
return 1
}
return 0
}
export function compareNativeChatDisplayMessages(
a: NativeChatMessage,
b: NativeChatMessage
): number {
const rank = messageSortRank(a)
const difference = rank - messageSortRank(b)
return difference || (rank === 0 ? compareNativeChatTranscriptMessages(a, b) : 0)
}
export function orderNativeChatDisplayEntries<T>(
entries: readonly T[],
messageOf: (entry: T) => NativeChatMessage
): T[] {
return [0, 1, 2, 3].flatMap((rank) => {
const tier = entries.filter((entry) => messageSortRank(messageOf(entry)) === rank)
return rank === 0 ? orderNativeChatTranscriptEntries(tier, messageOf) : tier
})
}
export function orderNativeChatDisplayMessages(
messages: readonly NativeChatMessage[]
): NativeChatMessage[] {
return orderNativeChatDisplayEntries(messages, (message) => message)
}
@@ -25,8 +25,10 @@ export type NativeChatPendingSend = {
writeUnconfirmedAt?: number
text: string
imagePaths?: string[]
/** Epoch ms when the send was issued, so the queued bubble sorts to the end. */
/** Client clock at send time; display placement never compares it with the host clock. */
sentAt: number
/** Host working epoch observed at send time; absent for an idle send, null for an unknown epoch. */
queuedDuringWorkingEpoch?: number | null
/** Match only after the last authoritative row at send time so repeated prompts cannot bind older turns. */
afterMessageId?: string | null
/** Timestamp of that boundary in the transcript host's clock domain. */
@@ -204,10 +206,24 @@ export function visibleNativeChatPendingSends<T extends NativeChatPendingSend>(
})
}
/** Scrape-source echoes yield to real transcript rows; send timestamps keep queued bubbles at the end. */
export function nativeChatPendingSendPlacement(
entry: Pick<NativeChatPendingSend, 'queuedDuringWorkingEpoch'>,
workingEpoch?: number | null
): 'active' | 'queued' {
return entry.queuedDuringWorkingEpoch !== undefined &&
workingEpoch !== undefined &&
(entry.queuedDuringWorkingEpoch === null ||
workingEpoch === null ||
entry.queuedDuringWorkingEpoch === workingEpoch)
? 'queued'
: 'active'
}
/** Scrape-source echoes yield to real transcript rows. */
export function pendingSendsAsMessages(
pending: NativeChatPendingSend[],
existingMessages: readonly NativeChatMessage[] = []
existingMessages: readonly NativeChatMessage[] = [],
workingEpoch?: number | null
): NativeChatMessage[] {
return visibleNativeChatPendingSends(pending, existingMessages).map((entry) => ({
id: `pending:${entry.id}`,
@@ -217,6 +233,7 @@ export function pendingSendsAsMessages(
...(entry.text.trim().length > 0 ? [{ type: 'text' as const, text: entry.text }] : [])
],
timestamp: entry.sentAt,
pendingSend: nativeChatPendingSendPlacement(entry, workingEpoch),
source: 'scrape' as const
}))
}
@@ -224,3 +241,7 @@ export function pendingSendsAsMessages(
export function isPendingMessageId(id: string): boolean {
return id.startsWith('pending:')
}
export function isLaunchPromptMessageId(id: string): boolean {
return id.startsWith('launch-pending:')
}
@@ -400,9 +400,9 @@ describe('canonical source positions', () => {
marker.id,
'later',
'after',
NATIVE_CHAT_STREAMING_ID,
'launch-pending:1',
'pending:1',
'launch-pending:1',
NATIVE_CHAT_STREAMING_ID,
'queued'
])
// Contradictory clocks preserve the canonical subsequence, with one explicit supplemental placement.