Librarian: answer each query back to the bot that sent it

So one librarian can serve several bots (test + deploy) instead of firing
every result/pong at a single static CONJURER_MAIN_BOT.

* The bot includes its own callback address (CONJURER_SELF_CALLBACK) in
  every /query and /ping.
* The librarian stores that callback with the query (persisted with the
  request, so a replay after restart still answers the right bot) and, for
  results, in the OUTBOX entry ({target, payload}) so the resender delivers
  to the origin bot even across a librarian restart.
* Pongs go back to the pinging bot too - otherwise a second bot's health
  check would be ponged to the first and always time out, so it could
  never enable its librarian cog.
* Empty callback falls back to MAIN_BOT_ADDRESS, and a legacy OUTBOX entry
  (raw payload, pre-callback) is still delivered to the default bot, so the
  upgrade is seamless.

Tests: per-origin result delivery + legacy-shape fallback (outbox),
busy/idle pong routed to the callback bot vs default (lifecycle). Suite:
58 unit + 52 integration green.

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

conjurer_librarian/conjurer_librarian.py

@@ -154,49 +154,55 @@ 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.
+def _post_pong(app_logger, ping_uuid, callback="") -> None:
+    """POST a pong for ``ping_uuid`` back to the pinging bot. Non-fatal.
 
-    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."""
+    ``callback`` is the address of the bot that sent the ping, so a librarian
+    shared by several bots pongs each at its OWN address (empty => the static
+    MAIN_BOT_ADDRESS). Same return path a real result takes (bot's /conjurer),
+    so a delivered pong proves the librarian->that-bot leg works."""
+    target = f"{callback or MAIN_BOT_ADDRESS}{SEND_RESULTS}"
     try:
         requests.post(
-            f"{MAIN_BOT_ADDRESS}{SEND_RESULTS}",
-            json={"__pong__": ping_uuid},
-            headers=_service_headers(),
-            timeout=5,
+            target, 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)
 
 
-def _deliver_result(uuid, payload, app_logger, attempts=RESULT_SEND_ATTEMPTS) -> bool:
-    """POST one result to the bot, retrying with backoff. True only on HTTP 200.
+def _outbox_target_payload(entry):
+    """Unpack an OUTBOX entry into (target, payload).
 
-    The bot's /conjurer is idempotent (dedups by uuid), so re-POSTing a result
-    it already has is safe - it just answers 200 again. That is what lets the
-    OUTBOX keep retrying until the result is truly acknowledged, without ever
-    double-delivering to the user.
-    """
-    target = f"{MAIN_BOT_ADDRESS}{SEND_RESULTS}"
+    New shape: {"target": <bot address>, "payload": {uuid: result}}. Old shape
+    (from before per-origin callbacks) is the raw payload - delivered to the
+    default bot - so an upgrade doesn't strand results already on disk."""
+    if isinstance(entry, dict) and "target" in entry and "payload" in entry:
+        return entry["target"], entry["payload"]
+    return "", entry
+
+
+def _deliver_result(target, uuid, payload, app_logger, attempts=RESULT_SEND_ATTEMPTS) -> bool:
+    """POST one result to ``target``'s /conjurer, retrying with backoff. True only
+    on HTTP 200. ``target`` is the origin bot's address so a shared librarian
+    answers each bot where its query came from (empty => MAIN_BOT_ADDRESS). The
+    bot's /conjurer is idempotent, so re-POSTing what it already has is safe."""
+    url = f"{target or MAIN_BOT_ADDRESS}{SEND_RESULTS}"
     for attempt in range(1, max(1, attempts) + 1):
         try:
             response = requests.post(
-                target, json=payload, headers=_service_headers(), timeout=60
+                url, json=payload, headers=_service_headers(), timeout=60
             )
             if response.status_code == 200:
-                app_logger.info("Result %s delivered (HTTP 200) on attempt %d", uuid, attempt)
+                app_logger.info("Result %s delivered to %s (HTTP 200) attempt %d", uuid, url, attempt)
                 return True
             app_logger.warning(
-                "Result %s: bot returned HTTP %s (attempt %d/%d): %s",
-                uuid, response.status_code, attempt, attempts, response.text[:300],
+                "Result %s: %s returned HTTP %s (attempt %d/%d): %s",
+                uuid, url, response.status_code, attempt, attempts, response.text[:300],
             )
         except requests.exceptions.RequestException as exc:
             app_logger.warning(
-                "Result %s delivery failed (attempt %d/%d): %s", uuid, attempt, attempts, exc
+                "Result %s delivery to %s failed (attempt %d/%d): %s",
+                uuid, url, attempt, attempts, exc,
             )
         if attempt < attempts:
             time.sleep(RESULT_SEND_BACKOFF * attempt)
@@ -204,13 +210,12 @@ def _deliver_result(uuid, payload, app_logger, attempts=RESULT_SEND_ATTEMPTS) ->
 
 
 def _resend_once(app_logger) -> None:
-    """One sweep of the OUTBOX: try to deliver every un-acked result, once each.
-
-    Removes each entry only after a positive ACK, so nothing is dropped until
-    the bot has it. Corrupt/unreadable entries are skipped by DiskQueue.items().
-    """
-    for uuid, payload, _ts in _outbox.items():
-        if _deliver_result(uuid, payload, app_logger, attempts=1):
+    """One sweep of the OUTBOX: try to deliver every un-acked result, once each,
+    to the ORIGIN bot recorded with it. Removes each entry only after a positive
+    ACK. Corrupt/unreadable entries are skipped by DiskQueue.items()."""
+    for uuid, entry, _ts in _outbox.items():
+        target, payload = _outbox_target_payload(entry)
+        if _deliver_result(target, uuid, payload, app_logger, attempts=1):
             _outbox.remove(uuid)
 
 
@@ -243,7 +248,10 @@ def replay_requests(app_logger) -> None:
     app_logger.info("Replaying %d unfinished search request(s) after restart", len(pending))
     for uuid, payload, _ts in pending:
         try:
-            cl = Librarian(app, payload["query"], uuid, payload.get("deep_search", False))
+            cl = Librarian(
+                app, payload["query"], uuid,
+                payload.get("deep_search", False), payload.get("callback", ""),
+            )
         except Exception as exc:  # pylint: disable=broad-exception-caught
             app_logger.warning("Cannot replay request %s (dropping): %s", uuid, exc)
             _forget_search(uuid)
@@ -269,7 +277,7 @@ class Librarian(object):
     Represents a librarian object that performs search and refinement operations on queries.
     """
 
-    def __init__(self, _app, query, uuid, _deep_search) -> None:
+    def __init__(self, _app, query, uuid, _deep_search, callback="") -> None:
         """
         Initializes a Librarian object.
 
@@ -340,6 +348,9 @@ class Librarian(object):
         self.search_result_from_cr = {}
         self.done = False
         self.deep_search = _deep_search
+        # Where to send this search's result back to (the bot that asked). Lets
+        # one librarian serve several bots; empty => static MAIN_BOT_ADDRESS.
+        self.callback = callback or ""
         # Set True when a graceful shutdown interrupts this search mid-scan; the
         # worker then leaves the request + checkpoint in place instead of
         # delivering, so a restart resumes it.
@@ -630,7 +641,9 @@ class BackgroundTaskSearch(threading.Thread):
                     "PING %s pulled off internal queue - ponging back (no search)",
                     ping_uuid,
                 )
