docs(hardcover): list the API key scopes Shelfmark needs (#1243)

Hardcover's August 2026 token system replaced blanket access with
per-token scopes, and nothing in the docs said which ones Shelfmark
actually uses. A key missing write:library or write:lists still passes
Test Connection -- the reading-status and auto-remove-on-download calls
just fail silently afterwards.

Verified against a live hc_pat_ key: every scope in the table backs a
query or mutation the provider really issues, and the omitted ones
(journal, goals, reviews, prompts, notifications, account) are absent
from the provider entirely.

Refs #1240

docs/environment-variables.md

@@ -1977,7 +1977,7 @@ Move deletes the job from your usenet client after import; Copy keeps it in the
 | Variable | Description | Type | Default |
 |----------|-------------|------|---------|
 | `HARDCOVER_ENABLED` | Enable Hardcover as a metadata provider for book searches | boolean | `false` |
-| `HARDCOVER_API_KEY` | Get your API key from hardcover.app/account/api | string (secret) | _none_ |
+| `HARDCOVER_API_KEY` | Get your API key from hardcover.app/account/api (starts with hc_pat_) | string (secret) | _none_ |
 | `HARDCOVER_DEFAULT_SORT` | Default sort order for Hardcover search results. | string (choice) | `relevance` |
 | `HARDCOVER_EXCLUDE_COMPILATIONS` | Filter out compilations, anthologies, and omnibus editions from search results | boolean | `false` |
 | `HARDCOVER_EXCLUDE_UNRELEASED` | Filter out books with a release year in the future | boolean | `false` |
@@ -1999,7 +1999,7 @@ Enable Hardcover as a metadata provider for book searches
 
 **API Key**
 
-Get your API key from hardcover.app/account/api
+Get your API key from hardcover.app/account/api (starts with hc_pat_)
 
 - **Type:** string (secret)
 - **Default:** _none_
@@ -2304,6 +2304,7 @@ Override destination based on content type metadata.
 | `EXT_BYPASSER_URL` | URL of the external bypasser service (e.g., FlareSolverr). | string | `http://flaresolverr:8191` |
 | `EXT_BYPASSER_PATH` | API path for the external bypasser. | string | `/v1` |
 | `EXT_BYPASSER_TIMEOUT` | Timeout for external bypasser requests in milliseconds. | number | `60000` |
+| `BYPASS_BROWSER_IDLE_TIMEOUT` | How long the bypass helper process may sit unused before it is shut down. Higher keeps more searches fast, lower frees memory sooner. | number | `180` |
 
 <details>
 <summary>Detailed descriptions</summary>
@@ -2359,6 +2360,17 @@ Timeout for external bypasser requests in milliseconds.
 - **Requires restart:** Yes
 - **Constraints:** min: 10000, max: 300000
 
+#### `BYPASS_BROWSER_IDLE_TIMEOUT`
+
+**Bypasser Idle Timeout (seconds)**
+
+How long the bypass helper process may sit unused before it is shut down. Higher keeps more searches fast, lower frees memory sooner.
+
+- **Type:** number
+- **Default:** `180`
+- **Requires restart:** Yes
+- **Constraints:** min: 30, max: 3600
+
 </details>
 
 ### Direct Download: Mirrors

pyproject.toml

@@ -19,7 +19,7 @@ dependencies = [
     "psutil",
     "emoji",
     "rarfile",
-    "qbittorrent-api>=2026.8.0",
+    "qbittorrent-api>=2026.8.1",
     "transmission-rpc",
     "authlib>=1.7.2,<1.8",
     "apprise>=1.12.0",

readme.md

@@ -95,6 +95,30 @@ volumes:
 - Aggregates releases from multiple configured sources
 - Full audiobook support
 
+### Hardcover API Key
+
+Hardcover powers metadata search in Universal mode. Create a token at
+[hardcover.app/account/api](https://hardcover.app/account/api) — current keys start with `hc_pat_`
+and are far shorter than the JWTs Hardcover issued before August 2026.
+
+Tick these seven scopes on the token screen:
+
+| Scope | Used for |
+|-------|----------|
+| `read:catalog` | Metadata search, plus book, edition, author and series lookups |
+| `read:library` | Your reading status and shelf counts |
+| `read:lists` | Your lists and the books on them |
+| `read:me:content` | Test Connection and the "Connected as" label |
+| `read:users` | Usernames shown alongside lists |
+| `write:library` | Setting a book's reading status from Shelfmark |
+| `write:lists` | Adding and removing books from lists, including auto-remove on download |
+
+The two `write:` scopes matter only if you set reading status from Shelfmark or leave
+**Auto-Remove from List on Download** enabled (it is on by default) — without them those actions
+fail silently. Everything else Hardcover offers (journal, goals, reviews, prompts, notifications,
+account) can stay unticked. The `all` scope works too, but it grants full account access including
+deletion, so prefer the list above.
+
 ### Environment Variables
 
 Environment variables work for initial setup and Docker deployments. They serve as defaults that can be overridden in the web interface.

shelfmark/bypass/challenge.py

@@ -0,0 +1,52 @@
+"""Challenge-page detection shared by the bypassers and the HTTP retry path.
+
+Kept out of `internal_bypasser` so the HTTP layer can recognise an interstitial
+without importing SeleniumBase: that module is imported lazily precisely because its
+browser dependencies are optional, and external-bypasser setups run without them.
+"""
+
+# Matched against lowercased text, so every entry must be lowercase.
+CLOUDFLARE_INDICATORS = [
+    "just a moment",
+    "verify you are human",
+    "verifying you are human",
+    "cloudflare.com/products/turnstile",
+]
+
+DDOS_GUARD_INDICATORS = [
+    "ddos-guard",
+    "ddos guard",
+    "checking your browser before accessing",
+    "complete the manual check to continue",
+    "could not verify your browser automatically",
+]
+
+# Markers that exist only in raw markup: the bypassers scan rendered innerText, where
+# a script src or a <title> never appears. The title match is scoped to the tag on
+# purpose - hosts word the rest of that sentence differently, and matching "checking
+# your browser" as free text would trip on any page that merely discusses a challenge.
+_RAW_HTML_MARKERS = (
+    "<title>checking your browser",
+    "/cdn-cgi/challenge-platform",
+    "/.well-known/ddos-guard/",
+)
+
+# An interstitial is a few KB of markup. Past that it is a real page that happens to
+# mention a marker - a protected site links its own DDoS-Guard endpoints on every page.
+MAX_CHALLENGE_HTML_CHARS = 64 * 1024
+
+
+def challenge_marker(html: str) -> str | None:
+    """Return the marker proving `html` is an unsolved challenge page, or None.
+
+    Only meaningful for a response that already carries a challenge status: the
+    markers appear on protected sites' real pages too, so the status is what
+    separates "blocked" from "served".
+    """
+    if not html or len(html) > MAX_CHALLENGE_HTML_CHARS:
+        return None
+    lowered = html.lower()
+    for marker in (*_RAW_HTML_MARKERS, *DDOS_GUARD_INDICATORS, *CLOUDFLARE_INDICATORS):
+        if marker in lowered:
+            return marker
+    return None

shelfmark/bypass/internal_bypasser.py

@@ -27,6 +27,7 @@ from seleniumbase import cdp_driver
 from seleniumbase.undetected.cdp_driver.connection import ProtocolException
 
 from shelfmark.bypass import BypassCancelledError
+from shelfmark.bypass.challenge import CLOUDFLARE_INDICATORS, DDOS_GUARD_INDICATORS
 from shelfmark.bypass.cookie_store import (
     clear_cf_cookies,
     export_store,
@@ -57,28 +58,32 @@ _LOADING_BODY_LENGTH_MAX = 50
 _PAGE_BODY_PREVIEW_CHARS = 500
 _BROWSER_START_TIMEOUT_SECONDS = 45.0
 _BYPASS_SUBPROCESS_TIMEOUT_SECONDS = 420.0
-# Same budget as the Docker helper process, applied to the in-process CDP path so both
-# branches of get() are bounded the same way.
-_IN_PROCESS_BYPASS_TIMEOUT_SECONDS = _BYPASS_SUBPROCESS_TIMEOUT_SECONDS
+# How long a cancelled bypass may take to close its browser before the calling thread
+# stops waiting for it. Counted on top of the bypass deadline, so every budget below is
+# set to leave room for it.
+_CDP_UNWIND_GRACE_SECONDS = 15.0
+# Same wall-clock budget as the Docker helper process, applied to the in-process CDP path
+# so both branches of get() are bounded the same way: the deadline plus the unwind grace
+# comes to _BYPASS_SUBPROCESS_TIMEOUT_SECONDS either way.
+_IN_PROCESS_BYPASS_TIMEOUT_SECONDS = _BYPASS_SUBPROCESS_TIMEOUT_SECONDS - _CDP_UNWIND_GRACE_SECONDS
 _BYPASS_CHILD_ENV = "SHELFMARK_INTERNAL_BYPASSER_CHILD"
+# The helper bounds each bypass below the parent's deadline, so it is the side that gives
+# up first: it still gets to report the timeout and close its browser, and stays available
+# for the next request. A parent that hit its deadline first could only kill the helper,
+# throwing away a process the next request would have to start again. The 30s covers the
+# unwind grace as well, so a helper that times out and closes its browser as slowly as it
+# is allowed to still answers with 15s to spare.
+_CHILD_BYPASS_TIMEOUT_SECONDS = _BYPASS_SUBPROCESS_TIMEOUT_SECONDS - 30.0
+# The helper publishes its answer by writing the result file the request named, so the
+# parent waits by watching for that file rather than by reading a stream it would have to
+# demultiplex from the helper's log output.
+_HELPER_RESULT_POLL_SECONDS = 0.05
+# Closing the helper's stdin asks it to shut down; this is how long it may take to finish
+# what it is doing and exit before its session is killed instead.
+_HELPER_SHUTDOWN_GRACE_SECONDS = 15.0
+_HELPER_IDLE_TIMEOUT_DEFAULT = 180.0
 _PARENT_WATCHDOG_INTERVAL_SECONDS = 5.0
 
-# Challenge detection indicators
-CLOUDFLARE_INDICATORS = [
-    "just a moment",
-    "verify you are human",
-    "verifying you are human",
-    "cloudflare.com/products/turnstile",
-]
-
-DDOS_GUARD_INDICATORS = [
-    "ddos-guard",
-    "ddos guard",
-    "checking your browser before accessing",
-    "complete the manual check to continue",
-    "could not verify your browser automatically",
-]
-
 
 class _DisplayState(TypedDict):
     ffmpeg: subprocess.Popen[bytes] | None
@@ -222,14 +227,33 @@ class _CdpWorker:
             msg = "CDP worker loop failed to start"
             raise RuntimeError(msg)
 
+    @staticmethod
+    async def _bounded(coro: Any, timeout: float | None) -> Any:
+        """Run the coroutine under its deadline, on the loop that owns it.
+
+        The deadline has to be enforced from inside the loop rather than by the calling
+        thread: asyncio.wait_for() cancels the bypass and then *waits for it to unwind*,
+        so `finally: await _close_cdp_driver(driver)` has finished by the time this
+        raises. Cancelling from outside returns the moment the cancellation is scheduled,
+        which in a helper serving many requests let the abandoned bypass close its browser
+        while the next one was already opening its own - on the same loop, sharing the
+        DISPLAY globals and one process group.
+        """
+        if timeout is None:
+            return await coro
+        return await asyncio.wait_for(coro, timeout)
+
     def run(self, coro: Any, timeout: float | None = None) -> Any:
         self.start()
         if not self._loop or self._loop.is_closed():
             msg = "CDP worker loop not available"
             raise RuntimeError(msg)
-        future = asyncio.run_coroutine_threadsafe(coro, self._loop)
+        future = asyncio.run_coroutine_threadsafe(self._bounded(coro, timeout), self._loop)
+        # Backstop for an unwind that wedges too: _close_cdp_driver awaits websockets that
+        # a dead browser may never answer, and _bounded cannot outlive its own cleanup.
+        wait_for = None if timeout is None else timeout + _CDP_UNWIND_GRACE_SECONDS
         try:
-            return future.result(timeout=timeout)
+            return future.result(timeout=wait_for)
         except TimeoutError:
             # Otherwise the coroutine keeps running in the worker loop after we stop
             # waiting, holding the browser and racing the next bypass.
@@ -820,13 +844,21 @@ def _run_bypass_in_current_process(url: str, retry: int, cancel_flag: Event | No
             if driver:
                 await _close_cdp_driver(driver)
 
-    if os.environ.get(_BYPASS_CHILD_ENV) == "1":
-        return asyncio.run(_run_bypass())
-    # Bound the wait: this path runs in-process (non-Docker installs), holds the module-wide
-    # LOCKED for its whole duration, and neither page.get() nor page.wait() has a timeout of
-    # its own. Without a deadline here a single wedged CDP session blocks every subsequent
-    # bypass in the process forever.
-    return _CDP_WORKER.run(_run_bypass(), timeout=_IN_PROCESS_BYPASS_TIMEOUT_SECONDS)
+    # Bound the wait: this holds the module-wide LOCKED for its whole duration, and neither
+    # page.get() nor page.wait() has a timeout of its own. Without a deadline here a single
+    # wedged CDP session blocks every subsequent bypass in the process forever.
+    #
+    # The helper goes through the worker too, rather than asyncio.run: that owns a loop for
+    # one call and closes it on the way out, so a helper serving many requests would build
+    # and tear down a loop per bypass and would carry no deadline of its own. The worker's
+    # loop lives in a thread, outlives any single bypass, and cancels the coroutine when the
+    # deadline passes.
+    timeout = (
+        _CHILD_BYPASS_TIMEOUT_SECONDS
+        if os.environ.get(_BYPASS_CHILD_ENV) == "1"
+        else _IN_PROCESS_BYPASS_TIMEOUT_SECONDS
+    )
+    return _CDP_WORKER.run(_run_bypass(), timeout=timeout)
 
 
 def _store_child_bypass_state(payload: dict[str, Any]) -> None:
@@ -868,6 +900,213 @@ def _terminate_helper_session(proc: subprocess.Popen[str]) -> None:
         proc.wait(timeout=5)
 
 
+def _part_path(result_path: Path) -> Path:
+    """Where the helper stages a result before renaming it into place."""
+    return result_path.with_name(result_path.name + ".part")
+
+
+class _BypassHelper:
+    """The helper subprocess that runs the bypasses, kept alive across them.
+
+    Spawning it costs about 4.5 seconds of interpreter start and imports before any work
+    begins, paid on every protected request - and a single search issues several. What it
+    keeps is the process, not the browser: each bypass still starts and closes its own
+    Chrome, so nothing accumulates between requests.
+
+    Protocol: one JSON request per line on stdin, answered by writing the result file that
+    request named. stdout and stderr stay attached to the parent's, so helper logs keep
+    showing up in `docker logs` as before.
+
+    Only one request is ever in flight - get() serializes every bypass behind LOCKED. The
+    lock here is for the idle reaper, which runs on a timer thread.
+    """
+
+    def __init__(self) -> None:
+        self._lock = threading.RLock()
+        self._proc: subprocess.Popen[str] | None = None
+        self._last_used = 0.0
+        self._idle_timer: threading.Timer | None = None
+
+    def _idle_timeout(self) -> float:
+        return _coerce_non_negative_float(
+            app_config.get("BYPASS_BROWSER_IDLE_TIMEOUT", _HELPER_IDLE_TIMEOUT_DEFAULT),
+            _HELPER_IDLE_TIMEOUT_DEFAULT,
+        )
+
+    def _spawn(self) -> subprocess.Popen[str]:
+        env_vars = os.environ.copy()
+        env_vars[_BYPASS_CHILD_ENV] = "1"
+        env_vars = _prepare_child_browser_env(env_vars)
+        return subprocess.Popen(
+            [sys.executable, "-m", "shelfmark.bypass.internal_bypasser"],
+            stdin=subprocess.PIPE,
+            text=True,
+            env=env_vars,
+            # Give the helper its own session: Chrome, Xvfb and ffmpeg inherit its process
+            # group, which is what lets the cleanup sweep tell this helper's browsers apart
+            # from a concurrent worker's (#1231) and lets us kill the whole tree below.
+            start_new_session=True,
+        )
+
+    def _running(self) -> subprocess.Popen[str] | None:
+        proc = self._proc
+        if proc is None:
+            return None
+        if proc.poll() is not None or proc.stdin is None or proc.stdin.closed:
+            return None
+        return proc
+
+    def _ensure_running(self) -> subprocess.Popen[str]:
+        proc = self._running()
+        if proc is not None:
+            return proc
+        if self._proc is not None:
+            logger.info("Bypass helper exited (code %s), starting a new one", self._proc.returncode)
+            self._discard()
+        self._proc = self._spawn()
+        return self._proc
+
+    def _discard(self, *, wait_for_exit: bool = True) -> None:
+        """Stop the helper and forget it.
+
+        `wait_for_exit` belongs to a helper that could still act on the closed pipe: an
+        idle one is sitting in its stdin read, notices EOF and exits on its own. A helper
+        dropped mid-bypass is blocked inside the solve and will not return to that read,
+        so the grace cannot end in anything but the kill below - and the caller waiting it
+        out is a user cancelling a download, holding LOCKED while every other bypass in
+        the worker queues behind them.
+        """
+        proc = self._proc
+        self._proc = None
+        if proc is None:
+            return
+
+        # Closing stdin ends the helper's request loop, so an idle helper gets to exit on
+        # its own. One mid-bypass cannot answer, and is killed below.
+        with suppress(OSError):
+            if proc.stdin is not None and not proc.stdin.closed:
+                proc.stdin.close()
+        if wait_for_exit:
+            try:
+                proc.wait(timeout=_HELPER_SHUTDOWN_GRACE_SECONDS)
+            except subprocess.TimeoutExpired:
+                logger.warning("Bypass helper did not exit on request, killing its session")
+
+        # Tear the session down either way: a helper killed mid-bypass leaves its Chrome
+        # and Xvfb running, and those leftovers are what made the next worker's browser
+        # fail to start. Harmless once it has already exited.
+        _terminate_helper_session(proc)
+
+    def _cancel_idle_timer(self) -> None:
+        if self._idle_timer is not None:
+            self._idle_timer.cancel()
+            self._idle_timer = None
+
+    def _arm_idle_timer(self) -> None:
+        self._cancel_idle_timer()
+        timeout = self._idle_timeout()
+        if self._proc is None or timeout <= 0:
+            return
+        timer = threading.Timer(timeout, self._reap_if_idle)
+        timer.daemon = True
+        self._idle_timer = timer
+        timer.start()
+
+    def _reap_if_idle(self) -> None:
+        with self._lock:
+            if self._proc is None:
+                return
+            idle_for = time.monotonic() - self._last_used
+            timeout = self._idle_timeout()
+            if idle_for < timeout:
+                # A bypass started while this timer was waiting for the lock.
+                self._arm_idle_timer()
+                return
+            logger.info("Closing idle bypass helper after %.0fs without work", idle_for)
+            self._discard()
+
+    def run(
+        self,
+        payload: dict[str, Any],
+        timeout: float,
+        cancel_flag: Event | None,
+    ) -> dict[str, Any]:
+        with self._lock:
+            self._cancel_idle_timer()
+            try:
+                return self._exchange(payload, timeout, cancel_flag)
+            finally:
+                self._last_used = time.monotonic()
+                self._arm_idle_timer()
+
+    def _exchange(
+        self,
+        payload: dict[str, Any],
+        timeout: float,
+        cancel_flag: Event | None,
+    ) -> dict[str, Any]:
+        request_line = json.dumps(payload) + "\n"
+        proc = self._ensure_running()
+        try:
+            self._write(proc, request_line)
+        except OSError as exc:
+            # A live helper can die between the liveness check and the write, so one retry
+            # on a fresh process. A fresh one failing here is a real failure.
+            logger.info("Bypass helper closed its pipe (%s), retrying on a new one", exc)
+            # Nothing to ask of a helper we cannot write to: its read end is already gone.
+            self._discard(wait_for_exit=False)
+            proc = self._ensure_running()
+            self._write(proc, request_line)
+
+        return self._await_result(proc, Path(str(payload["result_path"])), timeout, cancel_flag)
+
+    def _write(self, proc: subprocess.Popen[str], request_line: str) -> None:
+        if proc.stdin is None:
+            msg = "Bypass helper has no stdin pipe"
+            raise OSError(msg)
+        proc.stdin.write(request_line)
+        proc.stdin.flush()
+
+    def _await_result(
+        self,
+        proc: subprocess.Popen[str],
+        result_path: Path,
+        timeout: float,
+        cancel_flag: Event | None,
+    ) -> dict[str, Any]:
+        deadline = time.monotonic() + timeout
+        try:
+            while not result_path.exists():
+                if proc.poll() is not None:
+                    returncode = proc.returncode
+                    self._discard(wait_for_exit=False)
+                    msg = f"Internal bypasser helper exited without a result (code {returncode})"
+                    raise RuntimeError(msg)
+                if cancel_flag is not None and cancel_flag.is_set():
+                    # The helper is mid-bypass and cannot be told to stop, so it goes.
+                    self._discard(wait_for_exit=False)
+                    _check_cancellation(cancel_flag, "Bypass cancelled while waiting for helper")
+                if time.monotonic() >= deadline:
+                    self._discard(wait_for_exit=False)
+                    msg = "Internal bypasser helper process timed out"
+                    raise TimeoutError(msg)
+                time.sleep(_HELPER_RESULT_POLL_SECONDS)
+
+            return json.loads(result_path.read_text(encoding="utf-8"))
+        finally:
+            # Every way out of here is final for this request: either the answer has been
+            # read, or the helper that would have written it has just been killed. Nothing
+            # will write these paths afterwards and nothing will come looking for them, so
+            # they are cleaned on the failure paths too - otherwise every cancelled
+            # download and every wedged solve leaves one behind for the container's life.
+            for path in (result_path, _part_path(result_path)):
+                with suppress(OSError):
+                    path.unlink()
+
+
+_BYPASS_HELPER = _BypassHelper()
+
+
 def _get_via_subprocess(url: str, retry: int, cancel_flag: Event | None = None) -> str:
     """Run the browser bypass in a helper process isolated from gunicorn/gevent."""
     _check_cancellation(cancel_flag, "Bypass cancelled before helper process")
@@ -878,50 +1117,15 @@ def _get_via_subprocess(url: str, retry: int, cancel_flag: Event | None = None)
     # freshly spawned helper would otherwise pre-resolve AA hostnames against the system
     # resolver - which may be blocked or hijacked by the user's ISP. Pass the parent's
     # active DNS config so the helper mirrors it (e.g. DoH) when building Chrome's host
-    # resolver rules.
+    # resolver rules. Sent with every request, not just at spawn, because a helper outlives
+    # changes the parent makes to its DNS provider.
     payload = {
         "url": url,
         "retry": retry,
         "result_path": str(result_path),
         "dns_config": network.get_dns_config(),
     }
-    env_vars = os.environ.copy()
-    env_vars[_BYPASS_CHILD_ENV] = "1"
-    env_vars = _prepare_child_browser_env(env_vars)
-
-    proc = subprocess.Popen(
-        [sys.executable, "-m", "shelfmark.bypass.internal_bypasser"],
-        stdin=subprocess.PIPE,
-        text=True,
-        env=env_vars,
-        # Give the helper its own session: Chrome, Xvfb and ffmpeg inherit its process
-        # group, which is what lets the cleanup sweep tell this bypass's browsers apart
-        # from a concurrent worker's (#1231) and lets us kill the whole tree below.
-        start_new_session=True,
-    )
-    timed_out = False
-    try:
-        proc.communicate(json.dumps(payload), timeout=_BYPASS_SUBPROCESS_TIMEOUT_SECONDS)
-    except subprocess.TimeoutExpired:
-        timed_out = True
-    finally:
-        # Always tear the session down, not just on timeout: killing the helper alone
-        # leaves its Chrome and Xvfb running, and those leftovers are what made the next
-        # worker's browser fail to start in the first place.
-        _terminate_helper_session(proc)
-
-    if timed_out:
-        msg = "Internal bypasser helper process timed out"
-        raise TimeoutError(msg)
-
-    try:
-        result = json.loads(result_path.read_text())
-    except FileNotFoundError as exc:
-        msg = f"Internal bypasser helper exited without a result (code {proc.returncode})"
-        raise RuntimeError(msg) from exc
-    finally:
-        with suppress(OSError):
-            result_path.unlink()
+    result = _BYPASS_HELPER.run(payload, _BYPASS_SUBPROCESS_TIMEOUT_SECONDS, cancel_flag)
 
     if not isinstance(result, dict):
         msg = "Internal bypasser helper returned an invalid result"
@@ -1258,26 +1462,38 @@ def get_bypassed_page(
     return response_html
 
 
+def _dns_fingerprint(dns_config: dict[str, Any]) -> tuple[str, tuple[str, ...], bool]:
+    """Reduce a DNS config to what has to match for two of them to be the same one."""
+    provider = str(dns_config.get("provider") or "").strip().lower()
+    servers = dns_config.get("servers") if provider == "manual" else None
+    server_list = tuple(str(server) for server in servers) if isinstance(servers, list) else ()
+    return (provider, server_list, bool(dns_config.get("doh_enabled")))
+
+
 def _apply_parent_dns_config(dns_config: dict[str, Any]) -> None:
     """Mirror the parent process's active DNS provider in this helper subprocess.
 
-    DNS state is in-memory only, so a fresh helper defaults to system DNS and would
-    pre-resolve AA hostnames (for Chrome's --host-resolver-rules) against a resolver
-    that may be blocked/hijacked. Re-applying the parent's provider keeps the helper on
-    the same DoH/custom resolver the parent already validated.
+    DNS state is in-memory only, so a helper left to itself would pre-resolve AA hostnames
+    (for Chrome's --host-resolver-rules) against a resolver that may be blocked or
+    hijacked. Re-applying the parent's provider keeps the helper on the same DoH/custom
+    resolver the parent already validated.
+
+    Compared against what this process is *actually* resolving through, rather than
+    against the last config it happened to be handed. The helper now outlives the request,
+    so it has to be able to travel back to auto as well as away from it - which a user
+    flipping CUSTOM_DNS in settings does live, without a restart - and asking the network
+    module what it is doing beats keeping a second, drifting copy of that answer here.
     """
-    provider = str(dns_config.get("provider") or "").strip().lower()
-    # "auto" means the parent has not rotated off system DNS yet, so the helper's own
-    # default initialization already matches it - nothing to override.
-    if not provider or provider == "auto":
+    wanted = _dns_fingerprint(dns_config)
+    provider, servers, use_doh = wanted
+    if not provider:
         return
-    manual_servers = dns_config.get("servers") if provider == "manual" else None
+    # set_dns_provider() rebuilds the resolvers, so it is worth doing only on a real change.
+    if wanted == _dns_fingerprint(network.get_dns_config()):
+        return
+
     try:
-        network.set_dns_provider(
-            provider,
-            manual_servers,
-            use_doh=bool(dns_config.get("doh_enabled")),
-        )
+        network.set_dns_provider(provider, list(servers) or None, use_doh=use_doh)
     except (OSError, RuntimeError, ValueError) as exc:
         logger.warning("Could not apply parent DNS config (%s): %s", provider, exc)
 
@@ -1314,9 +1530,20 @@ def _start_parent_watchdog() -> None:
     ).start()
 
 
-def _run_child_process() -> int:
-    """CLI entrypoint used by the Docker helper subprocess."""
-    request = json.loads(sys.stdin.read() or "{}")
+def _publish_result(result_path: Path, payload: dict[str, Any]) -> None:
+    """Write the result file atomically.
+
+    The parent decides the request is answered the moment this path exists, so it must
+    never observe a half-written file. Rename within the same directory is atomic.
+    """
+    tmp_path = _part_path(result_path)
+    tmp_path.write_text(json.dumps(payload), encoding="utf-8")
+    tmp_path.replace(result_path)
+
+
+def _handle_child_request(request_line: str) -> int:
+    """Answer one request from the parent."""
+    request = json.loads(request_line or "{}")
     result_path = Path(str(request["result_path"]))
     url = str(request["url"])
     retry = _coerce_positive_int(
@@ -1327,6 +1554,15 @@ def _run_child_process() -> int:
     if isinstance(dns_config, dict):
         _apply_parent_dns_config(dns_config)
 
+    # The parent owns the cookie store; this process only solves. Starting each request
+    # from an empty store is what a helper spawned per request gave for free, and losing
+    # it is what let clearance the parent had deliberately purged for some *other* host
+    # survive here and get merged back over the parent's copy by the export below - the
+    # dead-cookie resurrection that http.py's _redirect_loop_handoff purges to avoid.
+    # Nothing is lost by dropping it: get() below re-checks cached cookies, and the
+    # parent already ran that same check against a store that is a superset of this one.
+    clear_cf_cookies()
+
     try:
         html = get(url, retry=retry)
         cookies, user_agents = export_store()
@@ -1336,7 +1572,7 @@ def _run_child_process() -> int:
             "cookies": cookies,
             "user_agents": user_agents,
         }
-        result_path.write_text(json.dumps(payload), encoding="utf-8")
+        _publish_result(result_path, payload)
     except Exception as exc:  # noqa: BLE001 - helper boundary must serialize failures.
         payload = {
             "ok": False,
@@ -1344,11 +1580,28 @@ def _run_child_process() -> int:
             "error": str(exc),
             "traceback": traceback.format_exc(),
         }
-        result_path.write_text(json.dumps(payload), encoding="utf-8")
+        _publish_result(result_path, payload)
         return 1
     return 0
 
 
+def _run_child_process() -> int:
+    """CLI entrypoint used by the Docker helper subprocess.
+
+    Serves one request per line of stdin until the parent closes the pipe, so a burst of
+    protected requests - a single search is several - pays the interpreter start and imports
+    once instead of per request. Each bypass still gets its own browser, closed before the
+    answer is published.
+    """
+    exit_code = 0
+    for line in sys.stdin:
+        request_line = line.strip()
+        if not request_line:
+            continue
+        exit_code = _handle_child_request(request_line)
+    return exit_code
+
+
 if __name__ == "__main__":
     # Started here rather than in _run_child_process() so it only ever watches a real
     # spawned helper, never a test or an embedded call.

shelfmark/config/settings.py

@@ -1655,6 +1655,19 @@ def cloudflare_bypass_settings() -> list[SettingsField]:
             requires_restart=True,
             show_when={"field": "USING_EXTERNAL_BYPASSER", "value": True},
         ),
+        NumberField(
+            key="BYPASS_BROWSER_IDLE_TIMEOUT",
+            label="Bypasser Idle Timeout (seconds)",
+            description=(
+                "How long the bypass helper process may sit unused before it is shut down. "
+                "Higher keeps more searches fast, lower frees memory sooner."
+            ),
+            default=180,
+            min_value=30,
+            max_value=3600,
+            requires_restart=True,
+            show_when={"field": "USING_EXTERNAL_BYPASSER", "value": False},
+        ),
     ]
 
 

shelfmark/download/clients/base_handler.py

@@ -843,6 +843,9 @@ class ExternalClientHandler(DownloadHandler, ABC):
                             expected_hash=request.expected_hash,
                             seeding_time_limit=request.seeding_time_limit,
                             ratio_limit=request.ratio_limit,
+                            # rTorrent has no category concept, so its audiobook label
+                            # can only be chosen from the content type (#1235).
+                            content_type=task.content_type,
                         )
                     except Exception as e:
                         if not refresh_attempted:

shelfmark/download/clients/rtorrent.py

@@ -11,7 +11,12 @@ from urllib.parse import urlparse
 
 from shelfmark.core.config import config
 from shelfmark.core.logger import setup_logger
-from shelfmark.core.utils import get_hardened_xmlrpc_client
+from shelfmark.core.utils import (
+    get_hardened_xmlrpc_client,
+)
+from shelfmark.core.utils import (
+    is_audiobook as check_audiobook,
+)
 from shelfmark.download.clients import (
     DownloadClient,
     DownloadStatus,
@@ -173,7 +178,8 @@ class RTorrentClient(DownloadClient):
 
             commands = []
 
-            is_audiobook = kwargs.get("content_type") == "audiobook"
+            content_type = kwargs.get("content_type")
+            is_audiobook = check_audiobook(content_type if isinstance(content_type, str) else None)
             default_label = (
                 self._audiobook_label if is_audiobook and self._audiobook_label else self._label
             )

shelfmark/download/http.py

@@ -11,6 +11,7 @@ import requests
 from tqdm import tqdm
 
 from shelfmark.bypass import BypassCancelledError, cookie_store
+from shelfmark.bypass.challenge import challenge_marker
 from shelfmark.core.config import config as app_config
 from shelfmark.core.logger import setup_logger
 from shelfmark.core.request_helpers import coerce_bool, normalize_positive_int
@@ -233,6 +234,22 @@ def _is_retryable_error(e: Exception) -> bool:
 _DEAD_MIRROR_CODES = (410, 451)
 
 
+def _response_challenge_marker(response: requests.Response) -> str | None:
+    """The challenge marker in a response body, or None if it carries no challenge.
+
+    Content type is checked first so a JSON or octet-stream error body is never
+    decoded just to be scanned; a missing header is scanned anyway, since an
+    interstitial served without one is still an interstitial.
+    """
+    content_type = response.headers.get("Content-Type", "")
+    if content_type and "html" not in content_type.lower():
+        return None
+    try:
+        return challenge_marker(response.text)
+    except UnicodeDecodeError, ValueError:
+        return None
+
+
 def _fatal_mirror_reason(e: Exception) -> str | None:
     """Return why ``e`` proves the mirror is unusable, or None if it may recover.
 
@@ -455,6 +472,35 @@ def html_get_page(
                         )
                         continue
 
+                # A 503 still serving a challenge is protection, not a busy origin. The
+                # handshake above has nothing left to echo back, and 503 is in
+                # RETRYABLE_CODES, so without this the request spends every attempt on
+                # the same wall: the bypasser is only ever reached from the 403 branch
+                # and the AA redirect rescues. Gate on the body, not the status, so a
+                # genuine overloaded-origin 503 keeps its retry path.
+                if response.status_code == _HTTP_STATUS_SERVICE_UNAVAILABLE:
+                    marker = _response_challenge_marker(response)
+                    if marker and _bypass_handoff_allowed():
+                        if cookies:
+                            # Challenged while presenting clearance means those cookies
+                            # are dead; same reasoning as the 403 branch below.
+                            logger.debug(
+                                "503 challenge with cookies presented; purging: %s", current_url
+                            )
+                            _purge_clearance(current_url)
+                        logger.info(
+                            "503 challenge detected (%s); switching to bypasser: %s",
+                            marker,
+                            current_url,
+                        )
+                        return _run_bypasser(current_url)
+                    if marker:
+                        logger.debug(
+                            "503 challenge (%s) but no bypasser handoff available: %s",
+                            marker,
+                            current_url,
+                        )
+
                 if is_aa_url and response.is_redirect:
                     location = response.headers.get("Location", "")
                     if not location:

shelfmark/metadata_providers/hardcover.py

@@ -48,6 +48,10 @@ HARDCOVER_PAGE_SIZE = 25  # Hardcover API returns max 25 results per page
 HARDCOVER_MIN_AUTHOR_PARTS = 2
 HARDCOVER_MIN_TYPEAHEAD_QUERY_LENGTH = 2
 HARDCOVER_MAX_SERIES_OPTIONS = 7
+# Hardcover hands out short opaque tokens now ("hc_pat_...") instead of the ~500 char
+# JWTs it used to, so the length floor only applies to keys without that prefix.
+HARDCOVER_API_KEY_PREFIX = "hc_pat_"
+HARDCOVER_BEARER_PREFIX_PATTERN = re.compile(r"^bearer\s+", re.IGNORECASE)
 HARDCOVER_API_KEY_MIN_LENGTH = 100
 HARDCOVER_LIST_URL_PATTERN = re.compile(
     r"^/(?:@([\w.-]+)/)?lists?/([\w-]+)/?$",
@@ -670,7 +674,7 @@ def _normalize_series_position(value: Any) -> float | None:
 def _normalize_hardcover_api_key(value: object) -> str:
     """Normalize Hardcover API keys, stripping copied auth-header prefixes."""
     normalized_value = normalize_optional_text(value) or ""
-    return normalized_value.removeprefix("Bearer ").strip()
+    return HARDCOVER_BEARER_PREFIX_PATTERN.sub("", normalized_value.strip()).strip()
 
 
 def _normalize_search_text(value: str) -> str:
@@ -3019,12 +3023,13 @@ def _test_hardcover_connection(current_values: dict[str, Any] | None = None) ->
         _save_connected_user(None, None)
         return {"success": False, "message": "API key is required"}
 
-    if key_len < HARDCOVER_API_KEY_MIN_LENGTH:
+    is_prefixed_key = api_key.startswith(HARDCOVER_API_KEY_PREFIX)
+    if not is_prefixed_key and key_len < HARDCOVER_API_KEY_MIN_LENGTH:
         return {
             "success": False,
             "message": (
-                f"API key seems too short ({key_len} chars). "
-                f"Expected {HARDCOVER_API_KEY_MIN_LENGTH}+ chars."
+                f"API key seems too short ({key_len} chars). Expected a key starting "
+                f"with {HARDCOVER_API_KEY_PREFIX} or {HARDCOVER_API_KEY_MIN_LENGTH}+ chars."
             ),
         }
 
@@ -3131,7 +3136,7 @@ def hardcover_settings() -> list[SettingsField]:
         PasswordField(
             key="HARDCOVER_API_KEY",
             label="API Key",
-            description="Get your API key from hardcover.app/account/api",
+            description="Get your API key from hardcover.app/account/api (starts with hc_pat_)",
             required=True,
         ),
         ActionButton(

shelfmark/release_sources/audiobookbay/scraper.py

@@ -2,7 +2,7 @@
 
 import re
 import time
-from urllib.parse import quote
+from urllib.parse import quote, quote_plus
 
 import requests
 from bs4 import BeautifulSoup
@@ -10,6 +10,7 @@ from bs4 import BeautifulSoup
 from shelfmark.core.config import config
 from shelfmark.core.logger import setup_logger
 from shelfmark.download import http as downloader
+from shelfmark.release_sources.audiobookbay.utils import normalize_search_punctuation
 
 logger = setup_logger(__name__)
 
@@ -98,8 +99,10 @@ def _encode_search_query(query: str, *, exact_phrase: bool) -> str:
         and not (search_query.startswith('"') and search_query.endswith('"'))
     ):
         search_query = f'"{search_query}"'
-    # Keep ABB-friendly encoding style (spaces as '+') while percent-encoding quotes.
-    return search_query.replace('"', "%22").replace(" ", "+")
+    # Keep ABB's space-as-'+' style, but percent-encode everything else: a bare
+    # '&' would otherwise start a new query parameter, '%' would open an invalid
+    # escape, and a literal '+' would arrive as a space.
+    return quote_plus(search_query)
 
 
 def _normalize_result_url(url: str, hostname: str) -> str:
@@ -153,6 +156,9 @@ def search_audiobookbay(
 
     """
     results = []
+    # ABB matches the stored, untexturized title, so a curly apostrophe reaching
+    # the search returns nothing at all rather than merely ranking worse.
+    query = normalize_search_punctuation(query)
     rate_limit_delay = _coerce_non_negative_float(config.get("ABB_RATE_LIMIT_DELAY", 1.0), 1.0)
     session = requests.Session()
 

shelfmark/release_sources/audiobookbay/source.py

@@ -23,7 +23,11 @@ from shelfmark.release_sources import (
     register_source,
 )
 from shelfmark.release_sources.audiobookbay import scraper
-from shelfmark.release_sources.audiobookbay.utils import normalize_hostname, parse_size
+from shelfmark.release_sources.audiobookbay.utils import (
+    normalize_hostname,
+    normalize_search_punctuation,
+    parse_size,
+)
 
 logger = setup_logger(__name__)
 MIN_RELEVANCE_QUERY_WORD_LENGTH = 2
@@ -227,10 +231,12 @@ class AudiobookBaySource(ReleaseSource):
                         deduped_queries[index + 1].lower(),
                     )
 
-            # Extract query words for relevance checking
+            # Extract query words for relevance checking. Both sides of the
+            # comparison are punctuation-normalized: scraped titles carry the
+            # typographic forms WordPress renders, queries carry the ASCII ones.
             query_words = {
                 word.lower()
-                for word in query_lower.split()
+                for word in normalize_search_punctuation(query_lower).split()
                 if len(word) > MIN_RELEVANCE_QUERY_WORD_LENGTH
             }
 
@@ -239,7 +245,7 @@ class AudiobookBaySource(ReleaseSource):
                 try:
                     raw_title = result["title"]
                     title, author = _split_title_and_author(raw_title)
-                    title_for_filter = raw_title.lower()
+                    title_for_filter = normalize_search_punctuation(raw_title).lower()
 
                     # Basic relevance check: ensure title contains at least one query word
                     # This filters out homepage "Latest" feed items that may leak through

shelfmark/release_sources/audiobookbay/utils.py

@@ -2,6 +2,63 @@
 
 import re
 
+# WordPress texturizes punctuation on output only: a post stored as "The
+# Stranger's Wife" is rendered as "The Stranger’s Wife". ABB's search matches the
+# stored value, so a query carrying the typographic form matches nothing -- and
+# because ABB ANDs its search terms, one such term empties the entire result set.
+# Book metadata and phone keyboards both hand us the typographic forms, so map
+# them back before they reach a search or a title comparison.
+_ASCII_PUNCTUATION = str.maketrans(
+    {
+        # Single quotes
+        "‘": "'",  # left single quotation mark
+        "’": "'",  # right single quotation mark
+        "‚": "'",  # single low-9 quotation mark
+        "‛": "'",  # single high-reversed-9 quotation mark
+        "′": "'",  # prime
+        "´": "'",  # acute accent
+        "`": "'",  # grave accent
+        # Double quotes
+        "“": '"',  # left double quotation mark
+        "”": '"',  # right double quotation mark
+        "„": '"',  # double low-9 quotation mark
+        "‟": '"',  # double high-reversed-9 quotation mark
+        "″": '"',  # double prime
+        # Dashes
+        "‐": "-",  # hyphen
+        "‑": "-",  # non-breaking hyphen
+        "‒": "-",  # figure dash
+        "–": "-",  # en dash
+        "—": "-",  # em dash
+        "―": "-",  # horizontal bar
+        "−": "-",  # minus sign
+        "﹘": "-",  # small em dash
+        "﹣": "-",  # small hyphen-minus
+        "－": "-",  # fullwidth hyphen-minus
+        # Ellipsis
+        "…": "...",  # horizontal ellipsis
+    }
+)
+
+
+def normalize_search_punctuation(text: str) -> str:
+    """Replace typographic punctuation with the ASCII forms ABB stores.
+
+    Each character is mapped individually rather than collapsing runs, so an
+    ASCII "--" is left alone: only characters ABB cannot have stored are
+    rewritten.
+
+    Args:
+        text: A search query, or a scraped title being compared against one.
+
+    Returns:
+        The text with curly quotes, dashes and ellipses mapped to ASCII.
+
+    """
+    if not text:
+        return text
+    return text.translate(_ASCII_PUNCTUATION)
+
 
 def normalize_hostname(raw: str | None) -> str:
     """Normalize a user-supplied hostname for URL construction.

tests/audiobookbay/test_scraper.py

@@ -317,6 +317,44 @@ class TestSearchAudiobookbay:
         assert "s=%22test+query%22" in requested_url
         assert "cat=undefined%2Cundefined" in requested_url
 
+    @patch("shelfmark.release_sources.audiobookbay.scraper.downloader.html_get_page")
+    @patch("shelfmark.release_sources.audiobookbay.scraper.config.get")
+    def test_search_audiobookbay_normalizes_curly_apostrophe(self, mock_config_get, mock_html_get):
+        """Test curly apostrophes are searched as the ASCII form ABB stores."""
+        mock_config_get.return_value = 0.0
+        mock_html_get.return_value = (SAMPLE_SEARCH_HTML, "https://audiobookbay.lu/?s=x")
+
+        scraper.search_audiobookbay(
+            "the stranger’s wife",
+            max_pages=1,
+            hostname="audiobookbay.lu",
+        )
+
+        requested_url = mock_html_get.call_args.args[0]
+        assert "s=the+stranger%27s+wife" in requested_url
+        assert "%E2%80%99" not in requested_url
+
+    @patch("shelfmark.release_sources.audiobookbay.scraper.downloader.html_get_page")
+    @patch("shelfmark.release_sources.audiobookbay.scraper.config.get")
+    def test_search_audiobookbay_percent_encodes_reserved_characters(
+        self, mock_config_get, mock_html_get
+    ):
+        """Test reserved characters cannot break out of the search parameter."""
+        mock_config_get.return_value = 0.0
+        mock_html_get.return_value = (SAMPLE_SEARCH_HTML, "https://audiobookbay.lu/?s=x")
+
+        scraper.search_audiobookbay(
+            "sense & sensibility 100% c++",
+            max_pages=1,
+            hostname="audiobookbay.lu",
+        )
+
+        requested_url = mock_html_get.call_args.args[0]
+        assert "s=sense+%26+sensibility+100%25+c%2B%2B" in requested_url
+        # The only surviving '&' introduces the legacy category parameter.
+        assert requested_url.count("&") == 1
+        assert requested_url.endswith("&cat=undefined%2Cundefined")
+
     @patch("shelfmark.release_sources.audiobookbay.scraper.downloader.html_get_page")
     @patch("shelfmark.release_sources.audiobookbay.scraper.config.get")
     def test_search_audiobookbay_always_uses_legacy_category_query(

tests/audiobookbay/test_source.py

@@ -306,6 +306,42 @@ class TestAudiobookBaySource:
         assert len(results) == 1
         assert results[0].title == "Test Book by Test Author"
 
+    @patch("shelfmark.release_sources.audiobookbay.source.scraper.search_audiobookbay")
+    def test_search_relevance_filtering_spans_typographic_punctuation(self, mock_search):
+        """Test an ASCII query still matches the typographic title ABB renders."""
+        mock_search.return_value = [
+            {
+                "title": "The Stranger’s Wife — Anna‑Lou Weatherley",
+                "link": "https://audiobookbay.lu/abss/the-strangers-wife/",
+                "format": "M4B",
+                "size": "259 MB",
+                "language": "English",
+            },
+        ]
+
+        source = AudiobookBaySource()
+        book = BookMetadata(
+            provider="test",
+            provider_id="123",
+            title="Stranger's",
+            authors=["Anna-Lou Weatherley"],
+        )
+        # Every query word carries punctuation, so the result survives only when
+        # both sides of the comparison are normalized.
+        plan = ReleaseSearchPlan(
+            languages=["en"],
+            isbn_candidates=[],
+            author="",
+            title_variants=[ReleaseSearchVariant(title="Stranger's", author="")],
+            grouped_title_variants=[],
+        )
+
+        results = source.search(book, plan, content_type="audiobook")
+
+        assert len(results) == 1
+        # The release keeps the title as ABB rendered it; only matching normalizes.
+        assert results[0].title == "The Stranger’s Wife — Anna‑Lou Weatherley"
+
     @patch("shelfmark.release_sources.audiobookbay.source.scraper.search_audiobookbay")
     def test_search_result_mapping(self, mock_search):
         """Test conversion of scraper results to Release objects."""

tests/audiobookbay/test_utils.py

@@ -2,7 +2,52 @@
 Tests for AudiobookBay utility functions.
 """
 
-from shelfmark.release_sources.audiobookbay.utils import parse_size
+from shelfmark.release_sources.audiobookbay.utils import normalize_search_punctuation, parse_size
+
+
+class TestNormalizeSearchPunctuation:
+    """Tests for the normalize_search_punctuation function."""
+
+    def test_curly_apostrophe_becomes_ascii(self):
+        """ABB matches the stored ASCII apostrophe, not the rendered curly one."""
+        assert normalize_search_punctuation("The Stranger’s Wife") == "The Stranger's Wife"
+
+    def test_all_single_quote_variants(self):
+        """Every single-quote lookalike collapses to the ASCII apostrophe."""
+        for variant in ("‘", "’", "‚", "‛", "′", "´", "`"):
+            assert normalize_search_punctuation(f"don{variant}t") == "don't"
+
+    def test_all_double_quote_variants(self):
+        """Every double-quote lookalike collapses to the ASCII double quote."""
+        for variant in ("“", "”", "„", "‟", "″"):
+            assert normalize_search_punctuation(f"{variant}quoted{variant}") == '"quoted"'
+
+    def test_all_dash_variants(self):
+        """Every dash lookalike collapses to the ASCII hyphen."""
+        for variant in ("‐", "‑", "‒", "–", "—", "―", "−", "﹘", "﹣", "－"):
+            assert normalize_search_punctuation(f"anna{variant}lou") == "anna-lou"
+
+    def test_ellipsis_expands_to_three_dots(self):
+        """WordPress renders '...' as a single ellipsis character."""
+        assert normalize_search_punctuation("And Then…") == "And Then..."
+
+    def test_ascii_query_is_unchanged(self):
+        """A query that is already ASCII passes through untouched."""
+        assert normalize_search_punctuation("The Stranger's Wife") == "The Stranger's Wife"
+
+    def test_ascii_dash_runs_are_not_collapsed(self):
+        """Only characters ABB cannot have stored are rewritten."""
+        assert normalize_search_punctuation("Book -- Subtitle") == "Book -- Subtitle"
+
+    def test_other_punctuation_is_preserved(self):
+        """Colons and commas carry search signal and are left alone."""
+        assert normalize_search_punctuation("Weatherley: Book 3, Part 1") == (
+            "Weatherley: Book 3, Part 1"
+        )
+
+    def test_empty_query(self):
+        """An empty query is returned as-is."""
+        assert normalize_search_punctuation("") == ""
 
 
 class TestParseSize:

tests/bypass/test_internal_bypasser.py

@@ -1,6 +1,5 @@
 import asyncio
 import json
-import subprocess
 import threading
 from pathlib import Path
 
@@ -545,30 +544,59 @@ def test_cleanup_is_skipped_without_proc(monkeypatch, tmp_path):
     assert internal_bypasser._cleanup_orphan_processes() == 0
 
 
+class _FakeHelperStdin:
+    """The request pipe: a write is how the helper receives one request."""
+
+    def __init__(self, process):
+        self._process = process
+        self.closed = False
+
+    def write(self, data):
+        self._process.serve(data)
+
+    def flush(self):
+        return None
+
+    def close(self):
+        self.closed = True
+
+
 class _FakeHelperProcess:
-    """Stand-in for the bypass helper subprocess."""
+    """Stand-in for the bypass helper subprocess.
+
+    The helper serves one request per line of stdin and answers by writing the result file
+    the request named, so that is what this fakes: a write produces an answer.
+    """
 
     def __init__(self, *_args, **kwargs):
         self.kwargs = kwargs
         self.pid = 4242
-        self.returncode = 0
-        self.timed_out = False
+        self.returncode = None
+        self.answers = True
         self.killed = False
         self.waited = False
+        self.stdin = _FakeHelperStdin(self)
+        self.requests: list[dict] = []
 
-    def communicate(self, payload, timeout=None):
-        if self.timed_out:
-            raise subprocess.TimeoutExpired(cmd="helper", timeout=timeout)
+    def serve(self, payload):
         request = json.loads(payload)
+        self.requests.append(request)
+        if not self.answers:
+            return
         result = {"ok": True, "html": "<html>solved</html>", "cookies": {}, "user_agents": {}}
         Path(request["result_path"]).write_text(json.dumps(result), encoding="utf-8")
-        return "", ""
+
+    def poll(self):
+        return self.returncode
 
     def kill(self):
         self.killed = True
+        self.returncode = -9
 
     def wait(self, timeout=None):
         self.waited = True
+        if self.returncode is None:
+            self.returncode = 0
         return self.returncode
 
 
@@ -578,13 +606,47 @@ def _patch_helper_subprocess(monkeypatch, internal_bypasser, process, killed_gro
     monkeypatch.setattr(
         internal_bypasser.os, "killpg", lambda pgid, _sig: killed_groups.append(pgid)
     )
+    # A fresh helper per test: the module-level one is shared, and a process parked by one
+    # test would be handed to the next.
+    helper = internal_bypasser._BypassHelper()
+    monkeypatch.setattr(internal_bypasser._BypassHelper, "_idle_timeout", lambda _self: 0.0)
+    monkeypatch.setattr(internal_bypasser, "_BYPASS_HELPER", helper)
+    return helper
 
 
 def test_helper_runs_in_its_own_session_and_is_torn_down(monkeypatch):
     """Regression test for issue #1231: the helper's Chrome and Xvfb must belong to the
     helper's own process group, and the whole group must die with it - otherwise the
     leftovers break the next worker's browser and can only be cleared by a sweep broad
-    enough to kill a concurrent worker's browser too."""
+    enough to kill a concurrent worker's browser too.
+
+    The helper outlives a single request, so the teardown happens when it is dropped rather
+    than after every solve. Each bypass still closes its own browser, so what survives in
+    between is the process, not a Chrome.
+    """
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    processes: list[_FakeHelperProcess] = []
+    killed_groups: list[int] = []
+
+    def _make_process(*args, **kwargs):
+        process = _FakeHelperProcess(*args, **kwargs)
+        processes.append(process)
+        return process
+
+    helper = _patch_helper_subprocess(monkeypatch, internal_bypasser, _make_process, killed_groups)
+
+    assert internal_bypasser._get_via_subprocess("https://example.com", 1) == "<html>solved</html>"
+    assert processes[0].kwargs["start_new_session"] is True
+    assert killed_groups == [], "the helper was torn down after a single request"
+
+    helper._discard()
+
+    assert killed_groups == [processes[0].pid]
+
+
+def test_helper_serves_a_second_request_without_respawning(monkeypatch):
+    """The interpreter start and imports are paid once, not per protected request."""
     import shelfmark.bypass.internal_bypasser as internal_bypasser
 
     processes: list[_FakeHelperProcess] = []
@@ -597,9 +659,14 @@ def test_helper_runs_in_its_own_session_and_is_torn_down(monkeypatch):
 
     _patch_helper_subprocess(monkeypatch, internal_bypasser, _make_process, killed_groups)
 
-    assert internal_bypasser._get_via_subprocess("https://example.com", 1) == "<html>solved</html>"
-    assert processes[0].kwargs["start_new_session"] is True
-    assert killed_groups == [processes[0].pid]
+    internal_bypasser._get_via_subprocess("https://example.com/one", 1)
+    internal_bypasser._get_via_subprocess("https://example.com/two", 1)
+
+    assert len(processes) == 1
+    assert [request["url"] for request in processes[0].requests] == [
+        "https://example.com/one",
+        "https://example.com/two",
+    ]
 
 
 def test_helper_timeout_kills_the_whole_session(monkeypatch):
@@ -611,11 +678,12 @@ def test_helper_timeout_kills_the_whole_session(monkeypatch):
 
     def _make_process(*args, **kwargs):
         process = _FakeHelperProcess(*args, **kwargs)
-        process.timed_out = True
+        process.answers = False  # accepts the request, never writes a result
         processes.append(process)
         return process
 
     _patch_helper_subprocess(monkeypatch, internal_bypasser, _make_process, killed_groups)
+    monkeypatch.setattr(internal_bypasser, "_BYPASS_SUBPROCESS_TIMEOUT_SECONDS", 0.1)
 
     with pytest.raises(TimeoutError):
         internal_bypasser._get_via_subprocess("https://example.com", 1)

tests/bypass/test_warm_helper.py

@@ -0,0 +1,658 @@
+"""Tests for keeping the bypass helper process alive between requests.
+
+The browser is deliberately not kept: every bypass starts and closes its own Chrome. What
+survives is the helper process, whose interpreter start and imports are pure overhead.
+"""
+
+import asyncio
+import json
+
+import pytest
+
+
+class _FakeStdin:
+    def __init__(self) -> None:
+        self.closed = False
+        self.written: list[str] = []
+
+    def write(self, data: str) -> None:
+        if self.closed:
+            raise BrokenPipeError("stdin is closed")
+        self.written.append(data)
+
+    def flush(self) -> None:
+        return None
+
+    def close(self) -> None:
+        self.closed = True
+
+
+class _FakeProc:
+    """Enough of subprocess.Popen for the helper's process bookkeeping."""
+
+    _next_pid = 90001
+
+    def __init__(self) -> None:
+        self.stdin = _FakeStdin()
+        self.returncode: int | None = None
+        self.waited = False
+        # A pid nothing may actually be signalled by: _terminate_helper_session is patched
+        # out in these tests, and a stray killpg on a live pid would take out the test run.
+        type(self)._next_pid += 1
+        self.pid = type(self)._next_pid
+
+    def poll(self) -> int | None:
+        return self.returncode
+
+    def wait(self, timeout: float | None = None) -> int:
+        self.waited = True
+        if self.returncode is None:
+            self.returncode = 0
+        return self.returncode
+
+    def kill(self) -> None:
+        self.returncode = -9
+
+
+def _helper_with_fake_spawn(monkeypatch, procs: list[_FakeProc], terminated=None):
+    """Build a helper that hands out fake processes and never arms a real timer."""
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    def _spawn(_self) -> _FakeProc:
+        proc = _FakeProc()
+        procs.append(proc)
+        return proc
+
+    def _terminate(proc) -> None:
+        if terminated is not None:
+            terminated.append(proc)
+
+    monkeypatch.setattr(internal_bypasser._BypassHelper, "_spawn", _spawn)
+    monkeypatch.setattr(internal_bypasser._BypassHelper, "_idle_timeout", lambda _self: 0.0)
+    monkeypatch.setattr(internal_bypasser, "_terminate_helper_session", _terminate)
+    return internal_bypasser._BypassHelper()
+
+
+def _answered_payload(tmp_path, name: str = "result.json") -> dict:
+    """A request whose result file already exists, so the helper resolves immediately."""
+    result_path = tmp_path / name
+    result_path.write_text(json.dumps({"ok": True, "html": "<html/>"}), encoding="utf-8")
+    return {"url": "https://example.com", "retry": 1, "result_path": str(result_path)}
+
+
+def test_helper_serves_consecutive_requests_from_one_process(monkeypatch, tmp_path):
+    """The point of the whole thing: request two and three must not re-pay the spawn."""
+    procs: list[_FakeProc] = []
+    helper = _helper_with_fake_spawn(monkeypatch, procs)
+
+    for i in range(3):
+        result = helper.run(_answered_payload(tmp_path, f"r{i}.json"), timeout=5, cancel_flag=None)
+        assert result["ok"] is True
+
+    assert len(procs) == 1, "each request spawned its own helper"
+    assert len(procs[0].stdin.written) == 3
+    assert all(line.endswith("\n") for line in procs[0].stdin.written), (
+        "requests must be newline-delimited or the helper's loop cannot split them"
+    )
+
+
+def test_helper_respawns_after_the_previous_one_died(monkeypatch, tmp_path):
+    """A helper can be reaped while idle; the next request must not fail on it."""
+    procs: list[_FakeProc] = []
+    helper = _helper_with_fake_spawn(monkeypatch, procs)
+
+    helper.run(_answered_payload(tmp_path, "a.json"), timeout=5, cancel_flag=None)
+    procs[0].returncode = 1  # died between requests
+
+    result = helper.run(_answered_payload(tmp_path, "b.json"), timeout=5, cancel_flag=None)
+
+    assert result["ok"] is True
+    assert len(procs) == 2
+
+
+def test_helper_retries_once_when_the_pipe_breaks_on_write(monkeypatch, tmp_path):
+    """poll() can still say alive when the far end is already gone."""
+    procs: list[_FakeProc] = []
+    helper = _helper_with_fake_spawn(monkeypatch, procs)
+
+    helper.run(_answered_payload(tmp_path, "a.json"), timeout=5, cancel_flag=None)
+    procs[0].stdin.closed = True  # pipe gone, but poll() still reports running
+
+    result = helper.run(_answered_payload(tmp_path, "b.json"), timeout=5, cancel_flag=None)
+
+    assert result["ok"] is True
+    assert len(procs) == 2
+
+
+def test_helper_reports_a_helper_that_exits_without_answering(monkeypatch, tmp_path):
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    procs: list[_FakeProc] = []
+    helper = _helper_with_fake_spawn(monkeypatch, procs)
+
+    payload = {
+        "url": "https://example.com",
+        "retry": 1,
+        "result_path": str(tmp_path / "never-written.json"),
+    }
+
+    def _die_on_write(_self, proc, _line) -> None:
+        proc.returncode = 3
+
+    monkeypatch.setattr(internal_bypasser._BypassHelper, "_write", _die_on_write)
+
+    with pytest.raises(RuntimeError, match="exited without a result"):
+        helper.run(payload, timeout=5, cancel_flag=None)
+
+
+def test_helper_times_out_and_discards_the_wedged_process(monkeypatch, tmp_path):
+    procs: list[_FakeProc] = []
+    helper = _helper_with_fake_spawn(monkeypatch, procs)
+
+    payload = {
+        "url": "https://example.com",
+        "retry": 1,
+        "result_path": str(tmp_path / "never-written.json"),
+    }
+
+    with pytest.raises(TimeoutError):
+        helper.run(payload, timeout=0.05, cancel_flag=None)
+
+    assert helper._proc is None, "a wedged helper must not be handed to the next request"
+
+
+def test_idle_reaper_rearms_when_work_arrived_while_it_waited(monkeypatch, tmp_path):
+    """The timer fires on its own thread and can lose the race against a new request."""
+    import time
+
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    procs: list[_FakeProc] = []
+    helper = _helper_with_fake_spawn(monkeypatch, procs)
+    helper.run(_answered_payload(tmp_path), timeout=5, cancel_flag=None)
+
+    rearmed: list[bool] = []
+    monkeypatch.setattr(internal_bypasser._BypassHelper, "_idle_timeout", lambda _self: 3600.0)
+    monkeypatch.setattr(
+        internal_bypasser._BypassHelper, "_arm_idle_timer", lambda _self: rearmed.append(True)
+    )
+    helper._last_used = time.monotonic()
+
+    helper._reap_if_idle()
+
+    assert rearmed == [True]
+    assert helper._proc is not None, "helper was killed despite recent work"
+
+
+def test_idle_reaper_closes_a_genuinely_idle_helper(monkeypatch, tmp_path):
+    import time
+
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    procs: list[_FakeProc] = []
+    helper = _helper_with_fake_spawn(monkeypatch, procs)
+    helper.run(_answered_payload(tmp_path), timeout=5, cancel_flag=None)
+
+    monkeypatch.setattr(internal_bypasser._BypassHelper, "_idle_timeout", lambda _self: 60.0)
+    helper._last_used = time.monotonic() - 120
+
+    helper._reap_if_idle()
+
+    assert helper._proc is None
+    assert procs[0].stdin.closed
+
+
+def test_discard_tears_down_the_whole_session(monkeypatch, tmp_path):
+    """Dropping the helper must reach its browser tree, not just the helper itself.
+
+    The helper is a session leader (start_new_session), so a Chrome left behind by one
+    killed mid-bypass would keep a process group alive that the cleanup sweep is then not
+    allowed to reclaim - the leak #1231 was about.
+    """
+    procs: list[_FakeProc] = []
+    terminated: list[_FakeProc] = []
+    helper = _helper_with_fake_spawn(monkeypatch, procs, terminated)
+
+    helper.run(_answered_payload(tmp_path), timeout=5, cancel_flag=None)
+    helper._discard()
+
+    assert terminated == [procs[0]]
+
+
+def test_helper_asks_before_it_kills(monkeypatch, tmp_path):
+    """An idle helper should get to exit on its own; the kill is the fallback."""
+    procs: list[_FakeProc] = []
+    helper = _helper_with_fake_spawn(monkeypatch, procs)
+
+    helper.run(_answered_payload(tmp_path), timeout=5, cancel_flag=None)
+    helper._discard()
+
+    assert procs[0].stdin.closed, "stdin must be closed to end the helper's request loop"
+    assert procs[0].returncode == 0, "an idle helper should have exited on its own"
+
+
+def _bypass_with_recorded_driver(monkeypatch, get_impl):
+    """Wire up a bypass whose browser creation and closing are observable."""
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    driver = object()
+    closed: list[object] = []
+
+    async def _create(_url):
+        return driver
+
+    async def _close(drv):
+        closed.append(drv)
+
+    monkeypatch.setattr(internal_bypasser, "_create_cdp_browser", _create)
+    monkeypatch.setattr(internal_bypasser, "_get", get_impl)
+    monkeypatch.setattr(internal_bypasser, "_close_cdp_driver", _close)
+    return driver, closed
+
+
+def test_successful_bypass_closes_its_browser(monkeypatch):
+    """A living helper must not accumulate browsers: each bypass ends with Chrome gone."""
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    async def _get(_url, _driver, _cancel=None):
+        return "<html>ok</html>"
+
+    driver, closed = _bypass_with_recorded_driver(monkeypatch, _get)
+
+    result = internal_bypasser._run_bypass_in_current_process("https://example.com", 1)
+
+    assert result == "<html>ok</html>"
+    assert closed == [driver]
+
+
+def test_failed_bypass_closes_its_browser(monkeypatch):
+    """The same has to hold when the bypass raises on its way out."""
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    async def _get(_url, _driver, _cancel=None):
+        raise internal_bypasser.BypassCancelledError("cancelled")
+
+    driver, closed = _bypass_with_recorded_driver(monkeypatch, _get)
+
+    with pytest.raises(internal_bypasser.BypassCancelledError):
+        internal_bypasser._run_bypass_in_current_process("https://example.com", 1)
+
+    assert closed == [driver]
+
+
+def test_child_process_serves_every_line_it_is_given(monkeypatch, tmp_path):
+    """One helper, several requests: the loop is what saves the repeated process start."""
+    import io
+
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    urls: list[str] = []
+
+    def _fake_get(url, retry=None, cancel_flag=None):
+        urls.append(url)
+        return f"<html>{url}</html>"
+
+    requests = [
+        {"url": "https://example.com/one", "retry": 1, "result_path": str(tmp_path / "1.json")},
+        {"url": "https://example.com/two", "retry": 1, "result_path": str(tmp_path / "2.json")},
+    ]
+    stdin = io.StringIO("\n".join(json.dumps(request) for request in requests) + "\n")
+
+    monkeypatch.setattr(internal_bypasser, "get", _fake_get)
+    monkeypatch.setattr(internal_bypasser.sys, "stdin", stdin)
+
+    assert internal_bypasser._run_child_process() == 0
+    assert urls == ["https://example.com/one", "https://example.com/two"]
+
+    for index, request in enumerate(requests, start=1):
+        result = json.loads((tmp_path / f"{index}.json").read_text(encoding="utf-8"))
+        assert result["ok"] is True
+        assert result["html"] == f"<html>{request['url']}</html>"
+
+
+def test_child_process_keeps_serving_after_a_failed_request(monkeypatch, tmp_path):
+    """One failing URL must not take the helper - and everything queued - down."""
+    import io
+
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    def _fake_get(url, retry=None, cancel_flag=None):
+        if url.endswith("boom"):
+            raise RuntimeError("bypass exploded")
+        return "<html>ok</html>"
+
+    requests = [
+        {"url": "https://example.com/boom", "retry": 1, "result_path": str(tmp_path / "1.json")},
+        {"url": "https://example.com/fine", "retry": 1, "result_path": str(tmp_path / "2.json")},
+    ]
+    stdin = io.StringIO("\n".join(json.dumps(request) for request in requests) + "\n")
+
+    monkeypatch.setattr(internal_bypasser, "get", _fake_get)
+    monkeypatch.setattr(internal_bypasser.sys, "stdin", stdin)
+
+    assert internal_bypasser._run_child_process() == 0
+
+    failed = json.loads((tmp_path / "1.json").read_text(encoding="utf-8"))
+    assert failed["ok"] is False
+    assert failed["error"] == "bypass exploded"
+
+    served = json.loads((tmp_path / "2.json").read_text(encoding="utf-8"))
+    assert served["ok"] is True
+
+
+def test_result_file_becomes_visible_only_when_complete(tmp_path):
+    """The parent treats the file's existence as the answer, so no partial writes."""
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    result_path = tmp_path / "result.json"
+    internal_bypasser._publish_result(result_path, {"ok": True, "html": "<html/>"})
+
+    assert json.loads(result_path.read_text(encoding="utf-8"))["ok"] is True
+    assert list(tmp_path.iterdir()) == [result_path], "temporary file was left behind"
+
+
+def test_child_bypass_runs_on_the_long_lived_worker_loop(monkeypatch):
+    """A helper serving many requests must not build and close a loop per bypass.
+
+    asyncio.run() owns the loop for one call and closes it on the way out, which is why the
+    child goes through the worker unconditionally: one loop for the process's lifetime.
+    """
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    monkeypatch.setenv("SHELFMARK_INTERNAL_BYPASSER_CHILD", "1")
+
+    loops: list[asyncio.AbstractEventLoop] = []
+
+    async def _record_loop(_url, _driver, _cancel=None):
+        loops.append(asyncio.get_running_loop())
+        return "<html>ok</html>"
+
+    _bypass_with_recorded_driver(monkeypatch, _record_loop)
+
+    internal_bypasser._run_bypass_in_current_process("https://example.com", 1)
+    internal_bypasser._run_bypass_in_current_process("https://example.com", 1)
+
+    assert len(loops) == 2
+    assert loops[0] is loops[1], "second bypass ran on a different loop than the first"
+    assert not loops[0].is_closed()
+
+
+def test_child_bypass_carries_its_own_deadline(monkeypatch):
+    """The child bounds itself, rather than relying only on the parent's deadline."""
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    monkeypatch.setenv("SHELFMARK_INTERNAL_BYPASSER_CHILD", "1")
+
+    timeouts: list[float | None] = []
+    real_run = internal_bypasser._CDP_WORKER.run
+
+    def _record_timeout(coro, timeout=None):
+        timeouts.append(timeout)
+        return real_run(coro, timeout=timeout)
+
+    async def _get(_url, _driver, _cancel=None):
+        return "<html>ok</html>"
+
+    _bypass_with_recorded_driver(monkeypatch, _get)
+    monkeypatch.setattr(internal_bypasser._CDP_WORKER, "run", _record_timeout)
+
+    internal_bypasser._run_bypass_in_current_process("https://example.com", 1)
+
+    assert timeouts == [internal_bypasser._CHILD_BYPASS_TIMEOUT_SECONDS]
+
+
+def test_child_deadline_leaves_the_parent_room_to_hear_the_answer(monkeypatch):
+    """If the parent gave up first it could only kill the helper, losing a warm process."""
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    # The child's worst case is its deadline plus the grace it is given to close the
+    # browser after that deadline cancels the bypass, and all of it has to fit inside the
+    # parent's wait - otherwise the parent gives up first and kills a helper that was
+    # about to answer.
+    assert (
+        internal_bypasser._CHILD_BYPASS_TIMEOUT_SECONDS
+        + internal_bypasser._CDP_UNWIND_GRACE_SECONDS
+        < internal_bypasser._BYPASS_SUBPROCESS_TIMEOUT_SECONDS
+    )
+
+
+def test_in_process_bypass_keeps_the_parents_budget(monkeypatch):
+    """Non-Docker installs run in-process, where there is no helper to outlive anything."""
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    monkeypatch.delenv("SHELFMARK_INTERNAL_BYPASSER_CHILD", raising=False)
+
+    timeouts: list[float | None] = []
+    real_run = internal_bypasser._CDP_WORKER.run
+
+    def _record_timeout(coro, timeout=None):
+        timeouts.append(timeout)
+        return real_run(coro, timeout=timeout)
+
+    async def _get(_url, _driver, _cancel=None):
+        return "<html>ok</html>"
+
+    _bypass_with_recorded_driver(monkeypatch, _get)
+    monkeypatch.setattr(internal_bypasser._CDP_WORKER, "run", _record_timeout)
+
+    internal_bypasser._run_bypass_in_current_process("https://example.com", 1)
+
+    assert timeouts == [internal_bypasser._IN_PROCESS_BYPASS_TIMEOUT_SECONDS]
+
+
+def test_timed_out_bypass_finishes_unwinding_before_the_call_returns():
+    """A helper serving the next request must not race the browser teardown of the last.
+
+    The deadline cancels the bypass, but cancelling from the calling thread only schedules
+    that - it returns while `finally: await _close_cdp_driver(driver)` is still running.
+    In a helper that now outlives the request, the next bypass would open its Chrome on the
+    same loop while the abandoned one was still closing its own, sharing the DISPLAY
+    globals and one process group.
+    """
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    events: list[str] = []
+
+    async def _wedged():
+        try:
+            await asyncio.sleep(30)
+        finally:
+            # Teardown that yields, the way closing websockets and Chrome does.
+            await asyncio.sleep(0.05)
+            events.append("browser closed")
+
+    with pytest.raises(TimeoutError):
+        internal_bypasser._CDP_WORKER.run(_wedged(), timeout=0.1)
+
+    assert events == ["browser closed"], "run() returned before the bypass had unwound"
+
+
+def test_unwind_that_wedges_does_not_hold_the_caller_forever(monkeypatch):
+    """The grace is a bound, not a promise: cleanup can hang on a dead browser too."""
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    monkeypatch.setattr(internal_bypasser, "_CDP_UNWIND_GRACE_SECONDS", 0.1)
+
+    async def _wedged_on_both_ends():
+        try:
+            await asyncio.sleep(30)
+        finally:
+            await asyncio.sleep(30)
+
+    with pytest.raises(TimeoutError):
+        internal_bypasser._CDP_WORKER.run(_wedged_on_both_ends(), timeout=0.1)
+
+
+def test_cancelling_does_not_wait_out_the_shutdown_grace(monkeypatch, tmp_path):
+    """The grace only helps a helper that can still read its stdin.
+
+    One dropped mid-bypass is blocked inside the solve and will never reach its read loop,
+    so waiting it out cannot end in anything but the kill - while the user who asked to
+    cancel, and every bypass queued behind them on LOCKED, waits for it.
+    """
+    import threading
+
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    procs: list[_FakeProc] = []
+    helper = _helper_with_fake_spawn(monkeypatch, procs)
+
+    cancel_flag = threading.Event()
+    cancel_flag.set()
+    payload = {
+        "url": "https://example.com",
+        "retry": 1,
+        "result_path": str(tmp_path / "never-written.json"),
+    }
+
+    with pytest.raises(internal_bypasser.BypassCancelledError):
+        helper.run(payload, timeout=5, cancel_flag=cancel_flag)
+
+    assert not procs[0].waited, "a helper wedged mid-bypass was given the full exit grace"
+    assert procs[0].stdin.closed
+
+
+def test_idle_helper_still_gets_its_grace(monkeypatch, tmp_path):
+    """The reaper drops a helper that *is* in its read loop, and that one gets to exit."""
+    procs: list[_FakeProc] = []
+    helper = _helper_with_fake_spawn(monkeypatch, procs)
+
+    helper.run(_answered_payload(tmp_path), timeout=5, cancel_flag=None)
+    helper._discard()
+
+    assert procs[0].waited, "an idle helper should be asked to exit before being killed"
+
+
+def test_failed_request_leaves_no_result_files_behind(monkeypatch, tmp_path):
+    """Result paths are unique per request, so anything left is left for good."""
+    procs: list[_FakeProc] = []
+    helper = _helper_with_fake_spawn(monkeypatch, procs)
+
+    result_path = tmp_path / "result.json"
+    # A helper killed part-way through _publish_result leaves the staging file.
+    (tmp_path / "result.json.part").write_text('{"ok": tr', encoding="utf-8")
+    payload = {"url": "https://example.com", "retry": 1, "result_path": str(result_path)}
+
+    with pytest.raises(TimeoutError):
+        helper.run(payload, timeout=0.05, cancel_flag=None)
+
+    assert list(tmp_path.iterdir()) == []
+
+
+def test_child_does_not_export_cookies_left_by_an_earlier_request(monkeypatch, tmp_path):
+    """The parent owns the store; a warm helper must not push its own history back over it.
+
+    http.py purges a host's clearance the moment that host challenges a request carrying
+    it. A helper that kept its store across requests would still be holding the purged
+    cookies, and the next solve - for some entirely different host - would export them and
+    the parent would merge them straight back in.
+    """
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    def _solve_host(url, retry=None, cancel_flag=None):
+        # A solve fills the store for the host it solved, which is all it should report.
+        host = url.rsplit("/", 1)[-1]
+        internal_bypasser.import_store({host: {"cf_clearance": "fresh"}}, {host: "UA"})
+        return "<html>ok</html>"
+
+    monkeypatch.setattr(internal_bypasser, "get", _solve_host)
+    internal_bypasser.clear_cf_cookies()
+
+    for index, host in enumerate(("first.example", "second.example")):
+        internal_bypasser._handle_child_request(
+            json.dumps(
+                {
+                    "url": f"https://example.com/{host}",
+                    "retry": 1,
+                    "result_path": str(tmp_path / f"{index}.json"),
+                }
+            )
+        )
+
+    second = json.loads((tmp_path / "1.json").read_text(encoding="utf-8"))
+    assert list(second["cookies"]) == ["second.example"], (
+        "the helper exported clearance won by an earlier request"
+    )
+    assert list(second["user_agents"]) == ["second.example"]
+
+    internal_bypasser.clear_cf_cookies()
+
+
+def _record_dns_calls(monkeypatch):
+    """Stand in for the network module: report a resolver state, record changes to it.
+
+    A helper starts on system DNS, which is what the parent reports as "auto".
+    """
+    import shelfmark.bypass.internal_bypasser as internal_bypasser
+
+    calls: list[tuple] = []
+    state = {"provider": "auto", "servers": [], "doh_enabled": False}
+
+    def _set(provider, servers=None, use_doh=None):
+        calls.append((provider, servers, use_doh))
+        state.update({"provider": provider, "servers": servers or [], "doh_enabled": bool(use_doh)})
+
+    monkeypatch.setattr(internal_bypasser.network, "set_dns_provider", _set)
+    monkeypatch.setattr(internal_bypasser.network, "get_dns_config", lambda: dict(state))
+    return internal_bypasser, calls
+
+
+def test_helper_follows_the_parent_back_to_auto_dns(monkeypatch):
+    """A user flipping CUSTOM_DNS back to auto applies live - the helper has to hear it.
+
+    The old early-return on "auto" was correct only because a fresh helper had never been
+    told anything else. One that outlives the request has, and would go on resolving AA
+    through a resolver the parent has already abandoned.
+    """
+    internal_bypasser, calls = _record_dns_calls(monkeypatch)
+
+    internal_bypasser._apply_parent_dns_config(
+        {"provider": "cloudflare", "servers": [], "doh_enabled": True}
+    )
+    internal_bypasser._apply_parent_dns_config(
+        {"provider": "auto", "servers": [], "doh_enabled": False}
+    )
+
+    assert calls == [("cloudflare", None, True), ("auto", None, False)]
+
+
+def test_helper_does_not_reinitialize_dns_for_an_unchanged_config(monkeypatch):
+    """set_dns_provider() rebuilds resolvers; every request would pay for it otherwise."""
+    internal_bypasser, calls = _record_dns_calls(monkeypatch)
+
+    for _ in range(3):
+        internal_bypasser._apply_parent_dns_config(
+            {"provider": "quad9", "servers": [], "doh_enabled": True}
+        )
+
+    assert calls == [("quad9", None, True)]
+
+
+def test_fresh_helper_leaves_auto_dns_alone(monkeypatch):
+    """A helper starts on system DNS, which is what the parent reports as auto."""
+    internal_bypasser, calls = _record_dns_calls(monkeypatch)
+
+    internal_bypasser._apply_parent_dns_config(
+        {"provider": "auto", "servers": [], "doh_enabled": False}
+    )
+
+    assert calls == []
+
+
+def test_failed_dns_apply_is_retried_on_the_next_request(monkeypatch):
+    """A provider that did not land leaves the resolver where it was, so the next request
+    sees the same mismatch and tries again."""
+    internal_bypasser, calls = _record_dns_calls(monkeypatch)
+
+    def _explode(provider, servers=None, use_doh=None):
+        calls.append((provider, servers, use_doh))
+        msg = "resolver unreachable"
+        raise OSError(msg)
+
+    monkeypatch.setattr(internal_bypasser.network, "set_dns_provider", _explode)
+
+    config = {"provider": "google", "servers": [], "doh_enabled": True}
+    internal_bypasser._apply_parent_dns_config(config)
+    internal_bypasser._apply_parent_dns_config(config)
+
+    assert calls == [("google", None, True), ("google", None, True)]

tests/download/test_http_challenge_503.py

@@ -0,0 +1,127 @@
+"""Tests for handing a 503 that carries a browser challenge to the bypasser."""
+
+import requests
+
+_CHALLENGE_HTML = (
+    "<html><head><title>Checking your browser before accessing z-lib.gd</title>"
+    "<script src='/.well-known/ddos-guard/check.js'></script></head>"
+    "<body>Please wait...</body></html>"
+)
+
+
+class _FakeResponse:
+    """Minimal stand-in for requests.Response covering what html_get_page touches."""
+
+    def __init__(
+        self,
+        status_code: int,
+        *,
+        url: str = "https://z-lib.gd/md5/abc",
+        text: str = "",
+        cookies: dict[str, str] | None = None,
+    ) -> None:
+        self.status_code = status_code
+        self.url = url
+        self.text = text
+        self.cookies = cookies or {}
+        self.headers = {"Content-Type": "text/html;charset=utf-8"}
+        self.is_redirect = False
+
+    def raise_for_status(self) -> None:
+        if self.status_code >= 400:
+            error = requests.exceptions.HTTPError(f"{self.status_code} Error")
+            error.response = self
+            raise error
+
+
+def _neutralize_network(monkeypatch, http):
+    monkeypatch.setattr(http, "_apply_cf_bypass", lambda _url, _headers: {})
+    monkeypatch.setattr(http, "get_proxies", lambda _url: {})
+    monkeypatch.setattr(http, "get_ssl_verify", lambda _url: True)
+    monkeypatch.setattr(http.network, "should_rotate_dns_for_url", lambda _url: False)
+    monkeypatch.setattr(http.time, "sleep", lambda _seconds: None)
+
+
+def test_503_challenge_is_handed_to_the_bypasser(monkeypatch):
+    """The reissued-cookie 503 from #1233 reaches the bypasser instead of retrying."""
+    import shelfmark.download.http as http
+
+    _neutralize_network(monkeypatch, http)
+    monkeypatch.setattr(http, "_is_cf_bypass_enabled", lambda: True)
+
+    attempts: list[dict[str, str]] = []
+    bypassed: list[str] = []
+
+    def fake_get(_url: str, **kwargs):
+        attempts.append(dict(kwargs["cookies"]))
+        # Hit 1 issues the cookie; every later hit re-serves the challenge unchanged,
+        # which is what leaves the handshake with nothing to echo back.
+        if len(attempts) == 1:
+            return _FakeResponse(503, cookies={"bsrv": "1"})
+        return _FakeResponse(503, text=_CHALLENGE_HTML, cookies={"bsrv": "1"})
+
+    def fake_bypass(url: str, _selector=None, _cancel_flag=None):
+        bypassed.append(url)
+        return "<html>real page</html>"
+
+    monkeypatch.setattr(http.requests, "get", fake_get)
+    monkeypatch.setattr(http, "get_bypassed_page", fake_bypass)
+
+    html = http.html_get_page("https://z-lib.gd/md5/abc", retry=10, success_delay=0)
+
+    assert html == "<html>real page</html>"
+    assert bypassed == ["https://z-lib.gd/md5/abc"]
+    # The handshake still gets its echo; the challenge ends the loop on the second hit
+    # rather than burning all ten attempts.
+    assert attempts == [{}, {"bsrv": "1"}]
+
+
+def test_plain_503_still_retries_without_bypassing(monkeypatch):
+    """An overloaded origin has no challenge marker, so its retry path is untouched."""
+    import shelfmark.download.http as http
+
+    _neutralize_network(monkeypatch, http)
+    monkeypatch.setattr(http, "_is_cf_bypass_enabled", lambda: True)
+
+    attempts: list[dict[str, str]] = []
+    bypassed: list[str] = []
+
+    def fake_get(_url: str, **kwargs):
+        attempts.append(dict(kwargs["cookies"]))
+        return _FakeResponse(503, text="<html><body>Service Unavailable</body></html>")
+
+    monkeypatch.setattr(http.requests, "get", fake_get)
+    monkeypatch.setattr(
+        http, "get_bypassed_page", lambda url, *_a, **_k: bypassed.append(url) or ""
+    )
+
+    html = http.html_get_page("https://z-lib.gd/md5/abc", retry=3, success_delay=0)
+
+    assert html == ""
+    assert bypassed == []
+    assert attempts == [{}, {}, {}]
+
+
+def test_503_challenge_respects_disabled_bypasser_fallback(monkeypatch):
+    """Best-effort fetches must not stall on a minutes-long solve."""
+    import shelfmark.download.http as http
+
+    _neutralize_network(monkeypatch, http)
+    monkeypatch.setattr(http, "_is_cf_bypass_enabled", lambda: True)
+
+    bypassed: list[str] = []
+    monkeypatch.setattr(
+        http.requests,
+        "get",
+        lambda _url, **_kwargs: _FakeResponse(503, text=_CHALLENGE_HTML),
+    )
+    monkeypatch.setattr(
+        http, "get_bypassed_page", lambda url, *_a, **_k: bypassed.append(url) or ""
+    )
+
+    html = http.html_get_page(
+        "https://z-lib.gd/md5/abc", retry=1, success_delay=0, allow_bypasser_fallback=False
+    )
+
+    assert html == ""
+    assert bypassed == []

tests/metadata/test_hardcover_api_key.py

@@ -0,0 +1,53 @@
+import pytest
+
+from shelfmark.metadata_providers import hardcover
+from shelfmark.metadata_providers.hardcover import (
+    HardcoverProvider,
+    _test_hardcover_connection,
+)
+
+# Hardcover replaced its ~500 char JWTs with short opaque tokens.
+PAT = "hc_pat_" + "a" * 32
+
+
+@pytest.fixture(autouse=True)
+def _no_config_writes(monkeypatch):
+    """Keep the connection test from touching the on-disk provider config."""
+    monkeypatch.setattr(hardcover, "_save_connected_user", lambda user_id, username: None)
+
+
+class TestHardcoverApiKey:
+    def test_personal_access_token_is_accepted(self, monkeypatch):
+        monkeypatch.setattr(
+            HardcoverProvider,
+            "_execute_query",
+            lambda self, query, variables: {"me": [{"id": 1, "username": "alex"}]},
+        )
+
+        result = _test_hardcover_connection({"HARDCOVER_API_KEY": PAT})
+
+        assert result == {"success": True, "message": "Connected as: alex"}
+
+    def test_short_key_without_the_prefix_is_rejected(self):
+        result = _test_hardcover_connection({"HARDCOVER_API_KEY": "eyJhbGciOiJIUzI1NiJ9.short"})
+
+        assert result["success"] is False
+        assert "too short" in result["message"]
+
+    def test_short_prefixed_key_still_reaches_the_api(self, monkeypatch):
+        """A key wearing the hc_pat_ prefix is Hardcover's to accept or reject."""
+        monkeypatch.setattr(
+            HardcoverProvider,
+            "_execute_query",
+            lambda self, query, variables: None,
+        )
+
+        result = _test_hardcover_connection({"HARDCOVER_API_KEY": "hc_pat_ab"})
+
+        assert result == {"success": False, "message": "API request failed - check your API key"}
+
+    @pytest.mark.parametrize("pasted", [f"Bearer {PAT}", f"bearer {PAT}", f"  {PAT}  "])
+    def test_pasted_auth_header_noise_is_stripped(self, pasted):
+        provider = HardcoverProvider(api_key=pasted)
+
+        assert provider.session.headers["Authorization"] == f"Bearer {PAT}"

tests/prowlarr/test_handler.py

@@ -727,6 +727,62 @@ class TestProwlarrHandlerSeedCriteria:
             assert call_kwargs["ratio_limit"] == 1.25
 
 
+class TestProwlarrHandlerContentType:
+    """Regression tests for issue #1235 — content type must reach the client."""
+
+    def test_download_passes_content_type_to_client(self):
+        """rTorrent picks its audiobook label from content_type, not category."""
+        mock_client = MagicMock()
+        mock_client.name = "rtorrent"
+        mock_client.find_existing.return_value = None
+        mock_client.add_download.return_value = "download_id"
+
+        with (
+            patch(
+                "shelfmark.release_sources.prowlarr.handler.get_release",
+                return_value={
+                    "protocol": "torrent",
+                    "title": "Test Release",
+                    "magnetUrl": "magnet:?xt=urn:btih:abc123",
+                },
+            ),
+            patch(
+                "shelfmark.release_sources.prowlarr.handler.get_client",
+                return_value=mock_client,
+            ),
+            patch(
+                "shelfmark.release_sources.prowlarr.handler.remove_release",
+            ),
+            patch("shelfmark.release_sources.prowlarr.handler.config.get", return_value=True),
+            patch.object(
+                ProwlarrHandler,
+                "_poll_and_complete",
+                return_value=None,
+            ),
+        ):
+            handler = ProwlarrHandler()
+            task = DownloadTask(
+                task_id="content-type-pass-through",
+                source="prowlarr",
+                title="Test Audiobook",
+                content_type="audiobook",
+            )
+            cancel_flag = Event()
+            recorder = ProgressRecorder()
+
+            handler.download(
+                task=task,
+                cancel_flag=cancel_flag,
+                progress_callback=recorder.progress_callback,
+                status_callback=recorder.status_callback,
+            )
+
+            call_kwargs = mock_client.add_download.call_args.kwargs
+            assert call_kwargs["content_type"] == "audiobook"
+            # rTorrent gets no category, so content_type is its only audiobook signal.
+            assert call_kwargs["category"] is None
+
+
 class TestProwlarrHandlerExistingDownload:
     """Tests for handling existing downloads."""
 

tests/prowlarr/test_rtorrent_client.py

@@ -563,6 +563,36 @@ class TestRTorrentClientAudiobookLabel:
         assert "d.custom1.set=books" in args[2]
         assert "d.custom1.set=audiobooks" not in args[2]
 
+    def test_uses_audiobook_label_for_compound_content_type(self, monkeypatch):
+        """Issue #1235 — an exact "audiobook" match missed forms like "book (audiobook)"."""
+        config_values = {
+            "RTORRENT_URL": "http://localhost:8080/RPC2",
+            "RTORRENT_LABEL": "books",
+            "RTORRENT_AUDIOBOOK_LABEL": "audiobooks",
+            "RTORRENT_DOWNLOAD_DIR": "/downloads",
+        }
+        mock_rpc, mock_xmlrpc, mock_torrent_info = self._make_client(monkeypatch, config_values)
+
+        with patch.dict("sys.modules", {"xmlrpc.client": mock_xmlrpc}):
+            with patch(
+                "shelfmark.download.clients.torrent_utils.extract_torrent_info",
+                return_value=mock_torrent_info,
+            ):
+                if "shelfmark.download.clients.rtorrent" in sys.modules:
+                    del sys.modules["shelfmark.download.clients.rtorrent"]
+                from shelfmark.download.clients.rtorrent import RTorrentClient
+
+                client = RTorrentClient()
+                client.add_download(
+                    "magnet:?xt=urn:btih:abc123",
+                    "Test Audiobook",
+                    content_type="Book (Audiobook)",
+                )
+
+        args = mock_rpc.load.start.call_args[0]
+        assert "d.custom1.set=audiobooks" in args[2]
+        assert "d.custom1.set=books" not in args[2]
+
 
 class TestRTorrentClientGetStatus:
     """Tests for RTorrentClient.get_status()."""

uv.lock

@@ -1243,16 +1243,16 @@ wheels = [
 
 [[package]]
 name = "qbittorrent-api"
-version = "2026.8.0"
+version = "2026.8.1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "packaging" },
     { name = "requests" },
     { name = "urllib3" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/eb/d2/4a6d6ae6aaca1ba83039d27f28e5bf30bacef27ff4eb635c729d2d582363/qbittorrent_api-2026.8.0.tar.gz", hash = "sha256:4127a9c1d5c7ad9cd4e8de16f83daeea2e8b2ed962f99f284d238f927479391a", size = 1430678, upload-time = "2026-08-01T21:05:55.88Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/04/00/b7f41dfe8af6c3991f7182f9c14c46851379b8e66ac74c819f0952e9923d/qbittorrent_api-2026.8.1.tar.gz", hash = "sha256:9642c4528baee67216eab87022ce297048ac96df2f8c9f1cf920de078eb50b62", size = 1436760, upload-time = "2026-08-16T20:30:28.35Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/5b/81/b456688c20e75cae5964700b58712630930b9dfaf32ce7c6a8881658d0d0/qbittorrent_api-2026.8.0-py3-none-any.whl", hash = "sha256:03b3db598d351194fd3c65f1d3e53843f65f85bef1d5077791efaa2e26869d2b", size = 68196, upload-time = "2026-08-01T21:05:54.376Z" },
+    { url = "https://files.pythonhosted.org/packages/e6/60/77c7b63675f12ba492dee9f8aabab0fc61ced5d2131a83bd7e92dcc335c1/qbittorrent_api-2026.8.1-py3-none-any.whl", hash = "sha256:c17260e416ec5832c3ed7aff46fb0e2b47e70b90de182e89c34133b9d4ce5a6e", size = 71970, upload-time = "2026-08-16T20:30:26.67Z" },
 ]
 
 [[package]]
@@ -1516,7 +1516,7 @@ requires-dist = [
     { name = "python-socketio" },
     { name = "python-xlib", marker = "extra == 'browser'" },
     { name = "pyvirtualdisplay", marker = "extra == 'browser'" },
-    { name = "qbittorrent-api", specifier = ">=2026.8.0" },
+    { name = "qbittorrent-api", specifier = ">=2026.8.1" },
     { name = "rarfile" },
     { name = "requests", extras = ["socks"] },
     { name = "seleniumbase", marker = "extra == 'browser'", specifier = "==4.51.12" },