Gate librarian cog on a full ping round-trip, not a bare GET

The librarian health check was a plain GET to '/', which only proved
Flask was listening - not that the service could actually take a query,
run it through its internal queue+worker, and answer back. So the cog
could load against a librarian whose worker was wedged or that couldn't
reach the bot on the return leg.

Replace it with a ping that travels the SAME path a real search does, on
both sides:
  bot: QueryControl -> OUT_COMM_Q -> scan_queue -> awaiting_q
  librarian: POST /ping -> librarian_queue -> worker pulls it off
             (no Crossref/DOI search) -> pongs back with the same uuid
  bot: /conjurer -> incoming_q -> scan_incoming matches uuid, wakes waiter
The cog enables only when that whole loop closes within 3s. This also
proves the librarian->bot return path, which a GET never did.

Safety: uuid is random per ping; the wait and POST are both bounded so
startup can't stall; a pong that finds no waiter is dropped (never
orphaned into IN_COMM_Q, which would make the cog post a bogus 'no
results' message); and a ping whose pong never returns is swept out of
awaiting_q after PING_TTL_SECONDS so nothing leaks. All awaiting_q writes
stay within scan_queue (append) and scan_incoming (remove) - no locks,
no cross-thread mutation.

Integration tests cover: OK round-trip, timeout when accepted-but-no-pong,
unreachable, non-200, orphan-pong-dropped, and that real results still
reach IN_COMM_Q. Suite: 24 integration + 41 unit green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

ai_commands.py

@@ -427,6 +427,11 @@ class Events(commands.Cog):
                     return
                 message_content_lower = message_content_lower.replace("imaginuje sobie: ", "")
                 self.logger.debug("Wywolanie obrazka: %s", message_content_lower)
