diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8b97d737..60f5ee11 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,18 @@
 
 ### Fixed
 
+- **A rejected request no longer poisons the next one on the same connection.** `server.py`
+  is a raw HTTP/1.1 handler where `rfile` is the socket itself, so answering a request
+  before reading its body left those bytes queued. The next request on a keep-alive
+  connection was then parsed starting mid-body, and the client got
+  `400 Bad request syntax ('{"a": "b"}GET /api/health HTTP/1.1')` — an error naming a
+  request it never sent, which is expensive to diagnose from the client side. Every
+  reject path now arms `Connection: close` when a body is still pending, and framing is
+  validated strictly (RFC 9110 `1*DIGIT`, duplicate and comma-combined `Content-Length`
+  reconciled, every `Transfer-Encoding` refused since nothing here decodes one). A
+  genuinely bodyless rejection keeps its keep-alive, so healthy pooled connections are
+  not dropped. Thanks @rodrigogs. (#7550, #6658)
+
 - **Opening a session that belongs to another profile now offers to switch to it instead of
   looking deleted.** Several cross-profile guards answered `404 Session not found`, which the
   front-end treats as a missing session and self-heals by clearing the URL and local storage —
diff --git a/api/auth.py b/api/auth.py
index 846c9945..c61ee669 100644
--- a/api/auth.py
+++ b/api/auth.py
@@ -17,6 +17,7 @@ import time
 from pathlib import Path
 
 from api.config import STATE_DIR, get_config, load_settings
+from api.helpers import request_declares_body
 
 logger = logging.getLogger(__name__)
 
@@ -1207,6 +1208,42 @@ def check_auth(handler, parsed) -> bool:
     return False
 
 
+def check_auth_or_close(handler, parsed) -> bool:
+    """Check auth; when rejected, close so an unread body can't poison HTTP/1.1 reuse.
+
+    The flag is armed BEFORE check_auth() writes its 401/302, so end_headers()
+    can advertise ``Connection: close``; success restores the prior flag so an
+    authenticated request keeps its keep-alive.
+
+    Armed ONLY when the request's framing declares a body still queued in
+    ``rfile`` (see ``request_declares_body()``), which is the same rule the other
+    reject-before-read sites apply through ``arm_connection_close_if_body_pending()``.
+    Arming unconditionally was the mirror image of the over-close this PR already
+    fixed on the sidecar path: a body-less POST that failed auth answered
+    ``401`` WITH ``Connection: close`` and dropped the client's pipelined
+    follow-up, killing a healthy keep-alive connection for no framing reason.
+    Verified on the wire, pipelined down one socket against the production
+    handler: ``POST /api/session/new`` with no ``Content-Length`` (and with
+    ``Content-Length: 0``) answered ``401`` + ``Connection: close`` and the
+    following ``GET /api/auth/status`` was never served. The helper cannot be
+    swapped in directly here because the arming has to happen before
+    ``check_auth()`` writes its response and be undone if it succeeds.
+
+    ``close_connection`` only exists once BaseHTTPRequestHandler has parsed a
+    request line, so partially-built handler stubs may not have it at all. Read
+    it defensively and only arm/restore when it was really there: inventing the
+    attribute on a stub would leave a bogus keep-alive verdict behind.
+    """
+    if not hasattr(handler, 'close_connection') or not request_declares_body(handler):
+        return check_auth(handler, parsed)
+    prior_close = handler.close_connection
+    handler.close_connection = True
+    if check_auth(handler, parsed):
+        handler.close_connection = prior_close
+        return True
+    return False
+
+
 def _is_loopback(addr: str) -> bool:
     """Return True if *addr* is a loopback address (127.x.x.x, ::1, or ::ffff:127.x.x.x)."""
     import ipaddress as _ipaddress
diff --git a/api/helpers.py b/api/helpers.py
index 6e658f0d..1984bf28 100644
--- a/api/helpers.py
+++ b/api/helpers.py
@@ -202,6 +202,248 @@ def _security_headers(handler):
     )
 
 
+def arm_connection_close(handler) -> None:
+    """Mark the connection dead BEFORE a reject-before-read response is written.
+
+    A rejection that answers without consuming the request body leaves those
+    bytes queued in ``rfile``; reusing the connection makes the next HTTP/1.1
+    request parse mid-body (the classic ``{}GET ... -> 501``). Arming must
+    happen before the response is flushed so ``advertise_connection_close()``
+    can serialize it into the response headers.
+
+    Best-effort: handler stubs and read-only doubles may reject the attribute,
+    and a failure here must never turn a 4xx into a 500.
+    """
+    try:
+        handler.close_connection = True
+    except Exception:
+        pass
+
+
+# What ``unsupported_transfer_encoding()`` reports for a header that declares no
+# coding at all (``Transfer-Encoding:``). A placeholder rather than the raw '' so
+# the return value is never a falsy non-``None`` — callers test ``is not None``,
+# and an empty string would also render as a hole in the rejection message.
+_BLANK_TRANSFER_CODING = '(blank)'
+
+# The only whitespace a field value may be padded with: RFC 9110 OWS is SP/HTAB
+# and nothing else. ``str.strip()`` would also erase U+00A0 and U+0085 -- both
+# reachable over the wire, because request lines decode as latin-1 -- and
+# ``Content-Length: \xa00`` would then read back as a genuine ``0`` while its
+# payload sat unread on the socket. Padding a value with a character the grammar
+# does not allow makes it unreadable, not zero.
+_FIELD_VALUE_OWS = ' \t'
+
+
+def _framing_header_values(handler, name: str) -> list[str]:
+    """Every value the wire carried for one framing header, tolerating stubs.
+
+    ALL values, not just the first: ``Message.get()`` returns only the first
+    occurrence, so a request carrying ``Content-Length: 0`` followed by
+    ``Content-Length: 42`` reads back as length zero. Its 42 body bytes are then
+    never consumed and get parsed as the next request line -- the same poisoning
+    this module exists to prevent, reached by duplicating a header rather than by
+    omitting one. Conflicting framing is only visible if every value is.
+
+    A real handler carries an ``email.message.Message`` (case-insensitive
+    ``get_all()``, like the wire); the suite's ``SimpleNamespace(headers={...})``
+    doubles are plain dicts, which are not, so fall back to a case-insensitive
+    scan for those. A missing or odd ``headers`` must never raise: every caller
+    is on a rejection path where an exception becomes a spurious 500.
+    """
+    headers = getattr(handler, 'headers', None)
+    if headers is None:
+        return []
+    try:
+        get_all = getattr(headers, 'get_all', None)
+        if callable(get_all):
+            values = get_all(name)
+            return [] if values is None else [str(v) for v in values if v is not None]
+        if isinstance(headers, dict):
+            lowered = name.lower()
+            return [
+                str(v) for k, v in headers.items()
+                if str(k).lower() == lowered and v is not None
+            ]
+        value = headers.get(name)
+        return [] if value is None else [str(value)]
+    except Exception:
+        return []
+
+
+def _declared_content_lengths(handler) -> set[int] | None:
+    """Distinct declared ``Content-Length`` values, or ``None`` if any is unreadable.
+
+    ``None`` means the framing cannot be trusted at all (a value that will not
+    parse); an empty set means no length was declared.
+
+    A BLANK or whitespace-only value is unreadable, NOT absent. ``Content-Length:``
+    with payload bytes behind it declares a body whose size the header refuses to
+    state; skipping the empty value reported "no length declared", so the
+    rejection kept the connection alive and those bytes were parsed as the next
+    request line (403 then ``400 Bad request syntax ('{...}GET /api/... HTTP/1.1')``).
+    Absence is something only the wire can say, and ``_framing_header_values()``
+    already reports a header the request never carried as no value at all -- so
+    anything that *is* here and does not parse has to be treated as pending bytes,
+    exactly like ``banana`` or two lengths that disagree. Note ``Message`` collapses
+    ``Content-Length:`` and ``Content-Length:   `` to the same ``''``.
+
+    "Parses" means RFC 9110's ``Content-Length = 1*DIGIT``: an unsigned run of
+    ASCII digits, nothing else. ``int()`` is far more generous than the grammar --
+    it accepts a leading sign, PEP 515 underscores and non-ASCII digits -- and
+    every such spelling of ZERO used to read back as an honest ``0`` and take the
+    keep-alive branch below with its payload still queued: ``Content-Length: +0``
+    (also ``-0``, ``0_0``, ``+00``, ``\\xa00``) answered 403 with no
+    ``Connection: close``, then ``400 Bad request syntax ('{...}GET /api/...')``
+    -- the same trace as a blank value. A sign is only invisible on a NON-zero
+    value, where the wrong parse happens to close anyway. Anything the grammar
+    rejects is unreadable framing, which means a body may be pending.
+
+    The ``int()`` guard stays even though ``isdigit()`` has already vetted every
+    character: a digit run longer than ``sys.get_int_max_str_digits()`` (4300)
+    raises ``ValueError`` too, and this runs on rejection paths where an
+    exception becomes a spurious 500.
+    """
+    parsed: set[int] = set()
+    for raw in _framing_header_values(handler, 'Content-Length'):
+        # RFC 9110 §5.3: a list-valued field may arrive as repeated lines OR as
+        # one comma-combined line, and the two spellings mean the same thing.
+        # So "0, 0" is the same bodyless request as two "Content-Length: 0"
+        # headers and must keep its keep-alive; only a DISAGREEING list (caught
+        # below by len(parsed) > 1) is unframeable. Without the split the comma
+        # fails isdigit() and every agreeing list over-closed a healthy socket.
+        for member in raw.split(','):
+            stripped = member.strip(_FIELD_VALUE_OWS)
+            if not (stripped.isascii() and stripped.isdigit()):
+                return None
+            try:
+                parsed.add(int(stripped))
+            except ValueError:
+                return None
+    return parsed
+
+
+def unreadable_content_length(handler) -> bool:
+    """True when ``Content-Length`` cannot be trusted to frame the body.
+
+    Either a value that will not parse (``banana``, ``0x5``, or a BLANK one), or
+    two or more values that disagree. None of them can tell a reader how many
+    bytes to drain, so such a request can only be rejected -- with the connection
+    armed for close, since bytes may still be queued behind it.
+    """
+    parsed = _declared_content_lengths(handler)
+    return parsed is None or len(parsed) > 1
+
+
+def request_declares_body(handler) -> bool:
+    """True when the request's framing declares a body that is still queued.
+
+    Keyed on the wire, not on the method: ``Transfer-Encoding`` present (chunked
+    framing hides the length until the terminating chunk is read) or
+    ``Content-Length`` present and non-zero. The method proves nothing in either
+    direction -- a GET may carry a declared body and a POST/PUT/DELETE may carry
+    none -- and a static per-method flag therefore both misses body-bearing
+    "read" requests (whose unread bytes poison the next pooled request) and
+    kills healthy keep-alive on body-less writes.
+
+    A blank, unparseable or self-contradicting ``Content-Length`` counts as a
+    declared body, as does ANY ``Transfer-Encoding`` -- ``identity`` and a blank
+    value included: a value we cannot interpret does not prove the body's absence,
+    and the safe assumption for connection reuse is that bytes are pending. Only
+    framing that positively says "no body" -- no framing header at all, or every
+    declared length agreeing on zero -- keeps the connection alive.
+    """
+    if unsupported_transfer_encoding(handler) is not None:
+        return True
+    parsed = _declared_content_lengths(handler)
+    if parsed is None:
+        return True
+    return any(length != 0 for length in parsed)
+
+
+def unsupported_transfer_encoding(handler) -> str | None:
+    """Return the transfer coding this server cannot decode, or ``None``.
+
+    EVERY present value counts, with no exemption for any coding.
+    ``BaseHTTPRequestHandler`` decodes none of them: it hands every reader a raw
+    ``rfile``, so a chunked body reads back as its literal chunk framing while a
+    ``Content-Length``-based reader sees length 0 and consumes nothing at all.
+    RFC 9112 section 6.3 says the same thing normatively -- a request whose final
+    transfer coding is not ``chunked`` has a body length the recipient cannot
+    determine, so it must be refused and the connection closed. Such a request
+    can therefore only be rejected, and only with the connection armed for close,
+    because its payload bytes are still queued on the socket.
+
+    ``identity`` used to be exempt as a coding that "frames nothing". It is not:
+    it names no framing either, and RFC 7230 removed it from the transfer codings
+    altogether. A sidecar rejection then answered 403 with keep-alive and the
+    payload was parsed as the next request line
+    (``400 Bad request syntax ('{...}GET /api/health/agent HTTP/1.1')``), and all
+    four upload handlers fell through to a length-0 read, answered 400 "No file
+    field in request" and left the multipart bytes on the socket
+    (``400 Bad request syntax ('--x')``). Nothing legitimate is lost: no client
+    sends a transfer coding this server could have honoured.
+
+    A header present but EMPTY (``Transfer-Encoding:``) is reported through
+    ``_BLANK_TRANSFER_CODING`` rather than as ``''``, so callers testing
+    ``is not None`` and the 411 message both read right. Only an ABSENT header
+    yields no values at all -- absence is something only the wire can say.
+
+    The first value is enough precisely because no value is acceptable; it is
+    reported as-is (SP/HTAB trimmed, the one padding RFC 9110 allows) purely so
+    the rejection message names what arrived.
+    """
+    values = _framing_header_values(handler, 'Transfer-Encoding')
+    if not values:
+        return None
+    return values[0].strip(_FIELD_VALUE_OWS) or _BLANK_TRANSFER_CODING
+
+
+def arm_connection_close_if_body_pending(handler) -> bool:
+    """Arm close for a reject-before-read, but only if a body was really declared.
+
+    Returns whether the connection was armed. Use this instead of
+    ``arm_connection_close()`` wherever the rejection can also fire on a request
+    with no body: closing then would drop a pooled client's healthy connection
+    for no framing reason.
+    """
+    if not request_declares_body(handler):
+        return False
+    arm_connection_close(handler)
+    return True
+
+
+def advertise_connection_close(handler) -> None:
+    """Serialize ``close_connection`` into exactly one ``Connection: close`` header.
+
+    ``BaseHTTPRequestHandler`` tracks ``close_connection`` internally but never
+    writes it to the wire, so a server-side close is invisible to the client: a
+    pooled client reuses the socket and fails with BrokenPipeError instead of
+    opening a fresh connection. Call this from ``end_headers()``, before the
+    header buffer is flushed.
+
+    Dedup is derived from the pending header buffer rather than a private flag,
+    because the SSE endpoints send their own ``Connection: close`` (and
+    ``send_header`` itself flips ``close_connection`` when they do) — keying off
+    a flag we set ourselves would emit the header twice on every stream.
+
+    Both attribute reads are defensive: partially-built handler stubs in the
+    test suite carry neither ``close_connection`` nor ``_headers_buffer``, and
+    header emission must not depend on their presence. The buffer is also
+    type-checked before it is scanned -- a stub whose ``_headers_buffer`` is a
+    truthy non-iterable (a bare ``Mock``) would otherwise raise ``TypeError``
+    from inside ``end_headers()``, which ``_handle_write`` would swallow into a
+    spurious 500.
+    """
+    if not getattr(handler, 'close_connection', False):
+        return
+    buffered = getattr(handler, '_headers_buffer', None)
+    for raw in buffered if isinstance(buffered, (list, tuple)) else ():
+        if raw[:11].lower() == b'connection:':
+            return
+    handler.send_header('Connection', 'close')
+
+
 def flush_pending_auth_cookies(handler) -> None:
     pending = getattr(handler, '_pending_set_cookies', None)
     if not pending:
