fix(interrupt/auth): prevent /stop swallow (incl. Bedrock) and empty-provider credential corruption

agent/agent_runtime_helpers.py

@@ -729,7 +729,14 @@ def recover_with_credential_pool(
     # that seeded the pool.
     current_provider = (getattr(agent, "provider", "") or "").strip().lower()
     pool_provider = (getattr(pool, "provider", "") or "").strip().lower()
-    if current_provider and pool_provider and current_provider != pool_provider:
+    # Guard: skip credential pool recovery when the pool is scoped to a
+    # different provider than the agent.  Only guard when the pool has a
+    # known provider — an empty pool provider means "unscoped" (applies to
+    # any provider).  An empty agent provider is treated as a mismatch
+    # because swapping the pool's credentials would set base_url/api_key
+    # without fixing the empty provider field, leaving the agent in a
+    # corrupted state (provider="" model="").
+    if pool_provider and current_provider != pool_provider:
         # Custom endpoints use two naming conventions for the SAME provider:
         # the agent carries the generic ``custom`` label while the pool is
         # keyed ``custom:<name>`` (see CUSTOM_POOL_PREFIX). A literal string

agent/chat_completion_helpers.py

@@ -1881,6 +1881,16 @@ def _on_reasoning(text):
             t.join(timeout=0.3)
             if agent._interrupt_requested:
                 raise InterruptedError("Agent interrupted during Bedrock API call")
+        # Worker exited before the poll loop observed the interrupt flag. The
+        # Bedrock stream callback breaks out and returns a PARTIAL response
+        # without raising on interrupt (see bedrock_adapter.py
+        # stream_converse_with_callbacks / on_interrupt_check), so result[
+        # "response"] is populated with error=None and the in-loop raise above
+        # never fires. Re-check here so /stop is not silently swallowed on the
+        # Bedrock path — mirrors the post-worker guard on the main streaming
+        # loop. (#59999 area)
+        if agent._interrupt_requested:
+            raise InterruptedError("Agent interrupted during Bedrock API call (post-worker)")
         if result["error"] is not None:
             raise result["error"]
         return result["response"]
@@ -2902,6 +2912,13 @@ def _call():
             except Exception:
                 pass
             raise InterruptedError("Agent interrupted during streaming API call")
+    # Worker thread exited before the main thread's poll loop could check
+    # the interrupt flag.  If the worker returned early due to an interrupt
+    # (e.g. _call_anthropic() detected _interrupt_requested and returned
+    # None), the InterruptedError above was never raised.  Re-check the
+    # flag here so /stop is not silently swallowed.  (#59999 area)
+    if agent._interrupt_requested:
+        raise InterruptedError("Agent interrupted during streaming API call (post-worker)")
     if result["error"] is not None:
         if deltas_were_sent["yes"]:
             # Streaming failed AFTER some tokens were already delivered to

tests/agent/test_bedrock_interrupt_post_worker.py

@@ -0,0 +1,88 @@
+"""Regression: /stop must not be swallowed on the Bedrock streaming path.
+
+Companion to the OpenAI/Anthropic streaming post-worker guard. The Bedrock
+Converse stream callback (bedrock_adapter.stream_converse_with_callbacks) breaks
+out of its event loop on interrupt and returns a PARTIAL response WITHOUT
+raising. The worker thread then sets result["response"] and exits cleanly with
+agent._interrupt_requested still True. Without a post-worker re-check in the
+poll loop, interruptible_streaming_api_call would return that partial response
+and silently swallow the /stop signal.
+"""
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from agent import chat_completion_helpers as cch
+
+
+class _FakeAgent:
+    api_mode = "bedrock_converse"
+    _interrupt_requested = False  # not interrupted at entry (passes pre-flight)
+    _disable_streaming = False
+    reasoning_callback = None
+    stream_delta_callback = None
+
+    def _has_stream_consumers(self):
+        return False
+
+    def _fire_stream_delta(self, text):
+        pass
+
+    def _fire_tool_gen_started(self, name):
+        pass
+
+    def _fire_reasoning_delta(self, text):
+        pass
+
+    def _safe_print(self, *a, **k):
+        pass
+
+
+def test_bedrock_stream_interrupt_not_swallowed_post_worker():
+    """A /stop arriving MID-stream: the pre-flight check (top of function) has
+    already passed, the worker's stream callback breaks and returns a partial
+    response WITHOUT raising, leaving _interrupt_requested True. The post-worker
+    re-check must raise InterruptedError instead of returning the partial."""
+    agent = _FakeAgent()
+
+    partial = SimpleNamespace(choices=[], usage=None, stop_reason="interrupted")
+
+    # Simulate the real adapter: on interrupt it breaks out and returns a
+    # partial response WITHOUT raising. Flip the interrupt flag here to model
+    # /stop arriving mid-stream (after the pre-flight check, during the worker).
+    def _fake_stream(*args, **kwargs):
+        agent._interrupt_requested = True
+        return partial
+
+    fake_client = SimpleNamespace(converse_stream=lambda **kw: {"stream": []})
+
+    with patch("agent.bedrock_adapter._get_bedrock_runtime_client", return_value=fake_client), \
+         patch("agent.bedrock_adapter.stream_converse_with_callbacks", side_effect=_fake_stream), \
+         patch("agent.bedrock_adapter.normalize_converse_response", side_effect=lambda r: r), \
+         patch("agent.bedrock_adapter.is_stale_connection_error", return_value=False), \
+         patch("agent.bedrock_adapter.is_streaming_access_denied_error", return_value=False), \
+         patch("agent.bedrock_adapter.invalidate_runtime_client", lambda *a, **k: None):
+        api_kwargs = {"__bedrock_region__": "us-east-1", "__bedrock_converse__": True}
+        with pytest.raises(InterruptedError):
+            cch.interruptible_streaming_api_call(agent, api_kwargs)
+
+
+def test_bedrock_stream_returns_normally_when_not_interrupted():
+    """Sanity: with no interrupt, the same path returns the response (guard
+    must not fire spuriously)."""
+    agent = _FakeAgent()
+    agent._interrupt_requested = False
+
+    resp = SimpleNamespace(choices=[], usage=None, stop_reason="end_turn")
+    fake_client = SimpleNamespace(converse_stream=lambda **kw: {"stream": []})
+
+    with patch("agent.bedrock_adapter._get_bedrock_runtime_client", return_value=fake_client), \
+         patch("agent.bedrock_adapter.stream_converse_with_callbacks", return_value=resp), \
+         patch("agent.bedrock_adapter.normalize_converse_response", side_effect=lambda r: r), \
+         patch("agent.bedrock_adapter.is_stale_connection_error", return_value=False), \
+         patch("agent.bedrock_adapter.is_streaming_access_denied_error", return_value=False), \
+         patch("agent.bedrock_adapter.invalidate_runtime_client", lambda *a, **k: None):
+        api_kwargs = {"__bedrock_region__": "us-east-1", "__bedrock_converse__": True}
+        out = cch.interruptible_streaming_api_call(agent, api_kwargs)
+        assert out is resp

tests/agent/test_credential_pool_routing.py

@@ -155,6 +155,9 @@ def _make_agent_with_pool(self, pool_entries=3):
 
         pool = MagicMock()
         pool.has_credentials.return_value = True
+        # Must be set explicitly — MagicMock.provider returns a truthy
+        # child mock, which would trigger the provider-mismatch guard.
+        pool.provider = ""
 
         # mark_exhausted_and_rotate returns next entry until exhausted
         self._rotation_index = 0

tests/run_agent/test_credential_pool_interrupt.py

@@ -27,6 +27,9 @@ def _make_pool(entries):
     pool = MagicMock()
     pool.entries = entries
     pool.current.return_value = entries[0]
+    # Must be set explicitly — MagicMock.provider returns a truthy
+    # child mock, which would trigger the provider-mismatch guard.
+    pool.provider = ""
     return pool