+                # Every error branch below must RETURN: otherwise control falls
+                # through to `if response:` with `response` unbound (the call
+                # raised) -> UnboundLocalError, crashing the handler right after
+                # the friendly message was already sent.
+                response = None
                 try:
                     response = await OPENAICLIENT.images.generate(
                         model="dall-e-3",
@@ -440,10 +445,12 @@ class Events(commands.Cog):
                     await discord_friendly_reply(
                         message, f"*Kondziu patrzy na terminal, czeka, czeka, czeka,.... Jeszcze chwile czeka Przypierdala w niego pięścią....* Nie mogę się połączyć z Openai spróbuj od nowa. *Na ekranie pojawia się*: {e}"
                     )
+                    return
                 except openai.APIConnectionError as e:
                     await discord_friendly_reply(
                         message, f"*Kondziu patrzy na terminal, chwile się zastanawia. Przypierdala w niego pięścią....* Nie mogę się połączyć z Openai. *Na ekranie pojawia się*: {e}"
                     )
+                    return
                 except openai.BadRequestError as e:
                     # Handle invalid request error, e.g. validate parameters or log
                     if message.author.nick:
@@ -461,27 +468,31 @@ class Events(commands.Cog):
                     await discord_friendly_reply(
                         message, f"Sorki, cenzura: {resp}. Jak chcesz to są kanały na nudle #sexy-foteczky i #kanal-do-fapania *Na ekranie pojawia się: {e}"
                     )
+                    return
                 except openai.AuthenticationError as e:
                     # Handle authentication error, e.g. check credentials or log
                     await discord_friendly_reply(
                         message, f"*Kondziu patrzy na terminal, chwile się zastanawia. Przypierdala w niego pięścią....* Wołaj szefa - coś się z hasłem zjebało. *Na terminalu pojawia się:* {e}"
                     )
+                    return
                 except openai.PermissionDeniedError as e:
                     # Handle permission error, e.g. check scope or log
+                    # (was accidentally passing a (message, text) TUPLE as one arg)
                     await discord_friendly_reply(
-                        (
-                            message, f"*Kondziu patrzy na terminal, chwile się zastanawia. Przypierdala w niego pięścią....* Wołaj szefa - coś się z uprawnieniami zjebało. *Na terminalu pojawia się:* {e}"
-                        )
+                        message, f"*Kondziu patrzy na terminal, chwile się zastanawia. Przypierdala w niego pięścią....* Wołaj szefa - coś się z uprawnieniami zjebało. *Na terminalu pojawia się:* {e}"
                     )
+                    return
                 except openai.RateLimitError as e:
                     await discord_friendly_reply(
                         message, f"*Kondziu patrzy na terminal* Wołaj szefa. Zapłacić rachunki za AI trzeba. Jak chcesz to się na #zebranie dorzuć. {e}"
                     )
+                    return
                 except openai.APIError as e:
                     # Handle API error, e.g. retry or log
                     await discord_friendly_reply(
                         message, f"*Kondziu nurkuje za bar, terminal wybucha. Przed tobą ląduje pergamin zapisany pięknym gotykiem a na nim*: {e}"
                     )
+                    return
                 if response:
                     self.logger.info(response)
                     image_url = response.data[0].url

ai_functions.py

@@ -541,10 +541,12 @@ async def get_random_cyclic_message(client):
     # trunk-ignore(bandit/B311)
     ai_check = random.randint(0, 10)
     logger.info("Losowa wypowiedź")
-    if ai_check < 2:
+    if ai_check < 2 and CYCLIC_WORDS:
         logger.info("Predefiniowana")
+        # randrange(n) is 0..n-1; randint(0, n) was inclusive and could return n
+        # -> list(...)[n] IndexError. Guarded on empty CYCLIC_WORDS above.
         # trunk-ignore(bandit/B311)
-        messnum = random.randint(0, len(CYCLIC_WORDS))
+        messnum = random.randrange(len(CYCLIC_WORDS))
         logger.debug(messnum)
         logger.debug(len(CYCLIC_WORDS))
         mess_key = list(CYCLIC_WORDS.keys())[messnum]

bot.py

@@ -30,17 +30,23 @@ import discord
 import requests
 from discord.ext import commands
 
-from communication_subroutine import comm_subroutine
+from communication_subroutine import comm_subroutine, librarian_ping
 from constants import (
     ENCODING,
     FILE_SERVICE_ADDRESS,
     GET_MP3,
+    LIBRARIAN_PING,
     LIBRARIAN_SERVICE_ADDRESS,
     LOGFILE,
     RADIO_SERVICE_ADDRESS,
     TOKEN,
+    service_headers,
 )
 
+# Round-trip health check budget for the librarian ping (bot -> librarian
+# internal queue -> pong back). Deliberately short so startup never stalls.
+LIBRARIAN_PING_TIMEOUT = 3.0
+
 logger = logging.getLogger("discord")
 logger.setLevel(logging.DEBUG)
 formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
@@ -110,9 +116,11 @@ SERVICE_EXTENSION_GROUPS = {
         "health_url": f"{RADIO_SERVICE_ADDRESS}/ping",
         "extensions": ["radio_commands"],
     },
-    # librarian: DOI / Crossref search
+    # librarian: DOI / Crossref search. NOTE: this one is NOT a plain GET - it
+    # is gated on a full ping round-trip (see _load_service_groups); the URL
+    # here is only the label used in the "unreachable" log line.
     "librarian": {
-        "health_url": f"{LIBRARIAN_SERVICE_ADDRESS}/",
+        "health_url": f"{LIBRARIAN_SERVICE_ADDRESS}{LIBRARIAN_PING}",
         "extensions": ["librarian_commands"],
     },
 }
@@ -120,13 +128,15 @@ SERVICE_EXTENSION_GROUPS = {
 SERVICE_RECHECK_SECONDS = 300
 
 
-def _service_alive(url: str) -> bool:
-    """True when the service answers HTTP at all (any status code counts)."""
+def _service_health(url: str):
+    """Return None when the service answers HTTP at all (any status counts),
+    otherwise the connection error explaining WHY it's unreachable (refused vs
+    timeout vs DNS - the difference points straight at the cause)."""
     try:
         requests.get(url, timeout=3)
-        return True
-    except requests.exceptions.RequestException:
-        return False
+        return None
+    except requests.exceptions.RequestException as exc:
+        return exc
 
 
 async def _load_extension_safe(name: str) -> bool:
@@ -152,15 +162,37 @@ async def _load_service_groups() -> bool:
         missing = [e for e in group["extensions"] if e not in client.extensions]
         if not missing:
             continue
-        alive = await asyncio.to_thread(_service_alive, group["health_url"])
-        if not alive:
-            logger.warning(
-                "Service '%s' unreachable (%s) - cogs stay disabled: %s",
-                service,
-                group["health_url"],
-                ", ".join(missing),
+        if service == "librarian":
+            # A plain GET only proves Flask is up. The librarian is only useful
+            # once its internal queue + worker are flowing, so prove exactly that
+            # with a ping that must complete the full round-trip (see
+            # communication_subroutine.librarian_ping).
+            alive = await asyncio.to_thread(
+                librarian_ping,
+                LIBRARIAN_SERVICE_ADDRESS,
+                LIBRARIAN_PING,
+                service_headers(),
+                LIBRARIAN_PING_TIMEOUT,
             )
-            continue
+            if not alive:
+                logger.warning(
+                    "Service 'librarian' ping round-trip failed (%s) - cogs stay disabled: %s",
+                    group["health_url"],
+                    ", ".join(missing),
+                )
+                continue
+        else:
+            err = await asyncio.to_thread(_service_health, group["health_url"])
+            if err is not None:
+                logger.warning(
+                    "Service '%s' unreachable (%s) [%s: %s] - cogs stay disabled: %s",
+                    service,
+                    group["health_url"],
+                    type(err).__name__,
+                    err,
+                    ", ".join(missing),
+                )
+                continue
         logger.info("Service '%s' is alive - enabling: %s", service, ", ".join(missing))
         for extension in missing:
             if await _load_extension_safe(extension):
@@ -204,6 +236,16 @@ async def on_ready():
     for extension in CORE_EXTENSIONS:
         await _load_extension_safe(extension)
 
+    # Log the ACTUALLY-resolved service addresses. When one shows the built-in
+    # default (192.168.1.15:5000) it means the matching CONJURER_* env var never
+    # reached the process - the single most common cause of "service unreachable"
+    # confusion. Printing them makes env-vs-default obvious at a glance.
+    logger.info(
+        "Resolved service addresses -> musician(file): %s | librarian: %s | radio: %s",
+        FILE_SERVICE_ADDRESS,
+        LIBRARIAN_SERVICE_ADDRESS,
+        RADIO_SERVICE_ADDRESS,
+    )
     await _load_service_groups()
     logger.info("Sensors: online")
 

communication_subroutine.py

@@ -4,10 +4,12 @@ import os
 import re
 import threading
 import time
+import uuid as uuidlib
 from queue import Empty, Queue
 from typing import Optional
 from urllib import request as urequest
 
+import requests
 from flask import Flask, abort, jsonify, request
 from waitress import serve
 
@@ -26,6 +28,11 @@ SRCHTITLE = re.compile(rb"StreamTitle=\\*(?P<title>[^;]*);").search
 
 awaiting_q = []
 incoming_q = Queue()
+# A health-check ping whose pong never comes back (dead/dropped librarian) would
+# otherwise leave its record in awaiting_q forever. scan_incoming sweeps ping
+# records older than this. Kept well above the ping timeout so a slow-but-alive
+# round-trip is never swept out from under a waiter.
+PING_TTL_SECONDS = 30
 app = Flask(__name__)
 
 
@@ -232,17 +239,50 @@ def scan_incoming(stop_event: Optional[threading.Event] = None):
         if stop_event and stop_event.is_set():
             logger.info("scan_incoming: stop requested")
             break
+        # Sweep stale health-check pings first: if a librarian is dead the pong
+        # never arrives, so drop ping records past their TTL. scan_incoming is
+        # the sole remover of awaiting_q, so this needs no lock (scan_queue only
+        # appends). Snapshot with list() so removal during iteration is safe.
+        now = time.monotonic()
+        for record in list(awaiting_q):
+            if (
+                getattr(record, "is_ping", False)
+                and getattr(record, "answered", None) is not None
+                and not record.answered.is_set()
+                and now - getattr(record, "created", now) > PING_TTL_SECONDS
+            ):
+                awaiting_q.remove(record)
         try:
             answer = incoming_q.get(block=False)
             logger.info("DATA FOUND")
-            record_stored = False
+            # Health-check pong (shape: {"__pong__": uuid}). Close the waiter's
+            # event and drop its record. It must NEVER fall through to the
+            # result/orphan path below, or the librarian cog would later pull it
+            # off IN_COMM_Q and post a bogus "no results" message to Discord.
+            if isinstance(answer, dict) and "__pong__" in answer:
+                pong_uuid = answer["__pong__"]
+                for record in list(awaiting_q):
+                    if getattr(record, "uuid", None) == pong_uuid:
+                        event = getattr(record, "answered", None)
+                        if event is not None:
+                            event.set()
+                        awaiting_q.remove(record)
+                logger.info("PONG matched for %s", pong_uuid)
+                continue
+            # Collect matched records and drop them from awaiting_q afterwards -
+            # they used to stay forever (awaiting_q only ever grew), leaking
+            # memory over the bot's uptime and letting a reused UUID re-match a
+            # stale record.
+            matched = []
             for record in awaiting_q:
                 if record.uuid in answer.keys():
-                    record_stored = True
                     record.stop = True
                     record.entries = answer[record.uuid]
                     IN_COMM_Q.put(record)
-            if not record_stored:
+                    matched.append(record)
+            for record in matched:
+                awaiting_q.remove(record)
+            if not matched:
                 for key in answer.keys():
                     record = QueryControl("Orphaned", key, "Orphan", None)
                     record.stop = True
@@ -252,6 +292,55 @@ def scan_incoming(stop_event: Optional[threading.Event] = None):
             time.sleep(1)
 
 
+def librarian_ping(address: str, endpoint: str, headers: Optional[dict] = None,
+                   timeout: float = 3.0) -> bool:
+    """Health-check the librarian by round-tripping a ping through the FULL path.
+
+    The ping is a pseudo-query that exercises exactly the same machinery a real
+    search does, on BOTH sides:
+
+    * bot out: a ``QueryControl`` rides ``OUT_COMM_Q`` -> ``scan_queue`` ->
+      ``awaiting_q`` just like a real query,
+    * librarian: it is POSTed to the librarian, which must pull it off its OWN
+      internal queue and answer WITHOUT running a search,
+    * bot in: the pong comes back over ``/conjurer`` -> ``incoming_q`` ->
+      ``scan_incoming``, which matches it by uuid and sets our event.
+
+    Returns True only when that whole loop closes within ``timeout``. Never
+    blocks longer than roughly ``timeout`` and cannot deadlock: the POST is
+    bounded, the wait is bounded, and a ping whose pong never arrives is swept
+    out of ``awaiting_q`` by ``scan_incoming`` (PING_TTL_SECONDS).
+    """
+    logger = logging.getLogger("discord")
+    ping_uuid = str(uuidlib.uuid4())
+    answered = threading.Event()
+    query = QueryControl("healthcheck", ping_uuid, "__ping__", None)
+    query.is_ping = True
+    query.answered = answered
+    query.created = time.monotonic()
+    # Enter the bot-side comm queue BEFORE the POST, so the record is already in
+    # awaiting_q by the time the pong can come back (no lost-wakeup race).
+    OUT_COMM_Q.put(query)
+    try:
+        response = requests.post(
+            f"{address}{endpoint}",
+            json={"UUID": ping_uuid},
+            headers=headers or {},
+            timeout=timeout,
+        )
+    except requests.exceptions.RequestException as exc:
+        logger.info("Librarian ping POST failed (%s): %s", ping_uuid, exc)
+        return False  # stale record is swept by scan_incoming
+    if response.status_code != 200:
+        logger.info("Librarian ping rejected (%s): HTTP %s", ping_uuid, response.status_code)
+        return False
+    if answered.wait(timeout):
+        logger.info("Librarian ping round-trip OK (%s)", ping_uuid)
+        return True
+    logger.info("Librarian ping timed out after %ss (%s)", timeout, ping_uuid)
+    return False
+
+
 def get_stream_title(tag: bytes) -> str:
     title = ""
     if m := SRCHTITLE(tag):
@@ -270,10 +359,13 @@ def id3(url: str) -> dict:
         resp.read(
             metaint
         )  # this isn't seekable so, arbitrarily read to the point we want
+        # Guard the headers: an Icecast stream that omits icy-name / icy-genre
+        # (e.g. while the radio is down) made `.title()` raise AttributeError on
+        # None, 500-ing the /prepped_tracks "next" handler that calls this.
         tagdata = dict(
             site_url=resp.headers.get("icy-url"),
-            name=resp.headers.get("icy-name").title(),
-            genre=resp.headers.get("icy-genre").title(),
+            name=(resp.headers.get("icy-name") or "").title(),
+            genre=(resp.headers.get("icy-genre") or "").title(),
             title=get_stream_title(resp.read(255)),
         )
         return tagdata

conjurer_betoniarka/betoniarka.py

@@ -203,6 +203,12 @@ def wyszukaj(word_list, how_many, _logger=None, write_to=None):
 # ---------------------------------------------------------------- tailer
 def scan_tracks():
     """Tail the radio logs and forward play events to the bot."""
+    # On a fresh deploy Liquidsoap may not have written its logs yet; wait for
+    # them instead of dying with FileNotFoundError, which used to silently kill
+    # the now-playing forwarder until the container was restarted.
+    while not (RADIOLOG_PATH.exists() and PERSISTENCE_PATH.exists()):
+        logger.info("Waiting for radio logs (%s, %s)...", RADIOLOG_PATH, PERSISTENCE_PATH)
+        time.sleep(5)
     with open(RADIOLOG_PATH, "r", encoding=ENCODING) as log_file:
         log_file.seek(os.stat(RADIOLOG_PATH).st_size)
         prev_size = os.stat(PERSISTENCE_PATH).st_size

conjurer_librarian/conjurer_librarian.py

@@ -411,7 +411,30 @@ class BackgroundTaskSearch(threading.Thread):
         while True:
             database = None
             ndb_database = None
-            librarian = librarian_queue.get()
+            item = librarian_queue.get()
+            # Health-check ping: it has flowed through the internal queue and is
+            # now pulled off it - that is the whole point. Pong it straight back
+            # with the same uuid and DO NOT run a search.
+            if isinstance(item, dict) and "__ping__" in item:
+                ping_uuid = item["__ping__"]
+                self.app.logger.info(
+                    "PING %s pulled off internal queue - ponging back (no search)",
+                    ping_uuid,
+                )
+                try:
+                    await asyncio.to_thread(
+                        requests.post,
+                        f"{MAIN_BOT_ADDRESS}{SEND_RESULTS}",
+                        json={"__pong__": ping_uuid},
+                        headers=_service_headers(),
+                        timeout=5,
+                    )
+                except requests.exceptions.RequestException as exc:
+                    self.app.logger.warning(
+                        "PING pong send failed for %s: %s", ping_uuid, exc
+                    )
+                continue
+            librarian = item
             self.app.logger.info("STARTED")
             result = await librarian.answer_query(librarian.deep_search)
             result = {librarian.uuid: result}
@@ -530,6 +553,29 @@ async def query_database():
     return return_data
 
 
+@app.route("/ping", methods=["POST"])
+async def ping_roundtrip():
+    _authorize_request()
+    """
+    Health-check round-trip.
+
+    Puts a lightweight ping marker onto the SAME internal ``librarian_queue``
+    that real searches go through and returns 200 immediately. The background
+    worker pulls it off the queue and pongs it back to the bot with the same
+    uuid, WITHOUT running any Crossref/DOI search. A successful pong therefore
+    proves the whole pipeline (HTTP in -> internal queue -> worker -> HTTP out)
+    is flowing, not just that Flask is up.
+    """
+    record = json.loads(request.data)
+    ping_uuid = record["UUID"]
+    app.logger.info("PING received %s - queued for round-trip", ping_uuid)
+    librarian_queue.put({"__ping__": ping_uuid})
+    return (
+        jsonify(isError=False, message="ping-queued", statusCode=200, data=ping_uuid),
+        200,
+    )
+
+
 @app.route("/get_partial_result", methods=["POST"])
 async def get_partial():
     _authorize_request()

conjurer_librarian/search_bot.py

@@ -148,8 +148,11 @@ def consumer(in_q, control_q, doi, live_results, result_list, control_dict, expe
     print(f"Consumer thread started: {no} no")
     empty_counter = 0
     alive_no = 0
+    # DOI -> result item, so a line is matched with one O(1) dict lookup instead
+    # of scanning every queried DOI. Items are shared with result_list, so
+    # setting exists here is seen by everyone.
+    doi_index = {item["DOI"]: item for item in result_list}
     while True:
-        done_check = True
         try:
             data = in_q.get(block=True, timeout = 1)
             if data is _sentinel:
@@ -161,16 +164,21 @@ def consumer(in_q, control_q, doi, live_results, result_list, control_dict, expe
                 alive_no += 1
                 print(f"C{no}__{alive_no}\r", end="")
 
-                for item in result_list:
-                    if item["DOI"] in data and not item["exists"]:
-                        print(f"HIT in {no} content {data[0]} line {data[1]} file {data[2]} {item['exists']}")
-                        _logger.info(data)
-                        _logger.info("HIT")
-                        item["exists"] = True
-                        live_results.append(item)
-                    done_check = done_check and item["exists"]
-                if done_check:
-                    control_q.put(_sentinel)
+                # Each DB line is a DOI (optionally followed by metadata). Match
+                # the WHOLE first token exactly - the old `item["DOI"] in data`
+                # was a substring test, so a DOI that is a prefix of a longer one
+                # (10.1/1 vs 10.1/12) produced a false 'exists' hit.
+                parts = data.split()
+                line_doi = parts[0] if parts else ""
+                item = doi_index.get(line_doi)
+                if item is not None and not item["exists"]:
+                    print(f"HIT in {no}: {line_doi}")
+                    _logger.info("HIT %s", line_doi)
+                    item["exists"] = True
+                    live_results.append(item)
+                    # All found? Signal producers to stop early (rare -> cheap).
+                    if all(it["exists"] for it in result_list):
+                        control_q.put(_sentinel)
         except Empty:
             empty_counter += 1
             time.sleep(1)

constants.py

@@ -90,6 +90,9 @@ CREATE_PRIO_PLAYLIST = "/create_priority_playlist"
 REQUEST_MUSIC = "/request_radio_file"
 CLEAR_PRIO = "/clear_pr_pls"
 SEND_QUERY = "/query"
+# Health-check round-trip: a pseudo-query that the librarian must pull off its
+# own internal queue and answer (same uuid) WITHOUT running a real search.
+LIBRARIAN_PING = "/ping"
 TIME_BETWEEN_CALLS = 100000
 LAST_SPONTANEOUS_CALL = datetime.now()
 

librarian_commands.py

@@ -52,8 +52,13 @@ class DataModule(commands.Cog):
                 # check if current path is a file
                 if os.path.isfile(os.path.join(DIR_PATH_SADOX, path)):
                     res.append(path)
+            if not res:
+                await ctx.send("*Conjurer grzebie w pustej skrzyni* Nie ma dziś żadnych komiksów.")
+                return
+            # randrange(len) is 0..len-1; the old randrange(0, len-1) never picked
+            # the last file and raised ValueError('empty range') on a single file.
             # trunk-ignore(bandit/B311)
-            filename = res[random.randrange(0, len(res) - 1)]
+            filename = res[random.randrange(len(res))]
             # select random page
             file = open(DIR_PATH_SADOX + filename, "rb")
             if  True:

tests/integration/test_librarian_ping.py

@@ -0,0 +1,121 @@
+"""Integration: the librarian health check is a full comm round-trip, not a GET.
+
+``librarian_ping`` injects a pseudo-query into the SAME machinery a real search
+uses - it rides ``OUT_COMM_Q`` -> ``scan_queue`` -> ``awaiting_q``, the librarian
+is expected to pull it off its own queue and pong it back over ``/conjurer`` ->
+``incoming_q`` -> ``scan_incoming``, which matches it by uuid and wakes the
+waiter. These tests stand in for the librarian with a stubbed ``requests.post``
+and assert the loop closes (and, crucially, that a pong never leaks into
+``IN_COMM_Q`` where the librarian cog would mistake it for a real result).
+"""
+import threading
+import time
+
+import pytest
+
+import communication_subroutine as cs
+
+
+def _drain(queue):
+    while not queue.empty():
+        queue.get()
+
+
+@pytest.fixture
+def comm_threads():
+    """Run scan_queue + scan_incoming (the two workers librarian_ping relies on)
+    for the duration of a test, on cleared shared state."""
+    cs.awaiting_q.clear()
+    _drain(cs.incoming_q)
+    _drain(cs.OUT_COMM_Q)
+    _drain(cs.IN_COMM_Q)
+    stop = threading.Event()
+    workers = [
+        threading.Thread(target=cs.scan_queue, kwargs={"stop_event": stop}, daemon=True),
+        threading.Thread(target=cs.scan_incoming, kwargs={"stop_event": stop}, daemon=True),
+    ]
+    for worker in workers:
+        worker.start()
+    yield
+    stop.set()
+    for worker in workers:
+        worker.join(timeout=3)
+
+
+def _await_in_awaiting(ping_uuid, timeout=2):
+    """Block until scan_queue has moved the ping into awaiting_q."""
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if any(getattr(r, "uuid", None) == ping_uuid for r in list(cs.awaiting_q)):
+            return True
+        time.sleep(0.01)
+    return False
+
+
+class _Resp:
+    def __init__(self, status_code=200):
+        self.status_code = status_code
+
+
+def test_ping_round_trip_ok(comm_threads, monkeypatch):
+    # Stub librarian: only pong AFTER the record is in awaiting_q, mirroring the
+    # real network latency that always lets scan_queue win.
+    def fake_post(url, json=None, headers=None, timeout=None):
+        ping_uuid = json["UUID"]
+        _await_in_awaiting(ping_uuid)
+        cs.incoming_q.put({"__pong__": ping_uuid})
+        return _Resp(200)
+
+    monkeypatch.setattr(cs.requests, "post", fake_post)
+
+    assert cs.librarian_ping("http://lib", "/ping", {}, timeout=3.0) is True
+    # A pong must NEVER reach the cog's inbound queue...
+    assert cs.IN_COMM_Q.empty()
+    # ...and the ping record must be cleaned out of awaiting_q.
+    assert not any(getattr(r, "is_ping", False) for r in list(cs.awaiting_q))
+
+
+def test_ping_times_out_when_librarian_accepts_but_never_pongs(comm_threads, monkeypatch):
+    monkeypatch.setattr(cs.requests, "post", lambda *a, **k: _Resp(200))
+    start = time.monotonic()
+    assert cs.librarian_ping("http://lib", "/ping", {}, timeout=0.3) is False
+    # Bounded: it must not block much beyond the timeout.
+    assert time.monotonic() - start < 2.0
+
+
+def test_ping_false_when_librarian_unreachable(comm_threads, monkeypatch):
+    def boom(*a, **k):
+        raise cs.requests.exceptions.RequestException("no route to host")
+
+    monkeypatch.setattr(cs.requests, "post", boom)
+    assert cs.librarian_ping("http://lib", "/ping", {}, timeout=0.3) is False
+
+
+def test_ping_false_on_non_200(comm_threads, monkeypatch):
+    monkeypatch.setattr(cs.requests, "post", lambda *a, **k: _Resp(503))
+    assert cs.librarian_ping("http://lib", "/ping", {}, timeout=0.3) is False
+
+
+def test_orphan_pong_is_dropped_not_enqueued(comm_threads):
+    # A pong with no matching waiter (e.g. after the ping already timed out) must
+    # be silently dropped - never turned into an "Orphaned" IN_COMM_Q record that
+    # makes the librarian cog post a bogus "no results" message.
+    cs.incoming_q.put({"__pong__": "no-such-uuid"})
+    deadline = time.time() + 3
+    while time.time() < deadline and not cs.incoming_q.empty():
+        time.sleep(0.02)
+    time.sleep(0.2)  # give scan_incoming a beat to (not) enqueue anything
+    assert cs.IN_COMM_Q.empty()
+
+
+def test_real_result_still_reaches_in_comm_q(comm_threads):
+    # Guard the existing path: a normal {uuid: {...}} result must still match its
+    # QueryControl and land in IN_COMM_Q for the cog to render.
+    query = cs.QueryControl("user", "real-uuid", "jakieś zapytanie", None)
+    cs.OUT_COMM_Q.put(query)
+    assert _await_in_awaiting("real-uuid")
+    cs.incoming_q.put({"real-uuid": {"10.1/x": {"Title": ["Tytuł"], "type": "article"}}})
+    got = cs.IN_COMM_Q.get(timeout=3)
+    assert got.uuid == "real-uuid"
+    assert got.stop is True
+    assert "10.1/x" in got.entries

tests/unit/test_search_bot.py

@@ -94,6 +94,33 @@ def test_survives_invalid_utf8_byte_and_still_finds_later_doi(tmp_path, monkeypa
     assert hit, "DOI after the bad byte was not found - the file was aborted mid-read"
 
 
+def test_doi_match_is_exact_not_substring(tmp_path, monkeypatch):
+    # A DB line "10.1/12" must NOT satisfy a search for "10.1/1" (the old
+    # `doi in line` substring test did). The exact DOI must still be found.
+    monkeypatch.setattr(search_bot, "DATABASE_PATH", str(tmp_path) + "/")
+    (tmp_path / "0_chunk.txt").write_text(
+        "10.1/12\n10.1/1\n10.2/999\n", encoding="utf-8"
+    )
+
+    finished, result = _run_bounded([("10.1/1", "DATA"), ("10.9/absent", "DATA")])
+
+    assert finished
+    by_doi = {r["DOI"]: r["exists"] for r in result}
+    assert by_doi["10.1/1"] is True  # exact line present -> found
+    assert by_doi["10.9/absent"] is False
+
+
+def test_doi_match_handles_line_with_trailing_metadata(tmp_path, monkeypatch):
+    # Lines of the form "<DOI>\t<metadata>" still match on the first token.
+    monkeypatch.setattr(search_bot, "DATABASE_PATH", str(tmp_path) + "/")
+    (tmp_path / "0_chunk.txt").write_text("10.5/abc\tsome title here\n", encoding="utf-8")
+
+    finished, result = _run_bounded([("10.5/abc", "DATA")])
+
+    assert finished
+    assert result[0]["exists"] is True
+
+
 def test_discover_chunk_files_sorted_numerically(tmp_path, monkeypatch):
     monkeypatch.setattr(search_bot, "DATABASE_PATH", str(tmp_path) + "/")
     for n in (0, 2, 10, 1):