hermes-agent/gateway/session_transcript.py

"""SessionStore transcript I/O: SQLite append with a per-session retry queue, compression-reroute
following, FTS corruption recovery, rewrite/rewind/load. Mixin split out of ``gateway/session.py``;
bound onto ``SessionStore`` via the MRO."""

from __future__ import annotations

import contextlib
import logging
import threading
import time
from agent.turn_context import extract_api_content_sidecar
from typing import TYPE_CHECKING, Any, Dict, List, Optional

if TYPE_CHECKING:
    from gateway.session import SessionEntry

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


class TranscriptReadError(RuntimeError):
    """Raised when persisted history cannot be read safely."""

    def __init__(self, session_id: str) -> None:
        self.session_id = session_id
        super().__init__(f"transcript read failed for session {session_id}")


def _spool_dropped(session_id: str, message: Dict[str, Any]):
    """Spool one evicted/undeliverable message to disk (same machinery as the shutdown flush, so it
    is replayed after DB recovery); path or None."""
    try:
        from gateway.shutdown_flush import spool_dropped_transcript_message
        return spool_dropped_transcript_message(session_id, message)
    except Exception:
        return None


# Message keys persisted only for assistant rows (None otherwise).
_ASSISTANT_ONLY_KEYS = (
    "reasoning", "reasoning_content", "reasoning_details", "codex_reasoning_items",
    "codex_message_items",
)


