diff --git a/agent/activity_tracking.py b/agent/activity_tracking.py
new file mode 100644
index 0000000000..1ce2cf160a
--- /dev/null
+++ b/agent/activity_tracking.py
@@ -0,0 +1,149 @@
+"""Turn-liveness activity tracking for ``AIAgent`` (gateway watchdog + session activity persistence).
+
+``_touch_activity`` is the single write path; persistence is rate-limited and never raises.
+Extracted from ``run_agent.py``; every method resolves through ``AIAgent``'s MRO unchanged.
+"""
+import logging
+import os
+import threading
+import time
+from typing import Optional
+
+from agent.session_activity import ActivityProvenance
+
+# Same logger name as the origin module so log records / caplog filters are unchanged.
+logger = logging.getLogger("run_agent")
+
+
+class ActivityTrackingMixin:
+    """Liveness timestamps/labels and rate-limited session activity persistence."""
+
+    def _liveness_activity_lock(self) -> "threading.Lock":
+        """Shared lock for the activity clock and its generation counter.
+
+        ``_touch_activity`` stamps under it and the liveness watchdog samples/commits under it, so a stall
+        observation can never abort a turn that resumed in between. Lazy so ``__new__``-built doubles work.
+        """
+        _lock = getattr(self, "_turn_liveness_activity_lock", None)
+        if _lock is None:
+            _lock = threading.Lock()
+            self._turn_liveness_activity_lock = _lock
+        return _lock
+
+    def _touch_activity(
+        self,
+        desc: str,
+        *,
+        provenance: Optional[ActivityProvenance] = None,
+        force_persist: bool = False,
+    ) -> None:
+        """Update the last-activity timestamp and description (thread-safe).
+
+        Bumps a monotonic generation under ``_liveness_activity_lock`` so the watchdog can bind a stall
+        observation
+        to the exact ``(generation, timestamp)`` it sampled. Also bridges (rate-limited, best-effort) to the
+        kanban
+        heartbeat when this is a dispatcher-spawned worker, and to the durable SessionDB activity projection.
+        ``provenance`` names special writers (compression); ``force_persist`` bypasses the SessionDB rate
+        limit.
+        """
+        from agent.session_activity import (
+            bound_activity_description,
+            normalize_activity_provenance,
+            reset_session_activity_persist_window,
+        )
+
+        # Lazy per-instance lock, inline so SimpleNamespace doubles binding _touch_activity without the
+        # class keep working (tests/run_agent/test_session_activity_persist.py).
+        _clock_lock = getattr(self, "_turn_liveness_activity_lock", None)
+        if _clock_lock is None:
+            _clock_lock = threading.Lock()
+            self._turn_liveness_activity_lock = _clock_lock
+        with _clock_lock:
+            self._turn_liveness_activity_generation = (
+                getattr(self, "_turn_liveness_activity_generation", 0) + 1
+            )
+            self._last_activity_ts = time.time()
+            self._last_activity_desc = bound_activity_description(desc)
+            self._last_activity_provenance = normalize_activity_provenance(provenance)
+            # Real progress invalidates a reserved abort claim; an in-flight watchdog interrupt must abandon
+            # itself at the final mutation edge.
+            self._turn_liveness_abort_claim = None
+        if os.environ.get("HERMES_KANBAN_TASK"):
+            try:
+                from tools.kanban_tools import (
+                    heartbeat_current_worker_from_env,
+                    inject_new_comments_from_env,
+                )
+                heartbeat_current_worker_from_env()
+                # Fold any new operator notes into the running turn (OUT-OF-BAND
+                # steer) so the user can talk to a live task without a restart.
+                inject_new_comments_from_env(self)
+            except Exception:
+                # Never let the bridge break the loop; this guard covers import-time failures.
+                pass
+        if force_persist:
+            reset_session_activity_persist_window(self)
+        self._persist_session_activity_if_due()
+
+    def _persist_session_activity_if_due(self) -> None:
+        """Best-effort durable activity heartbeat for SessionDB consumers.
+
+        Cadence pinned by ``SESSION_ACTIVITY_HEARTBEAT_MIN_INTERVAL_SECONDS`` (config-independent). Fail-open:
+        a failed write never raises into the agent loop.
+        """
+        session_id = getattr(self, "session_id", None)
+        session_db = getattr(self, "_session_db", None)
+        if not session_id or session_db is None:
+            return
+        touch = getattr(session_db, "touch_session_activity", None)
+        if not callable(touch):
+            return
+        from agent.session_activity import (
+            SESSION_ACTIVITY_HEARTBEAT_MIN_INTERVAL_SECONDS,
+            normalize_activity_provenance,
+        )
+
+        now_mono = time.monotonic()
+        last_mono = getattr(self, "_session_activity_last_persist_mono", 0.0)
+        if (now_mono - last_mono) < SESSION_ACTIVITY_HEARTBEAT_MIN_INTERVAL_SECONDS:
+            return
+        self._session_activity_last_persist_mono = now_mono
+        try:
+            touch(
+                session_id,
+                getattr(self, "_last_activity_ts", None),
+                description=getattr(self, "_last_activity_desc", None),
+                provenance=normalize_activity_provenance(
+                    getattr(self, "_last_activity_provenance", None)
+                ),
+            )
+        except Exception:
+            # Heartbeat is observation-only; never let its I/O break the loop.
+            logger.debug(
+                "session activity heartbeat write failed (ignored)",
+                exc_info=True,
+            )
+
+    def _reset_activity_labels_after_turn(self) -> None:
+        """Drop mid-turn activity labels once the turn is no longer running.
+
+        Keeps ``_last_activity_ts`` so idle/watchdog clocks stay continuous across turns; clears description +
+        provenance so idle agents / SessionDB listings stop advertising the last mid-turn stamp.
+        """
+        from agent.session_activity import ActivityProvenance
+
+        self._last_activity_desc = ""
+        self._last_activity_provenance = ActivityProvenance.UNKNOWN
+        session_id = getattr(self, "session_id", None)
+        session_db = getattr(self, "_session_db", None)
+        if not session_id or session_db is None:
+            return
+        clear = getattr(session_db, "clear_session_activity_labels", None)
+        if not callable(clear):
+            return
+        try:
+            clear(session_id)
+        except Exception:
+            # Never let durable cleanup I/O break turn teardown.
+            pass
diff --git a/agent/interrupt_control.py b/agent/interrupt_control.py
new file mode 100644
index 0000000000..f852f0a2fe
--- /dev/null
+++ b/agent/interrupt_control.py
@@ -0,0 +1,433 @@
+"""Interrupt / steer / redirect control surface for ``AIAgent``.
+
+Soft/hard interrupt requests, tool-thread interrupt propagation, pending steer/redirect queues.
+Extracted from ``run_agent.py``; every method resolves through ``AIAgent``'s MRO unchanged.
+"""
+import logging
+import threading
+from typing import Optional
+
+from agent.interrupt_compat import request_hard_interrupt
+from tools.interrupt import set_interrupt as _set_interrupt
+
+# Same logger name as the origin module so log records / caplog filters are unchanged.
+logger = logging.getLogger("run_agent")
+
+
+class InterruptControlMixin:
+    """interrupt()/hard_interrupt()/clear_interrupt()/steer()/redirect() (see module docstring)."""
+
+    def interrupt(
+        self,
+        message: Optional[str] = None,
+        *,
+        hard_cancel: bool = False,
+        tool_reason: Optional[str] = None,
+        require_generation: Optional[int] = None,
+    ) -> bool:
+        """Request the agent to interrupt its current tool-calling loop (call from another thread).
+
+        ``message``: new message to include in the response context. ``hard_cancel``: explicit stop;
+        compression
+        may honor it even while ordinary interrupts are masked. ``tool_reason``: trusted fixed category safe
+        for
+        tool output. ``require_generation``: activity-generation claim — the interrupt is published only if
+        the
+        turn's generation still matches at the final mutation edge (claim reserved under the activity lock,
+        consumed together with the first observable publication); returns False if the turn resumed meanwhile.
+        """
+        if require_generation is not None:
+            # RESERVE the abort's generation claim under the SAME lock `_touch_activity` stamps with. Real
+            # progress invalidates it; it is CONSUMED at the final mutation edge, so a resumed turn abandons
+            # the abort.
+            with self._liveness_activity_lock():
+                if (
+                    getattr(self, "_turn_liveness_activity_generation", 0)
+                    != require_generation
+                ):
+                    return False
+                self._turn_liveness_abort_claim = require_generation
+
+        # A hard stop and redirect share one lock so /stop cannot race with an
+        # accepted correction and accidentally turn itself into a retry.
+        def _wait_for_compression_commit() -> None:
+            # Pre-claim half of hard-cancel admission (#99758 P1): wait out a commit that already crossed its
+            # boundary but mutate NOTHING — cancelling a pending fence is irreversible and must wait until the
+            # generation claim survived the final mutation edge (_cancel_pending_compression_commit).
+            fence = vars(self).get("_active_compression_commit_fence")
+            if fence is None:
+                return
+            if not getattr(fence, "commit_in_flight", False):
+                # No commit in flight — cancel_before_commit here WOULD cancel the pending commit; leave it to
+                # the
+                # destructive half.
+                return
+            cancel_before_commit = getattr(
+                type(fence), "cancel_before_commit", None
+            )
+            if callable(cancel_before_commit):
+                try:
+                    # A commit holds the fence lock through finish_commit: this blocks until it finishes and
+                    # returns
+                    # False WITHOUT setting _cancelled.
+                    cancel_before_commit(fence)
+                except Exception:
+                    logger.debug(
+                        "Compression hard-cancel fence wait failed",
+                        exc_info=True,
+                    )
+
+        def _cancel_pending_compression_commit() -> None:
+            # Destructive half of hard-cancel admission (#99758 P1): runs only AFTER the claim survived, so a
+            # declined abort never leaves the fence cancelled. A commit that started meanwhile owns the fence
+            # and completes on its own; only a still-pending commit is cancelled here.
+            fence = vars(self).get("_active_compression_commit_fence")
+            if fence is None:
+                return
+            if getattr(fence, "commit_in_flight", False):
+                return
+            cancel_before_commit = getattr(
+                type(fence), "cancel_before_commit", None
+            )
+            if callable(cancel_before_commit):
+                try:
+                    # Marks the fence cancelled (or waits out a just-started commit) without touching the
+                    # hard-stop
+                    # Event, which was published at the claim edge.
+                    cancel_before_commit(fence)
+                except Exception:
+                    logger.debug(
+                        "Compression hard-cancel fence admission failed",
+                        exc_info=True,
+                    )
+
+        def _publish_interrupt_state() -> None:
+            self._interrupt_requested = True
+            self._interrupt_message = message
+            self._tool_interrupt_reason = tool_interrupt_reason
+            if hard_cancel:
+                _hard_event = getattr(
+                    self, "_hard_interrupt_requested", None
+                )
+                if _hard_event is not None:
+                    _hard_event.set()
+
+        def _consume_claim_and_publish_first_state() -> bool:
+            # Final mutation edge: claim consumption and the FIRST observable interrupt publication are ONE
+            # activity-lock critical section, so either the claim survives and commits before any later
+            # activity stamp, or the stamp landed first and the abort declines without publishing.
+            if require_generation is None:
+                # No claim to race: publish WITHOUT the liveness lock. Bare AIAgent stand-ins in other suites
+                # lack
+                # the liveness seam and would AttributeError.
+                _publish_interrupt_state()
+                return True
+            with self._liveness_activity_lock():
+                if (
+                    getattr(self, "_turn_liveness_abort_claim", None)
+                    != require_generation
+                ):
+                    return False
+                self._turn_liveness_abort_claim = None
+                _publish_interrupt_state()
+            return True
+
+        # Tool cancellation attribution stays separate from _interrupt_message, which may carry the user's
+        # full next message.
+        tool_interrupt_reason = (
+            (tool_reason or "explicit stop requested")
+            if hard_cancel
+            else ("user sent a new message" if message else "user interrupt")
+        )
+
+        _redirect_lock = getattr(self, "_pending_redirect_lock", None)
+        if _redirect_lock is not None:
+            with _redirect_lock:
+                # The blocking in-flight-commit wait runs BEFORE the atomic claim edge (redirect lock still
+                # held);
+                # the destructive pending-commit cancel runs AFTER the claim survives (#99758 P1).
+                if hard_cancel:
+                    _wait_for_compression_commit()
+                if not _consume_claim_and_publish_first_state():
+                    return False
+                if hard_cancel:
+                    _cancel_pending_compression_commit()
+                self._pending_redirect = None
+        else:
+            if hard_cancel:
+                _wait_for_compression_commit()
+            if not _consume_claim_and_publish_first_state():
+                return False
+            if hard_cancel:
+                _cancel_pending_compression_commit()
+            self._pending_redirect = None
+
+        # Codex app-server owns its model/tool loop and watches a private
+        # interrupt event rather than Hermes' per-thread flag.
+        if getattr(self, "api_mode", None) == "codex_app_server":
+            _codex_session = getattr(self, "_codex_session", None)
+            _request_interrupt = getattr(_codex_session, "request_interrupt", None)
+            if callable(_request_interrupt):
+                try:
+                    _request_interrupt()
+                except Exception:
+                    logger.debug(
+                        "Failed to interrupt Codex app-server turn",
+                        exc_info=True,
+                    )
+
+        # Cron turns request on the conversation thread (no nested interrupt-worker deadlock); their client
+        # is registered here so this cross-thread interrupt can still shut the sockets.
+        _abort_active_request = getattr(self, "_active_request_abort", None)
+        if callable(_abort_active_request):
+            try:
+                _abort_active_request("interrupt_abort")
+            except Exception:
+                logger.debug("Failed to abort active inline request", exc_info=True)
+        # Scope the tool interrupt to this agent's execution thread so other in-process agents are unaffected.
+        if self._execution_thread_id is not None:
+            _set_interrupt(
+                True,
+                self._execution_thread_id,
+                reason=tool_interrupt_reason,
+            )
+            self._interrupt_thread_signal_pending = False
+        else:
+            # Interrupt arrived before run_conversation bound the execution thread: defer the tool-level
+            # signal instead of targeting the caller thread.
+            self._interrupt_thread_signal_pending = True
+        # Fan out to concurrent-tool worker tids: is_interrupted() inside a tool only sees its own tid, so
+        # without this a hung concurrent tool runs to its own timeout. getattr covers __init__-less stubs.
+        _tracker = getattr(self, "_tool_worker_threads", None)
+        _tracker_lock = getattr(self, "_tool_worker_threads_lock", None)
+        if _tracker is not None and _tracker_lock is not None:
+            with _tracker_lock:
+                _worker_tids = list(_tracker)
+            for _wtid in _worker_tids:
+                try:
+                    _set_interrupt(True, _wtid, reason=tool_interrupt_reason)
+                except Exception:
+                    pass
+        # Propagate interrupt to any running child agents (subagent delegation)
+        with self._active_children_lock:
+            children_copy = list(self._active_children)
+        for child in children_copy:
+            try:
+                if hard_cancel:
+                    request_hard_interrupt(
+                        child,
+                        message,
+                        tool_reason=tool_interrupt_reason,
+                    )
+                else:
+                    child.interrupt(message)
+            except Exception as e:
+                logger.debug("Failed to propagate interrupt to child agent: %s", e)
+        if not self.quiet_mode:
+            print("\n⚡ Interrupt requested" + (f": '{message[:40]}...'" if message and len(message) > 40 else f": '{message}'" if message else ""))
+        return True
+
+    def hard_interrupt(
+        self,
+        message: Optional[str] = None,
+        *,
+        tool_reason: Optional[str] = None,
+    ) -> None:
+        """Request an explicit stop while preserving the ``interrupt()`` ABI.
+
+        Frontends feature-detect this and fall back to legacy ``interrupt()`` for third-party agents.
+        """
+        # Bypass dynamic dispatch: legacy subclasses may override interrupt(message=None) without hard_cancel.
+        InterruptControlMixin.interrupt(
+            self,
+            message,
+            hard_cancel=True,
+            tool_reason=tool_reason,
+        )
+
+    def clear_interrupt(self, *, preserve_redirect: bool = False) -> bool:
+        """Clear the interrupt request and per-thread tool signal.
+
+        ``preserve_redirect`` is only for the conversation loop rebuilding the same logical turn after
+        cancelling a model request; public hard-stop paths clear everything.
+        """
+        _redirect_lock = getattr(self, "_pending_redirect_lock", None)
+        if _redirect_lock is not None:
+            with _redirect_lock:
+                if preserve_redirect and not self._pending_redirect:
+                    return False
+                self._interrupt_requested = False
+                self._interrupt_message = None
+                self._tool_interrupt_reason = None
+                getattr(self, "_hard_interrupt_requested", threading.Event()).clear()
+                if not preserve_redirect:
+                    self._pending_redirect = None
+        else:
+            if preserve_redirect and not getattr(self, "_pending_redirect", None):
+                return False
+            self._interrupt_requested = False
+            self._interrupt_message = None
+            self._tool_interrupt_reason = None
+            getattr(self, "_hard_interrupt_requested", threading.Event()).clear()
+            if not preserve_redirect:
+                self._pending_redirect = None
+        self._interrupt_thread_signal_pending = False
+        if self._execution_thread_id is not None:
+            _set_interrupt(False, self._execution_thread_id)
+        # Also clear worker-thread bits so no stale interrupt survives a turn boundary onto a recycled tid.
+        # getattr covers __init__-less test stubs.
+        _tracker = getattr(self, "_tool_worker_threads", None)
+        _tracker_lock = getattr(self, "_tool_worker_threads_lock", None)
+        if _tracker is not None and _tracker_lock is not None:
+            with _tracker_lock:
+                _worker_tids = list(_tracker)
+            for _wtid in _worker_tids:
+                try:
+                    _set_interrupt(False, _wtid)
+                except Exception:
+                    pass
+        # A hard interrupt supersedes any pending /steer — its target iteration will no longer happen.
+        _steer_lock = getattr(self, "_pending_steer_lock", None)
+        if _steer_lock is not None:
+            with _steer_lock:
+                self._pending_steer = None
+        return True
+
+    def steer(self, text: str) -> bool:
+        """Inject user text into the next tool result without interrupting the current tool.
+
+        The text is appended to the LAST tool result once the batch finishes, so the model sees it on its next
+        iteration. Thread-safe; multiple calls concatenate with newlines. Returns False for empty text.
+        """
+        if not text or not text.strip():
+            return False
+        cleaned = text.strip()
+        _lock = getattr(self, "_pending_steer_lock", None)
+        if _lock is None:
+            # __init__-less test stubs: fall back to a direct attribute set.
+            existing = getattr(self, "_pending_steer", None)
+            self._pending_steer = (existing + "\n" + cleaned) if existing else cleaned
+            return True
+        with _lock:
+            if self._pending_steer:
+                self._pending_steer = self._pending_steer + "\n" + cleaned
+            else:
+                self._pending_steer = cleaned
+        return True
+
+    def redirect(self, text: str) -> bool:
+        """Redirect the active turn without converting it into a new task.
+
+        During a model request this cancels only that request: completed messages/tool results are kept, the
+        displayed partial reasoning becomes assistant context, the correction is appended as a real user
+        message,
+        and the loop retries. During tool execution it degrades to ``steer()``; Codex app-server uses native
+        ``turn/steer``. Returns False when there is no live turn or the text is empty.
+        """
+        if not text or not text.strip():
+            return False
+        cleaned = text.strip()
+
+        # Codex owns its internal reasoning/tool loop, so use its first-class
+        # active-turn steering protocol rather than interrupting the subprocess.
+        if getattr(self, "api_mode", None) == "codex_app_server":
+            _codex_session = getattr(self, "_codex_session", None)
+            _native_steer = getattr(_codex_session, "request_steer", None)
+            if callable(_native_steer):
+                _redirect_lock = getattr(self, "_pending_redirect_lock", None)
+                if _redirect_lock is not None:
+                    with _redirect_lock:
+                        if self._interrupt_requested:
+                            return False
+                elif self._interrupt_requested:
+                    return False
+                try:
+                    return bool(_native_steer(cleaned))
+                except Exception:
+                    logger.debug("Codex app-server turn/steer failed", exc_info=True)
+                    return False
+
+        # Never kill a tool to deliver guidance; the steer drain puts it on the final tool result.
+        if getattr(self, "_executing_tools", False):
+            return self.steer(cleaned)
+
+        _model_active = getattr(self, "_model_request_active", None)
+        _redirect_lock = getattr(self, "_pending_redirect_lock", None)
+        if _redirect_lock is None:
+            if _model_active is None or not _model_active.is_set():
+                return False
+            existing = getattr(self, "_pending_redirect", None)
+            if self._interrupt_requested and not existing:
+                return False
+            self._pending_redirect = (
+                f"{existing}\n\n[Additional user correction]\n{cleaned}"
+                if existing
+                else cleaned
+            )
+            self._interrupt_requested = True
+            self._interrupt_message = None
+        else:
+            with _redirect_lock:
+                if _model_active is None or not _model_active.is_set():
+                    # The response completed before we acquired the state lock.
+                    # Reject so the surface queues a new turn.
+                    return False
+                if self._interrupt_requested and not self._pending_redirect:
+                    return False
+                if self._pending_redirect:
+                    self._pending_redirect = (
+                        f"{self._pending_redirect}\n\n"
+                        f"[Additional user correction]\n{cleaned}"
+                    )
+                else:
+                    self._pending_redirect = cleaned
+                self._interrupt_requested = True
+                self._interrupt_message = None
+
+        # Interrupt only the model request. Do not fan out to tool workers or
+        # child agents as interrupt() does.
+        _execution_thread_id = getattr(self, "_execution_thread_id", None)
+        if _execution_thread_id is not None:
+            _set_interrupt(True, _execution_thread_id)
+            self._interrupt_thread_signal_pending = False
+        else:
+            self._interrupt_thread_signal_pending = True
+        _abort_active_request = getattr(self, "_active_request_abort", None)
+        if callable(_abort_active_request):
+            try:
+                _abort_active_request("redirect_abort")
+            except Exception:
+                logger.debug("Failed to abort request for redirect", exc_info=True)
+        return True
+
+    def _has_pending_redirect(self) -> bool:
+        """Return whether an active-turn redirect is waiting to be applied."""
+        _redirect_lock = getattr(self, "_pending_redirect_lock", None)
+        if _redirect_lock is None:
+            return bool(getattr(self, "_pending_redirect", None))
+        with _redirect_lock:
+            return bool(self._pending_redirect)
+
+    def _drain_pending_redirect(self) -> Optional[str]:
+        """Return and clear pending active-turn correction text."""
+        _redirect_lock = getattr(self, "_pending_redirect_lock", None)
+        if _redirect_lock is None:
+            text = getattr(self, "_pending_redirect", None)
+            self._pending_redirect = None
+            return text
+        with _redirect_lock:
+            text = self._pending_redirect
+            self._pending_redirect = None
+        return text
+
+    def _drain_pending_steer(self) -> Optional[str]:
+        """Return the pending steer text (if any) and clear the slot; None when nothing is pending."""
+        _lock = getattr(self, "_pending_steer_lock", None)
+        if _lock is None:
+            text = getattr(self, "_pending_steer", None)
+            self._pending_steer = None
+            return text
+        with _lock:
+            text = self._pending_steer
+            self._pending_steer = None
+        return text
diff --git a/agent/rate_limit_credits.py b/agent/rate_limit_credits.py
new file mode 100644
index 0000000000..6b3643530b
--- /dev/null
+++ b/agent/rate_limit_credits.py
@@ -0,0 +1,211 @@
+"""Rate-limit / credits header capture and low-credit notices for ``AIAgent``.
+
+Parses provider response headers into ``_rate_limit_state`` / ``_credits_state`` and emits sticky notices.
+Extracted from ``run_agent.py``; every method resolves through ``AIAgent``'s MRO unchanged.
+"""
+import logging
+import os
+from typing import Any
+
+from utils import is_truthy_value
+
+# Same logger name as the origin module so log records / caplog filters are unchanged.
+logger = logging.getLogger("run_agent")
+
+
+class RateLimitCreditsMixin:
+    """Rate-limit + credits header capture and notices (see module docstring)."""
+
+    def _capture_rate_limits(self, http_response: Any) -> None:
+        """Parse x-ratelimit-* headers from an HTTP response and cache the state.
+
+        Called after each streaming call; the httpx Response is available as ``stream.response``.
+        """
+        if http_response is None:
+            return
+        headers = getattr(http_response, "headers", None)
+        if not headers:
+            return
+        try:
+            from agent.rate_limit_tracker import parse_rate_limit_headers
+            state = parse_rate_limit_headers(headers, provider=self.provider)
+            if state is not None:
+                self._rate_limit_state = state
+        except Exception:
+            pass  # Never let header parsing break the agent loop
+
+    def get_rate_limit_state(self):
+        """Return the last captured RateLimitState, or None."""
+        return self._rate_limit_state
+
+    def _capture_anthropic_response_headers(self, http_response: Any) -> None:
+        """Capture out-of-band state from Anthropic Messages response headers.
+
+        The SDK's aggregated ``Message`` drops headers, where Portal puts rate-limit and credits state. Fail-
+        open.
+        """
+        self._capture_rate_limits(http_response)
+        self._capture_credits(http_response)
+
+    def _capture_credits(self, http_response: Any) -> None:
+        """Parse x-nous-credits-* headers, cache CreditsState, fire threshold notices.
+
+        The PARSE is swallowed (miss → keep last-known); the notice EVALUATION is a separate block that WARNS
+        on
+        failure so a depletion-notice bug cannot vanish silently.
+        """
+        # Dev test fixture (HERMES_DEV_CREDITS_FIXTURE): inject a chosen notice state
+        # each turn for repeatable testing, bypassing real headers. Throwaway scaffolding.
+        try:
+            from agent.credits_tracker import dev_fixture_credits_state
+            _fixture = dev_fixture_credits_state()
+        except Exception:
+            _fixture = None
+        if _fixture is not None:
+            self._credits_state = _fixture
+            if self._credits_session_start_micros is None:
+                self._credits_session_start_micros = _fixture.remaining_micros
+            _latch = getattr(self, "_credits_latch", None)
+            if isinstance(_latch, dict):
+                # Only seen_below_90 — priming seen_grant_unspent would fire grant_spent on first observation.
+                _latch["seen_below_90"] = True  # let warn90 fire without a real crossing
+            _used = _fixture.used_fraction
+            logger.info(
+                "credits ▸ [FIXTURE] remaining=%d (%s) · paid=%s · denom=%s · used=%s "
+                "(real headers bypassed — `echo clear` / unset HERMES_DEV_CREDITS_FIXTURE to restore)",
+                _fixture.remaining_micros,
+                _fixture.remaining_usd or "?",
+                _fixture.paid_access,
+                _fixture.denominator_kind,
+                ("%.0f%%" % (_used * 100)) if _used is not None else "n/a",
+            )
+            self._emit_credits_notices()
+            return
+        if http_response is None:
+            return
+        headers = getattr(http_response, "headers", None)
+        if not headers:
+            return
+        _dev = is_truthy_value(os.environ.get("HERMES_DEV_CREDITS"))
+
+        # ── Parse (fail-open → miss; never overwrite good state with None) ──
+        try:
+            from agent.credits_tracker import parse_credits_headers
+            state = parse_credits_headers(headers, provider=self.provider)
+        except Exception:
+            return  # parse error → treat as a miss, keep last-known
+        if state is None:
+            if _dev:
+                logger.info(
+                    "credits ▸ response had no valid x-nous-credits-* headers "
+                    "(miss — producer off / non-Nous path / >TTL stale)"
+                )
+            return
+
+        # retain-last-known: only overwrite on a fresh valid parse
+        self._credits_state = state
+        # Latch session-start remaining the first time we ever see a header
+        if self._credits_session_start_micros is None:
+            self._credits_session_start_micros = state.remaining_micros
+        if _dev:
+            # HERMES_DEV_CREDITS: stream each capture to agent.log — watch live with
+            # `hermes logs -f` (grep 'credits ▸'). Dev-only; silent for normal users.
+            spent = self.get_credits_spent_micros()
+            used = state.used_fraction
+            logger.info(
+                "credits ▸ remaining=%d (%s) · paid=%s · denom=%s · used=%s "
+                "· Δspent=%s · age=%s%s",
+                state.remaining_micros,
+                state.remaining_usd or "?",
+                state.paid_access,
+                state.denominator_kind,
+                ("%.0f%%" % (used * 100)) if used is not None else "n/a",
+                ("%.1f¢" % (spent / 10000)) if spent is not None else "n/a",
+                ("%.0fs" % state.age_seconds) if state.age_seconds != float("inf") else "n/a",
+                (" · disabled=%s" % state.disabled_reason) if state.disabled_reason else "",
+            )
+
+        # Threshold notices — shared with the cold-start seed (see _emit_credits_notices).
+        self._emit_credits_notices()
+
+    def _emit_credits_notices(self) -> None:
+        """Run the threshold policy on the current credits state and emit notices.
+
+        Shared by the warm path and the cold-start seed so an already-depleted session warns immediately. Runs
+        only
+        when a notice consumer is bound. WARNS on failure. Emits clears FIRST so depleted lands last (latest-
+        wins slot).
+        """
+        if getattr(self, "notice_callback", None) is None and getattr(self, "notice_clear_callback", None) is None:
+            return
+        if not self._credits_notices_enabled():
+            return
+        state = getattr(self, "_credits_state", None)
+        if state is None:
+            return
+        try:
+            from agent.credits_tracker import evaluate_credits_notices, is_free_tier_model, new_credits_latch
+            latch = getattr(self, "_credits_latch", None)
+            if latch is None:
+                latch = self._credits_latch = new_credits_latch()
+            # Free-model gate: a depleted account can still inference on a free model. Local data only.
+            model_is_free = is_free_tier_model(
+                getattr(self, "model", "") or "",
+                getattr(self, "base_url", "") or "",
+            )
+            to_show, to_clear = evaluate_credits_notices(state, latch, model_is_free=model_is_free)
+            for key in to_clear:        # clears FIRST …
+                self._emit_notice_clear(key)
+            for notice in to_show:      # … then shows (depleted lands last in a latest-wins slot)
+                self._emit_notice(notice)
+        except Exception:
+            logger.warning("credits notice evaluation/emit failed", exc_info=True)
+
+    def _credits_notices_enabled(self) -> bool:
+        """Whether credits notices are enabled (``display.credits_notices``).
+
+        Read once per agent and cached (governs UI noise, not correctness); fail-open True.
+        """
+        cached = getattr(self, "_credits_notices_enabled_cache", None)
+        if cached is not None:
+            return cached
+        enabled = True
+        try:
+            from hermes_cli.config import load_config as _load_config
+            _cfg = _load_config() or {}
+            _display = _cfg.get("display") if isinstance(_cfg, dict) else None
+            if isinstance(_display, dict) and "credits_notices" in _display:
+                enabled = bool(_display.get("credits_notices"))
+        except Exception:
+            enabled = True
+        self._credits_notices_enabled_cache = enabled
+        return enabled
+
+    def get_credits_state(self):
+        """Return the last captured CreditsState, or None."""
+        return self._credits_state
+
+    def get_credits_spent_micros(self):
+        """Session-cumulative micros spent = first_seen_remaining - current_remaining. None if no data."""
+        if self._credits_session_start_micros is None or self._credits_state is None:
+            return None
+        return self._credits_session_start_micros - self._credits_state.remaining_micros
+
+    def _check_openrouter_cache_status(self, http_response: Any) -> None:
+        """Read X-OpenRouter-Cache-Status from response headers and log it; HITs count in ``_or_cache_hits``."""
+        if http_response is None:
+            return
+        headers = getattr(http_response, "headers", None)
+        if not headers:
+            return
+        try:
+            status = headers.get("x-openrouter-cache-status")
+            if not status:
+                return
+            if status.upper() == "HIT":
+                self._or_cache_hits += 1
+                logger.info("OpenRouter response cache HIT (total: %d)", self._or_cache_hits)
+            else:
+                logger.debug("OpenRouter response cache %s", status.upper())
+        except Exception:
+            pass  # Never let header parsing break the agent loop
diff --git a/agent/turn_explainers.py b/agent/turn_explainers.py
new file mode 100644
index 0000000000..192d52b95f
--- /dev/null
+++ b/agent/turn_explainers.py
@@ -0,0 +1,356 @@
+"""File-mutation verification footers and turn-completion explanations for ``AIAgent``.
+
+The footer tells the model (and user) when a claimed file mutation did not land; the explainer
+summarises why a turn ended without a final answer.
+Extracted from ``run_agent.py``; every method resolves through ``AIAgent``'s MRO unchanged.
+"""
+import re
+from typing import Any, Dict, Optional
+
+from agent.tool_dispatch_helpers import (
+    _extract_error_preview,
+    _extract_file_mutation_targets,
+    _extract_landed_file_mutation_paths,
+)
+from agent.tool_result_classification import (
+    FILE_MUTATING_TOOL_NAMES as _FILE_MUTATING_TOOLS,
+    file_mutation_result_landed,
+)
+
+
+class TurnExplainersMixin:
+    """File-mutation failure footer + turn-completion explainer (see module docstring)."""
+
+    def _record_file_mutation_result(
+        self,
+        tool_name: str,
+        args: Dict[str, Any],
+        result: Any,
+        is_error: bool,
+    ) -> None:
+        """Record a ``write_file`` / ``patch`` outcome for the turn-end verifier.
+
+        Failures store ``{path: {error_preview, tool}}``; a later success on the same path removes the entry.
+        No-op when the per-turn state dict is not initialised (tool dispatched outside ``run_conversation``).
+        """
+        if tool_name not in _FILE_MUTATING_TOOLS:
+            return
+        state = getattr(self, "_turn_failed_file_mutations", None)
+        if state is None:
+            return
+        targets = _extract_file_mutation_targets(tool_name, args)
+        if not targets:
+            return
+        landed = file_mutation_result_landed(tool_name, result)
+        if landed:
+            landed_paths = _extract_landed_file_mutation_paths(tool_name, args, result)
+            changed = getattr(self, "_turn_file_mutation_paths", None)
+            if changed is not None:
+                changed.update(landed_paths)
+            # Feed the checkpoint agent-write ledger so /rollback's safe mode
+            # can tell Hermes-authored content from later user hand-edits.
+            mgr = getattr(self, "_checkpoint_mgr", None)
+            if mgr is not None and getattr(mgr, "enabled", False):
+                for _p in landed_paths:
+                    try:
+                        mgr.record_agent_write(_p)
+                    except Exception:
+                        pass
+        if is_error and not landed:
+            preview = _extract_error_preview(result)
+            for path in targets:
+                # Keep the FIRST error per path unless a later success replaces it.
+                if path not in state:
+                    state[path] = {
+                        "tool": tool_name,
+                        "error_preview": preview,
+                    }
+        else:
+            for path in targets:
+                state.pop(path, None)
+
+    def _file_mutation_verifier_enabled(self) -> bool:
+        """Check whether the per-turn file-mutation verifier footer is on.
+
+        ``display.file_mutation_verifier`` (default True), cached per agent; ``HERMES_FILE_MUTATION_VERIFIER``
+        overrides on every call and is never cached. A method so tests can patch one seam.
+        """
+        try:
+            import os as _os
+            env = _os.environ.get("HERMES_FILE_MUTATION_VERIFIER")
+            if env is not None:
+                return env.strip().lower() not in {"0", "false", "no", "off"}
+            cached = getattr(self, "_file_mutation_verifier_enabled_cache", None)
+            if cached is not None:
+                return cached
+            # Read from the persisted config.yaml so gateway and CLI share
+            # the same setting.  Import lazily to avoid a startup-time cycle.
+            try:
+                from hermes_cli.config import load_config as _load_config
+                _cfg = _load_config() or {}
+            except Exception:
+                _cfg = {}
+            _display = _cfg.get("display") if isinstance(_cfg, dict) else None
+            if isinstance(_display, dict) and "file_mutation_verifier" in _display:
+                enabled = bool(_display.get("file_mutation_verifier"))
+            else:
+                enabled = True  # safe default: verifier on
+            self._file_mutation_verifier_enabled_cache = enabled
+            return enabled
+        except Exception:
+            pass
+        return True  # safe default: verifier on
+
+    # Bare absolute / home / Windows-drive paths in a footer line. Mirrors the gateway's
+    # extract_local_files detector so anything it WOULD auto-attach is backticked first (#35584).
+    _FOOTER_PATH_RE = re.compile(
+        r"(?<![/:\w.`])(?:~/|/|[A-Za-z]:[/\\])(?:[\w.\-]+[/\\])*[\w.\-]+\.[\w]+",
+    )
+
+    @classmethod
+    def _neutralize_footer_paths(cls, text: str) -> str:
+        """Wrap bare file paths in backticks so the gateway's ``extract_local_files`` never auto-attaches
+        them.
+
+        The extractor skips paths inside inline-code spans. Already-backticked paths are left alone (no
+        double-wrap).
+        """
+        if not text:
+            return text
+        return cls._FOOTER_PATH_RE.sub(lambda m: f"`{m.group(0)}`", text)
+
+    @classmethod
+    def _format_file_mutation_failure_footer(cls, failed: Dict[str, Dict[str, Any]]) -> str:
+        """Render the per-turn failed-mutation dict as a user-facing footer.
+
+        Up to 10 paths with their first error preview, then an overflow count; empty string when nothing
+        failed.
+        Every path is backtick-wrapped via ``_neutralize_footer_paths`` so protected files cannot be auto-
+        delivered.
+        """
+        if not failed:
+            return ""
+        lines = [
+            "⚠️ File-mutation verifier: "
+            f"{len(failed)} file(s) were NOT modified this turn despite any "
+            "wording above that may suggest otherwise. Run `git status` or "
+            "`read_file` to confirm."
+        ]
+        shown = 0
+        for path, info in failed.items():
+            if shown >= 10:
+                break
+            preview = (info.get("error_preview") or "").strip()
+            tool = info.get("tool") or "patch"
+            if preview:
+                lines.append(f"  • `{path}` — [{tool}] {preview}")
+            else:
+                lines.append(f"  • `{path}` — [{tool}] failed")
+            shown += 1
+        remaining = len(failed) - shown
+        if remaining > 0:
+            lines.append(f"  • … and {remaining} more")
+        # Neutralize paths the preview echoed; the lookbehind prevents double-wrapping the bullet path.
+        return cls._neutralize_footer_paths("\n".join(lines))
+
+    def _turn_completion_explainer_enabled(self) -> bool:
+        """Check whether the end-of-turn completion explainer footer is on.
+
+        ``display.turn_completion_explainer`` (default True), cached per agent;
+        ``HERMES_TURN_COMPLETION_EXPLAINER``
+        overrides on every call and is never cached. Mirrors ``_file_mutation_verifier_enabled``.
+        """
+        try:
+            import os as _os
+            env = _os.environ.get("HERMES_TURN_COMPLETION_EXPLAINER")
+            if env is not None:
+                return env.strip().lower() not in {"0", "false", "no", "off"}
+            cached = getattr(self, "_turn_completion_explainer_enabled_cache", None)
+            if cached is not None:
+                return cached
+            # Read from the persisted config.yaml so gateway and CLI share
+            # the same setting.  Import lazily to avoid a startup-time cycle.
+            try:
+                from hermes_cli.config import load_config as _load_config
+                _cfg = _load_config() or {}
+            except Exception:
+                _cfg = {}
+            _display = _cfg.get("display") if isinstance(_cfg, dict) else None
+            if isinstance(_display, dict) and "turn_completion_explainer" in _display:
+                enabled = bool(_display.get("turn_completion_explainer"))
+            else:
+                enabled = True  # safe default: explainer on
+            self._turn_completion_explainer_enabled_cache = enabled
+            return enabled
+        except Exception:
+            pass
+        return True  # safe default: explainer on
+
+    @staticmethod
+    def _format_turn_completion_explanation(
+        turn_exit_reason: str, persistence_cause: Optional[str] = None
+    ) -> str:
+        """Render a user-facing explanation for an abnormal turn ending.
+
+        Maps ``turn_exit_reason`` to an actionable message so a turn with no usable reply is never silent.
+        ``persistence_cause`` refines ``session_persistence_failed`` wording (lock contention ≠ disk full).
+        Returns "" for non-abnormal reasons so callers can concatenate unconditionally.
+        """
+        if not turn_exit_reason:
+            return ""
+        reason = str(turn_exit_reason)
+
+        # Normal completion — stay quiet.  ``text_response(...)`` is the
+        # healthy terminal; anything that produced a real reply is fine.
+        if reason.startswith("text_response"):
+            return ""
+
+        prefix = "⚠️ No reply: "
+        if reason == "empty_response_exhausted":
+            return (
+                prefix
+                + "the model returned empty content after retries and any "
+                "fallback providers. Try `continue`, switch model/provider, "
+                "or inspect the tool output above."
+            )
+        if reason == "all_retries_exhausted_no_response":
+            return (
+                prefix
+                + "all API retries were exhausted before a response was "
+                "produced (provider errors / rate limits). Try `continue` "
+                "or switch provider."
+            )
+        if reason == "partial_stream_recovery":
+            return (
+                prefix
+                + "streaming stopped early and only a partial response was "
+                "recovered. Send `continue` to resume from where it stopped."
+            )
+        if reason == "fallback_prior_turn_content":
+            return (
+                prefix
+                + "no new content was produced this turn; showing recovered "
+                "prior context. Send `continue` to retry."
+            )
+        if reason == "interrupted_during_api_call":
+            return (
+                prefix
+                + "the request was interrupted mid-call before a reply was "
+                "received. Send `continue` to retry."
+            )
+        if reason == "budget_exhausted":
+            return (
+                prefix
+                + "the per-turn iteration/cost budget was exhausted before a "
+                "final answer. Send `continue` to keep going."
+            )
+        if reason == "ollama_runtime_context_too_small":
+            return (
+                prefix
+                + "the local model's context window was too small to finish. "
+                "Increase the context size or use a larger model."
+            )
+        if reason.startswith("max_iterations_reached"):
+            return (
+                prefix
+                + "the maximum tool-iteration limit was reached before a "
+                "final answer. Send `continue` to keep going, or raise "
+                "`max_iterations`."
+            )
+        if reason.startswith("error_near_max_iterations"):
+            return (
+                prefix
+                + "an error occurred near the iteration limit before a final "
+                "answer. Check the tool output above, then send `continue`."
+            )
+        if reason.startswith("repeated_outer_errors"):
+            return (
+                prefix
+                + "the turn kept failing with repeated errors and was stopped "
+                "early instead of retrying forever. Check the errors above, "
+                "then send `continue` to retry."
+            )
+        if reason == "pending_tool_result":
+            return (
+                prefix
+                + "the turn stopped while a tool result was still pending and "
+                "the model produced no follow-up text. Send `continue` to "
+                "let it summarize."
+            )
+        if reason == "session_persistence_failed":
+            cause = persistence_cause or "unknown"
+            if cause == "compression":
+                return (
+                    prefix
+                    + "the turn was stopped because another process was "
+                    "compressing this session. Your message should already be "
+                    "saved — please send it again after compression completes."
+                )
+            if cause == "compression_closed":
+                return (
+                    prefix
+                    + "the turn was stopped because this session was rotated "
+                    "by context compression and its live continuation could "
+                    "not be adopted. The storage itself is healthy — refresh "
+                    "the client (or start a new turn) so it picks up the new "
+                    "session id, then send your message again."
+                )
+            if cause == "turn_lease":
+                return (
+                    prefix
+                    + "the turn was stopped because another Hermes process "
+                    "took over this session. Your reply was not saved — wait "
+                    "for the other process to finish, then send your message "
+                    "again."
+                )
+            if cause == "locked":
+                return (
+                    prefix
+                    + "the turn was stopped because session storage was busy "
+                    "(another Hermes process was writing to the state "
+                    "database). Your message should already be saved — "
+                    "please send it again in a moment."
+                )
+            if cause == "replaced":
+                return (
+                    prefix
+                    + "the turn was stopped because the state database file "
+                    "was replaced underneath this process. Do not run "
+                    "`hermes doctor --fix` or in-place FTS repair — stop "
+                    "the process, restore the intended state.db, then "
+                    "restart. Unwritten messages were diverted to "
+                    "sessions/<session_id>.jsonl and, on the gateway, "
+                    "pending_messages/pending-*.json."
+                )
+            if cause == "corrupt":
+                return (
+                    prefix
+                    + "the turn was stopped because the state database "
+                    "reported structural corruption (the transcript would "
+                    "have been lost on restart). Freeing disk space will "
+                    "not help. Recovery options:\n"
+                    "1. Run `hermes doctor --fix`\n"
+                    "2. Salvage with: sqlite3 ~/.hermes/state.db \".recover\" "
+                    "(then replace state.db)\n"
+                    "3. Restore from a backup in ~/.hermes/backups/\n"
+                    "Then send your message again."
+                )
+            if cause == "disk":
+                return (
+                    prefix
+                    + "the turn was stopped because session storage could not "
+                    "be written (the transcript would have been lost on "
+                    "restart). This is often a full disk — free some space "
+                    "(or fix state.db permissions), then send your message "
+                    "again."
+                )
+            return (
+                prefix
+                + "the turn was stopped because session storage could not be "
+                "written (the transcript would have been lost on restart). "
+                "Check the state database health (`hermes doctor`), then "
+                "send your message again."
+            )
+        # Unknown/diagnostic-only reasons (e.g. "unknown", guardrail_halt
+        # which already surfaces its own message) — don't second-guess.
+        return ""
diff --git a/run_agent.py b/run_agent.py
index 4bb6ff417b..2a908382d1 100644
--- a/run_agent.py
+++ b/run_agent.py
@@ -133,7 +133,6 @@ from agent.process_bootstrap import (
     _get_proxy_for_base_url,  # noqa: F401  # re-exported for tests
 )
 from agent.iteration_budget import IterationBudget
