review follow-ups: canonical classifier in hermes_state, compression-busy=locked, hedged gateway wording, drop dead constant

- Move classify_persistence_error into hermes_state beside is_disk_full_error
  and delegate the disk bucket to it (fixes 'ENOSPC writing state.db' and
  'not enough space' classifying as unknown). run_agent keeps a thin lazy
  delegating wrapper so the documented import path and fast import survive.
- Classify CompressionSessionBusyError (and its RPC-wrapped message forms)
  as 'locked': the motivating #81227 failure mode stringifies to 'is being
  compressed by another writer', which the substring heuristic missed.
- Export PERSISTENCE_ERROR_CAUSES and iterate it in the cron explainer
  suppression instead of a hardcoded tuple, so a future cause bucket cannot
  silently desynchronize cron delivery.
- Hedge the gateway locked/unknown recovery wording ('should already be
  saved' instead of 'was recorded') to match the explainer - the early
  turn-start persist may also have failed.
- Drop STATE_DB_WAL_WARN_BYTES (speculative dead constant with no consumer;
  the pre-existing 50 MB doctor WAL check covers the warning).
- Tests: compression-busy classification, is_disk_full_error delegation,
  causes-tuple coverage; mutation-checked red-green.

cron/scheduler.py

@@ -4255,7 +4255,11 @@ def run_job(
             # one-argument render would let cause-refined explainer text slip
             # through and be delivered as a cron warning.
             _explainer_variants = []
-            for _cause in (None, "locked", "disk", "unknown"):
+            try:
+                from hermes_state import PERSISTENCE_ERROR_CAUSES as _causes
+            except Exception:
+                _causes = ("locked", "disk", "unknown")
+            for _cause in (None, *_causes):
                 try:
                     _variant = AIAgent._format_turn_completion_explanation(
                         turn_exit_reason, _cause

gateway/run.py

@@ -3511,8 +3511,8 @@ def _normalize_empty_agent_response(
             return (
                 "⚠️ Session storage was temporarily unavailable, so this "
                 "turn was stopped to protect your conversation history. "
-                "Your message was recorded — please send it again in a "
-                "moment."
+                "Your message should already be saved — please send it "
+                "again in a moment."
             )
         is_context_failure = any(
             p in error_str

hermes_cli/doctor.py

@@ -219,7 +219,6 @@ def check_info(text: str):
 # ── state.db health/stats thresholds (advisory only — module constants,
 # deliberately NOT config: doctor warnings are guidance, not policy) ──
 STATE_DB_SIZE_WARN_BYTES = 1 * 1024 * 1024 * 1024   # 1 GiB logical size
-STATE_DB_WAL_WARN_BYTES = 256 * 1024 * 1024          # 256 MiB WAL
 
 
 def _human_bytes(n) -> str:
@@ -312,8 +311,7 @@ def _render_state_db_stats(stats: dict, holders=None) -> list:
     # WAL runaway is deliberately NOT warned here: the pre-existing WAL
     # check later in the state.db section already warns above 50 MB and
     # offers a checkpoint via --fix; a second warning at a higher threshold
-    # would only duplicate it. STATE_DB_WAL_WARN_BYTES remains for callers
-    # (dashboards) that consume the stats dict without that legacy check.
+    # would only duplicate it.
 
     return lines
 

hermes_state.py

@@ -1367,6 +1367,58 @@ def is_disk_full_error(exc: BaseException | str | None) -> bool:
     return any(marker in lowered for marker in _DISK_FULL_MARKERS)
 
 
+# Every cause bucket classify_persistence_error can return. Consumers that
+# enumerate causes (e.g. the cron scheduler's explainer-variant suppression)
+# must iterate this tuple instead of hardcoding the list, so adding a bucket
+# can never silently desynchronize them.
+PERSISTENCE_ERROR_CAUSES = ("locked", "disk", "unknown")
+
+
+def classify_persistence_error(exc_or_str) -> str:
+    """Classify a session-persistence failure into a coarse cause bucket.
+
+    Fast-failing a turn on a SessionDB write error is deliberate (the
+    transcript would otherwise be lost on restart), but the *guidance* the
+    user gets must match the cause: sustained SQLite write-lock contention
+    ("database is locked" on a shared state.db) needs "storage was busy,
+    send it again", while a full disk or read-only database needs the
+    disk-space/permissions advice. Returns one of PERSISTENCE_ERROR_CAUSES:
+
+    * ``"locked"``  — lock/busy contention (another process holds the write
+      lock, or a live compression lease refused the write); transient,
+      retry-later guidance applies.
+    * ``"disk"``    — disk full / read-only / permission-shaped failures
+      (delegates the disk-full patterns to :func:`is_disk_full_error` so the
+      two classifiers can never drift apart — e.g. ENOSPC).
+    * ``"unknown"`` — anything else (or no visible exception at all).
+    """
+    if exc_or_str is None:
+        return "unknown"
+    # A refused write during a live compression lease is contention, not
+    # storage damage — but its message ("is being compressed by another
+    # writer" / "Compression lease lost") contains neither "locked" nor
+    # "busy", so it must be matched by type and by phrase (for strings that
+    # survived RPC wrapping).
+    if isinstance(exc_or_str, CompressionSessionBusyError):
+        return "locked"
+    text = str(exc_or_str).lower()
+    if (
+        "locked" in text
+        or "busy" in text
+        or "being compressed" in text
+        or "compression lease" in text
+    ):
+        return "locked"
+    if (
+        is_disk_full_error(exc_or_str)
+        or "disk" in text
+        or "readonly" in text
+        or "read-only" in text
+    ):
+        return "disk"
+    return "unknown"
+
+
 def _claim_repair_attempt(db_path: Path) -> bool:
     """Claim the one-shot repair attempt for *db_path* in this process.
 

run_agent.py

@@ -282,32 +282,18 @@ _DB_PERSISTED_MARKER = "_db_persisted"
 def classify_persistence_error(exc_or_str) -> str:
     """Classify a session-persistence failure into a coarse cause bucket.
 
-    Fast-failing a turn on a SessionDB write error is deliberate (the
-    transcript would otherwise be lost on restart), but the *guidance* the
-    user gets must match the cause: sustained SQLite write-lock contention
-    ("database is locked" on a shared state.db) needs "storage was busy,
-    send it again", while a full disk or read-only database needs the
-    disk-space/permissions advice. Returns one of:
-
-    * ``"locked"``  — lock/busy contention (another process holds the write
-      lock); transient, retry-later guidance applies.
-    * ``"disk"``    — disk full / read-only / permission-shaped failures.
-    * ``"unknown"`` — anything else (or no visible exception at all).
+    Thin delegating wrapper: the canonical implementation lives in
+    ``hermes_state`` (beside ``is_disk_full_error``, whose disk-full
+    patterns it reuses so the two classifiers can never drift apart).
+    Kept importable from this module because the turn-finalizer contract
+    and existing callers reference ``run_agent.classify_persistence_error``.
+    The import stays lazy to preserve this module's fast import path —
+    every real caller already has hermes_state loaded (the error being
+    classified came from a SessionDB write).
     """
-    if exc_or_str is None:
-        return "unknown"
-    text = str(exc_or_str).lower()
-    if "locked" in text or "busy" in text:
-        return "locked"
-    if (
-        "disk" in text
-        or "readonly" in text
-        or "read-only" in text
-        or "no space" in text
-        or "database or disk is full" in text
-    ):
-        return "disk"
-    return "unknown"
+    from hermes_state import classify_persistence_error as _impl
+
+    return _impl(exc_or_str)
 
 
 # Guard so the OpenRouter metadata pre-warm thread is only spawned once per

tests/run_agent/test_turn_completion_explainer.py

@@ -177,6 +177,66 @@ def test_classify_persistence_error_categories():
     assert classify_persistence_error("") == "unknown"
 
 
+def test_classify_persistence_error_reuses_disk_full_markers():
+    """The disk bucket delegates to hermes_state.is_disk_full_error, so
+    every marker that helper recognizes (ENOSPC, 'not enough space', ...)
+    must classify as 'disk' — the two classifiers can never drift apart."""
+    import errno
+
+    from run_agent import classify_persistence_error
+
+    assert classify_persistence_error("ENOSPC writing state.db") == "disk"
+    assert classify_persistence_error(
+        "There is not enough space on the disk"
+    ) == "disk"
+    assert classify_persistence_error(
+        OSError(errno.ENOSPC, "No space left on device")
+    ) == "disk"
+
+
+def test_classify_persistence_error_compression_busy_is_locked():
+    """A live compression lease refusing the write is contention, not
+    storage damage — but its message contains neither 'locked' nor 'busy',
+    so it must classify by exception type (and by phrase for RPC-wrapped
+    strings). This is the exact failure mode of issue #81227."""
+    from hermes_state import (
+        CompressionSessionBusyError,
+        SessionCompressionInProgressError,
+    )
+    from run_agent import classify_persistence_error
+
+    assert classify_persistence_error(
+        SessionCompressionInProgressError(
+            "Session 'abc' is being compressed by another writer"
+        )
+    ) == "locked"
+    assert classify_persistence_error(
+        CompressionSessionBusyError("Compression lease lost before publication: abc")
+    ) == "locked"
+    # RPC-wrapped string forms (exception type lost in transit).
+    assert classify_persistence_error(
+        "Session 'abc' is being compressed by another writer"
+    ) == "locked"
+    assert classify_persistence_error(
+        "Compression lease lost before publication: abc"
+    ) == "locked"
+
+
+def test_persistence_error_causes_tuple_matches_classifier():
+    """PERSISTENCE_ERROR_CAUSES must cover every value the classifier can
+    return (consumers like cron suppression iterate it)."""
+    from hermes_state import PERSISTENCE_ERROR_CAUSES, classify_persistence_error
+
+    probes = (
+        "database is locked",
+        "database or disk is full",
+        "something else entirely",
+        None,
+    )
+    for probe in probes:
+        assert classify_persistence_error(probe) in PERSISTENCE_ERROR_CAUSES
+
+
 # --------------------------------------------------------------------------
 # 2. Enable/disable seam
 # --------------------------------------------------------------------------

tests/test_state_db_stats.py

@@ -212,10 +212,10 @@ def test_render_does_not_duplicate_legacy_wal_warning():
     """A large WAL must NOT warn here: doctor's pre-existing WAL check
     (50 MB threshold, with a --fix checkpoint) already covers it, and a
     second warning at a higher threshold would duplicate the output."""
-    from hermes_cli.doctor import STATE_DB_WAL_WARN_BYTES, _render_state_db_stats
+    from hermes_cli.doctor import _render_state_db_stats
 
     lines = _render_state_db_stats(
-        _base_stats(wal_size_bytes=STATE_DB_WAL_WARN_BYTES + 1), holders=None
+        _base_stats(wal_size_bytes=256 * 1024 * 1024 + 1), holders=None
     )
     warns = [line for line in lines if line[0] == "warn"]
     blob = " ".join(" ".join(str(p) for p in line) for line in warns).lower()