"""Truncation recovery (``finish_reason == "length"``) for the conversation turn loop.

Handles thinking-budget exhaustion, repetition-dominated truncation, content-filter stream
stalls escalated to the fallback chain, text continuation nudges (up to 4, with the ceiling
exit that drops the fragment trail), truncated tool-call retries with max_tokens boosts, and
the final roll-back. Nothing here imports ``agent.conversation_loop`` at module level
(cycle); loop-internal helpers are imported lazily so tests patching them keep working.
"""

from __future__ import annotations

import logging
import re
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple

from agent.error_classifier import FailoverReason
from agent.message_metadata import append_message
from agent.message_sanitization import close_interrupted_tool_sequence
from agent.repetition_guard import is_repetition_dominated
from agent.turn_api_call import stop_thinking_spinner
from agent.turn_failure_copy import content_policy_copy, provider_label_for, site_copy, stamp_failure
from agent.turn_retry_state import TurnRetryState
from agent.usage_pricing import normalize_usage
from hermes_constants import PARTIAL_STREAM_STUB_ID

logger = logging.getLogger("agent.conversation_loop")

# codex_responses only reaches ``finish_reason == "length"`` for a tool call cut off by
# max_output_tokens (turn_response_check.py::_derive_finish_reason); text truncation stays on
# the Codex incomplete continuation, so the text branch below never double-continues it.
_CONTINUABLE_MODES = {"chat_completions", "bedrock_converse", "anthropic_messages", "codex_responses"}
_THINK_TAG_RE = re.compile(r'<(?:think|thinking|reasoning|REASONING_SCRATCHPAD)[^>]*>', re.IGNORECASE)
_TRUNCATED_FINAL = site_copy("truncated")
_FIRST_TRUNCATED_FINAL = _TRUNCATED_FINAL
# #106260: a stream that died on a context-overflow error after partial delivery must not seed a
# continuation — the transcript already cannot fit, and appending the partial stub grows every
# later request into the same overflow. End the turn via the recovery contract instead.
_CONTEXT_OVERFLOW_PARTIAL_FINAL = (
    "The request no longer fits the model's context window, so the partial "
    "response was not continued. Continue in a fresh session (/new; gateway "
    "chats are reset automatically)."
)

def collapse_continuation_trail(
    agent: Any, messages: List[Dict[str, Any]], current_turn_user_idx: Any, *,
    finish_reason: str, parts: Optional[List[str]] = None,
) -> str:
    """Drop this turn's ``_length_continuation_fragment``/``_nudge`` rows and append one
    assistant row holding the joined, think-stripped partial; returns that text ("" none).

    ``parts=None`` (retry exhaustion, #119001): the text comes from the fragment rows and
    nothing happens without a valid turn index or a trail — an unanswered synthetic nudge
    must never be persisted, and an earlier turn's rows must never be read. Explicit
    ``parts`` (the continuation ceiling) always appends, scanning from 0 without an index.
    """
    idx = current_turn_user_idx
    valid_idx = isinstance(idx, int) and idx >= 0
    if parts is None and not (valid_idx and idx < len(messages)):
        return ""
    turn_start = idx + 1 if valid_idx else 0
    fragment_parts: List[str] = []
    retained: List[Any] = []
    found_trail = False
    for message in messages[turn_start:]:
        if isinstance(message, dict) and (
            message.get("_length_continuation_fragment") or message.get("_length_continuation_nudge")
        ):
            found_trail = True
            content = message.get("content")
            if message.get("_length_continuation_fragment") and isinstance(content, str) and content:
                fragment_parts.append(content)
            continue
        retained.append(message)
    if parts is None and not found_trail:
        return ""
    messages[turn_start:] = retained
    from agent.conversation_loop import _join_truncated_parts
    raw = fragment_parts if parts is None else parts
    join_parts = []
    for item in raw:
        if isinstance(item, tuple) and len(item) == 2 and isinstance(item[0], str):
            join_parts.append(item)
        elif isinstance(item, str) and item:
            join_parts.append((item, False))
    partial = agent._strip_think_blocks(
        _join_truncated_parts(join_parts)
    ).strip()
    if partial:
        append_message(messages, {"role": "assistant", "content": partial, "finish_reason": finish_reason})
    agent._session_messages = messages
    return partial


_THINKING_EXHAUSTED = (
    "💭 Reasoning exhausted the output token budget — no visible response was produced.",
    "⚠️ **Thinking Budget Exhausted**\n\nThe model used all its output tokens on reasoning "
    "and had none left for the actual response.\n\nTo fix this:\n"
    "→ Lower reasoning effort: `/reasoning low` or `/reasoning minimal`\n"
    "→ Or switch to a larger/non-reasoning model with `/model`",
    "Model used all output tokens on reasoning with none left "
    "for the response. Try lowering reasoning effort or increasing max_tokens.",
)

def repetition_copy(stopping: str, outcome: str, refusal: str) -> Tuple[str, str, str]:
    """(log line, user copy, error) for a repetition-dominated abort; only the clauses naming
    where the turn stopped differ between the length path and the stop path."""
    return (
        f"🔁 Response dominated by repeated text — stopping {stopping}.",
        "⚠️ **Response Stopped — Repetition Detected**\n\nThe model fell into a repetition loop while "
        f"writing this response, {outcome}\n\n→ Switch to a different model with `/model`\n"
        "→ Or resend your message (your conversation history is preserved)",
        f"Model output entered a repetition loop{refusal} degenerate response.",
    )


_REPETITION_DOMINATED = repetition_copy(
    "instead of continuing a degenerate response",
    "so continuing would only produce more repeated text. The partial response was discarded.",
    " and was truncated mid-loop; refusing to continue a",
)
_REPETITION_STREAM_CUT = repetition_copy(
    "the stream mid-loop",
    "so the stream was stopped instead of running on. The partial response was discarded.",
    " and the stream was cut mid-loop; discarding the",
)
_CEILING_NO_TEXT = (
    "⚠️ **No visible answer was produced.** The model hit its output-token limit on every "
    "continuation attempt — its reasoning consumed the entire budget each time.\n\nTo fix this:\n"
    "→ Lower reasoning effort: `/reasoning low` or `/reasoning none`\n→ Or raise max_tokens for this model"
)
# Below this many free tokens the prompt itself filled the window: a continuation nudge +
# fragment costs ~100 tokens per attempt, so retrying only shrinks the room (#106120).
_MIN_CONTINUATION_HEADROOM = 512
_WINDOW_FILLED = (
    "⚠️ **Context window full.** The prompt used {prompt:,} of this model's {ctx:,}-token "
    "context window, leaving no room to answer in. This is a context-window limit, not an "
    "output-length limit.\n\nTo fix this:\n→ Compress the conversation with `/compress` or start "
    "a new session\n→ Or raise the model's context window (e.g. Ollama `num_ctx`)"
)


def _prompt_filled_window(agent: Any, response: Any) -> Optional[tuple[int, int]]:
    """``(prompt_tokens, context_length)`` when this response's usage shows the prompt left
    less than ``_MIN_CONTINUATION_HEADROOM`` in the window compression resolves for the
    model; ``None`` (keep continuing) when either number is unknown."""
    ctx = int(getattr(getattr(agent, "context_compressor", None), "context_length", 0) or 0)
    usage = getattr(response, "usage", None)
    if not (ctx and usage):
        return None
    prompt = normalize_usage(usage, provider=agent.provider, api_mode=agent.api_mode).prompt_tokens
    return (prompt, ctx) if prompt and ctx - prompt < _MIN_CONTINUATION_HEADROOM else None


def normalize_response_for_agent(agent: Any, response: Any) -> Any:
    """One OpenAI-style message from any transport; Anthropic strips the OAuth tool prefix."""
    if agent.api_mode == "anthropic_messages":
        return agent._get_transport().normalize_response(
            response, strip_tool_prefix=agent._is_anthropic_oauth
        )
    return agent._get_transport().normalize_response(response)


def partial_result(
    messages: List[Dict[str, Any]], api_call_count: int, final_response: str,
    error: Optional[str] = None, *, failed: bool = False, compression_exhausted: bool = False,
) -> Dict[str, Any]:
    """Typed incomplete-turn result (``partial`` unless ``failed``); ``error`` defaults to
    ``final_response``. ``compression_exhausted`` carries the #98722 typed bit the gateway
    consumes to reset/move future input to a clean session (see run_turn.py)."""
    result = {
        "final_response": final_response,
        "messages": messages,
        "api_calls": api_call_count,
        "completed": False,
        ("failed" if failed else "partial"): True,
        "error": final_response if error is None else error,
    }
    if compression_exhausted:
        result["compression_exhausted"] = True
    return result


@dataclass
class TruncationVerdict:
    """Outcome of ``recover_from_truncation``.

    ``action``: ``"return"`` (end the turn with ``result``), ``"break"`` (a
    ``_retry.restart_with_*`` flag is set — restart the API call), ``"continue"``
    (re-issue the same call immediately) or ``"fallthrough"`` (unreachable in practice:
    every path exits, kept for the contract). The remaining fields are the loop locals
    the handler may have rebound."""

    action: str
    result: Optional[Dict[str, Any]]
    messages: List[Dict[str, Any]]
    length_continue_retries: int
    truncated_response_parts: List[Tuple[str, bool]]
    truncated_tool_call_retries: int
    retry_count: int
    compression_attempts: int


