test(lsp): cover the select-based read-body timeout branch via os.pipe()

The DoS mitigation added in db389a730 wraps each message-body read in
``select()`` so a stalled client cannot pin the server forever. The
existing transport tests route through MockLspTransport, whose
``BytesIO`` raises ``UnsupportedOperation`` on ``fileno()`` -- so
``_read_body_with_timeout`` always falls through the BytesIO fast
path and the actual ``select()``-based mitigation code (the part
that runs in production) is never executed. That left ~19 new lines
uncovered, dragging total coverage below the 96.5% floor and
failing the coverage gate.

Two new scenarios drive the helper through an ``os.pipe()`` whose
read fd satisfies ``fileno()``, so ``use_select`` is True and the
real DoS-protection path runs:

* ``timeout=0.0`` makes the deadline already past on the first
  iteration, exercising the ``if timeout <= 0:`` early-exit warning.
* ``timeout=0.05`` lets ``select.select()`` run and time out with no
  ready descriptors, exercising the ``if not ready:`` warning.

Both paths log ``lsp.transport.read_timeout`` and return ``None``,
matching the production behaviour the helper was added to provide.

ISSUES CLOSED: #5566

features/lsp_server_stub.feature

@@ -483,3 +483,15 @@ Feature: LSP server stub JSON-RPC protocol
     When I check the MESSAGE_READ_TIMEOUT constant
     Then the MESSAGE_READ_TIMEOUT should be 30
     And MESSAGE_READ_TIMEOUT should be importable from lsp server module
+
+  @lsp_transport_edge @lsp_dos_protection
+  Scenario: Read body times out immediately with read_timeout zero on a real pipe
+    When I invoke read body with timeout 0.0 on an empty pipe expecting 10 bytes
+    Then the read body result should be None
+    And a transport read_timeout warning should be captured
+
+  @lsp_transport_edge @lsp_dos_protection
+  Scenario: Read body times out when select reports no data on a real pipe
+    When I invoke read body with timeout 0.05 on an empty pipe expecting 10 bytes
+    Then the read body result should be None
+    And a transport read_timeout warning should be captured

features/steps/lsp_server_stub_steps.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 import contextlib
 import io
 import json
+import os
 from collections.abc import Generator
 from typing import Any
 
@@ -807,3 +808,72 @@ def step_message_read_timeout_importable(context: Context) -> None:
     assert hasattr(_lsp_server_module, "MESSAGE_READ_TIMEOUT"), (
         "MESSAGE_READ_TIMEOUT attribute not found in lsp.server module"
     )
+
+
+# ---------------------------------------------------------------------------
+# Real-pipe timeout coverage for ``_read_body_with_timeout``
+# ---------------------------------------------------------------------------
+# The MockLspTransport uses ``BytesIO`` whose ``fileno()`` raises
+# ``UnsupportedOperation``, so the DoS mitigation's ``select()`` guard is
+# never exercised by transport tests routed through the mock.  These steps
+# wire a real ``os.pipe()`` into the server so the actual timeout code path
+# (the one that protects production from partial-body stalls) runs end to end.
+
+
+@when(
+    "I invoke read body with timeout {timeout:f} on an empty pipe expecting "
+    "{expected:d} bytes"
+)
+def step_read_body_pipe_no_data(
+    context: Context, timeout: float, expected: int
+) -> None:
+    """Exercise ``_read_body_with_timeout``'s select-based timeout path.
+
+    A pipe fd makes ``fileno()`` succeed so ``use_select`` is True and the
+    DoS-protection branch runs.  No bytes are written before the call:
+
+    * ``timeout=0.0`` makes the deadline already past on entry, exercising
+      the ``if timeout <= 0:`` early-exit warning.
+    * A small positive timeout (e.g. ``0.05``) lets ``select.select()`` run
+      and return no ready descriptors, exercising the ``if not ready:``
+      warning.
+
+    Both paths log ``lsp.transport.read_timeout`` and return ``None``.
+    """
+    read_fd, write_fd = os.pipe()
+    pipe_in = os.fdopen(read_fd, "rb", buffering=0)
+    server = LspServer(
+        input_stream=pipe_in,
+        output_stream=io.BytesIO(),
+        read_timeout=timeout,
+    )
+    try:
+        with _capture_structlogs() as captured:
+            result = server._read_body_with_timeout(expected)
+        context.lsp_read_body_result = result
+        context.lsp_captured_logs = list(captured)
+    finally:
+        os.close(write_fd)
+        pipe_in.close()
+
+
+@then("the read body result should be None")
+def step_read_body_result_none(context: Context) -> None:
+    """Assert the timeout path returned ``None`` (so the caller can ``_SKIP``)."""
+    assert context.lsp_read_body_result is None, (
+        f"expected None, got {context.lsp_read_body_result!r}"
+    )
+
+
+@then("a transport read_timeout warning should be captured")
+def step_transport_read_timeout_captured(context: Context) -> None:
+    """Assert at least one ``lsp.transport.read_timeout`` event was logged."""
+    matches = [
+        entry
+        for entry in context.lsp_captured_logs
+        if entry.get("event") == "lsp.transport.read_timeout"
+    ]
+    assert matches, (
+        "no lsp.transport.read_timeout warning in captured logs: "
+        f"{context.lsp_captured_logs}"
+    )