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<Set<string>>
   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: <provider detail>" 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<string, ClientSessionState>
+
+function Harness() {
+  const activeSessionIdRef = useRef<string | null>(SID)
+  const sessionStateByRuntimeIdRef = useRef(new Map<string, ClientSessionState>())
+  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(<Harness />)
+  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<string, unknown>) =>
+  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: <detail>" 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> = {}): ChatMessage {
+  return { id, role: 'assistant', parts: [{ type: 'text', text }], ...extra }
+}
+
+function assistantWithTool(id: string, text: string, extra: Partial<ChatMessage> = {}): 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> = {}): 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<string, InFlightTurnSnapshot>
+  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<string, InFlightTurnSnapshot>,
+      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<string, ReturnType<typeof setTimeout>>()
+const persistLatest = new Map<string, JournalableSessionState>()
+
+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}