refactor(agent/turn): phase helpers for truncation recovery, shared spinner-stop and partial_result helpers

agent/turn_api_call.py

@@ -2,12 +2,13 @@
 the attempt while another session's Nous Portal rate limit is active), ``perform_api_call``
 (streaming decision, MoA prepared-request handshake, LLM execution middleware wrapper, the
 redirect ``_model_request_active`` bracket and the response-vs-redirect crossing check) and
-``handle_api_interrupt`` (``InterruptedError`` mid-call). Extracted from
-``run_conversation``; nothing here imports ``agent.conversation_loop`` at module level.
+``handle_api_interrupt`` (``InterruptedError`` mid-call). Nothing here imports
+``agent.conversation_loop`` at module level (cycle).
 """
 
 from __future__ import annotations
 
+from contextlib import nullcontext
 from dataclasses import dataclass
 import logging
 import time
@@ -18,6 +19,16 @@ from agent.message_metadata import append_message
 logger = logging.getLogger("agent.conversation_loop")
 
 
+def stop_thinking_spinner(agent: Any, thinking_spinner: Any) -> None:
+    """Stop the spinner silently and clear the thinking callback; returns ``None`` so
+    callers can rebind ``thinking_spinner = stop_thinking_spinner(agent, thinking_spinner)``."""
+    if thinking_spinner:
+        thinking_spinner.stop("")
+    if agent.thinking_callback:
+        agent.thinking_callback("")
+    return None
+
+
 @dataclass
 class ApiCallVerdict:
     """``action``: ``"fallthrough"`` (``response`` is ready for verification) or ``"break"``
@@ -29,58 +40,44 @@ class ApiCallVerdict:
     interrupted: Any
 
 
+def _should_stream(agent: Any) -> bool:
+    """Streaming is preferred even without consumers (stale-stream / read-timeout health
+    checks); disabled on provider signal, ACP schemes, MoA without a display consumer, or
+    Mock clients in tests (SimpleNamespace, not stream iterators)."""
+    if getattr(agent, "_disable_streaming", False):
+        return False
+    _base = str(agent.base_url or "").lower()
+    if agent.provider in {"copilot-acp"} or _base.startswith(("acp://", "acp+tcp://")):
+        return False
+    if not agent._has_stream_consumers():
+        if agent.provider == "moa":
+            return False
+        from unittest.mock import Mock
+        if isinstance(getattr(agent, "client", None), Mock):
+            return False
+    return True
+
+
 def perform_api_call(
     agent: Any, *, api_kwargs: Any, _original_api_kwargs: Any, _llm_middleware_trace: Any,
     _moa_prepared_request: Any, _retry: Any, thinking_spinner: Any, retry_count: Any,
     api_call_count: Any, api_request_id: Any, effective_task_id: Any, turn_id: Any,
     interrupted: Any,
 ) -> ApiCallVerdict:
-    """Issue the request. Streaming is preferred even without consumers (stale-stream /
-    read-timeout health checks) and disabled per provider signal, ACP schemes, MoA without a
-    display consumer, or Mock clients in tests."""
+    """Issue the request (see ``_should_stream`` for the streaming decision)."""
     response = None
 
-    def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> ApiCallVerdict:
+    def _verdict(action: str) -> ApiCallVerdict:
         return ApiCallVerdict(
             action=action, response=response, thinking_spinner=thinking_spinner,
             interrupted=interrupted,
         )
 
-    # Always prefer streaming even without consumers: it gives stale-
-    # stream/read-timeout health checks that quiet callers otherwise lack.
-    # Falls back if unsupported.
     def _stop_spinner():
         nonlocal thinking_spinner
-        if thinking_spinner:
-            thinking_spinner.stop("")
-            thinking_spinner = None
-        if agent.thinking_callback:
-            agent.thinking_callback("")
+        thinking_spinner = stop_thinking_spinner(agent, thinking_spinner)
 
