From df5daa34479ddd292c4d0af76166b35f24f31bfd Mon Sep 17 00:00:00 2001
From: palanivelg <palanivelg@ieee.org>
Date: Wed, 8 Jul 2026 23:24:02 +0000
Subject: [PATCH 1/3] fix(endpoint-client): retry pre-response connection
 resets (opt-in)

Single-slot servers that recycle idle keep-alive sockets between requests
can close a socket at the instant the client reuses it. The write then
races the close and raises a connection reset before any response byte
arrives, which previously failed the request outright -- and on an
accuracy run, silently zeroed the affected sample.

Add Worker._read_headers_with_retry: on a connection failure before any
response byte, discard the dead connection and re-issue the identical
request on a fresh one. It never fires once headers/chunks have started,
so it cannot duplicate output. Gated by a new HTTPClientConfig field
transport_max_retries (default 0 = prior fail-fast behavior, so every
existing use case is unchanged); the edge-agentic accuracy reference
config opts in explicitly.

Adds unit tests covering retry-success, retries-exhausted,
retries-disabled, the no-reset fast path, and the opt-in default.
---
 .../online_edge_full_run.yaml                 |   6 +
 .../templates/concurrency_template_full.yaml  |   1 +
 .../templates/offline_template_full.yaml      |   1 +
 .../templates/online_template_full.yaml       |   1 +
 .../endpoint_client/config.py                 |  17 ++
 .../endpoint_client/worker.py                 |  46 ++++-
 .../unit/endpoint_client/test_worker_retry.py | 192 ++++++++++++++++++
 7 files changed, 258 insertions(+), 6 deletions(-)
 create mode 100644 tests/unit/endpoint_client/test_worker_retry.py

diff --git a/examples/11_Edge_Agentic_Example/online_edge_full_run.yaml b/examples/11_Edge_Agentic_Example/online_edge_full_run.yaml
index e8bd1e418..647aab3de 100644
--- a/examples/11_Edge_Agentic_Example/online_edge_full_run.yaml
+++ b/examples/11_Edge_Agentic_Example/online_edge_full_run.yaml
@@ -106,6 +106,12 @@ settings:
     num_workers: 1
     max_connections: 1
     warmup_connections: 0
+    # Single-slot edge servers (e.g. llama.cpp -np 1) close idle keep-alive
+    # sockets between samples; a reused-then-reset socket would otherwise zero
+    # the affected accuracy sample. Retry the request on a fresh connection when
+    # the reset happens before any response byte (safe: the server produced no
+    # output). Off by default globally; enabled here for the gated run.
+    transport_max_retries: 2
 
 endpoint_config:
   endpoints:
diff --git a/src/inference_endpoint/config/templates/concurrency_template_full.yaml b/src/inference_endpoint/config/templates/concurrency_template_full.yaml
index 73e9816a5..ce3f172fd 100644
--- a/src/inference_endpoint/config/templates/concurrency_template_full.yaml
+++ b/src/inference_endpoint/config/templates/concurrency_template_full.yaml
@@ -78,6 +78,7 @@ settings:
     worker_force_kill_timeout: 0.5  # Force kill timeout after graceful wait (seconds)
     insecure: false  # Skip TLS certificate verification
     max_idle_time: 4.0  # Discard connections idle longer than this (seconds)
+    transport_max_retries: 0  # Retries on a pre-response connection reset (0=disabled)
     min_required_connections: -1  # Min connections to initialize (-1=auto, 0=disabled)
     worker_gc_mode: relaxed  # Worker GC strategy | options: disabled, relaxed, system
   drain:
diff --git a/src/inference_endpoint/config/templates/offline_template_full.yaml b/src/inference_endpoint/config/templates/offline_template_full.yaml
index 9c6a602f5..681170fc7 100644
--- a/src/inference_endpoint/config/templates/offline_template_full.yaml
+++ b/src/inference_endpoint/config/templates/offline_template_full.yaml
@@ -78,6 +78,7 @@ settings:
     worker_force_kill_timeout: 0.5  # Force kill timeout after graceful wait (seconds)
     insecure: false  # Skip TLS certificate verification
     max_idle_time: 4.0  # Discard connections idle longer than this (seconds)
+    transport_max_retries: 0  # Retries on a pre-response connection reset (0=disabled)
     min_required_connections: -1  # Min connections to initialize (-1=auto, 0=disabled)
     worker_gc_mode: relaxed  # Worker GC strategy | options: disabled, relaxed, system
   drain:
diff --git a/src/inference_endpoint/config/templates/online_template_full.yaml b/src/inference_endpoint/config/templates/online_template_full.yaml
index 78965a20a..604fef9f9 100644
--- a/src/inference_endpoint/config/templates/online_template_full.yaml
+++ b/src/inference_endpoint/config/templates/online_template_full.yaml
@@ -79,6 +79,7 @@ settings:
     worker_force_kill_timeout: 0.5  # Force kill timeout after graceful wait (seconds)
     insecure: false  # Skip TLS certificate verification
     max_idle_time: 4.0  # Discard connections idle longer than this (seconds)
+    transport_max_retries: 0  # Retries on a pre-response connection reset (0=disabled)
     min_required_connections: -1  # Min connections to initialize (-1=auto, 0=disabled)
     worker_gc_mode: relaxed  # Worker GC strategy | options: disabled, relaxed, system
   drain:
diff --git a/src/inference_endpoint/endpoint_client/config.py b/src/inference_endpoint/endpoint_client/config.py
index a0198cb5c..ce170e4b4 100644
--- a/src/inference_endpoint/endpoint_client/config.py
+++ b/src/inference_endpoint/endpoint_client/config.py
@@ -170,6 +170,23 @@ class HTTPClientConfig(WithUpdatesMixin, BaseModel):
         4.0, description="Discard connections idle longer than this (seconds)"
     )
 
+    # Retry a request when the connection fails BEFORE any response byte is
+    # received (e.g. "connection closed before headers received" — the server
+    # closed an idle keep-alive socket at the instant we wrote the request).
+    # Safe because the server produced no output; the retry re-issues on a fresh
+    # connection. Never fires once headers/chunks have started (would risk
+    # duplicate output). Guards single-stream localhost runs (e.g. llama.cpp with
+    # -np 1) where a stale-socket reset otherwise zeroes an accuracy sample.
+    #
+    # Opt-in: default 0 preserves the prior fail-fast behavior for every use
+    # case. Configs that need it (e.g. the edge-agentic accuracy reference)
+    # enable it explicitly via `client.transport_max_retries`.
+    transport_max_retries: int = Field(
+        0,
+        ge=0,
+        description="Retries on a pre-response connection reset (0=disabled)",
+    )
+
     # Minimum required connections for http-client to initialize.
     # Will log warning if not enough ephemeral ports are available during warmup.
     #
