Durable result delivery: OUTBOX + idempotent INBOX so results never die

An 8h search result must survive a transient bot outage, an api/address
misroute, or a restart of either side. Make the librarian->bot result
path durably at-least-once with idempotent rendering:

Shared: durable_queue.DiskQueue - a dependency-free, atomically-written,
one-file-per-key disk queue (unit-tested), shared by both images
(added to Dockerfile.librarian; the bot already COPYs *.py).

Librarian (sender): finished results go to a persistent OUTBOX before
sending; delivery retries with backoff; an entry is removed only on a
positive ACK; a resender thread keeps flushing the OUTBOX, so a result
survives a bot outage AND a librarian restart (OUTBOX is on the state
volume) - it simply keeps trying until acked.

Bot (receiver): /conjurer is now idempotent and durable - each result is
persisted to an INBOX before acking and only queued if its uuid was not
already delivered (dropped as a duplicate) or already pending. Once the
cog actually renders it, mark_delivered() records the uuid and clears the
inbox, so the librarian's resends become no-ops. On startup the bot
replays any accepted-but-unrendered result from the INBOX, so a bot crash
mid-flight doesn't lose it. Pongs stay ephemeral.

Together: the librarian keeps a result until the bot confirms it; the bot
keeps it until it is on screen; duplicates never double-render. Combined
with the deploy return-path fix, an expensive result no longer vanishes.

Tests: unit test_durable_queue; integration test_librarian_outbox
(retry/backoff, resend survives outage) and test_result_durable_delivery
(persist, dedup pending, dedup delivered, replay, pong not persisted).
Suite: 55 unit + 39 integration green.

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

communication_subroutine.py

@@ -13,6 +13,9 @@ import requests
 from flask import Flask, abort, jsonify, request
 from waitress import serve
 
+from constants import DELIVERED_DIR, DELIVERED_MAX, RESULT_INBOX_DIR
+from durable_queue import DiskQueue
+
 HOST_ADDRESS = os.getenv("CONJURER_DISCORD_HOST", "192.168.1.92")
 PORT_ADDRESS = int(os.getenv("CONJURER_DISCORD_PORT", "5000"))
 ICECAST_ADDRESS = os.getenv("CONJURER_ICECAST", "http://192.168.1.12:8000")
@@ -33,9 +36,43 @@ incoming_q = Queue()
 # 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
+
+# Durable result delivery. Every incoming search result is persisted to _inbox
+# before we ack the librarian, and only removed once it has actually been
+# rendered to the user (its uuid recorded in _delivered). This makes /conjurer
+# idempotent (the librarian can safely resend until acked) and lets an accepted
+# result survive a bot restart mid-flight (replayed from _inbox on startup).
+_inbox = DiskQueue(RESULT_INBOX_DIR)
+_delivered = DiskQueue(DELIVERED_DIR)
 app = Flask(__name__)
 
 
+def mark_delivered(query_uuid) -> None:
+    """Record that a result was rendered to the user.
+
+    After this, the librarian's resends of that uuid are dropped as duplicates
+    and it is never replayed from the inbox again. Called by the cog once it has
+    actually posted the result to Discord."""
+    _delivered.put(query_uuid, {})
+    _delivered.prune(DELIVERED_MAX)
+    _inbox.remove(query_uuid)
+
+
+def replay_inbox() -> None:
+    """Re-queue INBOX results that were accepted but not yet rendered.
+
+    Recovers an expensive result that reached the bot (and was acked to the
+    librarian, so it won't be resent) but whose render was lost to a bot restart.
+    """
+    logger = logging.getLogger("discord")
+    for query_uuid, payload, _ts in _inbox.items():
+        if _delivered.contains(query_uuid):
+            _inbox.remove(query_uuid)
+            continue
+        logger.info("Replaying un-rendered result %s from INBOX", query_uuid)
+        incoming_q.put(payload)
+
+
 def _authorize_request() -> None:
     """Reject inbound calls lacking the shared key (no-op if key is unset)."""
     if API_KEY and request.headers.get("X-Conjurer-Api-Key") != API_KEY:
