diff --git a/agent/command_token_source.py b/agent/command_token_source.py
index 38ace24324..08c6d04763 100644
--- a/agent/command_token_source.py
+++ b/agent/command_token_source.py
@@ -51,7 +51,7 @@ def _mint(command: str, label: str) -> tuple[str, Optional[float]]:
 
     try:
         completed = subprocess.run(
-            command, shell=True, capture_output=True, text=True, timeout=_MINT_TIMEOUT_SECONDS,
+            command, shell=True, capture_output=True, text=True, errors="replace", timeout=_MINT_TIMEOUT_SECONDS,
             env=served_profile_child_env(inherit_credentials=True),
         )
     except subprocess.TimeoutExpired as exc:
diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py
index 3957b89a34..57af87b558 100644
--- a/agent/conversation_loop.py
+++ b/agent/conversation_loop.py
@@ -40,7 +40,7 @@ from agent.turn_retry_state import TurnRetryState
 from agent.turn_api_call import handle_api_interrupt, nous_rate_limit_guard, perform_api_call
 from agent.turn_api_error import handle_api_error
 from agent.turn_api_request import build_api_request
-from agent.turn_failure_copy import failed_turn_notice, site_copy
+from agent.turn_failure_copy import FAILED_TURN_DISPLAY_KIND, failed_turn_notice, site_copy
 from agent.turn_final_response import finish_text_response
 from agent.turn_finalizer import finalize_turn
 from agent.turn_iteration_prep import (
@@ -1693,7 +1693,9 @@ def _close_durable_failed_turn(agent, result: Any) -> None:
         # hedge over the whole list rather than under-report a possible side effect.
         start = result.get("current_turn_user_idx")
         turn_messages = messages[start:] if isinstance(start, int) and 0 <= start < len(messages) else messages
-        append_message(messages, {"role": "assistant", "content": failed_turn_notice(turn_messages)})
+        append_message(messages, {
+            "role": "assistant", "content": failed_turn_notice(turn_messages), "display_kind": FAILED_TURN_DISPLAY_KIND,
+        })
         agent._flush_messages_to_session_db(messages)
     except Exception:
         logger.debug("failed-turn boundary not written", exc_info=True)
diff --git a/agent/turn_failure_copy.py b/agent/turn_failure_copy.py
index a980a9ff2d..1686f28c63 100644
--- a/agent/turn_failure_copy.py
+++ b/agent/turn_failure_copy.py
@@ -43,6 +43,19 @@ PARTIAL_FAILED_TURN_NOTICE = (
     "This turn did not complete. Some actions may already have run; verify their effects "
     "before resending."
 )
+# ``messages.display_kind`` of that row: display-only (stripped before every provider request),
+# so renderers show a Hermes notice and room pollers never read it as the model's reply.
+FAILED_TURN_DISPLAY_KIND = "failed_turn"
+
+
+def untyped_failed_turn_display_kind(role: Any, content: Any) -> Optional[str]:
+    """``FAILED_TURN_DISPLAY_KIND`` for a boundary row persisted before the closers typed it
+    (exact notice text, so a real reply quoting it stays a reply); read-side only."""
+    if role == "assistant" and isinstance(content, str) and content.strip() in (
+        FAILED_TURN_NOTICE, PARTIAL_FAILED_TURN_NOTICE,
+    ):
+        return FAILED_TURN_DISPLAY_KIND
+    return None
 
 
 def failed_turn_notice(turn_messages: Any) -> str:
diff --git a/apps/desktop/e2e/fixtures.ts b/apps/desktop/e2e/fixtures.ts
index 46495af595..1f3995e486 100644
--- a/apps/desktop/e2e/fixtures.ts
+++ b/apps/desktop/e2e/fixtures.ts
@@ -73,6 +73,16 @@ function isCredentialEnvVar(name: string): boolean {
   return CREDENTIAL_SUFFIXES.some((suffix) => name.endsWith(suffix))
 }
 
+// Runtime state of whatever Hermes launched this run. A spec driven from inside
+// an agent's terminal inherits HERMES_YOLO_MODE, HERMES_INTERACTIVE,
+// HERMES_SESSION_ID…, and the sandboxed backend then skips approvals or binds
+// the caller's session — the approval spec failed locally on the leaked yolo
+// flag while CI (which never has these) stayed green. The fixtures set every
+// HERMES_* the app needs themselves; only the harness's own knobs pass.
+function isInheritedHermesRuntimeVar(name: string): boolean {
+  return name.startsWith('HERMES_') && !name.startsWith('HERMES_DESKTOP_') && !name.startsWith('HERMES_E2E_')
+}
+
 function stripCredentials(env: Record<string, string | undefined>): Record<string, string> {
   const clean: Record<string, string> = {}
 
@@ -81,7 +91,7 @@ function stripCredentials(env: Record<string, string | undefined>): Record<strin
       continue
     }
 
-    if (isCredentialEnvVar(key)) {
+    if (isCredentialEnvVar(key) || isInheritedHermesRuntimeVar(key)) {
       continue
     }
 
diff --git a/apps/desktop/e2e/group-approval-click-submits.spec.ts b/apps/desktop/e2e/group-approval-click-submits.spec.ts
new file mode 100644
index 0000000000..9c78e413c4
--- /dev/null
+++ b/apps/desktop/e2e/group-approval-click-submits.spec.ts
@@ -0,0 +1,128 @@
+import { APPROVAL_COMMAND_TRIGGER, MOCK_REPLY } from '../../../tests-js/scripts/mock-server'
+
+import { type MockBackendFixture, setupMockBackend, waitForAppReady } from './fixtures'
+import { expect, test } from './test'
+
+// #91706: a command-approval card in a Bot Mode group room could highlight
+// the clicked choice but never send it — the footer "Respond" button was
+// off-screen or covered, so the member's hidden session stayed blocked until
+// the approval expired. Approvals are a closed choice set: the click IS the
+// answer. Driven against the real Electron app, a real gateway with
+// `approvals: mode: "manual"`, and mock inference that runs a gated
+// `rm -rf` through the real terminal tool.
+
+const ROOM = 'Programmer, Reviewer'
+let fixture: MockBackendFixture | null = null
+
+async function openBots(page: MockBackendFixture['page']): Promise<void> {
+  const tab = page.getByRole('button', { name: 'Bots', exact: true }).or(page.getByRole('tab', { name: 'Bots', exact: true })).first()
+  await tab.click()
+  await expect(page.getByRole('button', { name: 'New bot or group chat' })).toBeVisible()
+}
+
+async function createAgent(page: MockBackendFixture['page'], name: string, title: string): Promise<void> {
+  await page.getByRole('button', { name: 'New bot or group chat' }).click()
+  await page.getByRole('menuitem', { name: 'New Bot' }).click()
+
+  const dialog = page.getByRole('dialog', { name: 'New Bot' })
+  await dialog.getByPlaceholder('inbox-triage').fill(name)
+  await dialog.getByPlaceholder('Inbox Triage').fill(title)
+  await dialog.getByRole('button', { name: 'Create Bot' }).click()
+  await expect(dialog).toBeHidden({ timeout: 30_000 })
+  await expect(page.getByRole('button', { name: new RegExp(`^${title}\\b`) }).first()).toBeVisible({ timeout: 30_000 })
+}
+
+async function createRoom(page: MockBackendFixture['page']) {
+  await openBots(page)
+  await createAgent(page, 'programmer', 'Programmer')
+  await createAgent(page, 'reviewer', 'Reviewer')
+
+  await page.getByRole('button', { name: 'New bot or group chat' }).click()
+  await page.getByRole('menuitem', { name: 'New Group Chat' }).click()
+
+  const dialog = page.getByRole('dialog', { name: 'New Group Chat' })
+
+  for (const title of ['Programmer', 'Reviewer']) {
+    await dialog.getByText(title, { exact: true }).locator('xpath=ancestor::label').getByRole('checkbox').click()
+  }
+
+  await dialog.getByRole('textbox', { name: 'Group name' }).fill(ROOM)
+  await dialog.getByRole('button', { name: 'Create Group (2)' }).click()
+
+  const groupTab = page.getByRole('tab', { name: new RegExp(`${ROOM} Close`) })
+  const groupComposer = page.getByRole('textbox', { name: `Message ${ROOM}` }).filter({ visible: true })
+  await expect(groupTab).toBeVisible({ timeout: 20_000 })
+  await expect(groupTab).toHaveAttribute('aria-selected', 'true')
+  await expect(groupComposer).toBeVisible()
+
+  return groupComposer
+}
+
+test.beforeEach(async () => {
+  fixture = await setupMockBackend({ extraConfig: 'approvals:\n  mode: "manual"\n' })
+  await waitForAppReady(fixture, 120_000)
+})
+
+test.afterEach(async () => {
+  await fixture?.cleanup()
+  fixture = null
+})
+
+test('clicking an approval choice in a group room submits it (#91706)', async () => {
+  test.setTimeout(240_000)
+  const page = fixture!.page
+  // Count approval.respond frames on the real gateway socket.
+  await page.evaluate(() => {
+    const send = WebSocket.prototype.send
+
+    ;(window as any).__approvalResponds = [] as string[]
+
+    WebSocket.prototype.send = function (data) {
+      const frame = JSON.parse(String(data))
+
+      if (frame.method === 'approval.respond') {
+        ;(window as any).__approvalResponds.push(JSON.stringify(frame.params))
+      }
+
+      return send.call(this, data)
+    }
+  })
+  const groupComposer = await createRoom(page)
+
+  await groupComposer.fill(`@programmer ${APPROVAL_COMMAND_TRIGGER}`)
+  await groupComposer.press('Enter')
+
+  // The member's gated terminal command surfaces as an approval card in the
+  // ROOM. Assert inside the room's own card: the same approval also reaches
+  // the Desktop's session-level approval surface, and a build that switches
+  // tabs on it would otherwise fail here on visibility instead of on the
+  // contract below (the click must be the submit).
+  const groupTab = page.getByRole('tab', { name: new RegExp(`${ROOM} Close`) })
+  const card = page.getByText(/wants to run a command/).locator('xpath=..')
+  const once = card.getByRole('button', { name: 'once', exact: true })
+
+  await expect(async () => {
+    if ((await groupTab.getAttribute('aria-selected')) !== 'true') {
+      await groupTab.click()
+    }
+
+    await expect(card).toBeVisible({ timeout: 5_000 })
+  }).toPass({ timeout: 90_000 })
+  await expect(once).toBeVisible()
+  console.log('APPROVAL CARD: visible; responds so far =', await page.evaluate(() => (window as any).__approvalResponds.length))
+  await page.screenshot({ path: test.info().outputPath('approval-card.png') })
+
+  await once.click()
+
+  // One approval.respond leaves the Desktop for the click itself — no second
+  // "Respond" click required (there is none to click for approvals).
+  await expect.poll(() => page.evaluate(() => (window as any).__approvalResponds.length), { timeout: 15_000 }).toBe(1)
+  await expect(card.getByRole('button', { name: 'Respond', exact: true })).toHaveCount(0)
+  expect(await page.evaluate(() => (window as any).__approvalResponds[0])).toContain('once')
+
+  // The blocked member resumes: the command runs and its reply lands in the room.
+  await expect(page.getByText(MOCK_REPLY, { exact: true }).first()).toBeVisible({ timeout: 90_000 })
+  await expect(once).toHaveCount(0)
+  console.log('APPROVAL: submitted on click; responds =', await page.evaluate(() => (window as any).__approvalResponds))
+  await page.screenshot({ path: test.info().outputPath('approval-after.png') })
+})
diff --git a/apps/desktop/e2e/group-member-backend-failure.spec.ts b/apps/desktop/e2e/group-member-backend-failure.spec.ts
new file mode 100644
index 0000000000..aa670da52d
--- /dev/null
+++ b/apps/desktop/e2e/group-member-backend-failure.spec.ts
@@ -0,0 +1,148 @@
+import {
+  PROVIDER_FAILURE_TRIGGER,
+  TOOL_THEN_FAILURE_TEXT,
+  TOOL_THEN_FAILURE_TRIGGER
+} from '../../../tests-js/scripts/mock-server'
+
+import { type MockBackendFixture, setupMockBackend, waitForAppReady } from './fixtures'
+import { expect, test } from './test'
+
+// #92760 "thinks then never speaks": when a member's backend fails the turn
+// (bad credentials, provider refusal), the gateway keeps the failed turn under
+// `session.resume.inflight` as `{ status: 'error' }` so a reconnecting client
+// can rebuild the error bubble. The room engine read that truthy object as
+// "still working" and kept extending the member's deadline toward the
+// 20-minute cap — the user saw a bot thinking forever with no error anywhere.
+// Real Electron + real gateway; the mock provider answers the member with a
+// non-retryable 401.
+
+const ROOM = 'Programmer, Reviewer'
+let fixture: MockBackendFixture | null = null
+
+async function openBots(page: MockBackendFixture['page']): Promise<void> {
+  const tab = page.getByRole('button', { name: 'Bots', exact: true }).or(page.getByRole('tab', { name: 'Bots', exact: true })).first()
+  await tab.click()
+  await expect(page.getByRole('button', { name: 'New bot or group chat' })).toBeVisible()
+}
+
+async function createAgent(page: MockBackendFixture['page'], name: string, title: string): Promise<void> {
+  await page.getByRole('button', { name: 'New bot or group chat' }).click()
+  await page.getByRole('menuitem', { name: 'New Bot' }).click()
+
+  const dialog = page.getByRole('dialog', { name: 'New Bot' })
+  await dialog.getByPlaceholder('inbox-triage').fill(name)
+  await dialog.getByPlaceholder('Inbox Triage').fill(title)
+  await dialog.getByRole('button', { name: 'Create Bot' }).click()
+  await expect(dialog).toBeHidden({ timeout: 30_000 })
+  await expect(page.getByRole('button', { name: new RegExp(`^${title}\\b`) }).first()).toBeVisible({ timeout: 30_000 })
+}
+
+async function createRoom(page: MockBackendFixture['page']) {
+  await openBots(page)
+  await createAgent(page, 'programmer', 'Programmer')
+  await createAgent(page, 'reviewer', 'Reviewer')
+
+  await page.getByRole('button', { name: 'New bot or group chat' }).click()
+  await page.getByRole('menuitem', { name: 'New Group Chat' }).click()
+
+  const dialog = page.getByRole('dialog', { name: 'New Group Chat' })
+
+  for (const title of ['Programmer', 'Reviewer']) {
+    await dialog.getByText(title, { exact: true }).locator('xpath=ancestor::label').getByRole('checkbox').click()
+  }
+
+  await dialog.getByRole('textbox', { name: 'Group name' }).fill(ROOM)
+  await dialog.getByRole('button', { name: 'Create Group (2)' }).click()
+
+  const groupTab = page.getByRole('tab', { name: new RegExp(`${ROOM} Close`) })
+  const groupComposer = page.getByRole('textbox', { name: `Message ${ROOM}` }).filter({ visible: true })
+  await expect(groupTab).toBeVisible({ timeout: 20_000 })
+  await expect(groupTab).toHaveAttribute('aria-selected', 'true')
+  await expect(groupComposer).toBeVisible()
+
+  return groupComposer
+}
+
+test.beforeEach(async () => {
+  fixture = await setupMockBackend()
+  await waitForAppReady(fixture, 120_000)
+})
+
+test.afterEach(async () => {
+  await fixture?.cleanup()
+  fixture = null
+})
+
+test('a member whose backend fails the turn is reported at once, not read as busy (#92760)', async () => {
+  test.setTimeout(240_000)
+  const page = fixture!.page
+  const groupComposer = await createRoom(page)
+
+  await groupComposer.fill(`@programmer ${PROVIDER_FAILURE_TRIGGER}`)
+  await groupComposer.press('Enter')
+  await expect.poll(() => fixture!.mock.receivedPrompts.some(p => p.includes(PROVIDER_FAILURE_TRIGGER)), { timeout: 60_000 }).toBe(true)
+
+  // The gateway has failed the turn; the room must say so within the base
+  // turn timeout instead of extending the deadline on the retained snapshot.
+  const activity = page.getByRole('button', { name: /^Activity/ })
+  const groupTab = page.getByRole('tab', { name: new RegExp(`${ROOM} Close`) })
+
+  await expect(async () => {
+    // A just-created bot's background intro turn can front that bot's chat
+    // tab and yank the center away from the room; re-select the room first.
+    if ((await groupTab.getAttribute('aria-selected')) !== 'true') {
+      await groupTab.click()
+    }
+
+    await expect(activity).toContainText('Programmer hit an error', { timeout: 5_000 })
+  }).toPass({ timeout: 120_000 })
+  // #117366: the row names the cause, not just the fact — the raw error's
+  // first line rides along so a stopped backend and a provider refusal differ.
+  await expect(activity).toContainText(/Programmer hit an error — \S+/)
+  await expect(page.getByRole('button', { name: 'Stop', exact: true })).toHaveCount(0, { timeout: 30_000 })
+
+  const room = await page.evaluate(name => {
+    const rooms = JSON.parse(localStorage.getItem('hermes.plugin.hermes-bots.group-chats') || '{}')
+
+    return { stranded: Object.keys(rooms[name]?.stranded || {}), running: Boolean(rooms[name]?.running) }
+  }, ROOM)
+
+  expect(room.stranded).toEqual([])
+  console.log('RETAINED FAILURE: activity =', await activity.textContent(), 'room =', JSON.stringify(room))
+  await page.screenshot({ path: test.info().outputPath('retained-failure-after.png') })
+})
+
+// The member spoke and called a tool before the provider failed. The session
+// ends on the tool row (no failed-turn boundary behind it) and only the
+// retained 401 says the turn died: the text written before the tool call must
+// not be posted into the room as the member's reply.
+test('a member that fails after pre-tool text is reported, and that text is not its reply', async () => {
+  test.setTimeout(240_000)
+  const page = fixture!.page
+  const groupComposer = await createRoom(page)
+
+  await groupComposer.fill(`@programmer ${TOOL_THEN_FAILURE_TRIGGER}`)
+  await groupComposer.press('Enter')
+
+  const activity = page.getByRole('button', { name: /^Activity/ })
+  const groupTab = page.getByRole('tab', { name: new RegExp(`${ROOM} Close`) })
+
+  await expect(async () => {
+    if ((await groupTab.getAttribute('aria-selected')) !== 'true') {
+      await groupTab.click()
+    }
+
+    await expect(activity).toContainText('Programmer hit an error', { timeout: 5_000 })
+  }).toPass({ timeout: 150_000 })
+
+  const room = await page.evaluate(name => {
+    const rooms = JSON.parse(localStorage.getItem('hermes.plugin.hermes-bots.group-chats') || '{}')
+
+    return { log: (rooms[name]?.log || []).map((entry: { text?: string }) => entry.text || ''), stranded: Object.keys(rooms[name]?.stranded || {}) }
+  }, ROOM)
+
+  expect(room.log.some((text: string) => text.includes(TOOL_THEN_FAILURE_TEXT))).toBe(false)
+  expect(room.stranded).toEqual([])
+  console.log('TOOL THEN FAILURE: activity =', await activity.textContent(), 'log =', JSON.stringify(room.log))
+  await page.screenshot({ path: test.info().outputPath('tool-then-failure-after.png') })
+})
diff --git a/apps/desktop/src/lib/chat-messages.test.ts b/apps/desktop/src/lib/chat-messages.test.ts
index 80951435ad..d9c18c0923 100644
--- a/apps/desktop/src/lib/chat-messages.test.ts
+++ b/apps/desktop/src/lib/chat-messages.test.ts
@@ -450,6 +450,18 @@ describe('toChatMessages', () => {
     ])
   })
 
