fix(storage): surface degraded session database state

apps/desktop/src/app/contrib/wiring.tsx

@@ -32,6 +32,7 @@ import {
 import { FloatingPet } from '@/components/pet/floating-pet'
 import { RemoteDisplayBanner } from '@/components/remote-display-banner'
 import { SendDiagnosticsHost } from '@/components/send-diagnostics-dialog'
+import { StorageDegradedBanner } from '@/components/storage-degraded-banner'
 import { TipHost } from '@/components/tips'
 import { emitGatewayEvent } from '@/contrib/events'
 import { getLatestSessionMessages } from '@/hermes'
@@ -1155,6 +1156,7 @@ export function ContribWiring({ children }: { children: ReactNode }) {
 
       {/* The full real overlay set (mirrors DesktopController's `overlays`). */}
       <RemoteDisplayBanner />
+      <StorageDegradedBanner />
       {!isAuxiliaryWindow() && <DesktopInstallOverlay />}
       {!isAuxiliaryWindow() && (
         <DesktopOnboardingOverlay

apps/desktop/src/app/shell/hooks/use-status-snapshot.test.ts

@@ -1,7 +1,10 @@
-import { act, cleanup, renderHook } from '@testing-library/react'
+import { act, cleanup, render, renderHook } from '@testing-library/react'
+import { createElement } from 'react'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
+import { StorageDegradedBanner } from '@/components/storage-degraded-banner'
 import { getStatus } from '@/hermes'
+import { $storageStatus } from '@/store/storage-status'
 
 import { deferred } from '../../../test/deferred'
 
@@ -22,6 +25,7 @@ async function flushAsync() {
 beforeEach(() => {
   vi.useFakeTimers()
   vi.spyOn(document, 'hasFocus').mockReturnValue(true)
+  $storageStatus.set('ok')
   vi.mocked(getStatus)
     .mockReset()
     .mockResolvedValue({} as never)
@@ -57,6 +61,21 @@ describe('useStatusSnapshot', () => {
     expect(requestGateway).toHaveBeenCalledTimes(2)
   })
 
+  it('publishes degraded storage and renders conservative recovery guidance', async () => {
+    vi.mocked(getStatus).mockResolvedValue({ storage: 'degraded' } as never)
+    const requestGateway = vi.fn(async () => ({ ok: true }) as never) as unknown as GatewayRequester
+
+    renderHook(() => useStatusSnapshot('open', requestGateway))
+    await flushAsync()
+
+    expect($storageStatus.get()).toBe('degraded')
+    const { container, getByText } = render(createElement(StorageDegradedBanner))
+    expect(getByText('Session database needs repair')).toBeTruthy()
+    expect(container.textContent).toContain(
+      'hermes sessions recover --source <state.db> --inspect-only'
+    )
+  })
+
   it('keeps the last authoritative readiness through a transient RPC failure', async () => {
     let refresh = 0
 

apps/desktop/src/app/shell/hooks/use-status-snapshot.ts

@@ -2,6 +2,7 @@ import { useEffect, useState } from 'react'
 
 import { getStatus } from '@/hermes'
 import { evaluateRuntimeReadiness, type RuntimeReadinessResult } from '@/lib/runtime-readiness'
+import { setStorageStatus } from '@/store/storage-status'
 import type { StatusResponse } from '@/types/hermes'
 
 // Statusbar health is ambient chrome, not live data — nothing the user acts on
@@ -68,6 +69,7 @@ export function useStatusSnapshot(
 
         if (statusResult.status === 'fulfilled') {
           setStatusSnapshot(statusResult.value)
+          setStorageStatus(statusResult.value.storage)
         }
 
         if (inferenceResult.status === 'fulfilled') {

apps/desktop/src/components/storage-degraded-banner.tsx

@@ -0,0 +1,32 @@
+import { useStore } from '@nanostores/react'
+
+import { AlertCircle } from '@/lib/icons'
+import { $storageStatus } from '@/store/storage-status'
+
+/** Persistent recovery guidance for a backend with degraded state.db health. */
+export function StorageDegradedBanner() {
+  const storageStatus = useStore($storageStatus)
+
+  if (storageStatus !== 'degraded') {
+    return null
+  }
+
+  return (
+    <div
+      className="pointer-events-none fixed inset-x-0 top-0 z-50 flex justify-center px-3 pt-2"
+      role="alert"
+    >
+      <div className="flex max-w-2xl items-start gap-2 rounded-md border border-destructive/45 bg-destructive/12 px-3 py-2 text-sm shadow-lg">
+        <AlertCircle aria-hidden className="mt-0.5 size-4 shrink-0 text-destructive" />
+        <div>
+          <p className="font-medium">Session database needs repair</p>
+          <p className="text-muted-foreground">
+            Storage is operating in a degraded mode. For structural corruption, stop profile writers and preserve the
+            {' '}database, then run <code>hermes sessions recover --source &lt;state.db&gt; --inspect-only</code> or
+            {' '}restore a snapshot. Use <code>hermes sessions repair</code> only when that repair path is safe.
+          </p>
+        </div>
+      </div>
+    </div>
+  )
+}

apps/desktop/src/store/storage-status.ts

@@ -0,0 +1,12 @@
+import { atom } from 'nanostores'
+
+export type StorageStatus = 'degraded' | 'ok'
+
+// The backend latches this state for its lifetime after confirmed corruption.
+// Keep the renderer's copy outside a particular sidebar/chat surface so the
+// warning remains visible while users navigate the app.
+export const $storageStatus = atom<StorageStatus>('ok')
+
+export function setStorageStatus(status: StorageStatus | undefined): void {
+  $storageStatus.set(status === 'degraded' ? 'degraded' : 'ok')
+}

apps/desktop/src/types/hermes.ts

@@ -469,6 +469,8 @@ export interface PaginatedSessions {
   /** Per-profile read failures from the cross-profile aggregator (e.g. a locked
    *  or corrupt state.db). Present only on `/api/profiles/sessions`. */
   errors?: Array<{ profile: string; error: string }>
+  /** The selected profile's backend-visible persistence state. */
+  storage?: 'degraded' | 'ok'
 }
 
 export interface RpcEvent<T = unknown> {
@@ -1257,6 +1259,8 @@ export interface StatusResponse {
   hermes_home: string
   latest_config_version: number
   release_date: string
+  /** Latched when the backend observes state.db corruption and pauses writes. */
+  storage?: 'degraded' | 'ok'
   version: string
 }
 

hermes_cli/web_routers/sessions.py

@@ -192,7 +192,15 @@ def get_sessions(
                 s["pinned"] = bool(s.get("pinned"))
             if not full:
                 _strip_session_list_rows(sessions)
-            return {"sessions": sessions, "total": total, "limit": limit, "offset": offset}
+            from hermes_state import get_storage_status
+
+            return {
+                "sessions": sessions,
+                "total": total,
+                "limit": limit,
+                "offset": offset,
+                "storage": get_storage_status(db.db_path),
+            }
         finally:
             db.close()
     except HTTPException:

hermes_cli/web_server.py

@@ -4098,6 +4098,9 @@ def _bounded_health_probe():
         # other detail that could carry secrets. The storage probe reuses the
         # gateway readiness state_db check (read-only, 1s-bounded) in an
         # executor so a wedged DB can't stall the event loop.
+        from hermes_state import get_storage_status
+
+        storage_status = get_storage_status(get_hermes_home() / "state.db")
         components: Dict[str, Any] = {
             "gateway": {
                 "status": "ok" if gateway_running and gateway_state in {"running", "draining"} else "degraded",
@@ -4109,9 +4112,16 @@ def _bounded_health_probe():
             from gateway.readiness import _probe_state_db
 
             storage_check = await run_in_threadpool(_probe_state_db, get_hermes_home())
-            components["storage"] = {"status": storage_check.get("status", "degraded")}
+            components["storage"] = {
+                "status": (
+                    "degraded"
+                    if storage_status == "degraded"
+                    else storage_check.get("status", "degraded")
+                )
+            }
         except Exception:
             components["storage"] = {"status": "degraded"}
+        status["storage"] = storage_status
         platform_states = [
             str(value.get("state") or value.get("status") or "").lower()
             for value in gateway_platforms.values()
@@ -12615,7 +12625,13 @@ def _open_session_db_at_path(db_path: Path, *, read_only: bool):
     """
     import sqlite3
 
-    from hermes_state import SessionDB, is_malformed_schema_error
+    from hermes_state import (
+        SessionDB,
+        get_storage_status,
+        is_malformed_db_error,
+        is_malformed_schema_error,
+        mark_storage_degraded,
+    )
 
     if not read_only:
         return SessionDB(db_path=db_path, read_only=False)
@@ -12659,20 +12675,34 @@ def _open_probed():
             # so route it through the same dispatch as malformed schema.
             is_malformed_schema_error(exc) or isinstance(exc, UnicodeDecodeError)
         ):
+            if is_malformed_db_error(exc):
+                mark_storage_degraded(db_path, exc)
+            raise
+        try:
+            SessionDB(db_path=db_path, read_only=False).close()
+        except (sqlite3.DatabaseError, UnicodeDecodeError) as heal_exc:
+            if is_malformed_db_error(heal_exc):
+                mark_storage_degraded(db_path, heal_exc)
             raise
-        SessionDB(db_path=db_path, read_only=False).close()
         try:
             return _open_probed()
         except (sqlite3.DatabaseError, UnicodeDecodeError) as still_stale:
             message = str(still_stale).lower()
             if "no such table" not in message and "no such column" not in message:
+                if is_malformed_db_error(still_stale):
+                    mark_storage_degraded(db_path, still_stale)
                 raise
             # The writable open succeeded but the store is STILL behind the
             # probe: reconciliation cannot fix this one. Serve reads without
             # the probe (queries touching the broken part will still fail,
             # everything else works) and stop paying the writable init per
             # poll.
             _session_db_heal_exhausted.add(str(db_path))
+            if get_storage_status(db_path) != "degraded":
+                mark_storage_degraded(
+                    db_path,
+                    RuntimeError("session database schema recovery was exhausted"),
+                )
             if str(db_path) not in _session_db_heal_warned:
                 _session_db_heal_warned.add(str(db_path))
                 _log.warning(

hermes_state.py

@@ -2104,6 +2104,65 @@ def apply_database_pragmas(
 _repair_attempted_paths: set[str] = set()
 _repair_attempt_lock = threading.Lock()
 
+# A corruption report must outlive the particular SessionDB instance that saw
+# it.  Long-running backends create short-lived SessionDBs for sidebar polls,
+# while the gateway keeps another instance for writes; a process-wide latch
+# makes both report the same safe, explicit state.  It intentionally resets on
+# process restart so a completed offline repair can be adopted without a hidden
+# persistent marker in a database that may itself be unreadable.
+# A present entry means storage is visibly degraded. Its boolean value answers
+# the separate safety question: whether canonical writes must be paused.
+_storage_status_by_path: Dict[str, bool] = {}
+_storage_status_lock = threading.Lock()
+
+
+def _storage_status_key(db_path: Path) -> str:
+    """Return a stable process-local key without requiring the path to exist."""
+    return str(Path(db_path).resolve(strict=False))
+
+
+def get_storage_status(db_path: Path) -> str:
+    """Return ``ok`` or the latched ``degraded`` state for *db_path*."""
+    with _storage_status_lock:
+        return (
+            "degraded"
+            if _storage_status_key(db_path) in _storage_status_by_path
+            else "ok"
+        )
+
+
+def _storage_writes_paused(db_path: Path) -> bool:
+    """Whether an unrecoverable error has paused writes for *db_path*."""
+    with _storage_status_lock:
+        return _storage_status_by_path.get(_storage_status_key(db_path), False)
+
+
+def mark_storage_degraded(
+    db_path: Path, exc: BaseException, *, pause_writes: bool = True
+) -> None:
+    """Expose corruption while pausing writes only when canonical rows are unsafe."""
+    key = _storage_status_key(db_path)
+    with _storage_status_lock:
+        already_degraded = key in _storage_status_by_path
+        writes_were_paused = _storage_status_by_path.get(key, False)
+        _storage_status_by_path[key] = writes_were_paused or pause_writes
+    if not already_degraded:
+        logger.error(
+            "state.db at %s entered storage_degraded%s: %s",
+            db_path,
+            "; persistence is paused until the profile is repaired"
+            if pause_writes
+            else "; canonical writes remain available while FTS is recovered",
+            exc,
+        )
+    elif pause_writes and not writes_were_paused:
+        logger.error(
+            "state.db at %s can no longer safely persist canonical rows; "
+            "persistence is paused until the profile is repaired: %s",
+            db_path,
+            exc,
+        )
+
 
 def is_malformed_db_error(exc: BaseException) -> bool:
     """True for explicit malformed-schema or generic corrupt-image errors.
@@ -4429,6 +4488,17 @@ class StateDbCorruptError(sqlite3.DatabaseError):
     """
 
 
+class StorageDegradedError(StateDbCorruptError):
+    """A peer refused writes after this process quarantined its state.db.
+
+    This is the process-wide counterpart to a handle's
+    :class:`StateDbCorruptError`: the fresh peer did not touch SQLite itself,
+    but persistence is equally terminal until recovery.  Subclassing retains
+    the established transcript-diversion and persistence-classification
+    contract for both refusal paths.
+    """
+
+
 _STATE_DB_CORRUPT_MSG = (
     "FATAL: state.db reported structural corruption (database disk image is "
     "malformed outside the FTS shadow tables) on a live handle; refusing further "
@@ -5865,7 +5935,17 @@ def _read_ctx(self) -> Iterator[sqlite3.Connection]:
         conn = self._checkout_read_conn()
         if conn is not None:
             try:
-                yield conn
+                try:
+                    yield conn
+                except sqlite3.DatabaseError as exc:
+                    if is_malformed_db_error(exc):
+                        # A read-only observer should still surface the
+                        # recovery guidance. Do not pause canonical writes:
+                        # a MATCH read can self-heal its derived FTS index.
+                        mark_storage_degraded(
+                            self.db_path, exc, pause_writes=False
+                        )
+                    raise
             finally:
                 returned = False
                 with self._read_conns_lock:
@@ -5892,7 +5972,14 @@ def _read_ctx(self) -> Iterator[sqlite3.Connection]:
                 # close() ran while a reader was still unwinding (#94736
                 # class) — reopen instead of yielding None to a .execute.
                 self._reopen_after_close_locked(context="read")
-            yield cast(sqlite3.Connection, self._conn)
+            try:
+                yield cast(sqlite3.Connection, self._conn)
+            except sqlite3.DatabaseError as exc:
+                if is_malformed_db_error(exc):
+                    # A read-only observer should still surface the recovery
+                    # guidance without blocking canonical writes.
+                    mark_storage_degraded(self.db_path, exc, pause_writes=False)
+                raise
 
     def _reopen_after_close_locked(self, context: str = "write") -> None:
         """Reopen the writer connection after ``close()`` raced a live caller.
@@ -5981,7 +6068,6 @@ def _reopen_after_close_locked(self, context: str = "write") -> None:
         # cannot have lost it, so no _init_schema here (no DDL races with
         # sibling processes during teardown).
         self._conn = conn
-
     # ── Core write helper ──
 
     @staticmethod
@@ -6310,6 +6396,11 @@ def _is_no_more_rows(exc: sqlite3.Error) -> bool:
 
         while True:
             self._raise_if_db_corrupt()
+            if _storage_writes_paused(self.db_path):
+                raise StorageDegradedError(
+                    "session database is degraded after a corruption error; "
+                    "persistence is paused until the profile is repaired"
+                )
             self._raise_if_db_replaced()
             fn_started = False
             try:
@@ -6411,11 +6502,16 @@ def _is_no_more_rows(exc: sqlite3.Error) -> bool:
                 # then retry the canonical write. The existing stale-open and
                 # explicit repair paths retain rebuild ownership.
                 if self._enter_fts_fail_open(exc):
+                    # FTS is derived from canonical messages. Its fail-open
+                    # mode is degraded and must be surfaced, but deliberately
+                    # keeps canonical persistence available.
+                    mark_storage_degraded(self.db_path, exc, pause_writes=False)
                     continue
                 # Bare SQLITE_CORRUPT / NOTADB that survived the replaced-file
                 # check and the FTS-scoped fail-open is structural damage:
                 # quarantine the handle (see StateDbCorruptError).
                 if self._is_structural_corruption_error(exc):
+                    mark_storage_degraded(self.db_path, exc)
                     self._halt_db_corrupt(exc)
                 raise
             except sqlite3.Error as exc: