diff --git a/passiveworkers/research.py b/passiveworkers/research.py
index f4815aa..e3d3956 100644
--- a/passiveworkers/research.py
+++ b/passiveworkers/research.py
@@ -39,6 +39,7 @@
 import os
 import re
 import socket
+import threading
 import time
 from functools import lru_cache
 from urllib.parse import urlparse
@@ -106,7 +107,7 @@ def _host_is_public(host: str) -> bool:
 _MAX_REDIRECTS = 3
 
 
-def _guarded_get(url: str, timeout: float, deadline: float | None = None):
+def _guarded_get(url: str, timeout: float, deadline: float | None = None, session=None):
     """A streaming GET that re-validates the host on the INITIAL request AND on EVERY redirect hop.
     A public URL must never be able to 30x-redirect to a loopback/private/CGNAT/metadata address —
     so we disable requests' automatic redirects and follow them manually, bounded and re-checked.
@@ -114,13 +115,15 @@ def _guarded_get(url: str, timeout: float, deadline: float | None = None):
     asker-influenced fetch (page evidence, shard_map fetch:true, the download_extract task type).
     `deadline` (a time.monotonic() value) bounds the whole chain: TimeoutError once it passes, and
     each hop's per-read timeout is clipped to the time left. Every response is closed except the
-    one returned, which the caller owns and must close.
+    one returned, which the caller owns and must close. `session` (default: the requests module)
+    lets fetch_extract route every hop through its deadline-watched session.
     Residual (documented, Phase-3+): getaddrinfo runs again inside requests, so a DNS-rebinding
     attacker could in principle swap a public→private address between this check and the connect
     (TOCTOU); the allowlist narrows it, full IP-pinning needs a custom connector. The timeout is
-    per socket read, so a server dripping its HEADERS can still overrun the deadline inside
-    requests.get; the body read (fetch_extract) is checked against it between chunks."""
+    per socket read, so on its own a server dripping its HEADERS would overrun the deadline inside
+    the get; fetch_extract's _FetchWatchdog shuts the socket down at the deadline to stop that."""
     import requests
+    get = (session or requests).get
     cur = url
     for _hop in range(_MAX_REDIRECTS + 1):
         host = (urlparse(cur).hostname or "").lower()
@@ -132,8 +135,8 @@ def _guarded_get(url: str, timeout: float, deadline: float | None = None):
             if left <= 0:
                 raise TimeoutError(f"fetch deadline exceeded before {cur[:80]}")
             hop_timeout = min(timeout, left)
-        r = requests.get(cur, headers={"User-Agent": _UA}, timeout=hop_timeout, stream=True,
-                         allow_redirects=False)
+        r = get(cur, headers={"User-Agent": _UA}, timeout=hop_timeout, stream=True,
+                allow_redirects=False)
         if not (r.is_redirect or r.is_permanent_redirect):
             return r
         loc = r.headers.get("Location") or ""
@@ -758,6 +761,86 @@ def _key(item: dict) -> str:
 _HTML_LIKE = ("text/html", "application/xhtml+xml", "text/plain")
 
 
