diff --git a/src/pinky_daemon/api.py b/src/pinky_daemon/api.py index d55c302d..b69cc42f 100644 --- a/src/pinky_daemon/api.py +++ b/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,6 +4995,30 @@ 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, @@ -4880,6 +5026,7 @@ async def transport_stop_failure(name: str, req: TransportStopFailureRequest): "auth_failure": auth_failure, "alert_routed": alert_routed, "turn_resolved": turn_resolved, + "auto_healed": auto_healed, } @app.post("/agents/{name}/transport/transcript-path") diff --git a/src/pinky_daemon/auth_alerts.py b/src/pinky_daemon/auth_alerts.py index b1258d97..446ea03e 100644 --- a/src/pinky_daemon/auth_alerts.py +++ b/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.""" diff --git a/tests/test_auth_alerts.py b/tests/test_auth_alerts.py index 5a47a9a6..9bd7676c 100644 --- a/tests/test_auth_alerts.py +++ b/tests/test_auth_alerts.py @@ -4,8 +4,11 @@ from types import SimpleNamespace from pinky_daemon.auth_alerts import ( + POISONED_TRANSCRIPT_ERROR_TYPE, TRANSPORT_FAILURE_POLICIES, AuthFailureTracker, + PoisonedTranscriptTracker, + classify_stop_failure, format_alert_message, resolve_operator_chat, ) @@ -490,3 +493,168 @@ def test_admin_watchdog_exposes_transport_alert_status(tmp_path): for et, st in tstatus.items(): assert st["status"] == "ok" # no failures yet assert st["agents_failing"] == [] + + +# ── classify_stop_failure (poisoned-transcript signature) ─────────── + +# Verbatim from production: the Anthropic 400 that wedged Barsik. +_REAL_POISON_MSG = ( + 'API Error: 400 {"type":"error","error":{"type":"invalid_request_error",' + '"message":"messages.5.content.70: `thinking` or `redacted_thinking` ' + "blocks in the latest assistant message cannot be modified. These blocks " + 'must remain as they were in the original response."}}' +) + + +def test_classify_detects_real_poison_message(): + """The exact production 400 → poisoned_transcript signature.""" + et, sig = classify_stop_failure("unknown", _REAL_POISON_MSG) + assert et == POISONED_TRANSCRIPT_ERROR_TYPE + assert sig is True + + +def test_classify_detects_must_remain_variant(): + et, sig = classify_stop_failure( + "invalid_request", + "thinking blocks must remain as they were in the original response", + ) + assert et == POISONED_TRANSCRIPT_ERROR_TYPE + assert sig is True + + +def test_classify_passes_through_non_poison(): + """Auth/billing/rate_limit are untouched — signature only matches the + thinking-block wording.""" + assert classify_stop_failure("authentication_failed", "401") == ( + "authentication_failed", + False, + ) + assert classify_stop_failure("rate_limit", "429 Too Many Requests") == ( + "rate_limit", + False, + ) + + +def test_classify_thinking_word_without_modify_phrase_is_not_poison(): + """A message merely mentioning 'thinking' must NOT trip the signature — + both halves (the word + the immutability phrase) are required.""" + et, sig = classify_stop_failure("server_error", "still thinking about it") + assert et == "server_error" + assert sig is False + + +def test_classify_empty_defaults_unknown(): + assert classify_stop_failure("", "") == ("unknown", False) + assert classify_stop_failure(None, None) == ("unknown", False) + + +# ── PoisonedTranscriptTracker ─────────────────────────────────────── + + +def _poison_tracker(now_ref): + return PoisonedTranscriptTracker( + sig_threshold=2, + loop_threshold=4, + loop_window_seconds=180, + reset_cooldown_seconds=600, + clock=lambda: now_ref[0], + ) + + +def test_signature_resets_on_second_hit_same_session(): + now = [1000.0] + tr = _poison_tracker(now) + d1 = tr.record_failure("barsik", "sess-A", is_signature=True) + assert d1["should_reset"] is False # one hit — below threshold + d2 = tr.record_failure("barsik", "sess-A", is_signature=True) + assert d2["should_reset"] is True + assert d2["reason"] == "signature" + + +def test_signature_streak_resets_when_session_changes(): + now = [1000.0] + tr = _poison_tracker(now) + assert tr.record_failure("barsik", "sess-A", is_signature=True)["should_reset"] is False + # New transcript → streak restarts, so a single hit on B doesn't fire. + d = tr.record_failure("barsik", "sess-B", is_signature=True) + assert d["should_reset"] is False + assert d["sig_count"] == 1 + + +def test_loop_threshold_fires_on_rapid_any_class_failures(): + now = [1000.0] + tr = _poison_tracker(now) + decisions = [] + for _ in range(4): + now[0] += 5 # rapid — well within the 180s window + decisions.append( + tr.record_failure("ivan", "sess-X", is_signature=False) + ) + assert [d["should_reset"] for d in decisions] == [False, False, False, True] + assert decisions[-1]["reason"] == "loop" + + +def test_loop_streak_breaks_when_failures_are_spread_out(): + """Sporadic rate_limits across a healthy long session must NOT trigger — + a gap past the window resets the loop counter.""" + now = [1000.0] + tr = _poison_tracker(now) + for _ in range(3): + now[0] += 5 + tr.record_failure("ivan", "sess-X", is_signature=False) + now[0] += 200 # gap > 180s window → streak broken + d = tr.record_failure("ivan", "sess-X", is_signature=False) + assert d["should_reset"] is False + assert d["loop_count"] == 1 + + +def test_no_session_id_never_resets(): + now = [1000.0] + tr = _poison_tracker(now) + for _ in range(5): + d = tr.record_failure("barsik", "", is_signature=True) + assert d["should_reset"] is False + assert d["reason"] == "no_session_id" + + +def test_cooldown_blocks_repeat_heal(): + now = [1000.0] + tr = _poison_tracker(now) + tr.record_failure("barsik", "sess-A", is_signature=True) + assert tr.record_failure("barsik", "sess-A", is_signature=True)["should_reset"] is True + tr.commit_reset("barsik") # heal landed → cooldown starts + # Fresh transcript after the heal, immediately poisoned again, but within + # cooldown → suppressed (no reset storm). + now[0] += 60 + d1 = tr.record_failure("barsik", "sess-B", is_signature=True) + d2 = tr.record_failure("barsik", "sess-B", is_signature=True) + assert d1["should_reset"] is False + assert d2["should_reset"] is False + assert d2["reason"] == "cooldown" + # Past the cooldown → heals again. + now[0] += 600 + tr.record_failure("barsik", "sess-B", is_signature=True) + assert tr.record_failure("barsik", "sess-B", is_signature=True)["should_reset"] is True + + +def test_commit_reset_clears_streak_and_anchor(): + now = [1000.0] + tr = _poison_tracker(now) + tr.record_failure("barsik", "sess-A", is_signature=True) + tr.record_failure("barsik", "sess-A", is_signature=True) + tr.commit_reset("barsik") + now[0] += 700 # past cooldown so only the streak-clear is under test + # Same session_id reappears but the streak was cleared → one hit, no fire. + d = tr.record_failure("barsik", "sess-A", is_signature=True) + assert d["should_reset"] is False + assert d["sig_count"] == 1 + + +def test_separate_agents_track_independently(): + now = [1000.0] + tr = _poison_tracker(now) + tr.record_failure("barsik", "s1", is_signature=True) + # Ivan's first hit must not inherit Barsik's streak. + d = tr.record_failure("ivan", "s2", is_signature=True) + assert d["should_reset"] is False + assert d["sig_count"] == 1 diff --git a/tests/test_stop_failure_hook.py b/tests/test_stop_failure_hook.py index 21440905..09789974 100644 --- a/tests/test_stop_failure_hook.py +++ b/tests/test_stop_failure_hook.py @@ -21,6 +21,7 @@ import json import os import tempfile +from types import SimpleNamespace from fastapi.testclient import TestClient @@ -489,3 +490,136 @@ async def handle_stop_failure(self, *a, **k): ) assert r.status_code == 200 assert r.json()["turn_resolved"] is False + + +# ── Poisoned-transcript auto-heal (#poison) ───────────────────────── + + +# Verbatim from production: the Anthropic 400 that crash-looped Barsik. +_POISON_MSG = ( + 'API Error: 400 {"type":"error","error":{"type":"invalid_request_error",' + '"message":"messages.5.content.70: `thinking` or `redacted_thinking` ' + "blocks in the latest assistant message cannot be modified. These blocks " + 'must remain as they were in the original response."}}' +) + + +class _FakeHealableSession: + """tmux stand-in exposing the surface the auto-heal duck-types: + ``handle_stop_failure`` (#108 resolve) + ``force_restart`` + ``_config``.""" + + def __init__(self, restart_ok: bool = True): + self.id = "fake-session-id" + self._config = SimpleNamespace( + force_fresh_context_once=False, + restart_reason="", + resume_handle="stale-handle", + wake_context="", + ) + self._restart_ok = restart_ok + self.restart_calls: list[bool] = [] + self.stop_failure_calls: list[tuple] = [] + + async def handle_stop_failure(self, error_type, message="", session_id=""): + self.stop_failure_calls.append((error_type, message, session_id)) + return True + + async def force_restart(self, *, bypass_guard: bool = False) -> bool: + self.restart_calls.append(bypass_guard) + return self._restart_ok + + +class TestStopFailureAutoHeal: + """A poisoned transcript (thinking-block 400) crash-loops the agent — every + resend re-sends the poison. The endpoint detects it (signature) and force- + restarts the tmux session onto a fresh context. Threshold/loop/cooldown + logic lives in PoisonedTranscriptTracker (test_auth_alerts.py); here we pin + that the endpoint classifies, gates on threshold, and drives the reset.""" + + def setup_method(self): + self._tmpdir = tempfile.mkdtemp() + self._db = os.path.join(self._tmpdir, "test.db") + + def _client(self, name: str = "dymok"): + app = _make_app(self._db) + client = TestClient(app) + r = client.post("/agents", json={"name": name, "model": "sonnet"}) + assert r.status_code == 200 + return client, app + + def _post(self, client, *, message: str, error_type: str = "unknown", + session_id: str = "sess-poison"): + return client.post( + "/agents/dymok/transport/stop-failure", + json={ + "error_type": error_type, + "message": message, + "session_id": session_id, + }, + ) + + def test_classifies_poison_message(self): + """The thinking-block 400 is reclassified poisoned_transcript even + though CC sent a coarse error_type.""" + client, app = self._client() + r = self._post(client, message=_POISON_MSG) + assert r.status_code == 200 + body = r.json() + assert body["error_type"] == "poisoned_transcript" + # First hit is below the signature threshold (2) → no heal yet. + assert body["auto_healed"] is False + + def test_second_poison_triggers_auto_heal(self): + """Second signature hit on the same session crosses the threshold → + force-fresh restart fires.""" + client, app = self._client() + fake = _FakeHealableSession() + app.state.broker._streaming["dymok"] = {"main": fake} + + assert self._post(client, message=_POISON_MSG).json()["auto_healed"] is False + r2 = self._post(client, message=_POISON_MSG) + assert r2.json()["auto_healed"] is True + + # The respawn ran with the guard bypassed (REPL is provably wedged)… + assert fake.restart_calls == [True] + # …and was set up to drop --continue (fresh transcript) + re-orient. + assert fake._config.force_fresh_context_once is True + assert fake._config.restart_reason == "poisoned_transcript_auto_heal" + assert fake._config.resume_handle == "" + + def test_non_poison_never_auto_heals(self): + """A rate_limit loop is transient/retryable — never auto-healed via + the signature path.""" + client, app = self._client() + fake = _FakeHealableSession() + app.state.broker._streaming["dymok"] = {"main": fake} + for _ in range(3): + r = self._post(client, message="429 Too Many Requests", + error_type="rate_limit") + assert r.json()["auto_healed"] is False + assert fake.restart_calls == [] + + def test_auto_heal_skipped_for_non_tmux_session(self): + """SDK/Codex sessions (no force_restart) can't be healed this way — + threshold is reached but the reset duck-outs to False.""" + client, app = self._client() + app.state.broker._streaming["dymok"] = {"main": object()} + self._post(client, message=_POISON_MSG) + r2 = self._post(client, message=_POISON_MSG) + assert r2.status_code == 200 + assert r2.json()["auto_healed"] is False + + def test_failed_restart_does_not_burn_cooldown(self): + """If the respawn fails, the heal isn't committed — so the next poison + retries instead of being silenced for the cooldown window.""" + client, app = self._client() + fake = _FakeHealableSession(restart_ok=False) + app.state.broker._streaming["dymok"] = {"main": fake} + self._post(client, message=_POISON_MSG) + r2 = self._post(client, message=_POISON_MSG) + assert r2.json()["auto_healed"] is False # force_restart returned False + # Cooldown not advanced → a subsequent poison still tries to heal. + r3 = self._post(client, message=_POISON_MSG) + assert r3.json()["auto_healed"] is False + # force_restart was attempted on each post past the threshold (2nd, 3rd). + assert fake.restart_calls == [True, True]