fix(gateway,desktop): keep compression busy refusals at debug, share the busy mapping, check the reason on Desktop

A session_busy refusal that names the compression lock now logs at debug like any other busy refusal, so Desktop's busy retry during an edit cannot flood the gateway log. The MUTATION_GUARD_REFUSALS-to-session_busy mapping moves to busy_refusal() beside the tuple. Desktop's isCompressionBusyError also requires reason session_busy, as the TUI's check does.

apps/desktop/src/app/session/hooks/use-prompt-actions/utils.ts

@@ -300,7 +300,9 @@ export function isSessionBusyError(error: unknown): boolean {
 
 /** A busy refusal because another writer holds the compression lock (`field: 'compression'`). */
 export function isCompressionBusyError(error: unknown): boolean {
-  return error instanceof JsonRpcGatewayError && (error.data as { field?: unknown } | undefined)?.field === 'compression'
+  const data = error instanceof JsonRpcGatewayError ? (error.data as { field?: unknown; reason?: unknown } | undefined) : null
+
+  return data?.reason === 'session_busy' && data.field === 'compression'
 }
 
 // prompt.submit refused because another surface (TUI, messaging gateway)

gateway/session_controls.py

@@ -58,7 +58,7 @@ def _log_refusal(profile, method, exc: RuntimeStoreError) -> None:
     args = (method, _profile_label(profile), exc.reason,
             f' field={exc.field}' if getattr(exc, 'field', None) else '',
             f' detail={exc.detail}' if getattr(exc, 'detail', None) else '')
-    if getattr(exc, 'field', None) or getattr(exc, 'detail', None):
+    if exc.reason != 'session_busy' and (getattr(exc, 'field', None) or getattr(exc, 'detail', None)):
         logger.warning(line, *args)
     else:
         logger.debug(line, *args)
@@ -719,16 +719,13 @@ class AuthorityConnection:
 
     async def mutate(self, ref, params):
         from gateway.session_mutations import mutate_session
-        from hermes_state_mutation_guards import MUTATION_GUARD_REFUSALS
+        from hermes_state_mutation_guards import MUTATION_GUARD_REFUSALS, busy_refusal
         try:
             result = await mutate_session(self.authority, self.actor, ref, params)
         except MUTATION_GUARD_REFUSALS as exc:
             # A live turn lease / compression lock is the same retryable busy verdict a
-            # running admission gets (REST maps it to 409), not an internal error. A held
-            # compression lock is named: Stop does not end it, so clients say to wait instead.
-            from hermes_state_errors import SessionCompressionInProgressError
-            compressing = isinstance(exc, SessionCompressionInProgressError)
-            raise RuntimeStoreError('session_busy', field='compression' if compressing else None) from exc
+            # running admission gets (REST maps it to 409), not an internal error.
+            raise busy_refusal(exc) from exc
         except RuntimeStoreError as exc:
             if exc.reason in {'revision_conflict', 'stale_generation'}:
                 # The fresh fences ride on the refusal, so the client retries once without a re-read.

hermes_state_mutation_guards.py

@@ -9,6 +9,13 @@ from hermes_state_runtime import RuntimeStoreError
 MUTATION_GUARD_REFUSALS = (SessionTurnLeaseLostError, SessionCompressionInProgressError)
 
 
+def busy_refusal(exc):
+    """The client-facing refusal for one of ``MUTATION_GUARD_REFUSALS``. A held compression lock
+    is named in ``field``: Stop does not end it, so clients say to wait instead."""
+    compressing = isinstance(exc, SessionCompressionInProgressError)
+    return RuntimeStoreError('session_busy', field='compression' if compressing else None)
+
+
 def require_not_executing(conn, session_ids):
     """Refuse while a turn is running (or its outcome is unknown) on any of ``session_ids``.
     Queued admissions are allowed: a follower waits on the logical owner and simply runs