diff --git a/src/inference_endpoint/endpoint_client/worker.py b/src/inference_endpoint/endpoint_client/worker.py
index a87b6889f..5fe96aa83 100644
--- a/src/inference_endpoint/endpoint_client/worker.py
+++ b/src/inference_endpoint/endpoint_client/worker.py
@@ -378,14 +378,47 @@ async def _fire_request(self, req: InFlightRequest) -> bool:
             logger.error(f"Request {req.query_id} failed: {type(e).__name__}: {e}")
             return False
 
+    async def _read_headers_with_retry(self, req: InFlightRequest) -> int | None:
+        """Read response headers, retrying on a pre-response connection reset.
+
+        A connection-level failure at this point means the server closed the
+        socket before sending any response byte (the idle keep-alive race:
+        server drops an idle -np 1 connection just as the client writes on it).
+        No output was produced, so re-issuing the identical request on a fresh
+        connection is safe and idempotent. Returns the HTTP status code, or None
+        if retries are exhausted (an error response has been emitted).
+        """
+        attempt = 0
+        while True:
+            conn = req.connection
+            try:
+                status_code, _ = await conn.protocol.read_headers()
+                return status_code
+            except ConnectionError as e:
+                # Dead connection: discard it (frees the pool slot) and, if
+                # retries remain, re-issue on a fresh connection. Retrying is
+                # only reached here because no headers arrived, so no partial
+                # output was delivered to the accumulator.
+                self._pool.release(conn)
+                if attempt >= self.http_config.transport_max_retries:
+                    await self._handle_error(req.query_id, e)
+                    return None
+                attempt += 1
+                new_conn = await self._pool.acquire()
+                new_conn.protocol.write(req.http_bytes)
+                req.connection = new_conn
+
     @profile
     async def _process_response(self, req: InFlightRequest) -> None:
         """Process response for a fired request."""
-        conn = req.connection
-
         try:
-            # Await headers and handle error status
-            status_code, _ = await conn.protocol.read_headers()
+            # Await headers, retrying on a pre-response connection reset.
+            status_code = await self._read_headers_with_retry(req)
+            if status_code is None:
+                # Retries exhausted; error already emitted.
+                return
+
+            conn = req.connection
             if status_code != 200:
                 error_body = await conn.protocol.read_body()
                 self._pool.release(conn)
@@ -406,8 +439,9 @@ async def _process_response(self, req: InFlightRequest) -> None:
             logger.warning(f"Request {req.query_id} failed: {type(e).__name__}: {e}")
 
         finally:
-            # Release connection back to pool if not already
-            self._pool.release(conn)
+            # Release the current connection (body handlers release early; this
+            # is idempotent and also covers the swapped connection after a retry)
+            self._pool.release(req.connection)
 
             # Clean up task reference
             current_task = asyncio.current_task()