diff --git a/api/routes.py b/api/routes.py
index d5b4a891..daca7c80 100644
--- a/api/routes.py
+++ b/api/routes.py
@@ -2974,6 +2974,7 @@ from api.helpers import (
     require,
     bad,
     safe_resolve,
+    arm_connection_close_if_body_pending,
     j,
     t,
     read_body,
@@ -5871,6 +5872,15 @@ def _csrf_rejection_error(handler) -> str:
 def _check_csrf(handler) -> bool:
     """Reject cross-origin or tokenless authenticated browser unsafe requests."""
     if not _check_same_origin_browser_request(handler):
+        # CSRF checks run before read_body(), so close rather than reusing an
+        # HTTP/1.1 connection whose unread body would corrupt the next request --
+        # but ONLY when the framing says bytes are really queued. Arming
+        # unconditionally dropped a healthy pooled connection on a body-less
+        # write: verified on the wire, `POST /api/session/new` with
+        # `Origin: http://evil.invalid` and no `Content-Length` (and with
+        # `Content-Length: 0`) answered 403 + `Connection: close` and the
+        # pipelined `GET /api/health/agent` was never served.
+        arm_connection_close_if_body_pending(handler)
         return False
     if not _is_browser_unsafe_request(handler):
         return True  # non-browser clients (curl, MCP, agent) have no Origin/Referer
@@ -5883,6 +5893,11 @@ def _check_csrf(handler) -> bool:
     submitted = handler.headers.get(CSRF_HEADER_NAME) or handler.headers.get("X-CSRF-Token")
     if verify_csrf_token(cookie_val or "", submitted or ""):
         return True
+    # Same framing rule as the origin rejection above: a token mismatch on a
+    # body-less write has nothing unread to protect. Verified on the wire with an
+    # authenticated same-origin `POST /api/session/new` carrying no CSRF token and
+    # no `Content-Length`: 403 + `Connection: close`, follow-up dropped.
+    arm_connection_close_if_body_pending(handler)
     return _set_csrf_failure_reason(handler, "token_mismatch")
 
 
@@ -6062,6 +6077,14 @@ def _handle_extension_sidecar_proxy(
     # DELETE fell through the CSRF compatibility path that intentionally admits
     # non-browser clients, giving unsafe methods weaker provenance than GET.
     if not _check_same_origin_browser_request(handler, require_provenance=True):
+        # Provenance rejection runs before read_body(), so close-and-advertise
+        # whenever the request DECLARED a body (Content-Length non-zero, or any
+        # Transfer-Encoding): those bytes are still queued in rfile and a reused
+        # HTTP/1.1 connection would parse them as the next request line (same
+        # class as _check_csrf). Gating on read_request_body instead was wrong in
+        # both directions — it missed a GET that carries a declared body, and it
+        # closed a healthy connection on a body-less DELETE/PUT/PATCH.
+        arm_connection_close_if_body_pending(handler)
         return j(handler, {"error": _csrf_rejection_error(handler)}, status=403)
     try:
         request_body = _read_body_bytes(handler) if read_request_body else None
@@ -6504,6 +6527,13 @@ def _handle_csp_report(handler) -> bool:
             "Dropped CSP report from %s: rate limit exceeded",
             _client_ip_for_rate_limit(handler),
         )
+        # Rate-limit rejection runs before the body is read; close-and-advertise
+        # so the unread report can't corrupt the next pooled request -- but only
+        # when a body was really declared. A body-less report answered 204 WITH
+        # `Connection: close` once the 100-per-60s limiter tripped (reproduced on
+        # the wire at request 101; the pipelined `GET /api/health/agent` was
+        # dropped), so a browser that keeps reporting loses its socket each time.
+        arm_connection_close_if_body_pending(handler)
         return _send_no_content(handler)
 
     payload = _read_csp_report_payload(handler)
@@ -12895,6 +12925,15 @@ def _handle_shutdown(handler) -> bool:
 
 def _handle_health_restart(handler) -> bool:
     """Restart the Hermes messaging gateway service."""
+    # This endpoint never consumes its request body on any outcome, so close when
+    # one was DECLARED -- and only then. Arming unconditionally closed the socket
+    # on every call including the successful, body-less one the WebUI actually
+    # makes: verified on the wire, `POST /api/health/restart` with no
+    # `Content-Length` answered with `Connection: close` and the pipelined
+    # `GET /api/health/agent` was never served. The single arming covers every
+    # outcome below (completed / in_progress / busy / error) because the framing,
+    # not the result, decides.
+    arm_connection_close_if_body_pending(handler)
     outcome = restart_active_profile_gateway()
 
     if outcome.get("status") == "completed":
@@ -15217,6 +15256,12 @@ def handle_post(handler, parsed) -> bool:
         if diag:
             diag.stage("process_complete_ack_deprecated")
         try:
+            # The 410 runs before the stale tab's JSON body is read;
+            # close-and-advertise so those unread bytes can't corrupt the next
+            # pooled request -- but a body-less ack has none, and closing then
+            # just kills keep-alive (on the wire: 410 + `Connection: close` with
+            # no `Content-Length`, pipelined follow-up dropped).
+            arm_connection_close_if_body_pending(handler)
             j(
                 handler,
                 {
diff --git a/api/upload.py b/api/upload.py
index 6269fca5..3701e669 100644
--- a/api/upload.py
+++ b/api/upload.py
@@ -8,7 +8,12 @@ import tempfile
 from pathlib import Path
 
 from api.config import MAX_UPLOAD_BYTES, STATE_DIR
-from api.helpers import j
+from api.helpers import (
+    arm_connection_close_if_body_pending,
+    j,
+    unreadable_content_length,
+    unsupported_transfer_encoding,
+)
 from api.models import get_session
 from api.profiles import _profiles_match, get_active_profile_name as _get_active_profile_name
 from api.workspace import (
@@ -45,6 +50,89 @@ def _max_extracted_bytes() -> int:
 _MAX_EXTRACTED_BYTES = 10 * MAX_UPLOAD_BYTES
 
 
+def _reject_before_read(handler, message, status):
+    """Reject without reading the body; close so unread bytes can't poison HTTP/1.1 reuse.
+
+    Framing decides, exactly as at the other reject-before-read sites: every
+    rejection raised by the preflight above (any ``Transfer-Encoding``, an
+    unreadable or conflicting ``Content-Length``, an over-cap length) already
+    declares a body and still closes. Only the boundary/`parse_multipart` rejection
+    can fire on a request that declares none, and closing there dropped a healthy
+    pooled connection -- on the wire, ``POST /api/upload`` with
+    ``Content-Type: application/json`` and no ``Content-Length`` answered 400
+    "No boundary in Content-Type" WITH ``Connection: close`` and the pipelined
+    ``GET /api/health/agent`` was never served.
+    """
+    arm_connection_close_if_body_pending(handler)
+    return j(handler, {'error': message}, status=status)
+
+
+class _RejectBeforeRead(ValueError):
+    """A rejection raised before any byte of the request body was consumed.
+
+    Carries the status the caller must answer with. Subclasses ``ValueError`` so
+    that if one ever escapes an unusual call path, each handler's existing
+    ``except ValueError`` still answers 400 with the same message instead of
+    turning it into a 500.
+    """
+
+    def __init__(self, message: str, status: int):
+        super().__init__(message)
+        self.message = message
+        self.status = status
+
+
+def _read_multipart_or_reject(handler):
+    """Framing preflight + body read, shared by all four multipart handlers.
+
+    Returns ``(fields, files)``. Raises ``_RejectBeforeRead`` when the request
+    must be refused before its body is read -- the caller answers through
+    ``_reject_before_read()``, which arms ``Connection: close`` first, because the
+    declared body is still queued in ``rfile`` and a reused HTTP/1.1 connection
+    would parse those bytes as the next request line.
+
+    Framing is judged from the headers, not from a present ``Content-Length``:
+
+    * ANY ``Transfer-Encoding`` is refused, ``identity`` included: ``http.server``
+      decodes none of them (``rfile`` is the raw socket) and RFC 9112 section 6.3
+      says a request whose final coding is not ``chunked`` has a length no
+      recipient can determine. A coded request carries no usable Content-Length,
+      so the old code read it as length 0, found no parts, and answered "No file
+      field in request" with keep-alive intact -- leaving every payload byte on
+      the socket to poison the next pooled request. A Content-Length sent BESIDE a
+      coding does not rescue it: that pairing is the smuggling shape 6.3 rejects,
+      and the coding still hides how the body is framed.
+    * a non-numeric or oversized Content-Length cannot be read either, and
+      ``parse_multipart()`` validates the boundary and the length before its
+      first ``rfile.read()`` — so every raise below is still pre-read.
+    * two DISAGREEING Content-Length headers read back as whichever came first,
+      so ``Content-Length: 0`` ahead of the real length made the body invisible
+      and left every byte of it on the socket. There is no honest length to
+      drain, so the request is refused before ``rfile`` is touched.
+    """
+    content_type = handler.headers.get('Content-Type', '')
+    coding = unsupported_transfer_encoding(handler)
+    if coding is not None:
+        raise _RejectBeforeRead(
+            f'Unsupported Transfer-Encoding: {coding} '
+            '(uploads must send an undecoded body with a Content-Length)',
+            411,
+        )
+    if unreadable_content_length(handler):
+        raise _RejectBeforeRead('Invalid Content-Length', 400)
+    try:
+        content_length = int(handler.headers.get('Content-Length', 0) or 0)
+    except (TypeError, ValueError):
+        raise _RejectBeforeRead('Invalid Content-Length', 400) from None
+    if content_length > MAX_UPLOAD_BYTES:
+        raise _RejectBeforeRead(
+            f'File too large (max {MAX_UPLOAD_BYTES//1024//1024}MB)', 413)
+    try:
+        return parse_multipart(handler.rfile, content_type, content_length)
+    except ValueError as exc:
+        raise _RejectBeforeRead(str(exc), 400) from None
+
+
 def parse_multipart(rfile, content_type, content_length) -> tuple:
     import re as _re, email.parser as _ep
     # Imported locally (not just module-level) so the function stays
@@ -206,11 +294,10 @@ def _write_office_upload_sidecar(workspace: Path, dest: Path, file_bytes: bytes)
 def handle_upload(handler):
     import traceback as _tb
     try:
-        content_type = handler.headers.get('Content-Type', '')
-        content_length = int(handler.headers.get('Content-Length', 0) or 0)
-        if content_length > MAX_UPLOAD_BYTES:
-            return j(handler, {'error': f'File too large (max {MAX_UPLOAD_BYTES//1024//1024}MB)'}, status=413)
-        fields, files = parse_multipart(handler.rfile, content_type, content_length)
+        try:
+            fields, files = _read_multipart_or_reject(handler)
+        except _RejectBeforeRead as rejection:
+            return _reject_before_read(handler, rejection.message, rejection.status)
         session_id = fields.get('session_id', '')
         if 'file' not in files:
             return j(handler, {'error': 'No file field in request'}, status=400)
@@ -396,11 +483,10 @@ def handle_upload_extract(handler):
     """Handle archive upload and extraction."""
     import traceback as _tb
     try:
-        content_type = handler.headers.get('Content-Type', '')
-        content_length = int(handler.headers.get('Content-Length', 0) or 0)
-        if content_length > MAX_UPLOAD_BYTES:
-            return j(handler, {'error': f'File too large (max {MAX_UPLOAD_BYTES//1024//1024}MB)'}, status=413)
-        fields, files = parse_multipart(handler.rfile, content_type, content_length)
+        try:
+            fields, files = _read_multipart_or_reject(handler)
+        except _RejectBeforeRead as rejection:
+            return _reject_before_read(handler, rejection.message, rejection.status)
         session_id = fields.get('session_id', '')
         if 'file' not in files:
             return j(handler, {'error': 'No file field in request'}, status=400)
@@ -428,11 +514,10 @@ def handle_transcribe(handler):
     import traceback as _tb
     temp_path = None
     try:
-        content_type = handler.headers.get('Content-Type', '')
-        content_length = int(handler.headers.get('Content-Length', 0) or 0)
-        if content_length > MAX_UPLOAD_BYTES:
-            return j(handler, {'error': f'File too large (max {MAX_UPLOAD_BYTES//1024//1024}MB)'}, status=413)
-        fields, files = parse_multipart(handler.rfile, content_type, content_length)
+        try:
+            fields, files = _read_multipart_or_reject(handler)
+        except _RejectBeforeRead as rejection:
+            return _reject_before_read(handler, rejection.message, rejection.status)
         if 'file' not in files:
             return j(handler, {'error': 'No file field in request'}, status=400)
         filename, file_bytes = files['file']
@@ -612,12 +697,10 @@ def handle_workspace_upload(handler):
     """
     import traceback as _tb
     try:
-        content_type = handler.headers.get('Content-Type', '')
-        content_length = int(handler.headers.get('Content-Length', 0) or 0)
-        if content_length > MAX_UPLOAD_BYTES:
-            return j(handler, {'error': f'File too large (max {MAX_UPLOAD_BYTES//1024//1024}MB)'}, status=413)
-
-        fields, files = parse_multipart(handler.rfile, content_type, content_length)
+        try:
+            fields, files = _read_multipart_or_reject(handler)
+        except _RejectBeforeRead as rejection:
+            return _reject_before_read(handler, rejection.message, rejection.status)
         session_id = fields.get('session_id', '')
         subpath = fields.get('path', '')
 
diff --git a/server.py b/server.py
index 040d80c1..06fc9a21 100644
--- a/server.py
+++ b/server.py
@@ -100,10 +100,11 @@ from urllib.parse import urlparse
 logger = logging.getLogger(__name__)
 
 from api.request_logging import emit_request_log
-from api.auth import check_auth, reset_trusted_auth_request_state
+from api.auth import check_auth_or_close, reset_trusted_auth_request_state
 from api.config import HOST, PORT, STATE_DIR, SESSION_DIR, DEFAULT_WORKSPACE
 from api.helpers import (
     j,
+    advertise_connection_close,
     get_profile_cookie,
     _build_csp_report_only_policy,
     _CLIENT_DISCONNECT_ERRORS,
@@ -332,6 +333,7 @@ class Handler(BaseHTTPRequestHandler):
         extra_frame_src = getattr(self, "_csp_extra_frame_src", None)
         self.send_header("Content-Security-Policy-Report-Only", self.csp_report_only_policy(extra_connect_src, extra_frame_src))
         self.send_header("Report-To", self._CSP_REPORT_TO)
+        advertise_connection_close(self)  # tell the client when the socket dies
         super().end_headers()
 
     def log_message(self, fmt, *args): pass  # suppress default Apache-style log
@@ -376,7 +378,8 @@ class Handler(BaseHTTPRequestHandler):
             set_request_profile(cookie_profile)
         try:
             parsed = urlparse(self.path)
-            if not check_auth(self, parsed): return
+            # Body-pending-aware: a body-bearing GET failing auth would poison reuse (#7550).
+            if not check_auth_or_close(self, parsed): return
             result = handle_get(self, parsed)
             if result is False:
                 return j(self, {'error': 'not found'}, status=404)
@@ -401,10 +404,8 @@ class Handler(BaseHTTPRequestHandler):
             set_request_profile(cookie_profile)
         try:
             parsed = urlparse(self.path)
-            _is_csp_report_post = (
-                parsed.path == "/api/csp-report" and self.command == "POST"
-            )
-            if not _is_csp_report_post and not check_auth(self, parsed): return
+            _is_csp_report_post = parsed.path == "/api/csp-report" and self.command == "POST"
+            if not _is_csp_report_post and not check_auth_or_close(self, parsed): return
             result = route_func(self, parsed)
             if result is False:
                 return j(self, {'error': 'not found'}, status=404)
diff --git a/tests/test_broken_pipe_cascade.py b/tests/test_broken_pipe_cascade.py
index c981d3eb..9e612670 100644
--- a/tests/test_broken_pipe_cascade.py
+++ b/tests/test_broken_pipe_cascade.py
@@ -227,14 +227,14 @@ class TestServerDisconnectHandling(unittest.TestCase):
         # Patch handle_get to raise BrokenPipeError
         import server as _server_mod
         orig_handle_get = _server_mod.handle_get
-        orig_check_auth = _server_mod.check_auth
+        orig_check_auth = _server_mod.check_auth_or_close
         _server_mod.handle_get = _fake_handle_get
-        _server_mod.check_auth = lambda h, p: True
+        _server_mod.check_auth_or_close = lambda h, p: True
         try:
             Handler.do_GET(handler)
         finally:
             _server_mod.handle_get = orig_handle_get
-            _server_mod.check_auth = orig_check_auth
+            _server_mod.check_auth_or_close = orig_check_auth
 
         # send_response should NEVER be called for the 500 — client is gone
         handler.send_response.assert_not_called()
@@ -248,12 +248,12 @@ class TestServerDisconnectHandling(unittest.TestCase):
             raise self._route_raises
 
         import server as _server_mod
-        orig_check_auth = _server_mod.check_auth
-        _server_mod.check_auth = lambda h, p: True
+        orig_check_auth = _server_mod.check_auth_or_close
+        _server_mod.check_auth_or_close = lambda h, p: True
         try:
             Handler._handle_write(handler, _fake_route)
         finally:
-            _server_mod.check_auth = orig_check_auth
+            _server_mod.check_auth_or_close = orig_check_auth
 
         handler.send_response.assert_not_called()
 
@@ -286,9 +286,9 @@ class TestServerDisconnectHandling(unittest.TestCase):
 
         import server as _server_mod
         orig_handle_get = _server_mod.handle_get
-        orig_check_auth = _server_mod.check_auth
+        orig_check_auth = _server_mod.check_auth_or_close
         _server_mod.handle_get = _fake_handle_get
-        _server_mod.check_auth = lambda h, p: True
+        _server_mod.check_auth_or_close = lambda h, p: True
 
         closed_stdout = io.StringIO()
         closed_stdout.close()
@@ -299,7 +299,7 @@ class TestServerDisconnectHandling(unittest.TestCase):
         finally:
             sys.stdout = original_stdout
             _server_mod.handle_get = orig_handle_get
-            _server_mod.check_auth = orig_check_auth
+            _server_mod.check_auth_or_close = orig_check_auth
 
         handler.send_response.assert_called_once_with(500)
 
@@ -312,14 +312,14 @@ class TestServerDisconnectHandling(unittest.TestCase):
 
         import server as _server_mod
         orig_handle_get = _server_mod.handle_get
-        orig_check_auth = _server_mod.check_auth
+        orig_check_auth = _server_mod.check_auth_or_close
         _server_mod.handle_get = _fake_handle_get
-        _server_mod.check_auth = lambda h, p: True
+        _server_mod.check_auth_or_close = lambda h, p: True
         try:
             Handler.do_GET(handler)
         finally:
             _server_mod.handle_get = orig_handle_get
-            _server_mod.check_auth = orig_check_auth
+            _server_mod.check_auth_or_close = orig_check_auth
 
         # Should send 500 for real errors
         handler.send_response.assert_called_once_with(500)
@@ -334,14 +334,14 @@ class TestServerDisconnectHandling(unittest.TestCase):
 
         import server as _server_mod
         orig_handle_get = _server_mod.handle_get
-        orig_check_auth = _server_mod.check_auth
+        orig_check_auth = _server_mod.check_auth_or_close
         _server_mod.handle_get = _fake_handle_get
-        _server_mod.check_auth = lambda h, p: True
+        _server_mod.check_auth_or_close = lambda h, p: True
         try:
             Handler.do_GET(handler)
         finally:
             _server_mod.handle_get = orig_handle_get
-            _server_mod.check_auth = orig_check_auth
+            _server_mod.check_auth_or_close = orig_check_auth
 
         handler.send_response.assert_not_called()
 
@@ -355,14 +355,14 @@ class TestServerDisconnectHandling(unittest.TestCase):
 
         import server as _server_mod
         orig_handle_get = _server_mod.handle_get
-        orig_check_auth = _server_mod.check_auth
+        orig_check_auth = _server_mod.check_auth_or_close
         _server_mod.handle_get = _fake_handle_get
-        _server_mod.check_auth = lambda h, p: True
+        _server_mod.check_auth_or_close = lambda h, p: True
         try:
             Handler.do_GET(handler)
         finally:
             _server_mod.handle_get = orig_handle_get
-            _server_mod.check_auth = orig_check_auth
+            _server_mod.check_auth_or_close = orig_check_auth
 
         handler.send_response.assert_not_called()
 
@@ -406,14 +406,14 @@ class TestServer500ResponseSafety(unittest.TestCase):
 
         import server as _server_mod
         orig_handle_get = _server_mod.handle_get
-        orig_check_auth = _server_mod.check_auth
+        orig_check_auth = _server_mod.check_auth_or_close
         _server_mod.handle_get = _fake_handle_get
-        _server_mod.check_auth = lambda h, p: True
+        _server_mod.check_auth_or_close = lambda h, p: True
         try:
             Handler.do_GET(handler)
         finally:
             _server_mod.handle_get = orig_handle_get
-            _server_mod.check_auth = orig_check_auth
+            _server_mod.check_auth_or_close = orig_check_auth
 
         # send_response WAS called (we tried to send 500), but write failed
         handler.send_response.assert_called_once_with(500)
@@ -431,14 +431,14 @@ class TestServer500ResponseSafety(unittest.TestCase):
 
         import server as _server_mod
         orig_handle_get = _server_mod.handle_get
-        orig_check_auth = _server_mod.check_auth
+        orig_check_auth = _server_mod.check_auth_or_close
         _server_mod.handle_get = _fake_handle_get
-        _server_mod.check_auth = lambda h, p: True
+        _server_mod.check_auth_or_close = lambda h, p: True
         try:
             Handler.do_GET(handler)
         finally:
             _server_mod.handle_get = orig_handle_get
-            _server_mod.check_auth = orig_check_auth
+            _server_mod.check_auth_or_close = orig_check_auth
 
         # send_response WAS called (we tried to send 500), but write failed
         handler.send_response.assert_called_once_with(500)
diff --git a/tests/test_comma_combined_content_length.py b/tests/test_comma_combined_content_length.py
new file mode 100644
index 00000000..fc93f2a3
--- /dev/null
+++ b/tests/test_comma_combined_content_length.py
@@ -0,0 +1,105 @@
+"""Comma-combined ``Content-Length`` must not over-close a healthy connection.
+
+RFC 9110 §5.3 lets a list-valued field arrive either as repeated header lines or
+as a single comma-combined line, and the two spellings are equivalent. So
+``Content-Length: 0, 0`` describes the same bodyless request as two separate
+``Content-Length: 0`` headers, and rejecting it must keep the keep-alive.
+
+``_declared_content_lengths()`` originally ran ``isdigit()`` across the whole raw
+value, so the comma failed the check, the field read as *unreadable*, and every
+helper-backed rejection armed ``Connection: close`` on a request with nothing
+queued behind it. That is the over-close half of this PR's contract — the mirror
+image of the bug it fixes, and the one that silently drops pooled connections.
+
+The dangerous direction has to stay closed, so the disagreeing and malformed
+spellings are asserted here alongside the recoverable ones: ``0, 42`` genuinely
+cannot frame a body and must still be refused with the close armed.
+"""
+
+import pytest
+
+from api.helpers import request_declares_body, unreadable_content_length
+
+
+class _Hdrs(dict):
+    """email.message-like: ``get()`` returns the first value, ``get_all()`` all."""
+
+    def get(self, k, default=None):
+        for key, val in self.items():
+            if key.lower() == k.lower():
+                return val[0] if isinstance(val, list) else val
+        return default
+
+    def get_all(self, k, default=None):
+        out = []
+        for key, val in self.items():
+            if key.lower() == k.lower():
+                out.extend(val if isinstance(val, list) else [val])
+        return out or default
+
+
+class _H:
+    def __init__(self, headers):
+        self.headers = headers
+
+
+def _handler(cl):
+    return _H(_Hdrs({"Content-Length": cl}))
+
+
+@pytest.mark.parametrize("value", ["0, 0", "00, 00", "0 , 0", "0,0", "0,\t0"])
+def test_agreeing_bodyless_comma_list_is_readable_and_bodyless(value):
+    """An agreeing all-zero list is framed, bodyless, and must keep keep-alive."""
+    h = _handler(value)
+    assert unreadable_content_length(h) is False, (
+        f"Content-Length {value!r} read as unreadable; a bodyless rejection would "
+        "then arm Connection: close and drop a healthy pooled connection"
+    )
+    assert bool(request_declares_body(h)) is False, (
+        f"Content-Length {value!r} declares no body, so nothing is queued"
+    )
+
+
+@pytest.mark.parametrize("value", ["42, 42", "42,42", "042, 42"])
+def test_agreeing_nonzero_comma_list_is_readable_with_a_body(value):
+    """An agreeing nonzero list frames a real body: readable, and a body IS pending."""
+    h = _handler(value)
+    assert unreadable_content_length(h) is False, value
+    assert bool(request_declares_body(h)) is True, value
+
+
+@pytest.mark.parametrize("value", ["0, 42", "42, 0", "0, 0, 42"])
+def test_disagreeing_comma_list_stays_unreadable(value):
+    """Disagreeing members cannot frame a body — must stay refused, close armed."""
+    h = _handler(value)
+    assert unreadable_content_length(h) is True, (
+        f"Content-Length {value!r} disagrees; there is no honest length to drain, "
+        "so it must be refused with the connection armed for close"
+    )
+    assert bool(request_declares_body(h)) is True, value
+
+
+@pytest.mark.parametrize(
+    "value",
+    ["0, banana", "banana, 0", "0,", ",0", "0, ,0", "0, +0", "0, \xa00", "0, 0x5"],
+)
+def test_malformed_comma_members_stay_unreadable(value):
+    """One bad member poisons the whole field: fail closed, never treat as zero."""
+    h = _handler(value)
+    assert unreadable_content_length(h) is True, (
+        f"Content-Length {value!r} has an unparseable member and must fail closed"
+    )
+    assert bool(request_declares_body(h)) is True, value
+
+
+def test_repeated_headers_and_comma_list_agree():
+    """The two RFC spellings of the same field must behave identically."""
+    repeated = _H(_Hdrs({"Content-Length": ["0", "0"]}))
+    combined = _handler("0, 0")
+    assert unreadable_content_length(repeated) == unreadable_content_length(combined)
+    assert bool(request_declares_body(repeated)) == bool(request_declares_body(combined))
+
+    repeated_bad = _H(_Hdrs({"Content-Length": ["0", "42"]}))
+    combined_bad = _handler("0, 42")
+    assert unreadable_content_length(repeated_bad) is True
+    assert unreadable_content_length(combined_bad) is True
diff --git a/tests/test_get_with_body_auth_reject_reuse.py b/tests/test_get_with_body_auth_reject_reuse.py
new file mode 100644
index 00000000..8caac837
--- /dev/null
+++ b/tests/test_get_with_body_auth_reject_reuse.py
@@ -0,0 +1,158 @@
+"""A body-bearing GET that fails auth must not poison a keep-alive connection.
+
+The sibling files cover the WRITE paths: a rejected POST/PUT/PATCH/DELETE arms the
+close through ``check_auth_or_close``. ``do_GET`` was the documented residual — it
+called plain ``check_auth``, so a GET that *declares a body* and then fails auth
+answered its 401/302 and left those bytes queued in ``rfile``. On a keep-alive
+socket the next request is then parsed starting mid-body, and the client gets an
+error about a request it never sent.
+
+A GET with a body is unusual but entirely legal, and it is what a proxy, a
+``fetch`` with a body, or a scripted client can produce.
+
+Both halves of the contract are asserted here, because the naive fix (close every
+auth-failed GET) would regress the body-less login-redirect flow:
+
+* body-bearing + auth-failed -> connection MUST close (framing unresynchronisable)
+* body-less  + auth-failed -> connection MUST stay alive (nothing queued)
+
+``test_unfixed_handler_reproduces_the_poisoning`` is the negative control: the same
+socket exchange against a handler that answers without arming the close shows the
+leftover body surfacing as a bogus ``400 Bad request syntax``, proving the exchange
+really does exercise the bug rather than passing vacuously.
+"""
+
+import socket
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+BODY = b'{"q": "search"}'
+
+
+def _get_with_body(path=b"/api/sessions"):
+    return (
+        b"GET " + path + b" HTTP/1.1\r\n"
+        b"Host: localhost\r\n"
+        b"Content-Type: application/json\r\n"
+        b"Content-Length: " + str(len(BODY)).encode() + b"\r\n"
+        b"\r\n" + BODY
+    )
+
+
+BODYLESS_GET = (
+    b"GET /api/sessions HTTP/1.1\r\n"
+    b"Host: localhost\r\n"
+    b"\r\n"
+)
+FOLLOWING_GET = b"GET /api/health HTTP/1.1\r\nHost: localhost\r\n\r\n"
+
+
+def _declares_body(handler):
+    """The production rule: framing declares a body, so it is queued in rfile."""
+    raw = handler.headers.get("Content-Length")
+    if raw is not None:
+        stripped = raw.strip(" \t")
+        if not (stripped.isascii() and stripped.isdigit()):
+            return True  # unreadable framing -> fail closed
+        try:
+            if int(stripped) > 0:
+                return True
+        except ValueError:
+            return True
+    return bool((handler.headers.get("Transfer-Encoding") or "").strip())
+
+
+class _AuthRejectingHandler(BaseHTTPRequestHandler):
+    """Mirrors do_GET's shape: reject before the body is ever read."""
+
+    protocol_version = "HTTP/1.1"
+    arm_close = True
+
+    def do_GET(self):
+        if self.path == "/api/health":
+            payload = b'{"status": "ok"}'
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(payload)))
+            self.end_headers()
+            self.wfile.write(payload)
+            return
+        # Auth failure BEFORE the body is read — the residual's exact shape.
+        if self.arm_close and _declares_body(self):
+            self.close_connection = True
+        payload = b'{"error": "unauthorized"}'
+        self.send_response(401)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(payload)))
+        if self.arm_close and _declares_body(self):
+            self.send_header("Connection", "close")
+        self.end_headers()
+        self.wfile.write(payload)
+
+    def log_message(self, *a):
+        pass
+
+
+class _UnfixedHandler(_AuthRejectingHandler):
+    arm_close = False
+
+
+def _exchange(handler_cls, request_bytes):
+    """Pipeline `request_bytes` + a follow-up GET down ONE socket; return the reply."""
+    srv = ThreadingHTTPServer(("127.0.0.1", 0), handler_cls)
+    threading.Thread(target=srv.serve_forever, daemon=True).start()
+    try:
+        with socket.create_connection(srv.server_address, timeout=5) as s:
+            s.sendall(request_bytes + FOLLOWING_GET)
+            s.settimeout(5)
+            chunks = []
+            while True:
+                try:
+                    b = s.recv(4096)
+                except socket.timeout:
+                    break
+                if not b:
+                    break
+                chunks.append(b)
+            return b"".join(chunks)
+    finally:
+        srv.shutdown()
+        srv.server_close()
+
+
+def test_body_bearing_get_that_fails_auth_closes_the_connection():
+    raw = _exchange(_AuthRejectingHandler, _get_with_body())
+    assert b"401" in raw, raw[:200]
+    assert b"Connection: close" in raw, f"401 did not advertise close: {raw[:200]}"
+    # The follow-up GET must NEVER be answered on this socket.
+    assert raw.count(b"HTTP/1.1") == 1, f"connection was reused: {raw[:300]}"
+    assert b'"status": "ok"' not in raw, f"follow-up GET was served: {raw[:300]}"
+    # And the leftover body must never surface as a bogus request line.
+    assert b"Bad request syntax" not in raw, raw[:300]
+    assert b"501" not in raw, raw[:300]
+
+
+def test_bodyless_get_that_fails_auth_keeps_keep_alive():
+    """The over-close half: nothing is queued, so the socket must stay usable."""
+    raw = _exchange(_AuthRejectingHandler, BODYLESS_GET)
+    assert b"401" in raw, raw[:200]
+    assert b"Connection: close" not in raw, f"over-closed a bodyless reject: {raw[:200]}"
+    # The pipelined follow-up IS served, proving keep-alive survived.
+    assert b'"status": "ok"' in raw, f"keep-alive was dropped: {raw[:300]}"
+    assert raw.count(b"HTTP/1.1") == 2, raw[:300]
+
+
+def test_unfixed_handler_reproduces_the_poisoning():
+    """Negative control: without the arming, the leftover body corrupts the stream."""
+    raw = _exchange(_UnfixedHandler, _get_with_body())
+    assert b"401" in raw, raw[:200]
+    poisoned = (
+        b"Bad request syntax" in raw
+        or b"400" in raw
+        or b"501" in raw
+        or b'"status": "ok"' not in raw
+    )
+    assert poisoned, (
+        "expected the unread body to poison the next request on this socket; "
+        f"got {raw[:300]!r}"
+    )
diff --git a/tests/test_issue1909_csp_report_only.py b/tests/test_issue1909_csp_report_only.py
index a22be893..7ef947be 100644
--- a/tests/test_issue1909_csp_report_only.py
+++ b/tests/test_issue1909_csp_report_only.py
@@ -144,7 +144,7 @@ def test_server_bypasses_auth_for_csp_report(monkeypatch):
         called["path"] = parsed.path
         return True
 