-    _use_streaming = True
-    # Provider signaled "stream not supported": stay non-streaming for the
-    # session.
-    if getattr(agent, "_disable_streaming", False):
-        _use_streaming = False
-    # ACP clients (`acp://` scheme, any vendor) return a plain
-    # SimpleNamespace, not a stream; mirrors the Responses API exclusion.
-    elif (
-        agent.provider in {"copilot-acp"}
-        or str(agent.base_url or "").lower().startswith("acp://")
-        or str(agent.base_url or "").lower().startswith("acp+tcp://")
-    ):
-        _use_streaming = False
-    # MoA streams only with a display/TTS consumer
-    # (MoAChatCompletions.create() honors stream=True); else complete-
-    # response path.
-    elif agent.provider == "moa" and not agent._has_stream_consumers():
-        _use_streaming = False
-    elif not agent._has_stream_consumers():
-        # No consumer: still stream for health checking, except Mock clients
-        # in tests (SimpleNamespace, not stream iterators).
-        from unittest.mock import Mock
-        if isinstance(getattr(agent, "client", None), Mock):
-            _use_streaming = False
+    _use_streaming = _should_stream(agent)
 
     def _perform_api_call(next_api_kwargs):
         if agent.api_mode == "codex_responses":
@@ -117,15 +114,14 @@ def perform_api_call(
 
     from hermes_cli.middleware import run_llm_execution_middleware
 
+    # The ``_model_request_active`` bracket is taken under the redirect lock when one exists,
+    # so redirect() can't observe a half-toggled flag.
     _model_request_active = getattr(agent, "_model_request_active", None)
     _redirect_lock = getattr(agent, "_pending_redirect_lock", None)
-    if _redirect_lock is not None:
-        with _redirect_lock:
-            if _model_request_active is not None:
-                _model_request_active.set()
-    elif _model_request_active is not None:
-        _model_request_active.set()
-    _redirect_crossed_response = False
+    _bracket = nullcontext() if _redirect_lock is None else _redirect_lock
+    with _bracket:
+        if _model_request_active is not None:
+            _model_request_active.set()
     try:
         response = run_llm_execution_middleware(
             api_kwargs, _perform_api_call, original_request=_original_api_kwargs,
@@ -135,25 +131,17 @@ def perform_api_call(
             api_call_count=api_call_count, middleware_trace=list(_llm_middleware_trace),
         )
     finally:
-        if _redirect_lock is not None:
-            with _redirect_lock:
-                if _model_request_active is not None:
-                    _model_request_active.clear()
-                _redirect_crossed_response = bool(
-                    agent._pending_redirect
-                )
-        else:
+        with _bracket:
             if _model_request_active is not None:
                 _model_request_active.clear()
-            _redirect_crossed_response = agent._has_pending_redirect()
+            _redirect_crossed_response = (
+                bool(agent._pending_redirect) if _redirect_lock is not None
+                else agent._has_pending_redirect()
+            )
     if _redirect_crossed_response:
         # Response and redirect can cross threads: discard the now-stale
         # response and rebuild from the correction rather than lose it.
-        if thinking_spinner:
-            thinking_spinner.stop("")
-            thinking_spinner = None
-        if agent.thinking_callback:
-            agent.thinking_callback("")
+        thinking_spinner = stop_thinking_spinner(agent, thinking_spinner)
         if agent.clear_interrupt(preserve_redirect=True):
             _retry.restart_with_redirected_messages = True
         else:
@@ -180,36 +168,18 @@ def handle_api_interrupt(
     """``InterruptedError`` during the provider call: a pending redirect keeps its correction
     queued for the outer-loop rebuild; otherwise keep any streamed partial text so the next
     turn has a record of the half-finished reply."""
-    from agent.conversation_loop import (
-        INTERRUPT_WAITING_FOR_MODEL_PREFIX,
-    )
+    from agent.conversation_loop import INTERRUPT_WAITING_FOR_MODEL_PREFIX
 
-    def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> ApiInterruptVerdict:
-        return ApiInterruptVerdict(
-            action=action,
-            thinking_spinner=thinking_spinner,
-            interrupted=interrupted,
-            final_response=final_response,
-
-        )
-
-    if thinking_spinner:
-        thinking_spinner.stop("")
-        thinking_spinner = None
-    if agent.thinking_callback:
-        agent.thinking_callback("")
-    if agent._has_pending_redirect():
-        # redirect() cancelled only this request: keep the correction
-        # queued, clear the cancellation bit, let the outer loop rebuild.
-        # Never materialize incomplete signed/encrypted reasoning items.
-        if agent.clear_interrupt(preserve_redirect=True):
-            _retry.restart_with_redirected_messages = True
-            return _verdict("break")
+    thinking_spinner = stop_thinking_spinner(agent, thinking_spinner)
+    # redirect() cancelled only this request: keep the correction queued, clear the
+    # cancellation bit, let the outer loop rebuild. Never materialize incomplete
+    # signed/encrypted reasoning items.
+    if agent._has_pending_redirect() and agent.clear_interrupt(preserve_redirect=True):
+        _retry.restart_with_redirected_messages = True
+        return ApiInterruptVerdict("break", thinking_spinner, interrupted, final_response)
     api_elapsed = time.time() - api_start_time
     agent._vprint(f"{agent.log_prefix}⚡ Interrupted during API call.", force=True)
     interrupted = True
-    # Keep assistant text already streamed before the stop, else the next
-    # turn has no record of the half-finished reply.
     _partial = agent._strip_think_blocks(
         getattr(agent, "_current_streamed_assistant_text", "") or ""
     ).strip()
@@ -219,8 +189,7 @@ def handle_api_interrupt(
     else:
         final_response = f"{INTERRUPT_WAITING_FOR_MODEL_PREFIX}{api_elapsed:.1f}s elapsed)."
     agent._persist_session(messages, conversation_history)
-    return _verdict("break")
-    return _verdict("fallthrough")
+    return ApiInterruptVerdict("break", thinking_spinner, interrupted, final_response)
 
 
 @dataclass
@@ -241,9 +210,7 @@ def nous_rate_limit_guard(
 ) -> NousRateGuardVerdict:
     """Skip the call if another session recorded a Nous Portal rate limit: every attempt (incl.
     SDK retries) counts against RPH. Never lets the guard itself break the agent loop."""
-    from agent.conversation_loop import (
-        _arm_fallback_restart,
-    )
+    from agent.conversation_loop import _arm_fallback_restart
 
     def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> NousRateGuardVerdict:
         return NousRateGuardVerdict(
@@ -251,9 +218,6 @@ def nous_rate_limit_guard(
             compression_attempts=compression_attempts, result=result,
         )
 
-    # ── Nous Portal rate limit guard ──────────────────────
-    # Skip the call if another session recorded a rate limit: every attempt
-    # (incl. SDK retries) counts against RPH.
     if agent.provider == "nous":
         try:
             from agent.nous_rate_guard import (
@@ -265,9 +229,7 @@ def nous_rate_limit_guard(
                     f"Nous Portal rate limit active — "
                     f"resets in {_fmt_nous_remaining(_nous_remaining)}."
                 )
-                agent._buffer_vprint(
-                    f"⏳ {_nous_msg} Trying fallback..."
-                )
+                agent._buffer_vprint(f"⏳ {_nous_msg} Trying fallback...")
                 agent._buffer_status(f"⏳ {_nous_msg}")
                 if agent._try_activate_fallback():
                     active_system_prompt = _arm_fallback_restart(
@@ -275,8 +237,7 @@ def nous_rate_limit_guard(
                     retry_count = 0
                     compression_attempts = 0
                     return _verdict("break")
-                # No fallback available — surface buffered context
-                # so user sees the rate-limit message that led here.
+                # No fallback — surface the buffered rate-limit context that led here.
                 agent._flush_status_buffer()
                 agent._persist_session(messages, conversation_history)
                 return _verdict("return", {
@@ -292,8 +253,6 @@ def nous_rate_limit_guard(
                     "failed": True,
                     "error": _nous_msg,
                 })
-        except ImportError:
-            pass
         except Exception:
             pass  # Never let rate guard break the agent loop
     return _verdict("fallthrough")

agent/turn_response_check.py

@@ -2,9 +2,8 @@
 spinner, validate the response shape (retry / eager fallback / terminal invalid-response
 result), derive ``finish_reason`` per api_mode, route content-policy refusals and
 ``length`` truncation, fold usage into the compressor, and mark the logical relay call
-complete. Extracted from ``run_conversation``; nothing here imports
-``agent.conversation_loop`` at module level (cycle) — loop-internal helpers resolve lazily so
-existing ``patch("agent.conversation_loop.X")`` sites keep intercepting.
+complete. Nothing here imports ``agent.conversation_loop`` at module level (cycle) —
+loop-internal helpers resolve lazily so ``patch("agent.conversation_loop.X")`` keeps intercepting.
 """
 
 from __future__ import annotations
@@ -14,6 +13,7 @@ import logging
 import time
 from typing import Any, Dict, Optional
 
+from agent.turn_api_call import stop_thinking_spinner
 from agent.turn_truncation import handle_content_policy_refusal, recover_from_truncation
 from agent.turn_usage import record_response_usage
 
@@ -42,6 +42,45 @@ class ResponseCheckVerdict:
     result: Optional[Dict[str, Any]] = None
 
 
+def _codex_finish_reason(response: Any) -> str:
+    """Responses API max-output exhaustion is a normal Codex incomplete turn: route it to
+    the Codex continuation path (``"incomplete"``), not the length rollback."""
+    status = getattr(response, "status", None)
+    if isinstance(status, str):
+        status = status.strip().lower()
+    incomplete_details = getattr(response, "incomplete_details", None)
+    if isinstance(incomplete_details, dict):
+        incomplete_reason = incomplete_details.get("reason")
+    else:
+        incomplete_reason = getattr(incomplete_details, "reason", None)
+    if incomplete_reason is not None:
+        incomplete_reason = str(incomplete_reason).strip().lower()
+    if status == "incomplete" and incomplete_reason in {"max_output_tokens", "length"}:
+        return "incomplete"
+    if status == "incomplete" and incomplete_reason == "content_filter":
+        return "content_filter"
+    return "stop"
+
+
+def _derive_finish_reason(agent: Any, response: Any, messages: Any) -> str:
+    if agent.api_mode == "codex_responses":
+        return _codex_finish_reason(response)
+    transport = agent._get_transport()
+    if agent.api_mode == "anthropic_messages":
+        return transport.map_finish_reason(response.stop_reason)
+    normalized = transport.normalize_response(response)  # Bedrock already normalized at dispatch
+    finish_reason = normalized.finish_reason
+    if agent.api_mode != "bedrock_converse" and agent._should_treat_stop_as_truncated(
+        finish_reason, normalized, messages
+    ):
+        agent._vprint(
+            f"{agent.log_prefix}⚠️  Treating suspicious Ollama/GLM stop response as truncated",
+            force=True,
+        )
+        return "length"
+    return finish_reason
+
+
 def check_api_response(
     agent: Any, *, response: Any, _retry: Any, thinking_spinner: Any, messages: Any,
     api_messages: Any, api_kwargs: Any, active_system_prompt: Any, conversation_history: Any,
@@ -54,10 +93,7 @@ def check_api_response(
     """Verify ``response`` in the original order. The retry buffer is NOT cleared on success
     (bytes back != usable content); ``_preflight_compression_blocked``/``_last_preflight_pressure``
     reset only when the usage fold re-arms the compression budget."""
-    from agent.conversation_loop import (
-        validate_response_shape,
-    )
-    api_duration = None
+    from agent.conversation_loop import validate_response_shape
 
     def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> ResponseCheckVerdict:
         return ResponseCheckVerdict(
@@ -74,25 +110,17 @@ def check_api_response(
 
     api_duration = time.time() - api_start_time
 
-    # Stop thinking spinner silently -- the response box or tool
-    # execution messages that follow are more informative.
-    if thinking_spinner:
-        thinking_spinner.stop("")
-        thinking_spinner = None
-    if agent.thinking_callback:
-        agent.thinking_callback("")
+    # Silent stop: the response box / tool messages that follow are more informative.
+    thinking_spinner = stop_thinking_spinner(agent, thinking_spinner)
 
     if not agent.quiet_mode:
         agent._vprint(f"{agent.log_prefix}⏱️  API call completed in {api_duration:.2f}s")
 
     if agent.verbose_logging:
-        # Log response with provider info if available
         resp_model = getattr(response, 'model', 'N/A') if response else 'N/A'
         logging.debug(f"API Response received - Model: {resp_model}, Usage: {response.usage if hasattr(response, 'usage') else 'N/A'}")
 
-    # Validate response shape before proceeding
     response_invalid, error_details = validate_response_shape(agent, response)
-
     if response_invalid:
         _iv = retry_invalid_response(
             agent, response=response, error_details=error_details, _retry=_retry,
@@ -112,54 +140,9 @@ def check_api_response(
             return _verdict(_iv.action, _iv.result)
 
     agent._turn_received_provider_response = True
+    finish_reason = _derive_finish_reason(agent, response, messages)
 
-    # Check finish_reason before proceeding
-    if agent.api_mode == "codex_responses":
-        status = getattr(response, "status", None)
-        if isinstance(status, str):
-            status = status.strip().lower()
-        incomplete_details = getattr(response, "incomplete_details", None)
-        incomplete_reason = None
-        if isinstance(incomplete_details, dict):
-            incomplete_reason = incomplete_details.get("reason")
-        else:
-            incomplete_reason = getattr(incomplete_details, "reason", None)
-        if incomplete_reason is not None:
-            incomplete_reason = str(incomplete_reason).strip().lower()
-        if status == "incomplete" and incomplete_reason in {"max_output_tokens", "length"}:
-            # Responses API max-output exhaustion is a normal Codex
-            # incomplete turn: use the Codex continuation path, not the
-            # length rollback.
-            finish_reason = "incomplete"
-        elif status == "incomplete" and incomplete_reason == "content_filter":
-            finish_reason = "content_filter"
-        else:
-            finish_reason = "stop"
-    elif agent.api_mode == "anthropic_messages":
-        _tfr = agent._get_transport()
-        finish_reason = _tfr.map_finish_reason(response.stop_reason)
-    elif agent.api_mode == "bedrock_converse":
-        # Bedrock response already normalized at dispatch — use transport
-        _bt_fr = agent._get_transport()
-        _bedrock_result = _bt_fr.normalize_response(response)
-        finish_reason = _bedrock_result.finish_reason
-    else:
-        _cc_fr = agent._get_transport()
-        _finish_result = _cc_fr.normalize_response(response)
-        finish_reason = _finish_result.finish_reason
-        assistant_message = _finish_result
-        if agent._should_treat_stop_as_truncated(
-            finish_reason, assistant_message, messages
-        ):
-            agent._vprint(
-                f"{agent.log_prefix}⚠️  Treating suspicious Ollama/GLM stop response as truncated",
-                force=True,
-            )
-            finish_reason = "length"
-
-    # ── Content-policy refusal (HTTP 200) ──────────────────
-    # Refusal finish reasons (``content_filter``, ``guardrail_intervened``)
-    # are deterministic: one fallback try, else return the refusal.
+    # HTTP-200 refusals are deterministic: one fallback try, else return the refusal.
     if finish_reason == "content_filter":
         _rv = handle_content_policy_refusal(
             agent, response, _retry, thinking_spinner=thinking_spinner, messages=messages,
@@ -194,12 +177,8 @@ def check_api_response(
         truncated_tool_call_retries = _tv.truncated_tool_call_retries
         retry_count = _tv.retry_count
         compression_attempts = _tv.compression_attempts
-        if _tv.action == "return":
-            return _verdict("return", _tv.result)
-        if _tv.action == "break":
-            return _verdict("break")
-        if _tv.action == "continue":
-            return _verdict("continue")
+        if _tv.action in ("return", "break", "continue"):
+            return _verdict(_tv.action, _tv.result)
 
     # Fold provider usage into compressor / anchors / session counters / state.db
     # (agent/turn_usage.py). A rearmed budget also clears the preflight-block latch.
@@ -213,10 +192,8 @@ def check_api_response(
         _preflight_compression_blocked = False
         _last_preflight_pressure = None
 
-    _retry.has_retried_429 = False  # Reset on success
-    # Don't clear the retry buffer: bytes back != usable content; it is
-    # cleared once genuine content lands. Clearing Nous rate-limit state
-    # proves the limit reset so other sessions may resume.
+    _retry.has_retried_429 = False
+    # Clearing Nous rate-limit state proves the limit reset so other sessions may resume.
     if agent.provider == "nous":
         try:
             from agent.nous_rate_guard import clear_nous_rate_limit
@@ -225,12 +202,9 @@ def check_api_response(
             pass
     from agent import relay_llm
 
-    relay_llm.complete_logical_call(
-        api_request_id, outcome="success"
-    )
+    relay_llm.complete_logical_call(api_request_id, outcome="success")
     agent._touch_activity(f"API call #{api_call_count} completed")
-    return _verdict("break")  # Success, exit retry loop
-    return _verdict("fallthrough")
+    return _verdict("break")
 
 
 @dataclass
@@ -278,20 +252,11 @@ def retry_invalid_response(
         status_code=getattr(getattr(response, "error", None), "code", None),
         retry_count=retry_count, max_retries=max_retries, retryable=True, reason="invalid_response",
     )
-    # Stop spinner silently — retry status is now buffered
-    # and only surfaced if every retry+fallback exhausts.
-    if thinking_spinner:
-        thinking_spinner.stop("")
-        thinking_spinner = None
-    if agent.thinking_callback:
-        agent.thinking_callback("")
-
-    # Invalid response — could be rate limiting, provider timeout,
-    # upstream server error, or malformed response.
+    # Retry status is buffered and only surfaced if every retry+fallback exhausts.
+    thinking_spinner = stop_thinking_spinner(agent, thinking_spinner)
     retry_count += 1
 
-    # Eager fallback: empty/malformed responses often mean rate limiting
-    # — switch now instead of extended backoff.
+    # Eager fallback: empty/malformed responses often mean rate limiting.
     if agent._fallback_index < len(agent._fallback_chain):
         agent._buffer_status("⚠️ Empty/malformed response — switching to fallback...")
     if agent._try_activate_fallback():
@@ -304,15 +269,12 @@ def retry_invalid_response(
     error_msg, provider_name, _failure_hint = describe_invalid_response(
         agent, response, api_duration
     )
-
     agent._buffer_vprint(f"⚠️  Invalid API response (attempt {retry_count}/{max_retries}): {', '.join(error_details)}")
     agent._buffer_vprint(f"   🏢 Provider: {provider_name}")
-    cleaned_provider_error = agent._clean_error_message(error_msg)
-    agent._buffer_vprint(f"   📝 Provider message: {cleaned_provider_error}")
+    agent._buffer_vprint(f"   📝 Provider message: {agent._clean_error_message(error_msg)}")
     agent._buffer_vprint(f"   ⏱️  {_failure_hint}")
 
     if retry_count >= max_retries:
-        # Try fallback before giving up
         if agent._has_pending_fallback():
             agent._buffer_status(f"⚠️ Max retries ({max_retries}) for invalid responses — trying fallback...")
         if agent._try_activate_fallback():
@@ -333,17 +295,15 @@ def retry_invalid_response(
             "completed": False,
             "api_calls": api_call_count,
             "error": _final_response,
-            "failed": True  # Mark as failure for filtering
+            "failed": True,
         })
 
-    # Backoff before retry — jittered exponential: 5s base, 120s cap
     wait_time = jittered_backoff(retry_count, base_delay=5.0, max_delay=120.0)
     agent._buffer_vprint(f"⏳ Retrying in {wait_time:.1f}s ({_failure_hint})...")
     logger.warning("Invalid API response (retry %d/%d): %s | Provider: %s", retry_count, max_retries, ', '.join(error_details), provider_name)
 
-    # A redirect cancels only the live request; the helper preserves the
-    # pending correction (restart_with_redirected_messages) instead of
-    # destroying it with clear_interrupt().
+    # A redirect cancels only the live request; the helper preserves the pending
+    # correction (restart_with_redirected_messages) instead of clear_interrupt()-ing it.
     _interrupted = interruptible_backoff_sleep(
         agent, wait_time, _retry, messages=messages, conversation_history=conversation_history,
         api_call_count=api_call_count,
@@ -355,5 +315,4 @@ def retry_invalid_response(
         return _verdict("return", _interrupted)
     if _retry.restart_with_redirected_messages:
         return _verdict("break")  # rebuild this iteration from the correction
-    return _verdict("continue")  # Retry the API call
-    return _verdict("fallthrough")
+    return _verdict("continue")