fix(brother_printer): truthful probe note, pyusb auto-install, stuck-job detection

The CUPS-fallback note blamed a missing pyusb when the probe was really
skipped because CUPS owned the printer. It now names the real reason.
run.sh installs python-pyusb for the system python3 that sudo runs.

A paper misfeed blocks the usb:// backend while CUPS keeps saying
"processing", and cancel leaves that backend hung so the queue never
recovers. The report now flags a job whose page counter is frozen for
15 s, and a backend of a cancelled job, with a y/N offer to SIGKILL it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AaBRWtXXWCs3K3yPHSruUC

docs/superpowers/contracts/brother-printer-stuck-job-2026-10-12.json

@@ -0,0 +1,16 @@
+{
+  "title": "brother_printer: truthful probe note, pyusb auto-install, stuck-job detection",
+  "objective": "./run.sh says why the USB probe was skipped instead of falsely blaming pyusb, installs python-pyusb itself, and reports a stalled job or a hung backend of a cancelled job instead of 'Printing / processing job'.",
+  "acceptance_criteria": [
+    "While CUPS prints, the fallback note reads 'USB status probe skipped: CUPS has a job on the printer'",
+    "run.sh checks /usr/bin/python3 for usb.core and runs pacman -S --needed python-pyusb when it is missing",
+    "A frozen job-impressions-completed counter for 15 s reports cups:job-stalled",
+    "A usb://Brother backend whose job is no longer queued reports cups:backend-orphaned and offers a SIGKILL",
+    "brother_printer tests pass at 100% branch coverage"
+  ],
+  "out_of_scope": [
+    "Physical paper-feed problems on the HL-1110",
+    "Changing the queue default resolution"
+  ],
+  "verifier": "python3 -m pytest -q -o addopts='--strict-markers --strict-config' --cov=python_pkg.brother_printer --cov-branch python_pkg/brother_printer/tests"
+}

docs/superpowers/evidence/brother-printer-stuck-job-2026-10-12.json

@@ -0,0 +1,48 @@
+{
+  "intent": "The Brother status report stops lying while a job is stuck: it names the real reason the USB probe was skipped, installs pyusb on its own, and flags a stalled job (paper misfeed) or a hung backend left behind by a cancelled job.",
+  "scope": [
+    "python_pkg/brother_printer/{_stuck_job,_cups_fallback,cups_service,constants,data_classes,display}.py, run.sh, tests",
+    "Non-goal: fixing the printer's paper pick-up, which is mechanical"
+  ],
+  "changes": [
+    "USBResult.probe_note carries _usb_probe_skip_reason(): CUPS busy / pyusb missing / printer silent; display prints it instead of 'pyusb not available'",
+    "run.sh ensure_pyusb: probes /usr/bin/python3 (what sudo runs; the user's python3 is a pyenv shim) and pacman-installs python-pyusb",
+    "_stuck_job.py: scans /proc for usb://Brother backends; orphaned backend of a cancelled job -> cups:backend-orphaned + y/N SIGKILL offer; job-impressions-completed frozen 15 s -> cups:job-stalled",
+    "41xxx status now reads 'Out of Paper / not feeding' since a misfeed sets the same paper-empty bit"
+  ],
+  "verification": [
+    {
+      "command": "./run.sh during stalled job 105 (2026-10-12 16:01)",
+      "result": "pass",
+      "evidence": "Note: USB status probe skipped: CUPS has a job on the printer; status obtained via CUPS only."
+    },
+    {
+      "command": "fake orphan (exec -a usb://Brother/... sleep 40) + display_usb_results with stdin y",
+      "result": "pass",
+      "evidence": "Queue blocked by a hung backend of a cancelled job / Kill hung backend PID 960784 ... Backend killed; process gone"
+    },
+    {
+      "command": "flag_stuck_job with frozen vs moving counter",
+      "result": "pass",
+      "evidence": "frozen: cups:job-stalled after 15.0s; moving: cleared after 1.0s"
+    },
+    {
+      "command": "python3 -m pytest -q --cov=python_pkg.brother_printer --cov-branch python_pkg/brother_printer/tests",
+      "result": "pass",
+      "evidence": "431 passed, coverage 100.00%"
+    },
+    {
+      "command": "lpr Krasnale-sie-rzadza.pdf (jobs 106, 107 pages 52-55), monitored via ipptool",
+      "result": "pass",
+      "evidence": "all 55 pages on paper, user-confirmed; printer back at 10001 Ready"
+    }
+  ],
+  "risks": [
+    "Stall window is 15 s: a healthy run froze the counter up to ~9 s, so a very slow first page could false-flag",
+    "A status run during printing now blocks up to 15 s while it samples progress"
+  ],
+  "rollback": [
+    "git revert the commit",
+    "Run ./run.sh on an idle printer and confirm Ready 10001"
+  ]
+}

