hermes-agent/gateway/run_turn_hygiene.py

"""Gateway turn hygiene phases; inherited by GatewayTurnMixin."""
from __future__ import annotations

import logging
from typing import TYPE_CHECKING
import asyncio
import dataclasses
import inspect
import json
import os
import queue
import threading
import time
from agent.i18n import t
from agent.session_activity import format_iteration_progress
from contextlib import nullcontext, suppress
from contextvars import copy_context
from gateway.config import Platform
from gateway.media_repair import repair_explicit_computer_use_media_paths
from gateway.platforms.base import BasePlatformAdapter
from gateway.platforms.event import MessageEvent
from gateway.session import (
    SessionSource, _session_key_namespace, build_channel_continuity_note,
    build_session_context,
)
from gateway.session_transcript import TranscriptReadError
from gateway.turn_context import TurnContext
from gateway.turn_lease import DEFAULT_LEASE_WAIT, TurnLeaseTimeoutError
from hermes_constants import get_hermes_home_override
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Tuple
from utils import base_url_hostname

if TYPE_CHECKING:  # string annotations only; never imported at runtime (cycle)
    from gateway.run import GatewayRunner
    from gateway.run_turn_runner import TurnRunner

# Log-record parity with the origin module.
logger = logging.getLogger("gateway.run")