+class _FetchWatchdog:
+    """Makes _FETCH_DEADLINE a hard wall clock. Per-read timeouts only bound ONE recv, so a server
+    dripping a byte every few seconds through its status line/headers, or through a gzip header
+    comment (which urllib3's read1 loops over internally), would hold the fetch thread for hours.
+    At the deadline this shuts down every socket the fetch opened, so whatever recv is blocked
+    returns at once and the fetch fails; fetch_extract then raises TimeoutError. It shuts down a
+    dup() of each socket: shutdown acts on the shared connection, and the dup stays valid after
+    TLS wrapping detaches the original object."""
+
+    def __init__(self, deadline: float):
+        self.fired = threading.Event()
+        self._socks: list[socket.socket] = []
+        self._lock = threading.Lock()
+        self._timer = threading.Timer(max(0.0, deadline - time.monotonic()), self._fire)
+        self._timer.name = "pw-fetch-deadline"
+        self._timer.daemon = True
+        self._timer.start()
+
+    def track(self, sock: socket.socket) -> None:
+        dup = sock.dup()
+        with self._lock:
+            self._socks.append(dup)
+            if self.fired.is_set():   # connected after the deadline: cut it straight away
+                _shutdown(dup)
+
+    def _fire(self) -> None:
+        with self._lock:
+            self.fired.set()
+            for s in self._socks:
+                _shutdown(s)
+
+    def close(self) -> None:
+        # cancel AND join: the timer thread must never outlive fetch_extract (a daemon thread still
+        # alive at interpreter exit is a shutdown hazard). join() only waits out a _fire already in
+        # progress, which takes the lock briefly and makes non-blocking shutdown() calls.
+        self._timer.cancel()
+        self._timer.join()
+        with self._lock:
+            for s in self._socks:
+                s.close()
+            self._socks.clear()
+
+
+def _shutdown(sock: socket.socket) -> None:
+    try:
+        sock.shutdown(socket.SHUT_RDWR)
+    except OSError:
+        pass
+
+
+def _watched_session(watchdog: _FetchWatchdog):
+    """A requests Session whose connections hand each new socket to `watchdog`. The connection
+    class is swapped per pool instance (a subclass of whatever the pool uses, so HTTPS and proxy
+    pools keep their behaviour); the session is per-fetch, so nothing outside it is affected."""
+    import requests
+    from requests.adapters import HTTPAdapter
+
+    def watched(base):
+        class _Watched(base):
+            _pw_watched = True
+
+            def _new_conn(self):
+                sock = super()._new_conn()
+                watchdog.track(sock)
+                return sock
+        return _Watched
+
+    class _Adapter(HTTPAdapter):
+        def get_connection_with_tls_context(self, *a, **kw):
+            pool = super().get_connection_with_tls_context(*a, **kw)
+            if not getattr(pool.ConnectionCls, "_pw_watched", False):
+                pool.ConnectionCls = watched(pool.ConnectionCls)
+            return pool
+
+    sess = requests.Session()
+    sess.mount("http://", _Adapter())
+    sess.mount("https://", _Adapter())
+    return sess
+
+
 def fetch_extract(url: str, max_chars: int = 6000, with_date: bool = False):
     """One polite, SSRF-guarded fetch of a PUBLIC http(s) page → sanitized main text.
     Shared by the researcher (page evidence) and batch fetch shards. Raises on failure —
@@ -765,10 +848,27 @@ def fetch_extract(url: str, max_chars: int = 6000, with_date: bool = False):
     from passiveworkers.sanitize import clean
     # SSRF-guarded GET that re-validates the host on every redirect hop (a public URL must not be
     # able to bounce to an internal address). Raises on a non-public hop or too many redirects.
-    # One wall-clock deadline for redirects + body: `timeout` alone is per socket read, so a server
-    # dripping a byte every 14 s would otherwise hold this thread for hours (F8).
+    # One wall-clock deadline for redirects + headers + body: `timeout` alone is per socket read, so
+    # a server dripping a byte every 14 s would otherwise hold this thread for hours (F8). The
+    # checks below fail fast between hops/chunks; the watchdog cuts a recv that is still blocked.
     deadline = time.monotonic() + _FETCH_DEADLINE
-    r = _guarded_get(url, timeout=15, deadline=deadline)
+    watchdog = _FetchWatchdog(deadline)
+    try:
+        with _watched_session(watchdog) as sess:
+            raw = _fetch_body(url, deadline, sess, watchdog)
+    except Exception as e:
+        if watchdog.fired.is_set():
+            raise TimeoutError("fetch deadline exceeded") from e
+        raise
+    finally:
+        watchdog.close()
+    text, date = _extract_main(raw)
+    text = clean(text)[:max_chars]
+    return (text, date) if with_date else text
+
+
+def _fetch_body(url: str, deadline: float, sess, watchdog: _FetchWatchdog) -> str:
+    r = _guarded_get(url, timeout=15, deadline=deadline, session=sess)
     try:
         r.raise_for_status()
         # A PDF/image/binary body decoded as UTF-8 ("replace") still LOOKS like extracted text and
@@ -790,9 +890,8 @@ def fetch_extract(url: str, max_chars: int = 6000, with_date: bool = False):
                 break
             buf.append(chunk)
             size += len(chunk)
+        if watchdog.fired.is_set():   # a cut socket can look like a clean EOF: never keep that body
+            raise TimeoutError("fetch deadline exceeded while reading the body")
     finally:
         r.close()
-    raw = b"".join(buf)[:_FETCH_CAP].decode("utf-8", "replace")
-    text, date = _extract_main(raw)
-    text = clean(text)[:max_chars]
-    return (text, date) if with_date else text
+    return b"".join(buf)[:_FETCH_CAP].decode("utf-8", "replace")
diff --git a/tests/conftest.py b/tests/conftest.py
index 2814f04..ca85913 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -8,6 +8,8 @@
 from __future__ import annotations
 
 import importlib
+import sys
+import threading
 
 import pytest
 
@@ -73,6 +75,30 @@ def post(self, url, **kw):
         return _ShimResp(self.client.post(self._path(url), **self._kw(kw)))
 
 
+@pytest.fixture(autouse=True)
+def _no_real_baseline_threads(monkeypatch):
+    """Never start the coordinator's REAL baseline thread in tests. Every POST /jobs starts a
+    `pw-baseline` daemon thread (net/coordinator_app.py) that polls for up to 900 s, tries a local
+    Ollama on loopback and then prints. One still running when the suite ends can print into
+    interpreter shutdown, and CPython aborts with "Fatal Python error: _enter_buffered_busy"
+    (test (3.10), PR #48, CI run 38124448503). Setting PW_BASELINE_LOCAL_MODEL="" is not enough:
+    the thread still starts, polls and prints. A module-level patch would not survive the
+    importlib.reload() calls many tests make, so the guard sits on Thread.start and checks the
+    module's CURRENT _baseline_async. Raising RuntimeError uses _start_baseline's own "can't start
+    a thread" path, which releases the slot. Tests that swap in their own _baseline_async
+    (tests/test_coordinator_limits.py) still get real threads."""
+    real_start = threading.Thread.start
+
+    def start(self):
+        if self.name == "pw-baseline":
+            capp = sys.modules.get("passiveworkers.net.coordinator_app")
+            fn = getattr(capp, "_baseline_async", None)
+            if getattr(fn, "__module__", None) == "passiveworkers.net.coordinator_app":
+                raise RuntimeError("the real baseline thread is disabled in tests")
+        return real_start(self)
+    monkeypatch.setattr(threading.Thread, "start", start)
+
+
 @pytest.fixture
 def shim_factory():
     """Returns `make(client, base) -> RequestsShim` to patch onto a module's `requests`."""
diff --git a/tests/test_research.py b/tests/test_research.py
index 5d36376..acbd508 100644
--- a/tests/test_research.py
+++ b/tests/test_research.py
@@ -69,6 +69,14 @@ def close(self):
         self.closed = True
 
 
+def _patch_get(monkeypatch, fake):
+    """fetch_extract GETs through its deadline-watched requests.Session; route that (and the bare
+    requests.get _guarded_get uses by default) to `fake(url, **kw)`."""
+    import requests
+    monkeypatch.setattr(requests, "get", fake)
+    monkeypatch.setattr(requests.Session, "get", lambda self, url, **kw: fake(url, **kw))
+
+
 def test_guarded_get_blocks_redirect_to_internal(monkeypatch):
     # a public URL must NOT be able to 30x-redirect to a metadata/private address (SSRF)
     import pytest
@@ -98,18 +106,16 @@ def test_guarded_get_bounds_redirect_chains(monkeypatch):
 
 def test_fetch_extract_blocks_redirect_to_metadata(monkeypatch):
     import pytest
-    import requests
     monkeypatch.setattr(R, "_host_is_public", lambda h: h != "169.254.169.254")
-    monkeypatch.setattr(requests, "get", lambda url, **kw: _Resp(location="http://169.254.169.254/"))
+    _patch_get(monkeypatch, lambda url, **kw: _Resp(location="http://169.254.169.254/"))
     with pytest.raises(ValueError):
         R.fetch_extract("http://good.example/page")
 
 
 def test_fetch_extract_happy_path_through_guarded_get(monkeypatch):
-    import requests
     monkeypatch.setattr(R, "_host_is_public", lambda h: True)
-    monkeypatch.setattr(requests, "get",
-                        lambda url, **kw: _Resp(body=b"<html><body><p>Hello SSRF-safe</p></body></html>"))
+    _patch_get(monkeypatch,
+               lambda url, **kw: _Resp(body=b"<html><body><p>Hello SSRF-safe</p></body></html>"))
     assert "Hello SSRF-safe" in R.fetch_extract("http://good.example/page")
 
 
@@ -170,7 +176,7 @@ def boom():
         raise requests.HTTPError("503")
     bad.raise_for_status = boom
     for resp, err in ((ok, None), (pdf, ValueError), (bad, requests.HTTPError)):
-        monkeypatch.setattr(requests, "get", lambda url, _r=resp, **kw: _r)
+        _patch_get(monkeypatch, lambda url, _r=resp, **kw: _r)
         if err:
             with pytest.raises(err):
                 R.fetch_extract("http://good.example/page")
@@ -190,7 +196,6 @@ def monotonic(self):
 def test_fetch_extract_deadline_stops_a_drip_fed_body(monkeypatch):
     # every read returns one byte and "takes" 10 s — per-read timeouts never fire, the total must
     import pytest
-    import requests
     import types
     clock = _Clock()
     monkeypatch.setattr(R, "time", types.SimpleNamespace(monotonic=clock.monotonic))
@@ -204,7 +209,7 @@ def read1(self, n=-1, decode_content=True):
             return b"x"
     resp = _Resp()
     resp.raw = _Drip()
-    monkeypatch.setattr(requests, "get", lambda url, **kw: resp)
+    _patch_get(monkeypatch, lambda url, **kw: resp)
     with pytest.raises(TimeoutError):
         R.fetch_extract("http://good.example/slow")
     assert len(reads) <= R._FETCH_DEADLINE / 10 + 1 and resp.closed
@@ -266,6 +271,127 @@ def serve():
         srv.close()
 
 
+def _drip_server(first: bytes, drip: bytes, give_up: float = 8.0):
+    """Loopback stub: read the request, send `first`, then one `drip` every 50 ms until stopped or
+    `give_up` seconds pass (so old code fails the timing assert instead of hanging the suite)."""
+    import socket
+    import threading
+    srv = socket.socket()
+    srv.bind(("127.0.0.1", 0))
+    srv.listen(1)
+    stop = threading.Event()
+
+    def serve():
+        conn, _ = srv.accept()
+        with conn:
+            try:
+                conn.recv(65536)
+                conn.sendall(first)
+                t0 = time.monotonic()
+                while not stop.wait(0.05) and time.monotonic() - t0 < give_up:
+                    conn.sendall(drip)
+            except OSError:
+                pass
+    th = threading.Thread(target=serve, daemon=True)
+    th.start()
+    return srv, stop, th
+
+
+_DRIPS = {
+    # per-read timeouts never fire on any of these; only a hard wall clock stops them
+    "http_headers": ("http", b"HTTP/1.1 200 OK\r\nX-Pad: ", b"a"),
+    "gzip_header_comment": ("http", b"HTTP/1.1 200 OK\r\nContent-Type: text/html\r\n"
+                            b"Content-Encoding: gzip\r\nConnection: close\r\n\r\n"
+                            b"\x1f\x8b\x08\x10\x00\x00\x00\x00\x00\xff", b"c"),
+    # CPython already bounds a TLS handshake by one deadline; kept so the watched session can't regress it
+    "tls_handshake": ("https", b"\x16\x03\x03\x40\x00", b"\x00"),   # a 16 KB record, 1 byte at a time
+}
+
+
+def test_fetch_extract_deadline_cuts_a_drip_before_the_body(monkeypatch):
+    # The between-chunk check can't see these: the drip sits inside requests' header parse or
+    # urllib3's own read1 loop over a gzip header. The watchdog must cut them.
+    import pytest
+    monkeypatch.setenv("NO_PROXY", "127.0.0.1")   # a runner's HTTP(S)_PROXY must not catch loopback
+    monkeypatch.setattr(R, "_host_is_public", lambda h: True)
+    monkeypatch.setattr(R, "_FETCH_DEADLINE", 0.5)
+    for name, (scheme, first, drip) in _DRIPS.items():
+        srv, stop, th = _drip_server(first, drip)
+        t0 = time.monotonic()
+        try:
+            with pytest.raises(TimeoutError):
+                R.fetch_extract(f"{scheme}://127.0.0.1:{srv.getsockname()[1]}/")
+            assert time.monotonic() - t0 < 3.0, name
+        finally:
+            stop.set()
+            th.join(2)
+            srv.close()
+
+
+def test_fetch_watchdog_cuts_a_socket_connected_after_the_deadline():
+    import socket
+    a, b = socket.socketpair()
+    wd = R._FetchWatchdog(time.monotonic() - 1)
+    try:
+        assert wd.fired.wait(2)
+        wd.track(a)
+        a.settimeout(2)
+        assert a.recv(1) == b""          # shut down at once, not left to block
+    finally:
+        wd.close()
+        a.close()
+        b.close()
+
+
+def test_fetch_extract_real_page_through_the_watched_session(monkeypatch):
+    # happy path over a real loopback socket: the swapped connection class still fetches, and the
+    # watchdog leaves no timer thread and no dup()ed fd behind
+    import os
+    srv, stop, th = _drip_server(b"HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nContent-Length: 39"
+                                 b"\r\nConnection: close\r\n\r\n<html><body><p>served</p></body></html>",
+                                 b"", give_up=0)
+    monkeypatch.setenv("NO_PROXY", "127.0.0.1")
+    monkeypatch.setattr(R, "_host_is_public", lambda h: True)
+    fds = os.listdir("/proc/self/fd") if os.path.isdir("/proc/self/fd") else None
+    try:
+        assert "served" in R.fetch_extract(f"http://127.0.0.1:{srv.getsockname()[1]}/")
+    finally:
+        stop.set()
+        th.join(2)
+        srv.close()
+    assert not _watchdog_threads()   # joined in fetch_extract's finally: no polling needed
+    if fds is not None:
+        assert len(os.listdir("/proc/self/fd")) <= len(fds)
+
+
+def _watchdog_threads():
+    import threading
+    return [t for t in threading.enumerate() if t.name == "pw-fetch-deadline" and t.is_alive()]
+
+
+def test_fetch_watchdog_thread_never_outlives_fetch_extract(monkeypatch):
+    # A daemon timer still alive at interpreter exit is a shutdown hazard, so fetch_extract must
+    # cancel AND join it on every exit path: success, error, and the deadline actually firing.
+    import pytest
+    monkeypatch.setattr(R, "_host_is_public", lambda h: True)
+    _patch_get(monkeypatch, lambda url, **kw: _Resp(body=b"<html><body><p>fine</p></body></html>"))
+    assert "fine" in R.fetch_extract("http://good.example/page")
+    assert not _watchdog_threads()
+
+    def boom(url, **kw):
+        raise ConnectionError("refused")
+    _patch_get(monkeypatch, boom)
+    with pytest.raises(ConnectionError):
+        R.fetch_extract("http://good.example/page")
+    assert not _watchdog_threads()
+
+    monkeypatch.setattr(R, "_FETCH_DEADLINE", 0.0)   # already due: the timer fires (or is cancelled) mid-call
+    _patch_get(monkeypatch, lambda url, **kw: _Resp(body=b"<p>late</p>"))
+    with pytest.raises(TimeoutError):
+        R.fetch_extract("http://good.example/page")
+    assert not _watchdog_threads()
+
+
 # --------------------------------------------------------- keyed web backends (R33/D49)
 def _row(name):
     return [{"title": name, "href": "https://good.example/x", "body": name}]
@@ -307,31 +433,28 @@ def test_searxng_rejects_non_http_scheme(monkeypatch):
 # --------------------------------------------------------- content-type gate (R14 review)
 def test_fetch_extract_skips_pdf_content_type(monkeypatch):
     import pytest
-    import requests
     monkeypatch.setattr(R, "_host_is_public", lambda h: True)
     resp = _Resp(body=b"%PDF-1.4 binary garbage")
     resp.headers = {"Content-Type": "application/pdf"}
-    monkeypatch.setattr(requests, "get", lambda url, **kw: resp)
+    _patch_get(monkeypatch, lambda url, **kw: resp)
     with pytest.raises(ValueError):
         R.fetch_extract("http://good.example/doc.pdf")
 
 
 def test_fetch_extract_allows_text_plain(monkeypatch):
-    import requests
     monkeypatch.setattr(R, "_host_is_public", lambda h: True)
     resp = _Resp(body=b"plain text body")
     resp.headers = {"Content-Type": "text/plain; charset=utf-8"}
-    monkeypatch.setattr(requests, "get", lambda url, **kw: resp)
+    _patch_get(monkeypatch, lambda url, **kw: resp)
     assert "plain text body" in R.fetch_extract("http://good.example/page.txt")
 
 
 def test_fetch_extract_allows_missing_content_type(monkeypatch):
     # some legitimately HTML-serving hosts omit Content-Type — stay permissive (regression-pin
     # the existing happy-path behavior, which relies on this).
-    import requests
     monkeypatch.setattr(R, "_host_is_public", lambda h: True)
-    monkeypatch.setattr(requests, "get",
-                        lambda url, **kw: _Resp(body=b"<html><body><p>Hello</p></body></html>"))
+    _patch_get(monkeypatch,
+               lambda url, **kw: _Resp(body=b"<html><body><p>Hello</p></body></html>"))
     assert "Hello" in R.fetch_extract("http://good.example/page")