From 8d8d1d61fe3f92e5cf5aa4a11ac7209d540d89a1 Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Fri, 24 Jul 2026 23:31:55 -0500 Subject: [PATCH] feat(desktop): crash-survivable in-flight turn journal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The renderer's session-state cache is memory-only and the backend's inflight snapshot dies with the backend process, so nothing survived a full app or machine death mid-turn: reopening the session showed the transcript up to the last committed turn and silently dropped everything the crashed turn had streamed. While a turn runs, the visible tail (user prompt + streamed assistant rows, tool calls included) is now journaled to localStorage — throttled off the delta-flush hot path, bounded (24 entries / 7 days), cleared the moment the turn settles. Session resume folds the journaled tail back onto the restored transcript. When the backend also has a live text-only inflight projection for the same turn, the journal overlays its richer structure onto that row (longer text wins, base row id kept so live deltas keep landing) instead of treating it as caught up — the ordering defect that dropped locally recorded tool progress in the original PR. Co-authored-by: Omar Baradei --- .../hooks/use-message-stream/gateway-event.ts | 21 +- .../session/hooks/use-message-stream/index.ts | 29 +- .../terminal-error-frame.test.tsx | 119 +++++ .../hooks/use-session-actions/index.ts | 58 +- .../hooks/use-session-actions/utils.ts | 11 +- .../session/hooks/use-session-state-cache.ts | 5 + apps/desktop/src/lib/chat-messages.test.ts | 17 +- apps/desktop/src/lib/chat-messages.ts | 14 +- .../src/lib/inflight-turn-journal.test.ts | 240 +++++++++ apps/desktop/src/lib/inflight-turn-journal.ts | 504 ++++++++++++++++++ apps/desktop/src/types/hermes.ts | 12 + hermes_cli/config.py | 11 + tests/tui_gateway/test_auto_continue.py | 423 +++++++++++++++ tui_gateway/turn_marker.py | 159 ++++++ 14 files changed, 1601 insertions(+), 22 deletions(-) create mode 100644 apps/desktop/src/app/session/hooks/use-message-stream/terminal-error-frame.test.tsx create mode 100644 apps/desktop/src/lib/inflight-turn-journal.test.ts create mode 100644 apps/desktop/src/lib/inflight-turn-journal.ts create mode 100644 tests/tui_gateway/test_auto_continue.py create mode 100644 tui_gateway/turn_marker.py diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts index 32f8eb1736d3..101fe7e59be5 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts +++ b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts @@ -128,7 +128,12 @@ interface GatewayEventDeps { nativeSubagentSessionsRef: MutableRefObject> appendAssistantDelta: (sessionId: string, delta: string) => void appendReasoningDelta: (sessionId: string, delta: string, replace?: boolean) => void - completeAssistantMessage: (sessionId: string, text: string, responsePreviewed?: boolean) => void + completeAssistantMessage: ( + sessionId: string, + text: string, + responsePreviewed?: boolean, + failure?: { error: string; partial: boolean } + ) => void failAssistantMessage: (sessionId: string, errorMessage: string) => void flushQueuedDeltas: (sessionId?: string) => void finalizeInterimAssistantMessage: (sessionId: string, text: string) => void @@ -612,7 +617,19 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) { playCompletionSound(sessionId) const finalText = coerceGatewayText(payload?.text) || coerceGatewayText(payload?.rendered) - completeAssistantMessage(sessionId, finalText, payload?.response_previewed) + + // Terminal error frames (status "error") carry the failure in + // structured fields: `error` is the message, and `partial` marks + // `text` as streamed output to keep rather than the error string. + const failure = + payload?.status === 'error' + ? { + error: coerceGatewayText(payload.error).trim() || finalText || 'Hermes reported an error', + partial: Boolean(payload.partial) + } + : undefined + + completeAssistantMessage(sessionId, finalText, payload?.response_previewed, failure) // Structured billing wall forwarded by the gateway (out of credits / // payment required) — cache it + raise a billing-specific toast. diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/index.ts b/apps/desktop/src/app/session/hooks/use-message-stream/index.ts index 53a17b193c61..de62d0bb89d7 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/index.ts +++ b/apps/desktop/src/app/session/hooks/use-message-stream/index.ts @@ -437,7 +437,12 @@ export function useMessageStream({ ) const completeAssistantMessage = useCallback( - (sessionId: string, text: string, responsePreviewed?: boolean) => { + ( + sessionId: string, + text: string, + responsePreviewed?: boolean, + failure?: { error: string; partial: boolean } + ) => { let shouldHydrate = false const completedState = updateSessionState(sessionId, state => { @@ -459,7 +464,13 @@ export function useMessageStream({ const streamId = state.streamId const finalText = renderMediaTags(text).trim() - const completionError = completionErrorText(finalText) + // Structured failure from the terminal frame wins over the legacy text + // heuristic ("Error: " texts don't match the regexes). + const completionError = failure?.error ?? completionErrorText(finalText) + // A partial failure's `text` is streamed output the user should keep, + // not the error string — settle it like a normal reply AND mark the + // bubble failed, instead of stripping the text. + const keepFailedPartialText = Boolean(failure?.partial && finalText) const interimBoundaryPending = state.interimBoundaryPending const replaceTextPart = (parts: ChatMessagePart[]) => { @@ -473,15 +484,21 @@ export function useMessageStream({ const completeMessage = (message: ChatMessage): ChatMessage => { const settled = { ...message, pending: false, interim: false } - return completionError - ? { ...settled, error: completionError, parts: message.parts.filter(part => part.type !== 'text') } - : { ...settled, parts: replaceTextPart(message.parts) } + if (completionError && !keepFailedPartialText) { + return { ...settled, error: completionError, parts: message.parts.filter(part => part.type !== 'text') } + } + + return { + ...settled, + parts: replaceTextPart(message.parts), + ...(completionError ? { error: completionError } : {}) + } } const newAssistantFromCompletion = (): ChatMessage => ({ id: `assistant-${Date.now()}`, role: 'assistant', - parts: completionError ? [] : [assistantTextPart(finalText)], + parts: completionError && !keepFailedPartialText ? [] : [assistantTextPart(finalText)], branchGroupId: state.pendingBranchGroup ?? undefined, ...(completionError && { error: completionError }) }) diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/terminal-error-frame.test.tsx b/apps/desktop/src/app/session/hooks/use-message-stream/terminal-error-frame.test.tsx new file mode 100644 index 000000000000..a2c775a91249 --- /dev/null +++ b/apps/desktop/src/app/session/hooks/use-message-stream/terminal-error-frame.test.tsx @@ -0,0 +1,119 @@ +import { QueryClient } from '@tanstack/react-query' +import { act, cleanup, render, waitFor } from '@testing-library/react' +import { useEffect, useRef } from 'react' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +import type { ClientSessionState } from '@/app/types' +import { chatMessageText } from '@/lib/chat-messages' +import { createClientSessionState } from '@/lib/chat-runtime' +import type { RpcEvent } from '@/types/hermes' + +import { useMessageStream } from './index' + +const SID = 'session-1' + +let handleEvent: ((event: RpcEvent) => void) | null = null +let sessionStates: Map + +function Harness() { + const activeSessionIdRef = useRef(SID) + const sessionStateByRuntimeIdRef = useRef(new Map()) + const queryClientRef = useRef(new QueryClient()) + + const stream = useMessageStream({ + activeSessionIdRef, + hydrateFromStoredSession: vi.fn(async () => undefined), + queryClient: queryClientRef.current, + refreshHermesConfig: vi.fn(async () => undefined), + refreshSessions: vi.fn(async () => undefined), + sessionStateByRuntimeIdRef, + updateSessionState: (sessionId, updater) => { + const current = sessionStateByRuntimeIdRef.current.get(sessionId) ?? createClientSessionState() + const next = updater(current) + sessionStateByRuntimeIdRef.current.set(sessionId, next) + sessionStates.set(sessionId, next) + + return next + } + }) + + useEffect(() => { + handleEvent = stream.handleGatewayEvent + }, [stream.handleGatewayEvent]) + + return null +} + +async function mountStream() { + sessionStates = new Map() + render() + await waitFor(() => expect(handleEvent).not.toBeNull()) +} + +const start = () => act(() => handleEvent!({ payload: {}, session_id: SID, type: 'message.start' })) +const delta = (text: string) => act(() => handleEvent!({ payload: { text }, session_id: SID, type: 'message.delta' })) + +const completeWithError = (payload: Record) => + act(() => handleEvent!({ payload: { status: 'error', ...payload }, session_id: SID, type: 'message.complete' })) + +function getState(): ClientSessionState { + return sessionStates.get(SID) ?? createClientSessionState() +} + +function lastAssistant() { + return [...getState().messages].reverse().find(m => m.role === 'assistant' && !m.hidden) +} + +describe('terminal error message.complete frames', () => { + beforeEach(() => { + handleEvent = null + }) + + afterEach(() => { + cleanup() + vi.restoreAllMocks() + }) + + it('marks the bubble failed from the structured error field, not the text heuristic', async () => { + await mountStream() + await start() + await delta('…') + + // "Error: " does not match the legacy completionErrorText regexes. + await completeWithError({ text: 'Error: invalid model slug', error: 'invalid model slug', recoverable: true }) + + const bubble = lastAssistant() + expect(bubble?.error).toBe('invalid model slug') + expect(getState().busy).toBe(false) + expect(getState().awaitingResponse).toBe(false) + }) + + it('keeps streamed partial text visible on a partial failure', async () => { + await mountStream() + await start() + await delta('half an ans') + + await completeWithError({ + text: 'half an ans', + error: 'connection reset mid-stream', + partial: true, + recoverable: true + }) + + const bubble = lastAssistant() + expect(bubble?.error).toBe('connection reset mid-stream') + expect(chatMessageText(bubble!)).toBe('half an ans') + expect(bubble?.pending).toBe(false) + }) + + it('falls back to the frame text when no error field is present', async () => { + await mountStream() + await start() + await delta('…') + + await completeWithError({ text: 'Error: something broke' }) + + const bubble = lastAssistant() + expect(bubble?.error).toBe('Error: something broke') + }) +}) diff --git a/apps/desktop/src/app/session/hooks/use-session-actions/index.ts b/apps/desktop/src/app/session/hooks/use-session-actions/index.ts index 95730277cd3f..38cf05920898 100644 --- a/apps/desktop/src/app/session/hooks/use-session-actions/index.ts +++ b/apps/desktop/src/app/session/hooks/use-session-actions/index.ts @@ -7,6 +7,7 @@ import { deleteSession, getSessionMessages, setSessionArchived } from '@/hermes' import { useI18n } from '@/i18n' import { type ChatMessage, preserveLocalAssistantErrors, toChatMessages } from '@/lib/chat-messages' import { isMissingRpcMethod } from '@/lib/gateway-rpc' +import { recoverInFlightTurnJournal } from '@/lib/inflight-turn-journal' import { setSessionYolo } from '@/lib/yolo-session' import { migrateSessionDraft } from '@/store/composer' import { clearQueuedPrompts, migrateQueuedPrompts } from '@/store/composer-queue' @@ -821,6 +822,10 @@ export function useSessionActions({ } let resumedRunning = false + // A recovered in-flight tail means the turn already produced output, so + // it resumes into the streaming state rather than the "awaiting first + // token" spinner. + let recoveredInFlightTail = false try { const watchWindow = isWatchWindow() @@ -912,15 +917,36 @@ export function useSessionActions({ return chatMessageArraysEquivalent(currentMessages, resumedMessages) ? currentMessages : resumedMessages })() + resumedRunning = Boolean((resumed as { running?: boolean }).running) + + // Crash-survivable turn progress: fold a journaled in-flight tail + // (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, { + 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. const messagesForView = - preferredMessages === currentMessages + inFlightRecovery.messages === currentMessages ? currentMessages - : preserveLocalAssistantErrors(preferredMessages, currentMessages) + : preserveLocalAssistantErrors(inFlightRecovery.messages, currentMessages) - if (sessionShouldHaveTranscript(stored) && messagesForView.length === 0) { + // Fail-latch on the PRE-recovery transcript: an orphan journal tail + // must not mask a lost transcript (a retry that reloads real history + // is safer than surfacing the in-flight turn alone). Recovery only + // ever appends, so this matches the final transcript's emptiness. + if (sessionShouldHaveTranscript(stored) && preferredMessages.length === 0) { setActiveSessionId(null) activeSessionIdRef.current = null setResumeFailedSessionId(storedSessionId) @@ -935,8 +961,6 @@ export function useSessionActions({ patchSessionWorkspace(storedSessionId, runtimeInfo?.cwd) - resumedRunning = Boolean((resumed as { running?: boolean }).running) - updateSessionState( resumed.session_id, state => ({ @@ -944,7 +968,18 @@ export function useSessionActions({ ...(runtimeInfo ?? {}), messages: messagesForView, busy: resumedRunning, - awaitingResponse: resumedRunning + awaitingResponse: resumedRunning && !recoveredInFlightTail, + ...(inFlightRecovery.applied + ? { + sawAssistantPayload: true, + // Point live deltas at the recovered row when the backend is + // still mid-turn; a settled recovery keeps the stream idle. + streamId: resumedRunning ? inFlightRecovery.streamId : null, + turnStartedAt: resumedRunning + ? (inFlightRecovery.turnStartedAt ?? state.turnStartedAt ?? Date.now()) + : state.turnStartedAt + } + : {}) }), storedSessionId ) @@ -982,7 +1017,14 @@ export function useSessionActions({ ? preserveLocalPendingTurnMessages($messages.get(), resumeStartMessages) : $messages.get() - setMessages(reconcileAuthoritativeMessages(fallback.messages, previousMessages)) + // Resume failed, so there is no live projection — the journal is the + // only carrier of a crashed turn's progress on this path. + const fallbackRecovery = recoverInFlightTurnJournal( + storedSessionId, + reconcileAuthoritativeMessages(fallback.messages, previousMessages) + ) + + setMessages(fallbackRecovery.messages) } catch (e) { // Fallback also failed: nothing to paint. Leave whatever messages are // already shown and fall through to arm the resume-failure latch so @@ -1034,7 +1076,7 @@ export function useSessionActions({ if (isCurrentResume()) { busyRef.current = resumedRunning setBusy(resumedRunning) - setAwaitingResponse(resumedRunning) + setAwaitingResponse(resumedRunning && !recoveredInFlightTail) } } }, diff --git a/apps/desktop/src/app/session/hooks/use-session-actions/utils.ts b/apps/desktop/src/app/session/hooks/use-session-actions/utils.ts index bc5cb6cec3b5..e5caa5f64a78 100644 --- a/apps/desktop/src/app/session/hooks/use-session-actions/utils.ts +++ b/apps/desktop/src/app/session/hooks/use-session-actions/utils.ts @@ -349,9 +349,13 @@ export function appendLiveSessionProjection( const inflightUser = projection.inflight?.user?.trim() ?? '' const inflightAssistant = projection.inflight?.assistant ?? '' const inflightStreaming = Boolean(projection.inflight?.streaming) + // A retained failed turn (the gateway keeps error snapshots replayable when + // the terminal frame may have been lost to a disconnect) — surface the + // failure on the projected row instead of rendering the partial as healthy. + const inflightError = projection.inflight?.error?.trim() ?? '' const queuedUser = projection.queued?.user?.trim() ?? '' - if (!inflightUser && !inflightAssistant && !inflightStreaming && !queuedUser) { + if (!inflightUser && !inflightAssistant && !inflightStreaming && !inflightError && !queuedUser) { return messages } @@ -376,12 +380,13 @@ export function appendLiveSessionProjection( // Keep a pending assistant boundary even before the first delta when a // queued user turn follows it. This preserves the two distinct turns. - if (inflightAssistant || inflightStreaming || (inflightUser && queuedUser)) { + if (inflightAssistant || inflightStreaming || inflightError || (inflightUser && queuedUser)) { projected.push({ id: `assistant-stream-${sessionId}`, role: 'assistant', parts: inflightAssistant ? [assistantTextPart(inflightAssistant)] : [], - pending: inflightStreaming + pending: inflightStreaming, + ...(inflightError ? { error: inflightError } : {}) }) } diff --git a/apps/desktop/src/app/session/hooks/use-session-state-cache.ts b/apps/desktop/src/app/session/hooks/use-session-state-cache.ts index 2e732cb59254..25521257c5bb 100644 --- a/apps/desktop/src/app/session/hooks/use-session-state-cache.ts +++ b/apps/desktop/src/app/session/hooks/use-session-state-cache.ts @@ -4,6 +4,7 @@ import { type MutableRefObject, useCallback, useEffect, useRef } from 'react' import type { ChatMessage } from '@/lib/chat-messages' import { preserveLocalAssistantErrors } from '@/lib/chat-messages' import { createClientSessionState } from '@/lib/chat-runtime' +import { persistInFlightTurnState } from '@/lib/inflight-turn-journal' import { setMutableRef } from '@/lib/mutable-ref' import { $busy, @@ -268,6 +269,10 @@ export function useSessionStateCache({ const previous = ensureSessionState(sessionId, storedSessionId) const next = updater({ ...previous, messages: previous.messages }) sessionStateByRuntimeIdRef.current.set(sessionId, next) + // Crash-survivable turn progress: journal the running turn's visible + // tail (throttled localStorage write; cleared the moment the turn + // settles) so a renderer/app death mid-turn can be recovered on resume. + persistInFlightTurnState(next) // Publishing to $sessionStates automatically fires transition side-effects // (watchdog, settle grace, unread marker, compression id rotation) inside // publishSessionState — no manual transition call needed. diff --git a/apps/desktop/src/lib/chat-messages.test.ts b/apps/desktop/src/lib/chat-messages.test.ts index 14bad6075bc7..b7cd083aa619 100644 --- a/apps/desktop/src/lib/chat-messages.test.ts +++ b/apps/desktop/src/lib/chat-messages.test.ts @@ -241,15 +241,28 @@ describe('toChatMessages', () => { content: 'opaque delegation context payload', display_kind: 'async_delegation_complete', timestamp: 5 + }, + { + role: 'user', + content: '[System note: Your previous turn was interrupted mid-run…]\n\noriginal prompt', + display_kind: 'auto_continue', + timestamp: 6 } ]) - expect(messages.map(message => message.role)).toEqual(['user', 'assistant', 'system', 'system']) + expect(messages.map(message => message.role)).toEqual([ + 'user', + 'assistant', + 'system', + 'system', + 'system' + ]) expect(messages.map(chatMessageText)).toEqual([ 'real user turn', 'real assistant reply', 'model changed', - 'background agent work finished' + 'background agent work finished', + 'resumed interrupted turn' ]) }) diff --git a/apps/desktop/src/lib/chat-messages.ts b/apps/desktop/src/lib/chat-messages.ts index 8c01d3214b36..f0d8e2b7c641 100644 --- a/apps/desktop/src/lib/chat-messages.ts +++ b/apps/desktop/src/lib/chat-messages.ts @@ -102,6 +102,12 @@ export type GatewayEventPayload = { // message.complete — signals the final text was already previewed via // interim_assistant_callback, so the UI can settle instead of duplicating. response_previewed?: boolean + // message.complete with status "error" — `text` is streamed partial output + // (keep it visible), not the error string. + partial?: boolean + // message.complete with status "error" — the failed turn was retained + // backend-side and will replay through session.resume's inflight payload. + recoverable?: boolean // Structured billing wall forwarded on message.complete when a turn fails // with FailoverReason.billing (shape mirrors @hermes/shared BillingBlock). billing?: BillingBlock @@ -339,6 +345,10 @@ function timelineDisplayContent(message: SessionMessage, content: string): strin return 'model changed' } + if (message.display_kind === 'auto_continue') { + return 'resumed interrupted turn' + } + if (message.display_kind === 'async_delegation_complete') { const count = timelineTaskCount(message.display_metadata) @@ -938,7 +948,9 @@ export function toChatMessages(messages: SessionMessage[]): ChatMessage[] { ) const displayRole = - message.display_kind === 'model_switch' || message.display_kind === 'async_delegation_complete' + message.display_kind === 'model_switch' || + message.display_kind === 'async_delegation_complete' || + message.display_kind === 'auto_continue' ? 'system' : message.role diff --git a/apps/desktop/src/lib/inflight-turn-journal.test.ts b/apps/desktop/src/lib/inflight-turn-journal.test.ts new file mode 100644 index 000000000000..ffa1c4300651 --- /dev/null +++ b/apps/desktop/src/lib/inflight-turn-journal.test.ts @@ -0,0 +1,240 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +import type { ChatMessage } from '@/lib/chat-messages' +import { + clearInFlightTurnJournal, + type JournalableSessionState, + mergeInFlightMessages, + persistInFlightTurnState, + readInFlightTurnJournal, + recoverInFlightTurnJournal +} from '@/lib/inflight-turn-journal' + +const STORAGE_KEY = 'hermes.desktop.inflightTurnJournal.v1' + +function user(id: string, text: string): ChatMessage { + return { id, role: 'user', parts: [{ type: 'text', text }] } +} + +function assistant(id: string, text: string, extra: Partial = {}): ChatMessage { + return { id, role: 'assistant', parts: [{ type: 'text', text }], ...extra } +} + +function assistantWithTool(id: string, text: string, extra: Partial = {}): ChatMessage { + return { + id, + role: 'assistant', + parts: [ + { type: 'tool-call', toolCallId: 'tc-1', toolName: 'terminal', args: { command: 'ls' } }, + { type: 'text', text } + ], + ...extra + } +} + +function journalState(overrides: Partial = {}): JournalableSessionState { + return { + awaitingResponse: false, + busy: true, + messages: [user('u1', 'do the thing'), assistant('assistant-stream-1', 'partial answer', { pending: true })], + storedSessionId: 'stored-1', + streamId: 'assistant-stream-1', + turnStartedAt: 1000, + ...overrides + } +} + +beforeEach(() => { + vi.useFakeTimers() + window.localStorage.clear() +}) + +afterEach(() => { + clearInFlightTurnJournal('stored-1') + vi.useRealTimers() +}) + +describe('persistInFlightTurnState', () => { + it('journals the running turn tail after the throttle window', () => { + persistInFlightTurnState(journalState()) + + expect(readInFlightTurnJournal('stored-1')).toBeNull() + + vi.advanceTimersByTime(400) + + const entry = readInFlightTurnJournal('stored-1') + expect(entry).not.toBeNull() + expect(entry?.streamId).toBe('assistant-stream-1') + expect(entry?.turnStartedAt).toBe(1000) + expect(entry?.messages.map(m => m.role)).toEqual(['user', 'assistant']) + }) + + it('coalesces rapid updates into one write carrying the latest state', () => { + persistInFlightTurnState(journalState()) + persistInFlightTurnState( + journalState({ + messages: [ + user('u1', 'do the thing'), + assistant('assistant-stream-1', 'partial answer grew', { pending: true }) + ] + }) + ) + + vi.advanceTimersByTime(400) + + const entry = readInFlightTurnJournal('stored-1') + const tail = entry?.messages.find(m => m.role === 'assistant') + expect(tail?.parts).toEqual([{ type: 'text', text: 'partial answer grew' }]) + }) + + it('clears the entry the moment the turn settles, cancelling pending writes', () => { + persistInFlightTurnState(journalState()) + vi.advanceTimersByTime(400) + expect(readInFlightTurnJournal('stored-1')).not.toBeNull() + + persistInFlightTurnState(journalState({ messages: [] })) + persistInFlightTurnState(journalState({ busy: false, awaitingResponse: false, streamId: null })) + + expect(readInFlightTurnJournal('stored-1')).toBeNull() + + vi.advanceTimersByTime(1000) + expect(readInFlightTurnJournal('stored-1')).toBeNull() + }) + + it('does not journal a turn with no recoverable assistant content yet', () => { + persistInFlightTurnState(journalState({ messages: [user('u1', 'do the thing')], streamId: null })) + + vi.advanceTimersByTime(400) + expect(readInFlightTurnJournal('stored-1')).toBeNull() + }) + + it('expires entries older than the max age', () => { + persistInFlightTurnState(journalState()) + vi.advanceTimersByTime(400) + + const raw = JSON.parse(window.localStorage.getItem(STORAGE_KEY)!) + raw.entries['stored-1'].updatedAt = Date.now() - 8 * 24 * 60 * 60 * 1000 + window.localStorage.setItem(STORAGE_KEY, JSON.stringify(raw)) + + expect(readInFlightTurnJournal('stored-1')).toBeNull() + }) +}) + +describe('recoverInFlightTurnJournal', () => { + function journalEntry(messages: ChatMessage[]) { + persistInFlightTurnState(journalState({ messages, streamId: messages.at(-1)?.id ?? null })) + vi.advanceTimersByTime(400) + } + + it('is a reference-preserving no-op when nothing is journaled', () => { + const base = [user('u1', 'do the thing')] + const result = recoverInFlightTurnJournal('stored-1', base) + + expect(result.applied).toBe(false) + expect(result.messages).toBe(base) + }) + + it('appends the full tail when the base transcript never saw the turn', () => { + journalEntry([ + user('u1', 'do the thing'), + assistantWithTool('assistant-stream-1', 'working on it', { pending: true }) + ]) + + const base = [user('u0', 'earlier turn'), assistant('a0', 'earlier reply')] + const result = recoverInFlightTurnJournal('stored-1', base) + + expect(result.applied).toBe(true) + expect(result.messages.map(m => m.id)).toEqual(['u0', 'a0', 'u1', 'assistant-stream-1']) + expect(result.streamId).toBe('assistant-stream-1') + }) + + it('appends only the assistant tail when the user row was persisted', () => { + journalEntry([ + user('u1', 'do the thing'), + assistantWithTool('assistant-stream-1', 'working on it', { pending: true }) + ]) + + const base = [user('db-u1', 'do the thing')] + const result = recoverInFlightTurnJournal('stored-1', base, { keepPending: false }) + + expect(result.applied).toBe(true) + expect(result.messages.map(m => m.id)).toEqual(['db-u1', 'assistant-stream-1']) + const tail = result.messages.at(-1)! + expect(tail.pending).toBe(false) + expect(tail.parts[0]).toMatchObject({ type: 'tool-call' }) + }) + + it('detects a committed reply as caught up and clears the entry', () => { + journalEntry([user('u1', 'do the thing'), assistant('assistant-stream-1', 'partial', { pending: true })]) + + const base = [user('db-u1', 'do the thing'), assistant('db-a1', 'full committed reply')] + const result = recoverInFlightTurnJournal('stored-1', base) + + expect(result.applied).toBe(false) + expect(result.caughtUp).toBe(true) + expect(result.messages).toBe(base) + expect(readInFlightTurnJournal('stored-1')).toBeNull() + }) + + it('overlays the backend text-only projection instead of dropping local tool progress', () => { + // Sweeper regression on #44339: a backend `inflight` assistant snapshot + // (text only) used to mark the richer local tail "caught up" and delete + // locally recorded tool calls. + journalEntry([ + user('u1', 'do the thing'), + assistantWithTool('assistant-stream-old', 'local part', { pending: true }) + ]) + + const base = [ + user('db-u1', 'do the thing'), + assistant('assistant-stream-rt9', 'longer partial text from the backend snapshot', { pending: true }) + ] + + const result = recoverInFlightTurnJournal('stored-1', base, { keepPending: true }) + + expect(result.applied).toBe(true) + expect(result.caughtUp).toBe(false) + expect(result.messages).toHaveLength(2) + + const merged = result.messages.at(-1)! + // Keeps the BASE projection row id so live deltas keep landing on it. + expect(merged.id).toBe('assistant-stream-rt9') + expect(result.streamId).toBe('assistant-stream-rt9') + // Journal structure survives; the longer backend text wins. + expect(merged.parts[0]).toMatchObject({ type: 'tool-call', toolName: 'terminal' }) + expect(merged.parts[1]).toMatchObject({ type: 'text', text: 'longer partial text from the backend snapshot' }) + // Still in flight — the journal must NOT be cleared. + expect(readInFlightTurnJournal('stored-1')).not.toBeNull() + }) + + it('keeps the journal text when it is longer than the projection text', () => { + journalEntry([ + user('u1', 'do the thing'), + assistantWithTool('assistant-stream-old', 'a much longer locally journaled partial answer', { pending: true }) + ]) + + const base = [user('db-u1', 'do the thing'), assistant('assistant-stream-rt9', 'thin', { pending: true })] + const result = recoverInFlightTurnJournal('stored-1', base, { keepPending: true }) + + const merged = result.messages.at(-1)! + expect(merged.id).toBe('assistant-stream-rt9') + expect(merged.parts[1]).toMatchObject({ type: 'text', text: 'a much longer locally journaled partial answer' }) + }) +}) + +describe('mergeInFlightMessages', () => { + it('treats an error-bearing assistant row as recoverable content', () => { + const tail = [user('u1', 'do the thing'), assistant('a-err', '', { error: 'provider exploded' })] + const result = mergeInFlightMessages([user('db-u1', 'do the thing')], tail) + + expect(result.applied).toBe(true) + expect(result.messages.at(-1)?.error).toBe('provider exploded') + }) + + it('ignores hidden rows when extracting nothing to recover', () => { + const result = mergeInFlightMessages([], [user('u1', 'x')]) + + expect(result.applied).toBe(false) + expect(result.caughtUp).toBe(false) + }) +}) diff --git a/apps/desktop/src/lib/inflight-turn-journal.ts b/apps/desktop/src/lib/inflight-turn-journal.ts new file mode 100644 index 000000000000..8056fa7af102 --- /dev/null +++ b/apps/desktop/src/lib/inflight-turn-journal.ts @@ -0,0 +1,504 @@ +import { type ChatMessage, type ChatMessagePart, chatMessageText } from '@/lib/chat-messages' + +/** + * Crash-survivable in-flight turn journal. + * + * While a session is busy, the visible tail of the running turn (user prompt + + * streamed assistant rows, tool calls included) is persisted to localStorage. + * If the renderer or the whole app dies mid-turn, session resume folds the + * journaled tail back onto the restored transcript, so streamed progress is + * not silently lost. The backend's own `inflight` snapshot (merged by + * `appendLiveSessionProjection`) covers reconnects while the backend is alive; + * this journal covers the cases where the backend died too — and it is richer, + * because the backend snapshot carries text only while the journal keeps the + * full part structure. + * + * Best-effort by design: storage failures must never break chat streaming. + */ + +const STORAGE_KEY = 'hermes.desktop.inflightTurnJournal.v1' +const STORE_VERSION = 1 +const MAX_ENTRIES = 24 +const MAX_AGE_MS = 7 * 24 * 60 * 60 * 1000 +/** Streaming repaints arrive every ~33ms; localStorage writes are synchronous. + * Trailing-edge throttle keeps the journal off the hot path — a crash costs at + * most this much of the newest tail. */ +const PERSIST_THROTTLE_MS = 400 + +export interface InFlightTurnSnapshot { + messages: ChatMessage[] + streamId: null | string + turnStartedAt: null | number + updatedAt: number +} + +export interface JournalableSessionState { + awaitingResponse: boolean + busy: boolean + messages: ChatMessage[] + storedSessionId: null | string + streamId: null | string + turnStartedAt: null | number +} + +interface JournalStore { + entries: Record + version: typeof STORE_VERSION +} + +export interface InFlightRecoveryResult { + applied: boolean + /** The base transcript already contains the journaled turn's completed + * reply — the journal entry is stale and has been cleared. */ + caughtUp: boolean + messages: ChatMessage[] + streamId: null | string + turnStartedAt: null | number +} + +function storage(): Storage | null { + try { + return typeof window === 'undefined' ? null : window.localStorage + } catch { + return null + } +} + +function emptyStore(): JournalStore { + return { entries: {}, version: STORE_VERSION } +} + +function loadStore(): JournalStore { + const store = storage() + + if (!store) { + return emptyStore() + } + + try { + const raw = store.getItem(STORAGE_KEY) + + if (!raw) { + return emptyStore() + } + + const parsed = JSON.parse(raw) + + if ( + !parsed || + parsed.version !== STORE_VERSION || + typeof parsed.entries !== 'object' || + Array.isArray(parsed.entries) + ) { + return emptyStore() + } + + return { + entries: parsed.entries as Record, + version: STORE_VERSION + } + } catch { + return emptyStore() + } +} + +function saveStore(journal: JournalStore): void { + const store = storage() + + if (!store) { + return + } + + try { + const entries = Object.fromEntries( + Object.entries(journal.entries) + .filter(([, entry]) => !isExpired(entry)) + .sort((a, b) => b[1].updatedAt - a[1].updatedAt) + .slice(0, MAX_ENTRIES) + ) + + if (Object.keys(entries).length === 0) { + store.removeItem(STORAGE_KEY) + + return + } + + store.setItem(STORAGE_KEY, JSON.stringify({ entries, version: STORE_VERSION })) + } catch { + // Quota/private-mode failures: the journal is a recovery aid, not truth. + } +} + +function isExpired(entry: InFlightTurnSnapshot, now = Date.now()): boolean { + return now - entry.updatedAt > MAX_AGE_MS +} + +function cloneMessages(messages: ChatMessage[]): ChatMessage[] { + try { + return JSON.parse(JSON.stringify(messages)) as ChatMessage[] + } catch { + return [] + } +} + +function normalizedText(value: string): string { + return value.replace(/\s+/g, ' ').trim() +} + +function attachmentSignature(message: ChatMessage): string { + return (message.attachmentRefs ?? []).join('\n') +} + +function userMessagesMatch(left: ChatMessage, right: ChatMessage): boolean { + return ( + left.role === 'user' && + right.role === 'user' && + normalizedText(chatMessageText(left)) === normalizedText(chatMessageText(right)) && + attachmentSignature(left) === attachmentSignature(right) + ) +} + +function partHasRecoverableContent(part: ChatMessagePart): boolean { + if (part.type === 'text' || part.type === 'reasoning') { + return typeof part.text === 'string' && part.text.trim().length > 0 + } + + return part.type === 'tool-call' +} + +function assistantHasRecoverableContent(message: ChatMessage): boolean { + return message.role === 'assistant' && (Boolean(message.error) || message.parts.some(partHasRecoverableContent)) +} + +/** A live-turn projection row (backend `inflight` via appendLiveSessionProjection, + * or a still-streaming local bubble) — as opposed to a completed transcript row. */ +function isLiveProjectionRow(message: ChatMessage): boolean { + return ( + Boolean(message.pending) || + message.id.startsWith('assistant-stream-') || + message.id.startsWith('inflight-assistant-') + ) +} + +/** Visible tail of the running turn: the streaming assistant row (plus any + * interim rows sealed after it) back to the user prompt that started it. */ +function recoverableTail(messages: ChatMessage[], streamId: null | string): ChatMessage[] { + const visible = messages.filter(message => !message.hidden) + let assistantIndex = -1 + + if (streamId) { + assistantIndex = visible.findIndex(message => message.id === streamId && assistantHasRecoverableContent(message)) + } + + if (assistantIndex < 0) { + for (let index = visible.length - 1; index >= 0; index -= 1) { + const message = visible[index] + + if (message.role === 'user') { + break + } + + if (assistantHasRecoverableContent(message)) { + assistantIndex = index + + break + } + } + } + + if (assistantIndex < 0) { + return [] + } + + let start = assistantIndex + + for (let index = assistantIndex - 1; index >= 0; index -= 1) { + if (visible[index].role === 'user') { + start = index + + break + } + } + + return cloneMessages(visible.slice(start)) +} + +function normalizeRecoveredTail(tail: ChatMessage[], keepPending: boolean): ChatMessage[] { + return cloneMessages(tail).map(message => + message.role === 'assistant' + ? { + ...message, + pending: keepPending ? (message.pending ?? true) : false + } + : { ...message, pending: false } + ) +} + +function assistantTextLength(message: ChatMessage): number { + return chatMessageText(message).length +} + +/** Merge the journal's last assistant row into the base's live projection row. + * + * The journal carries structure (tool calls, reasoning) the backend snapshot + * lacks; the backend text may be newer than the journal's last throttled + * write. Keep the journal's parts, but let the longer text win — and keep the + * BASE row's id so live deltas keep appending to the row the stream handler + * already targets. + */ +function overlayProjectionRow(projection: ChatMessage, journalRow: ChatMessage): ChatMessage { + // A projected error (retained failed turn) must survive the overlay. + const error = journalRow.error ?? projection.error + + const merged: ChatMessage = { + ...journalRow, + id: projection.id, + pending: projection.pending, + ...(error ? { error } : {}) + } + + if (assistantTextLength(projection) <= assistantTextLength(journalRow)) { + return merged + } + + // Backend text is newer than the journal's last throttled write — swap it + // into the journal's first text part, keeping tool calls and reasoning. + const projectionText = chatMessageText(projection) + const parts: ChatMessagePart[] = [] + let textReplaced = false + + for (const part of journalRow.parts) { + if (part.type !== 'text') { + parts.push(part) + } else if (!textReplaced) { + parts.push({ ...part, text: projectionText }) + textReplaced = true + } + } + + if (!textReplaced) { + parts.push({ type: 'text', text: projectionText }) + } + + return { ...merged, parts } +} + +export function mergeInFlightMessages( + baseMessages: ChatMessage[], + tailMessages: ChatMessage[], + options: { keepPending?: boolean } = {} +): InFlightRecoveryResult { + const noop: InFlightRecoveryResult = { + applied: false, + caughtUp: false, + messages: baseMessages, + streamId: null, + turnStartedAt: null + } + + const tail = normalizeRecoveredTail(tailMessages, Boolean(options.keepPending)) + + if (!tail.some(assistantHasRecoverableContent)) { + return noop + } + + const tailUserIndex = tail.findIndex(message => message.role === 'user') + const tailUser = tailUserIndex >= 0 ? tail[tailUserIndex] : null + const tailAssistants = tail.slice(tailUserIndex + 1) + const lastJournalRow = tailAssistants.findLast(assistantHasRecoverableContent) ?? null + const matchingUserIndex = tailUser ? baseMessages.findLastIndex(message => userMessagesMatch(message, tailUser)) : -1 + + if (matchingUserIndex < 0) { + // Base doesn't know this turn at all (user row was never persisted): + // append the whole tail. + const streamId = lastJournalRow?.id ?? null + + return { applied: true, caughtUp: false, messages: [...baseMessages, ...tail], streamId, turnStartedAt: null } + } + + const afterUser = baseMessages.slice(matchingUserIndex + 1) + + const completedReply = afterUser.find( + message => assistantHasRecoverableContent(message) && !isLiveProjectionRow(message) + ) + + if (completedReply) { + // The transcript already holds this turn's committed reply — the journal + // entry is stale. + return { ...noop, caughtUp: true } + } + + const projectionIndex = baseMessages.findIndex( + (message, index) => index > matchingUserIndex && message.role === 'assistant' && isLiveProjectionRow(message) + ) + + if (projectionIndex < 0) { + if (tailAssistants.length === 0) { + return noop + } + + const streamId = lastJournalRow?.id ?? null + + return { + applied: true, + caughtUp: false, + messages: [...baseMessages, ...tailAssistants], + streamId, + turnStartedAt: null + } + } + + // Backend projection row present (text-only): overlay the journal's + // structure onto it instead of treating it as "caught up" — that is how + // locally recorded tool progress used to get dropped. + const projection = baseMessages[projectionIndex] + const merged = lastJournalRow ? overlayProjectionRow(projection, lastJournalRow) : projection + + const sealedRows = tailAssistants.filter( + message => message !== lastJournalRow && assistantHasRecoverableContent(message) + ) + + const messages = [ + ...baseMessages.slice(0, projectionIndex), + ...sealedRows, + merged, + ...baseMessages.slice(projectionIndex + 1) + ] + + return { applied: true, caughtUp: false, messages, streamId: merged.id, turnStartedAt: null } +} + +const persistTimers = new Map>() +const persistLatest = new Map() + +function writeSnapshot(storedSessionId: string, state: JournalableSessionState): void { + const tail = recoverableTail(state.messages, state.streamId) + + if (tail.length === 0) { + return + } + + const journal = loadStore() + + journal.entries[storedSessionId] = { + messages: tail, + streamId: state.streamId, + turnStartedAt: state.turnStartedAt, + updatedAt: Date.now() + } + saveStore(journal) +} + +/** Persist the running turn's visible tail (throttled), or clear the entry the + * moment the turn settles. Call on every session-state commit. */ +export function persistInFlightTurnState(state: JournalableSessionState): void { + const storedSessionId = state.storedSessionId + + if (!storedSessionId) { + return + } + + if (!state.busy && !state.awaitingResponse && !state.streamId) { + clearInFlightTurnJournal(storedSessionId) + + return + } + + persistLatest.set(storedSessionId, state) + + if (persistTimers.has(storedSessionId)) { + return + } + + persistTimers.set( + storedSessionId, + setTimeout(() => { + persistTimers.delete(storedSessionId) + const latest = persistLatest.get(storedSessionId) + + persistLatest.delete(storedSessionId) + + if (latest) { + writeSnapshot(storedSessionId, latest) + } + }, PERSIST_THROTTLE_MS) + ) +} + +export function readInFlightTurnJournal(storedSessionId: null | string): InFlightTurnSnapshot | null { + if (!storedSessionId) { + return null + } + + const journal = loadStore() + const entry = journal.entries[storedSessionId] + + if (!entry) { + return null + } + + if (isExpired(entry)) { + delete journal.entries[storedSessionId] + saveStore(journal) + + return null + } + + return entry +} + +/** Fold a journaled in-flight tail back onto a restored transcript. A no-op + * returns `baseMessages` by reference so callers keep their fast-path ref. */ +export function recoverInFlightTurnJournal( + storedSessionId: null | string, + baseMessages: ChatMessage[], + options: { keepPending?: boolean } = {} +): InFlightRecoveryResult { + const snapshot = readInFlightTurnJournal(storedSessionId) + + if (!snapshot) { + return { + applied: false, + caughtUp: false, + messages: baseMessages, + streamId: null, + turnStartedAt: null + } + } + + const recovered = mergeInFlightMessages(baseMessages, snapshot.messages, options) + + if (recovered.caughtUp) { + clearInFlightTurnJournal(storedSessionId) + } + + return { + ...recovered, + streamId: recovered.applied ? (recovered.streamId ?? snapshot.streamId) : null, + turnStartedAt: recovered.applied ? snapshot.turnStartedAt : null + } +} + +export function clearInFlightTurnJournal(storedSessionId: null | string): void { + if (!storedSessionId) { + return + } + + const timer = persistTimers.get(storedSessionId) + + if (timer) { + clearTimeout(timer) + persistTimers.delete(storedSessionId) + } + + persistLatest.delete(storedSessionId) + + const journal = loadStore() + + if (!(storedSessionId in journal.entries)) { + return + } + + delete journal.entries[storedSessionId] + saveStore(journal) +} diff --git a/apps/desktop/src/types/hermes.ts b/apps/desktop/src/types/hermes.ts index 2b387b8c9a42..669813b0ab25 100644 --- a/apps/desktop/src/types/hermes.ts +++ b/apps/desktop/src/types/hermes.ts @@ -514,8 +514,20 @@ export interface SessionMessagesResponse { } export interface SessionResumeResponse { + /** Present when the backend found a fresh crash-interrupted turn and + * scheduled its automatic continuation; the turn arrives as a normal + * message.start stream right after this resume. */ + auto_continue?: { + attempt: number + interrupted_at: number + } inflight?: null | { assistant?: string + /** Retained failed turn: the error the terminal frame carried (the frame + * itself may have been lost to a disconnect). */ + error?: string + recoverable?: boolean + status?: string streaming?: boolean user?: string } diff --git a/hermes_cli/config.py b/hermes_cli/config.py index cc03c5a57b5c..63d6ae20e18a 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -3646,6 +3646,17 @@ DEFAULT_CONFIG = { # false - always keep GPU acceleration on, even over a remote display. # Bridged to the HERMES_DESKTOP_DISABLE_GPU env var the Electron app reads. "disable_gpu": "auto", + # Auto-continue a turn that was killed mid-run by an app/backend/machine + # crash: resuming that session re-submits the interrupted prompt (shown + # as a "resumed interrupted turn" event) if the interruption is fresh. + # A stale interruption just shows the recovered partial transcript. + "auto_continue": { + "enabled": True, + # How recent the interruption must be to auto-continue (minutes). + "freshness_minutes": 15, + # Crash-loop breaker: max automatic re-runs of one interrupted turn. + "max_attempts": 2, + }, }, diff --git a/tests/tui_gateway/test_auto_continue.py b/tests/tui_gateway/test_auto_continue.py new file mode 100644 index 000000000000..28a70e3e420e --- /dev/null +++ b/tests/tui_gateway/test_auto_continue.py @@ -0,0 +1,423 @@ +"""Crash-interrupted turns auto-continue on the next session.resume. + +A turn's durable marker (tui_gateway/turn_marker.py) is written when the turn +starts running and cleared when it concludes — success, handled error, or +interrupt. Only a process death leaves it behind, so a marker found at resume +time is positive proof the turn never finished. Contract pinned here: + +* the marker module round-trips, prunes stale entries, and tolerates a + corrupt sidecar; +* ``_run_prompt_submit`` writes the marker before the turn and clears it in + the ``finally`` on both the success and exception paths (a handled failure + is a concluded turn — its terminal frame + retained snapshot own recovery); +* ``_maybe_schedule_auto_continue`` re-submits a fresh interrupted prompt as + a continuation note (display_kind ``auto_continue``), refuses stale / + disabled / crash-looping / already-running cases, and bounds attempts via + the marker's attempt counter. +""" + +from __future__ import annotations + +import threading +import time +import types + +import pytest + +from tui_gateway import server +from tui_gateway.turn_marker import ( + clear_turn_marker, + read_turn_marker, + record_turn_start, +) + + +class _InlineThread: + """Run threads synchronously so tests observe final state.""" + + def __init__(self, target=None, daemon=None, args=(), kwargs=None): + self._target = target + self._args = args + self._kwargs = kwargs or {} + + def start(self): + if self._target is not None: + self._target(*self._args, **self._kwargs) + + def is_alive(self): + return False + + def join(self, timeout=None): + return None + + +def _session(agent=None, **extra): + return { + "agent": agent if agent is not None else types.SimpleNamespace(), + "session_key": "session-key", + "history": [], + "history_lock": threading.Lock(), + "history_version": 0, + "running": False, + "attached_images": [], + "image_counter": 0, + "cols": 80, + "slash_worker": None, + "show_reasoning": False, + "tool_progress_mode": "all", + "inflight_turn": None, + **extra, + } + + +@pytest.fixture() +def emits(monkeypatch): + captured: list = [] + monkeypatch.setattr( + server, + "_emit", + lambda event, sid, payload=None: captured.append((event, sid, payload)), + ) + return captured + + +@pytest.fixture() +def marker_home(monkeypatch, tmp_path): + """Point the server's marker storage at a temp HERMES_HOME.""" + monkeypatch.setattr(server, "_hermes_home", tmp_path) + return tmp_path + + +@pytest.fixture() +def turn_env(monkeypatch, tmp_path, marker_home): + """Neutralize the turn pipeline's environment-heavy side paths.""" + monkeypatch.setattr(server.threading, "Thread", _InlineThread) + monkeypatch.setattr(server, "_wire_callbacks", lambda sid: None) + monkeypatch.setattr(server, "_sync_agent_model_with_config", lambda sid, session: None) + monkeypatch.setattr(server, "_session_cwd", lambda session: str(tmp_path)) + monkeypatch.setattr(server, "_register_session_cwd", lambda session: None) + monkeypatch.setattr(server, "_tts_stream_begin", lambda: None) + monkeypatch.setattr(server, "_sync_session_key_after_compress", lambda *a, **k: None) + monkeypatch.setattr(server, "_get_usage", lambda agent: {}) + + +# ── Marker module ────────────────────────────────────────────────────── + + +def test_marker_roundtrip(tmp_path): + record_turn_start(tmp_path, "abc", "fix the bug", attempts=1) + + marker = read_turn_marker(tmp_path, "abc") + assert marker is not None + assert marker["prompt"] == "fix the bug" + assert marker["attempts"] == 1 + assert marker["started_at"] == pytest.approx(time.time(), abs=5) + + clear_turn_marker(tmp_path, "abc") + assert read_turn_marker(tmp_path, "abc") is None + + +def test_marker_clear_is_noop_without_entry(tmp_path): + clear_turn_marker(tmp_path, "missing") + assert read_turn_marker(tmp_path, "missing") is None + + +def test_marker_write_prunes_expired_entries(tmp_path): + import json + + path = tmp_path / "desktop" / "interrupted_turns.json" + path.parent.mkdir(parents=True) + path.write_text( + json.dumps({"old": {"attempts": 0, "prompt": "ancient prompt", "started_at": 1.0}}) + ) + + record_turn_start(tmp_path, "new", "current prompt") + + assert read_turn_marker(tmp_path, "old") is None + assert read_turn_marker(tmp_path, "new") is not None + + +def test_marker_survives_corrupt_sidecar(tmp_path): + path = tmp_path / "desktop" / "interrupted_turns.json" + path.parent.mkdir(parents=True) + path.write_text("{not json") + + assert read_turn_marker(tmp_path, "abc") is None + record_turn_start(tmp_path, "abc", "prompt") + assert read_turn_marker(tmp_path, "abc")["prompt"] == "prompt" + + +# ── Turn lifecycle owns the marker ───────────────────────────────────── + + +def test_concluded_turn_clears_marker(emits, turn_env, marker_home): + seen_mid_turn: list = [] + + def _run(message, **kwargs): + seen_mid_turn.append(read_turn_marker(marker_home, "session-key")) + return {"final_response": "done"} + + agent = types.SimpleNamespace( + session_id="session-key", run_conversation=_run, clear_interrupt=lambda: None + ) + session = _session(agent=agent, running=True) + + server._run_prompt_submit("rid", "sid", session, "do the thing") + + # Written before the turn ran (this is what survives a process death) … + assert seen_mid_turn and seen_mid_turn[0] is not None + assert seen_mid_turn[0]["prompt"] == "do the thing" + assert seen_mid_turn[0]["attempts"] == 0 + # … and cleared once the turn concluded. + assert read_turn_marker(marker_home, "session-key") is None + + +@pytest.mark.parametrize( + "result", [{"final_response": "done"}, {"error": "provider said no"}] +) +def test_marker_is_gone_by_the_terminal_frame(monkeypatch, turn_env, marker_home, result): + """The client treats message.complete as "turn over" and may quit right + then; post-turn work (titles, memory, goal hooks) keeps the thread alive + for a second or more afterwards. If the marker outlived the frame, that + quit looked like a crash and re-ran a finished turn on the next launch.""" + at_frame: list = [] + + def _emit(event, sid, payload=None): + if event == "message.complete": + at_frame.append(read_turn_marker(marker_home, "session-key")) + + monkeypatch.setattr(server, "_emit", _emit) + agent = types.SimpleNamespace( + session_id="session-key", + run_conversation=lambda message, **kwargs: result, + clear_interrupt=lambda: None, + ) + + server._run_prompt_submit("rid", "sid", _session(agent=agent, running=True), "do the thing") + + assert at_frame == [None] + + +def test_handled_failure_still_clears_marker(emits, turn_env, marker_home): + """An exception is a CONCLUDED turn (terminal frame + retained snapshot own + recovery) — only a process death may leave the marker behind.""" + + def _boom(message, **kwargs): + raise RuntimeError("provider exploded") + + agent = types.SimpleNamespace( + session_id="session-key", run_conversation=_boom, clear_interrupt=lambda: None + ) + session = _session(agent=agent, running=True) + + server._run_prompt_submit("rid", "sid", session, "do the thing") + + assert read_turn_marker(marker_home, "session-key") is None + + +def test_continuation_turn_records_attempt_and_original_prompt( + emits, turn_env, marker_home +): + """A continuation's marker must carry the attempt count (crash-loop + breaker) and the ORIGINAL prompt — recording its own recovery note would + nest note inside note on a second crash.""" + seen: list = [] + + def _run(message, **kwargs): + seen.append(read_turn_marker(marker_home, "session-key")) + return {"final_response": "done"} + + agent = types.SimpleNamespace( + session_id="session-key", run_conversation=_run, clear_interrupt=lambda: None + ) + session = _session( + agent=agent, + running=True, + _auto_continue_attempt=2, + _auto_continue_prompt="the original prompt", + ) + + server._run_prompt_submit("rid", "sid", session, server._auto_continue_note("the original prompt")) + + assert [(m["attempts"], m["prompt"]) for m in seen] == [(2, "the original prompt")] + # Consumed, so the NEXT user turn starts from a clean slate. + assert "_auto_continue_attempt" not in session + assert "_auto_continue_prompt" not in session + + +# ── Scheduling decision ──────────────────────────────────────────────── + + +@pytest.fixture() +def schedule_env(monkeypatch, marker_home): + monkeypatch.setattr(server.threading, "Thread", _InlineThread) + monkeypatch.setattr(server, "_start_agent_build", lambda sid, session: None) + monkeypatch.setattr(server, "_wait_agent", lambda session, rid, timeout=30.0: None) + monkeypatch.setattr(server, "_load_cfg", lambda: {}) + submitted: list = [] + monkeypatch.setattr( + server, + "_run_prompt_submit", + lambda rid, sid, session, text, **kw: submitted.append((text, kw)), + ) + return submitted + + +def test_fresh_marker_schedules_continuation(emits, schedule_env, marker_home): + record_turn_start(marker_home, "session-key", "fix the flaky test") + session = _session() + + result = server._maybe_schedule_auto_continue("sid", session, "session-key") + + assert result is not None + assert result["attempt"] == 1 + assert session["running"] is True + assert session["_auto_continue_attempt"] == 1 + (text, kwargs), = schedule_env + assert text.startswith("[System note: Your previous turn was interrupted") + assert "fix the flaky test" in text + assert kwargs["display_kind"] == "auto_continue" + assert ("message.start", "sid", None) in [(e, s, p) for e, s, p in emits] + + +def test_stale_marker_is_cleared_not_continued(schedule_env, marker_home, monkeypatch): + record_turn_start(marker_home, "session-key", "old prompt") + monkeypatch.setattr( + server, "time", types.SimpleNamespace(time=lambda: time.time() + 3600) + ) + + result = server._maybe_schedule_auto_continue("sid", _session(), "session-key") + + assert result is None + assert not schedule_env + assert read_turn_marker(marker_home, "session-key") is None + + +def test_config_widens_freshness_window(emits, schedule_env, marker_home, monkeypatch): + record_turn_start(marker_home, "session-key", "old prompt") + monkeypatch.setattr( + server, + "_load_cfg", + lambda: {"desktop": {"auto_continue": {"freshness_minutes": 120}}}, + ) + monkeypatch.setattr( + server, "time", types.SimpleNamespace(time=lambda: time.time() + 3600) + ) + + result = server._maybe_schedule_auto_continue("sid", _session(), "session-key") + + assert result is not None + assert len(schedule_env) == 1 + + +def test_exhausted_attempts_break_the_loop(schedule_env, marker_home): + record_turn_start(marker_home, "session-key", "crashy prompt", attempts=2) + + result = server._maybe_schedule_auto_continue("sid", _session(), "session-key") + + assert result is None + assert not schedule_env + assert read_turn_marker(marker_home, "session-key") is None + + +def test_disabled_by_config(schedule_env, marker_home, monkeypatch): + record_turn_start(marker_home, "session-key", "prompt") + monkeypatch.setattr( + server, + "_load_cfg", + lambda: {"desktop": {"auto_continue": {"enabled": False}}}, + ) + + result = server._maybe_schedule_auto_continue("sid", _session(), "session-key") + + assert result is None + assert not schedule_env + + +def test_no_marker_means_no_continuation(schedule_env, marker_home): + assert server._maybe_schedule_auto_continue("sid", _session(), "session-key") is None + assert not schedule_env + + +def test_running_session_wins_over_continuation(emits, schedule_env, marker_home): + """A real user prompt that raced the kickoff keeps its turn; the marker is + left for that turn's own conclusion to clear.""" + record_turn_start(marker_home, "session-key", "prompt") + session = _session(running=True) + + result = server._maybe_schedule_auto_continue("sid", session, "session-key") + + # Scheduled (the descriptor is returned), but the kickoff bailed. + assert result is not None + assert not schedule_env + assert session["_auto_continue_scheduled"] is False + assert read_turn_marker(marker_home, "session-key") is not None + # Nothing left behind for the racing user turn to inherit. + assert "_auto_continue_attempt" not in session + assert "_auto_continue_prompt" not in session + + +def test_double_schedule_is_guarded(emits, schedule_env, marker_home): + record_turn_start(marker_home, "session-key", "prompt") + session = _session() + + first = server._maybe_schedule_auto_continue("sid", session, "session-key") + second = server._maybe_schedule_auto_continue("sid", session, "session-key") + + assert first is not None + assert second is None + assert len(schedule_env) == 1 + + +def test_failed_agent_build_leaves_marker_for_retry( + emits, schedule_env, marker_home, monkeypatch +): + record_turn_start(marker_home, "session-key", "prompt") + monkeypatch.setattr( + server, + "_wait_agent", + lambda session, rid, timeout=30.0: {"error": {"message": "boom"}}, + ) + session = _session() + + result = server._maybe_schedule_auto_continue("sid", session, "session-key") + + assert result is not None + assert not schedule_env + assert session["_auto_continue_scheduled"] is False + assert read_turn_marker(marker_home, "session-key") is not None + + +# ── End to end: continuation runs a real turn and clears the marker ──── + + +def test_continuation_runs_through_turn_pipeline(emits, turn_env, marker_home, monkeypatch): + record_turn_start(marker_home, "session-key", "finish the migration") + monkeypatch.setattr(server, "_start_agent_build", lambda sid, session: None) + monkeypatch.setattr(server, "_wait_agent", lambda session, rid, timeout=30.0: None) + monkeypatch.setattr(server, "_load_cfg", lambda: {}) + + prompts: list = [] + + def _run(message, **kwargs): + prompts.append(message) + return {"final_response": "continued and finished"} + + agent = types.SimpleNamespace( + session_id="session-key", run_conversation=_run, clear_interrupt=lambda: None + ) + session = _session(agent=agent) + + result = server._maybe_schedule_auto_continue("sid", session, "session-key") + + assert result == { + "attempt": 1, + "interrupted_at": pytest.approx(time.time(), abs=5), + } + assert len(prompts) == 1 + assert "finish the migration" in prompts[0] + # The concluded continuation cleared both the marker and the turn state. + assert read_turn_marker(marker_home, "session-key") is None + assert session["running"] is False + completes = [p for e, _s, p in emits if e == "message.complete"] + assert completes and completes[0]["status"] == "complete" diff --git a/tui_gateway/turn_marker.py b/tui_gateway/turn_marker.py new file mode 100644 index 000000000000..933f384f37d9 --- /dev/null +++ b/tui_gateway/turn_marker.py @@ -0,0 +1,159 @@ +"""Durable interrupted-turn markers for the desktop/TUI auto-continue path. + +A running turn's progress lives only in process memory (the agent flushes to +SQLite at turn end, not mid-turn), so an app/backend/machine death mid-turn +leaves no durable trace of the interrupted prompt. This sidecar is that +trace: a marker is written when a turn starts running and cleared when the +turn concludes — success, handled error, or interrupt all clear it, so only +a process death leaves one behind. ``session.resume`` reads the marker to +decide whether to auto-continue the interrupted turn (see +``_maybe_schedule_auto_continue`` in ``tui_gateway/server.py``). + +Markers are stored per ``HERMES_HOME`` (callers pass the session's home so +profile sessions keep their state in their own profile directory) and the +file is bounded: writes prune entries older than ``_MAX_AGE_SECS`` and cap +the total count, so an unlucky streak of crashes can't grow it unboundedly. + +Every function is best-effort by design — marker bookkeeping must never +break a turn — so I/O errors degrade to "no marker" instead of raising. +""" + +from __future__ import annotations + +import json +import logging +import os +import tempfile +import threading +import time +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + +_MARKER_DIR = "desktop" +_MARKER_FILE = "interrupted_turns.json" +_MAX_AGE_SECS = 24 * 3600 +_MAX_ENTRIES = 32 +# Enough to re-submit any realistic prompt; guards the sidecar against a +# pathological multi-megabyte paste being journaled on every turn. +_MAX_PROMPT_CHARS = 64_000 + +_lock = threading.Lock() + + +def _marker_path(home: Path | str) -> Path: + return Path(home) / _MARKER_DIR / _MARKER_FILE + + +def _load(path: Path) -> dict[str, dict]: + try: + with open(path, encoding="utf-8") as f: + data = json.load(f) + except FileNotFoundError: + return {} + except Exception: + logger.debug("unreadable turn-marker file %s; starting fresh", path, exc_info=True) + return {} + if not isinstance(data, dict): + return {} + return {k: v for k, v in data.items() if isinstance(v, dict)} + + +def _prune(entries: dict[str, dict], now: float) -> dict[str, dict]: + fresh = { + key: entry + for key, entry in entries.items() + if now - float(entry.get("started_at") or 0) <= _MAX_AGE_SECS + } + if len(fresh) <= _MAX_ENTRIES: + return fresh + newest = sorted( + fresh.items(), + key=lambda item: float(item[1].get("started_at") or 0), + reverse=True, + )[:_MAX_ENTRIES] + return dict(newest) + + +def _store(path: Path, entries: dict[str, dict]) -> None: + if not entries: + path.unlink(missing_ok=True) + return + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".turn-marker-") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(entries, f) + os.replace(tmp, path) + except Exception: + try: + os.unlink(tmp) + except OSError: + pass + raise + + +def record_turn_start( + home: Path | str, session_key: str, prompt: str, *, attempts: int = 0 +) -> None: + """Persist the marker for a turn that is about to run. + + ``attempts`` counts how many auto-continues led to this run: 0 for a + user-initiated turn, N for the Nth automatic re-run — the crash-loop + breaker reads it back on the next resume. + """ + if not session_key or not prompt: + return + now = time.time() + entry = { + "attempts": max(0, int(attempts)), + "prompt": prompt[:_MAX_PROMPT_CHARS], + "started_at": now, + } + try: + with _lock: + path = _marker_path(home) + entries = _prune(_load(path), now) + entries[session_key] = entry + _store(path, entries) + except Exception: + logger.debug("failed to record turn marker for %s", session_key, exc_info=True) + + +def clear_turn_marker(home: Path | str, session_key: str) -> None: + """Remove the marker once its turn concluded (any outcome the client saw).""" + if not session_key: + return + try: + with _lock: + path = _marker_path(home) + entries = _load(path) + if session_key not in entries: + return + del entries[session_key] + _store(path, entries) + except Exception: + logger.debug("failed to clear turn marker for %s", session_key, exc_info=True) + + +def read_turn_marker(home: Path | str, session_key: str) -> dict[str, Any] | None: + """The marker left by a turn that never concluded, or None.""" + if not session_key: + return None + try: + with _lock: + entry = _load(_marker_path(home)).get(session_key) + except Exception: + return None + if not isinstance(entry, dict): + return None + prompt = str(entry.get("prompt") or "") + if not prompt.strip(): + return None + try: + started_at = float(entry.get("started_at") or 0) + attempts = max(0, int(entry.get("attempts") or 0)) + except (TypeError, ValueError): + return None + return {"attempts": attempts, "prompt": prompt, "started_at": started_at}