Librarian: busy-aware ping + per-query lost-result watchdog

Two refinements to the librarian health/delivery story, matching how it
actually behaves under load:

1. Busy-aware ping (case b - broken return path). A ping arriving while
   the worker is grinding a search no longer queues behind it (which made
   a healthy-but-busy librarian time out and look dead). The librarian
   tracks worker_busy and, when set, pongs back IMMEDIATELY without
   touching the queue. Being busy is fine - you can keep piling searches
   on. The ping still travels the librarian->bot return path, so it keeps
   catching the one thing it must: a disrupted/incompatible return path
   where queries vanish. Idle pings still go through the internal queue.

2. Per-query watchdog (case a - finished but result lost). The librarian
   now tracks every search uuid's lifecycle (queued -> processing ->
   gone) in active_queries, exposed via a new POST /query_status. After
   dispatching a search the bot records it in self.pending; watch_pending
   polls /query_status for each. While the librarian still knows the uuid
   the search is progressing - left alone. The moment a uuid VANISHES
   there while still pending on the bot, its result was computed but never
   delivered: after a grace window (to rule out an in-flight result) the
   bot posts a notice to the channel - but ONLY then. A normally delivered
   result is popped from self.pending by check_data_q and never flagged.

Hardening: the worker's search body is now wrapped in try/except/finally
so a crashing search can't kill the worker thread (which would freeze the
queue), and worker_busy / active_queries are always cleared. The grace
logic lives in a dependency-free librarian_watchdog.pending_verdict so it
is unit-testable without discord/pdf libs. /ping and /query_status are
plain (sync) views so they run without flask[async].

Tests: unit test_librarian_watchdog (verdict transitions); integration
test_librarian_query_lifecycle (query_status known/unknown + auth,
idle-ping-queues, busy-ping-pongs-directly). Suite: 28 integration + 48
unit green.

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

conjurer_librarian/conjurer_librarian.py

@@ -69,6 +69,19 @@ app = Flask(__name__)
 librarian_queue = Queue()
 librarian_list = []
 
+# Lifecycle of every real search uuid: "queued" (accepted, sitting in
+# librarian_queue) -> "processing" (worker pulled it) -> removed (worker
+# finished AND attempted to send the result). The bot's per-query watchdog polls
+# /query_status against this: a uuid that VANISHES from here without its result
+# reaching the bot is a lost result (finished-but-never-delivered) and gets
+# flagged in chat.
+active_queries: Dict[str, str] = {}
+_active_lock = threading.Lock()
+# Set while the worker is grinding a real search. A ping arriving during this
+# pongs back immediately WITHOUT queueing - being busy is healthy (you can keep
+# piling searches on), so "busy" must never look like "dead" to the health check.
+worker_busy = threading.Event()
+
 
 def _service_headers() -> Dict[str, str]:
     if API_KEY:
@@ -81,6 +94,25 @@ def _authorize_request() -> None:
         abort(401)
 
 