@@ -113,11 +150,30 @@ def answer_external_command():
     """
     _authorize_request()
     logger = logging.getLogger("discord")
-    logger.info(request)
     record = json.loads(request.data)
-    logger.info(record)
-    logger.info("DATA RECEIVED")
-    incoming_q.put(record)
+    logger.info("DATA RECEIVED: %s", record)
+    # Health-check pongs are ephemeral - never persisted or deduped.
+    if isinstance(record, dict) and "__pong__" in record:
+        incoming_q.put(record)
+        return jsonify("SUCCESS")
+    # Search results: idempotent, durable intake. Persist each uuid to the inbox
+    # before acking, and queue only a uuid we have NOT already delivered or
+    # accepted. This lets the librarian's resender retry safely (a duplicate is
+    # dropped, never double-rendered) and lets an accepted-but-unrendered result
+    # be replayed after a bot restart.
+    if isinstance(record, dict):
+        for query_uuid in list(record.keys()):
+            if _delivered.contains(query_uuid):
+                logger.info("Result %s already delivered - dropping duplicate", query_uuid)
+                continue
+            if _inbox.contains(query_uuid):
+                logger.info("Result %s already pending - dropping duplicate", query_uuid)
+                continue
+            single = {query_uuid: record[query_uuid]}
+            _inbox.put(query_uuid, single)
+            incoming_q.put(single)
+    else:
+        incoming_q.put(record)  # unexpected shape - preserve old behaviour
     return jsonify("SUCCESS")
 
 
@@ -405,6 +461,13 @@ def comm_subroutine(stop_event: Optional[threading.Event] = None):
     for worker in threads:
         worker.start()
 
+    # Recover any result that was accepted before a previous shutdown but never
+    # rendered - re-queue it now that scan_incoming is running.
+    try:
+        replay_inbox()
+    except Exception:  # pylint: disable=broad-exception-caught
+        logger.exception("INBOX replay on startup failed")
+
     try:
         while any(thread.is_alive() for thread in threads):
             if stop_event and stop_event.is_set():

conjurer_librarian/conjurer_librarian.py

@@ -18,6 +18,7 @@ import json
 import logging
 import os
 import threading
+import time
 from json.decoder import JSONDecodeError
 from logging import handlers
 from pathlib import Path
@@ -29,6 +30,7 @@ import lib_paths
 import scrape_bot
 import search_bot
 # import search_bot2 as search_bot
+from durable_queue import DiskQueue
 from flask import Flask, jsonify, request, abort
 from habanero import Crossref
 from waitress import serve
@@ -64,6 +66,16 @@ LOGFILE_PATH = _env_path(
     "CONJURER_LIBRARIAN_LOG", str(BASE_DIR / "librarian.log")
 )
 
+# Durable OUTBOX for finished results. A result is expensive (hours of compute),
+# so it is written here and only removed once the bot ACKs it (HTTP 200). Lives
+# on the librarian's persistent state volume, so it survives a librarian restart
+# and a transient bot outage; the resender thread keeps retrying until delivered.
+OUTBOX_DIR = _env("CONJURER_LIBRARIAN_OUTBOX", os.path.join(lib_paths.STATE_DIR, "outbox"))
+RESULT_SEND_ATTEMPTS = int(_env("CONJURER_RESULT_SEND_ATTEMPTS", "3"))
+RESULT_SEND_BACKOFF = float(_env("CONJURER_RESULT_SEND_BACKOFF", "2"))
+OUTBOX_RESEND_SECONDS = int(_env("CONJURER_OUTBOX_RESEND_SECONDS", "60"))
+_outbox = DiskQueue(OUTBOX_DIR)
+
 app = Flask(__name__)
 
 librarian_queue = Queue()
@@ -113,6 +125,63 @@ def _post_pong(app_logger, ping_uuid) -> None:
         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.
+
+    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}"
+    for attempt in range(1, max(1, attempts) + 1):
+        try:
+            response = requests.post(
+                target, 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)
+                return True
+            app_logger.warning(
+                "Result %s: bot returned HTTP %s (attempt %d/%d): %s",
+                uuid, 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
+            )
+        if attempt < attempts:
+            time.sleep(RESULT_SEND_BACKOFF * attempt)
+    return False
+
+
+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):
+            _outbox.remove(uuid)
+
+
+def outbox_resender(app_logger) -> None:
+    """Background loop: periodically flush the OUTBOX until the bot is reachable.
+
+    This is what makes an expensive result survive a transient bot outage or a
+    librarian restart - on restart the persisted OUTBOX is simply resent."""
+    pending = len(_outbox)
+    if pending:
+        app_logger.info("OUTBOX has %d un-acked result(s) on startup - will resend", pending)
+    while True:
+        try:
+            _resend_once(app_logger)
+        except Exception as exc:  # pylint: disable=broad-exception-caught
+            app_logger.exception("OUTBOX resend sweep failed: %s", exc)
+        time.sleep(OUTBOX_RESEND_SECONDS)
+
+
 # trunk-ignore(pylint/R0902)
 class Librarian(object):
     """
