refactor(approval): one session /yolo contract for CLI, TUI/Desktop and the messaging gateway

The persist / restore / carry logic for session YOLO had grown three copies
(the CLI mixin, tui_gateway, the gateway runner), each with its own key choice,
ordering and error handling. tools/approval_yolo.py now owns it:

- toggle_session_yolo: flip or set, persist BEFORE the live flip (a racing
  restore never revives an OFF), stored-copy-aware so a toggle after a restart
  turns it OFF; a failed write logs a warning and never blocks the flip.
- restore_session_yolo: re-arm a persisted flag in a fresh process; skipped
  under the frozen process --yolo.
- transfer_session_yolo: move the live flag to a continuation id.
- with_session_yolo: a row being created records a live flag (lazy first row,
  compression child, /branch).

Each surface keeps only its storage: the session row for CLI/TUI/Desktop, the
routing entry for the gateway. Removed: HermesCLI._transfer_session_yolo /
_persist_session_yolo, GatewayRunner._restore_session_yolo, the TUI-only
restore helper, and the TUI's post-publish set_session_yolo carry (the
compression publish now writes the child row with the flag).

Closes a gap the unification exposed: a CLI manual /compress that rotated the
session dropped YOLO (the next dangerous command prompted, and a --resume of
the continuation came back with it off).

agent/conversation_compression.py