+def _post_pong(app_logger, ping_uuid) -> None:
+    """POST a pong for ``ping_uuid`` back to the bot. Non-fatal on failure.
+
+    This is the SAME return path a real result takes (bot's /conjurer), so a
+    delivered pong proves the librarian->bot leg works - the one thing the ping
+    needs to establish. Used directly by the /ping route (busy ping, which skips
+    the queue) and, wrapped in a thread, by the worker (idle ping). Synchronous
+    so the /ping route can stay a plain (non-async) view."""
+    try:
+        requests.post(
+            f"{MAIN_BOT_ADDRESS}{SEND_RESULTS}",
+            json={"__pong__": ping_uuid},
+            headers=_service_headers(),
+            timeout=5,
+        )
+    except requests.exceptions.RequestException as exc:
+        app_logger.warning("PING pong send failed for %s: %s", ping_uuid, exc)
+
+
 # trunk-ignore(pylint/R0902)
 class Librarian(object):
     """
@@ -421,102 +453,108 @@ class BackgroundTaskSearch(threading.Thread):
                     "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
-                    )
+                await asyncio.to_thread(_post_pong, self.app.logger, ping_uuid)
                 continue
             librarian = item
-            self.app.logger.info("STARTED")
-            result = await librarian.answer_query(librarian.deep_search)
-            result = {librarian.uuid: result}
-            self.app.logger.info("Saving to file")
-
-            # Save results to "not_in_db.json" file
-            with open(lib_paths.NOT_IN_DB, "r+", encoding="utf-8") as ndb_file:
-                ndb_database = {}
-                try:
-                    ndb_database = json.load(ndb_file)
-                except JSONDecodeError:
-                    pass
-                if ndb_database:
-                    ndb_database.update(librarian.not_in_db)
-                else:
-                    ndb_database = librarian.not_in_db
-                ndb_file.truncate(0)
-                ndb_file.seek(0)
-                json.dump(ndb_database, ndb_file)
-
-            # Save results to "s_results.json" file
-            with open(lib_paths.S_RESULTS, "r+", encoding="utf-8") as s_file:
-                database = {}
-                try:
-                    database = json.load(s_file)
-                except JSONDecodeError:
-                    pass
-                if database:
-                    self.app.logger.info(database)
-                    self.app.logger.info(result)
-                    database.update(result)
-                else:
-                    database = result
-                self.app.logger.info("DUMPING DATA")
-                s_file.truncate(0)
-                s_file.seek(0)
-                json.dump(database, s_file)
-            self.app.logger.info("FINISHED")
-
-            # Send the result back to the bot. Log EXACTLY what goes out (target,
-            # uuid, how many DOIs and which) so the librarian log makes it plain a
-            # result was sent and what was in it.
-            payload = result  # shape: {uuid: {DOI: {"Title": ..., "type": ...}}}
-            hits = payload.get(librarian.uuid, {}) if isinstance(payload, dict) else {}
-            target = f"{MAIN_BOT_ADDRESS}{SEND_RESULTS}"
-            self.app.logger.info(
-                "SENDING result for %s to %s: %d DOI(s): %s",
-                librarian.uuid,
-                target,
-                len(hits),
-                list(hits.keys()),
-            )
-            # A failed send must NOT kill this worker - otherwise a bot that is
-            # momentarily down stalls every future query until the librarian is
-            # restarted. Log and carry on to the next queued search.
+            # Mark busy + processing for the whole search, and ALWAYS clear both
+            # (even on a crash) in finally: worker_busy so a ping doesn't wait
+            # behind us, and active_queries so the bot's watchdog can tell a
+            # finished-and-gone query from one still in flight.
+            worker_busy.set()
+            with _active_lock:
+                active_queries[str(librarian.uuid)] = "processing"
             try:
-                response = await asyncio.to_thread(
-                    requests.post,
-                    target,
-                    json=payload,
-                    headers=_service_headers(),
-                    timeout=360,
-                )
-                if response.status_code == 200:
-                    self.app.logger.info(
-                        "SENT result for %s -> HTTP 200 (bot accepted)", librarian.uuid
-                    )
-                else:
-                    self.app.logger.warning(
-                        "SENT result for %s but bot returned HTTP %s: %s",
-                        librarian.uuid,
-                        response.status_code,
-                        response.text[:500],
-                    )
-            except requests.exceptions.RequestException as exc:
-                self.app.logger.error(
-                    "FAILED to send result for %s to %s: %s",
+                self.app.logger.info("STARTED")
+                result = await librarian.answer_query(librarian.deep_search)
+                result = {librarian.uuid: result}
+                self.app.logger.info("Saving to file")
+
+                # Save results to "not_in_db.json" file
+                with open(lib_paths.NOT_IN_DB, "r+", encoding="utf-8") as ndb_file:
+                    ndb_database = {}
+                    try:
+                        ndb_database = json.load(ndb_file)
+                    except JSONDecodeError:
+                        pass
+                    if ndb_database:
+                        ndb_database.update(librarian.not_in_db)
+                    else:
+                        ndb_database = librarian.not_in_db
+                    ndb_file.truncate(0)
+                    ndb_file.seek(0)
+                    json.dump(ndb_database, ndb_file)
+
+                # Save results to "s_results.json" file
+                with open(lib_paths.S_RESULTS, "r+", encoding="utf-8") as s_file:
+                    database = {}
+                    try:
+                        database = json.load(s_file)
+                    except JSONDecodeError:
+                        pass
+                    if database:
+                        self.app.logger.info(database)
+                        self.app.logger.info(result)
+                        database.update(result)
+                    else:
+                        database = result
+                    self.app.logger.info("DUMPING DATA")
+                    s_file.truncate(0)
+                    s_file.seek(0)
+                    json.dump(database, s_file)
+                self.app.logger.info("FINISHED")
+
+                # Send the result back to the bot. Log EXACTLY what goes out
+                # (target, uuid, how many DOIs and which) so the librarian log
+                # makes it plain a result was sent and what was in it.
+                payload = result  # shape: {uuid: {DOI: {"Title": ..., "type": ...}}}
+                hits = payload.get(librarian.uuid, {}) if isinstance(payload, dict) else {}
+                target = f"{MAIN_BOT_ADDRESS}{SEND_RESULTS}"
+                self.app.logger.info(
+                    "SENDING result for %s to %s: %d DOI(s): %s",
                     librarian.uuid,
                     target,
-                    exc,
+                    len(hits),
+                    list(hits.keys()),
                 )
-            await asyncio.sleep(1)
+                # A failed send must NOT kill this worker - otherwise a bot that
+                # is momentarily down stalls every future query until the
+                # librarian is restarted. Log and carry on to the next search.
+                try:
+                    response = await asyncio.to_thread(
+                        requests.post,
+                        target,
+                        json=payload,
+                        headers=_service_headers(),
+                        timeout=360,
+                    )
+                    if response.status_code == 200:
+                        self.app.logger.info(
+                            "SENT result for %s -> HTTP 200 (bot accepted)", librarian.uuid
+                        )
+                    else:
+                        self.app.logger.warning(
+                            "SENT result for %s but bot returned HTTP %s: %s",
+                            librarian.uuid,
+                            response.status_code,
+                            response.text[:500],
+                        )
+                except requests.exceptions.RequestException as exc:
+                    self.app.logger.error(
+                        "FAILED to send result for %s to %s: %s",
+                        librarian.uuid,
+                        target,
+                        exc,
+                    )
+            except Exception as exc:  # pylint: disable=broad-exception-caught
+                # A crashing search must not kill the worker thread (which would
+                # freeze the whole queue). Log and move on; finally still clears
+                # busy/active so the query is correctly seen as "gone".
+                self.app.logger.exception("Search %s crashed: %s", librarian.uuid, exc)
+            finally:
+                worker_busy.clear()
+                with _active_lock:
+                    active_queries.pop(str(librarian.uuid), None)
+                await asyncio.sleep(1)
 
 
 # ==================================SERVER ROUTES==========================================
@@ -545,6 +583,10 @@ async def query_database():
     cl = Librarian(app, record["query"], uuid, deep_search)
     librarian_queue.put(cl)
     librarian_list.append(cl)
+    # The bot's per-query watchdog polls /query_status for this uuid; mark it
+    # "queued" now so it counts as known the moment we accept it.
+    with _active_lock:
+        active_queries[str(uuid)] = "queued"
     answer_data = (record["query"], record["UUID"], librarian_queue.qsize())
     return_data = (
         jsonify(isError=False, message="Success", statusCode=200, data=answer_data),
@@ -554,28 +596,65 @@ async def query_database():
 
 
 @app.route("/ping", methods=["POST"])
-async def ping_roundtrip():
+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.
+    Two cases, one guarantee - the pong always comes back over the librarian->bot
+    return path (the only thing the ping must prove):
+
+    * IDLE: put a ping marker onto the SAME internal ``librarian_queue`` real
+      searches use and return 200. The worker pulls it off and pongs it back,
+      so a successful pong proves the whole pipeline flows (queue + worker + the
+      return leg), not just that Flask is up.
+    * BUSY (a search is grinding): DO NOT queue - the ping would just wait behind
+      a possibly hours-long search and time out, making a perfectly healthy busy
+      librarian look dead. Pong back immediately instead. Being busy is fine; you
+      can keep piling searches on. The ping only needs to catch a BROKEN return
+      path, and the direct pong exercises exactly that.
     """
     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})
+    if worker_busy.is_set():
+        app.logger.info("PING %s while busy grinding - direct pong (skip queue)", ping_uuid)
+        _post_pong(app.logger, ping_uuid)
+    else:
+        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("/query_status", methods=["POST"])
+def query_status():
+    _authorize_request()
+    """
+    Per-query watchdog probe.
+
+    Returns whether ``UUID`` is still known to the librarian (queued or being
+    processed). The bot polls this after dispatching a search: while the uuid is
+    known the search is progressing; once it VANISHES here without the result
+    ever reaching the bot, the result was lost in transit and the bot tells the
+    user. A busy/queued search is never mistaken for a lost one.
+    """
+    record = json.loads(request.data)
+    uuid = str(record["UUID"])
+    with _active_lock:
+        state = active_queries.get(uuid, "unknown")
+    return (
+        jsonify(
+            isError=False,
+            message="Success",
+            statusCode=200,
+            data={"uuid": uuid, "known": state != "unknown", "state": state},
+        ),
+        200,
+    )
+
+
 @app.route("/get_partial_result", methods=["POST"])
 async def get_partial():
     _authorize_request()