fix: quitting the CLI no longer spams shutdown-race API errors onto the shell

When the TUI exits while the post-turn background review fork is still
mid-request, every further API attempt raises 'cannot schedule new
futures after interpreter shutdown'. The conversation loop treated this
as a retryable API error: un-gated ❌ prints leaked onto the user's
shell AFTER the TUI exited (call #4, #5, #6...) and the loop retried a
doomed request until the interpreter froze the thread.

Fix the class, not the site:
- tools/interpreter_shutdown.py: single shared shutdown predicate
  (matches both CPython message variants + sys.is_finalizing()).
- cron/scheduler.py, agent/tool_executor.py: existing per-site
  predicates now delegate to the shared home (tool_executor previously
  matched only the fuller variant).
- agent/conversation_loop.py: inner retry handler recognizes the
  shutdown signal and abandons the turn — one log warning, no print,
  no traceback, no debug dump, no retry; outer handler gets the same
  guard for shutdown errors raised outside the API call.
- The outer handler's bare print() now honors suppress_status_output
  (set by the background-review fork) instead of bypassing it.

Refs #55924 #58720 (same class in cron delivery), adjacent to #90683.

agent/conversation_loop.py

@@ -4675,6 +4675,39 @@ def run_conversation(
                 status_code = getattr(api_error, "status_code", None)
                 error_context = agent._extract_api_error_context(api_error)
 
+                # ── Interpreter finalization: abandon immediately ──
+                # The process is exiting (TUI quit, SIGTERM, one-shot done)
+                # while this turn — typically the post-turn review fork's
+                # daemon thread — is mid-flight. Retries, credential
+                # rotation, and fallbacks are all futile ("cannot schedule
+                # new futures..."), and the buffered ⚠️/❌ retry trace spams
+                # the shell after the TUI already exited. End the turn with
+                # a single log line: no print, no traceback, no debug dump,
+                # no retry. Same class as cron delivery (#55924/#58720) and
+                # concurrent tool submission — shared predicate.
+                from tools.interpreter_shutdown import interpreter_shutting_down
+
+                if interpreter_shutting_down(api_error):
+                    logger.warning(
+                        "%sInterpreter is shutting down — abandoning turn "
+                        "during API call #%d (%s)",
+                        agent.log_prefix, api_call_count, api_error,
+                    )
+                    _shutdown_summary = (
+                        "Turn abandoned: the process was shutting down "
+                        "before the model call could complete."
+                    )
+                    return {
+                        "final_response": _shutdown_summary,
+                        "messages": messages,
+                        "api_calls": api_call_count,
+                        "completed": False,
+                        "failed": True,
+                        "error": _shutdown_summary,
+                        "failure_reason": "interpreter_shutdown",
+                        "failure_retryable": False,
+                    }
+
                 # ── Classify the error for structured recovery decisions ──
                 _compressor = getattr(agent, "context_compressor", None)
                 _ctx_len = getattr(_compressor, "context_length", 200000) if _compressor else 200000
@@ -8377,6 +8410,34 @@ def run_conversation(
 
             _is_local_processing_error = _hit_local and not _hit_api
 
+            # Interpreter finalization: the process is exiting (TUI quit,
+            # SIGTERM, one-shot CLI done) while a background turn — most
+            # commonly the post-turn review fork's daemon thread — is still
+            # mid-request. Every further API attempt raises "cannot schedule
+            # new futures after interpreter shutdown", so retrying is futile
+            # and the un-gated ❌ prints spam the user's shell AFTER the TUI
+            # already exited (call #4, #5, #6...). Same shutdown-race class
+            # as cron delivery (#55924/#58720) and concurrent tool submission;
+            # shared predicate in tools/interpreter_shutdown.py. Log one
+            # warning, no traceback, no synthetic history append (nothing can
+            # persist it anymore), and leave the loop immediately.
+            from tools.interpreter_shutdown import interpreter_shutting_down
+
+            if interpreter_shutting_down(e):
+                logger.warning(
+                    "Interpreter is shutting down — abandoning turn after "
+                    "API call #%d (%s)",
+                    api_call_count,
+                    e,
+                )
+                _turn_exit_reason = "interpreter_shutdown"
+                failed = True
+                final_response = (
+                    "Turn abandoned: the process was shutting down before "
+                    "the model call could complete."
+                )
+                break
+
             if _is_local_processing_error:
                 error_msg = (
                     f"Error during local message processing after "
@@ -8384,10 +8445,20 @@ def run_conversation(
                 )
             else:
                 error_msg = f"Error during OpenAI-compatible API call #{api_call_count}: {str(e)}"
-            try:
-                print(f"❌ {error_msg}")
-            except (OSError, ValueError):
+            # The background-review fork sets suppress_status_output=True so
+            # lifecycle noise never reaches the user's terminal — but this
+            # bare print() bypassed it, leaking ❌ lines onto the shell after
+            # the TUI exited. Honor the same contract _vprint enforces
+            # (quiet_mode -q still shows hard failures, matching force=True
+            # semantics; suppress_status_output silences them). The
+            # logger.exception below still captures the full traceback.
+            if getattr(agent, "suppress_status_output", False):
                 logger.error(error_msg)
+            else:
+                try:
+                    print(f"❌ {error_msg}")
+                except (OSError, ValueError):
+                    logger.error(error_msg)
 
             # Emit the full traceback at ERROR level so it lands in both
             # agent.log AND errors.log.  Previously this was logged at DEBUG,

agent/tool_executor.py

@@ -274,7 +274,15 @@ def _ra():
 
 
 def _is_interpreter_shutdown_submit_error(exc: RuntimeError) -> bool:
-    return "cannot schedule new futures after interpreter shutdown" in str(exc)
+    """Shutdown-race predicate — shared home in ``tools.interpreter_shutdown``.
+
+    Delegates so all sites (cron delivery, conversation-loop retry, tool
+    submission) recognize both CPython shutdown-message variants instead of
+    each matching its own substring (the bug class behind #55924/#58720).
+    """
+    from tools.interpreter_shutdown import interpreter_shutting_down
+
+    return interpreter_shutting_down(exc)
 
 
 def _emit_terminal_post_tool_call(

cron/scheduler.py

@@ -1431,19 +1431,16 @@ def _interpreter_shutting_down(exc: Optional[BaseException] = None) -> bool:
     shutdown signal: the ``concurrent.futures`` module-global flag can be set
     a hair before ``sys.is_finalizing()`` flips, so matching the error text is
     a safe fallback for that race.
+
+    Thin wrapper — the predicate itself lives in
+    ``tools.interpreter_shutdown.interpreter_shutting_down`` (shared with the
+    conversation loop and the concurrent tool executor) so the shutdown-race
+    bug class is fixed in one place. Kept as a module symbol because tests
+    and callers throughout this file reference it by this name.
     """
-    if sys.is_finalizing():
-        return True
-    if exc is not None:
-        # Match the SHORT prefix deliberately: CPython emits two shutdown
-        # variants — "cannot schedule new futures after interpreter shutdown"
-        # (asyncio.run_coroutine_threadsafe / a torn-down default executor) and
-        # "cannot schedule new futures after shutdown" (a plain
-        # ThreadPoolExecutor). Both are documented in #58720. The common prefix
-        # catches both; the sibling agent/tool_executor._is_interpreter_shutdown_submit_error
-        # matches only the fuller "...after interpreter shutdown" form.
-        return "cannot schedule new futures" in str(exc).lower()
-    return False
+    from tools.interpreter_shutdown import interpreter_shutting_down
+
+    return interpreter_shutting_down(exc)
 
 
 # Backward-compatible module override used by tests and emergency monkeypatches.

tests/run_agent/test_interpreter_shutdown_turn_exit.py

@@ -0,0 +1,194 @@
+"""Interpreter-shutdown handling in the conversation loop's outer retry path.
+
+When the process starts tearing down (TUI quit, SIGTERM, one-shot exit) while
+a turn — most commonly the post-turn background-review fork's daemon thread —
+is still mid-request, every further API attempt raises ``RuntimeError:
+cannot schedule new futures after interpreter shutdown``.
+
+Before the fix, the outer loop treated that as a retryable API error: it
+printed an un-gated ``❌ Error during OpenAI-compatible API call #N`` line
+per attempt (spamming the user's shell AFTER the TUI already exited) and
+retried until the iteration budget or interpreter froze the thread.
+
+Now the loop recognizes the shutdown signal via the shared predicate in
+``tools.interpreter_shutdown`` (same class as cron delivery #55924/#58720
+and concurrent tool submission), logs one warning, and abandons the turn —
+no print, no traceback, no retry.
+"""
+
+from types import SimpleNamespace
+
+
+def _text_response(text: str):
+    return SimpleNamespace(
+        choices=[
+            SimpleNamespace(
+                message=SimpleNamespace(content=text, reasoning=None, tool_calls=[]),
+                finish_reason="stop",
+            )
+        ],
+        usage=None,
+    )
+
+
+class _ShutdownThenTextCompletions:
+    """First call raises the CPython shutdown error; later calls would succeed.
+
+    The 'would succeed' part is the sabotage detector: if the loop wrongly
+    retries after the shutdown signal, call #2 returns a normal response and
+    the assertions on call count / failed flag below fail loudly.
+    """
+
+    def __init__(self):
+        self.calls = 0
+
+    def create(self, **kwargs):
+        self.calls += 1
+        if self.calls == 1:
+            raise RuntimeError(
+                "cannot schedule new futures after interpreter shutdown"
+            )
+        return _text_response("should never be reached")
+
+
+class _AlwaysFailingCompletions:
+    def __init__(self):
+        self.calls = 0
+
+    def create(self, **kwargs):
+        self.calls += 1
+        raise Exception("API down")
+
+
+def _make_agent(monkeypatch, completions):
+    from run_agent import AIAgent
+
+    client = SimpleNamespace(chat=SimpleNamespace(completions=completions))
+    monkeypatch.setattr("run_agent.OpenAI", lambda **kwargs: client)
+    monkeypatch.setattr("run_agent.get_tool_definitions", lambda *a, **k: [])
+
+    agent = AIAgent(
+        model="test-model",
+        api_key="test-key",
+        base_url="http://localhost:8080/v1",
+        platform="cli",
+        max_iterations=4,
+        quiet_mode=True,
+        skip_memory=True,
+    )
+    agent._disable_streaming = True
+    return agent
+
+
+def test_shutdown_error_exits_loop_without_retry(monkeypatch, capsys):
+    completions = _ShutdownThenTextCompletions()
+    agent = _make_agent(monkeypatch, completions)
+
+    result = agent.run_conversation("hello")
+
+    # Precondition: the shutdown error actually fired on call #1.
+    assert completions.calls >= 1
+    # The core contract: NO retry after the shutdown signal.
+    assert completions.calls == 1, (
+        "conversation loop retried an API call after interpreter-shutdown "
+        f"signal (made {completions.calls} calls)"
+    )
+    assert result["failed"] is True
+    assert "shutting down" in result["final_response"]
+
+    # And no ❌ spam on the user's terminal.
+    out = capsys.readouterr()
+    assert "❌" not in out.out
+    assert "cannot schedule new futures" not in out.out
+
+
+def test_shutdown_variant_without_interpreter_word_also_exits(monkeypatch):
+    """CPython's plain-ThreadPoolExecutor variant omits 'interpreter'."""
+
+    class _Completions:
+        def __init__(self):
+            self.calls = 0
+
+        def create(self, **kwargs):
+            self.calls += 1
+            raise RuntimeError("cannot schedule new futures after shutdown")
+
+    completions = _Completions()
+    agent = _make_agent(monkeypatch, completions)
+
+    result = agent.run_conversation("hello")
+
+    assert completions.calls == 1
+    assert result["failed"] is True
+
+
+def test_suppress_status_output_gates_error_print_for_ordinary_api_errors(
+    monkeypatch, capsys
+):
+    """suppress_status_output routes ❌ lines to the log, not stdout.
+
+    This is the leak path from the field report: the background-review fork
+    runs with suppress_status_output=True (which gates _vprint and the
+    buffered retry trace), yet the bare print() in the outer error handler
+    bypassed it.
+    """
+    completions = _AlwaysFailingCompletions()
+    agent = _make_agent(monkeypatch, completions)
+    agent.suppress_status_output = True
+
+    result = agent.run_conversation("hello")
+
+    # Precondition: the ordinary (non-shutdown) error path actually retried
+    # up to the budget — this test exercises the print gating, not early exit.
+    assert completions.calls > 1
+    assert "API down" in result["final_response"]
+
+    out = capsys.readouterr()
+    assert "❌" not in out.out
+
+
+def test_non_quiet_mode_still_prints_error(monkeypatch, capsys):
+    completions = _AlwaysFailingCompletions()
+    agent = _make_agent(monkeypatch, completions)
+    agent.quiet_mode = False
+
+    agent.run_conversation("hello")
+
+    out = capsys.readouterr()
+    assert "❌" in out.out
+
+
+class TestSharedPredicate:
+    def test_matches_interpreter_variant(self):
+        from tools.interpreter_shutdown import interpreter_shutting_down
+
+        exc = RuntimeError("cannot schedule new futures after interpreter shutdown")
+        assert interpreter_shutting_down(exc) is True
+
+    def test_matches_plain_executor_variant(self):
+        from tools.interpreter_shutdown import interpreter_shutting_down
+
+        exc = RuntimeError("cannot schedule new futures after shutdown")
+        assert interpreter_shutting_down(exc) is True
+
+    def test_ignores_unrelated_errors(self):
+        from tools.interpreter_shutdown import interpreter_shutting_down
+
+        assert interpreter_shutting_down(RuntimeError("boom")) is False
+        assert interpreter_shutting_down(None) is False
+
+    def test_cron_wrapper_delegates(self):
+        from cron.scheduler import _interpreter_shutting_down
+
+        exc = RuntimeError("cannot schedule new futures after shutdown")
+        assert _interpreter_shutting_down(exc) is True
+        assert _interpreter_shutting_down(RuntimeError("boom")) is False
+
+    def test_tool_executor_wrapper_delegates(self):
+        from agent.tool_executor import _is_interpreter_shutdown_submit_error
+
+        # The tool-executor predicate previously matched ONLY the fuller
+        # variant; via the shared home it now catches both.
+        exc = RuntimeError("cannot schedule new futures after shutdown")
+        assert _is_interpreter_shutdown_submit_error(exc) is True
+        assert _is_interpreter_shutdown_submit_error(RuntimeError("boom")) is False

tools/interpreter_shutdown.py

@@ -0,0 +1,56 @@
+"""Shared interpreter-shutdown detection.
+
+Single home for the "is the Python interpreter finalizing?" predicate used
+by every subsystem whose background threads can outlive process teardown
+(cron delivery, concurrent tool submission, the conversation loop's retry
+path, background review forks).
+
+Once finalization starts, ``concurrent.futures`` refuses new work with
+``RuntimeError: cannot schedule new futures after interpreter shutdown`` and
+asyncio's default executor is gone — *any* further attempt to schedule work
+(an API retry, a thread-pool submit, ``asyncio.run``) is doomed and only
+produces noise: stray ``❌`` prints after the TUI exited, tracebacks in
+``errors.log``, and futile retry loops that burn iterations against a dying
+process (#55924, #58720, and the CLI-exit retry spam this module was
+extracted for).
+
+CPython emits two message variants depending on the failing site:
+
+- ``cannot schedule new futures after interpreter shutdown`` — the
+  module-global finalization flag (asyncio.run_coroutine_threadsafe, a
+  torn-down default executor, ThreadPoolExecutor.submit during teardown).
+- ``cannot schedule new futures after shutdown`` — a plain
+  ``ThreadPoolExecutor`` whose ``shutdown()`` ran.
+
+The common short prefix catches both. Matching the second variant is safe
+for shutdown detection at every current call site: the pools involved are
+either module-global daemons or ``with``-scoped locals that cannot be shut
+down mid-use by anything except interpreter finalization.
+
+Historically this predicate existed at three sites, each fixed
+independently as its own incident — ``cron/scheduler.py`` (#55924/#58720),
+``agent/tool_executor.py``, and nothing at all in the conversation loop's
+outer retry handler (the CLI-exit spam). One predicate, all sites.
+"""
+
+from __future__ import annotations
+
+import sys
+from typing import Optional
+
+_SHUTDOWN_SUBMIT_ERROR_PREFIX = "cannot schedule new futures"
+
+
+def interpreter_shutting_down(exc: Optional[BaseException] = None) -> bool:
+    """Return True when the Python interpreter is finalizing.
+
+    ``exc`` lets a caller also treat an already-raised scheduling error as a
+    shutdown signal: the ``concurrent.futures`` module-global flag can be set
+    a hair before ``sys.is_finalizing()`` flips, so matching the error text
+    is a safe fallback for that race.
+    """
+    if sys.is_finalizing():
+        return True
+    if exc is not None:
+        return _SHUTDOWN_SUBMIT_ERROR_PREFIX in str(exc).lower()
+    return False