@dataclass(kw_only=True)
class _Trunc(TruncationVerdict):
    """Working state for the truncation phases — the verdict itself, plus the read-only
    call context; phases mutate the loop-local fields and ``done()`` stamps the action."""

    agent: Any
    response: Any
    finish_reason: str
    conversation_history: Any
    api_call_count: int
    effective_task_id: Any
    current_turn_user_idx: Any
    action: str = "fallthrough"
    result: Optional[Dict[str, Any]] = None
    window_filled: Optional[tuple[int, int]] = None  # (prompt_tokens, context_length)

    def done(self, action: str, result: Optional[Dict[str, Any]] = None) -> TruncationVerdict:
        self.action, self.result = action, result
        return self

    def end_turn(
        self, final_response: str, error: Optional[str] = None, *,
        result_messages: Optional[List[Dict[str, Any]]] = None, cleanup: bool = True,
        failed: bool = False, compression_exhausted: bool = False,
        failure: Tuple[str, bool] = ("truncated", True),
    ) -> TruncationVerdict:
        """Persist and end the turn as partial (or ``failed``).

        ``compression_exhausted`` forwards the #98722 typed bit so the gateway can
        move future input off a bloated session (run_turn.py consumes it). ``failure`` is
        the ``(failure_reason, retryable)`` verdict for the UI descriptor.
        """
        agent = self.agent
        if cleanup:
            agent._cleanup_task_resources(self.effective_task_id)
        agent._persist_session(self.messages, self.conversation_history)
        return self.done("return", stamp_failure(partial_result(
            self.messages if result_messages is None else result_messages, self.api_call_count,
            final_response, error, failed=failed, compression_exhausted=compression_exhausted,
        ), *failure))

    @property
    def is_stub(self) -> bool:
        return getattr(self.response, "id", "") == PARTIAL_STREAM_STUB_ID


def _abort_reason(agent: Any, content: Any, has_tool_calls: bool) -> Optional[tuple]:
    """``(vprint, user response, error)`` when continuation must NOT be attempted:
    thinking exhausted the budget (reasoning blocks with no visible text after them —
    ``content=None`` from non-<think> models is normal truncation), or a repetition loop
    burned the budget on one fragment (reasoning stripped first)."""
    if has_tool_calls:
        return None
    if content and _THINK_TAG_RE.search(content) and not agent._has_content_after_think_block(content):
        return _THINKING_EXHAUSTED
    visible = agent._strip_think_blocks(content) if isinstance(content, str) else content
    if visible and is_repetition_dominated(visible):
        return _REPETITION_DOMINATED
    return None


def _content_filter_fallback(st: _Trunc, _retry: TurnRetryState) -> Optional[TruncationVerdict]:
    """Content-filter stream stall → fallback. ``_content_filter_terminated`` is
    content-deterministic, so escalate before retrying the primary; without a fallback
    fall through to normal continuation (best-effort, may loop)."""
    agent = st.agent
    if not (
        getattr(st.response, "_content_filter_terminated", False)
        and agent._fallback_index < len(agent._fallback_chain)
    ):
        return None
    agent._vprint(
        f"{agent.log_prefix}🛡️  Content filter terminated stream — activating fallback provider...",
        force=True, diagnostic=True,
    )
    agent._emit_diagnostic_status("Content filter terminated stream; switching to fallback...")
    if agent._try_activate_fallback():
        # Roll partial content back to the last clean turn so the fallback gets a
        # coherent continuation point; unmark survivors (their text left the partial).
        if st.truncated_response_parts:
            st.messages = agent._get_messages_up_to_last_assistant(st.messages)
        for _frag in st.messages:
            if isinstance(_frag, dict):
                _frag.pop("_length_continuation_fragment", None)
                _frag.pop("_length_continuation_nudge", None)
        agent._session_messages = st.messages
        st.length_continue_retries = 0
        st.truncated_response_parts = []
        st.retry_count = 0
        st.compression_attempts = 0
        _retry.primary_recovery_attempted = False
        _retry.restart_with_rebuilt_messages = True
        return st.done("break")
    agent._vprint(
        f"{agent.log_prefix}⚠️  No fallback provider configured — retrying with same provider "
        f"(may re-hit filter)...",
        force=True, diagnostic=True,
    )
    return None