-                await asyncio.to_thread(_post_pong, self.app.logger, ping_uuid)
+                await asyncio.to_thread(
+                    _post_pong, self.app.logger, ping_uuid, item.get("callback", "")
+                )
                 continue
             librarian = item
             # Mark busy + processing for the whole search, and ALWAYS clear both
@@ -680,13 +693,16 @@ class BackgroundTaskSearch(threading.Thread):
                 # retrying until the bot ACKs, and only then is it removed.
                 payload = result  # shape: {uuid: {DOI: {"Title": ..., "type": ...}}}
                 uuid = str(librarian.uuid)
+                target = librarian.callback  # answer the bot that asked
                 hits = payload.get(librarian.uuid, {}) if isinstance(payload, dict) else {}
-                _outbox.put(uuid, payload)
+                # OUTBOX entry carries the origin bot's address so the resender
+                # delivers it to the right bot even after a librarian restart.
+                _outbox.put(uuid, {"target": target, "payload": payload})
                 self.app.logger.info(
-                    "SENDING result for %s: %d DOI(s): %s (queued to OUTBOX)",
-                    uuid, len(hits), list(hits.keys()),
+                    "SENDING result for %s to %s: %d DOI(s): %s (queued to OUTBOX)",
+                    uuid, target or "default", len(hits), list(hits.keys()),
                 )
-                if await asyncio.to_thread(_deliver_result, uuid, payload, self.app.logger):
+                if await asyncio.to_thread(_deliver_result, target, uuid, payload, self.app.logger):
                     _outbox.remove(uuid)
                 else:
                     self.app.logger.warning(
@@ -731,11 +747,17 @@ async def query_database():
     record = json.loads(request.data)
     uuid = record["UUID"]
     deep_search = record["deep_search"]
-    app.logger.info("Query accepted %s: %s", uuid, record["query"])
+    # Where to answer THIS query - the bot that sent it. Persisted with the
+    # request so a replay after restart still answers the right bot.
+    callback = record.get("callback", "")
+    app.logger.info("Query accepted %s (callback %s): %s", uuid, callback or "default", record["query"])
     # Persist the request BEFORE enqueuing, so an accepted search survives a
     # restart (it is replayed on startup) - not just an in-progress one.
-    _requests.put(str(uuid), {"query": record["query"], "deep_search": deep_search})
-    cl = Librarian(app, record["query"], uuid, deep_search)
+    _requests.put(
+        str(uuid),
+        {"query": record["query"], "deep_search": deep_search, "callback": callback},
+    )
+    cl = Librarian(app, record["query"], uuid, deep_search, callback)
     librarian_queue.put(cl)
     librarian_list.append(cl)
     # The bot's per-query watchdog polls /query_status for this uuid; mark it
@@ -771,12 +793,15 @@ def ping_roundtrip():
     """
     record = json.loads(request.data)
     ping_uuid = record["UUID"]
+    # Pong goes back to the bot that pinged (carried through the queue when idle),
+    # so a shared librarian health-checks correctly for every bot.
+    callback = record.get("callback", "")
     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)
+        _post_pong(app.logger, ping_uuid, callback)
     else:
         app.logger.info("PING received %s - queued for round-trip", ping_uuid)
-        librarian_queue.put({"__ping__": ping_uuid})
+        librarian_queue.put({"__ping__": ping_uuid, "callback": callback})
     return (
         jsonify(isError=False, message="ping-queued", statusCode=200, data=ping_uuid),
         200,