feat(stop-failure): auto-heal poisoned-transcript crash loops

src/pinky_daemon/api.py

@@ -986,6 +986,8 @@ async def security_headers_middleware(request: Request, call_next):
     from pinky_daemon.auth_alerts import (
         TRANSPORT_FAILURE_POLICIES,
         AuthFailureTracker,
+        PoisonedTranscriptTracker,
+        classify_stop_failure,
         format_alert_message,
         resolve_operator_chat,
     )
@@ -1002,6 +1004,11 @@ async def security_headers_middleware(request: Request, call_next):
         )
         for et, pol in TRANSPORT_FAILURE_POLICIES.items()
     }
+    # Poisoned-transcript auto-heal: detects a tmux session crash-looping on a
+    # non-retryable, transcript-poisoning API error (thinking-block 400, or a
+    # tight any-class loop on one session) and force-restarts it onto a fresh
+    # transcript — host-global, like the trackers above.
+    poison_tracker = PoisonedTranscriptTracker()
 
     # Message broker — routes platform messages through approval checks to agent sessions
     _platform_adapters: dict[tuple[str, str], object] = {}
@@ -1508,6 +1515,7 @@ async def _broker_stop_all() -> dict:
     app.state.agents = agents
     app.state.auth_tracker = auth_tracker
     app.state.transport_failure_trackers = transport_failure_trackers
+    app.state.poison_tracker = poison_tracker
     app.state.conversation_store = store
     app.state.session_store = session_store
     app.state.session_event_store = session_event_store
@@ -2206,6 +2214,112 @@ async def _on_transport_failure_alert(agent_name: str, error_type: str) -> None:
             f"(reason={decision.get('reason')})"
         )
 