def _continue_text(st: _Trunc, _retry: TurnRetryState, assistant_message: Any) -> TruncationVerdict:
    """Text truncation (no tool calls): append the fragment + a continuation nudge (up to
    4), then the ceiling exit that drops the fragment trail and keeps the stitched partial.
    Never appends an interim assistant row with NO visible content — strict providers
    reject it with 400 — only the nudge."""
    from agent.conversation_loop import _get_continuation_prompt, _join_truncated_parts

    agent = st.agent
    messages = st.messages
    st.length_continue_retries += 1
    n = st.length_continue_retries
    _interim_content = getattr(assistant_message, "content", None)
    if not _interim_content and not st.is_stub:
        # Thinking-only truncation: continuing with thinking ON re-burns the budget.
        agent._ephemeral_reasoning_off = True
    if _interim_content:
        interim_msg = agent._build_assistant_message(assistant_message, st.finish_reason)
        interim_msg["_length_continuation_fragment"] = True  # ceiling exit drops these
        append_message(messages, interim_msg)
        st.truncated_response_parts.append((_interim_content, st.is_stub))

    filled = st.window_filled
    if n < 4 and filled is None:
        _dropped_tools = getattr(st.response, "_dropped_tool_names", None)
        if st.is_stub and _dropped_tools:
            agent._vprint(
                f"{agent.log_prefix}↻ Stream interrupted mid "
                f"tool-call ({', '.join(_dropped_tools[:3])}) — requesting chunked retry ({n}/4)...", diagnostic=True,
            )
        elif st.is_stub:
            agent._vprint(f"{agent.log_prefix}↻ Stream interrupted — requesting continuation ({n}/4)...", diagnostic=True)
        else:
            agent._vprint(f"{agent.log_prefix}↻ Requesting continuation ({n}/4)...", diagnostic=True)
        append_message(messages, {
            "role": "user", "content": _get_continuation_prompt(st.is_stub, _dropped_tools),
            "_length_continuation_nudge": True,
        })
        agent._session_messages = messages
        _retry.restart_with_length_continuation = True
        return st.done("break")

    # Unanswered continue nudges made every later turn re-truncate: drop the trail.
    partial_response = collapse_continuation_trail(
        agent, messages, st.current_turn_user_idx, finish_reason="length",
        parts=st.truncated_response_parts,
    )
    # The one-shot reasoning-off override must not leak into the next turn.
    agent._ephemeral_reasoning_off = False
    agent._vprint(
        f"{agent.log_prefix}⚠️  Not continuing — each attempt would only grow the prompt."
        if filled is not None else
        f"{agent.log_prefix}⚠️  Response still truncated after {n} continuation attempts — "
        + ("keeping the partial response received so far." if partial_response
           else "no visible text was produced."),
        force=True, diagnostic=True,
    )
    if filled is not None:
        notice = _WINDOW_FILLED.format(prompt=filled[0], ctx=filled[1])
        return st.end_turn(
            f"{partial_response}\n\n{notice}" if partial_response else notice,
            f"Prompt used {filled[0]} of {filled[1]} context tokens; no room to answer",
        )
    return st.end_turn(
        partial_response or _CEILING_NO_TEXT,
        "Response remained truncated after 4 continuation attempts",
    )


def _model_output_limit(agent: Any) -> Optional[int]:
    """The model's real max output tokens when Hermes knows it, else None."""
    if getattr(agent, "api_mode", None) != "anthropic_messages":
        return None
    # Local: only Anthropic-Messages turns need the adapter module.
    from agent.anthropic_adapter import _get_anthropic_max_output
    return _get_anthropic_max_output(getattr(agent, "model", None) or "")


def boosted_output_cap(agent: Any, requested_cap: Optional[int], n: int, base: Optional[int] = None) -> int:
    """Output budget for truncation retry ``n`` (1-based): ``base·2ⁿ``, never below the
    failed request's cap, at most ``max(32768, 2×cap)``, and never above the model's
    known output limit. ``base`` defaults to max_tokens, else the cap actually sent.

    A ceiling equal to the requested cap would re-send the same budget (#72770); a
    ceiling past the model limit only buys a provider 400 (#79715).
    """
    if base is None:
        base = agent.max_tokens or requested_cap or 4096
    anchor = requested_cap or base
    limit = _model_output_limit(agent)
    if limit and anchor >= limit:
        return anchor  # already at the model ceiling: doubling cannot help
    boost = min(max(base * (2 ** n), requested_cap or 0), max(32768, anchor * 2))
    return min(boost, limit) if limit else boost


