Split brother_printer PJL wire transport out of usb_query

usb_query.py 267 -> 180. The non-blocking fd handling - _drain_buffer,
_write_with_deadline, _read_nonblocking, _wait_for_pjl_response and
pjl_query - moves to _pjl_io.py.

usb_query imports the three names test_usb_query.py patches back, which
is safe here only because all three still have real call sites in the
code that stayed; an import-only name would be stripped by ruff's F401
sweep and break the patch. os and time also stay: they are patched on
usb_query as well as on _pjl_io.

test_usb_query.py needs no edit. Pass count is 402 before and after.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011w2Pj4Ek3ZNMQVYcFHgXDY

docs/superpowers/evidence/brother-printer-pjl-io-split-2026-08-16.json

@@ -0,0 +1,51 @@
+{
+  "intent": "Bring brother_printer/usb_query.py under the 250-line cap by moving the non-blocking PJL transport into its own module, with no behaviour change.",
+  "scope": [
+    "python_pkg/brother_printer/_pjl_io.py (new)",
+    "python_pkg/brother_printer/usb_query.py",
+    "python_pkg/brother_printer/tests/test_usb_query_part2.py (MOD and import path only)",
+    "Non-goal: no assertion changed, no test deleted, skipped or weakened"
+  ],
+  "changes": [
+    "Moved _drain_buffer, _write_with_deadline, _read_nonblocking, _wait_for_pjl_response and pjl_query verbatim into python_pkg/brother_printer/_pjl_io.py, together with the comment explaining why the fd is kept non-blocking; 267 -> 180 lines.",
+    "usb_query.py imports _drain_buffer, _write_with_deadline and pjl_query back, because test_usb_query.py patches all three on that module. Checked first that each is genuinely called by the code that stayed (_retry_pjl_query and _run_pjl_queries call all three), so ruff cannot strip them under F401 and no __all__ shim is needed.",
+    "Dropped contextlib and select from usb_query, which the moved block was their only user. Kept os and time despite the move: test_usb_query.py patches MOD.os.access/open/close and MOD.time.sleep, and _retry_pjl_query / query_usb_pjl still use both.",
+    "_pjl_io imports os, select, time and contextlib by name because test_usb_query_part2.py patches MOD.os.read, MOD.os.write, MOD.select.select and MOD.time.time on it.",
+    "test_usb_query_part2.py had its MOD constant and import module repointed at _pjl_io. test_usb_query.py needed no change, which is the signal the seam matched the existing test boundary."
+  ],
+  "verification": [
+    {
+      "command": "PYTHONPATH=. python3 -m pytest python_pkg/brother_printer/tests -q",
+      "result": "pass",
+      "evidence": "402 passed before the split and 402 passed after it - identical pass count, not merely a green run."
+    },
+    {
+      "command": "PYTHONPATH=. python3 -c \"from python_pkg.brother_printer import usb_query as u, _pjl_io as p; u._drain_buffer; u.pjl_query; u._write_with_deadline; u.time; u.os; u.Path; u.shutil; p.os; p.select; p.time; p.contextlib; p._read_nonblocking\"",
+      "result": "pass",
+      "evidence": "runtime probe OK - the three re-imported names are real attributes of usb_query, so test_usb_query.py's patches still reach them."
+    },
+    {
+      "command": "grep -cw '_drain_buffer|pjl_query|_write_with_deadline' over the kept region of usb_query.py",
+      "result": "pass",
+      "evidence": "5, 1 and 1 real call sites respectively - all three imports are used, so ruff's F401 sweep leaves them alone. Had any been import-only it would have needed an __all__ entry per docs/python-split-recipes.md."
+    },
+    {
+      "command": "wc -l python_pkg/brother_printer/usb_query.py python_pkg/brother_printer/_pjl_io.py",
+      "result": "pass",
+      "evidence": "180 and 109 lines, both well under the 250-line cap."
+    },
+    {
+      "command": "pre-commit run --files <the four touched files>",
+      "result": "pass",
+      "evidence": "ruff, ruff format, mypy, pylint, bandit and pytest-with-coverage all pass on the touched files."
+    }
+  ],
+  "risks": [
+    "The three re-imported names are load-bearing for mock.patch, not just for calls. If a future edit removes the last call site of one, ruff will strip the import and test_usb_query.py will fail with 'does not have the attribute' - the fix is an __all__ entry, not deleting the patch.",
+    "usb_query <- _pjl_io is one-way; _pjl_io must never import from usb_query."
+  ],
+  "rollback": [
+    "git revert the commit; usb_query.py returns to 267 lines and _pjl_io.py disappears.",
+    "After rollback, re-run the package suite and confirm it still reports 402 passed."
+  ]
+}