class GatewayTurnHygieneMixin:
    @dataclasses.dataclass
    class _HygienePlan:
        """Hygiene pre-check outcome for one turn."""

        needs_compress: bool
        approx_tokens: int
        msg_count: int
        warn_token_threshold: int

    @staticmethod
    def _hmwa_hygiene_read_config(hs, data):
        """Apply model / compression knobs from the gateway config onto ``hs`` (invalid values keep the defaults)."""
        # Resolve model name (same logic as run_sync)
        _model_cfg = data.get("model", {})
        if isinstance(_model_cfg, str):
            hs.model = _model_cfg
        elif isinstance(_model_cfg, dict):
            hs.model = _model_cfg.get("default") or _model_cfg.get("model") or hs.model
            _raw_ctx = _model_cfg.get("context_length")
            if _raw_ctx is not None:
                with suppress(TypeError, ValueError):
                    hs.config_context_length = int(_raw_ctx)
            hs.provider = _model_cfg.get("provider") or None
            hs.base_url = _model_cfg.get("base_url") or None

        # Only the enabled flag is shared with the agent's compression config (hygiene runs higher).
        _comp_cfg = data.get("compression", {})
        if not isinstance(_comp_cfg, dict):
            return
        hs.compression_enabled = str(_comp_cfg.get("enabled", True)).lower() in {"true", "1", "yes"}

        def _knob(key, current, cast, allow_zero=False):
            raw = _comp_cfg.get(key)
            if raw is None:
                return current
            try:
                parsed = cast(raw)
            except (TypeError, ValueError):
                return current
            return parsed if (parsed >= 0 if allow_zero else parsed > 0) else current

        hs.hard_msg_limit = _knob("hygiene_hard_message_limit", hs.hard_msg_limit, int)
        hs.timeout_seconds = _knob("hygiene_timeout_seconds", hs.timeout_seconds, float)
        hs.total_ceiling_seconds = _knob("hygiene_total_ceiling_seconds", hs.total_ceiling_seconds, float)
        # The ceiling can never be tighter than one idle window, or the extension loop would be dead code.
        hs.total_ceiling_seconds = max(hs.total_ceiling_seconds, hs.timeout_seconds)
        hs.max_turn_hold_seconds = _knob("hygiene_max_turn_hold_seconds", hs.max_turn_hold_seconds, float)
        hs.failure_cooldown_seconds = _knob(
            "hygiene_failure_cooldown_seconds", hs.failure_cooldown_seconds, float, allow_zero=True,
        )

    async def _hmwa_hygiene_settings(self, source, session_key):
        """Resolve model/provider/context-length + hygiene knobs (fail-soft: errors keep defaults).

        The 0.85 threshold is deliberately HIGHER than the agent's compressor (0.50): a safety net
        for sessions that grew between turns. ``max_turn_hold_seconds`` bounds the TURN wait
        (compressor keeps running detached, commit fenced); kept below transport idle-timeouts."""
        from gateway.run import _load_gateway_config
        hs = self._HygieneSettings(
            model="anthropic/claude-sonnet-4.6", threshold_pct=0.85, compression_enabled=True,
            hard_msg_limit=5000, timeout_seconds=30.0, total_ceiling_seconds=600.0,
            max_turn_hold_seconds=10.0, failure_cooldown_seconds=300.0, config_context_length=None,
            provider=None, base_url=None, api_key=None, data={},
        )
        try:
            hs.data = _load_gateway_config()
            if hs.data:
                self._hmwa_hygiene_read_config(hs, hs.data)
            configured_model, configured_provider, configured_base_url = hs.model, hs.provider, hs.base_url

            with suppress(Exception):
                hs.model, _hyg_runtime = self._resolve_session_agent_runtime(
                    source=source, session_key=session_key,
                    user_config=hs.data if isinstance(hs.data, dict) else None,
                )
                hs.provider = _hyg_runtime.get("provider") or hs.provider
                hs.base_url = _hyg_runtime.get("base_url") or hs.base_url
                hs.api_key = _hyg_runtime.get("api_key") or hs.api_key

            if hs.config_context_length is not None:
                try:
                    from hermes_cli.route_identity import should_clear_context_pin_async

                    if await should_clear_context_pin_async(
                        configured_model, hs.model, configured_base_url, hs.base_url,
                        configured_provider, hs.provider,
                    ):
                        hs.config_context_length = None
                except Exception:
                    hs.config_context_length = None

            # custom_providers per-model context_length fallback (as in run_agent.py); needs base_url.
            if hs.config_context_length is None and hs.base_url:
                with suppress(TypeError, ValueError):
                    try:
                        from hermes_cli.config import (
                            get_compatible_custom_providers as _gw_gcp,
                            get_custom_provider_context_length as _gw_gccl,
                        )
                        _hyg_custom_providers = _gw_gcp(hs.data)
                    except Exception:
                        _hyg_custom_providers = hs.data.get("custom_providers")
                        if not isinstance(_hyg_custom_providers, list):
                            _hyg_custom_providers = []
                    _hyg_custom_ctx = _gw_gccl(
                        model=hs.model, base_url=hs.base_url, custom_providers=_hyg_custom_providers,
                    )
                    if _hyg_custom_ctx:
                        hs.config_context_length = int(_hyg_custom_ctx)
        except Exception:
            pass
        return hs

    async def _hmwa_hygiene_plan(self, hs, history, session_entry, session_key):
        """Decide whether hygiene compression fires this turn (token/message thresholds, DB-backed
        failure cooldown, in-flight compression)."""
        from agent.model_metadata import estimate_messages_tokens_rough, get_model_context_length_async
        _hyg_context_length = await get_model_context_length_async(
            hs.model, base_url=hs.base_url or "", api_key=hs.api_key or "",
            config_context_length=hs.config_context_length, provider=hs.provider or "",
        )
        _compress_token_threshold = int(_hyg_context_length * hs.threshold_pct)
        _warn_token_threshold = int(_hyg_context_length * 0.95)
        _msg_count = len(history)

        # Real usage decides: the API-reported prompt count, else the anchor persisted on the session
        # row (real count + delta of what was appended since, survives gateway restarts), else the
        # rough estimate (runs 30-50% high, which only fires hygiene early — safe). Do NOT compensate
        # with a threshold multiplier.
        from agent.image_token_cost import image_cost_context, learned_image_token_cost
        _anchored = None
        # Images in any local delta/estimate are priced at the cost learned from this model's usage.
        with image_cost_context(learned_image_token_cost(hs.model, hs.base_url)):
            if session_entry.last_prompt_tokens <= 0:
                from agent.usage_anchor import persisted_anchor_tokens
                _session_db = getattr(self, "_session_db", None)
                _anchored = persisted_anchor_tokens(
                    getattr(_session_db, "_db", _session_db), session_entry.session_id, history,
                )
            if session_entry.last_prompt_tokens > 0:
                _approx_tokens, _token_source = session_entry.last_prompt_tokens, "actual"
            elif _anchored is not None:
                _approx_tokens, _token_source = _anchored, "anchored"
            else:
                _approx_tokens, _token_source = estimate_messages_tokens_rough(history), "estimated"

        # Hard safety valve: force compression at an extreme message count regardless of tokens,
        # breaking the disconnect → no token data → no compression spiral. 5000 clears 1M+ sessions.
        _needs_compress = _approx_tokens >= _compress_token_threshold or _msg_count >= hs.hard_msg_limit

        if _needs_compress:
            # DB-backed cooldown (shared with context_compressor.py): survives gateway restarts, so a
            # failing compression is not re-triggered on every restart.
            # The in-memory dict was reset on every restart, re-triggering the same failing compression and
            # wedging session storage (#74136).
            _session_db = getattr(self, "_session_db", None)
            _getter = getattr(getattr(_session_db, "_db", _session_db), "get_compression_failure_cooldown", None)
            if _getter is not None:
                _cooldown_state = None
                with suppress(Exception):
                    _cooldown_state = _getter(session_entry.session_id)
                if _cooldown_state and _cooldown_state.get("remaining_seconds", 0) > 0:
                    logger.info(
                        "Session hygiene: skipping compression for %s; "
                        "previous failure cooldown active for %.1fs",
                        session_entry.session_id, _cooldown_state["remaining_seconds"],
                    )
                    _needs_compress = False

        if _needs_compress and await self._session_has_compression_in_flight(session_key):
            # A prior compression still holds the durable lock (e.g. a shielded worker left by /stop):
            # another attempt would wait up to 600s behind a commit the fence will refuse.
            logger.info(
                "Session hygiene: skipping compression for %s; "
                "another compression is already in flight", session_entry.session_id,
            )
            _needs_compress = False

        if _needs_compress:
            logger.info(
                "Session hygiene: %s messages, ~%s tokens (%s) — auto-compressing "
                "(threshold: %s%% of %s = %s tokens)",
                _msg_count, f"{_approx_tokens:,}", _token_source,
                int(hs.threshold_pct * 100), f"{_hyg_context_length:,}", f"{_compress_token_threshold:,}",
            )
        return self._HygienePlan(_needs_compress, _approx_tokens, _msg_count, _warn_token_threshold)

    async def _hmwa_hygiene_wait_for_summary(self, attempt, hs, session_entry):
        """Progress-aware inline wait for the detached hygiene compressor. Returns the compressed
        transcript; raises ``HygieneTurnHoldExceeded`` (turn-hold budget) or
        ``asyncio.TimeoutError`` (idle/ceiling/fence cancel) for the caller's handlers. Idle timeout
        (fence ticks per streamed token) + hard ceiling (moved back by prompt reading) + turn-hold cap."""
        from agent.conversation_compression_deadline import effective_ceiling
        from gateway.run import HygieneTurnHoldExceeded, hygiene_wait_should_extend
        fence = attempt.commit_fence
        while True:
            if fence.is_cancelled:
                raise TimeoutError
            # Charge the idle budget from the LAST PROGRESS event, else silence can approach 2x timeout.
            _hyg_waited = time.monotonic() - attempt.wait_started
            _ceiling = effective_ceiling(fence, hs.total_ceiling_seconds, attempt.wait_started)
            _slice = min(max(hs.timeout_seconds - fence.seconds_since_progress(), 0.005),
                         max(_ceiling - _hyg_waited, 0.005))
            # Cap the slice at the remaining turn-hold budget so a continuously-streaming worker can't
            # hold the turn until the ceiling. Budget exhausted → immediate timeout → abandonment.
            _turn_hold_remaining = hs.max_turn_hold_seconds - (time.monotonic() - attempt.wait_started)
            _slice = 0.005 if _turn_hold_remaining <= 0 else min(_slice, max(_turn_hold_remaining, 0.005))
            # Short poll so a /stop or /restart cancel is not stuck behind a full idle window.
            _idle_left = max(hs.timeout_seconds - fence.seconds_since_progress(), 0.005)
            _slice = min(_slice, 0.25)
            try:
                _compressed, _ = await asyncio.wait_for(asyncio.shield(attempt.future), timeout=_slice)
                return _compressed
            except TimeoutError:
                if fence.is_cancelled:
                    raise
                _hyg_waited = time.monotonic() - attempt.wait_started
                _ceiling = effective_ceiling(fence, hs.total_ceiling_seconds, attempt.wait_started)
                _idle = fence.seconds_since_progress()
                # Never hold the TURN past the budget even while the summary streams: proceed on the
                # uncompressed transcript so the wire never trips a transport idle-timeout.
                if _hyg_waited >= hs.max_turn_hold_seconds:
                    logger.info(
                        "Session hygiene compression for session %s exceeded the turn-hold "
                        "budget (%.1fs >= %.1fs) — abandoning inline wait, proceeding "
                        "without compression this turn",
                        session_entry.session_id, _hyg_waited, hs.max_turn_hold_seconds,
                    )
                    raise HygieneTurnHoldExceeded(
                        f"turn-hold budget {hs.max_turn_hold_seconds:.1f}s "
                        f"elapsed after {_hyg_waited:.1f}s"
                    )
                if hygiene_wait_should_extend(
                    idle=_idle, timeout=hs.timeout_seconds, waited=_hyg_waited,
                    ceiling=_ceiling, fence_cancelled=fence.is_cancelled,
                ):
                    if _slice >= _idle_left - 1e-9:
                        logger.info(
                            "Session hygiene compression for session %s still streaming after "
                            "%.0fs (last progress %.1fs ago) — extending wait (ceiling %.0fs)",
                            session_entry.session_id, _hyg_waited, _idle, _ceiling,
                        )
                    continue
                raise

    async def _hmwa_hygiene_cancel_or_adopt(self, attempt, context):
        """Cancel the worker at the commit fence; on success release its lease and defer agent
        cleanup, returning ``None``. When the worker already crossed into its commit, consume and
        return the compressed transcript instead (a successful compaction is never a timeout; the
        turn may be held past the budget by up to the commit duration — by design). The lock-free
        ``commit_in_flight`` marker keeps the poll from spinning on a hung commit."""
        fence = attempt.commit_fence
        while not fence.commit_in_flight:
            cancelled = fence.try_cancel_before_commit()
            if cancelled is None:
                await asyncio.sleep(0.025)
            elif cancelled:
                fence.release_cancelled_compression_lock()
                self._hmwa_hygiene_defer_cleanup(attempt, context)
                return None
            else:
                break
        _compressed, _ = await attempt.future
        return _compressed

    def _hmwa_hygiene_defer_cleanup(self, attempt, context):
        """Hand the agent's cleanup to the still-running worker future and mark it deferred."""
        self._defer_agent_cleanup_until_future_done(attempt.future, attempt.agent, context=context)
        attempt.cleanup_deferred = True

    @staticmethod
    def _hmwa_hygiene_stamp(agent, desc, provenance_name, debug_label):
        from agent.session_activity import ActivityProvenance
        from gateway.run import _stamp_hygiene_compression_provenance
        _stamp_hygiene_compression_provenance(agent, desc, getattr(ActivityProvenance, provenance_name), debug_label)

    async def _hmwa_hygiene_notify(self, source, meta, message, what):
        """Best-effort user notice on the hygiene thread; failure is logged, never raised."""
        from gateway.bot_peer_notices import automatic_notices_allowed
        try:
            _adapter = self._delivery_adapter_for(source) if automatic_notices_allowed(source) else None
            if _adapter and source.chat_id:
                await _adapter.emit_warning(source.chat_id, message, metadata=meta,
                                            logical_platform=source.platform)
        except Exception as _werr:
            logger.warning("Failed to deliver %s to user: %s", what, _werr)

    async def _hmwa_hygiene_record_failure_cooldown(self, hs, session_key, session_id, reason):
        """Escalate the failure streak (off-loop) and persist the cooldown, when enabled."""
        from gateway.run import _hygiene_cooldown_for_failure, _record_hygiene_cooldown
        if hs.failure_cooldown_seconds < 0:
            return
        _hyg_cooldown = await asyncio.to_thread(
            _hygiene_cooldown_for_failure, self, session_key, hs.failure_cooldown_seconds,
        )
        _record_hygiene_cooldown(self, session_id, _hyg_cooldown, reason)

    async def _hmwa_hygiene_on_turn_hold(self, attempt, hs, session_entry, session_key, source):
        """``except HygieneTurnHoldExceeded`` body: keep or cancel the worker's commit admission,
        notify the user, and re-raise; returns the compressed transcript only when the worker
        was already committing.

        Turn-hold expiry is an availability boundary, not a failure: the streak must NOT advance,
        only flat retry spacing is recorded. A watermark-fenced commit (rows appended after
        compression start survive as cloned tail) KEEPS admission: the turn proceeds uncompressed
        now and the summary is adopted at the worker's fenced commit — always cancelling burned
        every attempt for thinking summary models. Without the fence a late commit could clobber
        newer turns, so cancel."""
        from gateway.run import (
            _HYGIENE_TURNHOLD_RETRY_SECONDS, _record_hygiene_cooldown, _reset_hygiene_failure_streak
        )
        fence = attempt.commit_fence
        _hyg_keep_admission = bool(getattr(fence, "commit_watermark_fenced", False)) and not fence.is_cancelled
        if _hyg_keep_admission:
            self._hmwa_hygiene_defer_cleanup(attempt, "session hygiene turn-hold")
            # NO retry-after here (it would also block the agent-side preflight compressor); spacing
            # comes from the durable compression lock. The done-callback records the flat retry-after
            # ONLY if the worker ends without committing anything.
            _sid, _skey, _agent = session_entry.session_id, session_key, attempt.agent

            def _hyg_adopt_or_space_retry(_fut, _gw=self, _sid=_sid, _skey=_skey, _agent=_agent):
                try:
                    _exc = _fut.exception()
                except (asyncio.CancelledError, Exception):
                    _committed = False
                else:
                    _committed = _exc is None and (
                        bool(getattr(_agent, "_last_compaction_in_place", False))
                        or getattr(_agent, "session_id", _sid) != _sid
                    )
                if _committed:
                    logger.info(
                        "Session hygiene compression for session %s finished after the "
                        "turn-hold was released — summary adopted at the watermark-fenced "
                        "commit boundary (#97963)", _sid,
                    )
                    try:
                        _reset_hygiene_failure_streak(_gw, _skey)
                    except Exception as _rs_err:
                        logger.debug("hygiene streak reset after deferred adoption failed: %s", _rs_err)
                else:
                    # Nothing to adopt (summary failed / fence refused / superseded): flat spacing so
                    # sustained traffic doesn't spawn and abandon a compressor every turn.
                    _record_hygiene_cooldown(
                        _gw, _sid, _HYGIENE_TURNHOLD_RETRY_SECONDS,
                        "hygiene compression deferred: turn-hold budget expired and the "
                        "detached attempt did not commit",
                    )

            attempt.future.add_done_callback(_hyg_adopt_or_space_retry)
            _log_suffix = (
                " — the watermark-fenced worker keeps its commit admission and the summary "
                "will be adopted when it finishes"
            )
        else:
            _adopted = await self._hmwa_hygiene_cancel_or_adopt(attempt, "session hygiene turn-hold")
            if _adopted is not None:
                return _adopted
            # Short flat retry-after, else every turn re-spawns, holds and cancels a compressor.
            _record_hygiene_cooldown(
                self, session_entry.session_id, _HYGIENE_TURNHOLD_RETRY_SECONDS,
                "hygiene compression deferred: turn-hold budget expired while the "
                "summary was still streaming",
            )
            _log_suffix = ""
        self._hmwa_hygiene_stamp(
            attempt.agent, "session hygiene compression turn-hold",
            "AGENT_COMPRESSION_TURNHOLD", "hygiene compression turn-hold activity stamp failed",
        )
        logger.info(
            "Session hygiene compression for session %s exceeded turn-hold budget (%.1fs); "
            "proceeding without compression this turn%s",
            session_entry.session_id, time.monotonic() - attempt.wait_started, _log_suffix,
        )
        await self._hmwa_hygiene_notify(
            source, attempt.meta, t("gateway.compress.turnhold_deferred"), "compression-turnhold notice",
        )
        raise

    async def _hmwa_hygiene_on_timeout(self, attempt, hs, session_entry, session_key, source):
        """``except asyncio.TimeoutError`` body: cancel at the commit fence, record the failure
        cooldown, warn the user, and re-raise; returns the compressed transcript only when the
        worker crossed the commit boundary first."""
        from gateway.run import _hygiene_compression_timeout_message
        fence = attempt.commit_fence
        _hyg_total_exhausted = fence.deadline_exceeded  # armed with the ceiling before the wait began
        if _hyg_total_exhausted:
            # The worker checks this deadline between digest calls; keep its lease until it exits so
            # an unchanged session cannot overlap a retry (the release below is then a no-op).
            fence.retain_compression_lock_until_worker_done()
        # Capture fence state BEFORE try_cancel (which itself sets is_cancelled).
        _hyg_fence_cancelled = fence.is_cancelled
        _adopted = await self._hmwa_hygiene_cancel_or_adopt(attempt, "session hygiene timeout")
        if _adopted is not None:
            return _adopted
        await self._hmwa_hygiene_record_failure_cooldown(
            hs, session_key, session_entry.session_id,
            "session hygiene compression " + (
                "cancelled at commit fence" if _hyg_fence_cancelled
                else "total ceiling exhausted" if _hyg_total_exhausted
                else "timed out with no output from the summary model"
            ),
        )
        self._hmwa_hygiene_stamp(
            attempt.agent,
            "session hygiene compression cancelled at commit fence" if _hyg_fence_cancelled
            else "session hygiene compression timed out",
            "AGENT_COMPRESSION_TIMEOUT", "hygiene compression timeout activity stamp failed",
        )
        if _hyg_fence_cancelled:
            logger.warning(
                "Session hygiene compression for session %s was cancelled at the "
                "commit fence; continuing without compression", session_entry.session_id,
            )
            raise
        _hyg_elapsed = time.monotonic() - attempt.wait_started
        if _hyg_total_exhausted:
            logger.warning(
                "Session hygiene compression for session %s reached its total ceiling after "
                "%.1fs (progress observed=%s); continuing without compression",
                session_entry.session_id, _hyg_elapsed, fence.progress_observed,
            )
        else:
            logger.warning(
                "Session hygiene compression for session %s made no progress for %.1fs "
                "(total wait %.1fs, ceiling %.1fs); continuing without compression",
                session_entry.session_id, fence.seconds_since_progress(), _hyg_elapsed, hs.total_ceiling_seconds,
            )
        await self._hmwa_hygiene_notify(
            source, attempt.meta,
            _hygiene_compression_timeout_message(
                total_exhausted=_hyg_total_exhausted, elapsed=_hyg_elapsed,
                idle_timeout=hs.timeout_seconds, progress_observed=fence.progress_observed,
            ),
            "compression-timeout warning",
        )
        raise

    def _hmwa_hygiene_on_unwind(self, attempt, hs, session_entry, session_key):
        """``except BaseException`` body (caller re-raises): revoke commit admission BEFORE the host
        unwinds so the detached worker can never commit later, and record a cooldown — otherwise
        the next turn re-arms hygiene and waits up to 600s behind a fence that refuses again."""
        from gateway.run import _hygiene_cooldown_for_failure, _record_hygiene_cooldown
        attempt.commit_fence.revoke_commit_admission()
        if not attempt.cleanup_deferred:
            self._hmwa_hygiene_defer_cleanup(attempt, "session hygiene unwind")
        if hs.failure_cooldown_seconds >= 0:
            try:
                _record_hygiene_cooldown(
                    self, session_entry.session_id,
                    _hygiene_cooldown_for_failure(self, session_key, hs.failure_cooldown_seconds),
                    "session hygiene compression cancelled at commit fence",
                )
            except Exception as _cd_err:
                logger.debug("hygiene unwind cooldown record failed: %s", _cd_err)

    async def _hmwa_hygiene_adopt_transcript(
        self, attempt, _compressed, history, plan, *, session_entry, source, _quick_key, run_generation,
    ):
        """Adopt a finished compression (rotation / in-place / refused); publishes the transcript to
        continue with on ``attempt.history``. Returns ``(rotated, in_place, new_count, new_tokens)``.

        Rewrite only on rotation (NEW session id): in-place compaction already soft-archived the
        old rows and rewrite_transcript() would DELETE them; neither rotation nor in-place signals
        FAILURE and an unconditional rewrite would leave only the summary. Write-before-repoint:
        a repoint-then-failed-rewrite would point the live entry at an empty session."""
        from gateway.run_turn import hygiene_no_commit_reason
        from agent.model_metadata import estimate_messages_tokens_rough
        _hyg_agent = attempt.agent
        # _compress_context rotates to a NEW session_id so the old transcript stays intact/searchable.
        _hyg_new_sid = _hyg_agent.session_id
        _hyg_rotated = _hyg_new_sid != session_entry.session_id
        _hyg_in_place = bool(getattr(_hyg_agent, "_last_compaction_in_place", False))
        # Anti-growth guard: refuse a compression that did not shrink the transcript (seen 427K→598K).
        _hyg_in_toks = estimate_messages_tokens_rough(history)
        _hyg_out_toks = estimate_messages_tokens_rough(_compressed)
        if _hyg_rotated and _hyg_out_toks > _hyg_in_toks:
            logger.warning(
                "Gateway hygiene compression for session %s would grow transcript (~%s -> ~%s "
                "tokens); keeping the original transcript unchanged",
                session_entry.session_id, f"{_hyg_in_toks:,}", f"{_hyg_out_toks:,}",
            )
            _hyg_rotated = False
            _compressed = history
        # Only persist a child transcript when rotation produced a NEW session id. In-place compaction does NOT
        # need a rewrite: archive_and_compact() has already soft-archived the previous active rows and
        # inserted the compacted messages as the new active set inside _compress_context(). Calling
        # rewrite_transcript() after in-place compaction would invoke replace_messages(active_only=False)
        # which DELETEs ALL rows — including the archived turns that archive_and_compact() deliberately
        # preserved (silent data loss, #61145). The danger this guards against (mirrors the /compress fix
        # #44794/#39704): if _compress_context returns a summary but neither rotates nor completes
        # archive_and_compact(), the session_id is unchanged for a FAILURE reason, and an unconditional
        # rewrite_transcript() would DELETE the original messages and replace them with only the compressed
        # summary (permanent data loss, #21301). Write-before-repoint (mirrors manual /compress): if we
        # repointed session_entry onto the child SID and rewrite_transcript then failed (lock/ENOSPC), the
        # live entry would already reference a brand-new empty session while the turn continues — the
        # conversation silently vanishes. Persist the child transcript first; only then rebind the live
        # entry.
        if _hyg_rotated:
            # Published child is already durable; a rewrite would drop rows cloned at publish.
            if not await self.async_session_store.persist_rotated_compression_child(
                    session_entry.session_id, _hyg_new_sid, _compressed):
                logger.error(
                    "Session hygiene: failed to persist compressed transcript for rotated session "
                    "%s → %s; keeping the live entry on the original session so the "
                    "conversation is not dropped", session_entry.session_id, _hyg_new_sid,
                )
                # Fail closed: treat like no rotation.
                _hyg_rotated = False
                _hyg_in_place = False
            else:
                session_entry.session_id = _hyg_new_sid
                # The held turn lease follows the rotation (alias keys still serialize on this turn).
                self._rebind_turn_lease(_quick_key, run_generation, _hyg_new_sid)
                await self.async_session_store._save()
                await asyncio.to_thread(
                    self._sync_telegram_topic_binding, source, session_entry, reason="hygiene-compression",
                )

        if _hyg_rotated or _hyg_in_place:
            # Persisted (rotation) or persisted by archive_and_compact() (in-place): reset token count.
            session_entry.last_prompt_tokens = 0
            attempt.history = _compressed
            _new_count = len(_compressed)
            _new_tokens = estimate_messages_tokens_rough(_compressed)
        else:
            # No rewrite happened — post-compression counts equal the pre-compression ones.
            _new_count = plan.msg_count
            _new_tokens = plan.approx_tokens
            logger.warning(
                "Gateway hygiene compression for session %s did not rotate or compact in place (%s) — "
                "preserving the original transcript instead of overwriting it with the summary (#21301).",
                session_entry.session_id, hygiene_no_commit_reason(_hyg_agent),
            )

        logger.info(
            "Session hygiene: compressed %s → %s msgs, ~%s → ~%s tokens",
            plan.msg_count, _new_count, f"{plan.approx_tokens:,}", f"{_new_tokens:,}",
        )
        if _new_tokens >= plan.warn_token_threshold:
            logger.warning("Session hygiene: still ~%s tokens after compression", f"{_new_tokens:,}")
        return _hyg_rotated, _hyg_in_place, _new_count, _new_tokens

    async def _hmwa_hygiene_apply_result(
        self, attempt, hs, _compressed, history, plan, *,
        session_entry, session_key, source, _quick_key, run_generation,
    ):
        """Adopt a finished hygiene compression, rebind the session + turn lease, record
        streak/cooldown, and warn the user on abort."""
        from gateway.run import _reset_hygiene_failure_streak, hygiene_compaction_recovered
        _hyg_rotated, _hyg_in_place, _new_count, _new_tokens = await self._hmwa_hygiene_adopt_transcript(
            attempt, _compressed, history, plan, session_entry=session_entry, source=source,
            _quick_key=_quick_key, run_generation=run_generation,
        )
        # Summary failure aborts the compressor (nothing dropped). Warn the user visibly — agent.log
        # is invisible on TG/Discord — so they know the chat is "frozen" and can /compress or /reset.
        _comp = getattr(attempt.agent, "context_compressor", None)
        _hyg_aborted = _comp is not None and getattr(_comp, "_last_compress_aborted", False)
        # A fence-cancelled _compress_context returns the original transcript with
        # _last_compress_aborted False: treat that no-op as an abort so hygiene records a cooldown
        # instead of retrying into the 600s wait. A committed rotate/in-place is never an abort.
        _hyg_fence_cancelled = bool(attempt.commit_fence.is_cancelled and not _hyg_rotated and not _hyg_in_place)
        if _hyg_fence_cancelled:
            _hyg_aborted = True
        # Recovery decision lives in the unit-tested predicate: the "neither rotated nor in place"
        # path reuses pre-compression counts, so a numbers-only check would read a no-op as success.
        if not _hyg_aborted and hygiene_compaction_recovered(
            aborted=_hyg_aborted, rotated=_hyg_rotated, in_place=_hyg_in_place,
            msg_count=plan.msg_count, new_count=_new_count, approx_tokens=plan.approx_tokens,
            new_tokens=_new_tokens,
        ):
            await asyncio.to_thread(_reset_hygiene_failure_streak, self, session_key)
        if _hyg_aborted:
            await self._hmwa_hygiene_record_failure_cooldown(
                hs, session_key, session_entry.session_id,
                "session hygiene compression cancelled at commit fence" if _hyg_fence_cancelled
                else getattr(_comp, "_last_summary_error", None),
            )
            self._hmwa_hygiene_stamp(
                attempt.agent, "session hygiene compression aborted",
                "AGENT_COMPRESSION_COOLDOWN", "hygiene compression abort activity stamp failed",
            )
            if not _hyg_fence_cancelled:
                # Force-redact: provider exception text may contain credentials; this reaches users.
                from agent.redact import redact_sensitive_text
                _err = redact_sensitive_text(
                    getattr(_comp, "_last_summary_error", None) or t("gateway.shared.unknown_error"), force=True)
                logger.warning("Session hygiene compression aborted: %s", _err)
                await self._hmwa_hygiene_notify(
                    source, attempt.meta, t("gateway.compress.hygiene_failed"), "compression-failure warning",
                )
        # Configured aux model failed, recovered on the main model: only the user can fix that config.
        elif _comp is not None and getattr(_comp, "_last_aux_model_failure_model", None):
            _aux_model = getattr(_comp, "_last_aux_model_failure_model", "")
            _aux_err = getattr(_comp, "_last_aux_model_failure_error", None) or t("gateway.shared.unknown_error")
            await self._hmwa_hygiene_notify(
                source, attempt.meta, t("gateway.compress.aux_failed", model=_aux_model, error=_aux_err),
                "aux-model-fallback notice",
            )

    async def _hmwa_hygiene_codex_compaction(self, hs, plan, history, session_entry, session_key, _hyg_runtime):
        """codex app-server runtime: the real context is the server-side thread, not the transcript
        mirror. The detached-agent path would only rewrite the mirror and its finally-eviction
        would destroy the live thread (next turn starts blank), so use the cached agent's
        thread/compact/start and KEEP it cached."""
        from gateway.run import run_codex_hygiene_compaction
        # codex app-server runtime: the model's real context is the app-server's server-side thread, not the
        # transcript mirror. See #73503.
        _hyg_codex_auto = "native"
        _hyg_comp_cfg = hs.data.get("compression") if isinstance(hs.data, dict) else None
        if isinstance(_hyg_comp_cfg, dict):
            _hyg_codex_auto = str(_hyg_comp_cfg.get("codex_app_server_auto", "native") or "native")
        _hyg_codex_outcome = await run_codex_hygiene_compaction(
            self, session_key, session_entry.session_id, auto_mode=_hyg_codex_auto, history=history,
            approx_tokens=plan.approx_tokens, timeout_seconds=hs.total_ceiling_seconds,
            failure_cooldown_seconds=hs.failure_cooldown_seconds,
        )
        logger.info(
            "Session hygiene (codex app-server): %s (session=%s, mode=%s, ~%s tokens)",
            _hyg_codex_outcome, session_entry.session_id, _hyg_codex_auto, f"{plan.approx_tokens:,}",
        )

    async def _hmwa_hygiene_build_agent(self, _hyg_model, _hyg_runtime, session_entry):
        """Build the detached hygiene ``AIAgent`` with the live session's system prompt. Returns
        ``(agent, sync_session_db)``."""
        from gateway.run import _GATEWAY_HYGIENE_PLATFORM, _seed_hygiene_system_prompt
        from run_agent import AIAgent
        try:
            _hyg_session_row = await self._session_db.get_session(session_entry.session_id)
        except Exception as exc:
            _hyg_session_row = None
            logger.warning(
                "Session hygiene could not restore the system prompt for session %s: %s. "
                "Preserving an empty prompt so the live turn rebuilds it with its "
                "configured providers.", session_entry.session_id, exc, exc_info=True,
            )
        _hyg_session_db = getattr(self._session_db, "_db", self._session_db)
        # With compression.checkpoint_required on, load the memory provider so the checkpoint exists
        # before any mutation; otherwise keep the fast path (no provider init).
        from hermes_cli.config import load_config as _load_cfg
        from utils import is_truthy_value as _is_truthy

        _hyg_checkpoint_required = _is_truthy(
            ((_load_cfg() or {}).get("compression") or {}).get("checkpoint_required"), default=False,
        )
        _hyg_agent = AIAgent(
            **_hyg_runtime, model=_hyg_model, max_iterations=4, quiet_mode=True,
            skip_memory=not _hyg_checkpoint_required, enabled_toolsets=["memory"],
            session_id=session_entry.session_id, session_db=_hyg_session_db,
        )
        _seed_hygiene_system_prompt(_hyg_agent, _hyg_session_row)
        # The stamp only marks this agent as no real surface. Since #104414 Platform is not a
        # restore-identity field, so it no longer forces the next live turn to rebuild; the seed's
        # retain flag is what keeps the reduced-toolset build out of the session row (#122822).
        _hyg_agent.platform = _GATEWAY_HYGIENE_PLATFORM
        return _hyg_agent, _hyg_session_db

    async def _hmwa_hygiene_detached_attempt(
        self, attempt, hs, plan, history, _hyg_msgs, _hyg_model, _hyg_runtime,
        source, session_entry, session_key, _quick_key, run_generation,
    ):
        """Run one detached hygiene compression attempt end to end; publishes the transcript to
        continue with (compressed or original) on ``attempt.history``."""
        from gateway.run import HygieneTurnHoldExceeded
        from agent.conversation_compression import CompressionCommitFence
        _hyg_agent, _hyg_session_db = await self._hmwa_hygiene_build_agent(_hyg_model, _hyg_runtime, session_entry)
        attempt.agent = _hyg_agent
        try:
            # Hygiene owns the session binding, so prefer in-place compaction over minting a
            # continuation child. Without a SessionDB this stays False.
            _hyg_agent.compression_in_place = True
            _bind_hyg_state = getattr(getattr(_hyg_agent, "context_compressor", None), "bind_session_state", None)
            if callable(_bind_hyg_state):
                _bind_hyg_state(_hyg_session_db, session_entry.session_id)
            # Never finalize on close() — that would end the live gateway session row.
            _hyg_agent._end_session_on_close = False
            _hyg_agent._print_fn = lambda *a, **kw: None

            loop = asyncio.get_running_loop()
            _hyg_commit_fence = CompressionCommitFence(total_ceiling_seconds=hs.total_ceiling_seconds)
            # Default executor (NOT self._get_executor): a hung summary must never occupy an
            # agent-work slot. MUST run in the caller's contextvars (multiplex secret scope).
            attempt.commit_fence = _hyg_commit_fence
            attempt.future = loop.run_in_executor(
                None,
                # But it MUST run inside the caller's contextvars: under multiplex_profiles the profile
                # secret scope / HERMES_HOME override live in ContextVars, and a bare run_in_executor worker
                # starts with an empty Context — the summary model's get_secret(<PROVIDER>_API_KEY) then
                # fails closed (UnscopedSecretError) and every hygiene compaction silently degrades to a
                # lossy truncation (#100849 bundle).
                copy_context().run,
                # task_id = the live turn's dedup bucket (session row id), never "" (= reset every task).
                lambda: _hyg_agent._compress_context(
                    _hyg_msgs, "", approx_tokens=plan.approx_tokens, commit_fence=_hyg_commit_fence,
                    task_id=session_entry.session_id or "default", trigger="gateway_hygiene",
                ),
            )
            # Register the live worker with shutdown NOW, not only once it is deferred: the default
            # executor is outside self._executor's quiesce, so an untracked in-flight summary would let
            # stop() close/checkpoint state.db under its late write (mirrors run_codex_hygiene_compaction).
            self._track_deferred_agent_worker(attempt.future, _hyg_agent)
            attempt.wait_started = time.monotonic()
            try:
                _compressed = await self._hmwa_hygiene_wait_for_summary(attempt, hs, session_entry)
            except HygieneTurnHoldExceeded:
                _compressed = await self._hmwa_hygiene_on_turn_hold(attempt, hs, session_entry, session_key, source)
            except TimeoutError:
                _compressed = await self._hmwa_hygiene_on_timeout(attempt, hs, session_entry, session_key, source)
            except BaseException:
                self._hmwa_hygiene_on_unwind(attempt, hs, session_entry, session_key)
                raise

            await self._hmwa_hygiene_apply_result(
                attempt, hs, _compressed, history, plan, session_entry=session_entry,
                session_key=session_key, source=source, _quick_key=_quick_key,
                run_generation=run_generation,
            )
        finally:
            # Evict the cached agent so the next turn rebuilds its system prompt.
            self._evict_cached_agent(session_key)
            if not attempt.cleanup_deferred:
                await self._cleanup_agent_resources_off_loop(_hyg_agent, context="session hygiene")

    async def _hmwa_run_session_hygiene(
        self, event, source, session_entry, session_key, history, _quick_key, run_generation,
    ):
        """Auto-compress pathologically large transcripts before the agent starts so oversized
        histories don't cause repeated truncation/context failures. Token source: the API's
        prompt_tokens from the last turn, else a char/4 estimate."""
        from gateway.run import HygieneTurnHoldExceeded
        if not history or len(history) < 4:
            return history

        hs = await self._hmwa_hygiene_settings(source, session_key)
        # Hygiene can never land with compression disabled; a sub-limit transcript is the identity (#111988).
        if not hs.compression_enabled:
            return self._bound_hygiene_payload(history, hs, session_entry)
        plan = await self._hmwa_hygiene_plan(hs, history, session_entry, session_key)
        # No compression this turn (under both thresholds, cooldown, or one already in flight): without
        # the bound the model would get the full uncompressed transcript.
        if not plan.needs_compress:
            return self._bound_hygiene_payload(history, hs, session_entry)

        attempt = self._HygieneAttempt(agent=None, meta=self._event_thread_metadata(event, source), history=history)
        try:
            _hyg_model, _hyg_runtime = self._resolve_session_agent_runtime(
                source=source, session_key=session_key,
                user_config=hs.data if isinstance(hs.data, dict) else None,
            )
            if str(_hyg_runtime.get("api_mode") or "").lower() == "codex_app_server":
                await self._hmwa_hygiene_codex_compaction(hs, plan, history, session_entry, session_key, _hyg_runtime)
            elif _hyg_runtime.get("api_key"):
                # Pass the FULL transcript (tool results included) as the agent loop does: filtering
                # to user/assistant starved the compressor (tool results are the bulk of context).
                _hyg_msgs = [m for m in history if m.get("role") in {"user", "assistant", "tool"}]
                if len(_hyg_msgs) >= 4:
                    await self._hmwa_hygiene_detached_attempt(
                        attempt, hs, plan, history, _hyg_msgs, _hyg_model, _hyg_runtime,
                        source, session_entry, session_key, _quick_key, run_generation,
                    )
        except HygieneTurnHoldExceeded:
            # Availability boundary, not a failure — already logged at INFO by the turn-hold handler.
            # Must not hit the generic "auto-compress failed" warning below: that log is how thinking-model
            # deployments read as permanently broken (#97963; surfaced by @686f6c61 in PR #99657).
            pass
        except Exception as e:
            logger.warning("Session hygiene auto-compress failed: %s", e)
        # A landed compression published a NEW transcript on attempt.history: leave it byte-identical.
        # Anything else (turn-hold, timeout, unwind, codex path) left the FULL uncompressed transcript
        # there — that is the fail-closed case (#111988).
        if attempt.history is history:
            return self._bound_hygiene_payload(history, hs, session_entry)
        return attempt.history

    @staticmethod
    def _bound_hygiene_payload(history, hs, session_entry):
        """``bound_model_input_without_hygiene`` over ``hs.hard_msg_limit``, with one INFO line when the
        cut is real. Below the limit this is the identity — no allocation, no behaviour change."""
        from gateway.run_turn import bound_model_input_without_hygiene
        bounded = bound_model_input_without_hygiene(history, hs.hard_msg_limit)
        if bounded is not history:
            logger.info(
                "Session hygiene did not land for %s: bounding the model payload to %s of %s "
                "messages (hard limit %s) — the stored transcript is unchanged",
                session_entry.session_id, len(bounded), len(history), hs.hard_msg_limit,
            )
        return bounded