class SessionTranscriptMixin:
    """SessionStore transcript I/O: SQLite append with a per-session retry queue,
    compression-reroute following, FTS corruption recovery, rewrite/rewind/load."""

    _MAX_PENDING_PER_SESSION = 200  # in-memory pending messages per session (DB broken)
    # Cooldown between FTS5 rebuild attempts (see _rebuild_fts_once); avoids permanently
    # disabling recovery after one failed attempt while still avoiding a rebuild storm against
    # a database that is corrupt on every write.
    _FTS_REBUILD_COOLDOWN_SECONDS = 300
    # Consecutive transcript-append failures for one session before escalating from WARNING to
    # ERROR (see _append_to_transcript_serialized); a session stalled past this many attempts is
    # no longer a transient blip and needs operator attention.
    _TRANSCRIPT_APPEND_FAILURE_ESCALATION_THRESHOLD = 3

    def _compression_tip_for_session_id(self, session_id: Optional[str]) -> Optional[str]:
        """Latest compression continuation for *session_id* (heals a mapping left pointing at a
        compressed parent by a restart or failed send)."""
        db = self._db_for_session_id(session_id) if session_id else None
        if db is None:
            return session_id
        try:
            return db.get_compression_tip(session_id) or session_id
        except Exception:
            logger.debug("Compression-tip lookup failed for session %s", session_id, exc_info=True)
            return session_id

    def _heal_compression_tip_locked(
        self, entry: "SessionEntry", original_session_id: Optional[str],
        canonical_session_id: Optional[str]) -> bool:
        """Rewrite *entry* to the compression continuation if stale. Lock held."""
        if not original_session_id or not canonical_session_id:
            return False
        if entry.session_id != original_session_id or canonical_session_id == original_session_id:
            return False
        logger.info(
            "SessionStore healed compressed session mapping: %s -> %s", entry.session_id,
            canonical_session_id)
        entry.session_id = canonical_session_id
        return True

    def advance_compression_session(
        self, session_key: str, expected_session_id: str, target_session_id: str,
    ) -> Optional[SessionEntry]:
        """CAS-advance one route along an already-verified compression lineage. Unlike
        ``switch_session`` this never ends/reopens SQLite rows (the compression transaction owns
        that). ``None`` means the route moved after the caller's snapshot (e.g. /new) — caller
        must fail closed."""
        if not session_key or not expected_session_id or not target_session_id:
            return None
        with self._lock:
            entry = self._entry_locked(session_key)
            if entry is None:
                return None
            if entry.session_id == target_session_id:
                return entry
            if not self._heal_compression_tip_locked(entry, expected_session_id, target_session_id):
                return None  # route moved (session_id != expected) or nothing to heal
            self._save()  # bookkeeping, not user activity: leave ``updated_at`` alone
            return entry

    def _get_transcript_drain_lock(self):
        """Return the lock that serializes pending-queue drain boundaries."""
        return self._lazy("_transcript_drain_lock", threading.RLock)

    def append_to_transcript(self, session_id: str, message: Dict[str, Any], skip_db: bool = False) -> None:
        """Serialize transcript draining across queue migration boundaries. A session with no usable
        store is NOT skipped: the write is queued and counted like any other failed append, so a
        dead/unopenable state.db escalates and spools instead of dropping turns silently
        (#114266)."""
        if skip_db:
            return
        with self._get_transcript_drain_lock():
            self._append_to_transcript_serialized(self._follow_reroutes(session_id), message)

    def _follow_reroutes(self, session_id: str) -> str:
        """Follow the compression reroute chain (cycle-guarded)."""
        reroutes = self._lazy("_transcript_reroutes", dict)
        seen = set()
        while session_id in reroutes and session_id not in seen:
            seen.add(session_id)
            session_id = reroutes[session_id]
        return session_id

    def _enqueue_transcript_message(self, session_id: str, message: Dict[str, Any]) -> list:
        """Queue *message* (retry lock held); evicts + spools the oldest past the cap."""
        pending = self._dirty_transcripts.setdefault(session_id, [])
        pending.append(dict(message))
        # Cap pending messages per session to avoid unbounded memory growth when the DB is persistently
        # broken. Spool the evicted oldest message to the on-disk pending spool (same machinery
        # flush_pending_to_file uses at shutdown) so a runtime cap rotation does not silently discard it
        # (#78182); it is replayed on the next successful transcript flush.
        if len(pending) > self._MAX_PENDING_PER_SESSION:
            spool_path = _spool_dropped(session_id, pending.pop(0))
            if spool_path is not None:
                self._lazy("_spooled_drop_sessions", set).add(session_id)
                logger.warning(
                    "Session DB transcript pending queue full for %s (cap=%d); spooled oldest "
                    "message to %s for replay after DB recovery", session_id,
                    self._MAX_PENDING_PER_SESSION, spool_path)
            else:
                logger.warning(
                    "Session DB transcript pending queue full for %s (cap=%d); dropping oldest "
                    "message to make room (on-disk spool unavailable)", session_id,
                    self._MAX_PENDING_PER_SESSION)
        return pending

    def _divert_transcript_after_db_replaced(
        self, session_id: str, queue_session_id: str, exc: Exception
    ) -> None:
        """Stop SQLite writes on a replaced/quarantined handle and divert the backlog to the on-disk
        spool + JSONL fallback: retrying cannot succeed and the FTS rebuild must never run here."""
        logger.error(
            "Session DB refused further writes on this handle for %s (%s); stopping SQLite writes "
            "and diverting pending transcripts to the on-disk fallback: %s", session_id,
            type(exc).__name__, exc)
        with self._transcript_retry_lock:
            remaining = list(self._dirty_transcripts.get(queue_session_id, []))
            self._dirty_transcripts.pop(queue_session_id, None)
            self._transcript_append_failures.pop(session_id, None)
        for dropped in remaining:
            if _spool_dropped(session_id, dropped) is None:
                logger.warning(
                    "pending fallback failed for replaced state.db transcript on %s", session_id,
                    exc_info=True)
        try:
            from hermes_state import divert_session_transcript_jsonl
            divert_session_transcript_jsonl(session_id, remaining)
        except Exception:
            logger.warning(
                "JSONL divert failed for replaced state.db transcript on %s", session_id,
                exc_info=True)

    def _live_compression_child(self, session_id: str) -> str:
        """Transitive compression tip of *session_id* if it is a different, still-live row, else ""
        (a depth-1 lookup misses multi-hop lineages). Uses the PARENT's proven owner handle: the
        child's id is unpublished until its write succeeds, so a by-id lookup would hit the ambient
        store."""
        owner_db = self._db_for_session_id(session_id)
        tip = owner_db.get_compression_tip(session_id) if owner_db is not None else None
        if tip and tip != session_id:
            tip_row = owner_db.get_session(tip)
            if tip_row is not None and tip_row.get("ended_at") is None:
                return str(tip)
        return ""

    def _migrate_transcript_queue_to_child(
        self, session_id: str, queue_session_id: str, child_id: str, pending: list, msg
    ) -> list:
        """Move the retry queue + failure counter from parent to child and record the reroute (retry
        lock held); returns the child's pending list. Older parent backlog must precede messages
        already queued directly on the child; routing is published only AFTER the queue moved
        (caller), so new child writes cannot bypass older parent backlog."""
        if pending and pending[0] is msg:
            pending.pop(0)
        existing_child_pending = self._dirty_transcripts.get(child_id, [])
        if pending:
            pending.extend(existing_child_pending)
            self._dirty_transcripts[child_id] = pending
        elif existing_child_pending:
            pending = existing_child_pending
        self._dirty_transcripts.pop(queue_session_id, None)
        previous_failures = self._transcript_append_failures.pop(queue_session_id, 0)
        if previous_failures:
            self._transcript_append_failures[child_id] = max(
                previous_failures, self._transcript_append_failures.get(child_id, 0))
        self._transcript_reroutes[session_id] = child_id
        return pending

    def _publish_transcript_reroute(self, session_id: str, child_id: str) -> None:
        """Repoint every route at the compression child and save (index authoritative again)."""
        with self._lock:
            for entry in self._entries.values():
                if entry.session_id == session_id:
                    entry.session_id = child_id
            self._save()
        (getattr(self, "_session_owner_hints", None) or {}).pop(child_id, None)

    def _append_to_transcript_serialized(self, session_id: str, message: Dict[str, Any]) -> None:
        """Append a message to a session's transcript (SQLite), draining the per-session retry
        queue.

        Args: skip_db: When True, skip the SQLite write. Used when the agent already persisted messages to
        SQLite via its own _flush_messages_to_session_db(), preventing the duplicate-write bug (#860).
        """
        with self._transcript_retry_lock:
            pending = self._enqueue_transcript_message(session_id, message)
            msg = pending[0]
        queue_session_id = session_id

        def _ack_head() -> bool:
            """Pop the acknowledged head (retry lock held). True if queue drained."""
            if pending and pending[0] is msg:
                pending.pop(0)
            if not pending:
                self._dirty_transcripts.pop(queue_session_id, None)
                self._transcript_append_failures.pop(session_id, None)
                return True
            return False

        # DB write outside the retry lock so other sessions can append.
        while True:
            # Spooled backlog (cap eviction or a stalled session) is older than ``msg``: replay it
            # first so recovery keeps transcript order; a still-dead DB just fails both.
            self._drain_spooled_drops(session_id)
            try:
                self._append_transcript_message(session_id, msg)
            except Exception as exc:
                from hermes_state import StateDbCorruptError, StateDbReplacedError
                from hermes_state_errors import CompressionSessionClosedError
                if isinstance(exc, (StateDbReplacedError, StateDbCorruptError)):
                    self._divert_transcript_after_db_replaced(session_id, queue_session_id, exc)
                    return
                if isinstance(exc, CompressionSessionClosedError):
                    # Adopt only a different, still-live compression tip, else fail closed.
                    _owner_key = self._owner_key_for_session_id(session_id)
                    child_id = self._live_compression_child(session_id)
                    if child_id:
                        # Record the child's owner BEFORE writing to it (the reroute is published
                        # only after the write succeeds — load-bearing for backlog order).
                        if _owner_key:
                            self._lazy("_session_owner_hints", dict)[child_id] = _owner_key
                        try:
                            self._append_transcript_message(child_id, msg)
                        except Exception as reroute_exc:
                            exc = reroute_exc
                        else:
                            with self._transcript_retry_lock:
                                pending = self._migrate_transcript_queue_to_child(
                                    session_id, queue_session_id, child_id, pending, msg)
                                queue_session_id = child_id
                            self._publish_transcript_reroute(session_id, child_id)
                            if not pending:
                                return
                            msg = pending[0]
                            session_id = child_id
                            continue
                    else:
                        # Permanent routing invariant failure, not a transient outage: drop it so it
                        # cannot poison later writes.
                        with self._transcript_retry_lock:
                            _ack_head()
                        logger.error(
                            "Session DB transcript append rejected for compression-ended %s with "
                            "no unique live child; not retrying", session_id)
                        return
                if self._is_fts_corruption_error(exc) and self._rebuild_fts_once():
                    try:
                        self._append_transcript_message(session_id, msg)
                    except Exception as retry_exc:
                        exc = retry_exc
                    else:
                        with self._transcript_retry_lock:
                            _ack_head()
                        continue
                with self._transcript_retry_lock:
                    failures = self._transcript_append_failures.get(session_id, 0) + 1
                    self._transcript_append_failures[session_id] = failures
                if failures >= self._TRANSCRIPT_APPEND_FAILURE_ESCALATION_THRESHOLD:
                    spooled = self._spool_stalled_backlog(session_id, queue_session_id)
                    logger.error(
                        "Session DB transcript append failed for %s (failure_count=%d, "
                        "pending=%d, spooled_to_disk=%d); session is stalled and needs operator "
                        "attention: %s", session_id, failures, len(pending), spooled, exc)
                else:
                    logger.warning(
                        "Session DB transcript append failed for %s (failure_count=%d, pending=%d); "
                        "will retry: %s", session_id, failures, len(pending), exc)
                return
            else:
                with self._transcript_retry_lock:
                    queue_empty = _ack_head()
                    if not queue_empty:
                        msg = pending[0]
                if queue_empty:
                    return
                continue

    def _spool_stalled_backlog(self, session_id: str, queue_session_id: str) -> int:
        """Move a stalled session's in-memory backlog (oldest first) to the on-disk pending spool so
        a crash/restart during the outage no longer loses it (#114266): ``recover_pending_to_db``
        replays it at boot, ``_drain_spooled_drops`` before the next live write. Stops at the first
        spool failure so order holds; whatever stays in memory remains under the cap."""
        with self._transcript_retry_lock:
            pending = self._dirty_transcripts.get(queue_session_id, [])
            backlog = list(pending)
        spooled = 0
        for message in backlog:
            if _spool_dropped(session_id, message) is None:
                break
            spooled += 1
        if spooled:
            self._lazy("_spooled_drop_sessions", set).add(session_id)
            with self._transcript_retry_lock:
                del pending[:spooled]
                if not pending:
                    self._dirty_transcripts.pop(queue_session_id, None)
        return spooled

    def _drain_spooled_drops(self, session_id: str) -> None:
        """Replay cap-dropped spooled transcript messages after DB recovery. Best-effort: replay
        failures keep the spool files for the next successful flush; nothing here may raise."""
        spooled_sessions = getattr(self, "_spooled_drop_sessions", None)
        if not spooled_sessions or session_id not in spooled_sessions:
            return
        try:
            from gateway.shutdown_flush import drain_transcript_spool
            # Inside an outage the append that follows logs/escalates the same failure; the
            # replay attempt is only the order-preserving probe, so its failure stays at DEBUG.
            with self._transcript_retry_lock:
                known_failing = bool(self._transcript_append_failures.get(session_id))
            _replayed, remaining = drain_transcript_spool(
                session_id, lambda message: self._append_transcript_message(session_id, message),
                db_known_failing=known_failing,
            )
            if not remaining:
                spooled_sessions.discard(session_id)
        except Exception as exc:
            logger.warning("Failed to drain transcript spool for %s: %s", session_id, exc)

    def _append_transcript_message(self, session_id: str, message: Dict[str, Any]) -> None:
        """Write one transcript row. Caller handles retry queuing."""
        _db = self._db_for_session_id(session_id)
        if _db is None:
            # Named profile with no resolvable home yet: defer (caller queues) instead of writing
            # into the ambient store.
            raise RuntimeError(
                f"no owning session store for {session_id}; deferring transcript write")
        is_assistant = message.get("role") == "assistant"
        _db.append_message(
            session_id=session_id,
            role=message.get("role", "unknown"),
            content=message.get("content"),
            tool_name=message.get("tool_name"),
            tool_calls=message.get("tool_calls"),
            tool_call_id=message.get("tool_call_id"),
            **{k: message.get(k) if is_assistant else None for k in _ASSISTANT_ONLY_KEYS},
            platform_message_id=(message.get("platform_message_id") or message.get("message_id")),
            observed=bool(message.get("observed")),
            timestamp=message.get("timestamp"),
            # Exact bytes sent to the API (prompt-cache-stable replay); must survive every
            # persistence path or the next replay diverges.
            api_content=extract_api_content_sidecar(message),
            # Presentation typing (e.g. "internal_notification"); DB-only.
            # "internal_notification" for self-injected async-delegation/background notification turns,
            # #82888). DB-only; stripped from provider-bound payloads.
            display_kind=message.get("display_kind"),
            display_metadata=message.get("display_metadata"),
        )

    @staticmethod
    def _is_fts_corruption_error(exc: Exception) -> bool:
        """True only when the failure is provably scoped to the FTS index. A bare SQLITE_CORRUPT
        can mean structural B-tree damage; only errors naming ``messages_fts`` or carrying FTS
        provenance (``SessionDB._is_fts_write_corruption_error``) may authorize the one-shot
        rebuild-and-retry; everything else takes the retry path.

        A generic ``database disk image is malformed`` (bare SQLITE_CORRUPT) can mean structural damage to
        canonical B-trees, not just the FTS shadow tables — treating it as FTS-only here made the store
        rebuild the index and retry transcript writes against a structurally corrupt database (#97940).
        """
        if "messages_fts" in str(exc).lower():
            return True
        import sqlite3
        from hermes_state import SessionDB
        return isinstance(exc, sqlite3.DatabaseError) and SessionDB._is_fts_write_corruption_error(exc)

    def _rebuild_fts_once(self) -> bool:
        """Attempt FTS5 ``rebuild``, at most once per ``_FTS_REBUILD_COOLDOWN_SECONDS`` window;
        True if any index was rebuilt.

        A permanent one-shot flag meant a single rebuild failure (e.g. a transient WAL
        split-brain guard hit) permanently disabled recovery for the life of the process, even
        though later corruption on the same store could be fixable. Retrying on a cooldown lets
        the store try again after the underlying condition (e.g. a foreign holder) has likely
        cleared, without hammering a database that is corrupt on every write.
        """
        now = time.monotonic()
        last_attempt = self._fts_rebuild_last_attempt_at
        if last_attempt is not None and (now - last_attempt) < self._FTS_REBUILD_COOLDOWN_SECONDS:
            return False
        db = self._db
        if db is None or not hasattr(db, "rebuild_fts"):
            return False
        self._fts_rebuild_last_attempt_at = now
        # WAL split-brain guard: skip when a foreign process holds state.db.
        foreign_holders = None
        if hasattr(db, "_foreign_state_db_holders"):
            foreign_holders = db._foreign_state_db_holders()
        if foreign_holders:
            logger.warning(
                "Skipping Session DB FTS rebuild while foreign processes hold the database or "
                "WAL sidecars (%s); canonical transcript writes remain available.", foreign_holders)
            return False
        try:
            rebuilt = db.rebuild_fts()
        except Exception as exc:
            logger.warning("Session DB FTS rebuild failed: %s", exc)
            return False
        if rebuilt:
            logger.warning("Rebuilt %d Session DB FTS index(es) after append corruption", rebuilt)
        return rebuilt > 0

    def _clear_dirty_transcript(self, session_id: str) -> None:
        """Drop queued pending messages so a rewrite/rewind doesn't re-insert them."""
        with self._transcript_retry_lock:
            self._dirty_transcripts.pop(session_id, None)
            self._transcript_append_failures.pop(session_id, None)

    def has_platform_message_id(self, session_id: str, platform_message_id: str) -> bool:
        """Whether a message with this platform_message_id is persisted (False without a DB).

        Thin wrapper over SessionDB.has_platform_message_id(). Returns False when no DB is available
        (in-memory sessions). Used by the gateway's transient-failure dedupe guard (#47237).
        """
        db = self._db_for_session_id(session_id)
        if not db:
            return False
        try:
            return db.has_platform_message_id(session_id, platform_message_id)
        except Exception:
            logger.debug("has_platform_message_id lookup failed", exc_info=True)
            return False

    def transcript_tail_role(self, session_id: str) -> Optional[str]:
        """Role of the newest live conversation row on the route ``load_transcript`` reads (``None``
        when empty, no DB, or the read fails — the boundary write would fail the same way)."""
        session_id = self._compression_tip_for_session_id(self._follow_reroutes(session_id))
        db = self._db_for_session_id(session_id)
        if not db:
            return None
        try:
            return db.latest_conversation_role(session_id)
        except Exception:
            logger.debug("transcript tail lookup failed for %s", session_id, exc_info=True)
            return None

    def rewrite_transcript(
        self, session_id: str, messages: List[Dict[str, Any]], active_only: bool = False,
        reject_active_turn_lease: bool = False) -> bool:
        """Replace a session's transcript (/retry, /compress). DESTRUCTIVE by default:
        ``active_only=False`` DELETEs every row incl. soft-archived compaction history (pass
        ``active_only=True`` for sessions that may carry archived rows). True when the write lands
        or there is no DB, False on failure — callers committing a destructive change on top
        (/compress repointing) must check it. ``reject_active_turn_lease`` is for user-initiated
        rewrites that do not own the cross-process turn lease."""
        db = self._db_for_session_id(session_id)
        if not db:
            return True
        with self._get_transcript_drain_lock():
            try:
                # Even when the current agent doesn't "own" persistence, the session on disk may already
                # carry compaction-archived rows — e.g. after a model switch or a /restore, both of which
                # mint a fresh agent with _session_db_created=False (so the check above is False) yet leave
                # the durable archived transcript in place. A full-history replace would DELETE those
                # archived rows just like the owned-agent case. Guard against it by replacing ONLY the live
                # (active=1) set unconditionally: on a fresh create/fork every row is active=1, so
                # active-only replace is behaviorally identical to the full replace — and when archived rows
                # DO exist they survive. An existence probe here (has_archived_messages) would fail OPEN
                # into the destructive replace on any DB error and can race a concurrent archive_and_compact
                # — the same probe failure mode #80216's /retry fix (gateway/slash_commands.py) deliberately
                # avoids.
                db.replace_messages(
                    session_id, messages, active_only=active_only,
                    reject_active_turn_lease=reject_active_turn_lease)
            except Exception as e:
                logger.debug("Failed to rewrite transcript in DB: %s", e)
                return False
            self._clear_dirty_transcript(session_id)
            return True

    def has_input_owner(self, session_id: str, owner: str) -> bool:
        """Find this accepted input on the canonical live continuation and its ancestors.

        Content and unrelated writers cannot establish ownership. Query only existence;
        compaction archives can contain many megabytes that replay never needs to load.
        """
        try:
            current = self._follow_reroutes(session_id)
            db = self._db_for_session_id(current)
            current = db.get_compression_tip(current) or current
            seen = set()
            while current and current not in seen:
                seen.add(current)
                if db.has_gateway_input_owner(current, owner):
                    return True
                row = db.get_session(current)
                if not row or not db._is_compression_child_row(row):
                    break
                current = row["parent_session_id"]
            return False
        except Exception as e:
            raise TranscriptReadError(session_id) from e

    def load_transcript(self, session_id: str) -> List[Dict[str, Any]]:
        """Load all messages from a session's transcript (state.db is canonical). Reads follow the
        same routing writes use — the in-memory reroute map, then the durable compression tip —
        otherwise the transcript "vanishes" while every message sits under the child."""
        if not self._db_for_session_id(session_id):
            return []
        session_id = self._follow_reroutes(session_id)
        with contextlib.suppress(Exception):
            # Durable successor survives restart; the reroute map doesn't.
            db = self._db_for_session_id(session_id)
            session_id = db.get_compression_tip(session_id) or session_id
        try:
            # repair_alternation: this feeds LIVE REPLAY; heal a durable user;user wedge once here.
            return self._db_for_session_id(session_id).get_messages_as_conversation(
                session_id, repair_alternation=True)
        except Exception as e:
            # Empty history is valid data; a failed canonical read is not — live-replay callers
            # must fail closed, not start from [].
            logger.error(
                "Transcript read failed for session %s; refusing to treat the conversation as "
                "empty: %s", session_id, e, exc_info=True)
            raise TranscriptReadError(session_id) from e

    def rewind_session(
        self, session_id: str, n: int = 1, *, require_retryable_composite: bool = False,
    ) -> Optional[Dict[str, Any]]:
        """Back up ``n`` user turns via soft-delete (``active=0``), mirroring CLI ``/undo [N]``.
        Returns ``{"rewound_count", "turns_undone", "target_text"}`` or ``None`` (no DB / no rewindable
        turn / persistence failure); ``n`` clamps to the oldest user turn. ``require_retryable_composite``
        is the gateway ``/retry`` guard: the selected turn must be a composite carrier whose live payload
        is losslessly replayable as text — that replay-policy ``ValueError`` propagates so /retry can
        explain why the carrier is unsafe."""
        db = self._db_for_session_id(session_id)
        if not db:
            return None
        from hermes_state_rewind import RewindTargetUnavailableError
        with self._get_transcript_drain_lock():
            try:
                outcome = db.rewind_user_turn(
                    session_id, -max(n, 1), require_retryable=require_retryable_composite,
                    require_composite=require_retryable_composite)
            except RewindTargetUnavailableError as e:
                logger.debug("rewind_session: %s", e)
                return None
            except ValueError:
                raise
            except Exception as e:
                logger.debug("rewind_session: rewind failed: %s", e)
                return None
            self._clear_dirty_transcript(session_id)
            return {"rewound_count": outcome.rewound_count, "turns_undone": outcome.turns_undone,
                    "target_text": outcome.live_text}