python_pkg/brother_printer/_pjl_io.py

@@ -0,0 +1,108 @@
+"""Non-blocking PJL transport over the raw /dev/usb/lp* file descriptor.
+
+Split out of :mod:`python_pkg.brother_printer.usb_query` to keep it under the
+250-line cap. ``tests/test_usb_query_part2.py`` patches ``os``, ``select`` and
+``time`` on this module, so those imports are load-bearing; the query callers
+that stay in ``usb_query`` import these functions back, which is what keeps
+``tests/test_usb_query.py`` patching them there.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import os
+import select
+import time
+
+# ── PJL over USB ─────────────────────────────────────────────────────
+
+
+# The printer's fd is opened and kept non-blocking. usblp blocks indefinitely
+# on a plain open() and on writes whenever the printer is not ready to talk -
+# stuck mid-job, wedged, or simply asleep - so a blocking fd turns "the printer
+# is unwell" into "this tool hangs forever", which is the worst possible answer
+# when a status report is exactly what you wanted.
+
+
+def _drain_buffer(fd: int) -> None:
+    """Read and discard any stale data from the USB buffer."""
+    with contextlib.suppress(OSError):
+        while os.read(fd, 4096):
+            pass
+
+
+def _write_with_deadline(fd: int, data: bytes, deadline: float) -> bool:
+    """Write all of data to a non-blocking fd. Returns False if the deadline passes.
+
+    Args:
+        fd: Non-blocking file descriptor for the printer.
+        data: Bytes to send.
+        deadline: Absolute monotonic-ish time (time.time()) to give up at.
+
+    Returns:
+        True when everything was written, False on timeout or write error.
+    """
+    sent = 0
+    while sent < len(data):
+        remaining = deadline - time.time()
+        if remaining <= 0:
+            return False
+        try:
+            sent += os.write(fd, data[sent:])
+        except BlockingIOError:
+            # Printer's buffer is full or it is not accepting data: wait for it
+            # to become writable rather than spinning or blocking forever.
+            _, writable, _ = select.select([], [fd], [], min(remaining, 0.5))
+            if not writable:
+                continue
+        except OSError:
+            return False
+    return True
+
+
+def _read_nonblocking(fd: int) -> bytes:
+    """Read all currently available data from a non-blocking fd."""
+    data = b""
+    with contextlib.suppress(OSError):
+        while True:
+            chunk = os.read(fd, 4096)
+            if not chunk:
+                break
+            data += chunk
+    return data
+
+
+def _wait_for_pjl_response(fd: int, deadline: float) -> bytes:
+    """Poll fd until PJL data arrives or deadline expires."""
+    response = b""
+    while time.time() < deadline:
+        remaining = deadline - time.time()
+        if remaining <= 0:
+            break
+        readable, _, _ = select.select([fd], [], [], min(remaining, 1.0))
+        if readable:
+            response += _read_nonblocking(fd)
+            if response and (b"=" in response or b"@PJL" in response):
+                break
+    return response
+
+
+def pjl_query(fd: int, cmd: str, timeout_sec: float = 5.0) -> str:
+    """Send a PJL command via raw fd and read the response.
+
+    Args:
+        fd: Non-blocking file descriptor for the printer.
+        cmd: PJL command, e.g. "@PJL INFO STATUS".
+        timeout_sec: Total budget for sending the command and reading the reply.
+
+    Returns:
+        The printer's reply, or "" if it did not answer in time. A silent
+        printer is a normal outcome here, not an error: it happens whenever the
+        printer is busy digesting a job.
+    """
+    deadline = time.time() + timeout_sec
+    pjl_cmd = f"\x1b%-12345X@PJL\r\n{cmd}\r\n\x1b%-12345X"
+    if not _write_with_deadline(fd, pjl_cmd.encode(), deadline):
+        return ""
+    response = _wait_for_pjl_response(fd, deadline)
+    return response.decode("ascii", errors="replace")

python_pkg/brother_printer/tests/test_usb_query_part2.py

@@ -1,4 +1,4 @@
-"""Tests for brother_printer.usb_query - the PJL wire protocol.
+"""Tests for brother_printer._pjl_io - the PJL wire protocol.
 
 The printer's fd is non-blocking throughout: usblp blocks forever on open
 and on writes whenever the printer is unwell, which is exactly when a status
@@ -9,7 +9,7 @@ from __future__ import annotations
 
 from unittest.mock import MagicMock, patch
 
-from python_pkg.brother_printer.usb_query import (
+from python_pkg.brother_printer._pjl_io import (
     _drain_buffer,
     _read_nonblocking,
     _wait_for_pjl_response,
@@ -17,7 +17,7 @@ from python_pkg.brother_printer.usb_query import (
     pjl_query,
 )
 
-MOD = "python_pkg.brother_printer.usb_query"
+MOD = "python_pkg.brother_printer._pjl_io"
 
 
 class TestDrainBuffer:

python_pkg/brother_printer/usb_query.py

@@ -1,18 +1,25 @@
-"""USB printer discovery and PJL query functions."""
+"""USB printer discovery and PJL query functions.
+
+The non-blocking PJL transport lives in _pjl_io.py. Its three entry points are
+imported back here because tests/test_usb_query.py patches them on this module.
+"""
 
 from __future__ import annotations
 
-import contextlib
 import errno
 import importlib
 import logging
 import os
 from pathlib import Path
-import select
 import shutil
 import time
 from typing import TYPE_CHECKING
 
+from python_pkg.brother_printer._pjl_io import (
+    _drain_buffer,
+    _write_with_deadline,
+    pjl_query,
+)
 from python_pkg.brother_printer._query import (
     printer_info_from_cups,
     run_command_text,
@@ -45,100 +52,6 @@ def find_usb_printer_dev() -> str | None:
     return str(devices[0]) if devices else None
 
 
-# ── PJL over USB ─────────────────────────────────────────────────────
-
-
-# The printer's fd is opened and kept non-blocking. usblp blocks indefinitely
-# on a plain open() and on writes whenever the printer is not ready to talk -
-# stuck mid-job, wedged, or simply asleep - so a blocking fd turns "the printer
-# is unwell" into "this tool hangs forever", which is the worst possible answer
-# when a status report is exactly what you wanted.
-
-
-def _drain_buffer(fd: int) -> None:
-    """Read and discard any stale data from the USB buffer."""
-    with contextlib.suppress(OSError):
-        while os.read(fd, 4096):
-            pass
-
-
-def _write_with_deadline(fd: int, data: bytes, deadline: float) -> bool:
-    """Write all of data to a non-blocking fd. Returns False if the deadline passes.
-
-    Args:
-        fd: Non-blocking file descriptor for the printer.
-        data: Bytes to send.
-        deadline: Absolute monotonic-ish time (time.time()) to give up at.
-
-    Returns:
-        True when everything was written, False on timeout or write error.
-    """
-    sent = 0
-    while sent < len(data):
-        remaining = deadline - time.time()
-        if remaining <= 0:
-            return False
-        try:
-            sent += os.write(fd, data[sent:])
-        except BlockingIOError:
-            # Printer's buffer is full or it is not accepting data: wait for it
-            # to become writable rather than spinning or blocking forever.
-            _, writable, _ = select.select([], [fd], [], min(remaining, 0.5))
-            if not writable:
-                continue
-        except OSError:
-            return False
-    return True
-
-
-def _read_nonblocking(fd: int) -> bytes:
-    """Read all currently available data from a non-blocking fd."""
-    data = b""
-    with contextlib.suppress(OSError):
-        while True:
-            chunk = os.read(fd, 4096)
-            if not chunk:
-                break
-            data += chunk
-    return data
-
-
-def _wait_for_pjl_response(fd: int, deadline: float) -> bytes:
-    """Poll fd until PJL data arrives or deadline expires."""
-    response = b""
-    while time.time() < deadline:
-        remaining = deadline - time.time()
-        if remaining <= 0:
-            break
-        readable, _, _ = select.select([fd], [], [], min(remaining, 1.0))
-        if readable:
-            response += _read_nonblocking(fd)
-            if response and (b"=" in response or b"@PJL" in response):
-                break
-    return response
-
-
-def pjl_query(fd: int, cmd: str, timeout_sec: float = 5.0) -> str:
-    """Send a PJL command via raw fd and read the response.
-
-    Args:
-        fd: Non-blocking file descriptor for the printer.
-        cmd: PJL command, e.g. "@PJL INFO STATUS".
-        timeout_sec: Total budget for sending the command and reading the reply.
-
-    Returns:
-        The printer's reply, or "" if it did not answer in time. A silent
-        printer is a normal outcome here, not an error: it happens whenever the
-        printer is busy digesting a job.
-    """
-    deadline = time.time() + timeout_sec
-    pjl_cmd = f"\x1b%-12345X@PJL\r\n{cmd}\r\n\x1b%-12345X"
-    if not _write_with_deadline(fd, pjl_cmd.encode(), deadline):
-        return ""
-    response = _wait_for_pjl_response(fd, deadline)
-    return response.decode("ascii", errors="replace")
-
-
 def _parse_status(resp: str, result: USBResult) -> bool:
     """Parse STATUS response into result. Returns True if code was found."""
     found = False