diff --git a/tests/unit/endpoint_client/test_worker_retry.py b/tests/unit/endpoint_client/test_worker_retry.py
new file mode 100644
index 000000000..e3b63d677
--- /dev/null
+++ b/tests/unit/endpoint_client/test_worker_retry.py
@@ -0,0 +1,192 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Unit tests for the worker's pre-response connection-reset retry.
+
+Covers ``Worker._read_headers_with_retry`` — the guard that re-issues a request
+on a fresh connection when the server closes the socket before sending any
+response byte (the idle keep-alive race that otherwise zeroes accuracy samples
+on single-stream localhost servers such as llama.cpp ``-np 1``).
+"""
+
+from types import SimpleNamespace
+
+import pytest
+from inference_endpoint.endpoint_client.config import HTTPClientConfig
+from inference_endpoint.endpoint_client.http import InFlightRequest
+from inference_endpoint.endpoint_client.worker import Worker
+
+
+@pytest.mark.unit
+def test_transport_retries_default_is_opt_in():
+    """Default is 0 (fail-fast) so the retry never changes behavior unless a
+    config opts in explicitly (e.g. the edge-agentic accuracy reference)."""
+    assert HTTPClientConfig().transport_max_retries == 0
+
+
+class _FakeProtocol:
+    """Protocol whose ``read_headers`` replays a scripted result sequence."""
+
+    def __init__(self, results):
+        # results: list where each item is either the sentinel "reset" (raise
+        # ConnectionResetError) or an int status code to return.
+        self._results = list(results)
+        self.written: list[bytes] = []
+
+    async def read_headers(self):
+        outcome = self._results.pop(0)
+        if outcome == "reset":
+            raise ConnectionResetError("Connection closed before headers received")
+        return outcome, {}
+
+    def write(self, data: bytes) -> None:
+        self.written.append(data)
+
+
+class _FakeConn:
+    def __init__(self, protocol: _FakeProtocol):
+        self.protocol = protocol
+
+
+class _FakePool:
+    """Records releases and hands out queued connections on ``acquire``."""
+
+    def __init__(self, acquire_queue: list[_FakeConn]):
+        self._acquire_queue = list(acquire_queue)
+        self.released: list[_FakeConn] = []
+        self.acquired: list[_FakeConn] = []
+
+    def release(self, conn) -> None:
+        self.released.append(conn)
+
+    async def acquire(self) -> _FakeConn:
+        conn = self._acquire_queue.pop(0)
+        self.acquired.append(conn)
+        return conn
+
+
+def _make_worker(pool: _FakePool, max_retries: int):
+    """Build a Worker with only the attributes the retry path touches."""
+    worker = Worker.__new__(Worker)
+    worker._pool = pool  # type: ignore[assignment]
+    worker.http_config = SimpleNamespace(  # type: ignore[assignment]
+        transport_max_retries=max_retries
+    )
+    worker._handle_error_calls = []  # type: ignore[attr-defined]
+
+    async def _record_error(query_id, error):
+        worker._handle_error_calls.append((query_id, error))
+
+    worker._handle_error = _record_error  # type: ignore[method-assign]
+    return worker
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_retry_reissues_on_fresh_connection_after_reset():
+    """A pre-response reset is retried on a fresh connection and succeeds."""
+    dead = _FakeConn(_FakeProtocol(["reset"]))
+    fresh = _FakeConn(_FakeProtocol([200]))
+    pool = _FakePool(acquire_queue=[fresh])
+    worker = _make_worker(pool, max_retries=2)
+
+    req = InFlightRequest(
+        query_id="q1",
+        http_bytes=b"POST /v1/chat/completions HTTP/1.1\r\n\r\n",
+        is_streaming=True,
+        connection=dead,
+    )
+
+    status = await worker._read_headers_with_retry(req)
+
+    assert status == 200
+    assert req.connection is fresh
+    # Dead connection discarded exactly once; request re-written on the fresh one.
+    assert pool.released == [dead]
+    assert fresh.protocol.written == [req.http_bytes]
+    assert worker._handle_error_calls == []
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_retry_exhausted_emits_error_and_returns_none():
+    """When every attempt resets, retries are exhausted and an error is emitted."""
+    dead = _FakeConn(_FakeProtocol(["reset"]))
+    also_dead = _FakeConn(_FakeProtocol(["reset"]))
+    pool = _FakePool(acquire_queue=[also_dead])
+    worker = _make_worker(pool, max_retries=1)
+
+    req = InFlightRequest(
+        query_id="q2",
+        http_bytes=b"POST /x HTTP/1.1\r\n\r\n",
+        is_streaming=False,
+        connection=dead,
+    )
+
+    status = await worker._read_headers_with_retry(req)
+
+    assert status is None
+    # Both connections were tried and discarded.
+    assert pool.released == [dead, also_dead]
+    assert len(worker._handle_error_calls) == 1
+    assert worker._handle_error_calls[0][0] == "q2"
+    assert isinstance(worker._handle_error_calls[0][1], ConnectionResetError)
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_retries_disabled_emits_error_without_reacquiring():
+    """transport_max_retries=0 => no re-issue; the reset surfaces immediately."""
+    dead = _FakeConn(_FakeProtocol(["reset"]))
+    pool = _FakePool(acquire_queue=[])  # acquire must never be called
+    worker = _make_worker(pool, max_retries=0)
+
+    req = InFlightRequest(
+        query_id="q3",
+        http_bytes=b"POST /x HTTP/1.1\r\n\r\n",
+        is_streaming=False,
+        connection=dead,
+    )
+
+    status = await worker._read_headers_with_retry(req)
+
+    assert status is None
+    assert pool.acquired == []
+    assert pool.released == [dead]
+    assert len(worker._handle_error_calls) == 1
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_headers_ok_first_try_never_touches_pool():
+    """No reset => no release/acquire churn, status returned directly."""
+    conn = _FakeConn(_FakeProtocol([200]))
+    pool = _FakePool(acquire_queue=[])
+    worker = _make_worker(pool, max_retries=2)
+
+    req = InFlightRequest(
+        query_id="q4",
+        http_bytes=b"POST /x HTTP/1.1\r\n\r\n",
+        is_streaming=False,
+        connection=conn,
+    )
+
+    status = await worker._read_headers_with_retry(req)
+
+    assert status == 200
+    assert req.connection is conn
+    assert pool.acquired == []
+    assert pool.released == []
+    assert worker._handle_error_calls == []

From b9ca904eb9df5019c6c9934827c97b26a7ecacd7 Mon Sep 17 00:00:00 2001
From: palanivelg <palanivelg@ieee.org>
Date: Wed, 8 Jul 2026 23:49:28 +0000
Subject: [PATCH 2/3] fix(review): attach fresh connection before re-issue
 write

Assign req.connection to the freshly acquired connection before writing
the retried request, so a write failure leaves the request owning a
releasable connection (the caller's finally-block releases it) instead of
leaking it. Adds a regression test for the failing-write path.
---
 .../endpoint_client/worker.py                 |  5 ++-
 .../unit/endpoint_client/test_worker_retry.py | 31 +++++++++++++++++++
 2 files changed, 35 insertions(+), 1 deletion(-)

diff --git a/src/inference_endpoint/endpoint_client/worker.py b/src/inference_endpoint/endpoint_client/worker.py
index 5fe96aa83..cda0e15f5 100644
--- a/src/inference_endpoint/endpoint_client/worker.py
+++ b/src/inference_endpoint/endpoint_client/worker.py
@@ -404,9 +404,12 @@ async def _read_headers_with_retry(self, req: InFlightRequest) -> int | None:
                     await self._handle_error(req.query_id, e)
                     return None
                 attempt += 1
+                # Attach the fresh connection to the request BEFORE writing, so
+                # a write failure still leaves req.connection pointing at it for
+                # the caller's finally-block release (no leaked connection).
                 new_conn = await self._pool.acquire()
-                new_conn.protocol.write(req.http_bytes)
                 req.connection = new_conn
+                new_conn.protocol.write(req.http_bytes)
 
     @profile
     async def _process_response(self, req: InFlightRequest) -> None:
diff --git a/tests/unit/endpoint_client/test_worker_retry.py b/tests/unit/endpoint_client/test_worker_retry.py
index e3b63d677..48e701526 100644
--- a/tests/unit/endpoint_client/test_worker_retry.py
+++ b/tests/unit/endpoint_client/test_worker_retry.py
@@ -190,3 +190,34 @@ async def test_headers_ok_first_try_never_touches_pool():
     assert pool.acquired == []
     assert pool.released == []
     assert worker._handle_error_calls == []
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_fresh_conn_attached_before_write_so_failing_write_is_not_leaked():
+    """If the re-issue write fails, req.connection already points to the fresh
+    connection so the caller's finally-block can release it (no leak)."""
+
+    class _RaisingWriteProtocol(_FakeProtocol):
+        def write(self, data: bytes) -> None:
+            raise ConnectionResetError("write on closed transport")
+
+    dead = _FakeConn(_FakeProtocol(["reset"]))
+    fresh = _FakeConn(_RaisingWriteProtocol([200]))
+    pool = _FakePool(acquire_queue=[fresh])
+    worker = _make_worker(pool, max_retries=2)
+
+    req = InFlightRequest(
+        query_id="q5",
+        http_bytes=b"POST /x HTTP/1.1\r\n\r\n",
+        is_streaming=False,
+        connection=dead,
+    )
+
+    with pytest.raises(ConnectionError):
+        await worker._read_headers_with_retry(req)
+
+    # Fresh connection is attached before the failing write, so the request
+    # still owns a releasable connection (the old one was already released).
+    assert req.connection is fresh
+    assert pool.released == [dead]

From 52e8064dc3f6e7730a4964e68d47b7d66e13ace9 Mon Sep 17 00:00:00 2001
From: palanivelg <palanivelg@ieee.org>
Date: Sun, 12 Jul 2026 20:29:49 +0000
Subject: [PATCH 3/3] fix(review): observe, bound, and disclose transport
 retries; guard partial closes

Address PR #404 review feedback on the opt-in transport_max_retries feature:

- Observability: log a warning on each pre-response reset recovery, count
  re-issues per sample on QueryResult.transport_retries, and aggregate them
  into a transport_retries_total counter (new SampleEventType.TRANSPORT_RETRY
  -> metrics registry -> snapshot -> Report) so a retry-recovered reset no
  longer returns with zero trace and the report can segregate retried samples.
- Partial-close correctness: only retry a TRUE zero-byte close. Track whether
  any response byte arrived (HttpResponseProtocol.bytes_received); a partial
  status-line/header close now surfaces as an error instead of re-issuing a
  request the server may already have processed. Docstring scoped to
  client-output idempotency.
- Release idempotency: ConnectionPool.release now clears in_use in the dead
  branch so a double-release short-circuits instead of firing a spurious
  waiter wake.
- Compliance/reproducibility: bound transport_max_retries with le=5, lock it in
  check_config_lock, and surface a manual attestation in check_submission so a
  recovered reset can't silently satisfy no_dropped_turns.