@@ -503,47 +572,24 @@ class BackgroundTaskSearch(threading.Thread):
                     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.
+                # Persist the result to the durable OUTBOX FIRST, then try to
+                # deliver it. Writing to disk before sending is the whole point:
+                # an expensive (hours-long) result now survives a failed send, a
+                # bot outage, or a librarian restart - the resender keeps
+                # retrying until the bot ACKs, and only then is it removed.
                 payload = result  # shape: {uuid: {DOI: {"Title": ..., "type": ...}}}
+                uuid = str(librarian.uuid)
                 hits = payload.get(librarian.uuid, {}) if isinstance(payload, dict) else {}
-                target = f"{MAIN_BOT_ADDRESS}{SEND_RESULTS}"
+                _outbox.put(uuid, payload)
                 self.app.logger.info(
-                    "SENDING result for %s to %s: %d DOI(s): %s",
-                    librarian.uuid,
-                    target,
-                    len(hits),
-                    list(hits.keys()),
+                    "SENDING result for %s: %d DOI(s): %s (queued to OUTBOX)",
+                    uuid, 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 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,
+                if await asyncio.to_thread(_deliver_result, uuid, payload, self.app.logger):
+                    _outbox.remove(uuid)
+                else:
+                    self.app.logger.warning(
+                        "Result %s not acked yet - left in OUTBOX for the resender", uuid
                     )
             except Exception as exc:  # pylint: disable=broad-exception-caught
                 # A crashing search must not kill the worker thread (which would
@@ -703,6 +749,11 @@ if __name__ == "__main__":
             target=scrape_bot.scraper, args=(app.logger,), daemon=True
         )
     )
+    # Durable delivery: keep flushing the OUTBOX so any result not yet acked by
+    # the bot (transient outage, or left over from before a restart) is resent.
+    threads.append(
+        threading.Thread(target=outbox_resender, args=(app.logger,), daemon=True)
+    )
     i = 0
     try:
         for worker in threads:

constants.py