python_pkg/brother_printer/_cups_fallback.py

@@ -17,6 +17,7 @@ from python_pkg.brother_printer._query import (
     printer_info_from_cups,
     run_command_text,
 )
+from python_pkg.brother_printer._stuck_job import flag_stuck_job
 from python_pkg.brother_printer.constants import (
     _CUPS_REASONS_TO_STATUS,
     _CUPS_STATE_TO_STATUS,
@@ -27,9 +28,11 @@ from python_pkg.brother_printer.constants import (
 )
 from python_pkg.brother_printer.consumables import estimate_consumable_life
 from python_pkg.brother_printer.cups_service import (
+    _cups_is_busy,
     _ensure_cups_running,
     _get_pyusb_device_info,
     _query_usb_port_status_raw,
+    _usb_probe_skip_reason,
 )
 from python_pkg.brother_printer.data_classes import (
     USBPortStatus,
@@ -201,8 +204,19 @@ def query_usb_via_cups() -> USBResult:
         result.online = "TRUE"
         return result
 
+    _apply_cups_only_status(result, printer_name, ipp)
+    return result
+
+
+def _apply_cups_only_status(
+    result: USBResult, printer_name: str, ipp: dict[str, str]
+) -> None:
+    """Fill the status from CUPS alone, when the USB port could not be read."""
+    state = ipp.get("printer-state", "")
+    result.cups.probe_note = _usb_probe_skip_reason(state)
+    if _cups_is_busy(state) and flag_stuck_job(result, printer_name):
+        return
+    reasons = ipp.get("printer-state-reasons", "none")
     result.status_code = _map_cups_to_status_code(state, reasons)
     result.display = ipp.get("printer-state-message", "")
     result.online = "TRUE" if state.lower() in {"idle", "processing"} else "FALSE"
-
-    return result

python_pkg/brother_printer/_stuck_job.py

@@ -0,0 +1,174 @@
+"""Detect a print job that CUPS calls "processing" but that is going nowhere.
+
+Two failure modes seen on the HL-1110 (2026-10-12), neither visible in
+``printer-state-reasons``:
+
+* **Stalled job** - the printer failed to pick up paper. It raises the
+  paper-empty bit, the ``usb://`` backend blocks, and CUPS keeps saying
+  "processing". Only the job's page counter freezes.
+* **Orphaned backend** - ``cancel`` removes the job but the blocked backend
+  survives, so the queue stays "now printing <queue>-0" forever and even
+  ``systemctl restart cups`` hangs. Killing the backend is the only cure.
+
+``job-impressions-completed`` counts pages the filters have *rasterised*, not
+sheets out of the printer: it runs ~5 pages ahead of the paper. That is fine
+for spotting a freeze, useless as "pages printed".
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import os
+from pathlib import Path
+import re
+import shutil
+import signal
+import time
+from typing import TYPE_CHECKING
+
+from python_pkg.brother_printer._query import run_command_text
+from python_pkg.brother_printer.constants import (
+    DERIVED_BACKEND_ORPHANED,
+    DERIVED_JOB_STALLED,
+    DIM,
+    GREEN,
+    RED,
+    RESET,
+    _out,
+    _prompt,
+)
+
+if TYPE_CHECKING:
+    from python_pkg.brother_printer.data_classes import USBResult
+
+# A healthy run on 2026-10-12 froze the counter for up to ~9 s between pages
+# (sampled every 3 s), so 15 s leaves headroom before calling it stalled.
+STALL_WINDOW_SEC = 15.0
+POLL_INTERVAL_SEC = 1.0
+PROC_DIR = Path("/proc")
+_BACKEND_PREFIX = "usb://Brother"
+_IMPRESSIONS_RE = re.compile(r"job-impressions-completed \(integer\) = (\d+)")
+
+
+@dataclass
+class StuckJob:
+    """A job (or its leftover backend) that is not making progress."""
+
+    kind: str  # "stalled" or "orphaned"
+    job_id: str
+    backend_pid: int = 0
+
+
+def find_backend_pids() -> dict[str, int]:
+    """Map CUPS job id -> PID of every running Brother ``usb://`` backend."""
+    backends: dict[str, int] = {}
+    for entry in PROC_DIR.iterdir():
+        if not entry.name.isdigit():
+            continue
+        try:
+            argv = (entry / "cmdline").read_bytes().split(b"\0")
+        except OSError:
+            continue  # process exited mid-scan, or not ours to read
+        # cmdline is NUL-terminated, so argv[1] is b"" when there is no job id.
+        if (
+            len(argv) > 1
+            and argv[1]
+            and argv[0].decode(errors="replace").startswith(_BACKEND_PREFIX)
+        ):
+            backends[argv[1].decode(errors="replace")] = int(entry.name)
+    return backends
+
+
+def _job_impressions(job_id: str) -> int | None:
+    """Read a job's rasterised-page counter, or None if CUPS will not say."""
+    ipptool_path = shutil.which("ipptool")
+    if not ipptool_path:
+        return None
+    uri = f"ipp://localhost/jobs/{job_id}"
+    output = run_command_text([ipptool_path, "-tv", uri, "get-job-attributes.test"])
+    match = _IMPRESSIONS_RE.search(output)
+    return int(match.group(1)) if match else None
+
+
+def _job_is_stalled(job_id: str) -> bool:
+    """Watch the page counter; True if it does not move within the window."""
+    start = _job_impressions(job_id)
+    if start is None:
+        return False
+    deadline = time.monotonic() + STALL_WINDOW_SEC
+    while time.monotonic() < deadline:
+        time.sleep(POLL_INTERVAL_SEC)
+        if _job_impressions(job_id) != start:
+            return False
+    return True
+
+
+def detect_stuck_job(active_job_ids: list[str]) -> StuckJob | None:
+    """Find an orphaned backend or a stalled active job.
+
+    Args:
+        active_job_ids: numeric ids of the jobs CUPS still lists for the queue.
+    """
+    backends = find_backend_pids()
+    for job_id, pid in sorted(backends.items()):
+        if job_id not in active_job_ids:
+            return StuckJob("orphaned", job_id, pid)
+    if active_job_ids and _job_is_stalled(active_job_ids[0]):
+        job_id = active_job_ids[0]
+        return StuckJob("stalled", job_id, backends.get(job_id, 0))
+    return None
+
+
+def kill_backend(pid: int) -> bool:
+    """SIGKILL a hung backend; SIGTERM is what it already ignored."""
+    try:
+        os.kill(pid, signal.SIGKILL)
+    except OSError:
+        return False
+    return True
+
+
+def offer_backend_kill(pid: int) -> None:
+    """Ask before killing an orphaned backend; nothing else unblocks the queue."""
+    _out()
+    choice = _prompt(f"  Kill hung backend PID {pid} to unblock the queue? [y/N]: ")
+    if choice.strip().lower() != "y":
+        _out(f"  {DIM}No changes made.{RESET}")
+    elif kill_backend(pid):
+        _out(f"  {GREEN}✓ Backend killed. CUPS will return to idle.{RESET}")
+    else:
+        _out(f"  {RED}Could not kill PID {pid} (already gone, or not root).{RESET}")
+
+
+def _active_job_ids(printer_name: str) -> list[str]:
+    """Numeric ids of the jobs CUPS still lists for the queue."""
+    lpstat_path = shutil.which("lpstat")
+    if not lpstat_path:
+        return []
+    output = run_command_text([lpstat_path, "-o", printer_name])
+    prefix = f"{printer_name}-"
+    return [
+        line.split()[0].removeprefix(prefix)
+        for line in output.splitlines()
+        if line.startswith(prefix)
+    ]
+
+
+def flag_stuck_job(result: USBResult, printer_name: str) -> bool:
+    """Replace a misleading "printing" status when the job is going nowhere."""
+    stuck = detect_stuck_job(_active_job_ids(printer_name))
+    if stuck is None:
+        return False
+    result.online = "TRUE"
+    if stuck.kind == "orphaned":
+        result.status_code = DERIVED_BACKEND_ORPHANED
+        result.display = (
+            f"backend PID {stuck.backend_pid} of cancelled job {stuck.job_id}"
+            ' still running (a leftover "filter failed" message is just'
+            " the cancel killing Ghostscript)"
+        )
+        result.cups.orphaned_backend_pid = stuck.backend_pid
+    else:
+        result.status_code = DERIVED_JOB_STALLED
+        result.display = f"job {stuck.job_id} has not advanced a page in 15 s"
+    return True

python_pkg/brother_printer/constants.py

@@ -125,11 +125,14 @@ BROTHER_STATUS_CODES: dict[int, tuple[str, str, str]] = {
 # on a single-tray A4 printer).  So match the family and ignore the digits.
 # Each entry is an inclusive low/high bound plus the usual status triple.
 
+_PAPER_ACTION = "Load paper; if the tray is full, fan and reseat the stack."
+
 _STATUS_CODE_RANGES: tuple[tuple[int, int, tuple[str, str, str]], ...] = (
     (
         41000,
         41999,
-        ("critical", "Out of Paper", "Load paper into the paper tray."),
+        # A failed pick-up raises the same paper-empty bit as an empty tray.
+        ("critical", "Out of Paper / not feeding", _PAPER_ACTION),
     ),
     (
         50000,
@@ -188,6 +191,8 @@ _ERROR_REASON_MAP: tuple[tuple[tuple[str, ...], str, str], ...] = (
 DERIVED_TONER_END = "estimate:toner-end"
 DERIVED_TONER_LOW = "estimate:toner-low"
 DERIVED_CUPS_ERROR = "cups:error"
+DERIVED_JOB_STALLED = "cups:job-stalled"
+DERIVED_BACKEND_ORPHANED = "cups:backend-orphaned"
 
 _DERIVED_STATUS_CODES: dict[str, tuple[str, str, str]] = {
     DERIVED_TONER_END: (
@@ -201,6 +206,16 @@ _DERIVED_STATUS_CODES: dict[str, tuple[str, str, str]] = {
         "Printer reports an error",
         "Check the printer display for details.",
     ),
+    DERIVED_JOB_STALLED: (
+        "critical",
+        "Print job stalled (no page progress in 15 s)",
+        "Paper likely failed to feed: reseat it or power-cycle the printer.",
+    ),
+    DERIVED_BACKEND_ORPHANED: (
+        "critical",
+        "Queue blocked by a hung backend of a cancelled job",
+        "Kill the hung backend (offered below); CUPS recovers on its own.",
+    ),
 }
 
 

python_pkg/brother_printer/cups_service.py

@@ -156,6 +156,17 @@ def _cups_is_busy(cups_state: str) -> bool:
     return "processing" in cups_state.lower() or "printing" in cups_state.lower()
 
 
+def _usb_probe_skip_reason(cups_state: str) -> str:
+    """Explain why _query_usb_port_status_raw returned None."""
+    if _cups_is_busy(cups_state):
+        return "USB status probe skipped: CUPS has a job on the printer"
+    try:
+        _import_or_raise("usb.core")
+    except ImportError:
+        return "pyusb not installed (run.sh installs it)"
+    return "printer did not answer the USB status probe"
+
+
 def _query_usb_port_status_raw(cups_state: str = "") -> USBPortStatus | None:
     """Query the printer's port status without disturbing it.
 

python_pkg/brother_printer/data_classes.py

@@ -72,6 +72,16 @@ class USBPortStatus:
     raw_byte: int = 0
 
 
+@dataclass
+class CupsDiagnosis:
+    """What the CUPS fallback learned beyond the status code itself."""
+
+    # Why port_status is None, in words for the report. Empty when it was read.
+    probe_note: str = ""
+    # PID of a usb:// backend still running for a cancelled job; 0 if none.
+    orphaned_backend_pid: int = 0
+
+
 @dataclass
 class USBResult:
     """Result from a USB PJL query."""
@@ -90,6 +100,7 @@ class USBResult:
     # (CUPS fallback mode), which forces the estimate onto the CUPS page log.
     page_count: str = ""
     port_status: USBPortStatus | None = None
+    cups: CupsDiagnosis = field(default_factory=CupsDiagnosis)
 
 
 @dataclass

python_pkg/brother_printer/display.py

@@ -16,6 +16,7 @@ from python_pkg.brother_printer._severity import (
     _SEVERITY_ICONS,
     _SEVERITY_SUMMARIES,
 )
+from python_pkg.brother_printer._stuck_job import offer_backend_kill
 from python_pkg.brother_printer._supply import _collect_supply_items
 from python_pkg.brother_printer.constants import (
     BOLD,
@@ -120,9 +121,9 @@ def _display_cups_fallback_note(result: USBResult) -> None:
         )
     else:
         _out(
-            f"  {DIM}Note: pyusb not available; status obtained via"
-            f" CUPS only. Detailed toner/drum levels are not"
-            f" available in this mode.{RESET}"
+            f"  {DIM}Note: {result.cups.probe_note or 'USB status probe failed'};"
+            f" status obtained via CUPS only. Detailed toner/drum levels"
+            f" are not available in this mode.{RESET}"
         )
 
 
@@ -141,6 +142,8 @@ def display_usb_results(result: USBResult) -> None:
 
     if result.device == "cups":
         _display_cups_fallback_note(result)
+        if result.cups.orphaned_backend_pid:
+            offer_backend_kill(result.cups.orphaned_backend_pid)
 
     _out()
     printer_total = int(result.page_count) if result.page_count.isdigit() else 0

python_pkg/brother_printer/run.sh

@@ -3,8 +3,8 @@ set -euo pipefail
 SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
 
-# No pip dependencies — script only uses stdlib (+pyusb for fallback info)
-# Requires root for USB hardware queries and CUPS management
+# Depends on pyusb (USB port-status probe when CUPS owns the printer) and runs
+# as root for USB hardware queries and CUPS management.
 #
 # Imports are absolute (python_pkg.brother_printer.*), so the monorepo root must
 # be importable. sudo's env_reset drops an exported PYTHONPATH, hence passing it
@@ -13,7 +13,21 @@ REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
 # Usage: ./run.sh              # auto-detect
 #        ./run.sh <printer_ip> # network/SNMP mode
 
+# The tool runs under the system python3 (sudo secure_path), so pyusb must come
+# from the distro package, not a venv.
+ensure_pyusb() {
+	/usr/bin/python3 -c 'import usb.core' 2>/dev/null && return 0
+	if ! command -v pacman >/dev/null; then
+		echo "pyusb missing and pacman not found; install python-pyusb for system python3." >&2
+		exit 1
+	fi
+	echo "Installing python-pyusb..."
+	sudo pacman -S --needed --noconfirm python-pyusb
+}
+
+ensure_pyusb
+
 if [[ $EUID -ne 0 ]]; then
-    exec sudo PYTHONPATH="$REPO_ROOT" python3 -m python_pkg.brother_printer "$@"
+	exec sudo PYTHONPATH="$REPO_ROOT" python3 -m python_pkg.brother_printer "$@"
 fi
 exec env PYTHONPATH="$REPO_ROOT" python3 -m python_pkg.brother_printer "$@"

python_pkg/brother_printer/tests/test_constants.py

@@ -138,7 +138,7 @@ class TestGetStatusInfo:
         """
         severity, text, _ = get_status_info("41213")
         assert severity == "critical"
-        assert text == "Out of Paper"
+        assert text == "Out of Paper / not feeding"
 
     def test_derived_codes_are_not_pjl(self) -> None:
         """States we infer ourselves must not masquerade as printer reports."""

python_pkg/brother_printer/tests/test_cups_fallback_query.py

@@ -100,3 +100,34 @@ class TestQueryUsbViaCups:
             result = query_usb_via_cups()
             assert result.status_code == "41000"
             assert result.online == "FALSE"
+
+
+class TestQueryUsbViaCupsStuckJob:
+    """A busy queue whose job goes nowhere must not read as "printing"."""
+
+    def _query(self, *, stuck: bool) -> tuple[object, MagicMock]:
+        with (
+            patch(f"{MOD}._ensure_cups_running", return_value=True),
+            patch(f"{MOD}.find_cups_printer_name", return_value="Brother"),
+            patch(f"{MOD}._get_pyusb_device_info", return_value={}),
+            patch(f"{MOD}.printer_info_from_cups", return_value={}),
+            patch(
+                f"{MOD}._get_cups_ipp_status",
+                return_value={"printer-state": "processing"},
+            ),
+            patch(f"{MOD}._get_cups_economode", return_value=""),
+            patch(f"{MOD}._query_usb_port_status_raw", return_value=None),
+            patch(f"{MOD}.flag_stuck_job", return_value=stuck) as flag,
+        ):
+            return query_usb_via_cups(), flag
+
+    def test_stuck_job_short_circuits(self) -> None:
+        result, flag = self._query(stuck=True)
+        flag.assert_called_once()
+        assert result.status_code == ""  # left for flag_stuck_job to set
+
+    def test_progressing_job_reads_as_printing(self) -> None:
+        result, flag = self._query(stuck=False)
+        flag.assert_called_once()
+        assert result.status_code == "10023"
+        assert "CUPS has a job" in result.cups.probe_note

python_pkg/brother_printer/tests/test_display.py

@@ -8,6 +8,7 @@ from unittest.mock import MagicMock, patch
 import pytest
 
 from python_pkg.brother_printer.data_classes import (
+    CupsDiagnosis,
     USBPortStatus,
     USBResult,
 )
@@ -157,11 +158,17 @@ class TestDisplayCupsFallbackNote:
             _display_cups_fallback_note(r)
             assert "USB port query" in out.getvalue()
 
-    def test_without_port_status(self) -> None:
+    def test_without_port_status_shows_reason(self) -> None:
+        r = USBResult(cups=CupsDiagnosis(probe_note="pyusb not installed"))
+        with patch("sys.stdout", new_callable=StringIO) as out:
+            _display_cups_fallback_note(r)
+            assert "Note: pyusb not installed;" in out.getvalue()
+
+    def test_without_port_status_or_reason(self) -> None:
         r = USBResult(port_status=None)
         with patch("sys.stdout", new_callable=StringIO) as out:
             _display_cups_fallback_note(r)
-            assert "pyusb not available" in out.getvalue()
+            assert "USB status probe failed" in out.getvalue()
 
 
 class TestDisplayUsbResults:
@@ -190,10 +197,29 @@ class TestDisplayUsbResults:
             patch(f"{MOD}.get_cups_queue_status"),
             patch(f"{MOD}.display_cups_queue_status"),
             patch(f"{MOD}._display_cups_fallback_note") as mock_fallback,
+            patch(f"{MOD}.offer_backend_kill") as mock_kill,
             patch("sys.stdout", new_callable=StringIO),
         ):
             display_usb_results(r)
             mock_fallback.assert_called_once()
+            mock_kill.assert_not_called()
+
+    def test_cups_device_offers_backend_kill(self) -> None:
+        r = USBResult(device="cups", cups=CupsDiagnosis(orphaned_backend_pid=4242))
+        with (
+            patch(f"{MOD}._display_report_header"),
+            patch(f"{MOD}._display_usb_device_info"),
+            patch(f"{MOD}._display_pjl_status"),
+            patch(f"{MOD}._display_page_count_estimate"),
+            patch(f"{MOD}._display_consumables_reference"),
+            patch(f"{MOD}.get_cups_queue_status"),
+            patch(f"{MOD}.display_cups_queue_status"),
+            patch(f"{MOD}._display_cups_fallback_note"),
+            patch(f"{MOD}.offer_backend_kill") as mock_kill,
+            patch("sys.stdout", new_callable=StringIO),
+        ):
+            display_usb_results(r)
+            mock_kill.assert_called_once_with(4242)
 
     def test_error(self) -> None:
         r = USBResult(error="fail")

python_pkg/brother_printer/tests/test_stuck_job.py

@@ -0,0 +1,196 @@
+"""Tests for brother_printer._stuck_job - stalled jobs and orphaned backends.
+
+/proc, ipptool, lpstat, time and os.kill are all patched: nothing here reads
+the live process table, talks to CUPS, sleeps, or signals a real process.
+"""
+
+from __future__ import annotations
+
+from io import StringIO
+from pathlib import Path
+from typing import TYPE_CHECKING
+from unittest.mock import MagicMock, patch
+
+from python_pkg.brother_printer import _stuck_job
+from python_pkg.brother_printer._stuck_job import (
+    StuckJob,
+    detect_stuck_job,
+    find_backend_pids,
+    flag_stuck_job,
+    kill_backend,
+    offer_backend_kill,
+)
+from python_pkg.brother_printer.constants import (
+    DERIVED_BACKEND_ORPHANED,
+    DERIVED_JOB_STALLED,
+)
+from python_pkg.brother_printer.data_classes import USBResult
+
+if TYPE_CHECKING:
+    import pytest
+
+MOD = "python_pkg.brother_printer._stuck_job"
+
+
+def _proc(tmp_path: Path, pid: str, cmdline: bytes | None) -> None:
+    """Create a fake /proc/<pid>; cmdline None means it is unreadable."""
+    entry = tmp_path / pid
+    entry.mkdir()
+    if cmdline is not None:
+        (entry / "cmdline").write_bytes(cmdline)
+
+
+class TestFindBackendPids:
+    def test_maps_job_id_to_pid(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _proc(tmp_path, "101", b"usb://Brother/HL-1110\x00105\x00kuhy\x00")
+        _proc(tmp_path, "102", b"/usr/bin/gs\x00-dQUIET\x00")
+        _proc(tmp_path, "103", b"usb://Brother/HL-1110\x00")  # no job argument
+        _proc(tmp_path, "104", None)  # exited mid-scan
+        (tmp_path / "self").mkdir()  # non-numeric entry
+        monkeypatch.setattr(_stuck_job, "PROC_DIR", tmp_path)
+        assert find_backend_pids() == {"105": 101}
+
+
+class TestJobImpressions:
+    @patch(f"{MOD}.shutil.which", return_value=None)
+    def test_no_ipptool(self, w: MagicMock) -> None:
+        assert _stuck_job._job_impressions("105") is None
+
+    @patch(f"{MOD}.run_command_text")
+    @patch(f"{MOD}.shutil.which", return_value="/usr/bin/ipptool")
+    def test_parses_counter(self, w: MagicMock, run: MagicMock) -> None:
+        run.return_value = "        job-impressions-completed (integer) = 7\n"
+        assert _stuck_job._job_impressions("105") == 7
+        assert "ipp://localhost/jobs/105" in run.call_args.args[0]
+
+    @patch(f"{MOD}.run_command_text", return_value="")
+    @patch(f"{MOD}.shutil.which", return_value="/usr/bin/ipptool")
+    def test_counter_absent(self, w: MagicMock, r: MagicMock) -> None:
+        assert _stuck_job._job_impressions("105") is None
+
+
+class TestJobIsStalled:
+    @patch(f"{MOD}._job_impressions", return_value=None)
+    def test_unknown_counter_is_not_stalled(self, i: MagicMock) -> None:
+        assert _stuck_job._job_is_stalled("105") is False
+
+    @patch(f"{MOD}.time")
+    @patch(f"{MOD}._job_impressions", side_effect=[7, 7, 8])
+    def test_progress_clears_it(self, i: MagicMock, t: MagicMock) -> None:
+        t.monotonic.side_effect = [0.0, 0.0, 1.0]
+        assert _stuck_job._job_is_stalled("105") is False
+
+    @patch(f"{MOD}.time")
+    @patch(f"{MOD}._job_impressions", return_value=7)
+    def test_frozen_counter_is_stalled(self, i: MagicMock, t: MagicMock) -> None:
+        t.monotonic.side_effect = [0.0, 0.0, 10.0, 15.0]
+        assert _stuck_job._job_is_stalled("105") is True
+        assert t.sleep.call_count == 2
+
+
+class TestDetectStuckJob:
+    @patch(f"{MOD}._job_is_stalled")
+    @patch(f"{MOD}.find_backend_pids", return_value={"104": 820758})
+    def test_orphaned_backend(self, f: MagicMock, stalled: MagicMock) -> None:
+        assert detect_stuck_job([]) == StuckJob("orphaned", "104", 820758)
+        stalled.assert_not_called()
+
+    @patch(f"{MOD}._job_is_stalled", return_value=True)
+    @patch(f"{MOD}.find_backend_pids", return_value={"105": 884180})
+    def test_stalled_job(self, f: MagicMock, s: MagicMock) -> None:
+        assert detect_stuck_job(["105"]) == StuckJob("stalled", "105", 884180)
+
+    @patch(f"{MOD}._job_is_stalled", return_value=True)
+    @patch(f"{MOD}.find_backend_pids", return_value={})
+    def test_stalled_before_backend_starts(self, f: MagicMock, s: MagicMock) -> None:
+        assert detect_stuck_job(["105"]) == StuckJob("stalled", "105", 0)
+
+    @patch(f"{MOD}._job_is_stalled", return_value=False)
+    @patch(f"{MOD}.find_backend_pids", return_value={"105": 884180})
+    def test_healthy_job(self, f: MagicMock, s: MagicMock) -> None:
+        assert detect_stuck_job(["105"]) is None
+
+    @patch(f"{MOD}.find_backend_pids", return_value={})
+    def test_nothing_running(self, f: MagicMock) -> None:
+        assert detect_stuck_job([]) is None
+
+
+class TestKillBackend:
+    @patch(f"{MOD}.os.kill")
+    def test_sigkill(self, kill: MagicMock) -> None:
+        assert kill_backend(42) is True
+        kill.assert_called_once_with(42, _stuck_job.signal.SIGKILL)
+
+    @patch(f"{MOD}.os.kill", side_effect=ProcessLookupError)
+    def test_already_gone(self, k: MagicMock) -> None:
+        assert kill_backend(42) is False
+
+
+class TestOfferBackendKill:
+    def _offer(self, answer: str, *, killed: bool = True) -> tuple[str, MagicMock]:
+        with (
+            patch(f"{MOD}._prompt", return_value=answer),
+            patch(f"{MOD}.kill_backend", return_value=killed) as kill,
+            patch("sys.stdout", new_callable=StringIO) as out,
+        ):
+            offer_backend_kill(42)
+            return out.getvalue(), kill
+
+    def test_declined(self) -> None:
+        out, kill = self._offer("")
+        kill.assert_not_called()
+        assert "No changes made" in out
+
+    def test_accepted(self) -> None:
+        out, kill = self._offer(" Y ")
+        kill.assert_called_once_with(42)
+        assert "Backend killed" in out
+
+    def test_kill_fails(self) -> None:
+        out, _kill = self._offer("y", killed=False)
+        assert "Could not kill PID 42" in out
+
+
+class TestActiveJobIds:
+    @patch(f"{MOD}.shutil.which", return_value=None)
+    def test_no_lpstat(self, w: MagicMock) -> None:
+        assert _stuck_job._active_job_ids("Brother") == []
+
+    @patch(f"{MOD}.run_command_text")
+    @patch(f"{MOD}.shutil.which", return_value="/usr/bin/lpstat")
+    def test_parses_ids(self, w: MagicMock, run: MagicMock) -> None:
+        run.return_value = (
+            "Brother-105 kuhy 160768 Mon 12 Oct 2026\nOther-7 kuhy 1 Mon 12 Oct 2026\n"
+        )
+        assert _stuck_job._active_job_ids("Brother") == ["105"]
+
+
+class TestFlagStuckJob:
+    @patch(f"{MOD}._active_job_ids", return_value=[])
+    @patch(f"{MOD}.detect_stuck_job", return_value=None)
+    def test_nothing_stuck(self, d: MagicMock, a: MagicMock) -> None:
+        result = USBResult(device="cups")
+        assert flag_stuck_job(result, "Brother") is False
+        assert result.status_code == ""
+
+    @patch(f"{MOD}._active_job_ids", return_value=[])
+    @patch(f"{MOD}.detect_stuck_job")
+    def test_orphaned(self, detect: MagicMock, a: MagicMock) -> None:
+        detect.return_value = StuckJob("orphaned", "104", 820758)
+        result = USBResult(device="cups")
+        assert flag_stuck_job(result, "Brother") is True
+        assert result.status_code == DERIVED_BACKEND_ORPHANED
+        assert result.cups.orphaned_backend_pid == 820758
+        assert "cancelled job 104" in result.display
+
+    @patch(f"{MOD}._active_job_ids", return_value=["105"])
+    @patch(f"{MOD}.detect_stuck_job")
+    def test_stalled(self, detect: MagicMock, a: MagicMock) -> None:
+        detect.return_value = StuckJob("stalled", "105", 884180)
+        result = USBResult(device="cups")
+        assert flag_stuck_job(result, "Brother") is True
+        assert result.status_code == DERIVED_JOB_STALLED
+        assert result.cups.orphaned_backend_pid == 0
+        assert "job 105" in result.display

python_pkg/brother_printer/tests/test_usb_probe_reason.py

@@ -0,0 +1,25 @@
+"""Tests for cups_service._usb_probe_skip_reason - the fallback note's reason."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+from python_pkg.brother_printer.cups_service import _usb_probe_skip_reason
+
+MOD = "python_pkg.brother_printer.cups_service"
+
+
+class TestUsbProbeSkipReason:
+    """The note must name the real reason, not blame a pyusb that is there."""
+
+    def test_busy_cups_wins(self) -> None:
+        reason = _usb_probe_skip_reason("processing")
+        assert "CUPS has a job" in reason
+
+    @patch(f"{MOD}._import_or_raise", side_effect=ImportError("usb.core"))
+    def test_pyusb_missing(self, m: MagicMock) -> None:
+        assert "pyusb not installed" in _usb_probe_skip_reason("idle")
+
+    @patch(f"{MOD}._import_or_raise", return_value=MagicMock())
+    def test_printer_silent(self, m: MagicMock) -> None:
+        assert "did not answer" in _usb_probe_skip_reason("idle")