- Tests: multi-retry-then-success, partial-close no-retry, in_use-tracking pool
  driving _process_response end-to-end (double-release path), dead-connection
  release idempotency, and observability assertions.
---
 .../services/metrics_aggregator/aggregator.py |  12 ++
 src/inference_endpoint/compliance/checker.py  |  31 ++++
 .../templates/concurrency_template_full.yaml  |   2 +-
 .../templates/offline_template_full.yaml      |   2 +-
 .../templates/online_template_full.yaml       |   2 +-
 src/inference_endpoint/core/record.py         |   3 +
 src/inference_endpoint/core/types.py          |   7 +
 .../endpoint_client/config.py                 |   5 +-
 .../endpoint_client/http.py                   |  26 ++-
 .../endpoint_client/worker.py                 |  63 +++++--
 .../load_generator/session.py                 |  15 ++
 src/inference_endpoint/metrics/report.py      |  11 ++
 tests/unit/compliance/test_checker.py         |  44 ++++-
 tests/unit/endpoint_client/test_http.py       |  27 +++
 .../unit/endpoint_client/test_worker_retry.py | 157 +++++++++++++++++-
 15 files changed, 382 insertions(+), 25 deletions(-)

diff --git a/src/inference_endpoint/async_utils/services/metrics_aggregator/aggregator.py b/src/inference_endpoint/async_utils/services/metrics_aggregator/aggregator.py
index 506ca3495..4a3541585 100644
--- a/src/inference_endpoint/async_utils/services/metrics_aggregator/aggregator.py
+++ b/src/inference_endpoint/async_utils/services/metrics_aggregator/aggregator.py
@@ -70,6 +70,11 @@ class MetricCounterKey(str, Enum):
     # tracked row still exists when the aggregator sees the ERROR.
     TRACKED_SAMPLES_FAILED = "tracked_samples_failed"
     TRACKED_DURATION_NS = "tracked_duration_ns"
+    # Total pre-response connection-reset re-issues across the run (opt-in
+    # transport_max_retries). Non-zero means the SUT reset idle connections
+    # that the client transparently retried — a "can't keep up" signal that
+    # would otherwise vanish once the retry converts the reset into a success.
+    TRANSPORT_RETRIES_TOTAL = "transport_retries_total"
     # Legacy MLPerf LoadGen Server "completed" window (final_query_all_samples_done_time).
     LEGACY_LOADGEN_WINDOW_DURATION_NS = "legacy_loadgen_window_duration_ns"
     # Total wall-clock duration since session start. Updated on every event as
@@ -368,6 +373,13 @@ async def process(self, records: list[EventRecord]) -> None:
                 logger.debug("Error event: %s", record)
                 continue
 
