532b6d1cb1
format_session_db_unavailable consumes the now-pinned cause table but its own two fallbacks (no recorded cause; network-drive gloss) still printed a bare `hermes doctor`. Same class; one test covers both fallbacks. Also: spell the corrupt action as plain concatenation instead of an f-string with escaped braces, and fix the finalizer comment to state the invariant (final_response is never rebound) rather than describe a local that no longer exists.
103 lines
4.1 KiB
Python
103 lines
4.1 KiB
Python
"""Plain-language copy for "session storage is unavailable / could not be written" notices.
|
|
|
|
One table keyed by ``classify_persistence_error``'s cause bucket feeds every surface (CLI banner,
|
|
gateway home-channel warning, TUI/Desktop RPC errors) so they agree on what happened, what to do,
|
|
and a machine-readable ``code`` a GUI can attach a "Run doctor" button to.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from hermes_state_errors import classify_persistence_error, is_disk_full_error
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class StorageFailure:
|
|
cause: str # classify_persistence_error bucket
|
|
code: str # machine-readable, stable: storage_locked | storage_readonly | storage_corrupt | disk_full | ...
|
|
gloss: str # what happened, one clause, lowercase start
|
|
action: str # what to do, one sentence naming the exact command
|
|
|
|
|
|
_DOCTOR = "Run `hermes {profile_arg}doctor --fix` to diagnose and repair."
|
|
|
|
# cause -> (code, gloss, action). "disk" is split by is_disk_full_error at lookup time.
|
|
_STORAGE_FAILURES: dict[str, tuple[str, str, str]] = {
|
|
"locked": (
|
|
"storage_locked",
|
|
"the session database is locked by another Hermes process",
|
|
"Wait a moment and try again; if it persists, stop the other Hermes process "
|
|
"(`hermes {profile_arg}gateway stop`).",
|
|
),
|
|
"disk_full": (
|
|
"disk_full",
|
|
"the disk is full",
|
|
"Free some disk space, then try again.",
|
|
),
|
|
"disk": (
|
|
"storage_readonly",
|
|
"the session database file is read-only or not writable",
|
|
_DOCTOR,
|
|
),
|
|
"corrupt": (
|
|
"storage_corrupt",
|
|
"the session database file is damaged",
|
|
_DOCTOR + " Recovery: `hermes {profile_arg}sessions recover --source <state.db> --inspect-only`.",
|
|
),
|
|
"fts_index": (
|
|
"storage_index_corrupt",
|
|
"the session search index is damaged (the messages themselves are intact)",
|
|
"Run `hermes {profile_arg}doctor --fix` (or `hermes {profile_arg}sessions repair`) to rebuild it.",
|
|
),
|
|
"replaced": (
|
|
"storage_replaced",
|
|
"the session database file was replaced while Hermes was running",
|
|
"Stop Hermes (`hermes {profile_arg}gateway stop`), run `hermes {profile_arg}doctor`, then start it again.",
|
|
),
|
|
"deleted_wal": (
|
|
"storage_replaced",
|
|
"the session database file was changed or replaced while Hermes was running",
|
|
"Stop Hermes (`hermes {profile_arg}gateway stop`), run `hermes {profile_arg}doctor`, then start it again.",
|
|
),
|
|
"compression": (
|
|
"storage_busy",
|
|
"another process is compressing this session",
|
|
"Send your message again once compression finishes.",
|
|
),
|
|
"compression_closed": (
|
|
"storage_session_rotated",
|
|
"this session was rotated by context compression",
|
|
"Refresh the client (or start a new turn) and send your message again.",
|
|
),
|
|
"turn_lease": (
|
|
"storage_busy",
|
|
"another Hermes process took over this session",
|
|
"Wait for it to finish, then send your message again.",
|
|
),
|
|
"unknown": (
|
|
"storage_unavailable",
|
|
"the session database could not be opened",
|
|
_DOCTOR,
|
|
),
|
|
}
|
|
|
|
|
|
def describe_storage_failure(exc_or_str) -> StorageFailure:
|
|
"""Plain-language description of a persistence failure (never raises)."""
|
|
cause = classify_persistence_error(exc_or_str)
|
|
key = "disk_full" if cause == "disk" and is_disk_full_error(exc_or_str) else cause
|
|
code, gloss, action = _STORAGE_FAILURES.get(key, _STORAGE_FAILURES["unknown"])
|
|
# Pin the copy-pasteable command to the failing profile — see profile_cli_selector.
|
|
from hermes_constants import profile_cli_selector
|
|
|
|
return StorageFailure(
|
|
cause=cause, code=code, gloss=gloss, action=action.replace("{profile_arg}", profile_cli_selector())
|
|
)
|
|
|
|
|
|
def storage_failure_details(exc_or_str, limit: int = 200) -> str:
|
|
"""Raw cause for a trailing, secondary "Details:" line (never the lead sentence)."""
|
|
text = " ".join(str(exc_or_str or "").split())
|
|
return text if len(text) <= limit else text[: limit - 3].rstrip() + "..."
|