fix(desktop): preserve full transcripts for running sessions
This commit is contained in:
Kinkoolino-Hermes
committed by
Teknium
Teknium
parent
bbb6cc7e99
commit
e4e7dd5c6e
@@ -612,6 +612,8 @@ function ResumeHarness({
|
||||
sessionStateByRuntimeIdRef?: MutableRefObject<Map<string, ClientSessionState>>
|
||||
}) {
|
||||
const ref = <T,>(value: T): MutableRefObject<T> => ({ current: value })
|
||||
const runtimeMapRef = runtimeIdByStoredSessionIdRef ?? ref(new Map<string, string>())
|
||||
const stateMapRef = sessionStateByRuntimeIdRef ?? ref(new Map<string, ClientSessionState>())
|
||||
|
||||
const actions = useSessionActions({
|
||||
activeSessionId: null,
|
||||
@@ -624,13 +626,16 @@ function ResumeHarness({
|
||||
navigate: vi.fn() as never,
|
||||
requestGateway,
|
||||
resetViewSync: vi.fn(),
|
||||
runtimeIdByStoredSessionIdRef: runtimeIdByStoredSessionIdRef ?? ref(new Map<string, string>()),
|
||||
runtimeIdByStoredSessionIdRef: runtimeMapRef,
|
||||
selectedStoredSessionId,
|
||||
selectedStoredSessionIdRef: ref<string | null>(selectedStoredSessionId),
|
||||
sessionStateByRuntimeIdRef: sessionStateByRuntimeIdRef ?? ref(new Map<string, ClientSessionState>()),
|
||||
sessionStateByRuntimeIdRef: stateMapRef,
|
||||
syncSessionStateToView: vi.fn(),
|
||||
updateSessionState: (sessionId, updater) => {
|
||||
const next = updater({} as ClientSessionState)
|
||||
const current = stateMapRef.current.get(sessionId) ?? ({} as ClientSessionState)
|
||||
const next = updater(current)
|
||||
|
||||
stateMapRef.current.set(sessionId, next)
|
||||
onStateUpdate?.(sessionId, next)
|
||||
|
||||
return next
|
||||
@@ -767,12 +772,16 @@ describe('resumeSession failure recovery', () => {
|
||||
expect($messages.get().map(message => message.id)).toContain('user-optimistic')
|
||||
})
|
||||
|
||||
it('restores the in-flight turn and queued user prompt after a full renderer restart', async () => {
|
||||
it('keeps the complete transcript with the live tail after a full renderer restart', async () => {
|
||||
const storedMessages = [
|
||||
{ content: 'earlier question', role: 'user', timestamp: 1 },
|
||||
{ content: 'earlier answer', role: 'assistant', timestamp: 2 }
|
||||
{ content: 'older question removed by compression', role: 'user', timestamp: 1 },
|
||||
{ content: 'older answer removed by compression', role: 'assistant', timestamp: 2 },
|
||||
{ content: 'recent question', role: 'user', timestamp: 3 },
|
||||
{ content: 'recent answer', role: 'assistant', timestamp: 4 }
|
||||
]
|
||||
|
||||
const compressedRuntimeMessages = storedMessages.slice(-2)
|
||||
|
||||
vi.mocked(getLatestSessionMessages).mockResolvedValue({ messages: storedMessages, session_id: 'stored-1' } as never)
|
||||
|
||||
const requestGateway = vi.fn(async (method: string) => {
|
||||
@@ -781,8 +790,8 @@ describe('resumeSession failure recovery', () => {
|
||||
session_id: 'runtime-1',
|
||||
session_key: 'stored-1',
|
||||
resumed: 'stored-1',
|
||||
message_count: storedMessages.length,
|
||||
messages: storedMessages,
|
||||
message_count: compressedRuntimeMessages.length,
|
||||
messages: compressedRuntimeMessages,
|
||||
running: true,
|
||||
inflight: {
|
||||
user: 'current prompt',
|
||||
@@ -810,11 +819,99 @@ describe('resumeSession failure recovery', () => {
|
||||
await resume!('stored-1', true)
|
||||
|
||||
const renderedMessages = JSON.stringify(resumedState?.messages)
|
||||
expect(renderedMessages).toContain('older question removed by compression')
|
||||
expect(renderedMessages).toContain('current prompt')
|
||||
expect(renderedMessages).toContain('partial answer')
|
||||
expect(renderedMessages).toContain('newest prompt')
|
||||
})
|
||||
|
||||
it('preserves a runtime-cache delta that arrives while cold resume waits for REST', async () => {
|
||||
const persisted = deferred<Awaited<ReturnType<typeof getLatestSessionMessages>>>()
|
||||
|
||||
const sessionStateByRuntimeIdRef: MutableRefObject<Map<string, ClientSessionState>> = {
|
||||
current: new Map()
|
||||
}
|
||||
|
||||
const compressedRuntimeMessages = [
|
||||
{ content: 'recent question', role: 'user', timestamp: 3 },
|
||||
{ content: 'recent answer', role: 'assistant', timestamp: 4 }
|
||||
]
|
||||
|
||||
vi.mocked(getLatestSessionMessages).mockReturnValue(persisted.promise)
|
||||
|
||||
const requestGateway = vi.fn(async (method: string) => {
|
||||
if (method === 'session.resume') {
|
||||
return {
|
||||
session_id: 'runtime-1',
|
||||
session_key: 'stored-1',
|
||||
resumed: 'stored-1',
|
||||
message_count: compressedRuntimeMessages.length,
|
||||
messages: compressedRuntimeMessages,
|
||||
running: true,
|
||||
inflight: {
|
||||
user: 'current prompt',
|
||||
assistant: 'partial A',
|
||||
streaming: true
|
||||
},
|
||||
info: {}
|
||||
} as never
|
||||
}
|
||||
|
||||
return {} as never
|
||||
})
|
||||
|
||||
let resumedState: ClientSessionState | undefined
|
||||
let resume: ((storedSessionId: string, replaceRoute?: boolean) => Promise<unknown>) | null = null
|
||||
|
||||
render(
|
||||
<ResumeHarness
|
||||
onReady={ready => (resume = ready)}
|
||||
onStateUpdate={(_sessionId, state) => (resumedState = state)}
|
||||
requestGateway={requestGateway}
|
||||
sessionStateByRuntimeIdRef={sessionStateByRuntimeIdRef}
|
||||
/>
|
||||
)
|
||||
await waitFor(() => expect(resume).not.toBeNull())
|
||||
|
||||
const resumePromise = resume!('stored-1', true)
|
||||
|
||||
await waitFor(() => expect(requestGateway).toHaveBeenCalledWith('session.resume', expect.anything()))
|
||||
|
||||
const runtimeState = clientState('stored-1')
|
||||
runtimeState.messages = [
|
||||
{
|
||||
id: 'assistant-stream-live-cold',
|
||||
role: 'assistant',
|
||||
parts: [{ type: 'text', text: ' + delta B' }],
|
||||
pending: true
|
||||
}
|
||||
]
|
||||
runtimeState.streamId = 'assistant-stream-live-cold'
|
||||
sessionStateByRuntimeIdRef.current.set('runtime-1', runtimeState)
|
||||
|
||||
await act(async () => {
|
||||
persisted.resolve({
|
||||
messages: [
|
||||
{ content: 'older question removed by compression', role: 'user', timestamp: 1 },
|
||||
{ content: 'older answer removed by compression', role: 'assistant', timestamp: 2 },
|
||||
...compressedRuntimeMessages
|
||||
],
|
||||
session_id: 'stored-1'
|
||||
} as never)
|
||||
await resumePromise
|
||||
})
|
||||
|
||||
const renderedText = JSON.stringify(resumedState?.messages)
|
||||
|
||||
const streamingAssistantRows = resumedState?.messages.filter(message => message.id.startsWith('assistant-stream-'))
|
||||
|
||||
expect(renderedText).toContain('older question removed by compression')
|
||||
expect(renderedText).toContain('partial A')
|
||||
expect(renderedText).toContain('delta B')
|
||||
expect(streamingAssistantRows).toHaveLength(1)
|
||||
expect(streamingAssistantRows?.[0].id).toBe('assistant-stream-live-cold')
|
||||
})
|
||||
|
||||
it('uses the continuation projection when resume rotates an equal-length stored transcript', async () => {
|
||||
const parentMessages = [
|
||||
{ content: 'question before compression', role: 'user', timestamp: 1 },
|
||||
@@ -1681,6 +1778,314 @@ describe('resumeSession warm-cache mapping integrity', () => {
|
||||
expect(renderedMessages).toContain('still here after wake too')
|
||||
})
|
||||
|
||||
it('keeps the complete persisted transcript when activating a compressed running session', async () => {
|
||||
const runtimeIdByStoredSessionIdRef: MutableRefObject<Map<string, string>> = {
|
||||
current: new Map([['stored-A', 'rt-A']])
|
||||
}
|
||||
|
||||
const state = clientState('stored-A')
|
||||
state.messages = [
|
||||
{
|
||||
id: 'runtime-user',
|
||||
role: 'user',
|
||||
parts: [{ type: 'text', text: 'recent prompt' }]
|
||||
},
|
||||
{
|
||||
id: 'runtime-assistant',
|
||||
role: 'assistant',
|
||||
parts: [{ type: 'text', text: 'recent answer' }]
|
||||
}
|
||||
]
|
||||
|
||||
const sessionStateByRuntimeIdRef: MutableRefObject<Map<string, ClientSessionState>> = {
|
||||
current: new Map([['rt-A', state]])
|
||||
}
|
||||
|
||||
const persistedMessages = [
|
||||
{ content: 'older prompt that compression removed', role: 'user', timestamp: 1 },
|
||||
{ content: 'older answer that compression removed', role: 'assistant', timestamp: 2 },
|
||||
{ content: 'recent prompt', role: 'user', timestamp: 3 },
|
||||
{ content: 'recent answer', role: 'assistant', timestamp: 4 }
|
||||
]
|
||||
|
||||
const compressedRuntimeMessages = persistedMessages.slice(2)
|
||||
|
||||
vi.mocked(getLatestSessionMessages).mockResolvedValue({
|
||||
messages: persistedMessages,
|
||||
session_id: 'stored-A'
|
||||
} as never)
|
||||
|
||||
const requestGateway = vi.fn(async (method: string) => {
|
||||
if (method === 'session.activate') {
|
||||
return {
|
||||
session_id: 'rt-A',
|
||||
session_key: 'stored-A',
|
||||
resumed: 'stored-A',
|
||||
message_count: compressedRuntimeMessages.length,
|
||||
messages: compressedRuntimeMessages,
|
||||
running: true,
|
||||
inflight: {
|
||||
user: 'current prompt',
|
||||
assistant: 'partial answer',
|
||||
streaming: true
|
||||
},
|
||||
info: {}
|
||||
} as never
|
||||
}
|
||||
|
||||
return {} as never
|
||||
})
|
||||
|
||||
let resumedState: ClientSessionState | undefined
|
||||
let resume: ((storedSessionId: string, replaceRoute?: boolean) => Promise<unknown>) | null = null
|
||||
|
||||
render(
|
||||
<ResumeHarness
|
||||
onReady={ready => (resume = ready)}
|
||||
onStateUpdate={(_sessionId, next) => (resumedState = next)}
|
||||
requestGateway={requestGateway}
|
||||
runtimeIdByStoredSessionIdRef={runtimeIdByStoredSessionIdRef}
|
||||
sessionStateByRuntimeIdRef={sessionStateByRuntimeIdRef}
|
||||
/>
|
||||
)
|
||||
await waitFor(() => expect(resume).not.toBeNull())
|
||||
await resume!('stored-A', true)
|
||||
|
||||
const renderedMessages = resumedState?.messages ?? []
|
||||
const renderedText = JSON.stringify(renderedMessages)
|
||||
|
||||
expect(renderedText).toContain('older prompt that compression removed')
|
||||
expect(renderedText).toContain('older answer that compression removed')
|
||||
expect(renderedText).toContain('recent prompt')
|
||||
expect(renderedText).toContain('recent answer')
|
||||
expect(renderedText).toContain('partial answer')
|
||||
expect(renderedMessages.filter(message => JSON.stringify(message).includes('current prompt'))).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('preserves live cache updates that arrive while the persisted transcript is loading', async () => {
|
||||
const runtimeIdByStoredSessionIdRef: MutableRefObject<Map<string, string>> = {
|
||||
current: new Map([['stored-A', 'rt-A']])
|
||||
}
|
||||
|
||||
const state = clientState('stored-A')
|
||||
state.messages = [
|
||||
{
|
||||
id: 'runtime-user',
|
||||
role: 'user',
|
||||
parts: [{ type: 'text', text: 'recent prompt' }],
|
||||
timestamp: 3
|
||||
},
|
||||
{
|
||||
id: 'runtime-assistant',
|
||||
role: 'assistant',
|
||||
parts: [{ type: 'text', text: 'recent answer' }],
|
||||
timestamp: 4
|
||||
},
|
||||
{
|
||||
id: 'user-inflight-rt-A',
|
||||
role: 'user',
|
||||
parts: [{ type: 'text', text: 'current prompt' }]
|
||||
},
|
||||
{
|
||||
id: 'assistant-stream-live-123',
|
||||
role: 'assistant',
|
||||
parts: [{ type: 'text', text: 'partial A' }],
|
||||
pending: true
|
||||
}
|
||||
]
|
||||
|
||||
const sessionStateByRuntimeIdRef: MutableRefObject<Map<string, ClientSessionState>> = {
|
||||
current: new Map([['rt-A', state]])
|
||||
}
|
||||
|
||||
const persisted = deferred<Awaited<ReturnType<typeof getLatestSessionMessages>>>()
|
||||
|
||||
const compressedRuntimeMessages = [
|
||||
{ content: 'recent prompt', role: 'user', timestamp: 3 },
|
||||
{ content: 'recent answer', role: 'assistant', timestamp: 4 }
|
||||
]
|
||||
|
||||
vi.mocked(getLatestSessionMessages).mockReturnValue(persisted.promise)
|
||||
|
||||
const requestGateway = vi.fn(async (method: string) => {
|
||||
if (method === 'session.activate') {
|
||||
return {
|
||||
session_id: 'rt-A',
|
||||
session_key: 'stored-A',
|
||||
resumed: 'stored-A',
|
||||
message_count: compressedRuntimeMessages.length,
|
||||
messages: compressedRuntimeMessages,
|
||||
running: true,
|
||||
inflight: {
|
||||
user: 'current prompt',
|
||||
assistant: 'partial A',
|
||||
streaming: true
|
||||
},
|
||||
info: {}
|
||||
} as never
|
||||
}
|
||||
|
||||
return {} as never
|
||||
})
|
||||
|
||||
let resumedState: ClientSessionState | undefined
|
||||
let resume: ((storedSessionId: string, replaceRoute?: boolean) => Promise<unknown>) | null = null
|
||||
|
||||
render(
|
||||
<ResumeHarness
|
||||
onReady={ready => (resume = ready)}
|
||||
onStateUpdate={(_sessionId, next) => (resumedState = next)}
|
||||
requestGateway={requestGateway}
|
||||
runtimeIdByStoredSessionIdRef={runtimeIdByStoredSessionIdRef}
|
||||
sessionStateByRuntimeIdRef={sessionStateByRuntimeIdRef}
|
||||
/>
|
||||
)
|
||||
await waitFor(() => expect(resume).not.toBeNull())
|
||||
|
||||
const resumePromise = resume!('stored-A', true)
|
||||
|
||||
await waitFor(() => expect(requestGateway).toHaveBeenCalledWith('session.activate', expect.anything()))
|
||||
|
||||
const liveState = sessionStateByRuntimeIdRef.current.get('rt-A')!
|
||||
|
||||
const liveMessages = liveState.messages.map(message =>
|
||||
message.id === 'assistant-stream-live-123'
|
||||
? { ...message, parts: [{ type: 'text' as const, text: 'partial A + delta B' }] }
|
||||
: message
|
||||
)
|
||||
|
||||
sessionStateByRuntimeIdRef.current.set('rt-A', {
|
||||
...liveState,
|
||||
messages: [
|
||||
...liveMessages,
|
||||
{
|
||||
id: 'user-racing',
|
||||
role: 'user',
|
||||
parts: [{ type: 'text', text: 'racing prompt' }]
|
||||
}
|
||||
]
|
||||
})
|
||||
|
||||
await act(async () => {
|
||||
persisted.resolve({
|
||||
messages: [
|
||||
{ content: 'older prompt', role: 'user', timestamp: 1 },
|
||||
{ content: 'older answer', role: 'assistant', timestamp: 2 },
|
||||
...compressedRuntimeMessages
|
||||
],
|
||||
session_id: 'stored-A'
|
||||
} as never)
|
||||
await resumePromise
|
||||
})
|
||||
|
||||
const renderedText = JSON.stringify(resumedState?.messages)
|
||||
|
||||
expect(renderedText).toContain('older prompt')
|
||||
expect(renderedText).toContain('partial A + delta B')
|
||||
expect(renderedText).toContain('racing prompt')
|
||||
|
||||
const streamingAssistantRows = resumedState?.messages.filter(message => message.id.startsWith('assistant-stream-'))
|
||||
|
||||
expect(streamingAssistantRows).toHaveLength(1)
|
||||
expect(streamingAssistantRows?.[0].id).toBe('assistant-stream-live-123')
|
||||
})
|
||||
|
||||
it('does not duplicate an in-flight user prompt already present in the persisted suffix', async () => {
|
||||
const runtimeIdByStoredSessionIdRef: MutableRefObject<Map<string, string>> = {
|
||||
current: new Map([['stored-A', 'rt-A']])
|
||||
}
|
||||
|
||||
const state = clientState('stored-A')
|
||||
state.messages = [
|
||||
{
|
||||
id: 'runtime-user',
|
||||
role: 'user',
|
||||
parts: [{ type: 'text', text: 'earlier prompt' }]
|
||||
},
|
||||
{
|
||||
id: 'runtime-assistant',
|
||||
role: 'assistant',
|
||||
parts: [{ type: 'text', text: 'earlier answer' }]
|
||||
},
|
||||
{
|
||||
id: 'user-optimistic',
|
||||
role: 'user',
|
||||
parts: [{ type: 'text', text: 'current prompt' }]
|
||||
},
|
||||
{
|
||||
id: 'assistant-stream-rt-A',
|
||||
role: 'assistant',
|
||||
parts: [{ type: 'text', text: 'partial answer' }],
|
||||
pending: true
|
||||
}
|
||||
]
|
||||
|
||||
const sessionStateByRuntimeIdRef: MutableRefObject<Map<string, ClientSessionState>> = {
|
||||
current: new Map([['rt-A', state]])
|
||||
}
|
||||
|
||||
const compressedRuntimeMessages = [
|
||||
{ content: 'earlier prompt', role: 'user', timestamp: 1 },
|
||||
{ content: 'earlier answer', role: 'assistant', timestamp: 2 }
|
||||
]
|
||||
|
||||
const persistedMessages = [
|
||||
{ content: 'older prompt removed by compression', role: 'user', timestamp: -1 },
|
||||
{ content: 'older answer removed by compression', role: 'assistant', timestamp: 0 },
|
||||
...compressedRuntimeMessages,
|
||||
{ content: 'current prompt', role: 'user', timestamp: 3 }
|
||||
]
|
||||
|
||||
vi.mocked(getLatestSessionMessages).mockResolvedValue({
|
||||
messages: persistedMessages,
|
||||
session_id: 'stored-A'
|
||||
} as never)
|
||||
|
||||
const requestGateway = vi.fn(async (method: string) => {
|
||||
if (method === 'session.activate') {
|
||||
return {
|
||||
session_id: 'rt-A',
|
||||
session_key: 'stored-A',
|
||||
resumed: 'stored-A',
|
||||
message_count: compressedRuntimeMessages.length,
|
||||
messages: compressedRuntimeMessages,
|
||||
running: true,
|
||||
inflight: {
|
||||
user: 'current prompt',
|
||||
assistant: 'partial answer',
|
||||
streaming: true
|
||||
},
|
||||
info: {}
|
||||
} as never
|
||||
}
|
||||
|
||||
return {} as never
|
||||
})
|
||||
|
||||
let resumedState: ClientSessionState | undefined
|
||||
let resume: ((storedSessionId: string, replaceRoute?: boolean) => Promise<unknown>) | null = null
|
||||
|
||||
render(
|
||||
<ResumeHarness
|
||||
onReady={ready => (resume = ready)}
|
||||
onStateUpdate={(_sessionId, next) => (resumedState = next)}
|
||||
requestGateway={requestGateway}
|
||||
runtimeIdByStoredSessionIdRef={runtimeIdByStoredSessionIdRef}
|
||||
sessionStateByRuntimeIdRef={sessionStateByRuntimeIdRef}
|
||||
/>
|
||||
)
|
||||
await waitFor(() => expect(resume).not.toBeNull())
|
||||
await resume!('stored-A', true)
|
||||
|
||||
const currentPromptRows = (resumedState?.messages ?? []).filter(message =>
|
||||
JSON.stringify(message).includes('current prompt')
|
||||
)
|
||||
|
||||
expect(currentPromptRows).toHaveLength(1)
|
||||
expect(JSON.stringify(resumedState?.messages)).toContain('partial answer')
|
||||
>>>>>>> bfb973e25c (fix(desktop): preserve full transcripts for running sessions)
|
||||
})
|
||||
|
||||
it('keeps a warm runtime and optimistic turn on a transient activation timeout', async () => {
|
||||
const runtimeIdByStoredSessionIdRef: MutableRefObject<Map<string, string>> = {
|
||||
current: new Map([['stored-A', 'rt-A']])
|
||||
|
||||
@@ -82,10 +82,13 @@ import {
|
||||
applyStoredSessionPreviewRuntimeInfo,
|
||||
type BranchMessage,
|
||||
chatMessageArraysEquivalent,
|
||||
dedupeInflightUserAgainstTranscript,
|
||||
isSessionGoneError,
|
||||
overlayConcurrentMessageChanges,
|
||||
patchSessionWorkspace,
|
||||
preserveLocalPendingTurnMessages,
|
||||
reconcileResumeMessages,
|
||||
removeRepresentedLocalLiveProjection,
|
||||
resolveResumedBusy,
|
||||
resolveSessionProfile,
|
||||
resolveStoredSession,
|
||||
@@ -137,17 +140,27 @@ function applyStoredUsage(stored: { input_tokens?: number | null; output_tokens?
|
||||
setCurrentUsage(current => ({ ...current, input, output, total: input + output }))
|
||||
}
|
||||
|
||||
function reconcileAuthoritativeChatMessages(
|
||||
authoritativeMessages: ChatMessage[],
|
||||
previousMessages: ChatMessage[],
|
||||
liveProjection?: Pick<SessionResumeResponse, 'inflight' | 'queued' | 'session_id'>
|
||||
): ChatMessage[] {
|
||||
const withLiveProjection = liveProjection
|
||||
? appendLiveSessionProjection(authoritativeMessages, liveProjection)
|
||||
: authoritativeMessages
|
||||
|
||||
const reconciled = reconcileResumeMessages(withLiveProjection, previousMessages)
|
||||
const withPendingTurn = preserveLocalPendingTurnMessages(reconciled, previousMessages)
|
||||
|
||||
return preserveLocalAssistantErrors(withPendingTurn, previousMessages)
|
||||
}
|
||||
|
||||
function reconcileAuthoritativeMessages(
|
||||
authoritativeMessages: SessionResumeResponse['messages'],
|
||||
previousMessages: ChatMessage[],
|
||||
liveProjection?: Pick<SessionResumeResponse, 'inflight' | 'queued' | 'session_id'>
|
||||
): ChatMessage[] {
|
||||
const authoritative = toChatMessages(authoritativeMessages)
|
||||
const withLiveProjection = liveProjection ? appendLiveSessionProjection(authoritative, liveProjection) : authoritative
|
||||
const reconciled = reconcileResumeMessages(withLiveProjection, previousMessages)
|
||||
const withPendingTurn = preserveLocalPendingTurnMessages(reconciled, previousMessages)
|
||||
|
||||
return preserveLocalAssistantErrors(withPendingTurn, previousMessages)
|
||||
return reconcileAuthoritativeChatMessages(toChatMessages(authoritativeMessages), previousMessages, liveProjection)
|
||||
}
|
||||
|
||||
// `session.create` params from the current profile + sticky-UI model/effort/fast,
|
||||
@@ -839,12 +852,12 @@ export function useSessionActions({
|
||||
Boolean(sessionStateByRuntimeIdRef.current.get(cachedRuntimeId)?.busy)
|
||||
)
|
||||
|
||||
// While idle, the persisted REST transcript is the display
|
||||
// authority: session.activate returns the runtime's compressed
|
||||
// context projection, not necessarily the complete conversation.
|
||||
// During a live turn, keep the runtime/cache projection so an
|
||||
// accepted but not-yet-persisted prompt or stream is never lost.
|
||||
if (!running && persistedTranscriptPromise) {
|
||||
// The persisted REST transcript is the display authority: a live
|
||||
// runtime may carry only the agent's compressed context projection,
|
||||
// which is intentionally smaller than the user-visible conversation.
|
||||
// Reconcile its in-flight/queued tail onto the complete transcript
|
||||
// instead of replacing durable history while the turn is running.
|
||||
if (persistedTranscriptPromise) {
|
||||
const persisted = await persistedTranscriptPromise
|
||||
|
||||
if (!isCurrentResume()) {
|
||||
@@ -869,10 +882,34 @@ export function useSessionActions({
|
||||
persistedMatchesActivatedSession &&
|
||||
(persisted.messages.length || !activatedMessages.length)
|
||||
) {
|
||||
activatedMessages = reconcileAuthoritativeMessages(persisted.messages, activatedMessages)
|
||||
const persistedMessages = toChatMessages(persisted.messages)
|
||||
const runtimeMessages = toChatMessages(activated.messages)
|
||||
const previousMessages = removeRepresentedLocalLiveProjection(cachedViewState.messages, activated)
|
||||
|
||||
const liveProjection = dedupeInflightUserAgainstTranscript(
|
||||
persistedMessages,
|
||||
runtimeMessages,
|
||||
activated
|
||||
)
|
||||
|
||||
activatedMessages = reconcileAuthoritativeChatMessages(
|
||||
persistedMessages,
|
||||
previousMessages,
|
||||
liveProjection
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
const currentMessages = sessionStateByRuntimeIdRef.current.get(cachedRuntimeId)?.messages
|
||||
|
||||
if (currentMessages) {
|
||||
activatedMessages = overlayConcurrentMessageChanges(
|
||||
activatedMessages,
|
||||
cachedViewState.messages,
|
||||
currentMessages
|
||||
)
|
||||
}
|
||||
|
||||
const activatedState = updateSessionState(
|
||||
cachedRuntimeId,
|
||||
state => ({
|
||||
@@ -967,6 +1004,7 @@ export function useSessionActions({
|
||||
|
||||
let prefetchApplied = false
|
||||
let prefetchedStoredSessionId: string | null = null
|
||||
let prefetchedTranscriptMessages: ChatMessage[] | null = null
|
||||
|
||||
// REST transcript prefetch and the gateway resume RPC are independent
|
||||
// — run them concurrently so a big session's wall time is
|
||||
@@ -975,6 +1013,8 @@ export function useSessionActions({
|
||||
// Watch windows skip the prefetch — lazy resume attaches the live mirror.
|
||||
const prefetchPromise = watchWindow ? null : getLatestSessionMessages(storedSessionId, sessionProfile)
|
||||
|
||||
let resumeRuntimeBaselineMessages: ChatMessage[] = []
|
||||
|
||||
const resumePromise = requestGateway<SessionResumeResponse>('session.resume', {
|
||||
session_id: storedSessionId,
|
||||
cols: 96,
|
||||
@@ -988,6 +1028,11 @@ export function useSessionActions({
|
||||
// background while the prefetch above paints the transcript.
|
||||
...(watchWindow ? { lazy: true } : { omit_messages: true }),
|
||||
...(sessionProfile ? { profile: sessionProfile } : {})
|
||||
}).then(resumed => {
|
||||
resumeRuntimeBaselineMessages =
|
||||
sessionStateByRuntimeIdRef.current.get(resumed.session_id)?.messages ?? resumeRuntimeBaselineMessages
|
||||
|
||||
return resumed
|
||||
})
|
||||
|
||||
// The rejection is consumed by the `await` below; this guard only
|
||||
@@ -1018,7 +1063,8 @@ export function useSessionActions({
|
||||
? preserveLocalPendingTurnMessages($messages.get(), resumeStartMessages)
|
||||
: $messages.get()
|
||||
|
||||
localSnapshot = reconcileAuthoritativeMessages(prefetchedResult.messages, previousMessages)
|
||||
prefetchedTranscriptMessages = toChatMessages(prefetchedResult.messages)
|
||||
localSnapshot = reconcileAuthoritativeChatMessages(prefetchedTranscriptMessages, previousMessages)
|
||||
prefetchApplied = true
|
||||
prefetchedStoredSessionId = prefetchedResult.session_id || storedSessionId
|
||||
}
|
||||
@@ -1037,26 +1083,63 @@ export function useSessionActions({
|
||||
|
||||
const hasLiveProjection = Boolean(resumed.inflight || resumed.queued)
|
||||
|
||||
const preferredMessages =
|
||||
prefetchApplied && prefetchMatchesResumedSession && !hasLiveProjection
|
||||
? localSnapshot
|
||||
: (() => {
|
||||
const previousMessages = resumedSameSelectedSession
|
||||
? preserveLocalPendingTurnMessages(currentMessages, resumeStartMessages)
|
||||
: currentMessages
|
||||
const preferredMessages = (() => {
|
||||
if (prefetchApplied && prefetchMatchesResumedSession) {
|
||||
if (hasLiveProjection && prefetchedTranscriptMessages) {
|
||||
const runtimeMessages = toChatMessages(resumed.messages)
|
||||
const previousMessages = removeRepresentedLocalLiveProjection(currentMessages, resumed)
|
||||
|
||||
// Omitted, not empty — same trap as the activate path above.
|
||||
// The REST prefetch IS the transcript here; the resume payload
|
||||
// only contributes the live tail, so graft rather than rebuild.
|
||||
// (Without a usable prefetch there is nothing better to stand
|
||||
// on, so the projection alone remains the degraded fallback.)
|
||||
const resumedMessages =
|
||||
resumed.messages_omitted && prefetchApplied && prefetchMatchesResumedSession
|
||||
? appendLiveSessionProjection(localSnapshot, resumed)
|
||||
: reconcileAuthoritativeMessages(resumed.messages, previousMessages, resumed)
|
||||
// Omitted-messages resumes stay safe here: `resumed.messages`
|
||||
// is empty, so `runtimeMessages` has no anchor and the dedupe
|
||||
// helper returns the projection unchanged, while the REST
|
||||
// prefetch below remains the authoritative transcript — the
|
||||
// same "graft, don't rebuild" outcome the pre-restructure
|
||||
// messages_omitted branch produced.
|
||||
const liveProjection = dedupeInflightUserAgainstTranscript(
|
||||
prefetchedTranscriptMessages,
|
||||
runtimeMessages,
|
||||
resumed
|
||||
)
|
||||
|
||||
return chatMessageArraysEquivalent(currentMessages, resumedMessages) ? currentMessages : resumedMessages
|
||||
})()
|
||||
const resumedMessages = reconcileAuthoritativeChatMessages(
|
||||
prefetchedTranscriptMessages,
|
||||
previousMessages,
|
||||
liveProjection
|
||||
)
|
||||
|
||||
const withConcurrentChanges = overlayConcurrentMessageChanges(
|
||||
resumedMessages,
|
||||
localSnapshot,
|
||||
currentMessages
|
||||
)
|
||||
|
||||
return chatMessageArraysEquivalent(currentMessages, withConcurrentChanges)
|
||||
? currentMessages
|
||||
: withConcurrentChanges
|
||||
}
|
||||
|
||||
if (!hasLiveProjection) {
|
||||
return localSnapshot
|
||||
}
|
||||
}
|
||||
|
||||
const previousMessages = resumedSameSelectedSession
|
||||
? preserveLocalPendingTurnMessages(currentMessages, resumeStartMessages)
|
||||
: currentMessages
|
||||
|
||||
const resumedMessages = reconcileAuthoritativeMessages(resumed.messages, previousMessages, resumed)
|
||||
|
||||
return chatMessageArraysEquivalent(currentMessages, resumedMessages) ? currentMessages : resumedMessages
|
||||
})()
|
||||
|
||||
const currentRuntimeMessages =
|
||||
sessionStateByRuntimeIdRef.current.get(resumed.session_id)?.messages ?? resumeRuntimeBaselineMessages
|
||||
|
||||
const preferredWithRuntimeChanges = overlayConcurrentMessageChanges(
|
||||
preferredMessages,
|
||||
resumeRuntimeBaselineMessages,
|
||||
currentRuntimeMessages
|
||||
)
|
||||
|
||||
// #70449: same stale-snapshot guard as the warm path — a turn that
|
||||
// started while the resume RPC was in flight has already marked the
|
||||
@@ -1071,20 +1154,16 @@ export function useSessionActions({
|
||||
// (persisted by use-session-state-cache while the turn streamed;
|
||||
// survives renderer/app death) back onto the restored transcript. The
|
||||
// backend's own inflight projection is already inside
|
||||
// `preferredMessages` (appendLiveSessionProjection), so this merge only
|
||||
// adds the locally recorded structure — tool calls, sealed interim
|
||||
// rows — that the backend's text-only snapshot cannot carry. A no-op
|
||||
// returns `preferredMessages` by reference, keeping the fast path
|
||||
// below intact.
|
||||
const inFlightRecovery = recoverInFlightTurnJournal(storedSessionId, preferredMessages, {
|
||||
// `preferredWithRuntimeChanges`, so this merge only adds the locally
|
||||
// recorded structure that the backend's text-only snapshot cannot carry.
|
||||
const inFlightRecovery = recoverInFlightTurnJournal(storedSessionId, preferredWithRuntimeChanges, {
|
||||
keepPending: resumedRunning
|
||||
})
|
||||
|
||||
recoveredInFlightTail = inFlightRecovery.applied
|
||||
|
||||
// Prefetch-hit fast path: `preferredMessages` IS the live `$messages`
|
||||
// array (already error-merged when `localSnapshot` was built), so reuse
|
||||
// the ref instead of rebuilding a throwaway transcript+Map every switch.
|
||||
// Prefetch-hit fast path: reuse the live array when neither runtime
|
||||
// changes nor in-flight recovery changed the reconciled transcript.
|
||||
const messagesForView =
|
||||
inFlightRecovery.messages === currentMessages
|
||||
? currentMessages
|
||||
|
||||
@@ -13,7 +13,7 @@ import {
|
||||
setSelectedStoredSessionId,
|
||||
workspaceCwdBelongsToSelectedSession
|
||||
} from '@/store/session'
|
||||
import type { SessionInfo } from '@/types/hermes'
|
||||
import type { SessionInfo, SessionResumeResponse } from '@/types/hermes'
|
||||
|
||||
import {
|
||||
appendLiveSessionProjection,
|
||||
@@ -22,9 +22,12 @@ import {
|
||||
chatMessageArraysEquivalent,
|
||||
chatMessagesEquivalent,
|
||||
chatPartsEquivalent,
|
||||
dedupeInflightUserAgainstTranscript,
|
||||
isSessionGoneError,
|
||||
overlayConcurrentMessageChanges,
|
||||
preserveLocalPendingTurnMessages,
|
||||
reconcileResumeMessages,
|
||||
removeRepresentedLocalLiveProjection,
|
||||
resolveResumedBusy,
|
||||
sessionMatchesStoredId,
|
||||
sessionShouldHaveTranscript,
|
||||
@@ -1447,3 +1450,177 @@ describe('resolveResumedBusy', () => {
|
||||
expect(resolveResumedBusy(true, true)).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
const runningProjection = (user: string): SessionResumeResponse =>
|
||||
({
|
||||
session_id: 'runtime-1',
|
||||
session_key: 'stored-1',
|
||||
resumed: 'stored-1',
|
||||
message_count: 2,
|
||||
messages: [],
|
||||
running: true,
|
||||
inflight: { user, assistant: 'partial answer', streaming: true }
|
||||
}) as SessionResumeResponse
|
||||
|
||||
describe('dedupeInflightUserAgainstTranscript', () => {
|
||||
it('retains the in-flight user source only when it already exists after the runtime anchor', () => {
|
||||
const runtime = [
|
||||
msg('runtime-user', 'user', 'earlier prompt', { timestamp: 1 }),
|
||||
msg('runtime-assistant', 'assistant', 'earlier answer', { timestamp: 2 })
|
||||
]
|
||||
|
||||
const persisted = [...runtime, msg('persisted-current', 'user', 'current prompt', { timestamp: 3 })]
|
||||
|
||||
const deduped = dedupeInflightUserAgainstTranscript(persisted, runtime, runningProjection('current prompt'))
|
||||
|
||||
expect(deduped.inflight?.user).toBe('current prompt')
|
||||
expect(deduped.inflight?.assistant).toBe('partial answer')
|
||||
})
|
||||
|
||||
it('preserves the assistant boundary before a queued turn when the persisted in-flight user has no delta', () => {
|
||||
const runtime = [
|
||||
msg('runtime-user', 'user', 'earlier prompt', { timestamp: 1 }),
|
||||
msg('runtime-assistant', 'assistant', 'earlier answer', { timestamp: 2 })
|
||||
]
|
||||
|
||||
const persisted = [...runtime, msg('persisted-current', 'user', 'current prompt', { timestamp: 3 })]
|
||||
|
||||
const projection = {
|
||||
...runningProjection('current prompt'),
|
||||
inflight: { user: 'current prompt', assistant: '', streaming: false },
|
||||
queued: { user: 'queued prompt' }
|
||||
}
|
||||
|
||||
const deduped = dedupeInflightUserAgainstTranscript(persisted, runtime, projection)
|
||||
const restored = appendLiveSessionProjection(persisted, deduped)
|
||||
|
||||
expect(restored.map(message => message.role)).toEqual(['user', 'assistant', 'user', 'assistant', 'user'])
|
||||
expect(restored.slice(-2).map(message => message.id)).toEqual([
|
||||
'assistant-stream-runtime-1',
|
||||
'user-queued-runtime-1'
|
||||
])
|
||||
})
|
||||
|
||||
it('preserves an intentionally repeated prompt when the match is before the runtime anchor', () => {
|
||||
const runtime = [
|
||||
msg('runtime-user', 'user', 'repeat this', { timestamp: 1 }),
|
||||
msg('runtime-assistant', 'assistant', 'finished answer', { timestamp: 2 })
|
||||
]
|
||||
|
||||
const projection = runningProjection('repeat this')
|
||||
const unchanged = dedupeInflightUserAgainstTranscript(runtime, runtime, projection)
|
||||
|
||||
expect(unchanged).toBe(projection)
|
||||
expect(unchanged.inflight?.user).toBe('repeat this')
|
||||
})
|
||||
|
||||
it('preserves a repeated in-flight prompt when the persisted match already has an answer', () => {
|
||||
const runtime = [
|
||||
msg('runtime-user', 'user', 'earlier prompt', { timestamp: 1 }),
|
||||
msg('runtime-assistant', 'assistant', 'earlier answer', { timestamp: 2 })
|
||||
]
|
||||
|
||||
const persisted = [
|
||||
...runtime,
|
||||
msg('persisted-repeat', 'user', 'repeat this', { timestamp: 3 }),
|
||||
msg('persisted-repeat-answer', 'assistant', 'finished repeat answer', { timestamp: 4 })
|
||||
]
|
||||
|
||||
const projection = runningProjection('repeat this')
|
||||
const unchanged = dedupeInflightUserAgainstTranscript(persisted, runtime, projection)
|
||||
|
||||
expect(unchanged).toBe(projection)
|
||||
expect(unchanged.inflight?.user).toBe('repeat this')
|
||||
})
|
||||
})
|
||||
|
||||
describe('removeRepresentedLocalLiveProjection', () => {
|
||||
it('removes only matched synthetic rows from the open local tail', () => {
|
||||
const previous = [
|
||||
msg('user-old-optimistic', 'user', 'current prompt'),
|
||||
msg('assistant-complete', 'assistant', 'finished answer'),
|
||||
msg('user-current', 'user', 'current prompt'),
|
||||
msg('assistant-stream-current', 'assistant', 'partial answer', { pending: true }),
|
||||
msg('user-queued-runtime', 'user', 'queued prompt'),
|
||||
msg('user-racing', 'user', 'new racing prompt')
|
||||
]
|
||||
|
||||
const projection = {
|
||||
...runningProjection('current prompt'),
|
||||
queued: { user: 'queued prompt' }
|
||||
}
|
||||
|
||||
const remaining = removeRepresentedLocalLiveProjection(previous, projection)
|
||||
|
||||
expect(remaining.map(message => message.id)).toEqual(['user-old-optimistic', 'assistant-complete', 'user-racing'])
|
||||
})
|
||||
|
||||
it('preserves an ambiguous text-identical local race prompt without a matching stream boundary', () => {
|
||||
const previous = [
|
||||
msg('runtime-assistant', 'assistant', 'finished answer'),
|
||||
msg('user-racing', 'user', 'repeat this')
|
||||
]
|
||||
|
||||
const projection = runningProjection('repeat this')
|
||||
|
||||
expect(removeRepresentedLocalLiveProjection(previous, projection)).toBe(previous)
|
||||
})
|
||||
|
||||
it('does not consume a generic racing user as the activation-owned queued row', () => {
|
||||
const previous = [
|
||||
msg('runtime-assistant', 'assistant', 'finished answer'),
|
||||
msg('user-current', 'user', 'current prompt'),
|
||||
msg('assistant-stream-current', 'assistant', 'partial answer', { pending: true }),
|
||||
msg('user-racing', 'user', 'repeat this')
|
||||
]
|
||||
|
||||
const projection = {
|
||||
...runningProjection('current prompt'),
|
||||
queued: { user: 'repeat this' }
|
||||
}
|
||||
|
||||
const remaining = removeRepresentedLocalLiveProjection(previous, projection)
|
||||
|
||||
expect(remaining.map(message => message.id)).toEqual(['runtime-assistant', 'user-racing'])
|
||||
})
|
||||
})
|
||||
|
||||
describe('overlayConcurrentMessageChanges', () => {
|
||||
it('does not replace an authoritative row with an unchanged baseline cache row', () => {
|
||||
const baseline = [msg('shared-assistant', 'assistant', 'stale cached answer')]
|
||||
const authoritative = [msg('shared-assistant', 'assistant', 'completed persisted answer')]
|
||||
|
||||
const overlaid = overlayConcurrentMessageChanges(authoritative, baseline, baseline)
|
||||
|
||||
expect(overlaid).toBe(authoritative)
|
||||
expect(overlaid[0].parts).toEqual([{ type: 'text', text: 'completed persisted answer' }])
|
||||
})
|
||||
|
||||
it('replaces an activation stream placeholder and appends rows created after the baseline', () => {
|
||||
const baseline = [msg('assistant-stream-runtime', 'assistant', 'partial A', { pending: true })]
|
||||
const authoritative = [msg('assistant-stream-activation', 'assistant', 'partial A', { pending: true })]
|
||||
|
||||
const current = [
|
||||
msg('assistant-stream-runtime', 'assistant', 'partial A + delta B', { pending: true }),
|
||||
msg('user-racing', 'user', 'racing prompt')
|
||||
]
|
||||
|
||||
const overlaid = overlayConcurrentMessageChanges(authoritative, baseline, current)
|
||||
|
||||
expect(overlaid.map(message => message.id)).toEqual(['assistant-stream-runtime', 'user-racing'])
|
||||
expect(overlaid[0].parts).toEqual([{ type: 'text', text: 'partial A + delta B' }])
|
||||
})
|
||||
|
||||
it('merges an activation prefix with a baseline-new runtime delta chunk', () => {
|
||||
const authoritative = [msg('assistant-stream-activation', 'assistant', 'partial A', { pending: true })]
|
||||
const current = [msg('assistant-stream-runtime', 'assistant', ' + delta B', { pending: true })]
|
||||
|
||||
const overlaid = overlayConcurrentMessageChanges(authoritative, [], current)
|
||||
|
||||
expect(overlaid.map(message => message.id)).toEqual(['assistant-stream-runtime'])
|
||||
expect(overlaid[0].parts).toEqual([
|
||||
{ type: 'text', text: 'partial A' },
|
||||
{ type: 'text', text: ' + delta B' }
|
||||
])
|
||||
})
|
||||
})
|
||||
|
||||
@@ -683,10 +683,17 @@ export function preserveLocalPendingTurnMessages(
|
||||
* memory. Stable ids let repeated activate/resume hydration reconcile instead
|
||||
* of growing duplicate rows.
|
||||
*/
|
||||
export function appendLiveSessionProjection(
|
||||
messages: ChatMessage[],
|
||||
projection: Pick<SessionResumeResponse, 'inflight' | 'queued' | 'session_id'>
|
||||
): ChatMessage[] {
|
||||
const safelyPersistedInflightUser = Symbol('safelyPersistedInflightUser')
|
||||
|
||||
type LiveSessionProjection = Pick<SessionResumeResponse, 'inflight' | 'queued' | 'session_id'> & {
|
||||
[safelyPersistedInflightUser]?: true
|
||||
}
|
||||
|
||||
type ReconciledSessionResumeResponse = SessionResumeResponse & {
|
||||
[safelyPersistedInflightUser]?: true
|
||||
}
|
||||
|
||||
export function appendLiveSessionProjection(messages: ChatMessage[], projection: LiveSessionProjection): ChatMessage[] {
|
||||
const inflightUser = projection.inflight?.user?.trim() ?? ''
|
||||
const inflightAssistant = projection.inflight?.assistant ?? ''
|
||||
const inflightStreaming = Boolean(projection.inflight?.streaming)
|
||||
@@ -762,7 +769,8 @@ export function appendLiveSessionProjection(
|
||||
message => textWithoutReferenceLines(chatMessageText(message)) === textWithoutReferenceLines(text)
|
||||
)
|
||||
|
||||
const inflightUserAlreadyPersisted = Boolean(inflightUser) && persistedInLatestRun(inflightUser)
|
||||
const inflightUserAlreadyPersisted =
|
||||
projection[safelyPersistedInflightUser] === true || (Boolean(inflightUser) && persistedInLatestRun(inflightUser))
|
||||
|
||||
if (inflightUser && !inflightUserAlreadyPersisted) {
|
||||
projected.push({
|
||||
@@ -894,6 +902,224 @@ export function appendLiveSessionProjection(
|
||||
return projected.length ? [...messages, ...projected] : messages
|
||||
}
|
||||
|
||||
function normalizedMessageText(message: ChatMessage): string {
|
||||
return chatMessageText(message).replace(/\s+/g, ' ').trim()
|
||||
}
|
||||
|
||||
function transcriptAnchorMatches(a: ChatMessage, b: ChatMessage): boolean {
|
||||
if (a.role !== b.role) {
|
||||
return false
|
||||
}
|
||||
|
||||
const aText = normalizedMessageText(a)
|
||||
const bText = normalizedMessageText(b)
|
||||
|
||||
if (a.timestamp !== undefined && b.timestamp !== undefined) {
|
||||
return a.timestamp === b.timestamp && aText === bText
|
||||
}
|
||||
|
||||
return Boolean(aText) && aText === bText
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark only an already-materialized `inflight.user` for visual suppression.
|
||||
*
|
||||
* A running gateway returns two independent truths: its compressed runtime
|
||||
* history plus the current in-flight turn, while REST may already have flushed
|
||||
* that user row into the complete persisted transcript. Global text dedupe is
|
||||
* unsafe because users may intentionally submit the same prompt twice. Instead,
|
||||
* find the last runtime message inside the persisted transcript and inspect only
|
||||
* the newer persisted suffix.
|
||||
*
|
||||
* Keep `inflight.user` intact because it also carries turn structure: a queued
|
||||
* prompt needs its assistant boundary even when the persisted user has no
|
||||
* assistant delta yet. The private marker lets the renderer suppress only that
|
||||
* duplicate bubble. If the histories have no safe common anchor, keep the
|
||||
* projection unchanged — a duplicate is recoverable, but dropping a real
|
||||
* accepted prompt is not.
|
||||
*/
|
||||
export function dedupeInflightUserAgainstTranscript(
|
||||
persistedMessages: ChatMessage[],
|
||||
runtimeMessages: ChatMessage[],
|
||||
projection: SessionResumeResponse
|
||||
): ReconciledSessionResumeResponse {
|
||||
const inflightUser = projection.inflight?.user?.replace(/\s+/g, ' ').trim() ?? ''
|
||||
|
||||
if (!inflightUser) {
|
||||
return projection
|
||||
}
|
||||
|
||||
let suffixStart = 0
|
||||
|
||||
if (runtimeMessages.length) {
|
||||
const runtimeAnchor = runtimeMessages[runtimeMessages.length - 1]
|
||||
let persistedAnchorIndex = -1
|
||||
|
||||
for (let index = persistedMessages.length - 1; index >= 0; index -= 1) {
|
||||
if (transcriptAnchorMatches(persistedMessages[index], runtimeAnchor)) {
|
||||
persistedAnchorIndex = index
|
||||
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if (persistedAnchorIndex < 0) {
|
||||
return projection
|
||||
}
|
||||
|
||||
suffixStart = persistedAnchorIndex + 1
|
||||
}
|
||||
|
||||
const persistedTail = persistedMessages.slice(suffixStart)
|
||||
const lastPersistedMessage = persistedTail[persistedTail.length - 1]
|
||||
|
||||
const persistedUserPresent =
|
||||
lastPersistedMessage?.role === 'user' && normalizedMessageText(lastPersistedMessage) === inflightUser
|
||||
|
||||
if (!persistedUserPresent) {
|
||||
return projection
|
||||
}
|
||||
|
||||
return { ...projection, [safelyPersistedInflightUser]: true }
|
||||
}
|
||||
|
||||
/**
|
||||
* Drop only synthetic local tail rows that the activation snapshot replaces.
|
||||
* Unmatched optimistic rows survive so a submit racing with activation is not
|
||||
* lost; completed transcript rows before the open tail are never considered.
|
||||
*/
|
||||
export function removeRepresentedLocalLiveProjection(
|
||||
previousMessages: ChatMessage[],
|
||||
projection: Pick<SessionResumeResponse, 'inflight' | 'queued'>
|
||||
): ChatMessage[] {
|
||||
const inflightUser = projection.inflight?.user?.replace(/\s+/g, ' ').trim() ?? ''
|
||||
const inflightAssistant = projection.inflight?.assistant?.replace(/\s+/g, ' ').trim() ?? ''
|
||||
const queuedUser = projection.queued?.user?.replace(/\s+/g, ' ').trim() ?? ''
|
||||
|
||||
const hasAssistantProjection = Boolean(
|
||||
projection.inflight?.assistant || projection.inflight?.streaming || (inflightUser && queuedUser)
|
||||
)
|
||||
|
||||
if (!inflightUser || !hasAssistantProjection) {
|
||||
return previousMessages
|
||||
}
|
||||
|
||||
let openTailStart = 0
|
||||
|
||||
for (let index = previousMessages.length - 1; index >= 0; index -= 1) {
|
||||
const message = previousMessages[index]
|
||||
|
||||
if (message.role === 'assistant' && !message.pending) {
|
||||
openTailStart = index + 1
|
||||
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
const inflightUserIndex = previousMessages.findIndex(
|
||||
(message, index) =>
|
||||
index >= openTailStart &&
|
||||
message.role === 'user' &&
|
||||
message.id.startsWith('user-') &&
|
||||
normalizedMessageText(message) === inflightUser
|
||||
)
|
||||
|
||||
const assistantIndex = inflightUserIndex + 1
|
||||
const assistant = previousMessages[assistantIndex]
|
||||
|
||||
const assistantMatches =
|
||||
inflightUserIndex >= openTailStart &&
|
||||
assistant?.role === 'assistant' &&
|
||||
assistant.id.startsWith('assistant-stream-') &&
|
||||
normalizedMessageText(assistant) === inflightAssistant
|
||||
|
||||
if (!assistantMatches) {
|
||||
return previousMessages
|
||||
}
|
||||
|
||||
let queuedUserIndex = -1
|
||||
|
||||
if (queuedUser) {
|
||||
queuedUserIndex = previousMessages.findIndex(
|
||||
(message, index) =>
|
||||
index > assistantIndex &&
|
||||
message.role === 'user' &&
|
||||
message.id.startsWith('user-queued-') &&
|
||||
normalizedMessageText(message) === queuedUser
|
||||
)
|
||||
}
|
||||
|
||||
return previousMessages.filter(
|
||||
(_message, index) => index !== inflightUserIndex && index !== assistantIndex && index !== queuedUserIndex
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Overlay messages that changed while activation waited on REST. Existing ids
|
||||
* replace the older activation row; only rows added or changed since the warm
|
||||
* cache baseline are appended. This is identity-based, never text-based.
|
||||
*/
|
||||
export function overlayConcurrentMessageChanges(
|
||||
nextMessages: ChatMessage[],
|
||||
baselineMessages: ChatMessage[],
|
||||
currentMessages: ChatMessage[]
|
||||
): ChatMessage[] {
|
||||
const baselineById = new Map(baselineMessages.map(message => [message.id, message]))
|
||||
const nextIndexById = new Map(nextMessages.map((message, index) => [message.id, index]))
|
||||
let changed = false
|
||||
const overlaid = [...nextMessages]
|
||||
|
||||
let activationStreamIndex = overlaid.findIndex(
|
||||
message =>
|
||||
message.role === 'assistant' && message.id.startsWith('assistant-stream-') && !baselineById.has(message.id)
|
||||
)
|
||||
|
||||
for (const current of currentMessages) {
|
||||
const baseline = baselineById.get(current.id)
|
||||
const changedSinceBaseline = !baseline || !chatMessagesEquivalent(baseline, current)
|
||||
|
||||
if (!changedSinceBaseline) {
|
||||
continue
|
||||
}
|
||||
|
||||
const nextIndex = nextIndexById.get(current.id)
|
||||
|
||||
if (nextIndex !== undefined) {
|
||||
if (!chatMessagesEquivalent(overlaid[nextIndex], current)) {
|
||||
overlaid[nextIndex] = current
|
||||
changed = true
|
||||
}
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
if (activationStreamIndex >= 0 && current.role === 'assistant' && current.id.startsWith('assistant-stream-')) {
|
||||
const activationStream = overlaid[activationStreamIndex]
|
||||
const activationText = chatMessageText(activationStream)
|
||||
const currentText = chatMessageText(current)
|
||||
|
||||
const replacement =
|
||||
activationText && !currentText.startsWith(activationText)
|
||||
? { ...current, parts: [...activationStream.parts, ...current.parts] }
|
||||
: current
|
||||
|
||||
nextIndexById.delete(activationStream.id)
|
||||
nextIndexById.set(current.id, activationStreamIndex)
|
||||
overlaid[activationStreamIndex] = replacement
|
||||
activationStreamIndex = -1
|
||||
changed = true
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
nextIndexById.set(current.id, overlaid.length)
|
||||
overlaid.push(current)
|
||||
changed = true
|
||||
}
|
||||
|
||||
return changed ? overlaid : nextMessages
|
||||
}
|
||||
|
||||
export interface BranchMessage {
|
||||
content: string
|
||||
role: ChatMessage['role']
|
||||
|
||||
Reference in New Issue
Block a user