@@ -230,6 +230,21 @@ MEMORY_COMPACT_THRESHOLD = int(os.getenv("CONJURER_MEMORY_COMPACT_THRESHOLD", "4
 MEMORY_KEEP_RECENT = int(os.getenv("CONJURER_MEMORY_KEEP_RECENT", "200"))
 MEMORY_COMPACT_HOURS = float(os.getenv("CONJURER_MEMORY_COMPACT_HOURS", "6"))
 
+# Durable result-delivery spool (librarian -> bot). The bot persists every
+# incoming search result to RESULT_INBOX_DIR before acking and only forgets it
+# once rendered (uuid recorded in DELIVERED_DIR), so an expensive (hours-long)
+# result survives a bot restart mid-flight and duplicate resends are idempotent.
+# Rooted under CONJURER_DATA_DIR when set (a mounted volume), else next to the
+# log file. DELIVERED_MAX bounds the remembered-uuid set.
+_STATE_ROOT = _DATA_DIR or (os.path.dirname(LOGFILE) or ".")
+RESULT_INBOX_DIR = os.getenv(
+    "CONJURER_RESULT_INBOX", os.path.join(_STATE_ROOT, "result_inbox")
+)
+DELIVERED_DIR = os.getenv(
+    "CONJURER_DELIVERED_DIR", os.path.join(_STATE_ROOT, "delivered_uuids")
+)
+DELIVERED_MAX = int(os.getenv("CONJURER_DELIVERED_MAX", "10000"))
+
 FILE_SERVICE_ADDRESS = os.getenv("CONJURER_FILE_SERVICE", "http://192.168.1.15:5000")
 RADIO_HARBOR_ADDRESS = os.getenv("CONJURER_RADIO_HARBOR", "http://192.168.1.15:54321")
 # Betoniarka (radio-operator service colocated with Liquidsoap). Falls back to

docker/Dockerfile.librarian

@@ -14,6 +14,9 @@ RUN pip install --no-cache-dir --upgrade pip \
     && pip install --no-cache-dir -r requirements_librarian.txt requests
 
 COPY conjurer_librarian/ ./
+# durable_queue lives at the repo root and is shared with the bot; the librarian
+# imports it for the durable result OUTBOX.
+COPY durable_queue.py ./
 
 ENV PYTHONUNBUFFERED=1 \
     CONJURER_LIBRARIAN_HOST=0.0.0.0 \

durable_queue.py

@@ -0,0 +1,104 @@
+"""Dependency-free, disk-backed queue for durable message delivery.
+
+One JSON file per key under a directory. Used on both sides of the
+librarian <-> bot result path so an expensive (hours-long) search result is
+never lost to a transient network failure or a restart:
+
+* the librarian keeps a result in its OUTBOX until the bot acks it,
+* the bot keeps a result in its INBOX until it is actually rendered, and
+  remembers delivered uuids so duplicate resends are idempotent.
+
+Only ``json`` + ``os`` are imported, so the logic is unit-testable without
+flask, discord, or the network. Writes are atomic (temp file + ``os.replace``)
+so a crash mid-write can never leave a half-written record that poisons replay.
+"""
+import json
+import os
+import tempfile
+import time
+
+
+def _safe_name(key: str) -> str:
+    """Filesystem-safe file stem for a key (uuids are safe; be defensive)."""
+    stem = "".join(c for c in str(key) if c.isalnum() or c in "-_.")
+    return stem or "_"
+
+
+class DiskQueue:
+    """A directory of ``<key>.json`` records, each ``{key, payload, ts}``."""
+
+    def __init__(self, directory: str):
+        # No disk touch here on purpose: constructing a DiskQueue at import time
+        # must not create directories (tests, read-only default paths). The
+        # directory is created lazily on the first put().
+        self.directory = directory
+
+    def _path(self, key) -> str:
+        return os.path.join(self.directory, _safe_name(key) + ".json")
+
+    def put(self, key, payload) -> None:
+        """Atomically write (overwrite) the record for ``key``."""
+        os.makedirs(self.directory, exist_ok=True)
+        record = {"key": str(key), "payload": payload, "ts": time.time()}
+        fd, tmp = tempfile.mkstemp(dir=self.directory, suffix=".tmp")
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8") as handle:
+                json.dump(record, handle)
+            os.replace(tmp, self._path(key))  # atomic on POSIX
+        finally:
+            if os.path.exists(tmp):
+                os.remove(tmp)
+
+    def contains(self, key) -> bool:
+        return os.path.exists(self._path(key))
+
+    def remove(self, key) -> None:
+        try:
+            os.remove(self._path(key))
+        except FileNotFoundError:
+            pass
+
+    def items(self):
+        """Return ``[(key, payload, ts), ...]`` oldest-first.
+
+        Unreadable / half-written / corrupt files are skipped (never raise),
+        so one bad file can't stall replay of the rest.
+        """
+        out = []
+        try:
+            names = os.listdir(self.directory)
+        except FileNotFoundError:
+            return out
+        for name in names:
+            if not name.endswith(".json"):
+                continue
+            try:
+                with open(os.path.join(self.directory, name), encoding="utf-8") as handle:
+                    record = json.load(handle)
+                out.append((record["key"], record["payload"], record.get("ts", 0)))
+            except (OSError, ValueError, KeyError, TypeError):
+                continue
+        out.sort(key=lambda triple: triple[2])
+        return out
+
+    def keys(self):
+        return [key for key, _payload, _ts in self.items()]
+
+    def __len__(self) -> int:
+        return len(self.items())
+
+    def prune(self, max_entries: int) -> int:
+        """Keep only the newest ``max_entries`` (by ts); drop the rest.
+
+        Used for the delivered-uuid set so it cannot grow without bound.
+        Returns how many were dropped.
+        """
+        if max_entries < 0:
+            return 0
+        entries = self.items()  # oldest first
+        excess = len(entries) - max_entries
+        dropped = 0
+        for key, _payload, _ts in entries[: max(0, excess)]:
+            self.remove(key)
+            dropped += 1
+        return dropped

librarian_commands.py

@@ -15,7 +15,13 @@ import requests
 from discord.ext import commands, tasks
 
 from ai_functions import handle_response
-from communication_subroutine import IN_COMM_Q, OUT_COMM_Q, QueryControl, submit_ai_query
+from communication_subroutine import (
+    IN_COMM_Q,
+    OUT_COMM_Q,
+    QueryControl,
+    mark_delivered,
+    submit_ai_query,
+)
 from constants import (
     DIR_PATH_SADOX,
     LIBRARIAN_SERVICE_ADDRESS,
@@ -169,6 +175,12 @@ class DataModule(commands.Cog):
                             await ctx.send(message)
                             message = ""
 
+                # The result is now on screen: mark it delivered so the
+                # librarian's resends become no-ops and it is dropped from the
+                # durable inbox (never replayed again). Done after the core
+                # render but before the optional AI review, which is a bonus.
+                mark_delivered(str(fresh_data.uuid))
+
                 # Optional AI pass: re-rank the (already Crossref-relevance-
                 # sorted) DOI list and review the sources. Enqueued to the AI
                 # worker so it runs on whatever backend $gadaj_teraz selected;

tests/integration/test_librarian_outbox.py

@@ -0,0 +1,99 @@
+"""Integration: the librarian's durable result OUTBOX + retrying delivery.
+
+An 8-hour search result must not be lost to a transient bot outage. The result
+is written to the OUTBOX before sending; delivery retries with backoff; the
+entry is removed only on a positive ACK; and the resender keeps flushing the
+OUTBOX (across restarts, since it is on the persistent state volume).
+"""
+import logging
+import sys
+import types
+
+import pytest
+
+# conjurer_librarian imports `from habanero import Crossref` at import time; the
+# integration job doesn't install habanero. Stub it (we never build a real
+# Librarian here).
+if "habanero" not in sys.modules:
+    _habanero = types.ModuleType("habanero")
+    _habanero.Crossref = object
+    sys.modules["habanero"] = _habanero
+
+import conjurer_librarian as lib  # noqa: E402
+from durable_queue import DiskQueue  # noqa: E402
+
+_LOG = logging.getLogger("test-outbox")
+_LOG.addHandler(logging.NullHandler())
+
+
+class _Resp:
+    def __init__(self, status_code, text=""):
+        self.status_code = status_code
+        self.text = text
+
+
+@pytest.fixture(autouse=True)
+def _no_backoff(monkeypatch):
+    # Never actually sleep during retry backoff in tests.
+    monkeypatch.setattr(lib.time, "sleep", lambda _s: None)
+
+
+@pytest.fixture
+def outbox(tmp_path, monkeypatch):
+    box = DiskQueue(str(tmp_path / "outbox"))
+    monkeypatch.setattr(lib, "_outbox", box)
+    return box
+
+
+def test_deliver_succeeds_first_try(monkeypatch):
+    calls = []
+    monkeypatch.setattr(lib.requests, "post", lambda *a, **k: calls.append(1) or _Resp(200))
+    assert lib._deliver_result("u1", {"u1": {}}, _LOG, attempts=3) is True
+    assert len(calls) == 1  # no needless retries after a 200
+
+
+def test_deliver_retries_then_succeeds(monkeypatch):
+    responses = iter([_Resp(503), _Resp(500), _Resp(200)])
+    monkeypatch.setattr(lib.requests, "post", lambda *a, **k: next(responses))
+    assert lib._deliver_result("u2", {"u2": {}}, _LOG, attempts=3) is True
+
+
+def test_deliver_returns_false_when_all_attempts_fail(monkeypatch):
+    def boom(*_a, **_k):
+        raise lib.requests.exceptions.RequestException("bot down")
+
+    monkeypatch.setattr(lib.requests, "post", boom)
+    assert lib._deliver_result("u3", {"u3": {}}, _LOG, attempts=2) is False
+
+
+def test_resend_once_removes_only_acked_entries(outbox, monkeypatch):
+    outbox.put("ok", {"ok": {}})
+    outbox.put("bad", {"bad": {}})
+
+    def fake_deliver(query_uuid, _payload, _logger, attempts=1):
+        return query_uuid == "ok"
+
+    monkeypatch.setattr(lib, "_deliver_result", fake_deliver)
+    lib._resend_once(_LOG)
+
+    assert not outbox.contains("ok")   # acked -> dropped
+    assert outbox.contains("bad")      # not acked -> kept for the next sweep
+
+
+def test_resend_keeps_result_until_bot_recovers(outbox, monkeypatch):
+    # Simulate: bot down for the first sweep, up for the second. The result must
+    # survive the outage and be delivered on recovery.
+    outbox.put("u9", {"u9": {"10.1/x": {"Title": ["P"], "type": "article"}}})
+    state = {"up": False}
+
+    def flaky_post(*_a, **_k):
+        return _Resp(200) if state["up"] else _Resp(502)
+
+    monkeypatch.setattr(lib.requests, "post", flaky_post)
+
+    lib._resend_once(_LOG)               # bot down
+    assert outbox.contains("u9")         # preserved, not lost
+
+    state["up"] = True
+    lib._resend_once(_LOG)               # bot recovered
+    assert not outbox.contains("u9")     # now delivered and cleared

tests/integration/test_result_durable_delivery.py

@@ -0,0 +1,92 @@
+"""Integration: the bot's DURABLE, idempotent result intake (/conjurer).
+
+The expensive-result guarantees on the bot side:
+* every result is persisted to the inbox before it is acked,
+* a resend of a not-yet-delivered result is dropped (no double render),
+* once rendered (mark_delivered) further resends are dropped and it leaves the
+  inbox,
+* on startup, an accepted-but-unrendered result is replayed from the inbox,
+* health-check pongs are never persisted.
+"""
+import pytest
+
+import communication_subroutine as cs
+from durable_queue import DiskQueue
+
+
+@pytest.fixture
+def spool(tmp_path, monkeypatch):
+    inbox = DiskQueue(str(tmp_path / "inbox"))
+    delivered = DiskQueue(str(tmp_path / "delivered"))
+    monkeypatch.setattr(cs, "_inbox", inbox)
+    monkeypatch.setattr(cs, "_delivered", delivered)
+    cs.API_KEY = None
+    while not cs.incoming_q.empty():
+        cs.incoming_q.get()
+    return inbox, delivered
+
+
+def _drain_incoming():
+    out = []
+    while not cs.incoming_q.empty():
+        out.append(cs.incoming_q.get())
+    return out
+
+
+def _payload(uuid):
+    return {uuid: {"10.1000/xyz": {"Title": ["A Real Paper"], "type": "journal-article"}}}
+
+
+def test_result_is_persisted_then_queued(spool):
+    inbox, _delivered = spool
+    resp = cs.app.test_client().post("/conjurer", json=_payload("u1"))
+    assert resp.status_code == 200
+    assert inbox.contains("u1")  # durable before ack
+    assert _drain_incoming() == [_payload("u1")]
+
+
+def test_resend_while_pending_is_not_requeued(spool):
+    client = cs.app.test_client()
+    client.post("/conjurer", json=_payload("u2"))
+    _drain_incoming()  # consume the first queueing
+    # Resend before it was rendered: inbox still holds it -> dropped, not doubled.
+    client.post("/conjurer", json=_payload("u2"))
+    assert _drain_incoming() == []
+
+
+def test_resend_after_delivery_is_dropped(spool):
+    inbox, delivered = spool
+    client = cs.app.test_client()
+    client.post("/conjurer", json=_payload("u3"))
+    _drain_incoming()
+    cs.mark_delivered("u3")
+    assert not inbox.contains("u3")
+    assert delivered.contains("u3")
+    # A late resend of an already-delivered result must not re-render.
+    client.post("/conjurer", json=_payload("u3"))
+    assert _drain_incoming() == []
+
+
+def test_replay_requeues_only_undelivered(spool):
+    inbox, delivered = spool
+    inbox.put("u4", _payload("u4"))
+    inbox.put("u5", _payload("u5"))
+    delivered.put("u5", {})  # u5 already shown to the user
+    cs.replay_inbox()
+    keys = [list(p.keys())[0] for p in _drain_incoming()]
+    assert keys == ["u4"]  # only the un-rendered one replayed
+    assert not inbox.contains("u5")  # the delivered one is cleaned from the inbox
+
+
+def test_empty_result_still_persisted_and_delivered(spool):
+    inbox, _delivered = spool
+    cs.app.test_client().post("/conjurer", json={"u6": {}})
+    assert inbox.contains("u6")
+    assert _drain_incoming() == [{"u6": {}}]
+
+
+def test_pong_is_not_persisted(spool):
+    inbox, _delivered = spool
+    cs.app.test_client().post("/conjurer", json={"__pong__": "ping-1"})
+    assert len(inbox) == 0
+    assert _drain_incoming() == [{"__pong__": "ping-1"}]

tests/unit/test_durable_queue.py

@@ -0,0 +1,76 @@
+"""Unit tests for the disk-backed durable queue used by result delivery."""
+import json
+
+from durable_queue import DiskQueue
+
+
+def test_put_contains_remove(tmp_path):
+    q = DiskQueue(str(tmp_path / "q"))
+    assert not q.contains("a")
+    q.put("a", {"hello": 1})
+    assert q.contains("a")
+    q.remove("a")
+    assert not q.contains("a")
+    q.remove("a")  # idempotent - no error on missing
+
+
+def test_put_overwrites_and_roundtrips_payload(tmp_path):
+    q = DiskQueue(str(tmp_path / "q"))
+    q.put("uuid-1", {"uuid-1": {"10.1/x": {"Title": ["P"], "type": "article"}}})
+    q.put("uuid-1", {"uuid-1": {"changed": True}})
+    items = q.items()
+    assert len(items) == 1
+    key, payload, _ts = items[0]
+    assert key == "uuid-1"
+    assert payload == {"uuid-1": {"changed": True}}
+
+
+def test_items_sorted_oldest_first(tmp_path, monkeypatch):
+    q = DiskQueue(str(tmp_path / "q"))
+    import durable_queue
+
+    times = iter([100.0, 200.0, 300.0])
+    monkeypatch.setattr(durable_queue.time, "time", lambda: next(times))
+    q.put("c", {})
+    q.put("a", {})
+    q.put("b", {})
+    assert [k for k, _p, _ts in q.items()] == ["c", "a", "b"]
+
+
+def test_corrupt_file_is_skipped_not_fatal(tmp_path):
+    directory = tmp_path / "q"
+    q = DiskQueue(str(directory))
+    q.put("good", {"ok": 1})
+    (directory / "broken.json").write_text("{ this is not json", encoding="utf-8")
+    keys = q.keys()
+    assert keys == ["good"]  # broken file skipped, good one survives
+
+
+def test_prune_keeps_newest(tmp_path, monkeypatch):
+    q = DiskQueue(str(tmp_path / "q"))
+    import durable_queue
+
+    times = iter([1.0, 2.0, 3.0, 4.0, 5.0])
+    monkeypatch.setattr(durable_queue.time, "time", lambda: next(times))
+    for key in ("k1", "k2", "k3", "k4", "k5"):
+        q.put(key, {})
+    dropped = q.prune(2)
+    assert dropped == 3
+    assert set(q.keys()) == {"k4", "k5"}
+
+
+def test_atomic_write_leaves_no_tmp_files(tmp_path):
+    directory = tmp_path / "q"
+    q = DiskQueue(str(directory))
+    q.put("a", {"x": 1})
+    leftover = [p.name for p in directory.iterdir() if p.suffix == ".tmp"]
+    assert leftover == []
+
+
+def test_key_with_slashes_is_sanitised(tmp_path):
+    q = DiskQueue(str(tmp_path / "q"))
+    q.put("../../etc/passwd", {"evil": 1})
+    # Stays inside the directory (no traversal), and round-trips by key.
+    files = list((tmp_path / "q").iterdir())
+    assert all(f.parent == tmp_path / "q" for f in files)
+    assert q.items()[0][1] == {"evil": 1}