-from agent.interrupt_compat import request_hard_interrupt
 
 
 from hermes_cli.env_loader import load_hermes_dotenv
@@ -177,6 +176,10 @@ from agent.stream_delivery import StreamDeliveryMixin
 from agent.status_output import StatusOutputMixin
 from agent.api_request_hooks import ApiRequestHooksMixin
 from agent.api_error_summary import ApiErrorSummaryMixin
+from agent.interrupt_control import InterruptControlMixin
+from agent.turn_explainers import TurnExplainersMixin
+from agent.activity_tracking import ActivityTrackingMixin
+from agent.rate_limit_credits import RateLimitCreditsMixin
 from agent.lazy_forward import forward as _forward, forward_static as _forward_static
 from agent.redact import redact_sensitive_text
 from agent.session_activity import ActivityProvenance
@@ -226,10 +229,6 @@ from agent.tool_guardrails import (
     append_toolguard_guidance,
     toolguard_synthetic_result,
 )
-from agent.tool_result_classification import (
-    FILE_MUTATING_TOOL_NAMES as _FILE_MUTATING_TOOLS,
-    file_mutation_result_landed,
-)
 from agent.trajectory import (
     convert_scratchpad_to_think,
     save_trajectory as _save_trajectory_to_file,
@@ -242,12 +241,9 @@ from agent.tool_dispatch_helpers import (
     _is_multimodal_tool_result,
     _multimodal_text_summary,
     _append_subdir_hint_to_multimodal,  # noqa: F401  # re-exported for tests that `from run_agent import _append_subdir_hint_to_multimodal`
-    _extract_file_mutation_targets,
-    _extract_landed_file_mutation_paths,
-    _extract_error_preview,
     _trajectory_normalize_msg,  # noqa: F401  # re-exported for tests that `from run_agent import _trajectory_normalize_msg`
 )
-from utils import atomic_json_write, base_url_host_matches, base_url_hostname, env_float, is_truthy_value, model_forces_max_completion_tokens
+from utils import atomic_json_write, base_url_host_matches, base_url_hostname, env_float, model_forces_max_completion_tokens
 
 
 # Flags marking ephemeral empty-response/prefill recovery scaffolding. The loop pops these before
@@ -360,6 +356,10 @@ class AIAgent(
     StatusOutputMixin,
     ApiRequestHooksMixin,
     ApiErrorSummaryMixin,
+    InterruptControlMixin,
+    TurnExplainersMixin,
+    ActivityTrackingMixin,
+    RateLimitCreditsMixin,
 ):
     """AI Agent with tool calling capabilities."""
 
@@ -1951,1081 +1951,8 @@ class AIAgent(
             if self.verbose_logging:
                 logging.warning(f"Failed to save session log: {e}")
 
-    def interrupt(
-        self,
-        message: Optional[str] = None,
-        *,
-        hard_cancel: bool = False,
-        tool_reason: Optional[str] = None,
-        require_generation: Optional[int] = None,
-    ) -> bool:
-        """Request the agent to interrupt its current tool-calling loop (call from another thread).
-
-        ``message``: new message to include in the response context. ``hard_cancel``: explicit stop;
-        compression
-        may honor it even while ordinary interrupts are masked. ``tool_reason``: trusted fixed category safe
-        for
-        tool output. ``require_generation``: activity-generation claim — the interrupt is published only if
-        the
-        turn's generation still matches at the final mutation edge (claim reserved under the activity lock,
-        consumed together with the first observable publication); returns False if the turn resumed meanwhile.
-        """
-        if require_generation is not None:
-            # RESERVE the abort's generation claim under the SAME lock `_touch_activity` stamps with. Real
-            # progress invalidates it; it is CONSUMED at the final mutation edge, so a resumed turn abandons
-            # the abort.
-            with self._liveness_activity_lock():
-                if (
-                    getattr(self, "_turn_liveness_activity_generation", 0)
-                    != require_generation
-                ):
-                    return False
-                self._turn_liveness_abort_claim = require_generation
-
-        # A hard stop and redirect share one lock so /stop cannot race with an
-        # accepted correction and accidentally turn itself into a retry.
-        def _wait_for_compression_commit() -> None:
-            # Pre-claim half of hard-cancel admission (#99758 P1): wait out a commit that already crossed its
-            # boundary but mutate NOTHING — cancelling a pending fence is irreversible and must wait until the
-            # generation claim survived the final mutation edge (_cancel_pending_compression_commit).
-            fence = vars(self).get("_active_compression_commit_fence")
-            if fence is None:
-                return
-            if not getattr(fence, "commit_in_flight", False):
-                # No commit in flight — cancel_before_commit here WOULD cancel the pending commit; leave it to
-                # the
-                # destructive half.
-                return
-            cancel_before_commit = getattr(
-                type(fence), "cancel_before_commit", None
-            )
-            if callable(cancel_before_commit):
-                try:
-                    # A commit holds the fence lock through finish_commit: this blocks until it finishes and
-                    # returns
-                    # False WITHOUT setting _cancelled.
-                    cancel_before_commit(fence)
-                except Exception:
-                    logger.debug(
-                        "Compression hard-cancel fence wait failed",
-                        exc_info=True,
-                    )
-
-        def _cancel_pending_compression_commit() -> None:
-            # Destructive half of hard-cancel admission (#99758 P1): runs only AFTER the claim survived, so a
-            # declined abort never leaves the fence cancelled. A commit that started meanwhile owns the fence
-            # and completes on its own; only a still-pending commit is cancelled here.
-            fence = vars(self).get("_active_compression_commit_fence")
-            if fence is None:
-                return
-            if getattr(fence, "commit_in_flight", False):
-                return
-            cancel_before_commit = getattr(
-                type(fence), "cancel_before_commit", None
-            )
-            if callable(cancel_before_commit):
-                try:
-                    # Marks the fence cancelled (or waits out a just-started commit) without touching the
-                    # hard-stop
-                    # Event, which was published at the claim edge.
-                    cancel_before_commit(fence)
-                except Exception:
-                    logger.debug(
-                        "Compression hard-cancel fence admission failed",
-                        exc_info=True,
-                    )
-
-        def _publish_interrupt_state() -> None:
-            self._interrupt_requested = True
-            self._interrupt_message = message
-            self._tool_interrupt_reason = tool_interrupt_reason
-            if hard_cancel:
-                _hard_event = getattr(
-                    self, "_hard_interrupt_requested", None
-                )
-                if _hard_event is not None:
-                    _hard_event.set()
-
-        def _consume_claim_and_publish_first_state() -> bool:
-            # Final mutation edge: claim consumption and the FIRST observable interrupt publication are ONE
-            # activity-lock critical section, so either the claim survives and commits before any later
-            # activity stamp, or the stamp landed first and the abort declines without publishing.
-            if require_generation is None:
-                # No claim to race: publish WITHOUT the liveness lock. Bare AIAgent stand-ins in other suites
-                # lack
-                # the liveness seam and would AttributeError.
-                _publish_interrupt_state()
-                return True
-            with self._liveness_activity_lock():
-                if (
-                    getattr(self, "_turn_liveness_abort_claim", None)
-                    != require_generation
-                ):
-                    return False
-                self._turn_liveness_abort_claim = None
-                _publish_interrupt_state()
-            return True
-
-        # Tool cancellation attribution stays separate from _interrupt_message, which may carry the user's
-        # full next message.
-        tool_interrupt_reason = (
-            (tool_reason or "explicit stop requested")
-            if hard_cancel
-            else ("user sent a new message" if message else "user interrupt")
-        )
-
-        _redirect_lock = getattr(self, "_pending_redirect_lock", None)
-        if _redirect_lock is not None:
-            with _redirect_lock:
-                # The blocking in-flight-commit wait runs BEFORE the atomic claim edge (redirect lock still
-                # held);
-                # the destructive pending-commit cancel runs AFTER the claim survives (#99758 P1).
-                if hard_cancel:
-                    _wait_for_compression_commit()
-                if not _consume_claim_and_publish_first_state():
-                    return False
-                if hard_cancel:
-                    _cancel_pending_compression_commit()
-                self._pending_redirect = None
-        else:
-            if hard_cancel:
-                _wait_for_compression_commit()
-            if not _consume_claim_and_publish_first_state():
-                return False
-            if hard_cancel:
-                _cancel_pending_compression_commit()
-            self._pending_redirect = None
-
-        # Codex app-server owns its model/tool loop and watches a private
-        # interrupt event rather than Hermes' per-thread flag.
-        if getattr(self, "api_mode", None) == "codex_app_server":
-            _codex_session = getattr(self, "_codex_session", None)
-            _request_interrupt = getattr(_codex_session, "request_interrupt", None)
-            if callable(_request_interrupt):
-                try:
-                    _request_interrupt()
-                except Exception:
-                    logger.debug(
-                        "Failed to interrupt Codex app-server turn",
-                        exc_info=True,
-                    )
-
-        # Cron turns request on the conversation thread (no nested interrupt-worker deadlock); their client
-        # is registered here so this cross-thread interrupt can still shut the sockets.
-        _abort_active_request = getattr(self, "_active_request_abort", None)
-        if callable(_abort_active_request):
-            try:
-                _abort_active_request("interrupt_abort")
-            except Exception:
-                logger.debug("Failed to abort active inline request", exc_info=True)
-        # Scope the tool interrupt to this agent's execution thread so other in-process agents are unaffected.
-        if self._execution_thread_id is not None:
-            _set_interrupt(
-                True,
-                self._execution_thread_id,
-                reason=tool_interrupt_reason,
-            )
-            self._interrupt_thread_signal_pending = False
-        else:
-            # Interrupt arrived before run_conversation bound the execution thread: defer the tool-level
-            # signal instead of targeting the caller thread.
-            self._interrupt_thread_signal_pending = True
-        # Fan out to concurrent-tool worker tids: is_interrupted() inside a tool only sees its own tid, so
-        # without this a hung concurrent tool runs to its own timeout. getattr covers __init__-less stubs.
-        _tracker = getattr(self, "_tool_worker_threads", None)
-        _tracker_lock = getattr(self, "_tool_worker_threads_lock", None)
-        if _tracker is not None and _tracker_lock is not None:
-            with _tracker_lock:
-                _worker_tids = list(_tracker)
-            for _wtid in _worker_tids:
-                try:
-                    _set_interrupt(True, _wtid, reason=tool_interrupt_reason)
-                except Exception:
-                    pass
-        # Propagate interrupt to any running child agents (subagent delegation)
-        with self._active_children_lock:
-            children_copy = list(self._active_children)
-        for child in children_copy:
-            try:
-                if hard_cancel:
-                    request_hard_interrupt(
-                        child,
-                        message,
-                        tool_reason=tool_interrupt_reason,
-                    )
-                else:
-                    child.interrupt(message)
-            except Exception as e:
-                logger.debug("Failed to propagate interrupt to child agent: %s", e)
-        if not self.quiet_mode:
-            print("\n⚡ Interrupt requested" + (f": '{message[:40]}...'" if message and len(message) > 40 else f": '{message}'" if message else ""))
-        return True
-
-    def hard_interrupt(
-        self,
-        message: Optional[str] = None,
-        *,
-        tool_reason: Optional[str] = None,
-    ) -> None:
-        """Request an explicit stop while preserving the ``interrupt()`` ABI.
-
-        Frontends feature-detect this and fall back to legacy ``interrupt()`` for third-party agents.
-        """
-        # Bypass dynamic dispatch: legacy subclasses may override interrupt(message=None) without hard_cancel.
-        AIAgent.interrupt(
-            self,
-            message,
-            hard_cancel=True,
-            tool_reason=tool_reason,
-        )
-
-    def clear_interrupt(self, *, preserve_redirect: bool = False) -> bool:
-        """Clear the interrupt request and per-thread tool signal.
-
-        ``preserve_redirect`` is only for the conversation loop rebuilding the same logical turn after
-        cancelling a model request; public hard-stop paths clear everything.
-        """
-        _redirect_lock = getattr(self, "_pending_redirect_lock", None)
-        if _redirect_lock is not None:
-            with _redirect_lock:
-                if preserve_redirect and not self._pending_redirect:
-                    return False
-                self._interrupt_requested = False
-                self._interrupt_message = None
-                self._tool_interrupt_reason = None
-                getattr(self, "_hard_interrupt_requested", threading.Event()).clear()
-                if not preserve_redirect:
-                    self._pending_redirect = None
-        else:
-            if preserve_redirect and not getattr(self, "_pending_redirect", None):
-                return False
-            self._interrupt_requested = False
-            self._interrupt_message = None
-            self._tool_interrupt_reason = None
-            getattr(self, "_hard_interrupt_requested", threading.Event()).clear()
-            if not preserve_redirect:
-                self._pending_redirect = None
-        self._interrupt_thread_signal_pending = False
-        if self._execution_thread_id is not None:
-            _set_interrupt(False, self._execution_thread_id)
-        # Also clear worker-thread bits so no stale interrupt survives a turn boundary onto a recycled tid.
-        # getattr covers __init__-less test stubs.
-        _tracker = getattr(self, "_tool_worker_threads", None)
-        _tracker_lock = getattr(self, "_tool_worker_threads_lock", None)
-        if _tracker is not None and _tracker_lock is not None:
-            with _tracker_lock:
-                _worker_tids = list(_tracker)
-            for _wtid in _worker_tids:
-                try:
-                    _set_interrupt(False, _wtid)
-                except Exception:
-                    pass
-        # A hard interrupt supersedes any pending /steer — its target iteration will no longer happen.
-        _steer_lock = getattr(self, "_pending_steer_lock", None)
-        if _steer_lock is not None:
-            with _steer_lock:
-                self._pending_steer = None
-        return True
-
-    def steer(self, text: str) -> bool:
-        """Inject user text into the next tool result without interrupting the current tool.
-
-        The text is appended to the LAST tool result once the batch finishes, so the model sees it on its next
-        iteration. Thread-safe; multiple calls concatenate with newlines. Returns False for empty text.
-        """
-        if not text or not text.strip():
-            return False
-        cleaned = text.strip()
-        _lock = getattr(self, "_pending_steer_lock", None)
-        if _lock is None:
-            # __init__-less test stubs: fall back to a direct attribute set.
-            existing = getattr(self, "_pending_steer", None)
-            self._pending_steer = (existing + "\n" + cleaned) if existing else cleaned
-            return True
-        with _lock:
-            if self._pending_steer:
-                self._pending_steer = self._pending_steer + "\n" + cleaned
-            else:
-                self._pending_steer = cleaned
-        return True
-
-    def redirect(self, text: str) -> bool:
-        """Redirect the active turn without converting it into a new task.
-
-        During a model request this cancels only that request: completed messages/tool results are kept, the
-        displayed partial reasoning becomes assistant context, the correction is appended as a real user
-        message,
-        and the loop retries. During tool execution it degrades to ``steer()``; Codex app-server uses native
-        ``turn/steer``. Returns False when there is no live turn or the text is empty.
-        """
-        if not text or not text.strip():
-            return False
-        cleaned = text.strip()
-
-        # Codex owns its internal reasoning/tool loop, so use its first-class
-        # active-turn steering protocol rather than interrupting the subprocess.
-        if getattr(self, "api_mode", None) == "codex_app_server":
-            _codex_session = getattr(self, "_codex_session", None)
-            _native_steer = getattr(_codex_session, "request_steer", None)
-            if callable(_native_steer):
-                _redirect_lock = getattr(self, "_pending_redirect_lock", None)
-                if _redirect_lock is not None:
-                    with _redirect_lock:
-                        if self._interrupt_requested:
-                            return False
-                elif self._interrupt_requested:
-                    return False
-                try:
-                    return bool(_native_steer(cleaned))
-                except Exception:
-                    logger.debug("Codex app-server turn/steer failed", exc_info=True)
-                    return False
-
-        # Never kill a tool to deliver guidance; the steer drain puts it on the final tool result.
-        if getattr(self, "_executing_tools", False):
-            return self.steer(cleaned)
-
-        _model_active = getattr(self, "_model_request_active", None)
-        _redirect_lock = getattr(self, "_pending_redirect_lock", None)
-        if _redirect_lock is None:
-            if _model_active is None or not _model_active.is_set():
-                return False
-            existing = getattr(self, "_pending_redirect", None)
-            if self._interrupt_requested and not existing:
-                return False
-            self._pending_redirect = (
-                f"{existing}\n\n[Additional user correction]\n{cleaned}"
-                if existing
-                else cleaned
-            )
-            self._interrupt_requested = True
-            self._interrupt_message = None
-        else:
-            with _redirect_lock:
-                if _model_active is None or not _model_active.is_set():
-                    # The response completed before we acquired the state lock.
-                    # Reject so the surface queues a new turn.
-                    return False
-                if self._interrupt_requested and not self._pending_redirect:
-                    return False
-                if self._pending_redirect:
-                    self._pending_redirect = (
-                        f"{self._pending_redirect}\n\n"
-                        f"[Additional user correction]\n{cleaned}"
-                    )
-                else:
-                    self._pending_redirect = cleaned
-                self._interrupt_requested = True
-                self._interrupt_message = None
-
-        # Interrupt only the model request. Do not fan out to tool workers or
-        # child agents as interrupt() does.
-        _execution_thread_id = getattr(self, "_execution_thread_id", None)
-        if _execution_thread_id is not None:
-            _set_interrupt(True, _execution_thread_id)
-            self._interrupt_thread_signal_pending = False
-        else:
-            self._interrupt_thread_signal_pending = True
-        _abort_active_request = getattr(self, "_active_request_abort", None)
-        if callable(_abort_active_request):
-            try:
-                _abort_active_request("redirect_abort")
-            except Exception:
-                logger.debug("Failed to abort request for redirect", exc_info=True)
-        return True
-
-    def _has_pending_redirect(self) -> bool:
-        """Return whether an active-turn redirect is waiting to be applied."""
-        _redirect_lock = getattr(self, "_pending_redirect_lock", None)
-        if _redirect_lock is None:
-            return bool(getattr(self, "_pending_redirect", None))
-        with _redirect_lock:
-            return bool(self._pending_redirect)
-
-    def _drain_pending_redirect(self) -> Optional[str]:
-        """Return and clear pending active-turn correction text."""
-        _redirect_lock = getattr(self, "_pending_redirect_lock", None)
-        if _redirect_lock is None:
-            text = getattr(self, "_pending_redirect", None)
-            self._pending_redirect = None
-            return text
-        with _redirect_lock:
-            text = self._pending_redirect
-            self._pending_redirect = None
-        return text
-
-    def _drain_pending_steer(self) -> Optional[str]:
-        """Return the pending steer text (if any) and clear the slot; None when nothing is pending."""
-        _lock = getattr(self, "_pending_steer_lock", None)
-        if _lock is None:
-            text = getattr(self, "_pending_steer", None)
-            self._pending_steer = None
-            return text
-        with _lock:
-            text = self._pending_steer
-            self._pending_steer = None
-        return text
-
-    def _record_file_mutation_result(
-        self,
-        tool_name: str,
-        args: Dict[str, Any],
-        result: Any,
-        is_error: bool,
-    ) -> None:
-        """Record a ``write_file`` / ``patch`` outcome for the turn-end verifier.
-
-        Failures store ``{path: {error_preview, tool}}``; a later success on the same path removes the entry.
-        No-op when the per-turn state dict is not initialised (tool dispatched outside ``run_conversation``).
-        """
-        if tool_name not in _FILE_MUTATING_TOOLS:
-            return
-        state = getattr(self, "_turn_failed_file_mutations", None)
-        if state is None:
-            return
-        targets = _extract_file_mutation_targets(tool_name, args)
-        if not targets:
-            return
-        landed = file_mutation_result_landed(tool_name, result)
-        if landed:
-            landed_paths = _extract_landed_file_mutation_paths(tool_name, args, result)
-            changed = getattr(self, "_turn_file_mutation_paths", None)
-            if changed is not None:
-                changed.update(landed_paths)
-            # Feed the checkpoint agent-write ledger so /rollback's safe mode
-            # can tell Hermes-authored content from later user hand-edits.
-            mgr = getattr(self, "_checkpoint_mgr", None)
-            if mgr is not None and getattr(mgr, "enabled", False):
-                for _p in landed_paths:
-                    try:
-                        mgr.record_agent_write(_p)
-                    except Exception:
-                        pass
-        if is_error and not landed:
-            preview = _extract_error_preview(result)
-            for path in targets:
-                # Keep the FIRST error per path unless a later success replaces it.
-                if path not in state:
-                    state[path] = {
-                        "tool": tool_name,
-                        "error_preview": preview,
-                    }
-        else:
-            for path in targets:
-                state.pop(path, None)
-
-    def _file_mutation_verifier_enabled(self) -> bool:
-        """Check whether the per-turn file-mutation verifier footer is on.
-
-        ``display.file_mutation_verifier`` (default True), cached per agent; ``HERMES_FILE_MUTATION_VERIFIER``
-        overrides on every call and is never cached. A method so tests can patch one seam.
-        """
-        try:
-            import os as _os
-            env = _os.environ.get("HERMES_FILE_MUTATION_VERIFIER")
-            if env is not None:
-                return env.strip().lower() not in {"0", "false", "no", "off"}
-            cached = getattr(self, "_file_mutation_verifier_enabled_cache", None)
-            if cached is not None:
-                return cached
-            # Read from the persisted config.yaml so gateway and CLI share
-            # the same setting.  Import lazily to avoid a startup-time cycle.
-            try:
-                from hermes_cli.config import load_config as _load_config
-                _cfg = _load_config() or {}
-            except Exception:
-                _cfg = {}
-            _display = _cfg.get("display") if isinstance(_cfg, dict) else None
-            if isinstance(_display, dict) and "file_mutation_verifier" in _display:
-                enabled = bool(_display.get("file_mutation_verifier"))
-            else:
-                enabled = True  # safe default: verifier on
-            self._file_mutation_verifier_enabled_cache = enabled
-            return enabled
-        except Exception:
-            pass
-        return True  # safe default: verifier on
-
-    # Bare absolute / home / Windows-drive paths in a footer line. Mirrors the gateway's
-    # extract_local_files detector so anything it WOULD auto-attach is backticked first (#35584).
-    _FOOTER_PATH_RE = re.compile(
-        r"(?<![/:\w.`])(?:~/|/|[A-Za-z]:[/\\])(?:[\w.\-]+[/\\])*[\w.\-]+\.[\w]+",
-    )
-
-    @classmethod
-    def _neutralize_footer_paths(cls, text: str) -> str:
-        """Wrap bare file paths in backticks so the gateway's ``extract_local_files`` never auto-attaches
-        them.
-
-        The extractor skips paths inside inline-code spans. Already-backticked paths are left alone (no
-        double-wrap).
-        """
-        if not text:
-            return text
-        return cls._FOOTER_PATH_RE.sub(lambda m: f"`{m.group(0)}`", text)
-
-    @classmethod
-    def _format_file_mutation_failure_footer(cls, failed: Dict[str, Dict[str, Any]]) -> str:
-        """Render the per-turn failed-mutation dict as a user-facing footer.
-
-        Up to 10 paths with their first error preview, then an overflow count; empty string when nothing
-        failed.
-        Every path is backtick-wrapped via ``_neutralize_footer_paths`` so protected files cannot be auto-
-        delivered.
-        """
-        if not failed:
-            return ""
-        lines = [
-            "⚠️ File-mutation verifier: "
-            f"{len(failed)} file(s) were NOT modified this turn despite any "
-            "wording above that may suggest otherwise. Run `git status` or "
-            "`read_file` to confirm."
-        ]
-        shown = 0
-        for path, info in failed.items():
-            if shown >= 10:
-                break
-            preview = (info.get("error_preview") or "").strip()
-            tool = info.get("tool") or "patch"
-            if preview:
-                lines.append(f"  • `{path}` — [{tool}] {preview}")
-            else:
-                lines.append(f"  • `{path}` — [{tool}] failed")
-            shown += 1
-        remaining = len(failed) - shown
-        if remaining > 0:
-            lines.append(f"  • … and {remaining} more")
-        # Neutralize paths the preview echoed; the lookbehind prevents double-wrapping the bullet path.
-        return cls._neutralize_footer_paths("\n".join(lines))
-
-    def _turn_completion_explainer_enabled(self) -> bool:
-        """Check whether the end-of-turn completion explainer footer is on.
-
-        ``display.turn_completion_explainer`` (default True), cached per agent;
-        ``HERMES_TURN_COMPLETION_EXPLAINER``
-        overrides on every call and is never cached. Mirrors ``_file_mutation_verifier_enabled``.
-        """
-        try:
-            import os as _os
-            env = _os.environ.get("HERMES_TURN_COMPLETION_EXPLAINER")
-            if env is not None:
-                return env.strip().lower() not in {"0", "false", "no", "off"}
-            cached = getattr(self, "_turn_completion_explainer_enabled_cache", None)
-            if cached is not None:
-                return cached
-            # Read from the persisted config.yaml so gateway and CLI share
-            # the same setting.  Import lazily to avoid a startup-time cycle.
-            try:
-                from hermes_cli.config import load_config as _load_config
-                _cfg = _load_config() or {}
-            except Exception:
-                _cfg = {}
-            _display = _cfg.get("display") if isinstance(_cfg, dict) else None
-            if isinstance(_display, dict) and "turn_completion_explainer" in _display:
-                enabled = bool(_display.get("turn_completion_explainer"))
-            else:
-                enabled = True  # safe default: explainer on
-            self._turn_completion_explainer_enabled_cache = enabled
-            return enabled
-        except Exception:
-            pass
-        return True  # safe default: explainer on
-
-    @staticmethod
-    def _format_turn_completion_explanation(
-        turn_exit_reason: str, persistence_cause: Optional[str] = None
-    ) -> str:
-        """Render a user-facing explanation for an abnormal turn ending.
-
-        Maps ``turn_exit_reason`` to an actionable message so a turn with no usable reply is never silent.
-        ``persistence_cause`` refines ``session_persistence_failed`` wording (lock contention ≠ disk full).
-        Returns "" for non-abnormal reasons so callers can concatenate unconditionally.
-        """
-        if not turn_exit_reason:
-            return ""
-        reason = str(turn_exit_reason)
-
-        # Normal completion — stay quiet.  ``text_response(...)`` is the
-        # healthy terminal; anything that produced a real reply is fine.
-        if reason.startswith("text_response"):
-            return ""
-
-        prefix = "⚠️ No reply: "
-        if reason == "empty_response_exhausted":
-            return (
-                prefix
-                + "the model returned empty content after retries and any "
-                "fallback providers. Try `continue`, switch model/provider, "
-                "or inspect the tool output above."
-            )
-        if reason == "all_retries_exhausted_no_response":
-            return (
-                prefix
-                + "all API retries were exhausted before a response was "
-                "produced (provider errors / rate limits). Try `continue` "
-                "or switch provider."
-            )
-        if reason == "partial_stream_recovery":
-            return (
-                prefix
-                + "streaming stopped early and only a partial response was "
-                "recovered. Send `continue` to resume from where it stopped."
-            )
-        if reason == "fallback_prior_turn_content":
-            return (
-                prefix
-                + "no new content was produced this turn; showing recovered "
-                "prior context. Send `continue` to retry."
-            )
-        if reason == "interrupted_during_api_call":
-            return (
-                prefix
-                + "the request was interrupted mid-call before a reply was "
-                "received. Send `continue` to retry."
-            )
-        if reason == "budget_exhausted":
-            return (
-                prefix
-                + "the per-turn iteration/cost budget was exhausted before a "
-                "final answer. Send `continue` to keep going."
-            )
-        if reason == "ollama_runtime_context_too_small":
-            return (
-                prefix
-                + "the local model's context window was too small to finish. "
-                "Increase the context size or use a larger model."
-            )
-        if reason.startswith("max_iterations_reached"):
-            return (
-                prefix
-                + "the maximum tool-iteration limit was reached before a "
-                "final answer. Send `continue` to keep going, or raise "
-                "`max_iterations`."
-            )
-        if reason.startswith("error_near_max_iterations"):
-            return (
-                prefix
-                + "an error occurred near the iteration limit before a final "
-                "answer. Check the tool output above, then send `continue`."
-            )
-        if reason.startswith("repeated_outer_errors"):
-            return (
-                prefix
-                + "the turn kept failing with repeated errors and was stopped "
-                "early instead of retrying forever. Check the errors above, "
-                "then send `continue` to retry."
-            )
-        if reason == "pending_tool_result":
-            return (
-                prefix
-                + "the turn stopped while a tool result was still pending and "
-                "the model produced no follow-up text. Send `continue` to "
-                "let it summarize."
-            )
-        if reason == "session_persistence_failed":
-            cause = persistence_cause or "unknown"
-            if cause == "compression":
-                return (
-                    prefix
-                    + "the turn was stopped because another process was "
-                    "compressing this session. Your message should already be "
-                    "saved — please send it again after compression completes."
-                )
-            if cause == "compression_closed":
-                return (
-                    prefix
-                    + "the turn was stopped because this session was rotated "
-                    "by context compression and its live continuation could "
-                    "not be adopted. The storage itself is healthy — refresh "
-                    "the client (or start a new turn) so it picks up the new "
-                    "session id, then send your message again."
-                )
-            if cause == "turn_lease":
-                return (
-                    prefix
-                    + "the turn was stopped because another Hermes process "
-                    "took over this session. Your reply was not saved — wait "
-                    "for the other process to finish, then send your message "
-                    "again."
-                )
-            if cause == "locked":
-                return (
-                    prefix
-                    + "the turn was stopped because session storage was busy "
-                    "(another Hermes process was writing to the state "
-                    "database). Your message should already be saved — "
-                    "please send it again in a moment."
-                )
-            if cause == "replaced":
-                return (
-                    prefix
-                    + "the turn was stopped because the state database file "
-                    "was replaced underneath this process. Do not run "
-                    "`hermes doctor --fix` or in-place FTS repair — stop "
-                    "the process, restore the intended state.db, then "
-                    "restart. Unwritten messages were diverted to "
-                    "sessions/<session_id>.jsonl and, on the gateway, "
-                    "pending_messages/pending-*.json."
-                )
-            if cause == "corrupt":
-                return (
-                    prefix
-                    + "the turn was stopped because the state database "
-                    "reported structural corruption (the transcript would "
-                    "have been lost on restart). Freeing disk space will "
-                    "not help. Recovery options:\n"
-                    "1. Run `hermes doctor --fix`\n"
-                    "2. Salvage with: sqlite3 ~/.hermes/state.db \".recover\" "
-                    "(then replace state.db)\n"
-                    "3. Restore from a backup in ~/.hermes/backups/\n"
-                    "Then send your message again."
-                )
-            if cause == "disk":
-                return (
-                    prefix
-                    + "the turn was stopped because session storage could not "
-                    "be written (the transcript would have been lost on "
-                    "restart). This is often a full disk — free some space "
-                    "(or fix state.db permissions), then send your message "
-                    "again."
-                )
-            return (
-                prefix
-                + "the turn was stopped because session storage could not be "
-                "written (the transcript would have been lost on restart). "
-                "Check the state database health (`hermes doctor`), then "
-                "send your message again."
-            )
-        # Unknown/diagnostic-only reasons (e.g. "unknown", guardrail_halt
-        # which already surfaces its own message) — don't second-guess.
-        return ""
-
     _apply_pending_steer_to_tool_results = _forward("agent.agent_runtime_helpers", "apply_pending_steer_to_tool_results")
 
-    def _liveness_activity_lock(self) -> "threading.Lock":
-        """Shared lock for the activity clock and its generation counter.
-
-        ``_touch_activity`` stamps under it and the liveness watchdog samples/commits under it, so a stall
-        observation can never abort a turn that resumed in between. Lazy so ``__new__``-built doubles work.
-        """
-        _lock = getattr(self, "_turn_liveness_activity_lock", None)
-        if _lock is None:
-            _lock = threading.Lock()
-            self._turn_liveness_activity_lock = _lock
-        return _lock
-
-    def _touch_activity(
-        self,
-        desc: str,
-        *,
-        provenance: Optional[ActivityProvenance] = None,
-        force_persist: bool = False,
-    ) -> None:
-        """Update the last-activity timestamp and description (thread-safe).
-
-        Bumps a monotonic generation under ``_liveness_activity_lock`` so the watchdog can bind a stall
-        observation
-        to the exact ``(generation, timestamp)`` it sampled. Also bridges (rate-limited, best-effort) to the
-        kanban
-        heartbeat when this is a dispatcher-spawned worker, and to the durable SessionDB activity projection.
-        ``provenance`` names special writers (compression); ``force_persist`` bypasses the SessionDB rate
-        limit.
-        """
-        from agent.session_activity import (
-            bound_activity_description,
-            normalize_activity_provenance,
-            reset_session_activity_persist_window,
-        )
-
-        # Lazy per-instance lock, inline so SimpleNamespace doubles binding _touch_activity without the
-        # class keep working (tests/run_agent/test_session_activity_persist.py).
-        _clock_lock = getattr(self, "_turn_liveness_activity_lock", None)
-        if _clock_lock is None:
-            _clock_lock = threading.Lock()
-            self._turn_liveness_activity_lock = _clock_lock
-        with _clock_lock:
-            self._turn_liveness_activity_generation = (
-                getattr(self, "_turn_liveness_activity_generation", 0) + 1
-            )
-            self._last_activity_ts = time.time()
-            self._last_activity_desc = bound_activity_description(desc)
-            self._last_activity_provenance = normalize_activity_provenance(provenance)
-            # Real progress invalidates a reserved abort claim; an in-flight watchdog interrupt must abandon
-            # itself at the final mutation edge.
-            self._turn_liveness_abort_claim = None
-        if os.environ.get("HERMES_KANBAN_TASK"):
-            try:
-                from tools.kanban_tools import (
-                    heartbeat_current_worker_from_env,
-                    inject_new_comments_from_env,
-                )
-                heartbeat_current_worker_from_env()
-                # Fold any new operator notes into the running turn (OUT-OF-BAND
-                # steer) so the user can talk to a live task without a restart.
-                inject_new_comments_from_env(self)
-            except Exception:
-                # Never let the bridge break the loop; this guard covers import-time failures.
-                pass
-        if force_persist:
-            reset_session_activity_persist_window(self)
-        self._persist_session_activity_if_due()
-
-    def _persist_session_activity_if_due(self) -> None:
-        """Best-effort durable activity heartbeat for SessionDB consumers.
-
-        Cadence pinned by ``SESSION_ACTIVITY_HEARTBEAT_MIN_INTERVAL_SECONDS`` (config-independent). Fail-open:
-        a failed write never raises into the agent loop.
-        """
-        session_id = getattr(self, "session_id", None)
-        session_db = getattr(self, "_session_db", None)
-        if not session_id or session_db is None:
-            return
-        touch = getattr(session_db, "touch_session_activity", None)
-        if not callable(touch):
-            return
-        from agent.session_activity import (
-            SESSION_ACTIVITY_HEARTBEAT_MIN_INTERVAL_SECONDS,
-            normalize_activity_provenance,
-        )
-
-        now_mono = time.monotonic()
-        last_mono = getattr(self, "_session_activity_last_persist_mono", 0.0)
-        if (now_mono - last_mono) < SESSION_ACTIVITY_HEARTBEAT_MIN_INTERVAL_SECONDS:
-            return
-        self._session_activity_last_persist_mono = now_mono
-        try:
-            touch(
-                session_id,
-                getattr(self, "_last_activity_ts", None),
-                description=getattr(self, "_last_activity_desc", None),
-                provenance=normalize_activity_provenance(
-                    getattr(self, "_last_activity_provenance", None)
-                ),
-            )
-        except Exception:
-            # Heartbeat is observation-only; never let its I/O break the loop.
-            logger.debug(
-                "session activity heartbeat write failed (ignored)",
-                exc_info=True,
-            )
-
-    def _reset_activity_labels_after_turn(self) -> None:
-        """Drop mid-turn activity labels once the turn is no longer running.
-
-        Keeps ``_last_activity_ts`` so idle/watchdog clocks stay continuous across turns; clears description +
-        provenance so idle agents / SessionDB listings stop advertising the last mid-turn stamp.
-        """
-        from agent.session_activity import ActivityProvenance
-
-        self._last_activity_desc = ""
-        self._last_activity_provenance = ActivityProvenance.UNKNOWN
-        session_id = getattr(self, "session_id", None)
-        session_db = getattr(self, "_session_db", None)
-        if not session_id or session_db is None:
-            return
-        clear = getattr(session_db, "clear_session_activity_labels", None)
-        if not callable(clear):
-            return
-        try:
-            clear(session_id)
-        except Exception:
-            # Never let durable cleanup I/O break turn teardown.
-            pass
-
-    def _capture_rate_limits(self, http_response: Any) -> None:
-        """Parse x-ratelimit-* headers from an HTTP response and cache the state.
-
-        Called after each streaming call; the httpx Response is available as ``stream.response``.
-        """
-        if http_response is None:
-            return
-        headers = getattr(http_response, "headers", None)
-        if not headers:
-            return
-        try:
-            from agent.rate_limit_tracker import parse_rate_limit_headers
-            state = parse_rate_limit_headers(headers, provider=self.provider)
-            if state is not None:
-                self._rate_limit_state = state
-        except Exception:
-            pass  # Never let header parsing break the agent loop
-
-    def get_rate_limit_state(self):
-        """Return the last captured RateLimitState, or None."""
-        return self._rate_limit_state
-
-    def _capture_anthropic_response_headers(self, http_response: Any) -> None:
-        """Capture out-of-band state from Anthropic Messages response headers.
-
-        The SDK's aggregated ``Message`` drops headers, where Portal puts rate-limit and credits state. Fail-
-        open.
-        """
-        self._capture_rate_limits(http_response)
-        self._capture_credits(http_response)
-
-    def _capture_credits(self, http_response: Any) -> None:
-        """Parse x-nous-credits-* headers, cache CreditsState, fire threshold notices.
-
-        The PARSE is swallowed (miss → keep last-known); the notice EVALUATION is a separate block that WARNS
-        on
-        failure so a depletion-notice bug cannot vanish silently.
-        """
-        # Dev test fixture (HERMES_DEV_CREDITS_FIXTURE): inject a chosen notice state
-        # each turn for repeatable testing, bypassing real headers. Throwaway scaffolding.
-        try:
-            from agent.credits_tracker import dev_fixture_credits_state
-            _fixture = dev_fixture_credits_state()
-        except Exception:
-            _fixture = None
-        if _fixture is not None:
-            self._credits_state = _fixture
-            if self._credits_session_start_micros is None:
-                self._credits_session_start_micros = _fixture.remaining_micros
-            _latch = getattr(self, "_credits_latch", None)
-            if isinstance(_latch, dict):
-                # Only seen_below_90 — priming seen_grant_unspent would fire grant_spent on first observation.
-                _latch["seen_below_90"] = True  # let warn90 fire without a real crossing
-            _used = _fixture.used_fraction
-            logger.info(
-                "credits ▸ [FIXTURE] remaining=%d (%s) · paid=%s · denom=%s · used=%s "
-                "(real headers bypassed — `echo clear` / unset HERMES_DEV_CREDITS_FIXTURE to restore)",
-                _fixture.remaining_micros,
-                _fixture.remaining_usd or "?",
-                _fixture.paid_access,
-                _fixture.denominator_kind,
-                ("%.0f%%" % (_used * 100)) if _used is not None else "n/a",
-            )
-            self._emit_credits_notices()
-            return
-        if http_response is None:
-            return
-        headers = getattr(http_response, "headers", None)
-        if not headers:
-            return
-        _dev = is_truthy_value(os.environ.get("HERMES_DEV_CREDITS"))
-
-        # ── Parse (fail-open → miss; never overwrite good state with None) ──
-        try:
-            from agent.credits_tracker import parse_credits_headers
-            state = parse_credits_headers(headers, provider=self.provider)
-        except Exception:
-            return  # parse error → treat as a miss, keep last-known
-        if state is None:
-            if _dev:
-                logger.info(
-                    "credits ▸ response had no valid x-nous-credits-* headers "
-                    "(miss — producer off / non-Nous path / >TTL stale)"
-                )
-            return
-
-        # retain-last-known: only overwrite on a fresh valid parse
-        self._credits_state = state
-        # Latch session-start remaining the first time we ever see a header
-        if self._credits_session_start_micros is None:
-            self._credits_session_start_micros = state.remaining_micros
-        if _dev:
-            # HERMES_DEV_CREDITS: stream each capture to agent.log — watch live with
-            # `hermes logs -f` (grep 'credits ▸'). Dev-only; silent for normal users.
-            spent = self.get_credits_spent_micros()
-            used = state.used_fraction
-            logger.info(
-                "credits ▸ remaining=%d (%s) · paid=%s · denom=%s · used=%s "
-                "· Δspent=%s · age=%s%s",
-                state.remaining_micros,
-                state.remaining_usd or "?",
-                state.paid_access,
-                state.denominator_kind,
-                ("%.0f%%" % (used * 100)) if used is not None else "n/a",
-                ("%.1f¢" % (spent / 10000)) if spent is not None else "n/a",
-                ("%.0fs" % state.age_seconds) if state.age_seconds != float("inf") else "n/a",
-                (" · disabled=%s" % state.disabled_reason) if state.disabled_reason else "",
-            )
-
-        # Threshold notices — shared with the cold-start seed (see _emit_credits_notices).
-        self._emit_credits_notices()
-
-    def _emit_credits_notices(self) -> None:
-        """Run the threshold policy on the current credits state and emit notices.
-
-        Shared by the warm path and the cold-start seed so an already-depleted session warns immediately. Runs
-        only
-        when a notice consumer is bound. WARNS on failure. Emits clears FIRST so depleted lands last (latest-
-        wins slot).
-        """
-        if getattr(self, "notice_callback", None) is None and getattr(self, "notice_clear_callback", None) is None:
-            return
-        if not self._credits_notices_enabled():
-            return
-        state = getattr(self, "_credits_state", None)
-        if state is None:
-            return
-        try:
-            from agent.credits_tracker import evaluate_credits_notices, is_free_tier_model, new_credits_latch
-            latch = getattr(self, "_credits_latch", None)
-            if latch is None:
-                latch = self._credits_latch = new_credits_latch()
-            # Free-model gate: a depleted account can still inference on a free model. Local data only.
-            model_is_free = is_free_tier_model(
-                getattr(self, "model", "") or "",
-                getattr(self, "base_url", "") or "",
-            )
-            to_show, to_clear = evaluate_credits_notices(state, latch, model_is_free=model_is_free)
-            for key in to_clear:        # clears FIRST …
-                self._emit_notice_clear(key)
-            for notice in to_show:      # … then shows (depleted lands last in a latest-wins slot)
-                self._emit_notice(notice)
-        except Exception:
-            logger.warning("credits notice evaluation/emit failed", exc_info=True)
-
-    def _credits_notices_enabled(self) -> bool:
-        """Whether credits notices are enabled (``display.credits_notices``).
-
-        Read once per agent and cached (governs UI noise, not correctness); fail-open True.
-        """
-        cached = getattr(self, "_credits_notices_enabled_cache", None)
-        if cached is not None:
-            return cached
-        enabled = True
-        try:
-            from hermes_cli.config import load_config as _load_config
-            _cfg = _load_config() or {}
-            _display = _cfg.get("display") if isinstance(_cfg, dict) else None
-            if isinstance(_display, dict) and "credits_notices" in _display:
-                enabled = bool(_display.get("credits_notices"))
-        except Exception:
-            enabled = True
-        self._credits_notices_enabled_cache = enabled
-        return enabled
-
-    def get_credits_state(self):
-        """Return the last captured CreditsState, or None."""
-        return self._credits_state
-
-    def get_credits_spent_micros(self):
-        """Session-cumulative micros spent = first_seen_remaining - current_remaining. None if no data."""
-        if self._credits_session_start_micros is None or self._credits_state is None:
-            return None
-        return self._credits_session_start_micros - self._credits_state.remaining_micros
-
-    def _check_openrouter_cache_status(self, http_response: Any) -> None:
-        """Read X-OpenRouter-Cache-Status from response headers and log it; HITs count in ``_or_cache_hits``."""
-        if http_response is None:
-            return
-        headers = getattr(http_response, "headers", None)
-        if not headers:
-            return
-        try:
-            status = headers.get("x-openrouter-cache-status")
-            if not status:
-                return
-            if status.upper() == "HIT":
-                self._or_cache_hits += 1
-                logger.info("OpenRouter response cache HIT (total: %d)", self._or_cache_hits)
-            else:
-                logger.debug("OpenRouter response cache %s", status.upper())
-        except Exception:
-            pass  # Never let header parsing break the agent loop
-
     def get_activity_summary(self) -> dict:
         """Return a snapshot of the agent's current activity for diagnostics.
 
diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py
index 05af3b8d1d..b687169141 100644
--- a/tests/run_agent/test_run_agent.py
+++ b/tests/run_agent/test_run_agent.py
@@ -872,7 +872,7 @@ class TestInit:
 
 class TestInterrupt:
     def test_interrupt_sets_flag(self, agent):
-        with patch("run_agent._set_interrupt"):
+        with patch("run_agent._set_interrupt"), patch("agent.interrupt_control._set_interrupt"):
             agent.interrupt()
             assert agent._interrupt_requested is True
 
@@ -900,7 +900,7 @@ class TestHydrateTodoStore:
             {"role": "user", "content": "hello"},
             {"role": "assistant", "content": "hi"},
         ]
-        with patch("run_agent._set_interrupt"):
+        with patch("run_agent._set_interrupt"), patch("agent.interrupt_control._set_interrupt"):
             agent._hydrate_todo_store(history)
         assert not agent._todo_store.has_items()
 
@@ -925,7 +925,7 @@ class TestHydrateTodoStore:
             },
         ]
 
-        with patch("run_agent._set_interrupt"):
+        with patch("run_agent._set_interrupt"), patch("agent.interrupt_control._set_interrupt"):
             agent._hydrate_todo_store(history)
 
         assert agent._todo_store.snapshot()["revision"] == 5
@@ -948,7 +948,7 @@ class TestHydrateTodoStore:
             },
         ]
 
-        with patch("run_agent._set_interrupt"):
+        with patch("run_agent._set_interrupt"), patch("agent.interrupt_control._set_interrupt"):
             agent._hydrate_todo_store(history)
 
         assert agent._todo_store.snapshot()["revision"] == 2
@@ -1778,6 +1778,7 @@ class TestExecuteToolCalls:
         with (
             patch("run_agent.handle_function_call", side_effect=KeyboardInterrupt),
             patch("run_agent._set_interrupt"),
+            patch("agent.interrupt_control._set_interrupt"),
             pytest.raises(KeyboardInterrupt),
         ):
             agent._execute_tool_calls_sequential(mock_msg, messages, "task-1")
@@ -1806,7 +1807,7 @@ class TestExecuteToolCalls:
             lambda hook_name, **kwargs: hook_calls.append((hook_name, kwargs)) or [],
         )
 
-        with patch("run_agent._set_interrupt"):
+        with patch("run_agent._set_interrupt"), patch("agent.interrupt_control._set_interrupt"):
             agent.interrupt()
 
         agent._execute_tool_calls(mock_msg, messages, "task-1")
@@ -3423,6 +3424,7 @@ class TestRunConversation:
             patch.object(agent, "_save_trajectory"),
             patch.object(agent, "_cleanup_task_resources"),
             patch("run_agent._set_interrupt"),
+            patch("agent.interrupt_control._set_interrupt"),
             patch.object(
                 agent, "_interruptible_api_call", side_effect=interrupt_side_effect
             ),