+            # --- Transport retry events ---
+            # A SampleEventType, but not a tracked lifecycle event: it only
+            # bumps the diagnostic recovery counter and never touches the table.
+            if ev == SampleEventType.TRANSPORT_RETRY:
+                registry.increment(MetricCounterKey.TRANSPORT_RETRIES_TOTAL.value)
+                continue
+
             # --- Sample events ---
             if (
                 not isinstance(ev, SampleEventType)
diff --git a/src/inference_endpoint/compliance/checker.py b/src/inference_endpoint/compliance/checker.py
index 83a4c6c07..fbb7c3321 100644
--- a/src/inference_endpoint/compliance/checker.py
+++ b/src/inference_endpoint/compliance/checker.py
@@ -157,6 +157,21 @@ def check_config_lock(config: dict[str, Any]) -> list[Check]:
         detail += f" target_concurrency={target_concurrency}"
     checks.append(Check("single_stream", single_stream, detail))
 
+    # Lock the pre-response reset retry: it is a reproducibility knob that lets
+    # the client silently re-issue after a SUT-side reset, so its value must be
+    # recorded (and within the schema bound) rather than passing for any value.
+    # Disclosure that a recovered reset didn't mask a dropped turn is a manual
+    # attestation added in check_submission (transport_retries_total lives in the
+    # run's result summary, not in this config).
+    retries = _get(config, "settings", "client", "transport_max_retries", default=0)
+    checks.append(
+        Check(
+            "transport_max_retries_locked",
+            isinstance(retries, int) and 0 <= retries <= 5,
+            f"transport_max_retries={retries}",
+        )
+    )
+
     return checks
 
 
@@ -309,6 +324,22 @@ def check_submission(
             report.add("config_valid", False, f"{config_path} failed to load: {e}")
         else:
             report.checks.extend(check_config_lock(config))
+            retries = _get(
+                config, "settings", "client", "transport_max_retries", default=0
+            )
+            if retries:
+                # A recovered pre-response reset produces a clean success, so
+                # the no_dropped_turns gate can't see it from scores.json alone.
+                # Require the submitter to confirm the disclosed retry value and
+                # that transport_retries_total (in the run's result summary) did
+                # not mask a dropped-and-re-issued turn.
+                report.notes.append(
+                    f"transport_max_retries={retries}: pre-response connection "
+                    "resets are retried. Confirm this matches the disclosed "
+                    "ruleset value and that no retry-recovered reset (see "
+                    "transport_retries_total in the run's result summary) masked "
+                    "a dropped turn."
+                )
 
     results_path = report_dir / "results.json"
     scores_path = report_dir / "scores.json"
diff --git a/src/inference_endpoint/config/templates/concurrency_template_full.yaml b/src/inference_endpoint/config/templates/concurrency_template_full.yaml
index ce3f172fd..acf45d918 100644
--- a/src/inference_endpoint/config/templates/concurrency_template_full.yaml
+++ b/src/inference_endpoint/config/templates/concurrency_template_full.yaml
@@ -78,7 +78,7 @@ settings:
     worker_force_kill_timeout: 0.5  # Force kill timeout after graceful wait (seconds)
     insecure: false  # Skip TLS certificate verification
     max_idle_time: 4.0  # Discard connections idle longer than this (seconds)
-    transport_max_retries: 0  # Retries on a pre-response connection reset (0=disabled)
+    transport_max_retries: 0  # Retries on a pre-response connection reset (0=disabled, max 5)
     min_required_connections: -1  # Min connections to initialize (-1=auto, 0=disabled)
     worker_gc_mode: relaxed  # Worker GC strategy | options: disabled, relaxed, system
   drain:
diff --git a/src/inference_endpoint/config/templates/offline_template_full.yaml b/src/inference_endpoint/config/templates/offline_template_full.yaml
index 681170fc7..28dbac973 100644
--- a/src/inference_endpoint/config/templates/offline_template_full.yaml
+++ b/src/inference_endpoint/config/templates/offline_template_full.yaml
@@ -78,7 +78,7 @@ settings:
     worker_force_kill_timeout: 0.5  # Force kill timeout after graceful wait (seconds)
     insecure: false  # Skip TLS certificate verification
     max_idle_time: 4.0  # Discard connections idle longer than this (seconds)
-    transport_max_retries: 0  # Retries on a pre-response connection reset (0=disabled)
+    transport_max_retries: 0  # Retries on a pre-response connection reset (0=disabled, max 5)
     min_required_connections: -1  # Min connections to initialize (-1=auto, 0=disabled)
     worker_gc_mode: relaxed  # Worker GC strategy | options: disabled, relaxed, system
   drain:
diff --git a/src/inference_endpoint/config/templates/online_template_full.yaml b/src/inference_endpoint/config/templates/online_template_full.yaml
index 604fef9f9..36a5f5bc4 100644
--- a/src/inference_endpoint/config/templates/online_template_full.yaml
+++ b/src/inference_endpoint/config/templates/online_template_full.yaml
@@ -79,7 +79,7 @@ settings:
     worker_force_kill_timeout: 0.5  # Force kill timeout after graceful wait (seconds)
     insecure: false  # Skip TLS certificate verification
     max_idle_time: 4.0  # Discard connections idle longer than this (seconds)
-    transport_max_retries: 0  # Retries on a pre-response connection reset (0=disabled)
+    transport_max_retries: 0  # Retries on a pre-response connection reset (0=disabled, max 5)
     min_required_connections: -1  # Min connections to initialize (-1=auto, 0=disabled)
     worker_gc_mode: relaxed  # Worker GC strategy | options: disabled, relaxed, system
   drain:
diff --git a/src/inference_endpoint/core/record.py b/src/inference_endpoint/core/record.py
index 9ac60e1b6..cf25ce94c 100644
--- a/src/inference_endpoint/core/record.py
+++ b/src/inference_endpoint/core/record.py
@@ -145,6 +145,9 @@ class SampleEventType(EventType):
     COMPLETE = "complete"
     RECV_FIRST = "recv_first"
     RECV_NON_FIRST = "recv_non_first"
+    # Emitted once per pre-response connection-reset re-issue (opt-in
+    # transport_max_retries). Diagnostic only — not a tracked sample event.
+    TRANSPORT_RETRY = "transport_retry"
 
 
 class EventRecord(msgspec.Struct, kw_only=True, frozen=True, gc=False):  # type: ignore[call-arg]
diff --git a/src/inference_endpoint/core/types.py b/src/inference_endpoint/core/types.py
index 6e1cbab46..5273f252b 100644
--- a/src/inference_endpoint/core/types.py
+++ b/src/inference_endpoint/core/types.py
@@ -402,6 +402,12 @@ class QueryResult(
         error: Structured error if query failed (None if successful).
         completed_at: High-resolution timestamp (nanoseconds, monotonic clock).
                       Auto-set in __post_init__ to prevent tampering.
+        transport_retries: Number of pre-response connection-reset re-issues the
+                      worker performed for this sample (0 = none). Non-zero marks
+                      a sample whose timed window folds in dead-socket wait +
+                      reconnect + a second server pass, so the report can
+                      segregate or exclude it from the server latency
+                      distribution. Appended last for array_like backward compat.
 
     Note:
         The completed_at field is intentionally set internally to prevent
@@ -422,6 +428,7 @@ class QueryResult(
     metadata: dict[str, Any] = msgspec.field(default_factory=dict)
     error: ErrorData | None = None
     completed_at: int | msgspec.UnsetType = msgspec.UNSET
+    transport_retries: int = 0
 
     def __post_init__(self):
         """Set completion timestamp automatically.
diff --git a/src/inference_endpoint/endpoint_client/config.py b/src/inference_endpoint/endpoint_client/config.py
index ce170e4b4..40285b7f4 100644
--- a/src/inference_endpoint/endpoint_client/config.py
+++ b/src/inference_endpoint/endpoint_client/config.py
@@ -181,10 +181,13 @@ class HTTPClientConfig(WithUpdatesMixin, BaseModel):
     # Opt-in: default 0 preserves the prior fail-fast behavior for every use
     # case. Configs that need it (e.g. the edge-agentic accuracy reference)
     # enable it explicitly via `client.transport_max_retries`.
+    # Bounded above so a genuinely-down server can't turn fail-fast into a long
+    # reconnect/re-write spin that masks a hard SUT failure and multiplies load.
     transport_max_retries: int = Field(
         0,
         ge=0,
-        description="Retries on a pre-response connection reset (0=disabled)",
+        le=5,
+        description="Retries on a pre-response connection reset (0=disabled, max 5)",
     )
 
     # Minimum required connections for http-client to initialize.
diff --git a/src/inference_endpoint/endpoint_client/http.py b/src/inference_endpoint/endpoint_client/http.py
index 210d3e192..618a805de 100644
--- a/src/inference_endpoint/endpoint_client/http.py
+++ b/src/inference_endpoint/endpoint_client/http.py
@@ -131,6 +131,7 @@ class HttpResponseProtocol(asyncio.Protocol):
         "_message_complete",
         "_connection_lost",
         "_exc",
+        "_bytes_received",
     )
 
     def __init__(self, loop: asyncio.AbstractEventLoop):
@@ -158,6 +159,10 @@ def __init__(self, loop: asyncio.AbstractEventLoop):
         self._message_complete: bool = False
         self._connection_lost: bool = False
         self._exc: Exception | None = None
+        # True once any response byte has been received for the current
+        # request. Distinguishes a true zero-byte close (safe to retry) from a
+        # partial-response close (server already processed the request).
+        self._bytes_received: bool = False
 
     def reset(self) -> None:
         """Reset protocol state for connection reuse."""
@@ -175,6 +180,7 @@ def reset(self) -> None:
         self._headers_complete = False
         self._message_complete = False
         self._exc = None
+        self._bytes_received = False
         # NOTE: Don't reset _connection_lost - that's transport state
 
     def _signal_stream_end(self) -> None:
@@ -196,6 +202,8 @@ def connection_made(self, transport: asyncio.Transport) -> None:  # type: ignore
         self._parser = httptools.HttpResponseParser(self)
 
     def data_received(self, data: bytes) -> None:
+        if data:
+            self._bytes_received = True
         # Lazy parser creation for better reset() performance
         if self._parser is None:
             self._parser = httptools.HttpResponseParser(self)
@@ -316,6 +324,17 @@ def should_close(self) -> bool:
         """Whether connection should be closed after this response."""
         return self._should_close or self._connection_lost or self._exc is not None
 
+    @property
+    def bytes_received(self) -> bool:
+        """Whether any response byte arrived for the current request.
+
+        False means a true zero-byte close (the server never wrote a byte);
+        True means partial or complete response bytes were parsed. Used to gate
+        the pre-response reset retry so a request the server may already have
+        processed is not re-issued.
+        """
+        return self._bytes_received
+
     def write(self, data: bytes) -> None:
         """Write data to transport."""
         if self._transport:
@@ -588,8 +607,11 @@ def release(self, conn: PooledConnection) -> None:
         if not conn.in_use:
             return
 
-        # Must close if: dead, server requested close, or error occurred
+        # Must close if: dead, server requested close, or error occurred.
+        # Clear in_use first so a second release() short-circuits above
+        # instead of re-entering this branch and firing a spurious waiter wake.
         if not conn.is_alive() or conn.protocol.should_close:
+            conn.in_use = False
             self._close_connection(conn)
             self._notify_waiter()
             return
@@ -798,9 +820,11 @@ class InFlightRequest:
         http_bytes: Serialized HTTP request for socket.write().
         is_streaming: Whether this is a streaming (SSE) request or not.
         connection: PooledConnection assigned to this request (set once request is fired).
+        transport_retries: Count of pre-response reset re-issues for this request.
     """
 
     query_id: str
     http_bytes: bytes
     is_streaming: bool
     connection: PooledConnection = field(default=None, repr=False)  # type: ignore[assignment]
+    transport_retries: int = 0
diff --git a/src/inference_endpoint/endpoint_client/worker.py b/src/inference_endpoint/endpoint_client/worker.py
index cda0e15f5..63a2dee32 100644
--- a/src/inference_endpoint/endpoint_client/worker.py
+++ b/src/inference_endpoint/endpoint_client/worker.py
@@ -28,6 +28,8 @@
 from typing import Any
 from urllib.parse import urlparse
 
+import msgspec
+
 from inference_endpoint.async_utils.transport import (
     ReceiverTransport,
     SenderTransport,
@@ -379,31 +381,45 @@ async def _fire_request(self, req: InFlightRequest) -> bool:
             return False
 
     async def _read_headers_with_retry(self, req: InFlightRequest) -> int | None:
-        """Read response headers, retrying on a pre-response connection reset.
-
-        A connection-level failure at this point means the server closed the
-        socket before sending any response byte (the idle keep-alive race:
-        server drops an idle -np 1 connection just as the client writes on it).
-        No output was produced, so re-issuing the identical request on a fresh
-        connection is safe and idempotent. Returns the HTTP status code, or None
-        if retries are exhausted (an error response has been emitted).
+        """Read response headers, retrying only on a true zero-byte close.
+
+        A connection-level failure here can mean the server closed the socket
+        before writing any response byte (the idle keep-alive race: an idle
+        ``-np 1`` connection dropped just as the client writes on it). Only a
+        true zero-byte close is retried — re-issuing on a fresh connection is
+        safe for *client* output because the accumulator has produced nothing.
+
+        If any response byte already arrived (partial status-line/header bytes
+        before the close), the server may have received and processed the
+        request, so a re-issue would duplicate server-side generation work; that
+        case surfaces the error instead of retrying. Each recovery is logged and
+        counted on ``req.transport_retries`` so the successful sample carries a
+        marker downstream. Returns the HTTP status code, or None if the reset is
+        not retryable or retries are exhausted (an error response was emitted).
         """
-        attempt = 0
         while True:
             conn = req.connection
             try:
                 status_code, _ = await conn.protocol.read_headers()
                 return status_code
             except ConnectionError as e:
-                # Dead connection: discard it (frees the pool slot) and, if
-                # retries remain, re-issue on a fresh connection. Retrying is
-                # only reached here because no headers arrived, so no partial
-                # output was delivered to the accumulator.
+                partial = conn.protocol.bytes_received
                 self._pool.release(conn)
-                if attempt >= self.http_config.transport_max_retries:
+                if (
+                    partial
+                    or req.transport_retries >= self.http_config.transport_max_retries
+                ):
                     await self._handle_error(req.query_id, e)
                     return None
-                attempt += 1
+                req.transport_retries += 1
+                logger.warning(
+                    "Request %s: retrying after pre-response connection reset "
+                    "(attempt %d/%d): %s",
+                    req.query_id,
+                    req.transport_retries,
+                    self.http_config.transport_max_retries,
+                    e,
+                )
                 # Attach the fresh connection to the request BEFORE writing, so
                 # a write failure still leaves req.connection pointing at it for
                 # the caller's finally-block release (no leaked connection).
@@ -469,7 +485,7 @@ async def _handle_streaming_body(self, req: InFlightRequest) -> None:
         self._pool.release(conn)
 
         # Send final complete back to main rank
-        self._responses.send(accumulator.get_final_output())
+        self._responses.send(self._tag_retries(accumulator.get_final_output(), req))
 
     @profile
     async def _handle_non_streaming_body(self, req: InFlightRequest) -> None:
@@ -487,7 +503,20 @@ async def _handle_non_streaming_body(self, req: InFlightRequest) -> None:
         result = self._adapter.decode_response(response_bytes, query_id)
 
         # Send result back to main rank
-        self._responses.send(result)
+        self._responses.send(self._tag_retries(result, req))
+
+    @staticmethod
+    def _tag_retries(result: QueryResult, req: InFlightRequest) -> QueryResult:
+        """Stamp a successful result with this request's retry count.
+
+        No-op when the request was not retried (the common path), so the hot
+        path pays nothing unless a pre-response reset was actually recovered.
+        """
+        if req.transport_retries:
+            msgspec.structs.force_setattr(
+                result, "transport_retries", req.transport_retries
+            )
+        return result
 
     async def _handle_error(self, query_id: str, error: Exception | str) -> None:
         """Send error response for a query."""
diff --git a/src/inference_endpoint/load_generator/session.py b/src/inference_endpoint/load_generator/session.py
index 0d991ab1f..9640140e4 100644
--- a/src/inference_endpoint/load_generator/session.py
+++ b/src/inference_endpoint/load_generator/session.py
@@ -600,6 +600,21 @@ def _handle_response(self, resp: QueryResult | StreamChunk) -> None:
                         data=resp.error,
                     )
                 )
+            # Surface pre-response connection-reset recoveries as a diagnostic
+            # counter (one event per re-issue). Emitted for every completed
+            # sample, warmup included, since a reset is a SUT-health signal
+            # regardless of the tracking window.
+            if resp.transport_retries:
+                retry_record = EventRecord(
+                    event_type=SampleEventType.TRANSPORT_RETRY,
+                    timestamp_ns=time.monotonic_ns(),
+                    sample_uuid=query_id,
+                    conversation_id=conv_id_str,
+                    turn=turn_num,
+                )
+                for _ in range(resp.transport_retries):
+                    self._publisher.publish(retry_record)
+
             if self._current_phase_type != PhaseType.WARMUP:
                 self._publisher.publish(
                     EventRecord(
diff --git a/src/inference_endpoint/metrics/report.py b/src/inference_endpoint/metrics/report.py
index f9472aa58..ffeeb41f6 100644
--- a/src/inference_endpoint/metrics/report.py
+++ b/src/inference_endpoint/metrics/report.py
@@ -152,6 +152,11 @@ class Report(msgspec.Struct, frozen=True):  # type: ignore[call-arg]
     # caller supplies them rather than reading them from the metrics snapshot.
     seeds: dict[str, int] | None = None
 
+    # Total pre-response connection-reset re-issues (opt-in transport_max_retries).
+    # Non-zero means the SUT reset idle connections that the client transparently
+    # retried; surfaced so a "clean" run still discloses the recovered resets.
+    transport_retries_total: int = 0
+
     @classmethod
     def from_snapshot(
         cls,
@@ -275,6 +280,7 @@ def _series_dict(key: str) -> dict[str, Any]:
             qps=qps,
             tps=tps,
             seeds=seeds,
+            transport_retries_total=_counter("transport_retries_total"),
         )
 
     def to_json(self, save_to: os.PathLike | None = None) -> bytes:
@@ -313,6 +319,11 @@ def display(
         fn(f"Total samples issued: {self.n_samples_issued}{newline}")
         fn(f"Total samples completed: {self.n_samples_completed}{newline}")
         fn(f"Total samples failed: {self.n_samples_failed}{newline}")
+        if self.transport_retries_total:
+            fn(
+                "Transport retries (pre-response resets recovered): "
+                f"{self.transport_retries_total}{newline}"
+            )
         if self.duration_ns is not None:
             fn(f"Duration: {self.duration_ns / 1e9:.2f} seconds{newline}")
         else:
diff --git a/tests/unit/compliance/test_checker.py b/tests/unit/compliance/test_checker.py
index e40655819..4a9ee8f27 100644
--- a/tests/unit/compliance/test_checker.py
+++ b/tests/unit/compliance/test_checker.py
@@ -90,10 +90,36 @@ def _accuracy_results(
 @pytest.mark.unit
 def test_config_lock_passes_on_compliant_config():
     checks = check_config_lock(_passing_config())
-    assert {c.name for c in checks} == {"temperature==0", "seed==42", "single_stream"}
+    assert {c.name for c in checks} == {
+        "temperature==0",
+        "seed==42",
+        "single_stream",
+        "transport_max_retries_locked",
+    }
     assert all(c.passed for c in checks)
 
 
+@pytest.mark.unit
+def test_config_lock_records_transport_max_retries():
+    config = _passing_config()
+    config["settings"]["client"]["transport_max_retries"] = 2
+    check = next(
+        c for c in check_config_lock(config) if c.name == "transport_max_retries_locked"
+    )
+    assert check.passed
+    assert "transport_max_retries=2" in check.detail
+
+
+@pytest.mark.unit
+def test_config_lock_fails_on_out_of_bound_transport_max_retries():
+    config = _passing_config()
+    config["settings"]["client"]["transport_max_retries"] = 9
+    check = next(
+        c for c in check_config_lock(config) if c.name == "transport_max_retries_locked"
+    )
+    assert not check.passed
+
+
 @pytest.mark.unit
 @pytest.mark.parametrize(
     "mutate",
@@ -274,6 +300,22 @@ def test_check_submission_uses_ruleset_thresholds(tmp_path):
     assert not report.passed
 
 
+@pytest.mark.unit
+def test_check_submission_surfaces_transport_retries_attestation(tmp_path):
+    # A nonzero transport_max_retries must be disclosed as a manual attestation
+    # so a retry-recovered reset can't silently satisfy no_dropped_turns.
+    config_with_retries = _VALID_CONFIG_YAML.replace(
+        "    max_connections: 1", "    max_connections: 1\n    transport_max_retries: 2"
+    )
+    (tmp_path / "config.yaml").write_text(config_with_retries)
+    (tmp_path / "results.json").write_text(
+        json.dumps(_accuracy_results(86.23, 87.96, 995))
+    )
+    report = check_submission(tmp_path)
+    assert report.passed
+    assert any("transport_max_retries=2" in n for n in report.notes)
+
+
 @pytest.mark.unit
 def test_check_submission_missing_artifacts(tmp_path):
     report = check_submission(tmp_path)
diff --git a/tests/unit/endpoint_client/test_http.py b/tests/unit/endpoint_client/test_http.py
index d9be6ddab..b14e8fc77 100644
--- a/tests/unit/endpoint_client/test_http.py
+++ b/tests/unit/endpoint_client/test_http.py
@@ -579,6 +579,33 @@ async def test_release_idempotent(self, pool):
         assert pool.idle_count == 1
         assert pool.total_count == 1
 
+    @pytest.mark.asyncio
+    async def test_release_dead_connection_idempotent(self, pool):
+        """Double-releasing a dead connection clears ``in_use`` and does not
+        fire a second (spurious) waiter wake — the retry-exhausted / disabled
+        paths release the dead connection once in the worker and again in the
+        ``finally`` block."""
+        conn = await pool.acquire()
+        conn.transport.close()  # dead: is_alive() is now False
+
+        notify_calls = 0
+        original_notify = pool._notify_waiter
+
+        def counting_notify() -> None:
+            nonlocal notify_calls
+            notify_calls += 1
+            original_notify()
+
+        pool._notify_waiter = counting_notify  # type: ignore[method-assign]
+
+        pool.release(conn)
+        assert conn.in_use is False
+        assert pool.total_count == 0
+        # Second release must short-circuit on `not in_use` instead of
+        # re-entering the dead branch and waking a waiter again.
+        pool.release(conn)
+        assert notify_calls == 1
+
     @pytest.mark.asyncio
     async def test_unlimited_pool(self, echo_server):
         """Pool with max_connections=None allows unlimited connections."""
diff --git a/tests/unit/endpoint_client/test_worker_retry.py b/tests/unit/endpoint_client/test_worker_retry.py
index 48e701526..964d0bcb8 100644
--- a/tests/unit/endpoint_client/test_worker_retry.py
+++ b/tests/unit/endpoint_client/test_worker_retry.py
@@ -24,6 +24,7 @@
 from types import SimpleNamespace
 
 import pytest
+from inference_endpoint.core.types import QueryResult
 from inference_endpoint.endpoint_client.config import HTTPClientConfig
 from inference_endpoint.endpoint_client.http import InFlightRequest
 from inference_endpoint.endpoint_client.worker import Worker
@@ -39,11 +40,15 @@ def test_transport_retries_default_is_opt_in():
 class _FakeProtocol:
     """Protocol whose ``read_headers`` replays a scripted result sequence."""
 
-    def __init__(self, results):
+    def __init__(self, results, *, bytes_received: bool = False, body: bytes = b"{}"):
         # results: list where each item is either the sentinel "reset" (raise
         # ConnectionResetError) or an int status code to return.
         self._results = list(results)
         self.written: list[bytes] = []
+        # bytes_received distinguishes a true zero-byte close (retryable) from a
+        # partial-response close (server may already have processed the request).
+        self.bytes_received = bytes_received
+        self._body = body
 
     async def read_headers(self):
         outcome = self._results.pop(0)
@@ -51,6 +56,9 @@ async def read_headers(self):
             raise ConnectionResetError("Connection closed before headers received")
         return outcome, {}
 
+    async def read_body(self) -> bytes:
+        return self._body
+
     def write(self, data: bytes) -> None:
         self.written.append(data)
 
@@ -58,6 +66,7 @@ def write(self, data: bytes) -> None:
 class _FakeConn:
     def __init__(self, protocol: _FakeProtocol):
         self.protocol = protocol
+        self.in_use = True
 
 
 class _FakePool:
@@ -77,13 +86,63 @@ async def acquire(self) -> _FakeConn:
         return conn
 
 
-def _make_worker(pool: _FakePool, max_retries: int):
+class _InUsePool:
+    """Fake pool that models ``in_use`` so a double-release is observably a
+    no-op the second time (mirrors ``ConnectionPool.release`` idempotency).
+
+    ``effective_releases`` counts only releases that actually transitioned a
+    connection out of use; ``release_calls`` counts every call. A correct
+    ``finally``-block double-release therefore shows two calls but one effect.
+    """
+
+    def __init__(self, acquire_queue: list[_FakeConn]):
+        self._acquire_queue = list(acquire_queue)
+        self.release_calls: list[_FakeConn] = []
+        self.effective_releases: list[_FakeConn] = []
+        self.acquired: list[_FakeConn] = []
+
+    def release(self, conn) -> None:
+        self.release_calls.append(conn)
+        if not conn.in_use:
+            return
+        conn.in_use = False
+        self.effective_releases.append(conn)
+
+    async def acquire(self) -> _FakeConn:
+        conn = self._acquire_queue.pop(0)
+        conn.in_use = True
+        self.acquired.append(conn)
+        return conn
+
+
+class _FakeResponses:
+    """Captures results the worker sends back to the main rank."""
+
+    def __init__(self) -> None:
+        self.sent: list = []
+
+    def send(self, item) -> None:
+        self.sent.append(item)
+
+
+class _FakeAdapter:
+    """Minimal adapter: decodes a non-streaming body into a QueryResult."""
+
+    @staticmethod
+    def decode_response(body: bytes, query_id: str):
+        return QueryResult(id=query_id)
+
+
+def _make_worker(pool, max_retries: int, *, responses=None):
     """Build a Worker with only the attributes the retry path touches."""
     worker = Worker.__new__(Worker)
     worker._pool = pool  # type: ignore[assignment]
     worker.http_config = SimpleNamespace(  # type: ignore[assignment]
         transport_max_retries=max_retries
     )
+    worker._active_tasks = set()  # type: ignore[attr-defined]
+    worker._responses = responses  # type: ignore[assignment]
+    worker._adapter = _FakeAdapter  # type: ignore[assignment]
     worker._handle_error_calls = []  # type: ignore[attr-defined]
 
     async def _record_error(query_id, error):
@@ -117,6 +176,8 @@ async def test_retry_reissues_on_fresh_connection_after_reset():
     assert pool.released == [dead]
     assert fresh.protocol.written == [req.http_bytes]
     assert worker._handle_error_calls == []
+    # The recovery is counted on the request so the success carries a marker.
+    assert req.transport_retries == 1
 
 
 @pytest.mark.unit
@@ -221,3 +282,95 @@ def write(self, data: bytes) -> None:
     # still owns a releasable connection (the old one was already released).
     assert req.connection is fresh
     assert pool.released == [dead]
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_multiple_retries_then_success():
+    """Two consecutive resets are recovered when max_retries allows it."""
+    dead1 = _FakeConn(_FakeProtocol(["reset"]))
+    dead2 = _FakeConn(_FakeProtocol(["reset"]))
+    fresh = _FakeConn(_FakeProtocol([200]))
+    pool = _FakePool(acquire_queue=[dead2, fresh])
+    worker = _make_worker(pool, max_retries=2)
+
+    req = InFlightRequest(
+        query_id="q6",
+        http_bytes=b"POST /x HTTP/1.1\r\n\r\n",
+        is_streaming=False,
+        connection=dead1,
+    )
+
+    status = await worker._read_headers_with_retry(req)
+
+    assert status == 200
+    assert req.connection is fresh
+    assert req.transport_retries == 2
+    # Both dead connections discarded in order; each re-issue re-wrote the body.
+    assert pool.released == [dead1, dead2]
+    assert pool.acquired == [dead2, fresh]
+    assert dead2.protocol.written == [req.http_bytes]
+    assert fresh.protocol.written == [req.http_bytes]
+    assert worker._handle_error_calls == []
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_partial_response_close_is_not_retried():
+    """A close AFTER some response bytes arrived is not a zero-byte close: the
+    server may already have processed the request, so it surfaces as an error
+    rather than a (server-side non-idempotent) re-issue."""
+    partial = _FakeConn(_FakeProtocol(["reset"], bytes_received=True))
+    pool = _FakePool(acquire_queue=[])  # acquire must never be called
+    worker = _make_worker(pool, max_retries=2)
+
+    req = InFlightRequest(
+        query_id="q7",
+        http_bytes=b"POST /x HTTP/1.1\r\n\r\n",
+        is_streaming=False,
+        connection=partial,
+    )
+
+    status = await worker._read_headers_with_retry(req)
+
+    assert status is None
+    assert pool.acquired == []
+    assert pool.released == [partial]
+    assert req.transport_retries == 0
+    assert len(worker._handle_error_calls) == 1
+    assert isinstance(worker._handle_error_calls[0][1], ConnectionResetError)
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_process_response_double_release_is_idempotent():
+    """Driving ``_process_response`` end-to-end after a recovered reset: the
+    non-streaming body handler releases the fresh connection, then the
+    ``finally`` block releases it again. With an ``in_use``-tracking pool the
+    second release is a no-op, and the sent result carries the retry count."""
+    dead = _FakeConn(_FakeProtocol(["reset"]))
+    fresh = _FakeConn(_FakeProtocol([200]))
+    pool = _InUsePool(acquire_queue=[fresh])
+    responses = _FakeResponses()
+    worker = _make_worker(pool, max_retries=2, responses=responses)
+
+    req = InFlightRequest(
+        query_id="q8",
+        http_bytes=b"POST /x HTTP/1.1\r\n\r\n",
+        is_streaming=False,
+        connection=dead,
+    )
+
+    await worker._process_response(req)
+
+    # fresh is released twice (body handler + finally) but only takes effect once.
+    assert pool.release_calls.count(fresh) == 2
+    assert pool.effective_releases.count(fresh) == 1
+    assert fresh.in_use is False
+    # The result reached the main rank tagged with the recovered retry.
+    assert len(responses.sent) == 1
+    result = responses.sent[0]
+    assert isinstance(result, QueryResult)
+    assert result.id == "q8"
+    assert result.transport_retries == 1
+    assert worker._handle_error_calls == []