def _retry_truncated_tool_call(st: _Trunc, api_kwargs: Any) -> TruncationVerdict:
    """Truncated tool call: re-run the same call (up to 4×) with a boosted max_tokens —
    a real output-cap truncation needs it, harmless for a network stall — else refuse to
    execute incomplete arguments."""
    agent = st.agent
    if st.truncated_tool_call_retries < 4:
        st.truncated_tool_call_retries += 1
        n = st.truncated_tool_call_retries
        if st.is_stub:
            agent._buffer_vprint(f"⚠️  Stream interrupted mid tool-call — retrying ({n}/4)...")
        else:
            agent._buffer_vprint(f"⚠️  Truncated tool call detected — retrying API call ({n}/4)...")
        agent._ephemeral_max_output_tokens = boosted_output_cap(
            agent, agent._requested_output_cap_from_api_kwargs(api_kwargs), n
        )
        return st.done("continue")  # don't append the broken response
    agent._flush_status_buffer()
    _failure = FailoverReason.timeout.value if st.is_stub else "truncated"
    if st.is_stub and getattr(st.response, "_clean_eof", False):
        # #102766: no transport error — the server (or a proxy) closed the stream cleanly
        # without a finish_reason, so "check your network" copy / a timeout stamp would mislead.
        agent._vprint(
            f"{agent.log_prefix}⚠️  Server kept closing the stream mid tool-call after 4 retries — the action was not executed.",
            force=True, diagnostic=True,
        )
        _final_response = site_copy("stream_closed_tool_call", label=provider_label_for(agent.provider))
        _failure = "truncated"
    elif st.is_stub:
        agent._vprint(
            f"{agent.log_prefix}⚠️  Stream kept dropping mid tool-call after 4 retries — the action was not executed.",
            force=True, diagnostic=True,
        )
        _final_response = site_copy("stream_dropped_tool_call", label=provider_label_for(agent.provider))
    else:
        agent._vprint(
            f"{agent.log_prefix}⚠️  Truncated tool call response detected again — refusing to execute incomplete tool arguments.",
            force=True, diagnostic=True,
        )
        _final_response = _TRUNCATED_FINAL
    agent._cleanup_task_resources(st.effective_task_id)
    # Prior tool batches can leave a tool-result tail; this path never reaches finalize_turn.
    close_interrupted_tool_sequence(st.messages, _final_response)
    return st.end_turn(
        _final_response, cleanup=False,
        failure=(_failure, True),
    )


def recover_from_truncation(
    agent: Any, response: Any, finish_reason: str, _retry: TurnRetryState, *,
    messages: List[Dict[str, Any]], conversation_history: Any, api_kwargs: Any, api_call_count: int,
    effective_task_id: Any, current_turn_user_idx: Any, length_continue_retries: int,
    truncated_response_parts: List[Tuple[str, bool]], truncated_tool_call_retries: int, retry_count: int,
    compression_attempts: int,
) -> TruncationVerdict:
    """Recover from a truncated response. Order is load-bearing: thinking exhaustion and
    repetition abort BEFORE any continuation; a content-filter stall escalates to the
    fallback chain BEFORE the primary is retried; text continuation (no tool calls) then
    truncated tool-call retry; finally roll back to the last complete assistant turn."""
    st = _Trunc(
        agent=agent, response=response, finish_reason=finish_reason,
        conversation_history=conversation_history, api_call_count=api_call_count,
        effective_task_id=effective_task_id, current_turn_user_idx=current_turn_user_idx,
        messages=messages, length_continue_retries=length_continue_retries,
        truncated_response_parts=truncated_response_parts,
        truncated_tool_call_retries=truncated_tool_call_retries, retry_count=retry_count,
        compression_attempts=compression_attempts,
    )
    if getattr(response, "_runaway_repetition", False):
        # The streaming call cut a live repetition loop: a continuation would only re-enter it,
        # whatever the partial or its tool calls look like.
        line, user_response, error = _REPETITION_STREAM_CUT
        agent._vprint(f"{agent.log_prefix}{line}", force=True, diagnostic=True)
        return st.end_turn(user_response, error)
    st.window_filled = _prompt_filled_window(agent, response)
    if st.is_stub and getattr(response, "_clean_eof", False):
        _banner = ("Response truncated — server ended the stream without ever sending finish_reason "
                   "(no transport error — the server or a proxy closed the stream cleanly)")
    elif st.is_stub:
        _banner = "Response truncated — stream ended before completion"
    elif st.window_filled:
        _banner = (f"Response truncated (finish_reason='length') - the prompt filled the context window "
                   f"({st.window_filled[0]:,}/{st.window_filled[1]:,} tokens)")
    else:
        _banner = "Response truncated (finish_reason='length') - model hit max output tokens"
    agent._vprint(f"{agent.log_prefix}⚠️  {_banner}", force=True, diagnostic=True)

    # #106260: a context-overflow error after partial delivery must not seed a
    # continuation. _partial_stream_stub marks such stubs _overflow_terminal and
    # leaves content empty; continuing would only re-send a larger request into
    # the same overflow. The stub path never raises, so this class never reached
    # recover_from_overflow's compress-and-retry on main either — ending the turn
    # replaces a growth loop, not a compression attempt.
    if getattr(st.response, "_overflow_terminal", False):
        agent._flush_status_buffer()
        agent._vprint(
            f"{agent.log_prefix}⚠️ Stream ended on a context-overflow error after "
            "partial delivery — not continuing (the request no longer fits the model's "
            "context window).",
            force=True, diagnostic=True,
        )
        # Prior tool batches can leave a tool-result tail; this path never reaches
        # finalize_turn (same as the truncated-tool-call terminal above).
        close_interrupted_tool_sequence(st.messages, _CONTEXT_OVERFLOW_PARTIAL_FINAL)
        # Carry the #98722 typed exhaustion bit so the gateway resets/moves future
        # input to a clean session instead of leaving this bloated one authoritative
        # for the next turn.
        return st.end_turn(
            _CONTEXT_OVERFLOW_PARTIAL_FINAL,
            error=_CONTEXT_OVERFLOW_PARTIAL_FINAL,
            failed=True,
            compression_exhausted=True,
            failure=("context_overflow", False),
        )

    _trunc_msg = normalize_response_for_agent(agent, response)
    _trunc_content = getattr(_trunc_msg, "content", None) if _trunc_msg else None
    _trunc_has_tool_calls = bool(getattr(_trunc_msg, "tool_calls", None)) if _trunc_msg else False

    abort = _abort_reason(agent, _trunc_content, _trunc_has_tool_calls)
    if abort is not None:
        line, user_response, error = abort
        agent._vprint(f"{agent.log_prefix}{line}", force=True, diagnostic=True)
        return st.end_turn(user_response, error)

    if agent.api_mode in _CONTINUABLE_MODES:
        cf = _content_filter_fallback(st, _retry)
        if cf is not None:
            return cf
        if _trunc_msg is not None:
            if not _trunc_has_tool_calls:
                return _continue_text(st, _retry, _trunc_msg)
            return _retry_truncated_tool_call(st, api_kwargs)

    if len(messages) > 1:
        agent._vprint(f"{agent.log_prefix}   ⏪ Rolling back to last complete assistant turn", diagnostic=True)
        return st.end_turn(
            _TRUNCATED_FINAL, result_messages=agent._get_messages_up_to_last_assistant(messages)
        )
    # First message was truncated - mark as failed
    agent._flush_status_buffer()
    agent._vprint(f"{agent.log_prefix}❌ First response truncated - cannot recover", force=True, diagnostic=True)
    return st.end_turn(_FIRST_TRUNCATED_FINAL, cleanup=False, failed=True)