+    async def _auto_heal_poisoned_session(
+        agent_name: str, session, *, reason: str
+    ) -> bool:
+        """Force a wedged tmux session onto a fresh transcript.
+
+        The poisoned message is permanent in the current transcript, so the
+        only escape is to drop ``--continue`` and respawn. We set the
+        one-shot ``force_fresh_context_once`` (which makes the next launch
+        omit ``--continue``) and call ``force_restart(bypass_guard=True)`` —
+        the same tmux-native respawn the inflight watchdog uses; by the time
+        we're here the REPL is provably wedged, so the persistence guard
+        would only strand the replay queue.
+
+        Only ``TmuxSession`` exposes ``force_restart`` + ``_config``;
+        SDK/Codex sessions duck-out to ``False`` (the poison loop is a tmux
+        REPL phenomenon). Returns True iff the respawn succeeded.
+        """
+        force_restart = getattr(session, "force_restart", None)
+        cfg = getattr(session, "_config", None)
+        if not callable(force_restart) or cfg is None:
+            _log(
+                f"api: poison auto-heal skipped for {agent_name} — session "
+                f"is not a tmux REPL (no force_restart/_config)"
+            )
+            return False
+
+        # Refresh orientation so the fresh boot wakes with current context
+        # (best-effort; a stale wake_context still beats staying wedged).
+        try:
+            cfg.wake_context = _build_streaming_wake_context(agent_name, commit=False)
+        except Exception as e:
+            _log(f"api: poison auto-heal wake-context rebuild failed for {agent_name}: {e}")
+        cfg.resume_handle = ""
+        cfg.restart_reason = "poisoned_transcript_auto_heal"
+        cfg.force_fresh_context_once = True
+
+        try:
+            ok = bool(await force_restart(bypass_guard=True))
+        except Exception as e:
+            _log(f"api: poison auto-heal force_restart raised for {agent_name}: {e}")
+            return False
+
+        if not ok:
+            _log(f"api: poison auto-heal force_restart returned False for {agent_name}")
+            return False
+
+        audit_meta = {
+            "reason": reason,
+            "source": "poison_auto_heal",
+            "label": "main",
+        }
+        try:
+            activity.log(
+                agent_name=agent_name,
+                event_type="force_restart",
+                title=f"{agent_name} auto-healed poisoned transcript",
+                metadata=audit_meta,
+            )
+            session_event_store.log(
+                session_id=getattr(session, "id", ""),
+                agent_name=agent_name,
+                event_type="force_restart",
+                metadata=audit_meta,
+            )
+        except Exception as e:
+            _log(f"api: poison auto-heal audit log failed for {agent_name}: {e}")
+        _log(
+            f"api: AUTO-HEALED poisoned transcript for {agent_name} "
+            f"(reason={reason}) — respawned on fresh context"
+        )
+        return True
+
+    async def _alert_operator_poison_heal(agent_name: str, decision: dict) -> None:
+        """Best-effort informational DM after an auto-heal — Brad always
+        wants to know a self-reset fired (a silent self-heal that hides a
+        recurring problem is worse than the wedge). Never raises."""
+        try:
+            chat_id, platform = resolve_operator_chat(
+                get_setting=agents.get_setting,
+                list_all_approved_users=agents.list_all_approved_users,
+            )
+        except Exception as e:
+            _log(f"api: poison auto-heal alert resolve_operator_chat raised: {e}")
+            return
+        if not chat_id:
+            _log(
+                f"api: poison auto-heal alert for {agent_name} suppressed — "
+                "no operator chat resolved"
+            )
+            return
+        try:
+            host_label = agents.get_setting("host_label", "") or ""
+        except Exception:
+            host_label = ""
+        trigger = decision.get("reason", "")
+        body = (
+            f"🔧 Auto-healed {agent_name} on {host_label or 'this host'}: "
+            f"poisoned transcript ({trigger}) detected — force-restarted onto a "
+            f"fresh context. The agent lost in-session context (reloads from "
+            f"memory on boot). No action needed; flagging for visibility."
+        )
+        try:
+            await _broker_send(agent_name, platform, chat_id, body)
+        except Exception as e:
+            _log(f"api: poison auto-heal alert send failed for {agent_name}: {e}")
+
     async def _make_streaming_resume_handle_callback(agent_name: str, label: str):
         """Persist a streaming session's SDK resume handle when captured.
 
@@ -4809,7 +4923,15 @@ async def transport_stop_failure(name: str, req: TransportStopFailureRequest):
         if not agent:
             raise HTTPException(404, f"Agent '{name}' not found")
 
-        error_type = (req.error_type or "unknown").strip() or "unknown"
+        # Classify FIRST. Claude Code reports the thinking-block integrity
+        # 400 (a poisoned, non-retryable transcript) as a coarse
+        # ``unknown``/``invalid_request``; ``classify_stop_failure`` upgrades
+        # it to ``poisoned_transcript`` from the error message body so logging
+        # + auto-heal can see it. Auth/billing/rate_limit types pass through
+        # unchanged (signature only matches the thinking-block wording).
+        error_type, is_poison_signature = classify_stop_failure(
+            req.error_type, req.message
+        )
 
         # Observability: every failure class is recorded.
         try:
@@ -4873,13 +4995,38 @@ async def transport_stop_failure(name: str, req: TransportStopFailureRequest):
                         f"api: StopFailure turn-resolve raised for {name}: {e}"
                     )
 
+        # Poisoned-transcript auto-heal. A thinking-block 400 (or a tight
+        # any-class loop on one session_id) can't be retried on the same
+        # transcript — every resend re-sends the poison. Detect it and force
+        # a fresh restart so the agent escapes the loop instead of wedging
+        # forever. Runs AFTER turn-resolve so the wedged caller is unblocked
+        # first; the tracker rate-limits to ≤1 heal/agent/10min. Fire-and-
+        # forget — a heal hiccup must never fail the hook.
+        auto_healed = False
+        try:
+            heal_decision = poison_tracker.record_failure(
+                name, req.session_id, is_poison_signature
+            )
+            if heal_decision.get("should_reset") and session is not None:
+                auto_healed = await _auto_heal_poisoned_session(
+                    name, session,
+                    reason=f"{error_type}/{heal_decision.get('reason')}",
+                )
+                if auto_healed:
+                    # Two-phase: only start the cooldown once the heal landed.
+                    poison_tracker.commit_reset(name)
+                    await _alert_operator_poison_heal(name, heal_decision)
+        except Exception as e:
+            _log(f"api: StopFailure poison auto-heal raised for {name}: {e}")
+
         return {
             "ok": True,
             "agent": name,
             "error_type": error_type,
             "auth_failure": auth_failure,
             "alert_routed": alert_routed,
             "turn_resolved": turn_resolved,
+            "auto_healed": auto_healed,
         }
 
     @app.post("/agents/{name}/transport/transcript-path")

src/pinky_daemon/auth_alerts.py

@@ -128,6 +128,185 @@ class FailureAlertPolicy:
 }
 
 
+# ── Poisoned-transcript detection (auto-heal) ───────────────────────
+#
+# Distinct from the alert classes above: those surface a problem a HUMAN
+# must fix (re-auth, top up billing, wait out throttling). A *poisoned
+# transcript* is one the daemon can fix ITSELF, and must — because it
+# never self-resolves and silently wedges the agent.
+#
+# THE FAILURE MODE
+#   With extended thinking on, every ``thinking`` block carries a crypto
+#   signature and Anthropic requires the latest assistant message's
+#   thinking blocks passed back byte-for-byte. When Claude Code cancels a
+#   sibling in a parallel tool batch (one tool errored), it reconstructs
+#   that assistant message and the thinking blocks no longer match → API
+#   400 "thinking blocks ... cannot be modified". That message is now baked
+#   into the transcript, so EVERY subsequent turn re-sends it and fails
+#   identically. ``claude --continue`` just resumes the poison. The agent
+#   crash-loops until a human force-restarts it onto a fresh transcript.
+#
+# THE FIX
+#   Detect it and force a fresh restart (drop ``--continue`` → new
+#   transcript), abandoning the poisoned history. Two triggers, belt +
+#   suspenders:
+#     1. Signature match — the StopFailure ``message`` carries the API
+#        error body; the "thinking ... cannot be modified" wording is a
+#        definitive, non-retryable poison signal. Claude Code reports this
+#        as a coarse ``unknown``/``invalid_request`` error_type, so the
+#        signature (not error_type) is what catches it.
+#     2. Tight-loop fallback — N rapid StopFailures (any class) on the SAME
+#        session_id within a short window. Catches future poison variants
+#        the string match misses, while the time window keeps sporadic
+#        rate_limits across a healthy long session from false-triggering.
+
+POISONED_TRANSCRIPT_ERROR_TYPE = "poisoned_transcript"
+
+# Auto-heal tuning.
+DEFAULT_POISON_SIG_THRESHOLD = 2     # consecutive signature hits on a session
+DEFAULT_POISON_LOOP_THRESHOLD = 4    # rapid any-class hits on a session
+DEFAULT_POISON_LOOP_WINDOW_SECONDS = 180     # "rapid" = within this gap
+DEFAULT_POISON_RESET_COOLDOWN_SECONDS = 600  # ≤1 auto-heal per agent / 10 min
+
+
+def classify_stop_failure(error_type: str, message: str) -> tuple[str, bool]:
+    """Map a raw StopFailure to ``(effective_error_type, is_poison_signature)``.
+
+    Detects the thinking-block integrity 400 from the error ``message``
+    regardless of the coarse ``error_type`` Claude Code sends for it (it
+    arrives as ``unknown`` / ``invalid_request``). When matched, the
+    effective type is upgraded to ``poisoned_transcript`` for logging +
+    auto-heal routing; otherwise the original ``error_type`` passes through
+    unchanged so existing auth/billing/rate_limit classification is
+    untouched.
+    """
+    msg = (message or "").lower()
+    if "thinking" in msg and (
+        "cannot be modified" in msg or "must remain as they were" in msg
+    ):
+        return POISONED_TRANSCRIPT_ERROR_TYPE, True
+    return (error_type or "unknown").strip() or "unknown", False
+
+
+@dataclass
+class _AgentPoisonState:
+    """Per-agent streak state for poisoned-transcript detection."""
+
+    session_id: str = ""
+    sig_count: int = 0          # consecutive signature hits on this session
+    loop_count: int = 0         # rapid any-class hits on this session
+    last_failure_at: float = 0.0
+    last_reset_at: float = 0.0  # rate-limit anchor (advanced on commit_reset)
+
+
+class PoisonedTranscriptTracker:
+    """Decides when a wedged tmux session should be auto-healed (#poison).
+
+    Per-agent state keyed on ``session_id``: a streak only accumulates
+    within one transcript and resets the moment the session_id changes
+    (i.e. after a heal spawns a fresh transcript). Two-phase like
+    ``AuthFailureTracker``: ``record_failure`` decides, ``commit_reset``
+    advances the cooldown — so a heal that fails to launch doesn't burn the
+    cooldown and the next failure retries.
+
+    Single-event-loop, no locking (mirrors ``AuthFailureTracker``).
+    """
+
+    def __init__(
+        self,
+        *,
+        sig_threshold: int = DEFAULT_POISON_SIG_THRESHOLD,
+        loop_threshold: int = DEFAULT_POISON_LOOP_THRESHOLD,
+        loop_window_seconds: int = DEFAULT_POISON_LOOP_WINDOW_SECONDS,
+        reset_cooldown_seconds: int = DEFAULT_POISON_RESET_COOLDOWN_SECONDS,
+        clock=time.time,
+    ) -> None:
+        self._sig_threshold = sig_threshold
+        self._loop_threshold = loop_threshold
+        self._loop_window = loop_window_seconds
+        self._cooldown = reset_cooldown_seconds
+        self._clock = clock
+        self._agents: dict[str, _AgentPoisonState] = defaultdict(_AgentPoisonState)
+
+    def record_failure(
+        self, agent_name: str, session_id: str, is_signature: bool
+    ) -> dict:
+        """Record one StopFailure; return a decision dict.
+
+        Keys: ``should_reset`` (bool), ``reason`` (signature|loop|
+        below_threshold|cooldown|no_session_id), ``sig_count``, ``loop_count``.
+
+        A non-empty ``session_id`` is required — it's the streak anchor; we
+        can't tell a tight loop apart from sporadic failures without it.
+        """
+        session_id = (session_id or "").strip()
+        if not session_id:
+            return {"should_reset": False, "reason": "no_session_id"}
+
+        now = self._clock()
+        st = self._agents[agent_name]
+
+        # New transcript → fresh streak. (A heal clears session_id via
+        # commit_reset, so the post-heal session's failures start clean.)
+        if session_id != st.session_id:
+            st.session_id = session_id
+            st.sig_count = 0
+            st.loop_count = 0
+            st.last_failure_at = 0.0
+
+        # Tight-loop counting: a gap longer than the window breaks the
+        # streak, so rate_limits sprinkled across a healthy long-lived
+        # session never accumulate to the loop threshold.
+        if st.last_failure_at and (now - st.last_failure_at) > self._loop_window:
+            st.loop_count = 0
+        st.loop_count += 1
+        if is_signature:
+            st.sig_count += 1
+        st.last_failure_at = now
+
+        if st.last_reset_at and now - st.last_reset_at < self._cooldown:
+            return {
+                "should_reset": False,
+                "reason": "cooldown",
+                "sig_count": st.sig_count,
+                "loop_count": st.loop_count,
+                "cooldown_remaining": int(
+                    self._cooldown - (now - st.last_reset_at)
+                ),
+            }
+
+        if st.sig_count >= self._sig_threshold:
+            reason = "signature"
+        elif st.loop_count >= self._loop_threshold:
+            reason = "loop"
+        else:
+            return {
+                "should_reset": False,
+                "reason": "below_threshold",
+                "sig_count": st.sig_count,
+                "loop_count": st.loop_count,
+            }
+
+        return {
+            "should_reset": True,
+            "reason": reason,
+            "sig_count": st.sig_count,
+            "loop_count": st.loop_count,
+        }
+
+    def commit_reset(self, agent_name: str) -> None:
+        """Mark an auto-heal as performed: start the cooldown and clear the
+        streak. Call ONLY after the force-restart succeeded — a failed heal
+        must leave the cooldown unadvanced so the next failure retries."""
+        st = self._agents[agent_name]
+        st.last_reset_at = self._clock()
+        st.sig_count = 0
+        st.loop_count = 0
+        # Drop the anchor so the post-heal transcript starts a clean streak
+        # even if its new session_id is briefly unknown to us.
+        st.session_id = ""
+
+
 @dataclass
 class _AgentFailures:
     """Sliding-window record of recent auth failures for one agent."""