-    monkeypatch.setattr("server.check_auth", fail_auth)
+    monkeypatch.setattr("server.check_auth_or_close", fail_auth)
     monkeypatch.setattr("server.clear_request_profile", lambda: None)
     monkeypatch.setattr("server.get_profile_cookie", lambda _handler: None)
 
diff --git a/tests/test_production_do_get_body_pending.py b/tests/test_production_do_get_body_pending.py
new file mode 100644
index 00000000..32929488
--- /dev/null
+++ b/tests/test_production_do_get_body_pending.py
@@ -0,0 +1,127 @@
+"""``Handler.do_GET`` itself must arm the close when a body is still queued.
+
+This is the production-wired companion to
+``test_get_with_body_auth_reject_reuse.py``. That file proves the HTTP-level
+consequence using a handler of the same shape; this one asserts the real
+``server.Handler.do_GET`` routes its auth rejection through the body-pending-aware
+path, so reverting ``server.py`` to plain ``check_auth`` fails a test.
+
+That distinction matters here: the PR's existing GET unit tests all pass with the
+residual reverted, because none of them reach ``do_GET``'s auth call.
+"""
+
+import types
+
+import pytest
+
+
+def _handler_double(headers, path="/api/sessions"):
+    """A handler stub shaped like the one the reject paths receive."""
+    h = types.SimpleNamespace()
+    h.path = path
+    h.headers = headers
+    h.close_connection = False
+    h.command = "GET"
+    h._headers_buffer = []
+    h.responses = []
+    h.sent = []
+    return h
+
+
+class _Hdrs(dict):
+    def get(self, k, default=None):
+        for key, val in self.items():
+            if key.lower() == k.lower():
+                return val[0] if isinstance(val, list) else val
+        return default
+
+    def get_all(self, k, default=None):
+        out = []
+        for key, val in self.items():
+            if key.lower() == k.lower():
+                out.extend(val if isinstance(val, list) else [val])
+        return out or default
+
+
+def test_do_get_source_uses_the_body_pending_aware_auth_call():
+    """``do_GET`` must not call bare ``check_auth``.
+
+    A source assertion is the right tool for this one: the call is a single line
+    inside a method whose surrounding machinery (socket, wfile, routing) would
+    have to be faked wholesale to reach it, and faking that is what let the
+    residual hide in the first place.
+    """
+    import pathlib
+
+    src = pathlib.Path(__file__).resolve().parents[1] / "server.py"
+    text = src.read_text()
+
+    start = text.index("def do_GET")
+    end = text.index("def ", start + 10)
+    body = text[start:end]
+
+    assert "check_auth_or_close(self, parsed)" in body, (
+        "do_GET must route its auth rejection through check_auth_or_close so a "
+        "body-bearing GET cannot leave unread bytes on a keep-alive socket"
+    )
+    assert "check_auth(self, parsed)" not in body, (
+        "do_GET still calls bare check_auth — a body-bearing GET that fails auth "
+        "will poison the next request on the connection"
+    )
+
+
+def test_check_auth_or_close_arms_only_when_a_body_is_pending():
+    """The over-close half, at the production helper."""
+    from api.auth import check_auth_or_close
+    import api.auth as auth_mod
+
+    calls = {}
+
+    def _deny(handler, parsed):
+        calls["hit"] = True
+        return False
+
+    real = auth_mod.check_auth
+    auth_mod.check_auth = _deny
+    try:
+        # Body pending -> close armed.
+        h = _handler_double(_Hdrs({"Content-Length": "15"}))
+        assert check_auth_or_close(h, None) is False
+        assert h.close_connection is True, "body-bearing reject must arm the close"
+
+        # Body-less -> keep-alive preserved.
+        h2 = _handler_double(_Hdrs({}))
+        assert check_auth_or_close(h2, None) is False
+        assert h2.close_connection is False, (
+            "a body-less reject must NOT close — that drops a healthy keep-alive"
+        )
+
+        # Unreadable framing -> fail closed.
+        h3 = _handler_double(_Hdrs({"Content-Length": "+0"}))
+        assert check_auth_or_close(h3, None) is False
+        assert h3.close_connection is True, (
+            "unreadable Content-Length must fail closed, not be treated as zero"
+        )
+    finally:
+        auth_mod.check_auth = real
+
+
+@pytest.mark.parametrize(
+    "value",
+    ["\xa00", "0\x85", "+0", "-0", "0_0", "\u0660", "-5", ""],
+)
+def test_adversarial_zero_spellings_declare_a_body(value):
+    """Every spelling of "zero" that ``int()`` accepts must still fail closed.
+
+    ``int()`` is far more generous than RFC 9110 ``Content-Length = 1*DIGIT``:
+    it tolerates unicode whitespace, a leading sign, PEP-515 underscores and
+    non-ASCII digits. Each of those parsed to an honest ``0`` would take the
+    keep-alive branch with the body still queued.
+    """
+    from api.helpers import request_declares_body
+
+    h = _handler_double(_Hdrs({"Content-Length": value}))
+    assert request_declares_body(h) is True, (
+        f"Content-Length {value!r} was treated as no-body; a queued body would "
+        "then poison the next request on this connection"
+    )
diff --git a/tests/test_rejected_write_connection_close.py b/tests/test_rejected_write_connection_close.py
new file mode 100644
index 00000000..e45fd46a
--- /dev/null
+++ b/tests/test_rejected_write_connection_close.py
@@ -0,0 +1,1294 @@
+"""Regression tests for HTTP/1.1 write rejection framing.
+
+A rejected POST can happen before its request body is consumed (authentication
+or CSRF failure). Keeping that connection alive leaves the body bytes in
+``rfile``; BaseHTTPRequestHandler then parses them as the next request method.
+The connection must close after those early rejections.
+"""
+from types import SimpleNamespace
+from typing import Any, cast
+
+import pytest
+
+
+def _rejected_write(monkeypatch, headers):
+    """Drive the production _handle_write through an auth rejection."""
+    import api.auth as auth
+    import server
+
+    handler = SimpleNamespace(
+        path="/api/session/new",
+        command="POST",
+        headers=headers,
+        close_connection=False,
+    )
+    monkeypatch.setattr(server, "reset_trusted_auth_request_state", lambda _handler: None)
+    monkeypatch.setattr(server, "get_profile_cookie", lambda _handler: None)
+    monkeypatch.setattr(server, "clear_request_profile", lambda: None)
+    monkeypatch.setattr(auth, "check_auth", lambda _handler, _parsed: False)
+
+    def route_must_not_run(_handler, _parsed):
+        raise AssertionError("route ran after auth rejection")
+
+    server.Handler._handle_write(cast(Any, handler), route_must_not_run)
+    return handler
+
+
+@pytest.mark.parametrize(
+    "headers",
+    [{"Content-Length": "15"}, {"Transfer-Encoding": "chunked"}, {"Content-Length": "banana"}],
+    ids=["content-length", "chunked", "unreadable-length"],
+)
+def test_auth_rejected_write_closes_connection(headers, monkeypatch):
+    """A rejected write whose framing declares a body leaves it unread — close."""
+    assert _rejected_write(monkeypatch, headers).close_connection is True
+
+
+@pytest.mark.parametrize(
+    "headers", [{}, {"Content-Length": "0"}], ids=["no-content-length", "zero-content-length"]
+)
+def test_bodyless_auth_rejected_write_keeps_connection(headers, monkeypatch):
+    """The over-close half at the auth gate: no declared body, nothing to protect.
+
+    `check_auth_or_close()` armed unconditionally, so a body-less POST that failed
+    auth answered 401 WITH `Connection: close` and dropped the client's pipelined
+    follow-up (reproduced on the wire — see the pooled-client file). Same rule as
+    the sidecar path: the framing decides, not the method or the site.
+    """
+    assert _rejected_write(monkeypatch, headers).close_connection is False
+
+
+def _csrf_origin_rejection(monkeypatch, headers):
+    import api.routes as routes
+
+    handler = SimpleNamespace(headers=headers, close_connection=False)
+    monkeypatch.setattr(routes, "_check_same_origin_browser_request", lambda _handler: False)
+
+    assert routes._check_csrf(handler) is False
+    return handler
+
+
+@pytest.mark.parametrize(
+    "headers",
+    [{"Content-Length": "15"}, {"Transfer-Encoding": "chunked"}, {"Content-Length": "banana"}],
+    ids=["content-length", "chunked", "unreadable-length"],
+)
+def test_origin_rejected_csrf_closes_connection(headers, monkeypatch):
+    assert _csrf_origin_rejection(monkeypatch, headers).close_connection is True
+
+
+@pytest.mark.parametrize(
+    "headers", [{}, {"Content-Length": "0"}], ids=["no-content-length", "zero-content-length"]
+)
+def test_bodyless_origin_rejected_csrf_keeps_connection(headers, monkeypatch):
+    """A cross-origin rejection of a body-less write must not drop the socket."""
+    assert _csrf_origin_rejection(monkeypatch, headers).close_connection is False
+
+
+def _csrf_token_rejection(monkeypatch, headers):
+    import api.auth as auth
+    import api.routes as routes
+
+    handler = SimpleNamespace(headers=headers, close_connection=False)
+    monkeypatch.setattr(routes, "_check_same_origin_browser_request", lambda _handler: True)
+    monkeypatch.setattr(routes, "_is_browser_unsafe_request", lambda _handler: True)
+    monkeypatch.setattr(auth, "is_auth_enabled", lambda: True)
+    monkeypatch.setattr(auth, "parse_cookie", lambda _handler: "session-cookie")
+    monkeypatch.setattr(auth, "verify_csrf_token", lambda _cookie, _token: False)
+
+    assert routes._check_csrf(handler) is False
+    return handler
+
+
+@pytest.mark.parametrize(
+    "headers",
+    [{"Content-Length": "15"}, {"Transfer-Encoding": "chunked"}, {"Content-Length": "banana"}],
+    ids=["content-length", "chunked", "unreadable-length"],
+)
+def test_token_rejected_csrf_closes_connection(headers, monkeypatch):
+    assert _csrf_token_rejection(monkeypatch, headers).close_connection is True
+
+
+@pytest.mark.parametrize(
+    "headers", [{}, {"Content-Length": "0"}], ids=["no-content-length", "zero-content-length"]
+)
+def test_bodyless_token_rejected_csrf_keeps_connection(headers, monkeypatch):
+    """A token mismatch on a body-less write has nothing unread either."""
+    assert _csrf_token_rejection(monkeypatch, headers).close_connection is False
+
+
+def test_sidecar_provenance_rejected_closes_connection(monkeypatch):
+    """Sidecar proxy provenance rejection also runs before read_body()."""
+    import api.routes as routes
+
+    handler = SimpleNamespace(
+        path="/api/extensions/ext1/sidecar/proxy",
+        command="POST",
+        headers={"Content-Length": "15"},
+        close_connection=False,
+    )
+    monkeypatch.setattr(routes, "_match_extension_sidecar_proxy_path", lambda _path: ("ext1", "/proxy"))
+    monkeypatch.setattr(routes, "_check_same_origin_browser_request", lambda _handler, **_: False)
+    monkeypatch.setattr(routes, "j", lambda _handler, _payload, status=200: None)
+
+    routes._handle_extension_sidecar_proxy(
+        handler, SimpleNamespace(path=handler.path, query=""), "POST", read_request_body=True
+    )
+
+    assert handler.close_connection is True
+
+
+def test_sidecar_get_provenance_rejected_keeps_connection(monkeypatch):
+    """A body-less GET rejection has nothing unread — keep-alive must survive.
+
+    Closing here would kill a pooled client's connection for no framing
+    reason (the deep-review gate's MUST-FIX 2).
+    """
+    import api.routes as routes
+
+    handler = SimpleNamespace(
+        path="/api/extensions/ext1/sidecar/proxy",
+        command="GET",
+        headers={},
+        close_connection=False,
+    )
+    monkeypatch.setattr(routes, "_match_extension_sidecar_proxy_path", lambda _path: ("ext1", "/proxy"))
+    monkeypatch.setattr(routes, "_check_same_origin_browser_request", lambda _handler, **_: False)
+    monkeypatch.setattr(routes, "j", lambda _handler, _payload, status=200: None)
+
+    routes._handle_extension_sidecar_proxy(handler, SimpleNamespace(path=handler.path, query=""), "GET")
+
+    assert handler.close_connection is False
+
+
+# ── Framing, not the method, decides whether a rejection must close ──────────
+#
+# The re-gate found both halves of the same mistake: `read_request_body` is a
+# static per-method flag, so it neither proves a body is pending (a body-less
+# DELETE closed a healthy connection) nor proves one is absent (a GET carrying a
+# declared body kept the connection and poisoned it). Both directions below run
+# against the production handler.
+
+
+def _rejected_sidecar_call(monkeypatch, method, headers, **kwargs):
+    """Drive a provenance-rejected sidecar proxy call; return the handler."""
+    import api.routes as routes
+
+    handler = SimpleNamespace(
+        path="/api/extensions/ext1/sidecar/proxy",
+        command=method,
+        headers=headers,
+        close_connection=False,
+    )
+    monkeypatch.setattr(routes, "_match_extension_sidecar_proxy_path", lambda _path: ("ext1", "/proxy"))
+    monkeypatch.setattr(routes, "_check_same_origin_browser_request", lambda _handler, **_: False)
+    monkeypatch.setattr(routes, "j", lambda _handler, _payload, status=200: None)
+
+    routes._handle_extension_sidecar_proxy(
+        cast(Any, handler), SimpleNamespace(path=handler.path, query=""), method, **kwargs
+    )
+    return handler
+
+
+@pytest.mark.parametrize(
+    "headers",
+    [
+        {"Content-Length": "12"},
+        {"Transfer-Encoding": "chunked"},
+        {"Content-Length": "banana"},  # unreadable framing proves nothing is absent
+    ],
+    ids=["content-length", "chunked", "garbage-content-length"],
+)
+def test_sidecar_get_with_a_declared_body_closes_connection(headers, monkeypatch):
+    """A GET may still declare a body — those bytes are unread, so close.
+
+    Reproduced by the gate as a 403 WITHOUT `Connection: close`, followed by
+    `501 Unsupported method ('{}GET')` on the same socket.
+    """
+    handler = _rejected_sidecar_call(monkeypatch, "GET", headers)
+
+    assert handler.close_connection is True
+
+
+@pytest.mark.parametrize("method", ["POST", "PUT", "PATCH", "DELETE"])
+@pytest.mark.parametrize(
+    "headers", [{}, {"Content-Length": "0"}], ids=["no-content-length", "zero-content-length"]
+)
+def test_bodyless_write_method_rejection_keeps_connection(method, headers, monkeypatch):
+    """A write method with no declared body has nothing unread — keep-alive.
+
+    `read_request_body=True` is passed for every one of these (that is what the
+    real unsafe-method call site does), so this is exactly the case the old
+    per-method gate got wrong in the over-close direction.
+    """
+    handler = _rejected_sidecar_call(monkeypatch, method, headers, read_request_body=True)
+
+    assert handler.close_connection is False
+
+
+@pytest.mark.parametrize(
+    ("headers", "expected"),
+    [
+        ({}, False),
+        ({"Content-Length": "0"}, False),
+        ({"Content-Length": "00"}, False),  # 1*DIGIT, so an honest zero
+        ({"Content-Length": " 0 "}, False),
+        ({"Content-Length": "0\t"}, False),
+        ({"Content-Length": ""}, True),
+        ({"Content-Length": "   "}, True),
+        ({"Content-Length": "\t"}, True),
+        ({"Content-Length": "12"}, True),
+        ({"Content-Length": "-1"}, True),
+        ({"Content-Length": "banana"}, True),
+        ({"Content-Length": "0x5"}, True),
+        ({"Content-Length": "+5"}, True),
+        # int() is looser than 1*DIGIT, and every one of these reads as ZERO --
+        # where the wrong parse is not merely wrong but silently keep-alive.
+        ({"Content-Length": "+0"}, True),
+        ({"Content-Length": "-0"}, True),
+        ({"Content-Length": "0_0"}, True),
+        ({"Content-Length": "+00"}, True),
+        ({"Content-Length": "0_0_0"}, True),
+        ({"Content-Length": "٠0"}, True),  # ARABIC-INDIC ZERO ahead of an ASCII one
+        ({"Content-Length": "٠"}, True),  # ARABIC-INDIC DIGIT ZERO: int() -> 0
+        ({"Content-Length": "０"}, True),  # FULLWIDTH DIGIT ZERO: int() -> 0
+        ({"Content-Length": "٥"}, True),  # ARABIC-INDIC DIGIT FIVE: int() -> 5
+        ({"Content-Length": "\xa00"}, True),  # NBSP-padded zero: int() strips it
+        ({"Content-Length": "0\x85"}, True),  # NEL-padded zero: likewise
+        ({"content-length": "7"}, True),
+        ({"Transfer-Encoding": "chunked"}, True),
+        ({"transfer-encoding": "Chunked"}, True),
+        # No coding is exempt: http.server decodes none of them (RFC 9112 6.3).
+        ({"Transfer-Encoding": "identity"}, True),
+        ({"Transfer-Encoding": "IDENTITY "}, True),
+        ({"Transfer-Encoding": "Identity"}, True),
+        ({"Transfer-Encoding": "\tidentity\t"}, True),
+        ({"Transfer-Encoding": "identity, chunked"}, True),
+        ({"Transfer-Encoding": "chunked, identity"}, True),
+        ({"Transfer-Encoding": "identity,identity"}, True),
+        ({"Transfer-Encoding": ","}, True),
+        ({"Transfer-Encoding": "banana"}, True),
+        ({"Transfer-Encoding": ""}, True),
+        ({"Transfer-Encoding": "  "}, True),
+        ({"Transfer-Encoding": "\xa0identity"}, True),  # not the token `identity`
+        # A coding beside a length: the coding decides, whatever the length says.
+        ({"Transfer-Encoding": "identity", "Content-Length": "0"}, True),
+        ({"Transfer-Encoding": "identity", "Content-Length": "15"}, True),
+    ],
+)
+def test_request_declares_body_reads_the_framing_headers(headers, expected):
+    from api.helpers import request_declares_body
+
+    handler = SimpleNamespace(headers=headers)
+
+    assert request_declares_body(cast(Any, handler)) is expected
+
+
+def test_request_declares_body_is_case_insensitive_on_a_real_message():
+    """Production handlers carry an email.message.Message, not a dict."""
+    from email.parser import Parser
+
+    from api.helpers import request_declares_body
+
+    message = Parser().parsestr("Host: x\r\ntransfer-encoding: Chunked\r\n\r\n")
+
+    assert request_declares_body(cast(Any, SimpleNamespace(headers=message))) is True
+
+
+def test_request_declares_body_tolerates_a_handler_without_headers():
+    """Never raise on a rejection path: an exception there becomes a 500."""
+    from api.helpers import request_declares_body
+
+    assert request_declares_body(cast(Any, SimpleNamespace())) is False
+    assert request_declares_body(cast(Any, SimpleNamespace(headers=None))) is False
+    assert request_declares_body(cast(Any, SimpleNamespace(headers=object()))) is False
+
+
+# ── A duplicated framing header hides the body from a first-value-only read ────
+#
+# `Message.get()` returns only the FIRST occurrence, so `Content-Length: 0`
+# ahead of the real length reads back as an empty body: nothing is drained and
+# the real bytes are parsed as the next request line. Verified live against the
+# running server before the fix -- 403 then
+# `400 Bad request syntax ('{"stale": true}GET /api/health/agent HTTP/1.1')`,
+# and on /api/upload a 400 then `400 Bad request syntax ('--x')`.
+
+
+def _message_with(raw_headers: str):
+    """Build the email.message.Message a production handler really carries."""
+    from email.parser import Parser
+
+    return SimpleNamespace(headers=Parser().parsestr(raw_headers + "\r\n"))
+
+
+def test_duplicate_conflicting_content_length_declares_a_body():
+    from api.helpers import request_declares_body
+
+    handler = _message_with("Host: x\r\nContent-Length: 0\r\nContent-Length: 15\r\n")
+
+    assert request_declares_body(cast(Any, handler)) is True
+
+
+def test_duplicate_agreeing_zero_content_length_declares_no_body():
+    """Repetition alone is not a conflict -- two honest zeroes still frame nothing."""
+    from api.helpers import request_declares_body
+
+    handler = _message_with("Host: x\r\nContent-Length: 0\r\nContent-Length: 0\r\n")
+
+    assert request_declares_body(cast(Any, handler)) is False
+
+
+@pytest.mark.parametrize(
+    "raw",
+    [
+        "Transfer-Encoding: identity\r\nTransfer-Encoding: chunked\r\n",
+        "Transfer-Encoding: chunked\r\nTransfer-Encoding: identity\r\n",
+        "Transfer-Encoding: identity\r\nTransfer-Encoding: identity\r\n",
+        "Transfer-Encoding: identity\r\ntransfer-encoding: identity\r\n",
+    ],
+    ids=["identity-then-chunked", "chunked-then-identity", "twice", "twice-mixed-case"],
+)
+def test_repeated_transfer_encoding_refuses_whatever_the_order(raw):
+    """A duplicated header cannot hide an undecodable coding behind an "unframed" one.
+
+    This used to have to look PAST a leading `identity` to find the chunked coding
+    (`Message.get()` returns only the first value, so it read as unframed and the
+    chunk bytes stayed queued). With no coding exempt the leading value already
+    refuses, so no ordering can mask anything -- which is what is pinned here,
+    rather than the specific coding a first-value read happened to miss.
+    """
+    from api.helpers import request_declares_body, unsupported_transfer_encoding
+
+    handler = _message_with("Host: x\r\n" + raw)
+
+    assert unsupported_transfer_encoding(cast(Any, handler)) is not None
+    assert request_declares_body(cast(Any, handler)) is True
+
+
+def test_unreadable_content_length_flags_conflict_and_garbage_only():
+    from api.helpers import unreadable_content_length
+
+    assert unreadable_content_length(cast(Any, _message_with("Content-Length: 5\r\n"))) is False
+    assert unreadable_content_length(cast(Any, _message_with("Host: x\r\n"))) is False
+    assert unreadable_content_length(
+        cast(Any, _message_with("Content-Length: 5\r\nContent-Length: 5\r\n"))
+    ) is False
+    assert unreadable_content_length(
+        cast(Any, _message_with("Content-Length: 0\r\nContent-Length: 9\r\n"))
+    ) is True
+    assert unreadable_content_length(cast(Any, _message_with("Content-Length: nope\r\n"))) is True
+
+
+def test_sidecar_get_with_duplicate_content_length_closes_connection(monkeypatch):
+    """The conflicting-framing GET reaches the production provenance rejection."""
+    import api.routes as routes
+
+    from email.parser import Parser
+
+    headers = Parser().parsestr("Host: x\r\nContent-Length: 0\r\nContent-Length: 15\r\n\r\n")
+    handler = SimpleNamespace(
+        path="/api/extensions/ext1/sidecar/proxy",
+        command="GET",
+        headers=headers,
+        close_connection=False,
+    )
+    monkeypatch.setattr(routes, "_match_extension_sidecar_proxy_path", lambda _path: ("ext1", "/proxy"))
+    monkeypatch.setattr(routes, "_check_same_origin_browser_request", lambda _handler, **_: False)
+    monkeypatch.setattr(routes, "j", lambda _handler, _payload, status=200: None)
+
+    routes._handle_extension_sidecar_proxy(
+        cast(Any, handler), SimpleNamespace(path=handler.path, query=""), "GET"
+    )
+
+    assert handler.close_connection is True
+
+
+# ── A BLANK framing value is unreadable, not absent ───────────────────────────
+#
+# `Content-Length:` with nothing after the colon reads back as '' from
+# `get_all()`, and the empty value was SKIPPED — so `_declared_content_lengths()`
+# reported "no length declared", `request_declares_body()` said no body was
+# pending, and the provenance rejection answered 403 with keep-alive intact.
+# Reproduced live against the running server on the otherwise-fixed head:
+#
+#   GET /api/extensions/probe/sidecar/ping  `Content-Length:`  {"stale": true}
+#     -> HTTP/1.1 403 Forbidden        (no Connection: close)
+#     -> HTTP/1.1 400 Bad request syntax
+#            ('{"stale": true}GET /api/health/agent HTTP/1.1')
+#
+# `Message` collapses `Content-Length:`, `Content-Length:   ` and
+# `Content-Length:\t` to the same '', so every blank spelling poisoned the socket
+# identically — as did a blank value duplicated, or paired with a real length.
+#
+# A blank `Transfer-Encoding:` was the same hole one header over: '' reached the
+# allowlist of codings that "framed nothing" (since deleted, along with the
+# `identity` entry that was its last member), so a header naming no coding at all
+# read as "no framing to worry about" and its payload poisoned the socket the same
+# way (verified live: 403 then the same bad-syntax 400). Absence is something only
+# the wire can say, and an absent header yields no values at all.
+
+
+_BLANK_FRAMING_HEADERS = [
+    "Content-Length:",
+    "Content-Length:   ",
+    "Content-Length:\t",
+    "Content-Length:\r\nContent-Length:",  # duplicated blank
+    "Content-Length:\r\nContent-Length: 15",  # blank ahead of a real length
+    "Content-Length: 15\r\nContent-Length:",  # and behind one
+    "Content-Length: 0\r\nContent-Length:",  # blank behind an honest zero
+    "Transfer-Encoding:",
+    "Transfer-Encoding:   ",
+    "Transfer-Encoding:\r\nTransfer-Encoding: chunked",
+]
+
+
+@pytest.mark.parametrize("raw", _BLANK_FRAMING_HEADERS)
+def test_blank_framing_value_declares_a_body(raw):
+    """On a real email.message.Message, as the production handler carries."""
+    from api.helpers import request_declares_body
+
+    handler = _message_with("Host: x\r\n" + raw + "\r\n")
+
+    assert request_declares_body(cast(Any, handler)) is True
+
+
+@pytest.mark.parametrize(
+    "raw",
+    [
+        "Content-Length:",
+        "Content-Length:   ",
+        "Content-Length:\r\nContent-Length:",
+        "Content-Length:\r\nContent-Length: 15",
+        "Content-Length: 0\r\nContent-Length:",
+    ],
+)
+def test_blank_content_length_is_unreadable(raw):
+    """The upload preflight gate must refuse it too — there is no length to drain."""
+    from api.helpers import unreadable_content_length
+
+    assert unreadable_content_length(cast(Any, _message_with("Host: x\r\n" + raw + "\r\n"))) is True
+
+
+def test_blank_transfer_encoding_is_reported_as_an_undecodable_coding():
+    """Named, non-empty and never None, so `is not None` callers and the 411 read right."""
+    from api.helpers import unsupported_transfer_encoding
+
+    coding = unsupported_transfer_encoding(cast(Any, _message_with("Host: x\r\nTransfer-Encoding:\r\n")))
+
+    assert coding is not None
+    assert coding, "an empty return value would be a falsy non-None trap for callers"
+    # Only an ABSENT header frames nothing; a present one always names a coding
+    # this server cannot decode, `identity` included.
+    assert unsupported_transfer_encoding(cast(Any, _message_with("Host: x\r\n"))) is None
+    assert unsupported_transfer_encoding(
+        cast(Any, _message_with("Host: x\r\nTransfer-Encoding: identity\r\n"))
+    ) == "identity"
+
+
+@pytest.mark.parametrize("raw", _BLANK_FRAMING_HEADERS)
+def test_sidecar_get_with_blank_framing_value_closes_connection(raw, monkeypatch):
+    """The reported case, through the production provenance rejection."""
+    import api.routes as routes
+
+    from email.parser import Parser
+
+    handler = SimpleNamespace(
+        path="/api/extensions/ext1/sidecar/proxy",
+        command="GET",
+        headers=Parser().parsestr("Host: x\r\n" + raw + "\r\n\r\n"),
+        close_connection=False,
+    )
+    monkeypatch.setattr(routes, "_match_extension_sidecar_proxy_path", lambda _path: ("ext1", "/proxy"))
+    monkeypatch.setattr(routes, "_check_same_origin_browser_request", lambda _handler, **_: False)
+    monkeypatch.setattr(routes, "j", lambda _handler, _payload, status=200: None)
+
+    routes._handle_extension_sidecar_proxy(
+        cast(Any, handler), SimpleNamespace(path=handler.path, query=""), "GET"
+    )
+
+    assert handler.close_connection is True
+
+
+@pytest.mark.parametrize(
+    "raw",
+    [
+        "",
+        "Content-Length: 0\r\n",
+        "Content-Length:  0 \r\n",
+        "Content-Length: 0\r\nContent-Length: 0\r\n",
+    ],
+)
+def test_blank_rule_does_not_over_close_a_bodyless_request(raw):
+    """The other half of the contract: framing that says "no body" keeps keep-alive.
+
+    No framing header at all, an OWS-padded zero and two AGREEING zeroes are
+    honest statements that nothing is queued, and none of them may be swept up by
+    the blank rule.
+    """
+    from api.helpers import request_declares_body
+
+    assert request_declares_body(cast(Any, _message_with("Host: x\r\n" + raw))) is False
+
+
+# ── A length that only `int()` calls zero is unreadable, not zero ──────────────
+#
+# The blank rule above closed the "no value" hole; this one closes the "value
+# `int()` is too generous about" hole beside it. RFC 9110 is
+# `Content-Length = 1*DIGIT` -- an unsigned run of ASCII digits -- but `int()`
+# also accepts a leading sign, PEP 515 underscores and non-ASCII digits, and
+# strips non-ASCII whitespace. Every spelling of ZERO below therefore parsed to
+# an honest `0`, took the "every declared length agrees on zero" keep-alive
+# branch, and left the payload queued. Reproduced live on the otherwise-fixed
+# head, pipelined down one socket:
+#
+#   GET /api/extensions/probe/sidecar/ping  `Content-Length: +0`  {"stale": true}
+#     -> HTTP/1.1 403 Forbidden        (no Connection: close)
+#     -> HTTP/1.1 400 Bad request syntax
+#            ('{"stale": true}GET /api/health/agent HTTP/1.1')
+#
+# byte-for-byte the trace the blank value produced, and on /api/upload a 400
+# then `400 Bad request syntax ('--x')`.
+#
+# A sign is only harmless on a NON-zero value (`+5` parses to 5, declares a body
+# and closes anyway), which is exactly why the family looked safe: the mistake is
+# invisible until the value is zero.
+#
+# Non-ASCII digits are unreadable by the same rule. An HTTP/1.1 request line
+# decodes as latin-1, so U+0660 cannot cross the wire in a header value (only
+# U+00B2/U+00B3/U+00B9 -- superscripts, which `int()` already rejects -- and the
+# U+0085/U+00A0 whitespace `int()` strips can); these helpers take any headers
+# mapping, though, so the grammar is enforced here rather than relying on the
+# transport to filter.
+
+_MALFORMED_ZERO_LENGTHS = [
+    "+0",
+    "-0",
+    "0_0",
+    "+00",
+    "0_0_0",
+    "\xa00",  # NBSP-padded zero: str.strip()/int() erase U+00A0
+    "0\x85",  # NEL-padded zero: likewise U+0085
+    "٠",  # ARABIC-INDIC DIGIT ZERO -- isdigit() and int() both accept it
+    "０",  # FULLWIDTH DIGIT ZERO
+    "𝟢",  # MATHEMATICAL MONOSPACE DIGIT ZERO
+]
+
+
+@pytest.mark.parametrize("value", _MALFORMED_ZERO_LENGTHS)
+def test_malformed_zero_content_length_declares_a_body(value):
+    """On a real email.message.Message, as the production handler carries."""
+    from api.helpers import request_declares_body
+
+    handler = _message_with(f"Host: x\r\nContent-Length: {value}\r\n")
+
+    assert request_declares_body(cast(Any, handler)) is True
+
+
+@pytest.mark.parametrize("value", _MALFORMED_ZERO_LENGTHS)
+def test_malformed_zero_content_length_is_unreadable(value):
+    """The upload preflight gate must refuse it too -- it is not a length."""
+    from api.helpers import unreadable_content_length
+
+    handler = _message_with(f"Host: x\r\nContent-Length: {value}\r\n")
+
+    assert unreadable_content_length(cast(Any, handler)) is True
+
+
+@pytest.mark.parametrize("value", _MALFORMED_ZERO_LENGTHS)
+def test_sidecar_get_with_a_malformed_zero_length_closes_connection(value, monkeypatch):
+    """Through the production provenance rejection, not just the helper."""
+    handler = _rejected_sidecar_call(
+        monkeypatch, "GET", _message_with(f"Host: x\r\nContent-Length: {value}\r\n").headers
+    )
+
+    assert handler.close_connection is True
+
+
+@pytest.mark.parametrize(
+    "value",
+    ["0", "00", "000", " 0 ", "0\t", "\t0", "0\r\nContent-Length: 0"],
+    ids=["zero", "double-zero", "triple-zero", "padded", "tab-behind", "tab-ahead", "two-agreeing"],
+)
+def test_digit_grammar_does_not_over_close_an_honest_zero(value):
+    """The over-close half: `1*DIGIT` zeroes, SP/HTAB padding and all, stay alive.
+
+    `00` is as valid as `0` (`1*DIGIT` sets no leading-zero rule), and RFC 9110
+    OWS around a field value is SP/HTAB -- so trimming those must not become
+    trimming everything `str.strip()` would.
+    """
+    from api.helpers import request_declares_body, unreadable_content_length
+
+    handler = _message_with(f"Host: x\r\nContent-Length: {value}\r\n")
+
+    assert request_declares_body(cast(Any, handler)) is False
+    assert unreadable_content_length(cast(Any, handler)) is False
+
+
+def test_absurdly_long_digit_run_is_unreadable_and_never_raises():
+    """`1*DIGIT` is necessary but not sufficient: `int()` caps the digit count.
+
+    A run longer than `sys.get_int_max_str_digits()` (4300) raises ValueError, and
+    this runs on rejection paths where an exception becomes a spurious 500 -- so
+    the parse stays guarded even though every character is now vetted first.
+    """
+    from api.helpers import request_declares_body, unreadable_content_length
+
+    handler = _message_with("Host: x\r\nContent-Length: " + "0" * 5000 + "\r\n")
+
+    assert request_declares_body(cast(Any, handler)) is True
+    assert unreadable_content_length(cast(Any, handler)) is True
+
+
+# ── The same audit one header over: NO transfer coding is decodable here ───────
+#
+# `identity` was exempt as a coding that "frames nothing", so `Transfer-Encoding:
+# identity` plus payload bytes read as body-less: the sidecar rejection answered
+# 403 with keep-alive and the payload became the next request line, and all four
+# upload handlers fell through to a length-0 read. RFC 9112 6.3 is explicit --
+# a request whose final transfer coding is not `chunked` has a length the
+# recipient cannot determine, so it must be refused and the connection closed --
+# and http.server decodes no coding at all, `chunked` included. Every spelling in
+# the neighbourhood is swept here because three prior rounds each fixed one value
+# of a framing rule and left an adjacent one poisoning the socket.
+#
+# The over-close half of this contract is pinned by the framing table above and by
+# `test_blank_rule_does_not_over_close_a_bodyless_request` /
+# `test_digit_grammar_does_not_over_close_an_honest_zero`: a request with NO
+# framing header, or one whose declared lengths all agree on zero, keeps its
+# keep-alive. An absent `Transfer-Encoding` is the only one that frames nothing.
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        "identity",
+        "IDENTITY",
+        "Identity",
+        "identity ",
+        "\tidentity\t",
+        "\xa0identity",
+        "identity\xa0",
+        "\xa0identity\xa0",
+        "\x85identity",
+        "identity, chunked",
+        "chunked, identity",
+        "identity,identity",
+        ",",
+        "banana",
+    ],
+    ids=[
+        "identity",
+        "upper",
+        "title",
+        "sp-padded",
+        "htab-padded",
+        "nbsp-ahead",
+        "nbsp-behind",
+        "nbsp-both",
+        "nel-ahead",
+        "identity-then-chunked",
+        "chunked-then-identity",
+        "identity-list",
+        "lone-comma",
+        "unknown-coding",
+    ],
+)
+def test_every_spelling_of_a_transfer_coding_declares_a_body(value):
+    """Case, OWS, non-OWS padding, comma lists, a bare comma, an unknown coding.
+
+    The ASCII `identity` spellings are the newly-closed hole (they normalized to
+    the exempt token); the U+00A0/U+0085-padded ones were the previous round's
+    (`str.strip()` erased the padding, so `\\xa0identity` read back as `identity`)
+    and they reach the wire because a request line decodes as latin-1. The comma
+    forms never matched the exempt token and already refused -- they are pinned so
+    a future normalizer that splits the list cannot reopen the hole.
+    """
+    from api.helpers import request_declares_body, unsupported_transfer_encoding
+
+    handler = _message_with(f"Host: x\r\nTransfer-Encoding: {value}\r\n")
+
+    assert unsupported_transfer_encoding(cast(Any, handler)) is not None
+    assert request_declares_body(cast(Any, handler)) is True
+
+
+@pytest.mark.parametrize("length", ["0", "15"], ids=["zero-length", "real-length"])
+def test_transfer_encoding_beside_a_content_length_still_declares_a_body(length):
+    """A coding plus a length is unframeable however honest the length looks.
+
+    `Transfer-Encoding: identity` with `Content-Length: 0` was the worst of the
+    pair: both halves read as "no body", so the rejection kept the socket and the
+    payload behind them was parsed as the next request line. RFC 9112 6.3 refuses
+    the combination outright (it is the request-smuggling shape), and the coding
+    has to win -- a length cannot describe a body the server cannot decode.
+    """
+    from api.helpers import request_declares_body
+
+    handler = _message_with(
+        f"Host: x\r\nTransfer-Encoding: identity\r\nContent-Length: {length}\r\n"
+    )
+
+    assert request_declares_body(cast(Any, handler)) is True
+
+
+def _rate_limited_csp_report(monkeypatch, headers):
+    import api.routes as routes
+
+    handler = SimpleNamespace(headers=headers, close_connection=False)
+    monkeypatch.setattr(routes, "_csp_report_rate_limited", lambda _handler: True)
+    monkeypatch.setattr(routes, "_send_no_content", lambda _handler: True)
+
+    assert routes._handle_csp_report(handler) is True
+    return handler
+
+
+@pytest.mark.parametrize(
+    "headers",
+    [{"Content-Length": "15"}, {"Transfer-Encoding": "chunked"}, {"Content-Length": "banana"}],
+    ids=["content-length", "chunked", "unreadable-length"],
+)
+def test_csp_report_rate_limited_closes_connection(headers, monkeypatch):
+    """A rate-limited CSP report is dropped before its body is read."""
+    assert _rate_limited_csp_report(monkeypatch, headers).close_connection is True
+
+
+@pytest.mark.parametrize(
+    "headers", [{}, {"Content-Length": "0"}], ids=["no-content-length", "zero-content-length"]
+)
+def test_bodyless_rate_limited_csp_report_keeps_connection(headers, monkeypatch):
+    """A body-less report dropped by the limiter must not cost the browser its socket."""
+    assert _rate_limited_csp_report(monkeypatch, headers).close_connection is False
+
+
+_RESTART_OUTCOMES = [
+    {"status": "completed"},
+    {"status": "in_progress"},
+    {"status": "busy"},
+    {"status": "error"},
+]
+
+
+@pytest.mark.parametrize("outcome", _RESTART_OUTCOMES, ids=lambda o: o["status"])
+@pytest.mark.parametrize(
+    "headers",
+    [{"Content-Length": "15"}, {"Transfer-Encoding": "chunked"}, {"Content-Length": "banana"}],
+    ids=["content-length", "chunked", "unreadable-length"],
+)
+def test_health_restart_closes_connection(headers, outcome, monkeypatch):
+    """health/restart never consumes its body on any outcome — so a declared one closes."""
+    import api.routes as routes
+
+    handler = SimpleNamespace(headers=headers, close_connection=False)
+    monkeypatch.setattr(routes, "restart_active_profile_gateway", lambda: outcome)
+    monkeypatch.setattr(routes, "j", lambda _handler, _payload, status=200: True)
+
+    routes._handle_health_restart(handler)
+    assert handler.close_connection is True
+
+
+@pytest.mark.parametrize("outcome", _RESTART_OUTCOMES, ids=lambda o: o["status"])
+@pytest.mark.parametrize(
+    "headers", [{}, {"Content-Length": "0"}], ids=["no-content-length", "zero-content-length"]
+)
+def test_bodyless_health_restart_keeps_connection(headers, outcome, monkeypatch):
+    """Including the SUCCESS path, which closed a healthy socket on every call.
+
+    The WebUI's restart button sends no body, so the old unconditional arming made
+    a 200 "restarted successfully" hang up the connection every single time. The
+    framing decides for all four outcomes, not the result.
+    """
+    import api.routes as routes
+
+    handler = SimpleNamespace(headers=headers, close_connection=False)
+    monkeypatch.setattr(routes, "restart_active_profile_gateway", lambda: outcome)
+    monkeypatch.setattr(routes, "j", lambda _handler, _payload, status=200: True)
+
+    routes._handle_health_restart(handler)
+    assert handler.close_connection is False
+
+
+def _deprecated_ack_post(monkeypatch, headers):
+    """Drive the deprecated /api/process-complete-ack 410, which runs pre-CSRF."""
+    import api.routes as routes
+
+    handler = SimpleNamespace(
+        path="/api/process-complete-ack",
+        command="POST",
+        headers=headers,
+        close_connection=False,
+    )
+    monkeypatch.setattr(
+        routes, "j", lambda _handler, _payload, status=200, extra_headers=None: True
+    )
+
+    assert routes.handle_post(
+        cast(Any, handler), SimpleNamespace(path=handler.path, query="")
+    ) is True
+    return handler
+
+
+@pytest.mark.parametrize(
+    "headers",
+    [{"Content-Length": "15"}, {"Transfer-Encoding": "chunked"}, {"Content-Length": "banana"}],
+    ids=["content-length", "chunked", "unreadable-length"],
+)
+def test_deprecated_ack_with_a_body_closes_connection(headers, monkeypatch):
+    """A stale tab's JSON ack is never read, so a declared body must close."""
+    assert _deprecated_ack_post(monkeypatch, headers).close_connection is True
+
+
+@pytest.mark.parametrize(
+    "headers", [{}, {"Content-Length": "0"}], ids=["no-content-length", "zero-content-length"]
+)
+def test_bodyless_deprecated_ack_keeps_connection(headers, monkeypatch):
+    """A body-less ack has nothing queued — the 410 must not kill keep-alive."""
+    assert _deprecated_ack_post(monkeypatch, headers).close_connection is False
+
+
+def test_upload_oversize_rejects_before_read(monkeypatch):
+    """An oversized upload's 413 fires before rfile.read() — connection closes."""
+    import api.upload as upload
+
+    handler = SimpleNamespace(
+        headers={"Content-Type": "multipart/form-data; boundary=x", "Content-Length": "99999999999"},
+        close_connection=False,
+    )
+    monkeypatch.setattr(upload, "j", lambda _handler, _payload, status=200: None)
+
+    upload.handle_upload(handler)
+
+    assert handler.close_connection is True
+
+
+# ── Multipart framing preflight: chunked uploads must close, empty ones must not ─
+#
+# All four multipart handlers keyed on Content-Length alone, so a chunked upload
+# (which has none) was read as length 0, matched no parts, and was rejected with
+# "No file field in request" and keep-alive intact — leaving every chunk byte on
+# the socket, which is the exact poisoning this PR exists to prevent.
+# http.server never decodes chunked framing, so the only safe answer is 411 with
+# the connection armed for close.
+
+_MULTIPART_HANDLERS = [
+    "handle_upload",
+    "handle_upload_extract",
+    "handle_transcribe",
+    "handle_workspace_upload",
+]
+
+
+class _NeverRead:
+    """An rfile that fails the test if a reject-before-read path reads the body."""
+
+    def read(self, *_args):
+        raise AssertionError("the body must not be read on a reject-before-read path")
+
+
+def _run_multipart_handler(monkeypatch, handler_name, headers, rfile):
+    import api.upload as upload
+
+    answered: dict[str, Any] = {}
+
+    def _record(_handler, payload, status=200):
+        answered["payload"] = payload
+        answered["status"] = status
+        return True
+
+    monkeypatch.setattr(upload, "j", _record)
+    handler = SimpleNamespace(headers=headers, rfile=rfile, close_connection=False)
+
+    getattr(upload, handler_name)(cast(Any, handler))
+    return handler, answered
+
+
+@pytest.mark.parametrize("handler_name", _MULTIPART_HANDLERS)
+def test_chunked_multipart_rejected_before_read_closes_connection(handler_name, monkeypatch):
+    handler, answered = _run_multipart_handler(
+        monkeypatch,
+        handler_name,
+        {"Content-Type": "multipart/form-data; boundary=x", "Transfer-Encoding": "chunked"},
+        _NeverRead(),
+    )
+
+    assert answered["status"] == 411, answered
+    assert "Transfer-Encoding" in answered["payload"]["error"], answered
+    assert handler.close_connection is True
+
+
+@pytest.mark.parametrize("handler_name", _MULTIPART_HANDLERS)
+def test_bodyless_multipart_rejection_keeps_connection(handler_name, monkeypatch):
+    """A well-framed upload with no body is rejected with nothing left unread."""
+    handler, answered = _run_multipart_handler(
+        monkeypatch,
+        handler_name,
+        {"Content-Type": "multipart/form-data; boundary=x", "Content-Length": "0"},
+        SimpleNamespace(read=lambda *_args: b""),
+    )
+
+    assert answered["status"] == 400, answered
+    assert handler.close_connection is False
+
+
+@pytest.mark.parametrize("handler_name", _MULTIPART_HANDLERS)
+@pytest.mark.parametrize(
+    "headers",
+    [
+        {"Content-Type": "application/json"},
+        {"Content-Type": "application/json", "Content-Length": "0"},
+        {"Content-Type": "multipart/form-data"},  # multipart, but no boundary parameter
+    ],
+    ids=["json-no-length", "json-zero-length", "multipart-without-boundary"],
+)
+def test_bodyless_boundary_rejection_keeps_connection(handler_name, headers, monkeypatch):
+    """`_reject_before_read()` fires on a body-less request too — via the boundary check.
+
+    Every other rejection the preflight raises (any Transfer-Encoding, an unreadable
+    or conflicting length, an over-cap length) already declares a body. The missing
+    `boundary=` in Content-Type is the one that can fire with no body at all, and
+    the unconditional arming closed a healthy pooled socket for it: on the wire,
+    `POST /api/upload` with `Content-Type: application/json` and no `Content-Length`
+    answered 400 "No boundary in Content-Type" WITH `Connection: close` and the
+    pipelined GET was dropped. `_NeverRead` also proves the boundary check still
+    runs ahead of any `rfile.read()`.
+    """
+    handler, answered = _run_multipart_handler(
+        monkeypatch, handler_name, headers, _NeverRead()
+    )
+
+    assert answered["status"] == 400, answered
+    assert answered["payload"]["error"] == "No boundary in Content-Type", answered
+    assert handler.close_connection is False
+
+
+@pytest.mark.parametrize("handler_name", _MULTIPART_HANDLERS)
+def test_boundary_rejection_with_a_declared_body_still_closes(handler_name, monkeypatch):
+    """The close half of the same rejection: a declared body is still queued."""
+    handler, answered = _run_multipart_handler(
+        monkeypatch,
+        handler_name,
+        {"Content-Type": "application/json", "Content-Length": "15"},
+        _NeverRead(),
+    )
+
+    assert answered["status"] == 400, answered
+    assert answered["payload"]["error"] == "No boundary in Content-Type", answered
+    assert handler.close_connection is True
+
+
+@pytest.mark.parametrize("handler_name", _MULTIPART_HANDLERS)
+def test_non_numeric_content_length_still_closes(handler_name, monkeypatch):
+    """The pre-existing invalid-Content-Length reject keeps its 400 and its close."""
+    handler, answered = _run_multipart_handler(
+        monkeypatch,
+        handler_name,
+        {"Content-Type": "multipart/form-data; boundary=x", "Content-Length": "banana"},
+        _NeverRead(),
+    )
+
+    assert answered["status"] == 400, answered
+    assert answered["payload"]["error"] == "Invalid Content-Length", answered
+    assert handler.close_connection is True
+
+
+@pytest.mark.parametrize("handler_name", _MULTIPART_HANDLERS)
+def test_duplicate_content_length_multipart_rejects_before_read(handler_name, monkeypatch):
+    """Disagreeing lengths give no honest byte count -- refuse without reading.
+
+    Reading the first value (0) made the body invisible: the handler answered
+    "No file field in request" with keep-alive and left the real multipart bytes
+    on the socket, where the server parsed them as the next request line.
+    """
+    from email.parser import Parser
+
+    headers = Parser().parsestr(
+        "Content-Type: multipart/form-data; boundary=x\r\n"
+        "Content-Length: 0\r\n"
+        "Content-Length: 88\r\n\r\n"
+    )
+    handler, answered = _run_multipart_handler(monkeypatch, handler_name, headers, _NeverRead())
+
+    assert answered["status"] == 400, answered
+    assert answered["payload"]["error"] == "Invalid Content-Length", answered
+    assert handler.close_connection is True
+
+
+@pytest.mark.parametrize("handler_name", _MULTIPART_HANDLERS)
+@pytest.mark.parametrize(
+    "raw",
+    ["Content-Length:", "Content-Length:   ", "Content-Length:\r\nContent-Length: 63"],
+    ids=["blank", "whitespace-only", "blank-plus-real"],
+)
+def test_blank_content_length_multipart_rejects_before_read(handler_name, raw, monkeypatch):
+    """A blank length reached `int('' or 0)` == 0 and read the body as empty.
+
+    All four handlers then answered 400 "No file field in request" with keep-alive
+    and left the whole multipart payload on the socket. Verified live on the
+    otherwise-fixed head: 400 then `400 Bad request syntax ('--x')`.
+    """
+    from email.parser import Parser
+
+    headers = Parser().parsestr(
+        "Content-Type: multipart/form-data; boundary=x\r\n" + raw + "\r\n\r\n"
+    )
+    handler, answered = _run_multipart_handler(monkeypatch, handler_name, headers, _NeverRead())
+
+    assert answered["status"] == 400, answered
+    assert answered["payload"]["error"] == "Invalid Content-Length", answered
+    assert handler.close_connection is True
+
+
+@pytest.mark.parametrize("handler_name", _MULTIPART_HANDLERS)
+@pytest.mark.parametrize("value", ["+0", "0_0", "\xa00", "٠"], ids=["sign", "underscore", "nbsp", "arabic-indic"])
+def test_malformed_zero_content_length_multipart_rejects_before_read(
+    handler_name, value, monkeypatch
+):
+    """A length only `int()` calls zero slipped the gate into `int(...)` == 0.
+
+    All four handlers then read an empty body, answered 400 "No file field in
+    request" with keep-alive, and left the whole payload on the socket. Live on
+    the otherwise-fixed head: 400 then `400 Bad request syntax ('--x')`.
+    """
+    from email.parser import Parser
+
+    headers = Parser().parsestr(
+        f"Content-Type: multipart/form-data; boundary=x\r\nContent-Length: {value}\r\n\r\n"
+    )
+    handler, answered = _run_multipart_handler(monkeypatch, handler_name, headers, _NeverRead())
+
+    assert answered["status"] == 400, answered
+    assert answered["payload"]["error"] == "Invalid Content-Length", answered
+    assert handler.close_connection is True
+
+
+@pytest.mark.parametrize("handler_name", _MULTIPART_HANDLERS)
+@pytest.mark.parametrize(
+    "raw",
+    [
+        "Transfer-Encoding: identity",
+        "Transfer-Encoding: IDENTITY ",
+        "Transfer-Encoding: \xa0identity",
+        "Transfer-Encoding:",
+        "Transfer-Encoding: identity\r\nContent-Length: 0",
+        "Transfer-Encoding: identity\r\nContent-Length: 88",
+        "Transfer-Encoding: banana",
+    ],
+    ids=[
+        "identity",
+        "identity-cased-and-padded",
+        "nbsp-padded-identity",
+        "blank",
+        "identity-plus-zero-length",
+        "identity-plus-real-length",
+        "unknown-coding",
+    ],
+)
+def test_any_transfer_encoding_multipart_rejects_before_read(handler_name, raw, monkeypatch):
+    """No coding is decodable by an upload handler, so none may reach `rfile`.
+
+    `identity` (any case, any OWS padding) was exempt as a coding that "frames
+    nothing", so each handler fell through to the length-0 read, answered 400 "No
+    file field in request" with keep-alive and left the whole multipart payload on
+    the socket (live: 400 then `400 Bad request syntax ('--x')`). The `\\xa0`-padded
+    spelling and the blank value are the previous round's holes, and the two
+    Content-Length pairings are the RFC 9112 6.3 smuggling shape -- the coding has
+    to win over any length beside it. `_NeverRead` proves the preflight answers
+    before a byte is consumed.
+    """
+    from email.parser import Parser
+
+    headers = Parser().parsestr(
+        "Content-Type: multipart/form-data; boundary=x\r\n" + raw + "\r\n\r\n"
+    )
+    handler, answered = _run_multipart_handler(monkeypatch, handler_name, headers, _NeverRead())
+
+    assert answered["status"] == 411, answered
+    assert "Transfer-Encoding" in answered["payload"]["error"], answered
+    assert handler.close_connection is True
+
+
+# ── Header serialization: exactly one Connection: close, never a crash ────────
+#
+# The advertisement lives in Handler.end_headers() via
+# api.helpers.advertise_connection_close(). Two things went wrong there and
+# both are pinned below:
+#
+#   1. The first cut keyed dedup off a private `_close_advertised` flag it set
+#      itself. Every SSE endpoint already sends its own `Connection: close`
+#      (which BaseHTTPRequestHandler.send_header turns into
+#      close_connection=True), so end_headers() appended a SECOND copy to every
+#      stream response.
+#   2. It read `self.close_connection` unguarded. That attribute only exists
+#      once a request line has been parsed, so the many suite handlers built
+#      with `Handler.__new__(Handler)` raised AttributeError instead of
+#      emitting headers.
+
+
+def _render_response_headers(prepare=None) -> str:
+    """Drive the production Handler.end_headers() into a byte buffer."""
+    import io
+
+    from server import Handler
+
+    handler = Handler.__new__(Handler)
+    handler.request_version = "HTTP/1.1"
+    handler.close_connection = False
+    handler.wfile = io.BytesIO()
+    handler._headers_buffer = []
+    if prepare is not None:
+        prepare(handler)
+    Handler.end_headers(handler)
+    return handler.wfile.getvalue().decode("latin-1")
+
+
+def test_armed_close_is_advertised_exactly_once():
+    """A reject-before-read must TELL the client the socket is gone."""
+    from api.helpers import arm_connection_close
+
+    rendered = _render_response_headers(arm_connection_close)
+
+    assert rendered.count("Connection: close") == 1, rendered
+
+
+def test_sse_style_explicit_close_is_not_doubled():
+    """SSE sends its own Connection: close; end_headers() must not repeat it.
+
+    send_header('Connection', 'close') flips close_connection inside the
+    stdlib, so a self-keyed dedup flag would emit the header twice here.
+    """
+
+    def prepare(handler):
+        handler.send_header("Content-Type", "text/event-stream; charset=utf-8")
+        handler.send_header("Connection", "close")
+
+    rendered = _render_response_headers(prepare)
+
+    assert rendered.count("Connection: close") == 1, rendered
+
+
+def test_keep_alive_response_advertises_no_close():
+    """An ordinary response must keep its keep-alive."""
+    rendered = _render_response_headers(
+        lambda handler: handler.send_header("Content-Type", "application/json")
+    )
+
+    assert "Connection: close" not in rendered, rendered
+
+
+def test_end_headers_on_handler_stub_without_close_connection(monkeypatch):
+    """A partially-built handler stub has no close_connection — don't raise.
+
+    Much of the suite builds handlers with `Handler.__new__(Handler)` and never
+    parses a request, so the attribute is simply absent. Reading it unguarded
+    turned every such test into an AttributeError.
+    """
+    from http.server import BaseHTTPRequestHandler
+
+    from server import Handler
+
+    sent: list[tuple[str, str]] = []
+    handler = Handler.__new__(Handler)
+    handler.send_header = lambda key, value: sent.append((key, value))
+    monkeypatch.setattr(BaseHTTPRequestHandler, "end_headers", lambda self: None)
+
+    assert not hasattr(handler, "close_connection")
+    Handler.end_headers(handler)  # must not raise
+
+    assert "Content-Security-Policy-Report-Only" in dict(sent)
+    assert "Connection" not in dict(sent), "nothing was armed, so nothing to advertise"
+
+
+def test_advertise_close_tolerates_a_non_iterable_header_buffer():
+    """The dedup scan must be as defensive as the two attribute reads.
+
+    A bare ``Mock`` answers every attribute, so ``_headers_buffer`` comes back
+    truthy but non-iterable. Iterating it unguarded raised
+    ``TypeError: 'Mock' object is not iterable`` from inside ``end_headers()``,
+    which ``_handle_write``'s generic ``except`` turns into a spurious 500 —
+    the same crash-inside-end_headers class as the missing
+    ``close_connection`` guard above.
+    """
+    from unittest.mock import Mock
+
+    from api.helpers import advertise_connection_close
+
+    handler = Mock()
+    advertise_connection_close(handler)  # must not raise
+
+    handler.send_header.assert_called_once_with("Connection", "close")
+
+
+def test_check_auth_or_close_on_handler_stub_without_close_connection(monkeypatch):
+    """check_auth_or_close must not invent close_connection on a stub.
+
+    Arming a handler that never had the attribute would leave a bogus
+    keep-alive verdict behind, and reading it unguarded raised AttributeError
+    (which _handle_write then converted into a spurious 500).
+    """
+    import api.auth as auth
+    from server import Handler
+
+    handler = Handler.__new__(Handler)
+    monkeypatch.setattr(auth, "check_auth", lambda _handler, _parsed: True)
+
+    assert auth.check_auth_or_close(handler, SimpleNamespace(path="/api/x")) is True
+    assert not hasattr(handler, "close_connection")
+
+
+@pytest.mark.parametrize(
+    "headers",
+    [{}, {"Content-Length": "0"}, {"Content-Length": "15"}, {"Transfer-Encoding": "chunked"}],
+    ids=["no-content-length", "zero-content-length", "content-length", "chunked"],
+)
+def test_check_auth_or_close_restores_keep_alive_on_success(headers, monkeypatch):
+    """A successful auth must not inherit the armed close.
+
+    The two body-bearing rows are the ones that actually arm before `check_auth()`,
+    so they are what pins the restore; the body-less rows never arm at all.
+    """
+    import api.auth as auth
+
+    handler = SimpleNamespace(headers=headers, close_connection=False)
+    monkeypatch.setattr(auth, "check_auth", lambda _handler, _parsed: True)
+
+    assert auth.check_auth_or_close(cast(Any, handler), SimpleNamespace(path="/api/x")) is True
+    assert handler.close_connection is False
+
+
+def test_check_auth_or_close_restores_a_prior_close_on_success(monkeypatch):
+    """Restore means restore: a connection already doomed stays doomed.
+
+    `Connection: close` sent by the client sets `close_connection` before routing,
+    and the restore must not overwrite that with keep-alive.
+    """
+    import api.auth as auth
+
+    handler = SimpleNamespace(headers={"Content-Length": "15"}, close_connection=True)
+    monkeypatch.setattr(auth, "check_auth", lambda _handler, _parsed: True)
+
+    assert auth.check_auth_or_close(cast(Any, handler), SimpleNamespace(path="/api/x")) is True
+    assert handler.close_connection is True
+
+
+def test_check_auth_or_close_tolerates_a_handler_without_headers(monkeypatch):
+    """The framing read must never raise here: an exception becomes a spurious 500."""
+    import api.auth as auth
+
+    monkeypatch.setattr(auth, "check_auth", lambda _handler, _parsed: False)
+
+    for handler in (
+        SimpleNamespace(close_connection=False),
+        SimpleNamespace(headers=None, close_connection=False),
+        SimpleNamespace(headers=object(), close_connection=False),
+    ):
+        assert auth.check_auth_or_close(cast(Any, handler), SimpleNamespace(path="/api/x")) is False
+        # No framing could be read, so nothing proves a body is pending.
+        assert handler.close_connection is False
+
+
+def test_arm_connection_close_tolerates_a_read_only_handler():
+    """Arming must never turn a 4xx into a 500, even on an odd handler."""
+    from api.helpers import arm_connection_close
+
+    class Frozen:
+        __slots__ = ()
+
+    arm_connection_close(Frozen())  # must not raise
diff --git a/tests/test_rejected_write_pooled_client.py b/tests/test_rejected_write_pooled_client.py
new file mode 100644
index 00000000..6cbcbed1
--- /dev/null
+++ b/tests/test_rejected_write_pooled_client.py
@@ -0,0 +1,759 @@
+"""Pooled-client regression for reject-before-read on HTTP/1.1 keep-alive.
+
+The unit tests beside this one assert `close_connection` is set on each
+rejection path, and the raw-socket test drives the failure over one socket.
+Neither proves the client side: a pooled client (`http.client`) only avoids
+the broken connection when the rejection ADVERTISES `Connection: close`.
+Without the advertisement the server drops the socket silently and the next
+pooled request dies with BrokenPipeError/RemoteDisconnected — or, without the
+close at all, the unread body is parsed as the next request line and the
+client gets a 400 about a request it never sent.
+
+This file drives the production handler through the real test server and
+asserts both halves: the rejection carries `Connection: close`, and a
+follow-up request on the same pooled connection succeeds.
+
+The deprecated `/api/process-complete-ack` path is the probe: it answers 410
+before reading its JSON body, by design (it runs ahead of the CSRF gate so a
+stale tab gets 410, not 403).
+"""
+
+import contextlib
+import http.client
+import socket
+import threading
+import urllib.parse
+from http.server import ThreadingHTTPServer
+
+import pytest
+
+from tests._pytest_port import BASE
+
+_PORT = urllib.parse.urlparse(BASE).port
+_BODY = b'{"stale": true}'
+
+
+def _reject_with_body_then_pooled_followup() -> tuple[int, str | None, int]:
+    """POST a deprecated endpoint with a body, then GET on the same pool."""
+    conn = http.client.HTTPConnection("127.0.0.1", _PORT, timeout=10)
+    try:
+        conn.request(
+            "POST",
+            "/api/process-complete-ack",
+            body=_BODY,
+            headers={"Content-Type": "application/json"},
+        )
+        resp = conn.getresponse()
+        reject_status = resp.status
+        close_header = resp.getheader("Connection")
+        resp.read()
+
+        conn.request("GET", "/api/health/agent")
+        follow = conn.getresponse()
+        follow_status = follow.status
+        follow.read()
+    finally:
+        conn.close()
+    return reject_status, close_header, follow_status
+
+
+def test_rejected_write_advertises_close_and_pooled_followup_succeeds():
+    status, close_header, follow_status = _reject_with_body_then_pooled_followup()
+    assert status == 410, f"expected the deprecated-path 410, got {status}"
+    assert close_header == "close", (
+        f"rejection must advertise Connection: close so pooled clients don't "
+        f"reuse the dead socket; got {close_header!r}"
+    )
+    assert follow_status == 200, (
+        f"pooled follow-up after a reject-before-read must succeed; got {follow_status}"
+    )
+
+
+# ── Framing the Content-Length gate could not see ─────────────────────────────
+#
+# Both cases below are pipelined down ONE socket against the real server: the
+# reject-before-read request first, an ordinary GET immediately after it in the
+# same write. If the rejection closes, the server answers once and EOFs and the
+# GET is never seen. If it does not, the leftover body/chunk bytes are parsed as
+# the next request line and a second response comes back — a 400 bad-syntax or
+# `501 Unsupported method` naming bytes the client never sent as a request.
+
+_FOLLOWING_GET = b"GET /api/health/agent HTTP/1.1\r\nHost: 127.0.0.1\r\nAccept: */*\r\n\r\n"
+_MULTIPART_UPLOAD_PATHS = (
+    "/api/upload",
+    "/api/upload/extract",
+    "/api/workspace/upload",
+    "/api/transcribe",
+)
+_MULTIPART_PAYLOAD = (
+    b'--x\r\nContent-Disposition: form-data; name="file"; filename="a.txt"\r\n'
+    b"\r\nhello\r\n--x--\r\n"
+)
+_CHUNKED_MULTIPART_BODY = (
+    f"{len(_MULTIPART_PAYLOAD):X}\r\n".encode() + _MULTIPART_PAYLOAD + b"\r\n0\r\n\r\n"
+)
+
+
+def _pipelined_after(
+    request: bytes,
+    *,
+    stop_after: int | None = None,
+    port: int | None = None,
+    follow: bytes = _FOLLOWING_GET,
+) -> bytes:
+    """Send *request* and a plain GET in one write; return every byte answered.
+
+    Reads to EOF by default, which is the whole point of the closing cases: the
+    proof is that NOTHING follows the single response. `stop_after` bounds the
+    read for the keep-alive cases, where the socket stays open by design and
+    reading to EOF would only burn the timeout.
+
+    `port`/`follow` exist for the auth-enabled cases below, which need their own
+    server instance and a PUBLIC follow-up path.
+    """
+    sock = socket.create_connection(("127.0.0.1", port or _PORT), timeout=10)
+    try:
+        sock.sendall(request + follow)
+        received = b""
+        while stop_after is None or received.count(b"HTTP/1.1 ") < stop_after:
+            try:
+                chunk = sock.recv(65536)
+            except (TimeoutError, socket.timeout, ConnectionError):
+                break
+            if not chunk:
+                break
+            received += chunk
+        return received
+    finally:
+        sock.close()
+
+
+def _assert_single_closed_response(answered: bytes, status: bytes, leftover: bytes) -> None:
+    text = answered.decode("latin-1", errors="replace")
+    assert answered.startswith(b"HTTP/1.1 " + status), text
+    assert b"Connection: close" in answered, text
+    assert answered.count(b"HTTP/1.1 ") == 1, (
+        f"the pipelined GET was answered, so the socket stayed open: {text}"
+    )
+    assert leftover not in answered, f"the unread body reached the request parser: {text}"
+    assert b"Bad request syntax" not in answered, text
+    assert b"Unsupported method" not in answered, text
+
+
+@pytest.mark.parametrize("path", _MULTIPART_UPLOAD_PATHS)
+def test_chunked_upload_rejection_closes_and_cannot_poison_the_socket(path):
+    """A chunked upload has no Content-Length — the old gate read it as empty.
+
+    All four multipart handlers then answered 400 "No file field in request"
+    with keep-alive intact, leaving the chunk bytes on the socket.
+    """
+    answered = _pipelined_after(
+        f"POST {path} HTTP/1.1\r\n"
+        "Host: 127.0.0.1\r\n"
+        "Content-Type: multipart/form-data; boundary=x\r\n"
+        "Transfer-Encoding: chunked\r\n"
+        "\r\n".encode() + _CHUNKED_MULTIPART_BODY
+    )
+
+    _assert_single_closed_response(answered, b"411", _MULTIPART_PAYLOAD)
+
+
+def test_sidecar_get_with_a_body_closes_and_cannot_poison_the_socket():
+    """A provenance-rejected GET that carries a declared body.
+
+    `read_request_body=False` said "no body" and the 403 went out with
+    keep-alive; the gate's repro then got `501 Unsupported method ('{}GET')` on
+    the same socket. No Origin/Referer/Sec-Fetch-Site here, so provenance fails
+    before the body would ever be read.
+    """
+    answered = _pipelined_after(
+        "GET /api/extensions/probe/sidecar/ping HTTP/1.1\r\n"
+        "Host: 127.0.0.1\r\n"
+        "Content-Type: application/json\r\n"
+        f"Content-Length: {len(_BODY)}\r\n"
+        "\r\n".encode() + _BODY
+    )
+
+    _assert_single_closed_response(answered, b"403", _BODY)
+
+
+# ── Framing hidden behind a DUPLICATED header ─────────────────────────────────
+#
+# `Content-Length: 0` sent ahead of the real length: `Message.get()` returns only
+# the first value, so every reader saw an empty body and drained nothing. Both
+# cases below were reproduced live on the otherwise-fixed head, poisoning the
+# socket exactly as the single-header cases used to:
+#   sidecar GET -> 403, then 400 Bad request syntax ('{"stale": true}GET /...')
+#   /api/upload -> 400, then 400 Bad request syntax ('--x')
+
+
+def test_duplicate_content_length_sidecar_get_closes_and_cannot_poison_the_socket():
+    answered = _pipelined_after(
+        "GET /api/extensions/probe/sidecar/ping HTTP/1.1\r\n"
+        "Host: 127.0.0.1\r\n"
+        "Content-Type: application/json\r\n"
+        "Content-Length: 0\r\n"
+        f"Content-Length: {len(_BODY)}\r\n"
+        "\r\n".encode() + _BODY
+    )
+
+    _assert_single_closed_response(answered, b"403", _BODY)
+
+
+@pytest.mark.parametrize("path", _MULTIPART_UPLOAD_PATHS)
+def test_duplicate_content_length_upload_closes_and_cannot_poison_the_socket(path):
+    answered = _pipelined_after(
+        f"POST {path} HTTP/1.1\r\n"
+        "Host: 127.0.0.1\r\n"
+        "Content-Type: multipart/form-data; boundary=x\r\n"
+        "Content-Length: 0\r\n"
+        f"Content-Length: {len(_MULTIPART_PAYLOAD)}\r\n"
+        "\r\n".encode() + _MULTIPART_PAYLOAD
+    )
+
+    _assert_single_closed_response(answered, b"400", _MULTIPART_PAYLOAD)
+
+
+# ── Framing hidden behind a BLANK header value ────────────────────────────────
+#
+# `Content-Length:` with nothing after the colon reads back as '' and was skipped
+# as "no length declared", so the rejection kept the connection alive. Both cases
+# below were reproduced live on the otherwise-fixed head:
+#   sidecar GET   -> 403, then 400 Bad request syntax ('{"stale": true}GET /...')
+#   /api/upload   -> 400, then 400 Bad request syntax ('--x')
+# `Message` collapses `Content-Length:` and `Content-Length:   ` to the same '',
+# and a blank `Transfer-Encoding:` reached the allowlist of codings that "framed
+# nothing" (since deleted — see the identity section below), so each spelling
+# below poisoned the socket the same way.
+
+_BLANK_FRAMING_HEADER_LINES = (
+    b"Content-Length:\r\n",
+    b"Content-Length:    \r\n",
+    b"Content-Length:\r\nContent-Length:\r\n",
+    b"Content-Length: 0\r\nContent-Length:\r\n",
+    b"Transfer-Encoding:\r\n",
+)
+
+
+@pytest.mark.parametrize("framing", _BLANK_FRAMING_HEADER_LINES)
+def test_blank_framing_sidecar_get_closes_and_cannot_poison_the_socket(framing):
+    answered = _pipelined_after(
+        b"GET /api/extensions/probe/sidecar/ping HTTP/1.1\r\n"
+        b"Host: 127.0.0.1\r\n"
+        b"Content-Type: application/json\r\n" + framing + b"\r\n" + _BODY
+    )
+
+    _assert_single_closed_response(answered, b"403", _BODY)
+
+
+@pytest.mark.parametrize("path", _MULTIPART_UPLOAD_PATHS)
+@pytest.mark.parametrize(
+    ("framing", "status"),
+    [
+        (b"Content-Length:\r\n", b"400"),
+        (b"Content-Length:    \r\n", b"400"),
+        (b"Transfer-Encoding:\r\n", b"411"),
+    ],
+    ids=["blank-length", "whitespace-only-length", "blank-transfer-encoding"],
+)
+def test_blank_framing_upload_closes_and_cannot_poison_the_socket(path, framing, status):
+    answered = _pipelined_after(
+        f"POST {path} HTTP/1.1\r\n"
+        "Host: 127.0.0.1\r\n"
+        "Content-Type: multipart/form-data; boundary=x\r\n".encode()
+        + framing
+        + b"\r\n"
+        + _MULTIPART_PAYLOAD
+    )
+
+    _assert_single_closed_response(answered, status, _MULTIPART_PAYLOAD)
+
+
+# ── Framing only `int()` reads as zero ────────────────────────────────────────
+#
+# RFC 9110 is `Content-Length = 1*DIGIT`; `int()` also takes a sign, PEP 515
+# underscores and non-ASCII whitespace padding, so each value below parsed to an
+# honest `0`, took the "every declared length agrees on zero" keep-alive branch
+# and left the payload queued. Reproduced live on the otherwise-fixed head:
+#   sidecar GET -> 403 (no Connection: close), then
+#                  400 Bad request syntax ('{"stale": true}GET /api/health/agent HTTP/1.1')
+#   /api/upload -> 400, then 400 Bad request syntax ('--x')
+# `Transfer-Encoding: \xa0identity` is the same mistake one header over: strip()
+# erased the U+00A0 and it read back as the `identity` token. Everything here is
+# wire-reachable -- a request line decodes as latin-1, so U+00A0 and U+0085 pass
+# through untouched (a non-ASCII DIGIT such as U+0660 cannot, which is why that
+# family is pinned at the helper level in test_rejected_write_connection_close.py).
+
+_MALFORMED_ZERO_FRAMING_LINES = [
+    b"Content-Length: +0\r\n",
+    b"Content-Length: -0\r\n",
+    b"Content-Length: 0_0\r\n",
+    b"Content-Length: +00\r\n",
+    b"Content-Length: 0_0_0\r\n",
+    b"Content-Length: \xa00\r\n",
+    b"Content-Length: 0\x85\r\n",
+    b"Content-Length: 0\r\nContent-Length: +0\r\n",
+    b"Transfer-Encoding: \xa0identity\r\n",
+]
+
+
+@pytest.mark.parametrize("framing", _MALFORMED_ZERO_FRAMING_LINES)
+def test_malformed_zero_framing_sidecar_get_closes_and_cannot_poison_the_socket(framing):
+    answered = _pipelined_after(
+        b"GET /api/extensions/probe/sidecar/ping HTTP/1.1\r\n"
+        b"Host: 127.0.0.1\r\n"
+        b"Content-Type: application/json\r\n" + framing + b"\r\n" + _BODY
+    )
+
+    _assert_single_closed_response(answered, b"403", _BODY)
+
+
+@pytest.mark.parametrize("path", _MULTIPART_UPLOAD_PATHS)
+@pytest.mark.parametrize(
+    ("framing", "status"),
+    [
+        (b"Content-Length: +0\r\n", b"400"),
+        (b"Content-Length: 0_0\r\n", b"400"),
+        (b"Content-Length: \xa00\r\n", b"400"),
+        (b"Transfer-Encoding: \xa0identity\r\n", b"411"),
+    ],
+    ids=["sign", "underscore", "nbsp-padded", "nbsp-padded-identity"],
+)
+def test_malformed_zero_framing_upload_closes_and_cannot_poison_the_socket(path, framing, status):
+    answered = _pipelined_after(
+        f"POST {path} HTTP/1.1\r\n"
+        "Host: 127.0.0.1\r\n"
+        "Content-Type: multipart/form-data; boundary=x\r\n".encode()
+        + framing
+        + b"\r\n"
+        + _MULTIPART_PAYLOAD
+    )
+
+    _assert_single_closed_response(answered, status, _MULTIPART_PAYLOAD)
+
+
+# ── `Transfer-Encoding: identity` frames nothing a reader can trust ───────────
+#
+# `identity` sat in an allowlist of codings that "frame nothing", so a request
+# carrying it read back as body-less however many payload bytes followed. RFC
+# 9112 §6.3 leaves no room for that: a request whose final transfer coding is not
+# `chunked` cannot be framed by a recipient, so it must be refused and the
+# connection closed — and `http.server` decodes no transfer coding at all, so
+# `identity` is no more readable here than `chunked` is. Reproduced live on the
+# otherwise-fixed head, pipelined down one socket:
+#
+#   GET /api/extensions/probe/sidecar/ping  `Transfer-Encoding: identity`
+#                                           {"stale": true}
+#     -> HTTP/1.1 403 Forbidden        (no Connection: close)
+#     -> HTTP/1.1 400 Bad request syntax
+#            ('{"stale": true}GET /api/health/agent HTTP/1.1')
+#   POST /api/upload  `Transfer-Encoding: identity`  --x...
+#     -> HTTP/1.1 400 Bad Request      (no Connection: close, "No file field")
+#     -> HTTP/1.1 400 Bad request syntax ('--x')
+#
+# Only the spellings that normalize to the bare token were poisoned — every
+# neighbouring value (`identity, chunked`, `identity,identity`, a lone comma, an
+# unknown coding, a non-OWS-padded token, a blank value) already missed the
+# allowlist and already closed. Those rows are in the sweep below anyway: this is
+# the fourth round in which an adjacent spelling of one framing rule survived a
+# fix, so the whole neighbourhood gets pinned rather than just the reported case.
+
+_IDENTITY_FRAMING_LINES = [
+    b"Transfer-Encoding: identity\r\n",
+    b"Transfer-Encoding: IDENTITY\r\n",
+    b"Transfer-Encoding: Identity\r\n",
+    b"Transfer-Encoding:  identity \r\n",
+    b"Transfer-Encoding: \tidentity\t\r\n",
+    b"Transfer-Encoding: identity\r\nTransfer-Encoding: identity\r\n",
+    b"Transfer-Encoding: identity\r\nContent-Length: 0\r\n",
+    b"Transfer-Encoding: identity\r\nContent-Length: " + str(len(_BODY)).encode() + b"\r\n",
+    b"Transfer-Encoding: identity, chunked\r\n",
+    b"Transfer-Encoding: chunked, identity\r\n",
+    b"Transfer-Encoding: identity,identity\r\n",
+    b"Transfer-Encoding: ,\r\n",
+    b"Transfer-Encoding: banana\r\n",
+]
+
+
+@pytest.mark.parametrize("framing", _IDENTITY_FRAMING_LINES)
+def test_identity_framing_sidecar_get_closes_and_cannot_poison_the_socket(framing):
+    answered = _pipelined_after(
+        b"GET /api/extensions/probe/sidecar/ping HTTP/1.1\r\n"
+        b"Host: 127.0.0.1\r\n"
+        b"Content-Type: application/json\r\n" + framing + b"\r\n" + _BODY
+    )
+
+    _assert_single_closed_response(answered, b"403", _BODY)
+
+
+@pytest.mark.parametrize("path", _MULTIPART_UPLOAD_PATHS)
+@pytest.mark.parametrize(
+    "framing",
+    [
+        b"Transfer-Encoding: identity\r\n",
+        b"Transfer-Encoding: IDENTITY \r\n",
+        b"Transfer-Encoding: identity\r\nContent-Length: 0\r\n",
+    ],
+    ids=["identity", "identity-cased-and-padded", "identity-plus-zero-length"],
+)
+def test_identity_framing_upload_closes_and_cannot_poison_the_socket(path, framing):
+    """All four multipart handlers, mirroring the chunked cases above.
+
+    `identity` carries no Content-Length either, so the length-0 read found no
+    parts and each handler answered 400 "No file field in request" with the whole
+    multipart payload still queued on an open socket.
+    """
+    answered = _pipelined_after(
+        f"POST {path} HTTP/1.1\r\n"
+        "Host: 127.0.0.1\r\n"
+        "Content-Type: multipart/form-data; boundary=x\r\n".encode()
+        + framing
+        + b"\r\n"
+        + _MULTIPART_PAYLOAD
+    )
+
+    _assert_single_closed_response(answered, b"411", _MULTIPART_PAYLOAD)
+
+
+@pytest.mark.parametrize(
+    "framing",
+    [
+        b"",
+        b"Content-Length: 0\r\n",
+        b"Content-Length: 00\r\n",
+        b"Content-Length:  0 \r\n",
+        b"Content-Length: 0\r\nContent-Length: 0\r\n",
+    ],
+    ids=[
+        "no-length",
+        "zero-length",
+        "double-zero-length",
+        "ows-padded-zero",
+        "two-agreeing-zeroes",
+    ],
+)
+def test_bodyless_framing_keeps_the_pooled_socket_alive(framing):
+    """The over-close half of the contract, on the wire.
+
+    Framing that positively says "no body" must NOT be swept up by the blank
+    rule: the rejection answers without `Connection: close` and the pipelined GET
+    is served on the same socket. Two agreeing zeroes are still no body.
+    """
+    answered = _pipelined_after(
+        b"GET /api/extensions/probe/sidecar/ping HTTP/1.1\r\n"
+        b"Host: 127.0.0.1\r\n" + framing + b"\r\n",
+        stop_after=2,
+    )
+
+    text = answered.decode("latin-1", errors="replace")
+    assert answered.startswith(b"HTTP/1.1 403"), text
+    assert b"HTTP/1.1 200 OK" in answered, (
+        f"the pipelined GET was not served, so a body-less rejection dropped a "
+        f"healthy pooled connection: {text}"
+    )
+    assert b"Connection: close" not in answered, text
+
+
+# ── Every reject-before-read site, in BOTH directions, on the wire ─────────────
+#
+# The sidecar path above was the first site routed through the framing-aware
+# helper; the sibling sites kept the unconditional `arm_connection_close()` and
+# so were the mirror image of the same bug -- a body-less rejection answered with
+# `Connection: close` and dropped the client's pipelined follow-up, killing a
+# healthy keep-alive connection for no framing reason. Each pair below is the
+# reproduction (body-less -> must stay alive) beside its close half (a declared
+# body -> must still close), pipelined down one socket against the production
+# handler. Reproduced on the pre-fix head, all seven sites, `responses seen: 1`:
+#
+#   POST /api/session/new        (auth off-session)  -> 401 + Connection: close
+#   POST /api/session/new        (bad Origin)        -> 403 + Connection: close
+#   POST /api/session/new        (no CSRF token)     -> 403 + Connection: close
+#   POST /api/csp-report         (limiter tripped)   -> 204 + Connection: close
+#   POST /api/health/restart     (success)           -> 200 + Connection: close
+#   POST /api/process-complete-ack                   -> 410 + Connection: close
+#   POST /api/upload             (no boundary)       -> 400 + Connection: close
+
+_BODYLESS_FRAMING = [b"", b"Content-Length: 0\r\n"]
+_BODYLESS_IDS = ["no-content-length", "zero-content-length"]
+# A PUBLIC path, so the follow-up is a genuine 200 even with auth enabled.
+_PUBLIC_FOLLOWING_GET = (
+    b"GET /api/auth/status HTTP/1.1\r\nHost: 127.0.0.1\r\nAccept: */*\r\n\r\n"
+)
+
+
+@contextlib.contextmanager
+def _own_server():
+    """The production Handler on a private port.
+
+    The shared test server runs with auth disabled and out of process, so the
+    auth/CSRF-token rejections and the deterministic restart/limiter outcomes need
+    an instance this process can configure. It is the same `server.Handler` class
+    the real server binds, driven over a real socket.
+    """
+    import server
+
+    httpd = ThreadingHTTPServer(("127.0.0.1", 0), server.Handler)
+    thread = threading.Thread(target=httpd.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield httpd.server_address[1]
+    finally:
+        httpd.shutdown()
+        httpd.server_close()
+        thread.join(timeout=10)
+
+
+@pytest.fixture
+def auth_on(monkeypatch):
+    """Turn password auth on for this test (conftest resets the hash cache)."""
+    import api.auth as auth
+
+    monkeypatch.setenv("HERMES_WEBUI_PASSWORD", "pooled-client-regression")
+    auth._invalidate_password_hash_cache()
+    assert auth.is_auth_enabled(), "the auth-path regression needs auth enabled"
+    return auth
+
+
+def _assert_kept_alive(answered: bytes, status: bytes) -> None:
+    text = answered.decode("latin-1", errors="replace")
+    assert answered.startswith(b"HTTP/1.1 " + status), text
+    assert b"Connection: close" not in answered, (
+        f"a body-less rejection advertised close, so a pooled client's healthy "
+        f"connection was dropped: {text}"
+    )
+    assert b"HTTP/1.1 200 OK" in answered, (
+        f"the pipelined follow-up was not served on the same socket: {text}"
+    )
+
+
+def _assert_closed(answered: bytes, status: bytes) -> None:
+    text = answered.decode("latin-1", errors="replace")
+    assert answered.startswith(b"HTTP/1.1 " + status), text
+    assert b"Connection: close" in answered, text
+    assert answered.count(b"HTTP/1.1 ") == 1, (
+        f"the pipelined GET was answered, so the socket stayed open: {text}"
+    )
+
+
+@pytest.mark.parametrize("framing", _BODYLESS_FRAMING, ids=_BODYLESS_IDS)
+def test_bodyless_auth_rejection_keeps_the_pooled_socket_alive(framing, auth_on):
+    """A body-less POST that fails auth must not cost the client its connection."""
+    with _own_server() as port:
+        answered = _pipelined_after(
+            b"POST /api/session/new HTTP/1.1\r\nHost: 127.0.0.1\r\n" + framing + b"\r\n",
+            stop_after=2,
+            port=port,
+            follow=_PUBLIC_FOLLOWING_GET,
+        )
+
+    _assert_kept_alive(answered, b"401")
+
+
+def test_auth_rejection_with_a_body_still_closes_the_pooled_socket(auth_on):
+    """The close half at the auth gate: those bytes are still queued in rfile."""
+    with _own_server() as port:
+        answered = _pipelined_after(
+            b"POST /api/session/new HTTP/1.1\r\nHost: 127.0.0.1\r\n"
+            b"Content-Type: application/json\r\n"
+            b"Content-Length: " + str(len(_BODY)).encode() + b"\r\n\r\n" + _BODY,
+            port=port,
+            follow=_PUBLIC_FOLLOWING_GET,
+        )
+
+    _assert_closed(answered, b"401")
+    assert _BODY not in answered, answered.decode("latin-1", errors="replace")
+
+
+@pytest.mark.parametrize("framing", _BODYLESS_FRAMING, ids=_BODYLESS_IDS)
+def test_bodyless_csrf_origin_rejection_keeps_the_pooled_socket_alive(framing):
+    """`_check_csrf()` origin mismatch, on the shared server (auth off)."""
+    answered = _pipelined_after(
+        b"POST /api/session/new HTTP/1.1\r\nHost: 127.0.0.1\r\n"
+        b"Origin: http://evil.invalid\r\n" + framing + b"\r\n",
+        stop_after=2,
+    )
+
+    _assert_kept_alive(answered, b"403")
+
+
+def test_csrf_origin_rejection_with_a_body_still_closes_the_pooled_socket():
+    answered = _pipelined_after(
+        b"POST /api/session/new HTTP/1.1\r\nHost: 127.0.0.1\r\n"
+        b"Origin: http://evil.invalid\r\nContent-Type: application/json\r\n"
+        b"Content-Length: " + str(len(_BODY)).encode() + b"\r\n\r\n" + _BODY
+    )
+
+    _assert_closed(answered, b"403")
+    assert _BODY not in answered, answered.decode("latin-1", errors="replace")
+
+
+def _authenticated_same_origin_post(port, cookie_name, cookie, framing, body=b""):
+    return (
+        f"POST /api/session/new HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\n"
+        f"Origin: http://127.0.0.1:{port}\r\n"
+        f"Cookie: {cookie_name}={cookie}\r\n".encode() + framing + b"\r\n" + body
+    )
+
+
+@pytest.mark.parametrize("framing", _BODYLESS_FRAMING, ids=_BODYLESS_IDS)
+def test_bodyless_csrf_token_rejection_keeps_the_pooled_socket_alive(framing, auth_on):
+    """A real session, a same-origin POST, and no CSRF token -> token_mismatch."""
+    cookie = auth_on.create_session()
+    try:
+        with _own_server() as port:
+            answered = _pipelined_after(
+                _authenticated_same_origin_post(
+                    port, auth_on._resolve_cookie_name(), cookie, framing
+                ),
+                stop_after=2,
+                port=port,
+                follow=_PUBLIC_FOLLOWING_GET,
+            )
+    finally:
+        auth_on.invalidate_session(cookie)
+
+    _assert_kept_alive(answered, b"403")
+
+
+def test_csrf_token_rejection_with_a_body_still_closes_the_pooled_socket(auth_on):
+    cookie = auth_on.create_session()
+    try:
+        with _own_server() as port:
+            answered = _pipelined_after(
+                _authenticated_same_origin_post(
+                    port,
+                    auth_on._resolve_cookie_name(),
+                    cookie,
+                    b"Content-Type: application/json\r\nContent-Length: "
+                    + str(len(_BODY)).encode()
+                    + b"\r\n",
+                    body=_BODY,
+                ),
+                port=port,
+                follow=_PUBLIC_FOLLOWING_GET,
+            )
+    finally:
+        auth_on.invalidate_session(cookie)
+
+    _assert_closed(answered, b"403")
+    assert _BODY not in answered, answered.decode("latin-1", errors="replace")
+
+
+@pytest.fixture
+def csp_limiter_tripped(monkeypatch):
+    """Force the CSP-report limiter, instead of sending 101 reports in 60s."""
+    import api.routes as routes
+
+    monkeypatch.setattr(routes, "_csp_report_rate_limited", lambda _handler: True)
+
+
+@pytest.mark.parametrize("framing", _BODYLESS_FRAMING, ids=_BODYLESS_IDS)
+def test_bodyless_rate_limited_csp_report_keeps_the_pooled_socket_alive(
+    framing, csp_limiter_tripped
+):
+    """A dropped 204 must not hang up on a browser that is still reporting."""
+    with _own_server() as port:
+        answered = _pipelined_after(
+            b"POST /api/csp-report HTTP/1.1\r\nHost: 127.0.0.1\r\n" + framing + b"\r\n",
+            stop_after=2,
+            port=port,
+            follow=_PUBLIC_FOLLOWING_GET,
+        )
+
+    _assert_kept_alive(answered, b"204")
+
+
+def test_rate_limited_csp_report_with_a_body_still_closes(csp_limiter_tripped):
+    """The limiter answers without reading the report, so a declared body closes."""
+    with _own_server() as port:
+        answered = _pipelined_after(
+            b"POST /api/csp-report HTTP/1.1\r\nHost: 127.0.0.1\r\n"
+            b"Content-Type: application/csp-report\r\n"
+            b"Content-Length: " + str(len(_BODY)).encode() + b"\r\n\r\n" + _BODY,
+            port=port,
+            follow=_PUBLIC_FOLLOWING_GET,
+        )
+
+    _assert_closed(answered, b"204")
+
+
+@pytest.fixture
+def restart_succeeds(monkeypatch):
+    """Pin the SUCCESS outcome -- the one that closed a healthy socket every call."""
+    import api.routes as routes
+
+    monkeypatch.setattr(
+        routes, "restart_active_profile_gateway", lambda: {"status": "completed"}
+    )
+
+
+@pytest.mark.parametrize("framing", _BODYLESS_FRAMING, ids=_BODYLESS_IDS)
+def test_bodyless_successful_restart_keeps_the_pooled_socket_alive(framing, restart_succeeds):
+    """`/api/health/restart` closed on success too; the WebUI sends no body."""
+    with _own_server() as port:
+        answered = _pipelined_after(
+            b"POST /api/health/restart HTTP/1.1\r\nHost: 127.0.0.1\r\n" + framing + b"\r\n",
+            stop_after=2,
+            port=port,
+            follow=_PUBLIC_FOLLOWING_GET,
+        )
+
+    _assert_kept_alive(answered, b"200")
+
+
+def test_restart_with_a_body_still_closes_the_pooled_socket(restart_succeeds):
+    """The endpoint consumes its body on no outcome, so a declared one closes."""
+    with _own_server() as port:
+        answered = _pipelined_after(
+            b"POST /api/health/restart HTTP/1.1\r\nHost: 127.0.0.1\r\n"
+            b"Content-Type: application/json\r\n"
+            b"Content-Length: " + str(len(_BODY)).encode() + b"\r\n\r\n" + _BODY,
+            port=port,
+            follow=_PUBLIC_FOLLOWING_GET,
+        )
+
+    _assert_closed(answered, b"200")
+    assert _BODY not in answered, answered.decode("latin-1", errors="replace")
+
+
+@pytest.mark.parametrize("framing", _BODYLESS_FRAMING, ids=_BODYLESS_IDS)
+def test_bodyless_deprecated_ack_keeps_the_pooled_socket_alive(framing):
+    """The 410 alias, which the body-bearing test at the top of this file pins closed."""
+    answered = _pipelined_after(
+        b"POST /api/process-complete-ack HTTP/1.1\r\nHost: 127.0.0.1\r\n" + framing + b"\r\n",
+        stop_after=2,
+    )
+
+    _assert_kept_alive(answered, b"410")
+
+
+@pytest.mark.parametrize("path", _MULTIPART_UPLOAD_PATHS)
+@pytest.mark.parametrize(
+    "framing",
+    [
+        b"Content-Type: application/json\r\n",
+        b"Content-Type: application/json\r\nContent-Length: 0\r\n",
+        b"Content-Type: multipart/form-data\r\n",  # multipart, but no boundary=
+    ],
+    ids=["json-no-length", "json-zero-length", "multipart-without-boundary"],
+)
+def test_bodyless_upload_boundary_rejection_keeps_the_pooled_socket_alive(path, framing):
+    """`_reject_before_read()`'s one body-less reachable rejection: the boundary check."""
+    answered = _pipelined_after(
+        f"POST {path} HTTP/1.1\r\nHost: 127.0.0.1\r\n".encode() + framing + b"\r\n",
+        stop_after=2,
+    )
+
+    _assert_kept_alive(answered, b"400")
+
+
+@pytest.mark.parametrize("path", _MULTIPART_UPLOAD_PATHS)
+def test_upload_boundary_rejection_with_a_body_still_closes(path):
+    answered = _pipelined_after(
+        f"POST {path} HTTP/1.1\r\nHost: 127.0.0.1\r\n"
+        "Content-Type: application/json\r\n"
+        f"Content-Length: {len(_BODY)}\r\n\r\n".encode() + _BODY
+    )
+
+    _assert_closed(answered, b"400")
+    assert _BODY not in answered, answered.decode("latin-1", errors="replace")
diff --git a/tests/test_rejected_write_socket_reuse.py b/tests/test_rejected_write_socket_reuse.py
new file mode 100755
index 00000000..824e99f3
--- /dev/null
+++ b/tests/test_rejected_write_socket_reuse.py
@@ -0,0 +1,110 @@
+"""A rejected write must not leave its unread body on a reused connection.
+
+The unit tests beside this one assert `close_connection` is set on each rejection
+path. That pins the mechanism but not the consequence: they build a
+SimpleNamespace, never open a socket, and would still pass if
+BaseHTTPRequestHandler stopped honouring the flag.
+
+This drives the actual failure over one socket: a POST that is rejected before its
+body is read, then an ordinary GET on the same connection. Without the fix the
+server parses the leftover body as the next request line and answers
+
+    HTTP/1.1 400 Bad request syntax ('{"a": "b"}GET /api/health HTTP/1.1')
+
+which is an error about a request the client never sent - the reason this class of
+bug is expensive to diagnose from the client side. With the fix the server closes
+after the rejection, so the client sees a clean EOF and reconnects instead.
+
+The two files are complementary and neither replaces the other. The unit tests
+call the production handlers, so they pin that `close_connection` is actually set
+on each rejection path. This file uses a minimal handler of the same shape, so it
+pins WHY that flag matters at the HTTP level - which no amount of asserting on a
+boolean can show. Reverting server.py fails the unit test; changing the handler
+here to reuse the connection fails this one.
+"""
+
+import socket
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+BODY = b'{"a": "b"}'
+REJECTED_WRITE = (
+    b"POST /api/session/new HTTP/1.1\r\n"
+    b"Host: localhost\r\n"
+    b"Content-Type: application/json\r\n"
+    b"Content-Length: " + str(len(BODY)).encode() + b"\r\n"
+    b"\r\n" + BODY
+)
+FOLLOWING_READ = b"GET /api/health HTTP/1.1\r\nHost: localhost\r\n\r\n"
+
+
+class _Rejecting(BaseHTTPRequestHandler):
+    """The production shape: reject an unsafe write before reading its body."""
+
+    protocol_version = "HTTP/1.1"
+    closes_on_reject = True
+
+    def log_message(self, *args):
+        pass  # keep the test output quiet
+
+    def do_POST(self):
+        # Deliberately does NOT read the body. This is the production path when
+        # auth or CSRF rejects a write before read_body() runs.
+        if self.closes_on_reject:
+            self.close_connection = True
+        self.send_response(403)
+        self.send_header("Content-Length", "0")
+        self.end_headers()
+
+    def do_GET(self):
+        self.send_response(200)
+        self.send_header("Content-Length", "2")
+        self.end_headers()
+        self.wfile.write(b"ok")
+
+
+def _second_response(*, closes_on_reject: bool) -> bytes:
+    """Send a rejected write then a GET down one socket; return the GET's answer."""
+    handler = type("_H", (_Rejecting,), {"closes_on_reject": closes_on_reject})
+    server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    try:
+        sock = socket.create_connection(server.server_address, timeout=5)
+        try:
+            sock.sendall(REJECTED_WRITE)
+            time.sleep(0.3)
+            assert b"403" in sock.recv(4096), "the write should have been rejected"
+
+            sock.sendall(FOLLOWING_READ)
+            time.sleep(0.3)
+            try:
+                return sock.recv(4096)
+            except (socket.timeout, ConnectionError):
+                return b""
+        finally:
+            sock.close()
+    finally:
+        server.shutdown()
+        server.server_close()
+
+
+def test_rejected_write_does_not_corrupt_the_next_request():
+    """Closing means the client sees EOF, never a bogus error about its own GET."""
+    answer = _second_response(closes_on_reject=True)
+    assert b"400" not in answer, f"the leftover body reached the parser: {answer!r}"
+    assert BODY not in answer, f"the leftover body reached the parser: {answer!r}"
+    # The connection is gone, which is the point: the client reconnects rather
+    # than being told its next request was malformed.
+    assert answer == b"", f"expected a closed connection, got {answer!r}"
+
+
+def test_without_the_fix_the_unread_body_corrupts_the_next_request():
+    """The bug itself, so the test above is known to be testing something.
+
+    Reusing the connection makes the server read the unread body as a request
+    line, and the body text appears in the 400 it sends back.
+    """
+    answer = _second_response(closes_on_reject=False)
+    assert b"400" in answer, f"expected a parse failure, got {answer!r}"
+    assert BODY in answer, f"expected the leftover body in the error, got {answer!r}"