_CODEX_REPLAY_KEYS = (
    "content", "reasoning", "reasoning_content", "reasoning_details",
    "codex_reasoning_items", "codex_message_items",
)

# Third return value of ``continue_codex_incomplete``: the reasoning-only stall was handed to a
# fallback provider — the caller re-syncs the system prompt identity and continues the turn.
CODEX_FALLBACK_ACTIVATED = "codex_fallback_activated"


def continue_codex_incomplete(
    agent: Any, assistant_message: Any, finish_reason: str, *, messages: List[Dict[str, Any]],
    conversation_history: Any, api_call_count: int, response: Any = None,
) -> Optional[Any]:
    """Codex Responses ``status=incomplete`` continuation (max 3 per turn).

    Appends the interim assistant message (deduped on visible content only — opaque
    provider state drifts per continuation; ``codex_reasoning_items`` are merged, not
    overwritten, because the earlier response holds the only native-compaction
    checkpoint) and, when a bare retry would be byte-identical, a user-role nudge — only
    after an assistant row, to preserve role alternation. Returns ``None`` to continue
    the turn loop, ``CODEX_FALLBACK_ACTIVATED`` when a reasoning-only stall was handed to
    the next fallback provider, or the terminal ``partial`` result once retries are exhausted.

    Reasoning-only stall ladder (#67321): a response with neither visible text nor a tool
    call advances ``_codex_reasoning_only_streak`` (a visible partial resets it; the aggregate
    ``_codex_incomplete_retries`` stays the cap for partials). Encrypted reasoning replays
    byte-for-byte, so after replay (1) and nudge (2) the third consecutive reasoning-only
    response goes to the configured fallback with the semantic ``incomplete_response`` reason
    instead of ending on the sentinel; when that response consumed the last iteration the
    fallback gets exactly one grace call (``_budget_grace_call`` is consumed by the next
    iteration, and the streak restarts from 0, so a second grace call is unreachable).

    When ``response`` hit ``max_output_tokens`` with no visible text (reasoning ate the
    whole budget), the next attempt goes out with reasoning off and a doubled output
    cap — the same one-shot overrides the chat-completions length path uses — because
    re-sending the identical budget and effort re-burns the budget identically (#90393)."""
    from agent.conversation_loop import _CODEX_INCOMPLETE_NUDGE
    from agent.turn_response_check import _codex_finish_reason

    agent._codex_incomplete_retries += 1
    n = agent._codex_incomplete_retries

    interim_msg = agent._build_assistant_message(assistant_message, finish_reason)
    interim_has_content = bool((interim_msg.get("content") or "").strip())
    _reasoning = interim_msg.get("reasoning")
    interim_has_reasoning = isinstance(_reasoning, str) and bool(_reasoning.strip())
    interim_has_codex_reasoning = bool(interim_msg.get("codex_reasoning_items"))
    interim_has_codex_message_items = bool(interim_msg.get("codex_message_items"))
    reasoning_only = not interim_has_content and not getattr(assistant_message, "tool_calls", None)
    agent._codex_reasoning_only_streak = agent._codex_reasoning_only_streak + 1 if reasoning_only else 0
    streak = agent._codex_reasoning_only_streak

    if interim_has_content or interim_has_reasoning or interim_has_codex_reasoning or interim_has_codex_message_items:
        last_msg = messages[-1] if messages else None
        last_is_dict = isinstance(last_msg, dict)
        last_interim_visible = agent._interim_assistant_visible_text(last_msg) if last_is_dict else ""
        current_interim_visible = agent._interim_assistant_visible_text(interim_msg)
        if last_interim_visible or current_interim_visible:
            same_visible_output = last_interim_visible == current_interim_visible
        else:
            # Neither has text eligible for interim delivery: compare raw content+reasoning.
            same_visible_output = last_is_dict and (
                (last_msg.get("content") or "") == (interim_msg.get("content") or "")
                and (last_msg.get("reasoning") or "") == (interim_msg.get("reasoning") or "")
            )
        if (
            last_is_dict
            and last_msg.get("role") == "assistant"
            and last_msg.get("finish_reason") == "incomplete"
            and same_visible_output
        ):
            # Duplicate: refresh replay state in place, no re-emitted commentary.
            for _key in _CODEX_REPLAY_KEYS:
                if _key not in interim_msg:
                    continue
                if _key == "codex_reasoning_items":
                    from agent.native_compaction import merge_interim_reasoning_items
                    last_msg[_key] = merge_interim_reasoning_items(last_msg.get(_key), interim_msg[_key])
                else:
                    last_msg[_key] = interim_msg[_key]
        else:
            append_message(messages, interim_msg)
            agent._emit_interim_assistant_message(interim_msg)

    if reasoning_only and streak >= 3:
        if agent._try_activate_fallback(reason=FailoverReason.incomplete_response):
            # The trigger may have consumed the turn budget; without a grace call the loop
            # exits before the fallback is ever asked.
            if api_call_count >= agent.max_iterations or agent.iteration_budget.remaining <= 0:
                agent._budget_grace_call = True
            agent._codex_incomplete_retries = 0
            agent._codex_reasoning_only_streak = 0
            if not agent.quiet_mode:
                agent._vprint(
                    f"{agent.log_prefix}↻ Codex reasoning-only stall after {streak} attempts — "
                    f"switching to fallback {agent.model} ({agent.provider})", diagnostic=True,
                )
            agent._emit_diagnostic_wait("↻ model stuck on internal reasoning — switching to fallback provider")
            agent._session_messages = messages
            return CODEX_FALLBACK_ACTIVATED
        # No fallback left: fall through to the terminal sentinel.
    elif n < 3 or reasoning_only:
        # A reasoning-only streak below 3 continues even once partials used up the aggregate
        # cap, so the mixed partial-then-stall variant reaches the ladder above.
        # If the interim has nothing the Responses converter will replay, a bare retry is
        # byte-identical; a replayable interim holding only a ``compaction`` checkpoint
        # ALSO re-sends identically. One bare retry, then always nudge.
        interim_replayable = interim_has_content or interim_has_codex_reasoning or interim_has_codex_message_items
        if not interim_replayable or n >= 2:
            _last_msg = messages[-1] if messages else None
            if isinstance(_last_msg, dict):
                _already_nudged = (
                    _last_msg.get("role") == "user" and _last_msg.get("content") == _CODEX_INCOMPLETE_NUDGE
                )
                # Alternation guard: the nudge may only follow an assistant row.
                if not _already_nudged and _last_msg.get("role") == "assistant":
                    append_message(messages, {"role": "user", "content": _CODEX_INCOMPLETE_NUDGE})
        if not interim_has_content and _codex_finish_reason(response) == "incomplete":
            agent._ephemeral_reasoning_off = True
            # No configured cap means the provider's own ceiling was hit: the observed
            # output_tokens IS that ceiling, so seed the escalation from it (else 4096).
            usage = getattr(response, "usage", None)
            observed = getattr(usage, "output_tokens", None) if not isinstance(usage, dict) else usage.get("output_tokens")
            agent._ephemeral_max_output_tokens = boosted_output_cap(
                agent, None, n, base=agent.max_tokens or int(observed or 0) or 4096
            )
        if not agent.quiet_mode:
            agent._vprint(f"{agent.log_prefix}↻ Codex response incomplete; continuing turn ({n}/3)", diagnostic=True)
        # Spinner/heartbeat notice: these retries can take minutes and otherwise look
        # like infinite thinking.
        # #70773: same FD-recycle corruption vector as #67142. The shared OpenAI client's connection pool
        # must NOT be closed from this watchdog/poll thread — worker threads from previous stale-killed
        # attempts may still be unwinding their SSL BIOs. The request-local client is already closed above
        # via _close_request_client_once. The shared client will be replaced lazily by
        # _ensure_primary_openai_client on the next request.
        # Surface the continuation on the live spinner/status line (CLI/TUI/Desktop) and gateway heartbeat:
        # each of these retries can spend minutes waiting on the provider, and without a distinct notice the
        # user only sees a generic thinking spinner ("infinite thinking", #64434).
        agent._emit_diagnostic_wait(
            f"↻ model returned reasoning with no final answer — asking it to continue ({n}/3)"
        )
        agent._session_messages = messages
        return None

    agent._codex_incomplete_retries = 0
    agent._codex_reasoning_only_streak = 0
    agent._persist_session(messages, conversation_history)
    return partial_result(
        messages, api_call_count, "Codex response remained incomplete after 3 continuation attempts"
    )