+  // Hermes closes a failed turn with an assistant-role row (agent/turn_failure_copy.py);
+  // painted as the model's reply it read as the assistant refusing the request.
+  it('renders the failed-turn boundary as a Hermes notice, not a model reply', () => {
+    const messages = toChatMessages([
+      { role: 'user', content: 'do the thing', timestamp: 1 },
+      { role: 'assistant', content: 'Your request was not processed.', display_kind: 'failed_turn', timestamp: 2 }
+    ])
+
+    expect(messages.map(message => message.role)).toEqual(['user', 'system'])
+    expect(chatMessageText(messages[1])).toBe('Your request was not processed.')
+  })
+
   // A backend older than this app serves display_metadata as unparsed JSON
   // text. Indexing into that string used to throw and fail the whole resume.
   it.each([
diff --git a/apps/desktop/src/lib/chat-messages/hydration.ts b/apps/desktop/src/lib/chat-messages/hydration.ts
index aa21ce40e0..ecf79177ed 100644
--- a/apps/desktop/src/lib/chat-messages/hydration.ts
+++ b/apps/desktop/src/lib/chat-messages/hydration.ts
@@ -346,7 +346,9 @@ export function toChatMessages(messages: SessionMessage[]): ChatMessage[] {
       message.display_kind === 'async_delegation_complete' ||
       message.display_kind === 'process_complete' ||
       message.display_kind === 'auto_continue' ||
-      message.display_kind === 'personality_switch'
+      message.display_kind === 'personality_switch' ||
+      // Hermes closing a failed turn, not the model speaking.
+      message.display_kind === 'failed_turn'
         ? 'system'
         : message.role
 
diff --git a/apps/desktop/src/lib/speech-text.test.ts b/apps/desktop/src/lib/speech-text.test.ts
index b66cb46283..b22d7c28d4 100644
--- a/apps/desktop/src/lib/speech-text.test.ts
+++ b/apps/desktop/src/lib/speech-text.test.ts
@@ -7,9 +7,7 @@ describe('sanitizeTextForSpeech', () => {
     // The "code block omitted" summary used to be read aloud as English text
     // (#86602). Code that can't be spoken should be silence, not a sentence.
     // The "here is code:" colon also closes: the voice never waits on it.
-    expect(sanitizeTextForSpeech('Here is code:\n```ts\nconst x = 1\n```\nDone.')).toBe(
-      'Here is code. Done.'
-    )
+    expect(sanitizeTextForSpeech('Here is code:\n```ts\nconst x = 1\n```\nDone.')).toBe('Here is code. Done.')
   })
 
   it('still keeps normal prose and inline code readable', () => {
@@ -94,9 +92,7 @@ Second sentence.`
 
   it('does not speak a placeholder word for URLs', () => {
     // Used to say the English word "link" (#86602); URLs are silence now.
-    expect(sanitizeTextForSpeech('See https://example.com/a-huge-page for details')).toBe(
-      'See for details'
-    )
+    expect(sanitizeTextForSpeech('See https://example.com/a-huge-page for details')).toBe('See for details')
   })
 
   it('keeps ~~strike~~ readable instead of speaking tildes', () => {
@@ -112,9 +108,9 @@ Second sentence.`
   it('closes a colon that a code block used to follow', () => {
     // The real repro: "one line added to the regex list:" then a code fence.
     // The voice hit the colon, found a wall of punctuation, and stuttered.
-    expect(
-      sanitizeTextForSpeech('One line added to the regex list:\n```ts\nconst x = 1\n```\nBye.')
-    ).toBe('One line added to the regex list. Bye.')
+    expect(sanitizeTextForSpeech('One line added to the regex list:\n```ts\nconst x = 1\n```\nBye.')).toBe(
+      'One line added to the regex list. Bye.'
+    )
   })
 
   it('closes a colon that ends the speakable text', () => {
diff --git a/apps/desktop/src/lib/voice-barge-in.test.ts b/apps/desktop/src/lib/voice-barge-in.test.ts
index 2853cbf53f..e597c7dd22 100644
--- a/apps/desktop/src/lib/voice-barge-in.test.ts
+++ b/apps/desktop/src/lib/voice-barge-in.test.ts
@@ -1,6 +1,10 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
-import { bargeInTriggerLevels, DEFAULT_BARGE_IN_THRESHOLD_MULTIPLIER, monitorSpeechDuringPlayback } from './voice-barge-in'
+import {
+  bargeInTriggerLevels,
+  DEFAULT_BARGE_IN_THRESHOLD_MULTIPLIER,
+  monitorSpeechDuringPlayback
+} from './voice-barge-in'
 
 // Drive the real monitor with a fake mic: `micLevel` is the byte-domain RMS
 // level the analyser reports, `playing` is the TTS-flowing flag, and each
diff --git a/apps/desktop/src/plugins/hermes-bots/group-external-writes.ts b/apps/desktop/src/plugins/hermes-bots/group-external-writes.ts
index 37881d0817..ad099d6d20 100644
--- a/apps/desktop/src/plugins/hermes-bots/group-external-writes.ts
+++ b/apps/desktop/src/plugins/hermes-bots/group-external-writes.ts
@@ -74,6 +74,15 @@ export const SYNTHETIC_USER_ROW_PREFIXES = [
   'Cronjob Response:'
 ]
 
+/** The Hermes-authored assistant row that closes a turn which failed before
+ *  the model answered (a provider 401, retry exhaustion, a refusal), typed
+ *  `display_kind: failed_turn` by `agent/turn_failure_copy.py`. A transcript
+ *  boundary, never the member's reply: read as one, the room posts it as the
+ *  bot speaking and loses the failure (#92760). */
+export function failedTurnBoundaryRow(row: GroupTranscriptRow): boolean {
+  return row.role === 'assistant' && row.display_kind === 'failed_turn'
+}
+
 /** A user row that carries no user words: typed scaffolding (auto-continue
  *  notes, steer markers, model-switch notices — anything but a skill
  *  invocation, which is the user's own `/command`) or an untyped row opening
@@ -100,7 +109,7 @@ export function externalGroupTranscriptRows(rows: GroupTranscriptRow[]): string[
   for (const row of rows) {
     const text = groupTranscriptRowText(row)
 
-    if (!text || (row.role !== 'user' && row.role !== 'assistant')) {
+    if (!text || (row.role !== 'user' && row.role !== 'assistant') || failedTurnBoundaryRow(row)) {
       continue
     }
 
diff --git a/apps/desktop/src/plugins/hermes-bots/group-test-utils.ts b/apps/desktop/src/plugins/hermes-bots/group-test-utils.ts
index af3600f40c..1292cf4e31 100644
--- a/apps/desktop/src/plugins/hermes-bots/group-test-utils.ts
+++ b/apps/desktop/src/plugins/hermes-bots/group-test-utils.ts
@@ -26,6 +26,7 @@ import { vi } from 'vitest'
 /** One message in a scripted session transcript, in the gateway's own shape. */
 export interface ScriptedMessage {
   content: string
+  display_kind?: string
   role: string
 }
 
diff --git a/apps/desktop/src/plugins/hermes-bots/group-turns.test.ts b/apps/desktop/src/plugins/hermes-bots/group-turns.test.ts
index bd979d29a4..1335369d9c 100644
--- a/apps/desktop/src/plugins/hermes-bots/group-turns.test.ts
+++ b/apps/desktop/src/plugins/hermes-bots/group-turns.test.ts
@@ -20,6 +20,10 @@ vi.mock('@hermes/plugin-sdk', async () => {
   return pluginSdkMock(host)
 })
 
+// agent/turn_failure_copy.py::PARTIAL_FAILED_TURN_NOTICE
+const PARTIAL_NOTICE =
+  'This turn did not complete. Some actions may already have run; verify their effects before resending.'
+
 interface Room {
   chat: typeof groupChat
   gateway: ScriptedGateway
@@ -333,6 +337,120 @@ describe('session-gone classification', () => {
     }
   })
 
+  // The core loop closes a failed turn with a Hermes-authored assistant row
+  // typed `display_kind: failed_turn` (agent/turn_failure_copy.py). That row is
+  // a transcript boundary, not the member speaking: read as the reply, the
+  // room posted it as the bot's answer, re-drove the member and hid the error.
+  it('reports a failed turn whose transcript Hermes closed with the failed-turn notice', async () => {
+    let now = 1_000_000
+    const clock = vi.spyOn(Date, 'now').mockImplementation(() => (now += 60_000))
+
+    const room = await loadRoom({
+      turn: () => [{ content: 'Your request was not processed.', display_kind: 'failed_turn', role: 'assistant' }]
+    })
+
+    const request = host.request as (method: string, params?: Record<string, unknown>) => Promise<unknown>
+    let submitted = false
+
+    // The real gateway keeps the tombstone AND the closed transcript.
+    host.request = async (method: string, params: Record<string, unknown> = {}) => {
+      const result = (await request(method, params)) as Record<string, unknown>
+
+      submitted = submitted || method === 'prompt.submit'
+
+      return method === 'session.resume' && submitted
+        ? { ...result, inflight: { error: 'HTTP 401: invalid_api_key', status: 'error', streaming: false } }
+        : result
+    }
+
+    try {
+      await expect(room.turns.runGroupChatMemberTurn('Room', LOCAL_MEMBER, 'hi', 't1', [])).rejects.toThrow(
+        'HTTP 401: invalid_api_key'
+      )
+    } finally {
+      clock.mockRestore()
+    }
+  })
+
+  // The member spoke, called a tool, then the provider failed. The core closer
+  // writes no boundary behind a tool row, so the retained error is the only
+  // evidence; with a boundary but the retained error gone (backend restarted)
+  // the turn still failed. Either way the pre-tool text is not the reply.
+  it.each([
+    ['the retained provider error', [], 'HTTP 401: invalid_api_key'],
+    [
+      'the failed-turn notice once the retained error is gone',
+      [{ content: PARTIAL_NOTICE, display_kind: 'failed_turn', role: 'assistant' }],
+      null
+    ]
+  ])('reports a turn that failed after pre-tool text with %s', async (_label, boundary, retained) => {
+    let now = 1_000_000
+    const clock = vi.spyOn(Date, 'now').mockImplementation(() => (now += 60_000))
+
+    const room = await loadRoom({
+      turn: () => [
+        { content: 'Let me check the repo first.', role: 'assistant' },
+        { content: 'ok', role: 'tool' },
+        ...boundary
+      ]
+    })
+
+    const request = host.request as (method: string, params?: Record<string, unknown>) => Promise<unknown>
+    let submitted = false
+
+    host.request = async (method: string, params: Record<string, unknown> = {}) => {
+      const result = (await request(method, params)) as Record<string, unknown>
+
+      submitted = submitted || method === 'prompt.submit'
+
+      return method === 'session.resume' && submitted && retained
+        ? { ...result, inflight: { error: retained, status: 'error', streaming: false } }
+        : result
+    }
+
+    try {
+      await expect(room.turns.runGroupChatMemberTurn('Room', LOCAL_MEMBER, 'hi', 't1', [])).rejects.toThrow(
+        retained ?? PARTIAL_NOTICE
+      )
+    } finally {
+      clock.mockRestore()
+    }
+  })
+
+  // The previous turn failed with the same error: only `turn_started_at` tells
+  // this turn's retained failure from the leftover one.
+  it('reports a failure identical to the previous turn’s instead of posting pre-tool text', async () => {
+    let now = 1_000_000
+    const clock = vi.spyOn(Date, 'now').mockImplementation(() => (now += 60_000))
+
+    const room = await loadRoom({
+      turn: () => [
+        { content: 'Let me check the repo first.', role: 'assistant' },
+        { content: 'ok', role: 'tool' }
+      ]
+    })
+
+    const request = host.request as (method: string, params?: Record<string, unknown>) => Promise<unknown>
+    const inflight = { error: 'HTTP 401: invalid_api_key', status: 'error', streaming: false }
+    let submitted = false
+
+    host.request = async (method: string, params: Record<string, unknown> = {}) => {
+      const result = (await request(method, params)) as Record<string, unknown>
+
+      submitted = submitted || method === 'prompt.submit'
+
+      return method === 'session.resume' ? { ...result, inflight, turn_started_at: submitted ? 200 : 100 } : result
+    }
+
+    try {
+      await expect(room.turns.runGroupChatMemberTurn('Room', LOCAL_MEMBER, 'hi', 't1', [])).rejects.toThrow(
+        inflight.error
+      )
+    } finally {
+      clock.mockRestore()
+    }
+  })
+
   // A turn that dies BEFORE its prompt is committed (agent-init failure,
   // no-agent refusal) leaves a retained `{ status: 'error' }` and a transcript
   // that never grew — the failure must still surface instead of the poll
@@ -1261,6 +1379,98 @@ describe('stranded harvest', () => {
     ])
   })
 
+  // A late turn that spoke, called a tool and then hit a provider failure: the
+  // pre-tool text is not a late reply, whether the retained error or only the
+  // failed-turn row (retained error gone) says the turn failed.
+  it.each([
+    ['the retained provider error', [], 'HTTP 401: invalid_api_key'],
+    [
+      'the failed-turn notice alone',
+      [{ content: PARTIAL_NOTICE, display_kind: 'failed_turn', role: 'assistant' }],
+      null
+    ]
+  ])('reports a late turn that failed after pre-tool text with %s', async (_label, boundary, retained) => {
+    const room = await loadRoom()
+    const activity = await import('./group-activity')
+
+    room.chat.updateGroupChat('Broke', current => {
+      current.sessions = { research: 'sid-research' }
+      current.stranded = { research: 0 }
+
+      return current
+    })
+    room.gateway.sessions.set('sid-research', {
+      messages: [
+        { content: roomPrompt('Broke'), role: 'user' },
+        { content: 'Let me check the repo first.', role: 'assistant' },
+        { content: 'ok', role: 'tool' },
+        ...boundary
+      ],
+      profile: 'research',
+      runtime: 'rt-research',
+      stored: 'sid-research',
+      title: 'Group: Broke'
+    })
+
+    if (retained) {
+      const request = host.request as (method: string, params?: Record<string, unknown>) => Promise<unknown>
+
+      host.request = async (method: string, params: Record<string, unknown> = {}) => {
+        const result = (await request(method, params)) as Record<string, unknown>
+
+        return method === 'session.resume'
+          ? { ...result, inflight: { error: retained, status: 'error', streaming: false } }
+          : result
+      }
+    }
+
+    await room.turns.harvestStrandedGroupReply('Broke', { name: 'research', title: '' })
+
+    expect(log(room, 'Broke')).toHaveLength(0)
+    expect(room.chat.$groupChats.get().Broke.stranded?.research).toBeUndefined()
+    expect(activity.$groupActivity.get().Broke?.events.map(event => event.kind)).toEqual(['failed'])
+  })
+
+  // A later gateway turn in the same session failed: its retained error is not
+  // the stranded turn's, so the stranded turn's real late reply still posts.
+  it('posts the late reply when a later turn in the session failed', async () => {
+    const room = await loadRoom()
+    const activity = await import('./group-activity')
+
+    room.chat.updateGroupChat('Late', current => {
+      current.sessions = { research: 'sid-research' }
+      current.stranded = { research: 0 }
+
+      return current
+    })
+    room.gateway.sessions.set('sid-research', {
+      messages: [
+        { content: roomPrompt('Late'), role: 'user' },
+        { content: 'Here is the late answer.', role: 'assistant' },
+        { content: 'and the deploy?', role: 'user' }
+      ],
+      profile: 'research',
+      runtime: 'rt-research',
+      stored: 'sid-research',
+      title: 'Group: Late'
+    })
+
+    const request = host.request as (method: string, params?: Record<string, unknown>) => Promise<unknown>
+
+    host.request = async (method: string, params: Record<string, unknown> = {}) => {
+      const result = (await request(method, params)) as Record<string, unknown>
+
+      return method === 'session.resume'
+        ? { ...result, inflight: { error: 'HTTP 401: invalid_api_key', status: 'error', streaming: false } }
+        : result
+    }
+
+    await room.turns.harvestStrandedGroupReply('Late', { name: 'research', title: '' })
+
+    expect(log(room, 'Late').map(entry => entry.text)).toContain('Here is the late answer.')
+    expect(activity.$groupActivity.get().Late?.events.map(event => event.kind)).not.toContain('failed')
+  })
+
   it('never re-submits into a member the harvest just confirmed is still running', async () => {
     // research is confirmed busy on exactly its first two session.resume calls
     // — the number of harvest-only touches the FIXED code makes across two
diff --git a/apps/desktop/src/plugins/hermes-bots/group-turns.ts b/apps/desktop/src/plugins/hermes-bots/group-turns.ts
index 362b40880a..00af6bb52c 100644
--- a/apps/desktop/src/plugins/hermes-bots/group-turns.ts
+++ b/apps/desktop/src/plugins/hermes-bots/group-turns.ts
@@ -12,7 +12,12 @@ import { noteBotAttention } from './data'
 import { groupFailureReason, recordGroupActivity } from './group-activity'
 import { $groupChats, $groupClarify, appendGroupChatEntry, updateGroupChat } from './group-chat'
 import type { GroupChatRoom } from './group-chat'
-import { groupTranscriptRowText, mirrorExternalGroupWrites, syntheticGroupUserRow } from './group-external-writes'
+import {
+  failedTurnBoundaryRow,
+  groupTranscriptRowText,
+  mirrorExternalGroupWrites,
+  syntheticGroupUserRow
+} from './group-external-writes'
 import {
   followGroupChat,
   groupMemberAuthor,
@@ -41,10 +46,16 @@ export function isGroupPassText(text: unknown) {
  *  `content` is a plain string on most providers and a part array on the rest. */
 interface GroupTurnTranscriptMessage {
   content?: string | Array<string | { text?: string }>
+  display_kind?: string
   role?: string
   text?: string
 }
 
+/** What a finished turn left behind: the member's reply, or the notice of the
+ *  `failed_turn` row Hermes closed it with (the member never answered), or
+ *  null when no assistant row landed. */
+type GroupTurnPick = { failedNotice: string } | null | string
+
 /** #94376: pick the reply a finished turn should surface among the messages
  *  appended since `before`. Scans newest-first and prefers the last
  *  substantive (non-pass) assistant answer over a trailing pass — a Codex
@@ -52,8 +63,10 @@ interface GroupTurnTranscriptMessage {
  *  synthetic "(pass)" to the nudge itself, which must not hide the answer.
  *  When only pass text exists in range, returns the newest (last
  *  chronological) one rather than the oldest. Returns null only when no
- *  assistant message appears in that range. */
-function pickGroupTurnReply(messages: GroupTurnTranscriptMessage[], before: number): null | string {
+ *  assistant message appears in that range. A failed-turn boundary ends the
+ *  scan: text the member wrote before the tool call that preceded the
+ *  provider failure is not its reply. */
+function pickGroupTurnReply(messages: GroupTurnTranscriptMessage[], before: number): GroupTurnPick {
   let passText: null | string = null
 
   for (let i = messages.length - 1; i >= before; i--) {
@@ -72,6 +85,10 @@ function pickGroupTurnReply(messages: GroupTurnTranscriptMessage[], before: numb
 
     const replyText = String(text).trim()
 
+    if (failedTurnBoundaryRow(msg)) {
+      return passText ?? { failedNotice: replyText }
+    }
+
     if (isGroupPassText(replyText)) {
       if (passText === null) {
         passText = replyText
@@ -91,14 +108,17 @@ function pickGroupTurnReply(messages: GroupTurnTranscriptMessage[], before: numb
  *  stopping where an outside writer takes the session over. Newest-first
  *  (`pickGroupTurnReply`) would post a CLI answer written after the late reply
  *  as the turn reply — and the external-write mirror posts it again. Only
- *  passes in range → the last pass; no anchor row → scan from `before`. */
-function pickStrandedGroupTurnReply(messages: GroupTurnTranscriptMessage[], before: number): null | string {
+ *  passes in range → the last pass; no anchor row → scan from `before`. A
+ *  failed-turn boundary anywhere in the turn makes it a failure, even after
+ *  text the member wrote before its last tool call. */
+function pickStrandedGroupTurnReply(messages: GroupTurnTranscriptMessage[], before: number): GroupTurnPick {
   const anchor = messages.findIndex(
     (msg, i) =>
       i >= before && msg?.role === 'user' && groupTranscriptRowText(msg).startsWith(GROUP_PROMPT_HEADER_PREFIX)
   )
 
   let passText: null | string = null
+  let reply: null | string = null
 
   for (let i = anchor === -1 ? before : anchor; i < messages.length; i++) {
     const msg = messages[i]
@@ -112,6 +132,10 @@ function pickStrandedGroupTurnReply(messages: GroupTurnTranscriptMessage[], befo
       break
     }
 
+    if (failedTurnBoundaryRow(msg)) {
+      return { failedNotice: text }
+    }
+
     if (msg?.role !== 'assistant' || !text) {
       continue
     }
@@ -122,10 +146,10 @@ function pickStrandedGroupTurnReply(messages: GroupTurnTranscriptMessage[], befo
       continue
     }
 
-    return text
+    reply ??= text
   }
 
-  return passText
+  return reply ?? passText
 }
 
 /** A clarify question blocking inside a member's session, as `session.resume`
@@ -163,6 +187,8 @@ interface GroupSessionSnapshot {
   running?: boolean
   session_id?: string
   session_key?: string
+  /** Start time of the live or retained turn; the `inflight` snapshot omits it. */
+  turn_started_at?: null | number
 }
 
 /** Group turns are explicit user work. A member may be cold or retired when
@@ -196,6 +222,26 @@ export function retainedGroupTurnError(state: GroupSessionSnapshot | null | unde
   return null
 }
 
+/** Identity of the retained failed turn, else null. The `inflight` snapshot
+ *  carries no start time, so two identical consecutive failures differ only
+ *  by `turn_started_at`. */
+function retainedGroupTurnKey(state: GroupSessionSnapshot | null | undefined): null | string {
+  return retainedGroupTurnError(state) === null ? null : JSON.stringify([state?.turn_started_at ?? null, state?.inflight])
+}
+
+/** Does a user row follow the stranded turn's own prompt? Then a later turn
+ *  ran in the session, and a retained error belongs to that turn, not this one. */
+function laterTurnAfterStranded(messages: GroupTurnTranscriptMessage[], before: number): boolean {
+  const anchor = messages.findIndex(
+    (msg, i) =>
+      i >= before && msg?.role === 'user' && groupTranscriptRowText(msg).startsWith(GROUP_PROMPT_HEADER_PREFIX)
+  )
+
+  return messages.some(
+    (msg, i) => i > (anchor === -1 ? before - 1 : anchor) && msg?.role === 'user' && !syntheticGroupUserRow(msg)
+  )
+}
+
 /** Is the member's session still doing work this turn should wait for?
  *  Reading a retained failure as busy kept a dead turn's deadline sliding to
  *  the hard cap and left its stranded marker harvestable forever
@@ -1043,26 +1089,33 @@ async function pollGroupMemberTurn(context: GroupTurnPollContext): Promise<null
     // the transcript, so the tombstone — not the message count — is the only
     // evidence; a tombstone identical to the pre-submit one is an older turn's.
     const failure = retainedGroupTurnError(state)
-    const died = failure !== null && (messages.length > before || JSON.stringify(state?.inflight) !== context.leftover)
+    // Turn start replaces the snapshot and `turn_started_at`, so a retained
+    // error unlike the pre-submit one is THIS turn's: the member did not
+    // finish, and text it wrote before a tool call is not its reply.
+    const failedThisTurn = failure !== null && retainedGroupTurnKey(state) !== context.leftover
+    const died = failedThisTurn || (failure !== null && messages.length > before)
 
     if ((messages.length > before || died) && done) {
-      const replyText = messages.length > before ? pickGroupTurnReply(messages, before) : null
+      const pick = messages.length > before && !failedThisTurn ? pickGroupTurnReply(messages, before) : null
 
-      if (replyText !== null) {
+      if (typeof pick === 'string') {
         recordGroupActivity(context.group, {
-          kind: isGroupPassText(replyText) ? 'passed' : 'replied',
+          kind: isGroupPassText(pick) ? 'passed' : 'replied',
           member: groupMemberKey(member),
           thread
         })
 
-        return replyText
+        return pick
       }
 
       // The turn died on our prompt: surface the gateway's retained error
       // through the failed-turn path (activity row + roster badge) instead of
-      // reading the silence as a pass or sitting out the deadline.
-      if (failure !== null) {
-        throw new Error(failure)
+      // reading the silence as a pass or sitting out the deadline. A failed-turn
+      // row whose error is gone (backend restarted since) still failed.
+      const error = failure ?? pick?.failedNotice ?? null
+
+      if (error !== null) {
+        throw new Error(error)
       }
 
       recordGroupActivity(context.group, {
@@ -1122,7 +1175,7 @@ async function prepareGroupTurnBaseline(
 
     snapshot = pre
     before = Array.isArray(pre?.messages) ? pre.messages.length : pre?.message_count || 0
-    leftover = retainedGroupTurnError(pre) === null ? null : JSON.stringify(pre.inflight)
+    leftover = retainedGroupTurnKey(pre)
 
     if (pre?.session_id) {
       runtimeIds.add(pre.session_id)
@@ -1319,12 +1372,21 @@ export async function harvestStrandedGroupReply(group: string, member: GroupMemb
     const messages = Array.isArray(state?.messages) ? state.messages : []
     // A transcript that never grew is not proof of nothing: a turn that dies
     // before its prompt is committed leaves only the retained error behind.
-    const reply = messages.length > strandedBefore ? pickStrandedGroupTurnReply(messages, strandedBefore) : null
+    // The retained error is the stranded turn's own unless a later turn ran
+    // after it; then it is that turn's error and the late reply still posts.
+    // Text written before a failed tool step is no reply.
+    const retained = laterTurnAfterStranded(messages, strandedBefore) ? null : retainedGroupTurnError(state)
+
+    const pick =
+      retained === null && messages.length > strandedBefore ? pickStrandedGroupTurnReply(messages, strandedBefore) : null
+
+    const reply = typeof pick === 'string' ? pick : null
+    const failedNotice = typeof pick === 'string' ? null : (pick?.failedNotice ?? null)
 
     if (reply === null) {
       // The late turn died instead of answering: say so where the user looks
       // (activity row + roster badge) rather than consuming the marker silently.
-      const failure = retainedGroupTurnError(state)
+      const failure = retained ?? failedNotice
 
       if (failure !== null) {
         const reason = groupFailureReason(failure)
diff --git a/apps/desktop/src/store/voice-prefs.ts b/apps/desktop/src/store/voice-prefs.ts
index c5e9051f22..bb8cf68e1d 100644
--- a/apps/desktop/src/store/voice-prefs.ts
+++ b/apps/desktop/src/store/voice-prefs.ts
@@ -27,9 +27,7 @@ export const $voiceStopPhrase = atom<string | null>(BACKEND_DEFAULT_STOP_PHRASES
 
 /** How the desktop matcher should treat `voice.stop_phrases` (#117801). */
 export type VoiceStopPhraseConfig =
-  | { mode: 'default' }
-  | { mode: 'custom'; phrases: readonly string[] }
-  | { mode: 'disabled' }
+  { mode: 'default' } | { mode: 'custom'; phrases: readonly string[] } | { mode: 'disabled' }
 
 // Full matcher config — kept in sync with `$voiceStopPhrase` so the spoken
 // stop recognizer honours the same list the notice advertises.
diff --git a/apps/desktop/src/types/hermes.ts b/apps/desktop/src/types/hermes.ts
index e288842527..44da09adfa 100644
--- a/apps/desktop/src/types/hermes.ts
+++ b/apps/desktop/src/types/hermes.ts
@@ -636,6 +636,7 @@ export interface SessionMessage {
   display_kind?:
     | 'async_delegation_complete'
     | 'auto_continue'
+    | 'failed_turn'
     | 'hidden'
     | 'model_switch'
     | 'personality_switch'
diff --git a/contributors/emails/orangercc@gmail.com b/contributors/emails/orangercc@gmail.com
new file mode 100644
index 0000000000..4b6d67cb05
--- /dev/null
+++ b/contributors/emails/orangercc@gmail.com
@@ -0,0 +1 @@
+orangercat
diff --git a/gateway/run.py b/gateway/run.py
index 2db97938d6..2a352c2c56 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -1599,8 +1599,12 @@ _ensure_ssl_certs()
 
 sys.path.insert(0, str(Path(__file__).parent.parent))
 
-from hermes_constants import get_hermes_home, get_hermes_home_override
-_hermes_home = get_hermes_home()
+from hermes_constants import get_hermes_home, get_hermes_home_override, get_process_hermes_home
+# The PROCESS's own home, never an import-time ContextVar: a multiplexed backend (``hermes serve``)
+# first imports this module lazily from a session's agent build, under that session's routed profile
+# override, and the import-time config bridge below would then latch the secondary's terminal.* and
+# settings into the launch process env for every later launch-profile turn.
+_hermes_home = get_process_hermes_home()
 
 # Load ~/.hermes/.env first: user-managed env files must override stale shell exports on restart.
 from hermes_cli.env_loader import load_hermes_dotenv
diff --git a/gateway/run_turn.py b/gateway/run_turn.py
index 0d55a752b1..37b52a0944 100644
--- a/gateway/run_turn.py
+++ b/gateway/run_turn.py
@@ -18,7 +18,7 @@ import threading
 import time
 from agent.i18n import t
 from agent.session_activity import format_iteration_progress
-from agent.turn_failure_copy import FAILED_TURN_NOTICE, PARTIAL_FAILED_TURN_NOTICE
+from agent.turn_failure_copy import FAILED_TURN_DISPLAY_KIND, FAILED_TURN_NOTICE, PARTIAL_FAILED_TURN_NOTICE
 from contextlib import nullcontext, suppress
 from contextvars import copy_context
 from gateway.config import Platform
@@ -1722,7 +1722,7 @@ class GatewayTurnMixin:
         if await self.async_session_store.transcript_tail_role(session_id) != "user":
             return
         await self.async_session_store.append_to_transcript(session_id, {
-            "role": "assistant", "content": notice, "timestamp": time.time(),
+            "role": "assistant", "content": notice, "timestamp": time.time(), "display_kind": FAILED_TURN_DISPLAY_KIND,
         })
 
     def _hmwa_classify_turn_failure(self, agent_result, history, session_entry):
diff --git a/hermes_cli/update_cmd_fleet.py b/hermes_cli/update_cmd_fleet.py
index aab627c66b..4b6f80ba57 100644
--- a/hermes_cli/update_cmd_fleet.py
+++ b/hermes_cli/update_cmd_fleet.py
@@ -334,7 +334,10 @@ def _marker_only_restart_obsolete() -> bool:
     that died before its inventory was recorded, #115638) clears once every live gateway is
     current on the checkout — there is no recorded owed set, so the fleet running the code on disk
     is the whole of the evidence the marker's warning can be about, even after HEAD moved past
-    ``expected_sha`` by an out-of-band pull.
+    ``expected_sha`` by an out-of-band pull. With no live gateway at all, the inventory-less marker
+    asks the host instead (``update_cmd_fleet_gatewayless``): it clears when no profile left a
+    gateway that should be running and every live runtime is supervisor-owned or handed off, so a
+    Desktop-only install stops failing every later update (#118742).
 
     A serve/dashboard row whose supervisor owns the restart (Desktop backend, systemd/launchd
     unit, Windows service) is outside the gateway matrix's evidence, not evidence against it —
@@ -347,7 +350,7 @@ def _marker_only_restart_obsolete() -> bool:
     phase never touched it — this marker only stops re-warning about it on every later startup.
     """
     from hermes_cli.update_cmd_fleet_checkout import checkout_contains
-    from hermes_cli.update_serve_obligations import defer_manual_serve
+    from hermes_cli.update_cmd_fleet_gatewayless import host_owes_no_gateway_restart, runtime_outside_gateway_evidence
 
     try:
         fields = _obligation_fields()
@@ -366,10 +369,7 @@ def _marker_only_restart_obsolete() -> bool:
             for runtime in runtimes:
                 if not isinstance(runtime, dict):
                     return False
-                if runtime.get("kind") in ("serve", "dashboard") and (
-                    defer_manual_serve(runtime)
-                    or runtime.get("supervisor") in _SUPERVISOR_OWNED_SERVE_BACKENDS
-                ):
+                if runtime_outside_gateway_evidence(runtime):
                     continue
                 if runtime.get("kind") != "gateway":
                     return False
@@ -403,7 +403,20 @@ def _marker_only_restart_obsolete() -> bool:
         logger.debug("Fleet probe failed; keeping fleet-restart-pending marker: %s", exc)
         return False
     if not fleet:
-        return False  # Absence cannot prove recovery of the recorded inventory.
+        if owed is not None:
+            return False  # Absence cannot prove recovery of the recorded inventory.
+        # No recorded owed set and no live gateway: settle only when the host itself shows nothing
+        # the update could still owe a restart to, and HEAD still holds the code it pulled (#118742).
+        try:
+            gatewayless = (checkout_sha == expected_sha or checkout_contains(expected_sha)) and host_owes_no_gateway_restart()
+        except Exception as exc:
+            logger.debug("Gateway-less host probe failed; keeping fleet-restart-pending marker: %s", exc)
+            return False
+        if not gatewayless:
+            return False
+        _clear_fleet_restart_pending_marker()
+        logger.debug("Fleet-restart-pending marker discharged: host runs no gateway at %s", checkout_sha[:10])
+        return True
     covered = _fleet_covered_gateways(fleet)
     if covered is None:
         return False  # unidentified runtime: the matrix cannot vouch for it
diff --git a/hermes_cli/update_cmd_fleet_gatewayless.py b/hermes_cli/update_cmd_fleet_gatewayless.py
new file mode 100644
index 0000000000..bf2668c32c
--- /dev/null
+++ b/hermes_cli/update_cmd_fleet_gatewayless.py
@@ -0,0 +1,49 @@
+"""Gateway-less host evidence for the update-restart obligation (``update_cmd_fleet`` sibling).
+
+An inventory-less obligation (the pre-inventory writer, or a tail that died before recording one)
+has no owed set, so the live gateway matrix is its whole evidence. On a host that runs no gateway
+(the Desktop app alone, #118742) that matrix is empty forever and the obligation could never
+settle. An empty probe cannot tell that host from one whose gateway the dying update stopped, so
+these readers ask the live host directly. A historical receipt is never consulted: it can belong
+to an older update and says nothing about what runs now.
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict
+
+
+def runtime_outside_gateway_evidence(runtime: dict) -> bool:
+    """A serve/dashboard row the gateway matrix neither covers nor needs to.
+
+    Its supervisor owns the restart (Desktop backend, launchd/systemd unit, Windows service), or it
+    is a manual serve whose restart ``defer_manual_serve`` has handed to its own durable reminder.
+    Unclassified backends and failed transfers stay evidence against settlement (#115090, #111494).
+    """
+    from hermes_cli.update_cmd_fleet import _SUPERVISOR_OWNED_SERVE_BACKENDS
+    from hermes_cli.update_serve_obligations import defer_manual_serve
+
+    return runtime.get("kind") in ("serve", "dashboard") and (
+        defer_manual_serve(runtime) or runtime.get("supervisor") in _SUPERVISOR_OWNED_SERVE_BACKENDS
+    )
+
+
+def host_owes_no_gateway_restart() -> bool:
+    """True when no profile expects a gateway to be running and every live runtime is outside the matrix.
+
+    A ``gateway_state.json`` that does not say ``stopped``/``startup_failed`` belongs to a gateway
+    that went away without a clean stop, which is what an update that died mid-restart leaves
+    behind; it keeps the obligation. Profiles that never ran a gateway have no record at all.
+    """
+    from gateway.status import read_runtime_status
+    from hermes_cli.update_inventory import collect_runtime_inventory
+    from hermes_cli.update_receipt import _NOT_EXPECTED_STATES, _profile_homes
+
+    for _profile, home in _profile_homes():
+        record = read_runtime_status(home / "gateway_state.json")
+        if record is None:
+            continue
+        state = record.get("gateway_state") if isinstance(record, dict) else None
+        if not (isinstance(state, str) and state in _NOT_EXPECTED_STATES):
+            return False
+    return all(runtime_outside_gateway_evidence(asdict(runtime)) for runtime in collect_runtime_inventory().runtimes)
diff --git a/hermes_cli/web_routers/files.py b/hermes_cli/web_routers/files.py
index c2657aa70a..54591d684a 100644
--- a/hermes_cli/web_routers/files.py
+++ b/hermes_cli/web_routers/files.py
@@ -217,7 +217,10 @@ def _fs_default_cwd() -> str:
 
 def _fs_git_branch(cwd: str) -> str:
     try:
-        run_kwargs: Dict[str, Any] = {"capture_output": True, "text": True, "timeout": 2, "check": False}
+        # git emits UTF-8 (branch names, localized "not a git repository" stderr); the locale codec
+        # (cp936 on zh-CN Windows) raised inside communicate()'s reader threads on every poll (#83851).
+        run_kwargs: Dict[str, Any] = {"capture_output": True, "text": True, "encoding": "utf-8",
+                                      "errors": "replace", "timeout": 2, "check": False}
         if sys.platform == "win32":
             run_kwargs["creationflags"] = windows_hide_flags()
         result = subprocess.run(["git", "-C", cwd, "branch", "--show-current"], **run_kwargs)
diff --git a/hermes_cli/web_routers/sessions.py b/hermes_cli/web_routers/sessions.py
index e406017718..97a852b22a 100644
--- a/hermes_cli/web_routers/sessions.py
+++ b/hermes_cli/web_routers/sessions.py
@@ -554,10 +554,16 @@ def _with_tool_call_labels(message: dict) -> dict:
 def _project_for_display(messages: list) -> list:
     from agent.compaction_display import project_compaction_message_for_display
     from agent.context_compressor import is_compaction_summary_message
+    from agent.turn_failure_copy import untyped_failed_turn_display_kind
 
     projected_messages = []
     for message in messages:
         message = _with_tool_call_labels(message)
+        # Same read-side typing as session.resume (tui_gateway/session_history.py).
+        failed_turn = not message.get("display_kind") and untyped_failed_turn_display_kind(
+            message.get("role"), message.get("content"))
+        if failed_turn:
+            message = {**message, "display_kind": failed_turn}
         if not is_compaction_summary_message(message):
             projected_messages.append(message)
             continue
diff --git a/hermes_state.py b/hermes_state.py
index ef7fe49e65..f0a3c55413 100644
--- a/hermes_state.py
+++ b/hermes_state.py
@@ -979,14 +979,7 @@ class SessionDB(
         # mutations, not just idempotent UPSERTs.
         ioerr_begin_retried = False
         while True:
-            self._raise_if_db_corrupt()
-            if storage_state(self.db_path) == STORAGE_CORRUPT:
-                # Another handle in this process already saw structural damage on this file.
-                # Quarantine this one before it touches SQLite; the error type is the same
-                # StateDbCorruptError, so every transcript-diversion owner handles it unchanged.
-                self._halt_db_corrupt(sqlite3.DatabaseError(
-                    "database disk image is malformed (reported earlier in this process: "
-                    f"{storage_corrupt_reason(self.db_path)})"))
+            self._raise_if_db_corrupt(storage=True)
             # NOTE: the replaced/generation live probe runs INSIDE the lock below,
             # not here. close() mutates _conn and _db_sidecar_identity under that
             # same lock, ending the WAL generation (SQLite unlinks the -wal/-shm
@@ -1078,7 +1071,7 @@ class SessionDB(
                             self._raise_if_db_replaced()
                     # Corrupt FTS shadow tables fail every write via the sync triggers while canonical
                     # rows are intact: detach the derived indexes atomically and retry (never rebuild here).
-                    if self._enter_fts_fail_open(exc):
+                    if self._enter_fts_fail_open(exc, deadline=deadline, patience_s=patience_s):
                         continue
                     # What survives both checks is structural damage: quarantine.
                     if self._is_structural_corruption_error(exc):
@@ -1397,9 +1390,16 @@ class SessionDB(
             )
         return retire_without_close
 
-    def _raise_if_db_corrupt(self) -> None:
+    def _raise_if_db_corrupt(self, *, storage: bool = False) -> None:
         if self._db_corrupt:
             raise self._corrupt_error()
+        if storage and storage_state(self.db_path) == STORAGE_CORRUPT:
+            # Another handle in this process already saw structural damage on this file.
+            # Quarantine this one before it touches SQLite; the error type is the same
+            # StateDbCorruptError, so every transcript-diversion owner handles it unchanged.
+            self._halt_db_corrupt(sqlite3.DatabaseError(
+                "database disk image is malformed (reported earlier in this process: "
+                f"{storage_corrupt_reason(self.db_path)})"))
 
     def _sleep_before_write_retry(self, deadline: float, patience_s: float) -> bool:
         """Sleep one jitter interval if the budget allows; True = retry, False = deadline passed. Small
diff --git a/hermes_state_fts.py b/hermes_state_fts.py
index 2160efd1d3..3ced10cb2c 100644
--- a/hermes_state_fts.py
+++ b/hermes_state_fts.py
@@ -5,13 +5,14 @@ FTS-scoped corruption detection and the atomic fail-open trigger detach."""
 import logging
 import os
 import sqlite3
+import time
 from pathlib import Path
 from typing import Sequence
 
 from hermes_constants import get_hermes_home
 from hermes_state_common import (FTS_CJK_STALE_KEY, FTS_STALE_KEY, _FTS_CJK_TRIGGERS, _FTS_TRIGGERS,
     routed_sessions_setting)
-from hermes_state_errors import is_fts_scoped_corruption_error
+from hermes_state_errors import is_fts_scoped_corruption_error, is_sqlite_lock_error
 
 # caplog tests pin the "hermes_state" logger name.
 logger = logging.getLogger("hermes_state")
@@ -350,48 +351,68 @@ class SessionFtsSetupMixin:
         gateway transcript retry: see :func:`hermes_state_errors.is_fts_scoped_corruption_error`."""
         return is_fts_scoped_corruption_error(exc)
 
-    def _enter_fts_fail_open(self, exc: sqlite3.DatabaseError) -> bool:
+    def _enter_fts_fail_open(
+        self, exc: sqlite3.DatabaseError, *, deadline: float | None = None, patience_s: float | None = None,
+    ) -> bool:
         """Detach corrupt FTS indexes so canonical writes can continue. Breadcrumb +
         trigger drop commit atomically: once triggers are absent the index has a
-        gap of unknown extent, so nobody may reinstall them without a full rebuild."""
+        gap of unknown extent, so nobody may reinstall them without a full rebuild.
+
+        A busy write lock is waited out on the caller's write budget (default
+        ``_WRITE_PATIENCE_S``), like ``_execute_write``: the writer connection's busy
+        timeout is only 1 s, and the usual holder is a sibling writer detaching the
+        same corrupt index — giving up after 1 s cost that turn's canonical write."""
         if not self._fts_enabled or not self._is_fts_write_corruption_error(exc):
             return False
-        self._raise_if_db_corrupt()
-        try:
-            with self._lock:
-                self._raise_if_db_replaced()
-                if self._conn is None:
-                    self._reopen_after_close_locked(context="write")
-                self._conn.execute("BEGIN IMMEDIATE")
-                try:
-                    self._conn.execute(
-                        "INSERT INTO state_meta (key, value) VALUES (?, '1') "
-                        "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
-                        (FTS_STALE_KEY,),
-                    )
-                    cjk_triggers_present = self._conn.execute(
-                        "SELECT 1 FROM sqlite_master WHERE type = 'trigger' "
-                        f"AND name IN ({','.join('?' for _ in _FTS_CJK_TRIGGERS)}) "
-                        "LIMIT 1",
-                        _FTS_CJK_TRIGGERS,
-                    ).fetchone()
-                    if cjk_triggers_present:
+        if patience_s is None:
+            patience_s = self._WRITE_PATIENCE_S
+        if deadline is None:
+            deadline = time.monotonic() + patience_s
+        while True:
+            # Re-checked every attempt: a sibling may quarantine the file while we wait for the
+            # lock, and nothing may be committed on a quarantined handle.
+            self._raise_if_db_corrupt(storage=True)
+            try:
+                with self._lock:
+                    self._raise_if_db_replaced()
+                    if self._conn is None:
+                        self._reopen_after_close_locked(context="write")
+                    self._conn.execute("BEGIN IMMEDIATE")
+                    try:
                         self._conn.execute(
                             "INSERT INTO state_meta (key, value) VALUES (?, '1') "
                             "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
-                            (FTS_CJK_STALE_KEY,),
+                            (FTS_STALE_KEY,),
                         )
-                    self._drop_all_fts_triggers(self._conn.cursor())
-                    self._conn.commit()
-                except BaseException:
-                    self._conn.rollback()
-                    raise
-        except sqlite3.Error as detach_exc:
-            logger.error(
-                "Could not detach corrupt FTS indexes; canonical write still cannot proceed: %s",
-                detach_exc,
-            )
-            return False
+                        cjk_triggers_present = self._conn.execute(
+                            "SELECT 1 FROM sqlite_master WHERE type = 'trigger' "
+                            f"AND name IN ({','.join('?' for _ in _FTS_CJK_TRIGGERS)}) "
+                            "LIMIT 1",
+                            _FTS_CJK_TRIGGERS,
+                        ).fetchone()
+                        if cjk_triggers_present:
+                            self._conn.execute(
+                                "INSERT INTO state_meta (key, value) VALUES (?, '1') "
+                                "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
+                                (FTS_CJK_STALE_KEY,),
+                            )
+                        self._drop_all_fts_triggers(self._conn.cursor())
+                        self._conn.commit()
+                    except BaseException:
+                        self._conn.rollback()
+                        raise
+                break
+            except sqlite3.Error as detach_exc:
+                if (
+                    isinstance(detach_exc, sqlite3.OperationalError) and is_sqlite_lock_error(detach_exc)
+                    and self._sleep_before_write_retry(deadline, patience_s)
+                ):
+                    continue
+                logger.error(
+                    "Could not detach corrupt FTS indexes; canonical write still cannot proceed: %s",
+                    detach_exc,
+                )
+                return False
         self._fts_stale = True
         self._fts_enabled = False
         self._trigram_available = False
diff --git a/tests-js/scripts/mock-server.ts b/tests-js/scripts/mock-server.ts
index c9e558562c..6d1d62eff5 100644
--- a/tests-js/scripts/mock-server.ts
+++ b/tests-js/scripts/mock-server.ts
@@ -364,6 +364,19 @@ const TASK_PANEL_RESUME_SCRIPT: ScriptedTurn[] = [
 export const PROVIDER_FAILURE_TRIGGER = 'E2E_PROVIDER_FAILURE_TRIGGER'
 export const PROVIDER_FAILURE_MESSAGE = 'E2E invalid_api_key: the mock refused this completion on purpose'
 
+/**
+ * The same provider failure one step later: the first completion says
+ * TOOL_THEN_FAILURE_TEXT and calls a tool, the completion after the tool
+ * result is the 401. That pre-tool text is not the member's reply.
+ */
+export const TOOL_THEN_FAILURE_TRIGGER = 'E2E_TOOL_THEN_PROVIDER_401'
+export const TOOL_THEN_FAILURE_TEXT = 'Let me note the plan before answering.'
+
+const TOOL_THEN_FAILURE_TURN: ScriptedTurn = {
+  text: TOOL_THEN_FAILURE_TEXT,
+  toolCalls: [{ name: 'todo', args: { todos: [{ id: '1', content: 'Answer the room', status: 'in_progress' }] } }],
+}
+
 const BLOCKING_CLARIFY_TURN: ScriptedTurn = {
   text: '',
   toolCalls: [{ name: 'clarify', args: { question: BLOCKING_CLARIFY_QUESTION, choices: ['Yes', 'No'] } }],
@@ -760,7 +773,17 @@ export function startMockServer(options: MockServerOptions = {}): Promise<MockSe
             return
           }
 
-          if (userText.includes(PROVIDER_FAILURE_TRIGGER)) {
+          if (userText.includes(TOOL_THEN_FAILURE_TRIGGER) && !messages.some(message => message?.role === 'tool')) {
+            if (stream) {
+              streamScriptedTurn(res, model, TOOL_THEN_FAILURE_TURN)
+            } else {
+              nonStreamingScriptedTurn(res, model, TOOL_THEN_FAILURE_TURN)
+            }
+
+            return
+          }
+
+          if (userText.includes(PROVIDER_FAILURE_TRIGGER) || userText.includes(TOOL_THEN_FAILURE_TRIGGER)) {
             res.writeHead(401, { 'Content-Type': 'application/json' })
             res.end(JSON.stringify({ error: { code: 'invalid_api_key', message: PROVIDER_FAILURE_MESSAGE, type: 'invalid_request_error' } }))
 
diff --git a/tests/acp_adapter/test_failed_turn_closure.py b/tests/acp_adapter/test_failed_turn_closure.py
index 590290b76c..5db652681e 100644
--- a/tests/acp_adapter/test_failed_turn_closure.py
+++ b/tests/acp_adapter/test_failed_turn_closure.py
@@ -144,7 +144,7 @@ def test_acp_refusal_closes_the_turn_and_is_not_replayed_into_the_next_prompt(ac
     Invariant: the durable tail after a failed turn is a Hermes-authored assistant row (never
     provider text), and the next prompt reaches the provider as its own user row.
     """
-    from agent.turn_failure_copy import FAILED_TURN_NOTICE
+    from agent.turn_failure_copy import FAILED_TURN_DISPLAY_KIND, FAILED_TURN_NOTICE
 
     provider, prompt, conversation_rows, db, sid, conn, server = acp
 
@@ -155,6 +155,12 @@ def test_acp_refusal_closes_the_turn_and_is_not_replayed_into_the_next_prompt(ac
     assert [r[0] for r in rows] == ["user", "assistant"], rows
     assert rows[1][1] == FAILED_TURN_NOTICE  # no tool ran: no hedging, no provider detail
     assert db.latest_conversation_role(sid) == "assistant"
+    # Typed, so a renderer or room poller never has to match the English copy to tell the
+    # boundary from a model reply.
+    with sqlite3.connect(db.db_path) as c:
+        kinds = [r[0] for r in c.execute(
+            "SELECT display_kind FROM messages WHERE session_id = ? AND role = 'assistant'", (sid,))]
+    assert kinds == [FAILED_TURN_DISPLAY_KIND]
     # The refusal reaches the ACP client, but never becomes canonical assistant history.
     texts = [getattr(getattr(u, "content", None), "text", None) or getattr(u, "text", None) for u in conn.updates]
     assert any(_REFUSAL_DETAIL in (t or "") for t in texts)
@@ -165,6 +171,7 @@ def test_acp_refusal_closes_the_turn_and_is_not_replayed_into_the_next_prompt(ac
     sent = [m for m in provider.requests[-1]["messages"] if m["role"] != "system"]
     assert [m["role"] for m in sent] == ["user", "assistant", "user"], sent
     assert [m["content"] for m in sent if m["role"] == "user"] == [_REFUSED, _NEW_REQUEST]
+    assert sent[1] == {"role": "assistant", "content": FAILED_TURN_NOTICE}  # the type never reaches the wire
 
     # Turn 3: cancelled mid-turn. The interrupt envelope carries ``final_response=None``
     # (no assistant text yet); ``_finish_turn`` must still report ``cancelled`` — never crash
diff --git a/tests/agent/test_command_token_source.py b/tests/agent/test_command_token_source.py
index 71251e9963..78bd86ec1f 100644
--- a/tests/agent/test_command_token_source.py
+++ b/tests/agent/test_command_token_source.py
@@ -77,6 +77,18 @@ class TestMinting:
         assert "dbx" in message          # names the provider to fix
         assert "exited" in message       # states what happened
 
+    def test_undecodable_output_does_not_raise(self):
+        """One non-UTF-8 byte from the helper must not crash token minting.
+
+        A strict decode raised UnicodeDecodeError inside subprocess.run — a ValueError,
+        so neither the TimeoutExpired nor the OSError handler above caught it: provider
+        auth died with a traceback instead of the documented CommandTokenError path.
+        """
+        token, ttl = _mint("printf 'tok\\377'", "dbx")
+
+        assert ttl is None
+        assert token.startswith("tok")
+
 
 class TestNoCredentialLeak:
     def test_failure_message_excludes_command_output(self):
diff --git a/tests/e2e/core/parity/test_entrypoint_parity.py b/tests/e2e/core/parity/test_entrypoint_parity.py
index 79878c20c7..7bc0864786 100644
--- a/tests/e2e/core/parity/test_entrypoint_parity.py
+++ b/tests/e2e/core/parity/test_entrypoint_parity.py
@@ -79,10 +79,7 @@ DRIVERS: dict[str, Driver] = {
 # Cells that are red on current main for a tracked, open bug. Strict: the test
 # FAILS as soon as the cell turns green, so the entry is removed with the fix
 # instead of silently masking a later regression of the same cell.
-KNOWN_RED: dict[tuple[str, str], str] = {
-    # ACP builds its AIAgent without the configured agent.disabled_toolsets.
-    ("acp (stdio)", "toolset_restriction"): "#74582",
-}
+KNOWN_RED: dict[tuple[str, str], str] = {}
 
 
 @dataclass
diff --git a/tests/gateway/test_config_env_bridge_authority.py b/tests/gateway/test_config_env_bridge_authority.py
index 0f322e33a1..fc4bde4cd7 100644
--- a/tests/gateway/test_config_env_bridge_authority.py
+++ b/tests/gateway/test_config_env_bridge_authority.py
@@ -21,7 +21,9 @@ import pytest
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
 
 
-def _run_gateway_import(hermes_home: Path, initial_env: dict[str, str]) -> dict[str, str]:
+def _run_gateway_import(
+    hermes_home: Path, initial_env: dict[str, str], routed_home: Path | None = None
+) -> dict[str, str]:
     """Import gateway.run in a clean subprocess and return the post-import env.
 
     The bridge runs at module-import time, so simply importing is enough
@@ -33,6 +35,9 @@ def _run_gateway_import(hermes_home: Path, initial_env: dict[str, str]) -> dict[
         f"""
         import os, sys
         sys.path.insert(0, {str(PROJECT_ROOT)!r})
+        if {str(routed_home or "")!r}:
+            from hermes_constants import set_hermes_home_override
+            set_hermes_home_override({str(routed_home or "")!r})
 
         try:
             from gateway import run  # noqa: F401  — module import triggers bridge
@@ -50,6 +55,7 @@ def _run_gateway_import(hermes_home: Path, initial_env: dict[str, str]) -> dict[
             "HERMES_GATEWAY_BUSY_TEXT_MODE",
             "HERMES_GATEWAY_PLATFORM_CONNECT_TIMEOUT",
             "HERMES_TIMEZONE",
+            "TERMINAL_CWD",
         ):
             v = os.environ.get(k)
             if v is not None:
@@ -218,3 +224,22 @@ def test_env_platform_connect_timeout_wins_over_config(hermes_home: Path) -> Non
     )
 
     assert env.get("HERMES_GATEWAY_PLATFORM_CONNECT_TIMEOUT") == "120"
+
+
+def test_first_import_under_a_routed_override_bridges_the_process_home(tmp_path: Path) -> None:
+    """A multiplexed backend first imports gateway.run lazily inside a routed profile's session;
+    the import-time bridge must still write the LAUNCH home's config into the process env."""
+    import yaml
+
+    homes = {}
+    for name, turns in (("launch", 111), ("routed", 222)):
+        home = tmp_path / name
+        (home / "work").mkdir(parents=True)
+        cfg = {"agent": {"max_turns": turns}, "terminal": {"cwd": str(home / "work")}}
+        (home / "config.yaml").write_text(yaml.safe_dump(cfg), encoding="utf-8")
+        homes[name] = home
+
+    env = _run_gateway_import(homes["launch"], {}, routed_home=homes["routed"])
+
+    assert env.get("HERMES_MAX_ITERATIONS") == "111"
+    assert env.get("TERMINAL_CWD") == str(homes["launch"] / "work")
diff --git a/tests/gateway/test_failure_writer_ownership.py b/tests/gateway/test_failure_writer_ownership.py
index 86839736a4..71a946350f 100644
--- a/tests/gateway/test_failure_writer_ownership.py
+++ b/tests/gateway/test_failure_writer_ownership.py
@@ -5,7 +5,7 @@ import subprocess
 import sys
 from pathlib import Path
 
-from agent.turn_failure_copy import PARTIAL_FAILED_TURN_NOTICE
+from agent.turn_failure_copy import FAILED_TURN_DISPLAY_KIND, PARTIAL_FAILED_TURN_NOTICE
 
 
 def test_gateway_failure_writer_preserves_accepted_turn_identity(tmp_path):
@@ -228,8 +228,10 @@ def test_fresh_session_agent_flushed_failed_turn_is_closed(tmp_path):
                 response="x", agent_failed_early=True, hidden_reasoning_incomplete=False,
                 is_context_overflow_failure=False,
             )
-        roles = [m["role"] for m in db.get_messages(sid) if m["role"] != "session_meta"]
-        assert roles == ["user", "assistant"]
+        rows = [m for m in db.get_messages(sid) if m["role"] != "session_meta"]
+        assert [m["role"] for m in rows] == ["user", "assistant"]
+        # Typed like the core closer's row, or Desktop reads the boundary as the model's reply.
+        assert rows[-1]["display_kind"] == FAILED_TURN_DISPLAY_KIND
         assert store.transcript_tail_role(sid) == "assistant"
         db.close()
 
diff --git a/tests/hermes_cli/test_manual_serve_deferral.py b/tests/hermes_cli/test_manual_serve_deferral.py
index 7e5497ac34..05c1440454 100644
--- a/tests/hermes_cli/test_manual_serve_deferral.py
+++ b/tests/hermes_cli/test_manual_serve_deferral.py
@@ -84,16 +84,17 @@ def test_historical_manual_obligation_does_not_block_healthy_gateway(monkeypatch
     monkeypatch.setattr("hermes_cli.update_cmd._current_checkout_sha", lambda: "new")
     monkeypatch.setattr(process_identity, "_pid_alive_matches", lambda *a: alive)
     monkeypatch.setattr(update_receipt, "collect_fleet_versions", lambda **k: [{"profile": "default", "state": "current", "code_sha": "new"}] if gateway_present else [])
+    monkeypatch.setattr("hermes_cli.update_inventory.collect_runtime_inventory", lambda: UpdatePlan(runtimes=[runtime] if alive is not False else []))
     if marker:
         fleet._write_fleet_restart_pending_marker(expected_sha="new")
-    # An inventory-less marker never inherits inventory from a historical receipt, but it
-    # discharges when the live fleet provably serves its expected SHA (#115638).
-    pending = marker and not gateway_present
-    assert fleet._pending_fleet_restart_needed() is pending
+    # An inventory-less marker never inherits inventory from a historical receipt. It discharges
+    # when the live fleet provably serves its expected SHA (#115638), or when the host runs no
+    # gateway and the manual serve has its own reminder (#118742).
+    assert not fleet._pending_fleet_restart_needed()
     fleet._warn_pending_fleet_restart_on_startup()
     warning = capsys.readouterr().err
     assert ("serve [work] pid 900" in warning) is (alive is not False)
-    assert ("hermes gateway restart" in warning) is pending
+    assert "hermes gateway restart" not in warning
     assert json.loads((root / "latest.json").read_text()) == receipt
 
 
@@ -108,13 +109,16 @@ def test_stamped_manual_only_history_has_no_gateway_obligation(monkeypatch, caps
     monkeypatch.setattr("hermes_cli.update_cmd._current_checkout_sha", lambda: "new")
     monkeypatch.setattr(fleet, "_current_checkout_sha", lambda: "new")
     monkeypatch.setattr(update_receipt, "collect_fleet_versions", lambda **k: [])
+    monkeypatch.setattr("hermes_cli.update_inventory.collect_runtime_inventory", lambda: UpdatePlan(runtimes=[RuntimeRecord(**runtime)]))
     if marker:
         fleet._write_fleet_restart_pending_marker(expected_sha="new")
-    assert fleet._pending_fleet_restart_needed() is marker
+    # With no gateway on the host, the live manual serve carries its own reminder and an
+    # inventory-less marker has nothing left to hold (#118742).
+    assert not fleet._pending_fleet_restart_needed()
     fleet._warn_pending_fleet_restart_on_startup()
     warning = capsys.readouterr().err
     assert "serve [work] pid 900" in warning
-    assert ("hermes gateway restart" in warning) is marker
+    assert "hermes gateway restart" not in warning
 
 
 @pytest.mark.parametrize("manual_first", [True, False])
diff --git a/tests/hermes_cli/test_session_message_page_owner.py b/tests/hermes_cli/test_session_message_page_owner.py
index 3e43acdfcd..5d453f83ab 100644
--- a/tests/hermes_cli/test_session_message_page_owner.py
+++ b/tests/hermes_cli/test_session_message_page_owner.py
@@ -47,3 +47,40 @@ def test_message_pages_identify_the_serving_profile(tmp_path, monkeypatch, servi
     assert [row["content"] for row in older["messages"] + tail["messages"]] == [
         f"message-{index}" for index in range(199)
     ]
+
+
+def test_message_pages_type_untyped_failed_turn_rows(tmp_path, monkeypatch):
+    """Desktop cold-loads and pages through REST, not ``session.resume``: a failed-turn boundary
+    written before the closers typed it must reach it as ``failed_turn``, not model text."""
+    from agent.turn_failure_copy import FAILED_TURN_DISPLAY_KIND, FAILED_TURN_NOTICE, PARTIAL_FAILED_TURN_NOTICE
+    from hermes_state import SessionDB
+
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr("hermes_state.DEFAULT_DB_PATH", home / "state.db")
+    db = SessionDB(db_path=home / "state.db")
+    try:
+        db.create_session(session_id="s", source="desktop")
+        db.append_messages_batch("s", [
+            {"role": "user", "content": "a"},
+            {"role": "assistant", "content": FAILED_TURN_NOTICE},
+            {"role": "user", "content": "b"},
+            {"role": "assistant", "content": PARTIAL_FAILED_TURN_NOTICE},
+            {"role": "user", "content": "c"},
+            {"role": "assistant", "content": f"Quoting Hermes: {FAILED_TURN_NOTICE}"},
+        ])
+    finally:
+        db.close()
+
+    from hermes_cli.web_routers.sessions import manage_router
+
+    app = FastAPI()
+    app.include_router(manage_router)
+    with TestClient(app) as client:
+        rows = client.get("/api/sessions/s/messages").json()["messages"]
+
+    assert [row.get("display_kind") for row in rows] == [
+        None, FAILED_TURN_DISPLAY_KIND, None, FAILED_TURN_DISPLAY_KIND, None, None,
+    ]
diff --git a/tests/hermes_cli/test_update_scoped_reconciliation.py b/tests/hermes_cli/test_update_scoped_reconciliation.py
index 32764bcb20..1fabeed5f0 100644
--- a/tests/hermes_cli/test_update_scoped_reconciliation.py
+++ b/tests/hermes_cli/test_update_scoped_reconciliation.py
@@ -4,13 +4,24 @@ import json
 
 import pytest
 
-from hermes_cli import process_identity, update_cmd_fleet as fleet, update_receipt
+from hermes_cli import process_identity, update_cmd_fleet as fleet, update_inventory, update_receipt
+from hermes_cli.update_inventory import RuntimeRecord, UpdatePlan
 from hermes_constants import get_hermes_home
 import hermes_cli.update_host_obligation as host_obligation
 
 MANUAL = {"kind": "serve", "profile": "work", "pid": 900, "supervisor": "manual-serve", "restart_via": "respawn-argv", "code_sha": "old", "detail": {"create_time": 1000.0}}
 CURRENT = {"profile": "alpha", "state": "current", "code_sha": "new"}
 GATEWAY = {"kind": "gateway", "profile": "alpha", "code_sha": "old"}
+DEAD_PID = 2**22 - 7
+
+
+def write_gateway_state(home, state):
+    """``gateway_state.json`` for a gateway whose pid is gone; ``state=None`` omits the field."""
+    record = {"pid": DEAD_PID, "code_sha": "old"}
+    if state is not None:
+        record["gateway_state"] = state
+    home.mkdir(parents=True, exist_ok=True)
+    (home / "gateway_state.json").write_text(json.dumps(record))
 
 CASES = [
     ("receipt-successor", {"outcome": "failed", "plan": {"runtimes": [GATEWAY]}}, None, [CURRENT], False),
@@ -37,6 +48,11 @@ def seed(monkeypatch, old, marker, live, alive=True):
     monkeypatch.setattr(fleet, "_current_checkout_sha", lambda: "new")
     monkeypatch.setattr("hermes_cli.update_cmd._current_checkout_sha", lambda: "new")
     monkeypatch.setattr(update_receipt, "collect_fleet_versions", lambda **k: live)
+    # Hold the host constant at one that still owes a gateway (the update stopped it and nothing
+    # replaced it), so an empty live fleet stays unproven and only the receipt varies. Hosts that run
+    # no gateway settle on host evidence: test_gatewayless_host_settles_on_host_evidence.
+    write_gateway_state(get_hermes_home(), "running")
+    monkeypatch.setattr(update_inventory, "collect_runtime_inventory", UpdatePlan)
     if marker is not None:
         fleet._write_fleet_restart_pending_marker(expected_sha=marker)
     return target
@@ -64,28 +80,60 @@ def test_scoped_reconciliation_matrix(monkeypatch, capsys, name, old, marker, li
         assert target.read_bytes() == before
 
 
-@pytest.mark.parametrize("suffix", ["{", '{"version": 1, "inventory": ', "broken-line"])
-def test_malformed_obligation_stays_pending(monkeypatch, suffix):
-    """An unparseable record is an obligation whose terms are unknown — never a discharged one."""
-    seed(monkeypatch, {}, "new", [CURRENT])
-    marker = host_obligation.host_obligation_path()
-    marker.write_text(marker.read_text(encoding="utf-8") + suffix, encoding="utf-8")
+DESKTOP = RuntimeRecord(kind="serve", profile="default", pid=901, supervisor="desktop", restart_via="desktop-respawn")
+DESKTOP_RECEIPT = {"outcome": "failed", "plan": {"runtimes": [{"kind": "serve", "profile": "default", "supervisor": "desktop"}]}}
+
+# (name, receipt, live runtimes, gateway_state per profile, checkout, pending)
+GATEWAYLESS_CASES = [
+    # #118742: Desktop app only, no gateway was ever installed.
+    ("desktop-only", {}, [DESKTOP], {}, "new", False),
+    ("nothing-running", {}, [], {}, "new", False),
+    ("gateway-stopped-cleanly", {}, [DESKTOP], {"default": "stopped"}, "new", False),
+    ("gateway-startup-failed", {}, [], {"default": "startup_failed"}, "new", False),
+    # HEAD carries a local commit on top of the pulled SHA (#119367).
+    ("carried-local-commit", {}, [DESKTOP], {}, "hotfix", False),
+    # Receipts neither discharge nor block: the old manual row is history, not the live host.
+    ("manual-receipt-gatewayless-host", {"outcome": "success", "plan": {"runtimes": [MANUAL]}}, [], {}, "new", False),
+    ("desktop-receipt-stopped-gateway", DESKTOP_RECEIPT, [DESKTOP], {"default": "running"}, "new", True),
+    ("named-profile-gateway-gone", {}, [DESKTOP], {"work": "running"}, "new", True),
+    ("state-record-without-state", {}, [], {"default": None}, "new", True),
+    ("unclassified-serve", {}, [RuntimeRecord(kind="serve", profile="default", pid=902, supervisor="manual")], {}, "new", True),
+    ("manual-serve-without-identity", {}, [RuntimeRecord(kind="serve", profile="work", pid=903, supervisor="manual-serve", restart_via="respawn-argv")], {}, "new", True),
+    ("gateway-runtime-without-fleet-row", {}, [RuntimeRecord(kind="gateway", profile="default", pid=904, supervisor="manual")], {}, "new", True),
+    ("checkout-diverged", {}, [DESKTOP], {}, "elsewhere", True),
+]
+
+
+@pytest.mark.parametrize("name,receipt,runtimes,states,checkout,pending", GATEWAYLESS_CASES, ids=[case[0] for case in GATEWAYLESS_CASES])
+def test_gatewayless_host_settles_on_host_evidence(monkeypatch, capsys, name, receipt, runtimes, states, checkout, pending):
+    """An inventory-less marker with no live gateway settles on what the host runs now (#118742)."""
+    from hermes_cli.profiles import _get_default_hermes_home, _get_profiles_root
+
+    seed(monkeypatch, receipt, "new", [])
+    (get_hermes_home() / "gateway_state.json").unlink()
+    for profile, state in states.items():
+        write_gateway_state(_get_default_hermes_home() if profile == "default" else _get_profiles_root() / profile, state)
+    monkeypatch.setattr(update_inventory, "collect_runtime_inventory", lambda: UpdatePlan(runtimes=list(runtimes)))
+    monkeypatch.setattr(fleet, "_current_checkout_sha", lambda: checkout)
+    monkeypatch.setattr("hermes_cli.update_cmd._current_checkout_sha", lambda: checkout)
+    monkeypatch.setattr("hermes_cli.update_cmd_fleet_checkout.checkout_contains", lambda sha: checkout == "hotfix")
+
+    assert fleet._pending_fleet_restart_needed() is pending
+    assert host_obligation.host_obligation_path().exists() is pending
+    fleet._warn_pending_fleet_restart_on_startup()
+    assert ("hermes gateway restart" in capsys.readouterr().err) is pending
+
+
+def test_gatewayless_probe_failure_keeps_marker(monkeypatch):
+    seed(monkeypatch, {}, "new", [])
+    (get_hermes_home() / "gateway_state.json").unlink()
+
+    def unavailable():
+        raise OSError("process table unreadable")
+
+    monkeypatch.setattr(update_inventory, "collect_runtime_inventory", unavailable)
     assert fleet._pending_fleet_restart_needed()
-    assert marker.exists()
-
-
-def test_marker_reconciliation_collects_one_live_snapshot(monkeypatch):
-    seed(monkeypatch, {}, "new", [CURRENT])
-    fleet._write_fleet_restart_pending_marker(expected_sha="new", runtimes=[GATEWAY])
-    probes = []
-
-    def collect(**kwargs):
-        probes.append(True)
-        return [CURRENT] if len(probes) == 1 else []
-
-    monkeypatch.setattr(update_receipt, "collect_fleet_versions", collect)
-    assert not fleet._pending_fleet_restart_needed()
-    assert len(probes) == 1
+    assert host_obligation.host_obligation_path().exists()
 
 
 @pytest.mark.parametrize("completed_restart", [False, True])
@@ -122,3 +170,32 @@ def test_inventory_less_marker_settles_after_out_of_band_pull(monkeypatch, capsy
     assert ("hermes gateway restart" in capsys.readouterr().err) is pending
     assert fleet._pending_fleet_restart_needed() is pending
     assert marker.exists() is pending
+
+
+# Explicit `inventory=null` records no obligation either, so it settles on live-fleet
+# evidence like a missing line (#115638), and an explicit empty inventory owes nothing
+# (#115311); the variants below stay fail-closed.
+
+
+@pytest.mark.parametrize("suffix", ["{", '{"version": 1, "inventory": ', "broken-line"])
+def test_malformed_obligation_stays_pending(monkeypatch, suffix):
+    """An unparseable record is an obligation whose terms are unknown — never a discharged one."""
+    seed(monkeypatch, {}, "new", [CURRENT])
+    marker = host_obligation.host_obligation_path()
+    marker.write_text(marker.read_text(encoding="utf-8") + suffix, encoding="utf-8")
+    assert fleet._pending_fleet_restart_needed()
+    assert marker.exists()
+
+
+def test_marker_reconciliation_collects_one_live_snapshot(monkeypatch):
+    seed(monkeypatch, {}, "new", [CURRENT])
+    fleet._write_fleet_restart_pending_marker(expected_sha="new", runtimes=[GATEWAY])
+    probes = []
+
+    def collect(**kwargs):
+        probes.append(True)
+        return [CURRENT] if len(probes) == 1 else []
+
+    monkeypatch.setattr(update_receipt, "collect_fleet_versions", collect)
+    assert not fleet._pending_fleet_restart_needed()
+    assert len(probes) == 1
diff --git a/tests/hermes_cli/test_web_server_files.py b/tests/hermes_cli/test_web_server_files.py
index 344b4579f0..fd9ed4d078 100644
--- a/tests/hermes_cli/test_web_server_files.py
+++ b/tests/hermes_cli/test_web_server_files.py
@@ -419,3 +419,23 @@ def test_credential_dir_trees_blocked_on_subdir_descent(forced_files_client):
     assert [e["name"] for e in mcp_listing.json()["entries"]] == []
 
 
+
+
+def test_git_branch_decodes_utf8_under_a_gbk_default_codec(tmp_path, monkeypatch):
+    """#83851: the Desktop polls ``/api/fs/default-cwd``; on zh-CN Windows the serve process's default
+    subprocess codec is cp936, and git's UTF-8 output (branch names, localized stderr) raised
+    UnicodeDecodeError in communicate()'s reader threads on every poll. The branch must round-trip."""
+    import shutil
+    import subprocess
+
+    git = shutil.which("git")
+    if git is None:
+        pytest.skip("git not installed")
+    branch = "功能/✅-修复"  # UTF-8 bytes that are illegal multibyte sequences in GBK
+    subprocess.run([git, "init", "-q", str(tmp_path)], check=True)
+    subprocess.run([git, "-C", str(tmp_path), "symbolic-ref", "HEAD", f"refs/heads/{branch}"], check=True)
+    # subprocess resolves an unspecified text-mode codec through _text_encoding() → locale.getencoding()
+    # (cp936 on zh-CN Windows); patch that seam since run_tests.sh's PYTHONUTF8=1 short-circuits locale.
+    monkeypatch.setattr(subprocess, "_text_encoding", lambda: "gbk")
+
+    assert _rt_files._fs_git_branch(str(tmp_path)) == branch
diff --git a/tests/hermes_state/test_fts_index_fail_open.py b/tests/hermes_state/test_fts_index_fail_open.py
index 834d5e49ba..496ca0f7a2 100644
--- a/tests/hermes_state/test_fts_index_fail_open.py
+++ b/tests/hermes_state/test_fts_index_fail_open.py
@@ -11,14 +11,18 @@ user-facing guidance:
   ``messages`` (the turn proceeds);
 * an FTS-scoped error that still escapes (detach refused) classifies as ``fts_index`` and
   never quarantines the handle;
+* a sibling process holding the write lock when the detach runs is waited out, not a lost write;
+* a quarantine that lands while the detach waits stops it: nothing is committed on the file;
 """
 
 import sqlite3
+import threading
 from types import SimpleNamespace
 
 import pytest
 
-from hermes_state import SessionDB
+from hermes_state import SessionDB, StateDbCorruptError
+from hermes_state_health import mark_storage_corrupt, reset_storage_state
 from run_agent import AIAgent
 
 
@@ -115,7 +119,7 @@ def test_escaped_fts_only_error_is_index_scoped_not_quarantined(tmp_path, monkey
     try:
         _seed(db)
         _stomp_fts_shadow(db_path)
-        monkeypatch.setattr(db, "_enter_fts_fail_open", lambda exc: False)
+        monkeypatch.setattr(db, "_enter_fts_fail_open", lambda exc, **_: False)
         agent = _flush_agent(db, "s1")
 
         ok = agent._flush_messages_to_session_db(
@@ -131,3 +135,107 @@ def test_escaped_fts_only_error_is_index_scoped_not_quarantined(tmp_path, monkey
         assert "refused detach" not in _contents(db_path)
     finally:
         db.close()
+
+
+def test_detach_waits_out_a_sibling_holding_the_write_lock(tmp_path):
+    """Gateway + TUI hit the same corrupt index: one detaches while the other waits. A sibling
+    that takes the write lock between this writer's corruption error and its detach, and holds
+    it past the writer connection's 1 s busy timeout, must be waited out on the write budget —
+    the canonical row lands instead of escaping as 'database disk image is malformed'."""
+    db_path = tmp_path / "state.db"
+    db = SessionDB(db_path=db_path)
+    try:
+        _seed(db, rows=5)
+        _stomp_fts_shadow(db_path)
+        held, release = threading.Event(), threading.Event()
+
+        def sibling():
+            raw = sqlite3.connect(str(db_path), timeout=30, isolation_level=None)
+            raw.execute("BEGIN IMMEDIATE")
+            held.set()
+            release.wait(10)
+            raw.execute("COMMIT")
+            raw.close()
+
+        real_check = db._is_fts_write_corruption_error
+        holder = []
+
+        def check_then_contend(exc):
+            hit = real_check(exc)
+            if hit and not holder:  # the writer has rolled back; the sibling grabs the lock now
+                holder.append(threading.Thread(target=sibling))
+                holder[0].start()
+                assert held.wait(10)
+                threading.Timer(1.6, release.set).start()
+            return hit
+
+        db._is_fts_write_corruption_error = check_then_contend
+        db.append_message("s1", "user", "lands after the sibling lets go")
+        if not holder:
+            pytest.skip("this SQLite build defers FTS shadow corruption past the insert trigger")
+        holder[0].join(10)
+
+        assert _contents(db_path)[-1] == "lands after the sibling lets go"
+        assert db._fts_stale is True
+        assert db._db_corrupt is False
+    finally:
+        db.close()
+
+
+def test_quarantine_while_detach_waits_commits_nothing(tmp_path):
+    """The detach may now wait up to the write budget for the lock. A sibling that quarantines
+    this file meanwhile (structural corruption latched process-wide) must stop it: the retry
+    drops no triggers, commits no stale breadcrumb, and the corrupt error surfaces."""
+    db_path = tmp_path / "state.db"
+    db = SessionDB(db_path=db_path)
+    try:
+        _seed(db, rows=5)
+        _stomp_fts_shadow(db_path)
+        held, release = threading.Event(), threading.Event()
+
+        def sibling():
+            raw = sqlite3.connect(str(db_path), timeout=30, isolation_level=None)
+            raw.execute("BEGIN IMMEDIATE")
+            held.set()
+            release.wait(10)
+            raw.execute("COMMIT")
+            raw.close()
+
+        real_check, real_sleep = db._is_fts_write_corruption_error, db._sleep_before_write_retry
+        holder = []
+
+        def check_then_contend(exc):
+            hit = real_check(exc)
+            if hit and not holder:
+                holder.append(threading.Thread(target=sibling))
+                holder[0].start()
+                assert held.wait(10)
+            return hit
+
+        def quarantine_then_sleep(deadline, patience_s):
+            mark_storage_corrupt(db_path, "database disk image is malformed (sibling handle)")
+            release.set()
+            return real_sleep(deadline, patience_s)
+
+        db._is_fts_write_corruption_error = check_then_contend
+        db._sleep_before_write_retry = quarantine_then_sleep
+        with pytest.raises(StateDbCorruptError):
+            db.append_message("s1", "user", "must not land on a quarantined file")
+        if not holder:
+            pytest.skip("this SQLite build defers FTS shadow corruption past the insert trigger")
+        holder[0].join(10)
+
+        raw = sqlite3.connect(str(db_path))
+        try:
+            triggers = raw.execute(
+                "SELECT count(*) FROM sqlite_master WHERE type = 'trigger' AND name LIKE 'messages_fts%'"
+            ).fetchone()[0]
+            stale = raw.execute("SELECT value FROM state_meta WHERE key LIKE 'fts%stale%'").fetchall()
+        finally:
+            raw.close()
+        assert triggers > 0
+        assert stale == []
+        assert db._fts_stale is False
+    finally:
+        db.close()
+        reset_storage_state(db_path)
diff --git a/tests/tools/test_bot_mode_dm.py b/tests/tools/test_bot_mode_dm.py
index 20ff6cf632..898aede69b 100644
--- a/tests/tools/test_bot_mode_dm.py
+++ b/tests/tools/test_bot_mode_dm.py
@@ -1150,3 +1150,33 @@ def test_poll_reply_is_persisted_as_a_delivery_row_when_the_runner_exits(tmp_pat
     assert kw["display_kind"] == "process_complete"
     assert "PAYLOAD_SENTINEL_42" in kw["content"]
     assert procs[0].id in kw["content"]
+
+
+def test_local_turn_survives_undecodable_transport_output(tmp_path, capsys):
+    """A transport that exits 0 while printing a non-UTF-8 byte must still deliver.
+
+    A strict decode raised UnicodeDecodeError inside subprocess.run — a ValueError, so no
+    handler caught it and the delivery crashed instead of re-emitting the transport's
+    streams (stdout is the reply text the completion notification carries back).
+    """
+    dm_file = tmp_path / "dm.txt"
+    dm_file.write_text("hello", encoding="utf-8")
+    argv = [sys.executable, "-c", "import sys; sys.stdout.buffer.write(b'reply \\377')"]
+
+    assert bot_mode_dm._run_local_turn(argv, str(dm_file)) == 0
+    assert "reply" in capsys.readouterr().out
+
+
+def test_local_turn_relays_utf8_reply_under_a_gbk_default_codec(tmp_path, monkeypatch, capsys):
+    """#83851: the transport is a Hermes CLI child, which always writes UTF-8 stdio. Decoding it with
+    the host's default codec (cp936 on zh-CN Windows) crashed or garbled the reply; it must round-trip."""
+    dm_file = tmp_path / "dm.txt"
+    dm_file.write_text("hello", encoding="utf-8")
+    reply = "✅ 已完成…"
+    argv = [sys.executable, "-c", f"import sys; sys.stdout.buffer.write({reply.encode('utf-8')!r})"]
+    # subprocess resolves an unspecified text-mode codec through _text_encoding() → locale.getencoding();
+    # patch that seam since run_tests.sh's PYTHONUTF8=1 short-circuits the locale lookup.
+    monkeypatch.setattr(subprocess, "_text_encoding", lambda: "gbk")
+
+    assert bot_mode_dm._run_local_turn(argv, str(dm_file)) == 0
+    assert reply in capsys.readouterr().out
diff --git a/tests/tools/test_terminal_scope_multiplex.py b/tests/tools/test_terminal_scope_multiplex.py
index 1ad67eda64..271f310701 100644
--- a/tests/tools/test_terminal_scope_multiplex.py
+++ b/tests/tools/test_terminal_scope_multiplex.py
@@ -110,7 +110,9 @@ def test_routed_turn_reads_every_terminal_consumer_from_profile(
         assert cfg["cwd"] == str(b_cwd)
         assert cfg["docker_volumes"] == []
         assert cfg["docker_shared_container_key"] == ""
-        assert tt._resolve_container_task_id(None) == "default"
+        # Session-less (cron) work for routed B keys B's own environment, never the launch
+        # profile's shared "default" one (which carries A's env and terminal policy).
+        assert tt._resolve_container_task_id(None) == f"home:{os.path.realpath(home)}"
         assert gbase._parse_docker_volume_mounts() == []
         assert not any(
             "alpha-shared" in c for c in gbase._docker_sandbox_dir_candidates("agent:bee:x")
@@ -143,6 +145,30 @@ def test_profile_omitting_keys_gets_defaults_not_launch_values(tmp_path):
     assert json.loads(os.environ["TERMINAL_DOCKER_VOLUMES"])  # A unchanged
 
 
+def test_persistent_docker_routed_profile_keeps_one_container(tmp_path):
+    """Persistent Docker is profile-scoped: a routed profile's session-less (cron) work must key the
+    SAME container as its session-bound work, and never another profile's."""
+    import gateway.run as gw
+    import tools.terminal_tool as tt
+    from gateway.session_context import clear_session_vars, reset_session_vars, set_session_vars
+
+    docker = "terminal:\n  backend: docker\n  container_persistent: true\n"
+    keys = {}
+    for name in ("bee", "wasp"):
+        home = _profile(tmp_path, name, docker)
+        with gw._profile_runtime_scope(home):
+            cron_key = tt._resolve_container_task_id(None)
+            tokens = set_session_vars(session_key=f"agent:{name}:chat", profile=name)
+            try:
+                session_key = tt._resolve_container_task_id(None)
+            finally:
+                clear_session_vars(tokens)
+                reset_session_vars()
+        assert cron_key == session_key == f"profile:{name}"
+        keys[name] = cron_key
+    assert keys["bee"] != keys["wasp"]
+
+
 def test_malformed_profile_config_refuses_execution(tmp_path):
     """Unresolvable policy → refusal scope; terminal_tool refuses instead of
     running under the launch process's ambient policy (fail closed)."""
diff --git a/tests/tui_gateway/test_tui_gateway_server.py b/tests/tui_gateway/test_tui_gateway_server.py
index a5ba995927..ceb5bc73ea 100644
--- a/tests/tui_gateway/test_tui_gateway_server.py
+++ b/tests/tui_gateway/test_tui_gateway_server.py
@@ -2518,6 +2518,25 @@ def test_history_to_messages_preserves_tool_calls_for_resume_display():
     ]
 
 
+def test_history_to_messages_types_the_failed_turn_boundary_for_resume():
+    """Desktop keys the failed-turn boundary on ``display_kind`` (a room poller must not post it
+    as the member's reply); rows written before the closer typed it are typed on read."""
+    from agent.turn_failure_copy import FAILED_TURN_DISPLAY_KIND, FAILED_TURN_NOTICE, PARTIAL_FAILED_TURN_NOTICE
+
+    history = [
+        {"role": "user", "content": "a"},
+        {"role": "assistant", "content": FAILED_TURN_NOTICE, "display_kind": FAILED_TURN_DISPLAY_KIND},
+        {"role": "user", "content": "b"},
+        {"role": "assistant", "content": PARTIAL_FAILED_TURN_NOTICE},  # legacy untyped row
+        {"role": "user", "content": "c"},
+        {"role": "assistant", "content": f"Quoting Hermes: {FAILED_TURN_NOTICE}"},  # a real reply
+    ]
+
+    assert [m.get("display_kind") for m in server._history_to_messages(history)] == [
+        None, FAILED_TURN_DISPLAY_KIND, None, FAILED_TURN_DISPLAY_KIND, None, None,
+    ]
+
+
 def test_history_to_messages_drops_pure_compaction_scaffolding():
     from agent.context_compressor import (
         HISTORICAL_TASK_HEADING,
diff --git a/tools/bot_mode_dm.py b/tools/bot_mode_dm.py
index a1f6d01be3..70bcbfb991 100644
--- a/tools/bot_mode_dm.py
+++ b/tools/bot_mode_dm.py
@@ -416,7 +416,7 @@ def _run_local_turn(argv: list[str], dm_file: str, *, env: Optional[dict[str, st
 
     def _turn(turn_env=env):
         return subprocess.run([*argv, "--query-file", dm_file], check=False, stdin=subprocess.DEVNULL,
-                              capture_output=True, text=True, encoding="utf-8", env=turn_env)
+                              capture_output=True, text=True, encoding="utf-8", errors="replace", env=turn_env)
 
     proc = _turn()
     if proc.returncode != 0:
diff --git a/tools/browser_lightpanda.py b/tools/browser_lightpanda.py
index 67cb9e585e..0cdcc31a17 100644
--- a/tools/browser_lightpanda.py
+++ b/tools/browser_lightpanda.py
@@ -117,7 +117,7 @@ def _binary_supports_http_cache(binary: str) -> bool:
     try:
         proc = subprocess.run(
             [binary, "help"],
-            capture_output=True, text=True, timeout=3.0,
+            capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=3.0,
             stdin=subprocess.DEVNULL,
         )
         return _HTTP_CACHE_FLAG in ((proc.stdout or "") + (proc.stderr or ""))
diff --git a/tools/computer_use/cua_backend_daemon.py b/tools/computer_use/cua_backend_daemon.py
index 71dd46ec47..756cfd553a 100644
--- a/tools/computer_use/cua_backend_daemon.py
+++ b/tools/computer_use/cua_backend_daemon.py
@@ -169,7 +169,8 @@ class _EmbeddedCuaDaemon:
         env = self._sanitized_env()
         command = _embedded_daemon_spawn_command(self._command, self._serve_args(), platform=sys.platform)
         self._process = subprocess.Popen(command, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
-                                         stderr=subprocess.PIPE, text=True, env=env)
+                                         stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace",
+                                         env=env)
         self._owns_runtime = True
         threading.Thread(target=self._drain_stderr, args=(self._process,), name="hermes-cua-daemon-stderr", daemon=True).start()
         deadline = time.monotonic() + self._START_TIMEOUT_SECONDS
diff --git a/tools/self_repo_guard.py b/tools/self_repo_guard.py
index 0b03fe80f2..f0fb5aa169 100644
--- a/tools/self_repo_guard.py
+++ b/tools/self_repo_guard.py
@@ -446,7 +446,7 @@ def _read_git_alias(executable: str, target: Path, alias: str) -> str | None:
     with contextlib.suppress(OSError, subprocess.SubprocessError):
         result = subprocess.run(
             [executable, "-C", str(target), "config", "--get", f"alias.{alias}"],
-            capture_output=True, text=True, timeout=1, check=False)
+            capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=1, check=False)
         return (result.stdout.strip() or None) if result.returncode == 0 else None
     return None
 
diff --git a/tools/terminal_tool.py b/tools/terminal_tool.py
index 80cfaf9fd0..efb397fb9e 100644
--- a/tools/terminal_tool.py
+++ b/tools/terminal_tool.py
@@ -430,6 +430,30 @@ def _docker_session_isolation_enabled() -> bool:
     return _session_scope().docker_session_isolated
 
 
+def _routed_home_task_key(profile_scoped: bool) -> Optional[str]:
+    """Key for a session-less task serving a routed (non-launch) profile home, else None.
+
+    A multiplexed host runs every profile's cron jobs without a session key; collapsing them all onto
+    ``"default"`` made profile B's cron tool calls reuse the environment the launch profile's job
+    created (its ``.env`` residue, its bridged ``TERMINAL_*``, its shell), so B ran with A's settings.
+    Persistent Docker keys the profile name exactly like B's session-bound work, so B keeps ONE
+    container instead of a second one per home path.
+    """
+    from hermes_constants import get_hermes_home_override, profile_name_for_home
+    from tools.environments.local import _is_routed_home
+
+    override = get_hermes_home_override()
+    if not override or not _is_routed_home(override):
+        return None
+    profile = profile_name_for_home(override) if profile_scoped else None
+    if profile:
+        return "default" if profile == "default" else f"profile:{profile}"
+    try:
+        return f"home:{os.path.realpath(override)}"
+    except OSError:
+        return f"home:{override}"
+
+
 def _resolve_container_task_id(task_id: Optional[str]) -> str:
     """Map a tool-call ``task_id`` to the ``_active_environments`` key. Order matters —
     earlier branches are authoritative where they apply:
@@ -445,9 +469,10 @@ def _resolve_container_task_id(task_id: Optional[str]) -> str:
        default-profile gateway sessions share ONE container; other backends key
        ``session:<key>`` so switching profiles can't reuse another profile's
        SSHEnvironment on the wrong host.
-    4. No session key (CLI): ``shared:<key>`` when opted in (else a CLI run of a
-       keyed profile would split from its gateway sessions), else ``"default"``,
-       which subagent ids collapse onto to share the parent's container.
+    4. No session key (CLI, cron): ``shared:<key>`` when opted in (else a CLI run of a
+       keyed profile would split from its gateway sessions); a routed multiplexed profile
+       keys its own home (``profile:<name>`` under persistent Docker, matching branch 3);
+       else ``"default"``, which subagent ids collapse onto to share the parent's container.
     """
     if task_id and _has_isolation_overrides(task_id):
         return task_id
@@ -470,7 +495,7 @@ def _resolve_container_task_id(task_id: Optional[str]) -> str:
         # ONE container/cache slot (and sandbox dir) regardless of profile name (#84671).
         return f"shared:{shared}"
     if not session_key:
-        return "default"
+        return _routed_home_task_key(scope.docker_profile_scoped) or "default"
     if not scope.docker_profile_scoped:
         return f"session:{session_key}"
     profile = _current_session_profile() or "default"
diff --git a/tui_gateway/session_history.py b/tui_gateway/session_history.py
index af1954877b..b01a4129a3 100644
--- a/tui_gateway/session_history.py
+++ b/tui_gateway/session_history.py
@@ -179,6 +179,11 @@ _AUTO_CONTINUE_NOTE_PREFIX = "[System note: Your previous turn was interrupted m
 def _legacy_display_kind(role: str, text: str) -> str | None:
     """Display type of a synthetic row persisted untyped: new rows are typed at turn start (``persist_user_display_kind``);
     this prefix sniff migrates rows already on disk (a turn killed mid-run never reached the stamp)."""
+    # Imported functions are not rebound onto server.py (method_ctx.bind_module): import here.
+    from agent.turn_failure_copy import untyped_failed_turn_display_kind
+
+    if failed_turn := untyped_failed_turn_display_kind(role, text):
+        return failed_turn
     return "auto_continue" if role == "user" and text.lstrip().startswith(_AUTO_CONTINUE_NOTE_PREFIX) else None
 
 
diff --git a/website/docs/getting-started/updating.md b/website/docs/getting-started/updating.md
index 81d05a69af..9d8da85505 100644
--- a/website/docs/getting-started/updating.md
+++ b/website/docs/getting-started/updating.md
@@ -234,7 +234,7 @@ The same inventory is embedded in every real update's receipt (`~/.hermes/logs/u
 
 Every `hermes update` run writes a machine-readable receipt to `~/.hermes/logs/update_receipts/` (last 20 kept, `latest.json` always points at the most recent): the pre-update fleet plan, each step taken, anything skipped and why, the gateway restart outcome, and the final fleet version matrix. The SQLite runtime repair is one of those steps (`sqlite_runtime_repair`): a failed repair records the actual reason (for example the `uv sync` error) and the SQLite version pair, a deferred or not-applicable repair lands in the skips with its reason. After the restart phase the updater compares each live gateway's running code against the freshly updated checkout and prints a per-profile matrix — a gateway still serving pre-update code is reported loudly with the exact restart command, and the update exits non-zero so automation never treats a mixed-version fleet as healthy. Both `--plan` and the fleet check ask each running gateway directly over its local control socket (`gateway.sock` in the profile's data directory, a named pipe on Windows) when available, so version and supervisor information comes from the gateway itself; gateways from older versions are still discovered through their state files as before.
 
-A multiplexed default gateway is one process serving several profiles, so it appears once in the matrix and vouches for every profile in its `served_profiles` record. The same coverage clears the "A previous `hermes update` pulled new code but did not restart running gateways" hint: once that gateway (or, after a manual `git pull`, every gateway an update restarted) runs the current code, `hermes gateway restart` is enough — the hint no longer waits for the next `hermes update` to write a fresh receipt. The same is true of the restart obligation left by an update that died before recording which gateways it owed (or by an older updater that never recorded them): once every live gateway runs the current checkout, the obligation is retired and the hint stops. That obligation is recorded once per HOST, in the cross-profile rendezvous directory (`$HERMES_GATEWAY_LOCK_DIR`, else `$XDG_STATE_HOME/hermes/gateway-locks`) as `host-update-restart.json`, so every profile's CLI sees the same one: `hermes -p coder update` and `hermes -p writer update` restart the shared multiplexed gateway once between them, not once each. An obligation left behind by an older per-profile updater (`fleet_restart_pending` in one profile's Hermes home) is still read and cleared. An update whose pre-update plan found no gateway at all owes nothing and leaves no breadcrumb. A backend supervised by Desktop, systemd or launchd is restarted by its supervisor and never blocks this settlement; only a manual backend whose reminder could not be saved keeps the obligation open.
+A multiplexed default gateway is one process serving several profiles, so it appears once in the matrix and vouches for every profile in its `served_profiles` record. The same coverage clears the "A previous `hermes update` pulled new code but did not restart running gateways" hint: once that gateway (or, after a manual `git pull`, every gateway an update restarted) runs the current code, `hermes gateway restart` is enough — the hint no longer waits for the next `hermes update` to write a fresh receipt. The same is true of the restart obligation left by an update that died before recording which gateways it owed (or by an older updater that never recorded them): once every live gateway runs the current checkout, the obligation is retired and the hint stops. On a host that runs no gateway at all (the Desktop app alone), it is retired once no profile has a gateway that went away without a clean stop and every running backend is restarted by its own supervisor or has its own reminder. That obligation is recorded once per HOST, in the cross-profile rendezvous directory (`$HERMES_GATEWAY_LOCK_DIR`, else `$XDG_STATE_HOME/hermes/gateway-locks`) as `host-update-restart.json`, so every profile's CLI sees the same one: `hermes -p coder update` and `hermes -p writer update` restart the shared multiplexed gateway once between them, not once each. An obligation left behind by an older per-profile updater (`fleet_restart_pending` in one profile's Hermes home) is still read and cleared. An update whose pre-update plan found no gateway at all owes nothing and leaves no breadcrumb. A backend supervised by Desktop, systemd or launchd is restarted by its supervisor and never blocks this settlement; only a manual backend whose reminder could not be saved keeps the obligation open.
 
 ### Manual backend restart reminders