fix(runtime): recover reset streams and deduplicate terminal model errors

agent/agent_runtime_helpers.py

@@ -25,7 +25,7 @@
 from agent.credential_pool import (
     STATUS_EXHAUSTED, credential_pool_matches_provider, resolve_runtime_pool_key
 )
-from agent.error_classifier import FailoverReason
+from agent.error_classifier import TRANSPORT_ERROR_TYPES, FailoverReason
 from agent.turn_context import drop_stale_api_content
 from utils import base_url_host_matches, base_url_hostname, env_var_enabled, atomic_json_write
 logger = logging.getLogger(__name__)
@@ -1195,10 +1195,16 @@ def _load_primary_pool():
 
 
 # Transient transport failures worth one more attempt with a rebuilt client / connection pool.
-_TRANSIENT_TRANSPORT_ERRORS = frozenset({
-    "ReadTimeout", "ConnectTimeout", "PoolTimeout", "ConnectError", "RemoteProtocolError",
-    "APIConnectionError", "APITimeoutError",
-})
+# Derived from the canonical classifier so the gate cannot drift from "what counts as a
+# transport fault": a hand-maintained copy omitted ``ReadError`` — the shape a mid-response
+# ``[Errno 104] Connection reset by peer`` takes on routes that read the body themselves —
+# so the one connection the reset poisoned was never retired and every reset skipped the
+# rebuild. This widens WHICH failures get the single rebuild, not how many attempts anything
+# gets: recovery still fires at most once per API-call block, after the classifier already
+# called the error retryable and the normal retry budget is spent.
+# ``PoolTimeout`` stays a local addition: it is a fault of the pool this function rebuilds,
+# which is narrower than the classifier's question.
+_TRANSIENT_TRANSPORT_ERRORS = TRANSPORT_ERROR_TYPES | {"PoolTimeout"}
 _INLINE_REASONING_PATTERNS = tuple(
     re.compile(rf"<{tag}>(.*?)</{tag}>", re.DOTALL | re.IGNORECASE)
     for tag in ("think", "thinking", "thought", "reasoning", "REASONING_SCRATCHPAD")

agent/error_classifier.py

@@ -285,7 +285,11 @@ def billing_unverified(self) -> bool:
 
 # SSL names keep provider-wrapped SSL errors (chain lost) as transport, not
 # unknown; OpenAI SDK errors are not subclasses of Python builtins.
-_TRANSPORT_ERROR_TYPES = frozenset({
+# Public: this is the canonical answer to "is this exception type a transport fault?".
+# ``agent/agent_runtime_helpers.py`` derives its client-rebuild gate from it so the two
+# cannot drift (they did: ``ReadError`` — the fleet's most common reset shape — was
+# classified as transport here but missing from the rebuild gate).
+TRANSPORT_ERROR_TYPES = frozenset({
     "ReadTimeout", "ConnectTimeout", "PoolTimeout", "ConnectError", "RemoteProtocolError",
     "ConnectionError", "ConnectionResetError", "ConnectionAbortedError", "BrokenPipeError",
     "TimeoutError", "ReadError", "ServerDisconnectedError",
@@ -571,7 +575,7 @@ def _by_transport(c: _Ctx) -> Optional[Verdict]:
     # network call): as ``unknown`` it would burn every retry instantly.
     if c.error_type == "RuntimeError" and "consecutive stale attempts" in msg and "aborting this call" in msg:
         return _v(_R.timeout, **_ABORT_FALLBACK)
-    transport = c.error_type in _TRANSPORT_ERROR_TYPES or isinstance(c.error, (TimeoutError, ConnectionError, OSError))
+    transport = c.error_type in TRANSPORT_ERROR_TYPES or isinstance(c.error, (TimeoutError, ConnectionError, OSError))
     return _V_TIMEOUT if transport else None
 
 

gateway/run.py

@@ -391,6 +391,25 @@ def _gateway_surface_passes_raw_text(platform: Any) -> bool:
     r"cannot\s+connect", r"failed\s+to\s+establish", r"could\s+not\s+connect")
 _GATEWAY_CONNECTION_ERROR_RE = re.compile("(" + "|".join(_CONNECTION_ERROR_MARKERS) + ")", re.IGNORECASE)
 
+# An ESTABLISHED connection died mid-transfer. Says nothing about whether the endpoint is up —
+# in the reported incident an earlier call in the same turn had already been answered by it.
+_CONNECTION_INTERRUPTED_MARKERS = (
+    r"connection\s+reset", r"connection\s+aborted", r"errno\s+104", r"errno\s+103",
+    r"broken\s+pipe", r"server\s+disconnected", r"peer\s+closed\s+connection",
+    r"connection\s+was\s+closed", r"network\s+connection\s+lost", r"unexpected\s+eof",
+    r"incomplete\s+chunked\s+read", r"response\s+ended\s+prematurely", r"socket\s+hang\s+up",
+    r"(?:\w+\.)?remoteprotocolerror", r"(?:\w+\.)?readerror")
+_GATEWAY_CONNECTION_INTERRUPTED_RE = re.compile(
+    "(" + "|".join(_CONNECTION_INTERRUPTED_MARKERS) + ")", re.IGNORECASE)
+
+# Nothing accepted the connection / no path to the host: "the endpoint is not up" IS the diagnosis.
+_ENDPOINT_UNREACHABLE_MARKERS = (
+    r"connection\s+refused", r"actively\s+refused", r"winerror\s+10061", r"errno\s+111",
+    r"no\s+route\s+to\s+host", r"network\s+is\s+unreachable", r"cannot\s+connect",
+    r"failed\s+to\s+establish", r"could\s+not\s+connect", r"(?:\w+\.)?connect\s*(?:error|timeout)")
+_GATEWAY_ENDPOINT_UNREACHABLE_RE = re.compile(
+    "(" + "|".join(_ENDPOINT_UNREACHABLE_MARKERS) + ")", re.IGNORECASE)
+
 _GATEWAY_SECRET_PATTERNS = (
     re.compile(r"\bsk-[A-Za-z0-9][A-Za-z0-9_\-]{12,}\b"),
     re.compile(r"\bgh[pousr]_[A-Za-z0-9_]{20,}\b"), re.compile(r"\bxapp-\d+-[A-Za-z0-9\-]{20,}\b"),
@@ -600,14 +619,32 @@ def _format_exec_approval_fallback(
         + ", ".join(choices[:-1]) + f", or {choices[-1]}.")
 
 # Ordered: auth beats policy beats rate-limit beats connection; first match wins.
+#
+# The three connection rows are NOT interchangeable, and collapsing them is what made a
+# mid-response TCP reset read as "your model server is down" (2026-09-06) while the same turn
+# had already been answered by that endpoint:
+#   * interrupted  — an established connection died mid-transfer. Whether the endpoint is up is
+#                    unknown from this alone, so we must not guess.
+#   * unreachable  — nothing accepted the connection at all; "not running / unreachable" is the
+#                    actual diagnosis, and this is the case that wording was written for (#86570).
+#   * ambiguous    — connection-shaped but the cause was flattened away (an SDK-wrapped
+#                    ``APIConnectionError: Connection error.`` keeps neither). Name both
+#                    possibilities; assert neither.
 _PROVIDER_ERROR_REPLIES = (
     (_GATEWAY_AUTH_ERROR_RE, "⚠️ Provider authentication failed. Check the configured credentials; "
                              "raw provider details are in the gateway logs."),
     (_GATEWAY_PROVIDER_POLICY_RE, "⚠️ The model provider rejected the request. I kept the raw provider "
                                   "error out of chat; check gateway logs for details or try rephrasing."),
     (_GATEWAY_RATE_LIMIT_RE, "⏱️ The model provider is rate-limiting requests. Please wait a moment and try again."),
-    (_GATEWAY_CONNECTION_ERROR_RE, "⚠️ The model server is not responding — it looks like the configured "
-                                   "model endpoint is not running or is unreachable."))
+    (_GATEWAY_CONNECTION_INTERRUPTED_RE, "⚠️ The connection to the model provider was interrupted before the "
+                                         "reply arrived, and the retries hit the same problem. Please try "
+                                         "again; the transport details are in the gateway logs."),
+    (_GATEWAY_ENDPOINT_UNREACHABLE_RE, "⚠️ The model server is not responding — it looks like the configured "
+                                       "model endpoint is not running or is unreachable."),
+    (_GATEWAY_CONNECTION_ERROR_RE, "⚠️ The model request could not be completed over the network after "
+                                   "retries — the connection was either never established or dropped before "
+                                   "the reply arrived. Please try again; if it keeps happening, check that the "
+                                   "configured model endpoint is reachable. Details are in the gateway logs."))
 
 
 def _gateway_provider_error_reply(text: str) -> str:
@@ -689,7 +726,14 @@ def _prepare_gateway_status_message(platform: Any, event_type: str, message: str
     ):
         return None
     if _looks_like_gateway_provider_error(text):
-        return _gateway_provider_error_reply(text)
+        # The status and the turn's final response are the SAME failure: the agent's terminal
+        # paths (agent/turn_recovery.py) emit the envelope through status_callback and then
+        # return it as final_response, and _sanitize_gateway_final_response maps both onto the
+        # same reply here — so delivering this one posts the identical bubble twice
+        # (2026-09-06: three connection resets, one warning shown twice). Chat surfaces get the
+        # provider-failure category from the final response only; the raw diagnostic still goes
+        # to the logs and to the programmatic surfaces above.
+        return None
     return text
 
 

tests/gateway/test_connection_error_reply_wording.py

@@ -0,0 +1,139 @@
+"""Connection failures must be described accurately on chat surfaces.
+
+A mid-response `[Errno 104] Connection reset by peer` is NOT evidence that the
+configured endpoint is down — the same turn had already completed an earlier call
+against it. Telling the user "the endpoint is not running" sends them debugging a
+server that is running (reported 2026-09-06).
+
+The refusal case (nothing listening: ECONNREFUSED / WinError 10061 / no route) IS a
+"your endpoint is not up" diagnosis and must keep it, as must the auth, policy and
+rate-limit classifications.
+"""
+
+import pytest
+
+from gateway.config import Platform
+from gateway.run import (
+    _gateway_provider_error_reply,
+    _sanitize_gateway_final_response,
+)
+
+CHAT_PLATFORMS = [Platform.TELEGRAM, "slack", "feishu"]
+
+# Envelopes the agent's terminal path actually produces for a mid-transfer drop: an
+# established connection died, which says nothing about whether the endpoint is up.
+INTERRUPTED_ENVELOPES = [
+    "API call failed after 3 retries: httpx.ReadError: [Errno 104] Connection reset by peer",
+    "API call failed after 3 retries: ConnectionResetError: [Errno 104] Connection reset by peer",
+    "API call failed after 3 retries: httpx.RemoteProtocolError: peer closed connection "
+    "without sending complete message body",
+    "API call failed after 3 retries: httpx.ReadError: server disconnected without sending a response",
+]
+
+# Envelopes that really do mean "nothing is listening at the configured endpoint".
+UNREACHABLE_ENVELOPES = [
+    "API call failed after 3 retries: httpx.ConnectError: [Errno 111] Connection refused",
+    "API call failed after 3 retries: ConnectionError: [WinError 10061] No connection could "
+    "be made because the target machine actively refused it",
+    "API call failed after 3 retries: httpx.ConnectError: [Errno 113] No route to host",
+]
+
+# Connection-shaped, but the SDK flattened the cause away — neither diagnosis is supported.
+AMBIGUOUS_ENVELOPES = [
+    "API call failed after 3 retries: openai.APIConnectionError: Connection error.",
+    "❌ API failed after 3 retries — openai.APIConnectionError: Connection error.",
+]
+
+# Wording that ASSERTS the endpoint is down / not started.
+_DOWN_DIAGNOSIS_CLAIMS = ("is not running", "not responding", "is unreachable", "not started")
+
+
+def _claims_endpoint_is_down(reply: str) -> bool:
+    return any(claim in reply.lower() for claim in _DOWN_DIAGNOSIS_CLAIMS)
+
+
+@pytest.mark.parametrize("envelope", INTERRUPTED_ENVELOPES)
+def test_interrupted_connection_is_not_diagnosed_as_a_dead_endpoint(envelope):
+    """A reset/dropped response must be reported as an interruption, not a dead server."""
+    reply = _gateway_provider_error_reply(envelope)
+
+    assert reply
+    assert not _claims_endpoint_is_down(reply), reply
+
+
+@pytest.mark.parametrize("envelope", UNREACHABLE_ENVELOPES)
+def test_refused_connection_keeps_the_endpoint_down_diagnosis(envelope):
+    """A refused/unroutable connect is exactly the case the old wording was written for."""
+    reply = _gateway_provider_error_reply(envelope)
+
+    assert _claims_endpoint_is_down(reply), reply
+
+
+@pytest.mark.parametrize("envelope", AMBIGUOUS_ENVELOPES)
+def test_ambiguous_connection_error_asserts_neither_cause(envelope):
+    """An SDK-flattened ``Connection error.`` supports no diagnosis — so make none."""
+    reply = _gateway_provider_error_reply(envelope)
+
+    assert not _claims_endpoint_is_down(reply), reply
+    # It still has to be actionable: the user is told what to check, not what is broken.
+    assert "reachable" in reply.lower(), reply
+
+
+def test_the_three_connection_causes_are_distinct_categories():
+    """The causes must not collapse onto one message again."""
+    interrupted = {_gateway_provider_error_reply(e) for e in INTERRUPTED_ENVELOPES}
+    unreachable = {_gateway_provider_error_reply(e) for e in UNREACHABLE_ENVELOPES}
+    ambiguous = {_gateway_provider_error_reply(e) for e in AMBIGUOUS_ENVELOPES}
+
+    assert len(interrupted) == len(unreachable) == len(ambiguous) == 1
+    assert len(interrupted | unreachable | ambiguous) == 3
+
+
+@pytest.mark.parametrize(
+    "envelope, expected_marker",
+    [
+        (
+            "API call failed after 3 retries: HTTP 401 Unauthorized: incorrect api key provided",
+            "authentication",
+        ),
+        (
+            "API call failed after 3 retries: HTTP 429: rate limit exceeded for this model",
+            "rate-limiting",
+        ),
+        (
+            "API call failed after 3 retries: HTTP 400: request blocked under the provider "
+            "safety policy",
+            "rejected",
+        ),
+    ],
+    ids=["auth", "rate_limit", "policy"],
+)
+def test_other_provider_error_classifications_are_preserved(envelope, expected_marker):
+    """Auth beats policy beats rate-limit beats connection — that ordering still holds."""
+    reply = _gateway_provider_error_reply(envelope)
+
+    assert expected_marker in reply.lower(), reply
+
+
+@pytest.mark.parametrize("platform", CHAT_PLATFORMS)
+def test_reset_final_response_is_sanitized_and_secret_free(platform):
+    """The reset envelope still goes through redaction + the safe-category rewrite."""
+    raw = (
+        "API call failed after 3 retries: httpx.ReadError: [Errno 104] Connection reset "
+        "by peer (Authorization: Bearer sk-ABCDEF0123456789abcdef0123)"
+    )
+
+    sanitized = _sanitize_gateway_final_response(platform, raw)
+
+    assert "sk-ABCDEF" not in sanitized
+    assert "Errno 104" not in sanitized
+    assert not _claims_endpoint_is_down(sanitized), sanitized
+    assert sanitized.strip()
+
+
+@pytest.mark.parametrize("platform", ["local", "api_server", "webhook"])
+def test_programmatic_surfaces_keep_the_raw_connection_error(platform):
+    """CLI/API consumers still need the bottom exception, not a chat-safe category."""
+    raw = "API call failed after 3 retries: httpx.ReadError: [Errno 104] Connection reset by peer"
+
+    assert _sanitize_gateway_final_response(platform, raw) == raw

tests/gateway/test_local_model_connection_reply.py

@@ -11,8 +11,8 @@
 
 class TestGatewayConnectionErrorReply:
     def test_connection_error_strings_produce_specific_reply(self):
+        """A connect that was REFUSED/unroutable is the local-endpoint-down case."""
         samples = [
-            "openai.APIConnectionError",
             "httpx.ConnectError: connection refused",
             "ConnectionError: [WinError 10061] No connection could be made",
             "Errno 111 Connection refused",
@@ -24,6 +24,21 @@ def test_connection_error_strings_produce_specific_reply(self):
             assert "not responding" in reply.lower(), text
             assert "not running or is unreachable" in reply, text
 
+    def test_cause_free_connection_error_points_at_reachability_without_asserting_it(self):
+        """A bare ``APIConnectionError`` kept no cause: it may equally be a dropped reply.
+
+        Still a provider-error envelope, and still points the user at endpoint
+        reachability — but it no longer *states* that the endpoint is down (issue #15:
+        the same wording was shown for a mid-response TCP reset from a live endpoint).
+        """
+        text = "openai.APIConnectionError"
+
+        assert _looks_like_gateway_provider_error(text)
+        reply = _gateway_provider_error_reply(text)
+
+        assert "reachable" in reply.lower()
+        assert "not running or is unreachable" not in reply
+
     def test_broad_connection_phrases_still_map_once_classified(self):
         """Reply selector keeps the full phrase set; the gate does not."""
         for text in (

tests/gateway/test_telegram_noise_filter.py

@@ -255,20 +255,24 @@ def test_chat_gateways_drop_interrupt_sentinel(platform):
     assert _sanitize_gateway_final_response("local", sentinel) == sentinel
 
 
-def test_telegram_status_sanitizes_raw_provider_security_errors():
-    """Provider policy/security bodies should be replaced before chat delivery."""
+def test_telegram_status_never_delivers_raw_provider_security_errors():
+    """Provider policy/security bodies must not reach chat through the status lane.
+
+    The terminal failure envelope is delivered ONCE, as the turn's final response
+    (test_telegram_final_response_sanitizes_raw_provider_errors below) — the status
+    copy is dropped so the user does not get the same warning twice. Whatever the
+    delivery decision, the raw body/request id must never survive it.
+    """
     raw = (
         "❌ API failed after 3 retries — HTTP 400: request blocked because "
         "Operation contains cybersecurity risk. request_id=req_123"
     )
 
     sanitized = _prepare_gateway_status_message(Platform.TELEGRAM, "lifecycle", raw)
 
-    assert sanitized is not None
-    assert "provider rejected" in sanitized.lower()
-    assert "cybersecurity risk" not in sanitized.lower()
-    assert "HTTP 400" not in sanitized
-    assert "req_123" not in sanitized
+    assert sanitized is None
+    # The programmatic surfaces still get the full diagnostic.
+    assert _prepare_gateway_status_message("local", "lifecycle", raw) == raw
 
 
 def test_telegram_final_response_sanitizes_raw_provider_errors():