@dataclass
class RefusalVerdict:
    """Outcome of ``handle_content_policy_refusal``: ``"break"`` (fallback activated —
    restart armed on ``_retry``; caller resets retry/compression counters) or
    ``"return"`` (the typed content-policy result in ``result``). ``active_system_prompt``
    is the possibly re-synced system prompt."""

    action: str
    result: Optional[Dict[str, Any]]
    active_system_prompt: Any


def handle_content_policy_refusal(
    agent: Any, response: Any, _retry: TurnRetryState, *, thinking_spinner: Any,
    messages: List[Dict[str, Any]], api_messages: Any, api_kwargs: Any, active_system_prompt: Any,
    conversation_history: Any, api_call_count: int, effective_task_id: Any, turn_id: Any,
    api_request_id: Any, api_start_time: float, retry_count: int, max_retries: int,
) -> RefusalVerdict:
    """HTTP-200 refusal (``finish_reason`` ``content_filter`` / ``guardrail_intervened``).
    Deterministic for the unchanged prompt — never retried: one configured-fallback try,
    else surface the refusal (explanation may live only in the reasoning channel)."""
    from agent.conversation_loop import _arm_fallback_restart, _content_policy_blocked_result

    _refusal_result = normalize_response_for_agent(agent, response)
    _refusal_text = (getattr(_refusal_result, "content", None) or "").strip()
    if not _refusal_text:
        _refusal_text = (agent._extract_reasoning(_refusal_result) or "").strip()
    # Anthropic stop_reason=refusal carries its reason on stop_details (category + optional explanation),
    # not in a content block — without it a classifier halt reads as "(no text)" (#113689).
    _stop_details = (getattr(_refusal_result, "provider_data", None) or {}).get("stop_details")
    if not _refusal_text and isinstance(_stop_details, dict):
        _refusal_text = str(_stop_details.get("explanation") or "").strip() or (
            f"provider refusal category: {_stop_details['category']}" if _stop_details.get("category") else ""
        )

    agent._invoke_api_request_error_hook(
        task_id=effective_task_id, turn_id=turn_id, api_request_id=api_request_id,
        api_call_count=api_call_count, api_start_time=api_start_time, api_kwargs=api_kwargs,
        error_type="ContentPolicyBlocked",
        error_message=_refusal_text or "model declined to respond (content_filter)",
        status_code=None, retry_count=retry_count, max_retries=max_retries, retryable=False,
        reason=FailoverReason.content_policy_blocked.value,
    )
    stop_thinking_spinner(agent, thinking_spinner)

    if agent._has_pending_fallback():
        agent._buffer_diagnostic_status("⚠️ Model declined to respond (safety refusal) — trying fallback...")
    if agent._try_activate_fallback():
        active_system_prompt = _arm_fallback_restart(agent, api_messages, active_system_prompt, _retry)
        return RefusalVerdict("break", None, active_system_prompt)

    agent._flush_status_buffer()
    _refusal_log = _refusal_text[:500] + "..." if len(_refusal_text) > 500 else _refusal_text
    # native_stop_reason tells an Anthropic classifier refusal (``refusal``) from a Bedrock guardrail
    # block (``end_turn``); both arrive here as content_filter.
    logger.warning(
        "%sModel declined to respond (finish_reason=content_filter). model=%s provider=%s "
        "native_stop_reason=%s stop_details=%s refusal=%s",
        agent.log_prefix, agent.model, agent.provider,
        getattr(response, "stop_reason", None) or "n/a", _stop_details or "n/a",
        _refusal_log or "(no text)",
    )
    agent._emit_diagnostic_status("⚠️ The model declined to respond to this request (safety refusal).")
    _refusal_response = "⚠️ " + content_policy_copy(
        label=provider_label_for(agent.provider),
        summary=_refusal_text or "the model returned no explanation",
    )
    agent._cleanup_task_resources(effective_task_id)
    agent._persist_session(messages, conversation_history)
    return RefusalVerdict("return", _content_policy_blocked_result(
        messages, api_call_count, final_response=_refusal_response,
        error_detail=_refusal_text or "model declined (content_filter)",
    ), active_system_prompt)