@@ -3328,22 +3328,22 @@ def _publish_rotated_compaction(
         _foreign_tail_ceiling = agent._session_db.get_active_message_watermark(agent.session_id)
     with contextlib.suppress(Exception):  # best-effort — don't block compression on a flush error
         agent._flush_messages_to_session_db(messages, conversation_history=persisted_history)
-    # Publish closure + child + handoff in one transaction so no reader sees an
-    # empty child. Child stays on the parent's profile ("default" persists as NULL);
-    # publish also COALESCEs from the parent row for threads lacking HERMES_HOME.
+    # Publish closure + child + handoff in one transaction so no reader sees an empty child. Child stays on the
+    # parent's profile ("default" persists as NULL) and keeps a live /yolo; publish COALESCEs from the parent row.
     _profile_for_child = None
     with contextlib.suppress(Exception):
         from hermes_cli.profiles import get_active_profile_name
         _profile_for_child = get_active_profile_name()
-    if _profile_for_child == "default":
-        _profile_for_child = None
+    _profile_for_child = None if _profile_for_child == "default" else _profile_for_child
     old_title = agent._session_db.get_session_title(agent.session_id)
     new_session_id = mint_session_id()
     from agent.context_compressor import _DB_PERSISTED_MARKER
+    from tools.approval_yolo import with_session_yolo
     agent._session_db.publish_compression_child(
         parent_session_id=old_session_id, child_session_id=new_session_id,
         source=_compression_child_source(agent, old_session_id), model=agent.model,
-        model_config=agent._session_init_model_config, system_prompt=new_system_prompt, messages=compressed,
+        model_config=with_session_yolo(agent._session_init_model_config, old_session_id),
+        system_prompt=new_system_prompt, messages=compressed,
         cwd=getattr(agent, "working_directory", None), profile_name=_profile_for_child,
         compression_lock_holder=lease.holder, require_compression_lease=lease.holder is not None,
         require_lease_refresh=lease.holder is not None, lease_ttl_seconds=lease.ttl,

gateway/run_agent_cache.py

@@ -442,17 +442,6 @@ class GatewayAgentCacheMixin:
                 # Never fails the boundary itself, but a surviving ON would come back after a restart.
                 logger.warning("Failed to clear persisted yolo for session boundary %s", session_key, exc_info=True)
 
-    @staticmethod
-    def _restore_session_yolo(session_key: str, session_entry) -> None:
-        """Re-arm a persisted /yolo bypass after a gateway restart (the in-memory approval set starts
-        empty). Restore-only: ``_handle_yolo_command`` is the one writer and persists before it flips
-        the live flag, so a turn never sees a stale ON that it would wrongly revive."""
-        if not session_key or getattr(session_entry, "yolo", False) is not True:
-            return
-        from tools.approval import enable_session_yolo, is_session_yolo_enabled
-        if not is_session_yolo_enabled(session_key):
-            enable_session_yolo(session_key)
-
     def _begin_session_run_generation(self, session_key: str) -> int:
         """Claim a fresh, monotonically increasing run generation token (NEVER reset): a late result
         from a worker /stop or /new invalidated is recognized and dropped."""

gateway/run_turn.py

@@ -2049,13 +2049,13 @@ class GatewayTurnMixin:
         ``(_PreparedTurn, env_tokens)``; a ``str`` first element is a reply to send instead of
         running (history unreadable); ``None`` drops the turn (inbound text rejected)."""
         from gateway.run import _load_gateway_config
+        from tools.approval_yolo import restore_session_yolo
         _was_auto_reset, _is_new_session = await self._hmwa_open_session(session_entry, session_key, source)
-        self._restore_session_yolo(session_key, session_entry)
+        restore_session_yolo(session_key, session_entry.yolo is True)  # a restarted gateway's set starts empty
         context = build_session_context(source, self.config, session_entry)
         # Session context variables for tools (task-local, concurrency-safe)
         _session_env_tokens = self._set_session_env(context)
-        # Self-injected turns (MessageEvent(internal=True)) persist with a DB-only display_kind so
-        # UIs render timeline notices, not user bubbles; role/content untouched.
+        # Self-injected turns (internal=True) persist with a DB-only display_kind: timeline notices, not user bubbles.
         persist_user_display_kind = display_kind_for_event(event)
         _redact_pii = False  # privacy.redact_pii, re-read per message
         with suppress(Exception):

gateway/slash_commands.py

@@ -937,19 +937,15 @@ class GatewaySlashCommandsMixin(
     async def _handle_yolo_command(self, event: MessageEvent) -> Union[str, EphemeralReply]:
         """Handle /yolo — toggle dangerous command approval bypass for this session only. The flag is
         persisted on the routing entry so it survives a gateway restart."""
-        from tools.approval import disable_session_yolo, enable_session_yolo, is_session_yolo_enabled
+        from tools.approval_yolo import toggle_session_yolo
         session_key = self._session_key_for_source(event.source)
-        if self.session_store is not None:
+        persisted, persist = False, None
+        if (store := self.session_store) is not None:
             # A first-message /yolo has no routing entry yet; materialize it so the flag has a home.
-            entry = await self.async_session_store.get_or_create_session(event.source)
-            # After a restart only the persisted copy is set: the user still sees it ON.
-            enable = not (entry.yolo is True or is_session_yolo_enabled(session_key))
-            # Persist BEFORE flipping the live flag: a turn restoring in between then never revives a
-            # bypass that is being switched off.
-            await self.async_session_store.set_session_yolo(session_key, enable)
-        else:
-            enable = not is_session_yolo_enabled(session_key)
-        (enable_session_yolo if enable else disable_session_yolo)(session_key)
+            persisted = (await self.async_session_store.get_or_create_session(event.source)).yolo is True
+            persist = lambda on: store.set_session_yolo(session_key, on)  # noqa: E731
+        # Off the loop: the persist is a routing-store write (thread-safe, like AsyncSessionStore's calls).
+        enable = await asyncio.to_thread(toggle_session_yolo, session_key, persisted=persisted, persist=persist)
         return EphemeralReply(t("gateway.yolo.enabled" if enable else "gateway.yolo.disabled"))
 
     async def _handle_verbose_command(self, event: MessageEvent) -> str:

hermes_cli/cli_chat_turn_mixin.py

@@ -519,7 +519,8 @@ class CLIChatTurnMixin:
         # titling and the exit summary target the live child, not the ended parent.
         if (self.agent and getattr(self.agent, "session_id", None)
                 and self.agent.session_id != self.session_id):
-            self._transfer_session_yolo(self.session_id, self.agent.session_id)
+            from tools.approval_yolo import transfer_session_yolo
+            transfer_session_yolo(self.session_id, self.agent.session_id)
             self.session_id = self.agent.session_id
             self._write_terminal_breadcrumb()
             self._pending_title = None

hermes_cli/cli_commands_mixin.py

@@ -1103,21 +1103,21 @@ class CLICommandsMixin(CLICommandsSessionToolsMixin):
         parent_session_id = self.session_id
         # Create the child BEFORE ending the parent: a failed create_session must leave the session the
         # user is still on open, not ended with end_reason="branched" and no branch (#11030).
-        # The stable ``_branched_from`` marker keeps the branch visible in /resume + /sessions
-        # even after the parent is re-ended with a different end_reason.
-        # The child sends the parent's exact system prompt: a row without one makes the branch's first
-        # turn rebuild (re-probing the workspace), so the warm cache the copied transcript buys is
-        # lost at byte 0 whenever the repo moved since the parent's session start.
+        # The stable ``_branched_from`` marker keeps the branch visible in /resume + /sessions even after the
+        # parent is re-ended with a different end_reason; with_session_yolo keeps a live /yolo on the branch.
+        # The child sends the parent's exact system prompt: a row without one makes the branch's first turn rebuild
+        # (re-probing the workspace), losing the copied transcript's warm cache whenever the repo moved since.
         parent_prompt = getattr(self.agent, "_cached_system_prompt", None)
         if not isinstance(parent_prompt, str) or not parent_prompt:
             with suppress(Exception):
                 parent_prompt = (self._session_db.get_session(parent_session_id) or {}).get("system_prompt")
+        from tools.approval_yolo import transfer_session_yolo, with_session_yolo
         try:
             self._session_db.create_session(
                 session_id=new_session_id, source=os.environ.get("HERMES_SESSION_SOURCE", "cli"),
                 model=self.model, parent_session_id=parent_session_id, system_prompt=parent_prompt or None,
-                model_config={"max_iterations": self.max_turns, "reasoning_config": self.reasoning_config,
-                              "_branched_from": parent_session_id})
+                model_config=with_session_yolo({"max_iterations": self.max_turns, "reasoning_config":
+                                                self.reasoning_config, "_branched_from": parent_session_id}, parent_session_id))
         except Exception as e:
             return _cp(f"  {_gt('branch.create_failed', error=e)}")
         _end_current_session(self, "branched")
@@ -1132,7 +1132,7 @@ class CLICommandsMixin(CLICommandsSessionToolsMixin):
         with suppress(Exception):
             self._session_db.set_session_title(new_session_id, branch_title)
         # Switch to the new session
-        self._transfer_session_yolo(self.session_id, new_session_id)
+        transfer_session_yolo(self.session_id, new_session_id)
         self.session_id, self.session_start, self._pending_title = new_session_id, now, None
         self._resumed = True  # Prevents auto-title generation
         _sync_process_session_id(new_session_id)

hermes_cli/cli_session_mixin.py

@@ -167,22 +167,11 @@ class CLISessionMixin:
         _dim_notice(self, msg, quiet)
 
     def _restore_session_yolo(self, session_meta: dict, *, quiet: bool = False) -> None:
-        """Re-enable YOLO bypass on resume when the session row's ``model_config.yolo_mode``
-        says so — the in-memory ``tools.approval._session_yolo`` set starts empty in a fresh
-        process. No-op when already active or when the process was launched with ``--yolo``."""
-        try:
-            from hermes_state import SessionDB
-            from tools.approval import (
-                _YOLO_MODE_FROZEN, enable_session_yolo, is_session_yolo_enabled)
-        except Exception:
-            return
-        if _YOLO_MODE_FROZEN or not SessionDB.session_yolo_enabled(session_meta):
-            return
-        session_key = self.session_id or "default"
-        if is_session_yolo_enabled(session_key):
-            return
-        enable_session_yolo(session_key)
-        _dim_notice(self, t("cli.session.yolo_restored"), quiet)
+        """Re-arm the session row's persisted /yolo on every resume path (``tools.approval_yolo``)."""
+        from hermes_state import SessionDB
+        from tools.approval_yolo import restore_session_yolo
+        if restore_session_yolo(self.session_id or "default", SessionDB.session_yolo_enabled(session_meta)):
+            _dim_notice(self, t("cli.session.yolo_restored"), quiet)
 
     def _render_resume_history_panel_lines(self, panel) -> list[str]:
         """Render the resume panel at the current terminal width for resize replay."""
@@ -852,26 +841,6 @@ class CLISessionMixin:
 
             write_breadcrumb(self.session_id)
 
-    def _transfer_session_yolo(self, old_session_id: str, new_session_id: str) -> None:
-        """Move YOLO bypass state to a new session key when ``self.session_id`` is reassigned
-        mid-run (/branch, auto-compression rotation) — ``_session_yolo`` is keyed by id, so
-        without this the toggle silently reverts. Mirrors tui_gateway's rename path."""
-        if not old_session_id or not new_session_id or old_session_id == new_session_id:
-            return
-        try:
-            from tools.approval import (
-                disable_session_yolo, enable_session_yolo, is_session_yolo_enabled)
-        except Exception:
-            return
-        if is_session_yolo_enabled(old_session_id):
-            enable_session_yolo(new_session_id)
-            disable_session_yolo(old_session_id)
-            # Carry the persisted flag onto the continuation row so a later --resume restores
-            # it too. getattr: tests call this unbound against a minimal stand-in.
-            _persist = getattr(self, "_persist_session_yolo", None)
-            if _persist:
-                _persist(new_session_id, True)
-
     def _is_session_yolo_active(self) -> bool:
         """Whether YOLO bypass is on for this session: reads ``tools.approval._session_yolo``
         (not a stale env var) and honors the frozen process-start ``--yolo`` flag."""
@@ -896,8 +865,7 @@ class CLISessionMixin:
         """
         from cli import _cprint
         from hermes_cli.colors import Colors as _Colors
-        from tools.approval import (
-            _YOLO_MODE_FROZEN, disable_session_yolo, enable_session_yolo, is_session_yolo_enabled)
+        from tools.approval import _YOLO_MODE_FROZEN
 
         # A frozen process-level bypass short-circuits the approval gate ahead of the session
         # check — toggling "OFF" would be a false safety claim. Say so instead.
@@ -906,31 +874,17 @@ class CLISessionMixin:
             _cprint(f"  {t('cli.session.yolo_locked', state=state)}")
             return
 
+        from tools.approval_yolo import toggle_session_yolo
         session_key = self.session_id or "default"
-        # getattr: tests call this unbound against a minimal stand-in; persistence is best-effort.
-        _persist = getattr(self, "_persist_session_yolo", None)
-        if is_session_yolo_enabled(session_key):
-            disable_session_yolo(session_key)
-            if _persist:
-                _persist(session_key, False)
-            state = f"{_Colors.BOLD}{_Colors.RED}{t('cli.shared.label_off_upper')}{_Colors.RESET}"
-            _cprint(f"  {t('cli.session.yolo_off', state=state)}")
-        else:
-            enable_session_yolo(session_key)
-            if _persist:
-                _persist(session_key, True)
+        db = getattr(self, "_session_db", None)  # getattr: tests call this unbound against a stand-in
+        # The row is created lazily on the first turn; until then with_session_yolo records the toggle.
+        persist = (lambda on: db.set_session_yolo(session_key, on)) if db and session_key != "default" else None
+        if toggle_session_yolo(session_key, persist=persist):
             state = f"{_Colors.BOLD}{_Colors.GREEN}{t('cli.shared.label_on_upper')}{_Colors.RESET}"
             _cprint(f"  {t('cli.session.yolo_on', state=state)}")
-
-    def _persist_session_yolo(self, session_key: str, enabled: bool) -> None:
-        """Persist the YOLO flag to the session row so --resume restores it. Best-effort; the
-        in-memory toggle is authoritative. Skipped without a store or before the row exists
-        (rows are created lazily on the first turn)."""
-        db = getattr(self, "_session_db", None)
-        if db is None or not session_key or session_key == "default":
-            return
-        with contextlib.suppress(Exception):
-            db.set_session_yolo(session_key, enabled)
+        else:
+            state = f"{_Colors.BOLD}{_Colors.RED}{t('cli.shared.label_off_upper')}{_Colors.RESET}"
+            _cprint(f"  {t('cli.session.yolo_off', state=state)}")
 
     def _manual_compress(self, cmd_original: str = ""):
         """Manually trigger context compression.
@@ -985,6 +939,8 @@ class CLISessionMixin:
                 # generation point at the live continuation, not the ended parent.
                 agent_sid = getattr(self.agent, "session_id", None)
                 if agent_sid and agent_sid != self.session_id:
+                    from tools.approval_yolo import transfer_session_yolo
+                    transfer_session_yolo(self.session_id, agent_sid)
                     self.session_id = self.agent.session_id
                     self._write_terminal_breadcrumb()
                     self._pending_title = None

run_agent.py

@@ -305,20 +305,10 @@ class AIAgent(
             return None
 
     def _session_row_model_config(self) -> Any:
-        """``model_config`` for the session row: the init config plus the live YOLO bypass.
-
-        The row is created lazily on the first turn, so this is the only chance to record a pre-first-turn
-        /yolo toggle for ``hermes --resume``.
-        """
-        model_config = self._session_init_model_config
-        try:
-            from tools.approval import is_session_yolo_enabled
-            if is_session_yolo_enabled(self.session_id):
-                model_config = dict(model_config or {})
-                model_config["yolo_mode"] = True
-        except Exception:
-            pass
-        return model_config
+        """``model_config`` for the lazily created session row: the init config plus a live /yolo toggled
+        before the first turn (``hermes --resume`` restores it)."""
+        from tools.approval_yolo import with_session_yolo
+        return with_session_yolo(self._session_init_model_config, self.session_id)
 
     def _ensure_db_session(self) -> None:
         """Create the session DB row on first use; a transient failure leaves it to retry next turn."""

tests/gateway/test_yolo_command.py

@@ -66,6 +66,7 @@ async def test_yolo_survives_gateway_restart_and_dies_at_session_boundary(tmp_pa
     conversation boundary (/new, /resume) clears both copies so a restart cannot revive it."""
     from gateway.config import GatewayConfig
     from gateway.session import SessionStore
+    from tools.approval_yolo import restore_session_yolo
 
     def _runner():
         runner = _make_runner()
@@ -81,11 +82,11 @@ async def test_yolo_survives_gateway_restart_and_dies_at_session_boundary(tmp_pa
 
     disable_session_yolo(key)  # a new process starts with an empty in-memory approval set
     second = _runner()
-    second._restore_session_yolo(key, second.session_store.get_or_create_session(event.source))
+    restore_session_yolo(key, second.session_store.get_or_create_session(event.source).yolo)
     assert is_session_yolo_enabled(key) is True
 
     second._clear_session_boundary_security_state(key)
     assert is_session_yolo_enabled(key) is False
     third = _runner()
-    third._restore_session_yolo(key, third.session_store.get_or_create_session(event.source))
+    restore_session_yolo(key, third.session_store.get_or_create_session(event.source).yolo)
     assert is_session_yolo_enabled(key) is False

tests/hermes_cli/test_cli_yolo_resume_persistence.py

@@ -156,11 +156,6 @@ class TestToggleYoloPersists:
 
     def test_toggle_survives_missing_session_db(self):
         stand_in = SimpleNamespace(session_id=SESSION_ID, _session_db=None)
-        stand_in._persist_session_yolo = (
-            lambda key, enabled: HermesCLI._persist_session_yolo(
-                stand_in, key, enabled
-            )
-        )
         with patch("cli._cprint"):
             HermesCLI._toggle_yolo(stand_in)  # must not raise
         assert approval_module.is_session_yolo_enabled(SESSION_ID) is True
@@ -175,11 +170,6 @@ class TestEndToEndPersistAndRestore:
 
         # Process 1: user toggles /yolo ON — persisted to the row.
         cli_one = SimpleNamespace(session_id=SESSION_ID, _session_db=db)
-        cli_one._persist_session_yolo = (
-            lambda key, enabled: HermesCLI._persist_session_yolo(
-                cli_one, key, enabled
-            )
-        )
         with patch("cli._cprint"):
             HermesCLI._toggle_yolo(cli_one)
         assert approval_module.is_session_yolo_enabled(SESSION_ID) is True
@@ -213,11 +203,6 @@ class TestEndToEndPersistAndRestore:
             model_config={"yolo_mode": True},
         )
         cli_one = SimpleNamespace(session_id=SESSION_ID, _session_db=db)
-        cli_one._persist_session_yolo = (
-            lambda key, enabled: HermesCLI._persist_session_yolo(
-                cli_one, key, enabled
-            )
-        )
         approval_module.enable_session_yolo(SESSION_ID)
         with patch("cli._cprint"):
             HermesCLI._toggle_yolo(cli_one)  # OFF

tests/hermes_cli/test_cli_yolo_toggle.py

@@ -29,6 +29,7 @@ from unittest.mock import patch
 import pytest
 
 import tools.approval as approval_module
+from tools.approval_yolo import transfer_session_yolo
 from tools import approval_context
 from cli import HermesCLI
 
@@ -180,15 +181,14 @@ class TestSessionRotationTransfersYolo:
     compression continuation), YOLO state keyed under the old id must move
     to the new id. Otherwise the user's ``/yolo ON`` silently reverts on
     the next turn — the same UX failure mode this PR set out to fix.
-    Mirrors ``tui_gateway/server.py`` ~line 1297-1305."""
+    Shared with the TUI rotation path via ``tools.approval_yolo``."""
 
     def test_transfer_moves_yolo_to_new_session(self):
-        stand_in = _make_stand_in(session_id="old-id")
         try:
             approval_module.enable_session_yolo("old-id")
             assert approval_module.is_session_yolo_enabled("old-id") is True
 
-            HermesCLI._transfer_session_yolo(stand_in, "old-id", "new-id")
+            transfer_session_yolo("old-id", "new-id")
 
             assert approval_module.is_session_yolo_enabled("new-id") is True
             assert approval_module.is_session_yolo_enabled("old-id") is False
@@ -199,11 +199,10 @@ class TestSessionRotationTransfersYolo:
 
 
     def test_transfer_handles_empty_inputs_safely(self):
-        stand_in = _make_stand_in(session_id="x")
         # Both directions of empty input should be safe no-ops; nothing
         # to transfer from "" / to "".
-        HermesCLI._transfer_session_yolo(stand_in, "", "new")
-        HermesCLI._transfer_session_yolo(stand_in, "old", "")
+        transfer_session_yolo("", "new")
+        transfer_session_yolo("old", "")
         # Neither key should have been touched.
         assert approval_module.is_session_yolo_enabled("new") is False
         assert approval_module.is_session_yolo_enabled("old") is False

tests/hermes_cli/test_session_switch_mid_turn_guard.py

@@ -38,7 +38,7 @@ def cli(tmp_path, monkeypatch):
     obj.reasoning_config, obj._pending_title, obj._resumed, obj._pending_resume_sessions = {}, None, False, None
     obj.agent = _Agent("parent")
     obj.conversation_history = [{"role": "user", "content": "hi"}]
-    for name in ("_transfer_session_yolo", "_restore_session_cwd", "_restore_session_yolo",
+    for name in ("_restore_session_cwd", "_restore_session_yolo",
                  "_restore_session_model", "_display_resumed_history"):
         setattr(obj, name, lambda *a, **k: None)
     db.create_session(session_id="parent", source="cli", model="m")

tests/tools/test_approval_yolo.py

@@ -0,0 +1,74 @@
+"""The session /yolo contract every surface shares (``tools.approval_yolo``): a toggle persists before it flips
+the live flag, a fresh process re-arms it from the stored copy, a row created after the toggle records it, and a
+compression rotation publishes the continuation row with it (the CLI manual /compress dropped it before)."""
+
+import pytest
+
+from hermes_state import SessionDB
+from tools import approval
+from tools.approval_yolo import restore_session_yolo, toggle_session_yolo, transfer_session_yolo, with_session_yolo
+
+
+@pytest.fixture
+def db(tmp_path, monkeypatch):
+    monkeypatch.setattr(approval, "_YOLO_MODE_FROZEN", False)
+    database = SessionDB(db_path=tmp_path / "state.db")
+    yield database
+    for key in ("s1", "s2"):
+        approval.clear_session(key)
+    database.close()
+
+
+def test_toggle_restore_and_rotation_round_trip_through_the_row(db):
+    # Toggled before the row exists: the lazily created row records it.
+    assert toggle_session_yolo("s1", persist=lambda on: db.set_session_yolo("s1", on)) is True
+    db.create_session("s1", source="cli", model_config=with_session_yolo({"max_iterations": 5}, "s1"))
+    approval.clear_session("s1")  # a new process: the in-memory set starts empty
+    assert restore_session_yolo("s1", SessionDB.session_yolo_enabled(db.get_session("s1"))) is True
+    assert approval.is_session_yolo_enabled("s1")
+
+    # A rotation's continuation row is created carrying the live flag, which then moves to the new key.
+    db.create_session("s2", source="cli", model_config=with_session_yolo({}, "s1"), parent_session_id="s1")
+    transfer_session_yolo("s1", "s2")
+    assert approval.is_session_yolo_enabled("s2") and not approval.is_session_yolo_enabled("s1")
+    approval.clear_session("s2")
+    assert restore_session_yolo("s2", SessionDB.session_yolo_enabled(db.get_session("s2"))) is True
+
+    # After a restart only the stored copy is ON; a toggle switches it OFF and the OFF persists.
+    approval.clear_session("s2")
+    assert toggle_session_yolo("s2", persisted=True, persist=lambda on: db.set_session_yolo("s2", on)) is False
+    assert not SessionDB.session_yolo_enabled(db.get_session("s2"))
+    assert restore_session_yolo("s2", False) is False and not approval.is_session_yolo_enabled("s2")
+
+    # A failing write never blocks the live flip; a frozen process --yolo is never mirrored into the session set.
+    def broken(_on):
+        raise OSError("disk full")
+    assert toggle_session_yolo("s2", persist=broken) is True and approval.is_session_yolo_enabled("s2")
+    approval.clear_session("s2")
+    approval._YOLO_MODE_FROZEN = True
+    try:
+        assert restore_session_yolo("s2", True) is False and not approval.is_session_yolo_enabled("s2")
+    finally:
+        approval._YOLO_MODE_FROZEN = False
+
+
+def test_compression_rotation_publishes_the_child_row_with_the_live_flag(db):
+    import types
+    from agent import conversation_compression as cc
+    db.create_session("s1", source="cli")
+    db.append_message("s1", "user", "hello")
+    db.append_message("s1", "assistant", "world")
+    approval.enable_session_yolo("s1")
+    agent = types.SimpleNamespace(
+        _session_db=db, session_id="s1", platform="cli", model="m", _session_init_model_config={"max_iterations": 5},
+        working_directory=None, _memory_manager=None, context_compressor=types.SimpleNamespace(),
+        _flush_messages_to_session_db=lambda *a, **k: None, _persist_user_message_idx=None,
+        _session_messages=None, _gateway_session_key=None, _cached_system_prompt="sys")
+    cc._publish_rotated_compaction(
+        agent, [{"role": "user", "content": "hello"}, {"role": "assistant", "content": "world"}],
+        [{"role": "user", "content": "[handoff]"}],
+        new_system_prompt="sys", lease=types.SimpleNamespace(holder=None, ttl=60.0, watermark=None),
+        old_session_id="s1", compressed_user_turn_outcome="none")
+    assert agent.session_id != "s1"
+    assert SessionDB.session_yolo_enabled(db.get_session(agent.session_id))
+    approval.clear_session(agent.session_id)

tools/approval_yolo.py

@@ -0,0 +1,72 @@
+"""Session ``/yolo``: the one toggle / restore / carry contract every surface calls.
+
+The bypass itself is ``tools.approval``'s in-memory per-key set, so every process starts with it empty.
+Each surface persists it where its session identity lives: CLI, TUI and Desktop on the session row
+(``model_config.yolo_mode``), the messaging gateway on the routing entry, whose key outlives compression
+rotations. The surface hands its writer in as ``persist(enabled)``; everything else is decided here, so
+the CLI ``/yolo``, the TUI Shift+Tab, the Desktop zap and a Telegram ``/yolo`` behave the same.
+"""
+
+import logging
+from typing import Any, Callable, Optional
+
+from tools import approval
+
+logger = logging.getLogger(__name__)
+
+Persist = Optional[Callable[[bool], Any]]
+
+
+def _persist(persist: Persist, enabled: bool, session_key: str) -> None:
+    # Best-effort: the live flag is authoritative for this process; a failed write only costs the restore.
+    if persist is None:
+        return
+    try:
+        persist(enabled)
+    except Exception:
+        logger.warning("failed to persist session yolo=%s for %s", enabled, session_key, exc_info=True)
+
+
+def toggle_session_yolo(session_key: str, enabled: Optional[bool] = None, *, persisted: bool = False,
+                        persist: Persist = None) -> bool:
+    """Flip (``enabled=None``) or set the bypass for *session_key*; returns the new state.
+
+    *persisted* is the stored flag: after a restart only it is set and the user still sees ON, so a toggle
+    turns it OFF. The write runs BEFORE the live flip, so a turn restoring in between never revives a
+    bypass that is being switched off.
+    """
+    if enabled is None:
+        enabled = not (persisted or approval.is_session_yolo_enabled(session_key))
+    _persist(persist, enabled, session_key)
+    (approval.enable_session_yolo if enabled else approval.disable_session_yolo)(session_key)
+    return enabled
+
+
+def restore_session_yolo(session_key: str, persisted: bool) -> bool:
+    """Re-arm a persisted bypass in a fresh process; True when this call turned it on.
+
+    Key it on the id approvals are checked under (the stored session id / routing key), never a
+    transport-local id. Skipped under the frozen process ``--yolo``, which already bypasses everything.
+    """
+    if (not session_key or not persisted or approval._YOLO_MODE_FROZEN
+            or approval.is_session_yolo_enabled(session_key)):
+        return False
+    approval.enable_session_yolo(session_key)
+    return True
+
+
+def transfer_session_yolo(old_key: str, new_key: str) -> None:
+    """Move the live bypass when the conversation continues under a new id (compression rotation, /branch).
+    The new row records it at creation (``with_session_yolo``), so nothing is written here."""
+    if not old_key or not new_key or old_key == new_key or not approval.is_session_yolo_enabled(old_key):
+        return
+    approval.enable_session_yolo(new_key)
+    approval.disable_session_yolo(old_key)
+
+
+def with_session_yolo(model_config: Optional[dict], session_key: str) -> Optional[dict]:
+    """``model_config`` for a session row being created, carrying a live bypass. Rows are created lazily on
+    the first turn, so this is where a toggle made before the row existed gets recorded."""
+    if not approval.is_session_yolo_enabled(session_key):
+        return model_config
+    return {**(model_config or {}), "yolo_mode": True}

tui_gateway/methods_config_set.py

@@ -276,7 +276,6 @@ def _set_yolo(rid, params, key, value, session):
     # scope="session" (default; Shift+Tab) toggles ONLY this session's flag; scope="global"
     # (Shift+click the zap) flips persistent approvals.mode between "off" and "manual".
     scope = _word(params.get("scope") or "session")
-    from tools.approval import disable_session_yolo, enable_session_yolo, is_session_yolo_enabled
     raw = _word(value)
     if scope == "global":
         from tools.approval_context import _normalize_approval_mode
@@ -286,18 +285,15 @@ def _set_yolo(rid, params, key, value, session):
         _write_config_key("approvals.mode", "off" if enable else "manual")  # binary: no "smart" restore
         _emit_all_session_info()  # reflect the flip in every live indicator
     elif session:
-        skey = session["session_key"]
-        enable = _BOOL_WORDS.get(raw, not is_session_yolo_enabled(skey))
-        (enable_session_yolo if enable else disable_session_yolo)(skey)
-        # Persist like the CLI's /yolo so a resume in a new backend restores it. Row id prefers the agent's
-        # session_id: after compression the key can still name the ended parent (#20001).
-        row_id = getattr(session.get("agent"), "session_id", None) or skey
-        try:
+        from tools.approval_yolo import toggle_session_yolo
+        # Row id prefers the agent's session_id: after compression the key can still name the ended parent (#20001).
+        row_id = getattr(session.get("agent"), "session_id", None) or session["session_key"]
+
+        def persist(on):
             with _session_db(session) as db:
                 if db is not None:
-                    db.set_session_yolo(row_id, enable)
-        except Exception:
-            logger.warning("failed to persist session yolo flag for %s", row_id, exc_info=True)
+                    db.set_session_yolo(row_id, on)
+        enable = toggle_session_yolo(session["session_key"], _BOOL_WORDS.get(raw), persist=persist)
         _emit_session_info(params.get("session_id", ""), session)
     else:
         enable = _BOOL_WORDS.get(raw, not is_truthy_value(os.environ.get("HERMES_YOLO_MODE")))

tui_gateway/methods_session.py

@@ -132,7 +132,6 @@ def _session_row_summary(row: dict, *, tip_row: dict | None = None, resolved_id=
 
 from hermes_state_sessions import INTERNAL_LISTING_SOURCES
 
-from .methods_session_model_guard import restore_session_yolo as _restore_session_yolo
 
 # Hidden from human listings (kanban workers, tool integrations, one-shot runs); see INTERNAL_LISTING_SOURCES.
 _LISTING_DENY_SOURCES = frozenset(INTERNAL_LISTING_SOURCES)
@@ -1077,12 +1076,13 @@ def _(rid, params: dict) -> dict:
             live = _find_live_session_by_key(ctx.target, ctx.profile_home)
         if live is not None:
             return _resume_reuse_live(ctx, *live)
-        _restore_session_yolo(ctx.target, ctx.found)  # a new backend starts with an empty approval set
+        from hermes_state import SessionDB
+        from tools.approval_yolo import restore_session_yolo  # a fresh backend starts with an empty set
+        restore_session_yolo(ctx.target, SessionDB.session_yolo_enabled(ctx.found))
         if ctx.lazy:
             return _resume_lazy(ctx)
-        if ctx.eager_build:
-            return _resume_eager(ctx)
-        return _resume_deferred(ctx) if ctx.defer_history else _resume_cold(ctx)
+        return _resume_eager(ctx) if ctx.eager_build else (
+            _resume_deferred(ctx) if ctx.defer_history else _resume_cold(ctx))
     finally:
         # Refcounting alone does not release the sqlite fds: SessionDB pins ITSELF (atexit.register) once its
         # background token writer starts; only close() unregisters.

tui_gateway/methods_session_model_guard.py

@@ -1,5 +1,4 @@
-"""``session.create`` composer overrides, the model×provider coherence gate (#96817), and the
-per-session runtime state ``session.resume`` restores from the stored row.
+"""``session.create`` composer overrides and the model×provider coherence gate (#96817).
 
 A composer, script or older client can pin a model the selected provider cannot serve
 (``gpt-5.5`` on ``anthropic``); the session used to be minted fine and the FIRST turn died with
@@ -50,13 +49,3 @@ def create_overrides(params: dict) -> tuple:
     elif "fast" in params:
         service_tier_override = "priority" if is_truthy_value(params.get("fast")) else ""
     return model_override, reasoning_override, service_tier_override
-
-
-def restore_session_yolo(session_key: str, session_meta: dict | None) -> None:
-    """Re-arm a persisted session /yolo on ``session.resume``: a new backend process starts with an empty
-    in-memory approval set. Keyed on the stored id (``ctx.target``), the key approvals are checked under,
-    never the freshly minted runtime sid. Mirrors ``cli_session_mixin._restore_session_yolo``."""
-    from hermes_state import SessionDB
-    from tools.approval import enable_session_yolo
-    if session_key and SessionDB.session_yolo_enabled(session_meta):
-        enable_session_yolo(session_key)

tui_gateway/session_compression.py

@@ -334,14 +334,8 @@ def _sync_session_key_after_compress(
         from tools import approval
         with contextlib.suppress(Exception):
             approval.unregister_gateway_notify(old_key)
-        with contextlib.suppress(Exception):
-            if approval.is_session_yolo_enabled(old_key):
-                approval.enable_session_yolo(new_session_id)
-                approval.disable_session_yolo(old_key)
-                # The toggle persisted the now-ended parent row; carry it onto the continuation (CLI parity).
-                with _session_db(session) as db:
-                    if db is not None:
-                        db.set_session_yolo(new_session_id, True)
+        from tools.approval_yolo import transfer_session_yolo
+        transfer_session_yolo(old_key, new_session_id)
         with contextlib.suppress(Exception):
             approval.register_gateway_notify(new_session_id, lambda data: _emit_approval_request(sid, data))
     # Invalidate any in-flight ``_drain_queued_prompt`` claim taken under the pre-rotation key: a raced

tui_gateway/session_workdir.py

@@ -475,11 +475,8 @@ def _workdir_row_model_config(session: dict) -> tuple[str, dict]:
             model_config[flag] = True
     if isinstance(composer_profile := session.get("composer_override_profile"), dict):
         model_config["composer_override_profile"] = composer_profile
-    # A /yolo toggled before the first message had no row to persist to (rows are created lazily here).
-    from tools.approval import is_session_yolo_enabled
-    if is_session_yolo_enabled(session.get("session_key") or ""):
-        model_config["yolo_mode"] = True
-    return row_model, model_config
+    from tools.approval_yolo import with_session_yolo  # a /yolo toggled before the row existed
+    return row_model, with_session_yolo(model_config, session.get("session_key") or "")
 
 
 def _ensure_session_db_row(session: dict) -> bool: