chore: release v1.9.1

Ships the Wiki.js change listener supervision fix. Patch release: no API
change, no migration — the listener now reconnects after a database restart
instead of going silently deaf, and /health reports its subscription state.

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

CHANGELOG.md

@@ -7,6 +7,49 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [1.9.1] - 2026-08-08
+
+### Fixed
+
+- The Wiki.js change listener now survives a database restart. It held a single
+  `LISTEN` connection with no supervision, so when the connection dropped it was
+  gone permanently while `running` stayed `True` — the service kept reporting
+  healthy and silently stopped indexing every page edit until someone restarted
+  the container. This happened for real on 2026-08-08 when `postgres-shared` was
+  redeployed. It now detects the drop via asyncpg's termination callback and
+  reconnects with bounded exponential backoff (1s doubling to a 60s cap),
+  retrying indefinitely because a database in maintenance does come back and
+  giving up would recreate the same silent deafness.
+- `WikiChangeListener.running` is derived from the live connection instead of
+  being assigned once at startup, so it can no longer claim a subscription that
+  does not exist.
+
+### Added
+
+- `/health` reports the change listener under `services.wiki_listener`
+  (`subscribed`, `reconnects`, `last_gap_seconds`). Nothing previously exposed
+  its state anywhere, which is why a dead listener went unnoticed. It is
+  deliberately excluded from the overall healthy/degraded verdict: it recovers on
+  its own, and flipping the container unhealthy for the duration of a database
+  outage would add a restart loop to an incident rather than information.
+- On reconnect the listener logs the outage duration and warns that `NOTIFY`
+  events emitted during the gap were lost and cannot be replayed, pointing at
+  `POST /maintenance/integrity-check` to reconcile. A reconciliation pass is not
+  performed automatically — tenant attribution for a changed page is non-trivial
+  here, and guessing it wrong writes content into the wrong user's namespace.
+
+## [1.9.0] - 2026-07-20
+
+### Security
+
+- The Wiki.js integration buttons no longer embed an API key in the browser. The re-index and entity-link endpoints now authenticate via the NPM `/library-desk/` proxy (Authentik session for external users, LAN bypass for internal), verified by a trusted proxy marker header. The previously-embedded key was a full-privilege key served to every wiki visitor; it has been rotated out of service.
+- `verify_api_key` now uses a constant-time comparison.
+
+### Changed
+
+- `static/wikijs-integration.js` calls library-desk same-origin (`/library-desk/...`) with `credentials: same-origin` and no `Authorization` header. Update the Wiki.js code-injection snippet to `<script src="/library-desk/static/wikijs-integration.js">`.
+- Machine callers (the Scheduler) continue to use the Bearer API key against the container-network endpoints; only the two browser endpoints switched to proxy auth.
+
 ## [1.8.1] - 2026-07-19
 
 ### Changed

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "library-desk"
-version = "1.8.1"
+version = "1.9.1"
 description = "Coordination service for The Library system - HybridRAG queries, document ingestion, entity extraction, and knowledge consolidation"
 readme = "README.md"
 requires-python = ">=3.12"

src/core/dependencies.py

@@ -765,7 +765,8 @@ def get_volatile_cache_service() -> "VolatileCacheService":
 
 
 # Authentication
-from fastapi import Security, HTTPException
+import secrets
+from fastapi import Security, HTTPException, Request
 from fastapi.security import HTTPBearer
 
 security = HTTPBearer()
@@ -788,7 +789,7 @@ async def verify_api_key(
     Raises:
         HTTPException: If API key is invalid
     """
-    if credentials.credentials != settings.library_api_key:
+    if not secrets.compare_digest(credentials.credentials, settings.library_api_key):
         raise HTTPException(
             status_code=403,
             detail="Invalid API key"
@@ -796,6 +797,42 @@ async def verify_api_key(
     return credentials.credentials
 
 
+# Header set by NPM only on the authenticated /library-desk/ proxy location.
+# library-desk is bound to loopback (127.0.0.1:8089), so NPM is the only path
+# that can reach it and set this header — a client cannot forge it. NPM also
+# overwrites any client-supplied value via proxy_set_header.
+_PROXY_MARKER_HEADER = "x-library-desk-proxy"
+
+
+async def verify_browser_request(
+    request: Request,
+    settings: SettingsDep,
+) -> str:
+    """
+    Auth for browser-facing endpoints (the Wiki.js integration buttons).
+
+    Accepts the request when it arrives through the authenticated NPM proxy
+    location (Authentik session for external users, or the LAN bypass for
+    internal ones) — identified by the trusted proxy marker header. No secret
+    is carried in the browser. Machine callers may still authenticate with the
+    Bearer API key. Returns the acting user's identity.
+    """
+    if request.headers.get(_PROXY_MARKER_HEADER) == "1":
+        # Authentik injects the identity for externally-authenticated users;
+        # on the LAN bypass these are empty and the endpoint falls back to the
+        # user supplied in the request body.
+        return request.headers.get("x-authentik-email") or "lan"
+
+    # Fallback: server-to-server Bearer API key.
+    auth = request.headers.get("authorization", "")
+    if auth.startswith("Bearer ") and secrets.compare_digest(
+        auth[len("Bearer "):], settings.library_api_key
+    ):
+        return auth[len("Bearer "):]
+
+    raise HTTPException(status_code=401, detail="Unauthenticated")
+
+
 # Service type aliases for FastAPI endpoint dependencies
 # These are defined after the factory functions
 from src.services.vector_service import VectorService

src/main.py

@@ -123,10 +123,19 @@ async def health(settings: Settings = Depends(get_settings)) -> HealthResponse:
     # Check service connectivity
     service_health = await check_service_health()
 
-    # Overall status is healthy if at least Neo4j and Qdrant are up
+    # Overall status is healthy if at least Neo4j and Qdrant are up.
+    #
+    # The Wiki.js change listener is reported below but deliberately excluded
+    # from this decision. It supervises and reconnects itself, and a database
+    # restart would otherwise flip the container unhealthy for the duration of
+    # an outage it is already recovering from. It is reported so the state is
+    # observable at all — previously nothing anywhere exposed it, which is how a
+    # dead listener went unnoticed while this endpoint answered "healthy".
     all_healthy = service_health.get("neo4j", False) and service_health.get("qdrant", False)
     overall_status = "healthy" if all_healthy else "degraded"
 
+    wiki_listener = getattr(app.state, "wiki_listener", None)
+
     return HealthResponse(
         status=overall_status,
         app_name=settings.app_name,
@@ -152,6 +161,11 @@ async def health(settings: Settings = Depends(get_settings)) -> HealthResponse:
                 "url": settings.ollama_url,
                 "model": settings.ollama_llm_model,
                 "healthy": service_health.get("ollama", False)
+            },
+            "wiki_listener": {
+                "subscribed": bool(wiki_listener and wiki_listener.running),
+                "reconnects": getattr(wiki_listener, "reconnects", 0),
+                "last_gap_seconds": getattr(wiki_listener, "last_gap_seconds", None)
             }
         }
     )

src/routers/entity_linking.py

@@ -18,7 +18,7 @@ from src.core.dependencies import (
     get_wiki_service,
     get_graph_service,
     get_ingestion_service,
-    verify_api_key
+    verify_browser_request,
 )
 from src.services.wiki_service import WikiService
 from src.services.graph_service import GraphService
@@ -57,7 +57,7 @@ async def link_entities_in_page(
     wiki_service: WikiService = Depends(get_wiki_service),
     graph_service: GraphService = Depends(get_graph_service),
     ingestion_service: IngestionService = Depends(get_ingestion_service),
-    api_key: str = Depends(verify_api_key)
+    actor: str = Depends(verify_browser_request)
 ) -> EntityLinkingResult:
     """
     Find and link entities mentioned in a wiki page.

src/routers/ingestion.py

@@ -15,7 +15,8 @@ from src.models.ingestion import (
     BatchIngestionResult
 )
 from src.core.dependencies import (
-    get_ingestion_service, verify_api_key, RequiredUserQuery, JobManagerDep
+    get_ingestion_service, verify_api_key, verify_browser_request,
+    RequiredUserQuery, JobManagerDep
 )
 from src.jobs.job_manager import JobManager, JobStatus, JobType
 
@@ -65,7 +66,7 @@ async def ingest_page(
     request: IngestionRequest,
     ingestion: IngestionService = Depends(get_ingestion_service),
     job_manager: JobManagerDep = None,
-    api_key: str = Depends(verify_api_key)
+    actor: str = Depends(verify_browser_request)
 ):
     """
     Ingest a single wiki page into the knowledge base.

src/services/wiki_change_listener.py

@@ -27,21 +27,59 @@ class WikiChangeListener:
     NOTIFY events on INSERT/UPDATE/DELETE to the pages table.
     """
 
+    CHANNEL = 'wiki_page_changes'
+
+    # Bounded exponential backoff between reconnect attempts.
+    BACKOFF_INITIAL_SECONDS = 1.0
+    BACKOFF_MAX_SECONDS = 60.0
+
     def __init__(self):
         self.settings = get_settings()
         self.connection: Optional[asyncpg.Connection] = None
-        self.running = False
 
         # Loop prevention: Track recently processed pages
         # Key: page_id, Value: timestamp of last processing
         self._recent_notifications = {}
         self._debounce_seconds = self.settings.wikijs_change_listener_debounce_seconds
 
-    async def start(self):
-        """Start listening to database changes."""
-        logger.info("Starting Wiki.js database change listener")
+        # Supervision state. A single LISTEN connection does not heal itself the
+        # way an asyncpg pool does, so the drop has to be detected and repaired
+        # explicitly — see _supervise().
+        self._stopping = False
+        self._disconnected = asyncio.Event()
+        self._supervisor_task: Optional[asyncio.Task] = None
+        self._disconnected_at: Optional[datetime] = None
+        self.reconnects = 0
+        self.last_gap_seconds: Optional[float] = None
 
-        # Connect to Wiki.js PostgreSQL database
+    @property
+    def running(self) -> bool:
+        """
+        Whether a live subscription actually exists.
+
+        Derived rather than assigned. The previous implementation set a flag once
+        in start() and never revisited it, so after the connection dropped the
+        listener reported itself as running while being deaf to every event.
+        """
+        return (
+            not self._stopping
+            and self.connection is not None
+            and not self.connection.is_closed()
+        )
+
+    async def start(self):
+        """Start listening to database changes, and keep listening."""
+        logger.info("Starting Wiki.js database change listener")
+        self._stopping = False
+        self._disconnected.clear()
+
+        await self._connect()
+
+        self._supervisor_task = asyncio.create_task(self._supervise())
+        logger.info("Listening for Wiki.js page changes via PostgreSQL NOTIFY")
+
+    async def _connect(self):
+        """Open a connection and subscribe. Raises if the database is unreachable."""
         self.connection = await asyncpg.connect(
             host=self.settings.wikijs_db_host,
             port=self.settings.wikijs_db_port,
@@ -50,18 +88,136 @@ class WikiChangeListener:
             database=self.settings.wikijs_db_name
         )
 
-        # Listen to the wiki_page_changes channel
-        await self.connection.add_listener('wiki_page_changes', self._handle_notification)
+        await self.connection.add_listener(self.CHANNEL, self._handle_notification)
 
-        self.running = True
-        logger.info("Listening for Wiki.js page changes via PostgreSQL NOTIFY")
+        # Must be re-registered on every connection: asyncpg clears its
+        # termination listeners as soon as it fires them, so this is one-shot.
+        self.connection.add_termination_listener(self._on_connection_lost)
+
+    def _on_connection_lost(self, connection):
+        """
+        Called by asyncpg when the connection terminates.
+
+        Dispatched through loop.call_soon, so it must stay synchronous — the work
+        of reconnecting belongs to _supervise(), which this only wakes.
+        """
+        if self._stopping:
+            return
+
+        self._disconnected_at = datetime.now()
+        logger.error(
+            "Wiki.js change listener lost its database connection — "
+            "page changes are NOT being processed until it reconnects"
+        )
+        self._disconnected.set()
+
+    async def _supervise(self, max_iterations: Optional[int] = None) -> int:
+        """
+        Reconnect whenever the subscription drops.
+
+        Args:
+            max_iterations: Stop after N reconnect cycles (None = run forever;
+                            used by tests)
+
+        Returns:
+            Number of completed reconnect cycles
+        """
+        iterations = 0
+        while max_iterations is None or iterations < max_iterations:
+            await self._disconnected.wait()
+            if self._stopping:
+                break
+
+            self._disconnected.clear()
+            await self._reconnect_with_backoff()
+            iterations += 1
+
+        return iterations
+
+    async def _reconnect_with_backoff(self, max_attempts: Optional[int] = None) -> bool:
+        """
+        Re-establish the subscription, backing off between failures.
+
+        Keeps trying indefinitely by default: a database that is down for
+        maintenance will come back, and giving up would recreate exactly the
+        silent-deafness this supervision exists to prevent.
+        """
+        delay = self.BACKOFF_INITIAL_SECONDS
+        attempts = 0
+
+        while not self._stopping and (max_attempts is None or attempts < max_attempts):
+            attempts += 1
+            await self._close_connection()
+
+            try:
+                await self._connect()
+            except Exception as e:
+                logger.warning(
+                    f"Wiki.js change listener reconnect attempt {attempts} failed: {e}; "
+                    f"retrying in {delay:.0f}s"
+                )
+                await asyncio.sleep(delay)
+                delay = min(delay * 2, self.BACKOFF_MAX_SECONDS)
+                continue
+
+            self.reconnects += 1
+            gap = None
+            if self._disconnected_at is not None:
+                gap = (datetime.now() - self._disconnected_at).total_seconds()
+                self.last_gap_seconds = gap
+            self._disconnected_at = None
+
+            # NOTIFY is fire-and-forget: anything emitted while we were gone was
+            # delivered to nobody and cannot be replayed. Say so, and say what
+            # closes the gap, rather than reporting a clean recovery.
+            outage = f" after {gap:.0f}s" if gap is not None else ""
+            logger.warning(
+                f"Wiki.js change listener reconnected{outage} "
+                f"(reconnect #{self.reconnects}). NOTIFY events emitted during the "
+                f"outage were lost and cannot be replayed — run "
+                f"POST /maintenance/integrity-check to reconcile pages that "
+                f"changed while the listener was down."
+            )
+            return True
+
+        return False
+
+    async def _close_connection(self):
+        """Drop the current connection, tolerating one that is already dead."""
+        if not self.connection:
+            return
+
+        try:
+            if not self.connection.is_closed():
+                await self.connection.remove_listener(
+                    self.CHANNEL, self._handle_notification
+                )
+                await self.connection.close()
+        except Exception as e:
+            # A terminated connection raises on both calls; that is expected here.
+            logger.debug(f"Error closing Wiki.js listener connection: {e}")
+        finally:
+            self.connection = None
 
     async def stop(self):
         """Stop listening and close connection."""
-        if self.connection:
-            await self.connection.remove_listener('wiki_page_changes', self._handle_notification)
-            await self.connection.close()
-        self.running = False
+        self._stopping = True
+
+        # Wake the supervisor so it observes _stopping and exits rather than
+        # racing us to reconnect the connection we are about to close.
+        self._disconnected.set()
+
+        if self._supervisor_task:
+            self._supervisor_task.cancel()
+            try:
+                await self._supervisor_task
+            except asyncio.CancelledError:
+                pass
+            except Exception as e:
+                logger.debug(f"Wiki.js listener supervisor ended with: {e}")
+            self._supervisor_task = None
+
+        await self._close_connection()
         logger.info("Stopped Wiki.js change listener")
 
     async def _handle_notification(self, connection, pid, channel, payload):

static/wikijs-integration.js

@@ -2,21 +2,28 @@
  * Library Desk Integration for Wiki.js
  * Combined re-index and entity linking buttons
  *
- * Usage: Add to Wiki.js Code Injection:
- * <script src="http://192.168.86.149:8089/static/wikijs-integration.js"></script>
+ * Usage: Add to Wiki.js Code Injection (served same-origin behind Authentik):
+ * <script src="/library-desk/static/wikijs-integration.js"></script>
+ *
+ * Auth: none in the browser. Requests go same-origin through the NPM
+ * /library-desk/ location, which is gated by Authentik forward-auth with the
+ * LAN bypass — external users are authenticated, LAN users pass through, and
+ * library-desk trusts the proxy marker header. No API key is embedded here.
  */
 (function() {
   'use strict';
 
-  // Auto-detect Library Desk URL
+  // Same-origin base: the script is served from <origin>/library-desk/static/...,
+  // so strip '/static/...' to get the library-desk mount point on this origin.
   const scriptTag = document.currentScript;
   const scriptUrl = scriptTag ? scriptTag.src : '';
-  const libraryDeskUrl = scriptUrl ? scriptUrl.split('/static/')[0] : 'http://192.168.86.149:8089';
+  const libraryDeskUrl = scriptUrl
+    ? scriptUrl.replace(/^https?:\/\/[^/]+/, '').split('/static/')[0]
+    : '/library-desk';
 
   // Shared configuration
   const CONFIG = window.LIBRARY_DESK_CONFIG || {
     libraryDeskUrl: libraryDeskUrl,
-    apiKey: 'af88ed8f44bed81bdb20d0534f1c4547340b29e2aba4963f61a71b993d7eb6e5',
     user: 'jpmschweitzer',
     buttonPosition: 'toolbar', // 'toolbar' or 'floating'
     debug: true
@@ -229,8 +236,8 @@
       // Re-index directly
       const response = await fetch(CONFIG.libraryDeskUrl + '/ingest/page', {
         method: 'POST',
+        credentials: 'same-origin',
         headers: {
-          'Authorization': 'Bearer ' + CONFIG.apiKey,
           'Content-Type': 'application/json'
         },
         body: JSON.stringify({
@@ -334,8 +341,8 @@
       // Call entity linking endpoint
       const response = await fetch(CONFIG.libraryDeskUrl + '/entity-linking/link-page', {
         method: 'POST',
+        credentials: 'same-origin',
         headers: {
-          'Authorization': 'Bearer ' + CONFIG.apiKey,
           'Content-Type': 'application/json'
         },
         body: JSON.stringify({

tests/test_browser_auth.py

@@ -0,0 +1,57 @@
+"""
+Tests for verify_browser_request — the session/proxy auth used by the
+Wiki.js integration endpoints (no secret in the browser).
+"""
+import pytest
+from types import SimpleNamespace
+from fastapi import HTTPException
+from starlette.requests import Request
+
+from src.core.dependencies import verify_browser_request
+
+
+def _request(headers: dict) -> Request:
+    raw = [(k.lower().encode(), v.encode()) for k, v in headers.items()]
+    return Request({"type": "http", "method": "POST", "path": "/ingest/page", "headers": raw})
+
+
+SETTINGS = SimpleNamespace(library_api_key="server-secret-key")
+
+
+@pytest.mark.asyncio
+async def test_proxy_marker_with_authentik_identity_is_accepted():
+    req = _request({"X-Library-Desk-Proxy": "1", "X-Authentik-Email": "user@example.com"})
+    assert await verify_browser_request(req, SETTINGS) == "user@example.com"
+
+
+@pytest.mark.asyncio
+async def test_proxy_marker_on_lan_bypass_falls_back_to_lan():
+    # LAN bypass: proxy marker present, no Authentik identity headers.
+    req = _request({"X-Library-Desk-Proxy": "1"})
+    assert await verify_browser_request(req, SETTINGS) == "lan"
+
+
+@pytest.mark.asyncio
+async def test_valid_api_key_is_accepted_for_machine_callers():
+    req = _request({"Authorization": "Bearer server-secret-key"})
+    assert await verify_browser_request(req, SETTINGS) == "server-secret-key"
+
+
+@pytest.mark.asyncio
+async def test_no_marker_and_no_key_is_rejected():
+    with pytest.raises(HTTPException) as exc:
+        await verify_browser_request(_request({}), SETTINGS)
+    assert exc.value.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_forged_marker_value_is_rejected():
+    # Only the exact NPM-set value "1" is trusted.
+    with pytest.raises(HTTPException):
+        await verify_browser_request(_request({"X-Library-Desk-Proxy": "yes"}), SETTINGS)
+
+
+@pytest.mark.asyncio
+async def test_wrong_api_key_is_rejected():
+    with pytest.raises(HTTPException):
+        await verify_browser_request(_request({"Authorization": "Bearer wrong"}), SETTINGS)

tests/test_required_user.py

@@ -12,19 +12,21 @@ import pytest
 from fastapi.testclient import TestClient
 
 from src.main import app
-from src.core.dependencies import verify_api_key
+from src.core.dependencies import verify_api_key, verify_browser_request
 
 
 @pytest.fixture(scope="module")
 def client():
-    """TestClient with API-key auth stubbed out (no lifespan startup)."""
+    """TestClient with auth stubbed out (no lifespan startup)."""
     app.dependency_overrides[verify_api_key] = lambda: "test-key"
+    app.dependency_overrides[verify_browser_request] = lambda: "test-user"
     try:
         # No context manager: startup/lifespan events are NOT triggered,
         # so no connections to external services are attempted.
         yield TestClient(app)
     finally:
         app.dependency_overrides.pop(verify_api_key, None)
+        app.dependency_overrides.pop(verify_browser_request, None)
 
 
 QUERY_PARAM_ENDPOINTS = [

tests/test_wiki_change_listener.py

@@ -308,6 +308,10 @@ class TestWikiChangeListener:
     async def test_connection_lifecycle(self, listener, mock_settings):
         """Test listener connection start and stop lifecycle."""
         mock_connection = AsyncMock()
+        # asyncpg's is_closed() is synchronous. Left as an AsyncMock it returns a
+        # coroutine, which is always truthy, so the connection would read as
+        # closed the moment `running` started deriving from it.
+        mock_connection.is_closed = MagicMock(return_value=False)
 
         with patch('src.services.wiki_change_listener.asyncpg.connect', return_value=mock_connection) as mock_connect:
             # Start listener
@@ -426,3 +430,168 @@ class TestWikiChangeListenerIntegration:
 
         # This would test actual pg_notify() calls from triggers
         # and verify the listener receives and processes them
+
+
+class TestWikiChangeListenerReconnect:
+    """
+    Supervision and reconnect behaviour.
+
+    Regression cover for 2026-08-08: redeploying postgres-shared dropped the
+    listener's connection and it never came back. The container kept reporting
+    healthy because nothing observed the subscription, so Wiki.js page edits
+    silently stopped being indexed until someone restarted the service.
+    """
+
+    @pytest.mark.asyncio
+    async def test_running_is_false_when_connection_closed(self, listener):
+        """running reflects the live connection, not a flag set once at startup."""
+        connection = MagicMock()
+        connection.is_closed.return_value = False
+        listener.connection = connection
+        assert listener.running is True
+
+        # The exact failure mode: connection dies, nothing reassigns a flag.
+        connection.is_closed.return_value = True
+        assert listener.running is False
+
+    @pytest.mark.asyncio
+    async def test_running_is_false_while_stopping(self, listener):
+        """A listener being torn down does not advertise itself as subscribed."""
+        connection = MagicMock()
+        connection.is_closed.return_value = False
+        listener.connection = connection
+        listener._stopping = True
+        assert listener.running is False
+
+    @pytest.mark.asyncio
+    async def test_termination_callback_wakes_supervisor(self, listener):
+        """asyncpg's termination callback signals the supervisor and records the time."""
+        assert not listener._disconnected.is_set()
+
+        listener._on_connection_lost(MagicMock())
+
+        assert listener._disconnected.is_set()
+        assert listener._disconnected_at is not None
+
+    @pytest.mark.asyncio
+    async def test_termination_callback_ignored_while_stopping(self, listener):
+        """A deliberate shutdown must not trigger a reconnect."""
+        listener._stopping = True
+
+        listener._on_connection_lost(MagicMock())
+
+        assert not listener._disconnected.is_set()
+
+    @pytest.mark.asyncio
+    async def test_supervisor_reconnects_after_drop(self, listener):
+        """One drop produces one reconnect cycle."""
+        connection = MagicMock()
+        connection.is_closed.return_value = True
+        listener.connection = connection
+
+        with patch.object(listener, '_connect', new=AsyncMock()) as connect:
+            listener._disconnected.set()
+            cycles = await listener._supervise(max_iterations=1)
+
+        assert cycles == 1
+        connect.assert_awaited_once()
+        assert listener.reconnects == 1
+
+    @pytest.mark.asyncio
+    async def test_supervisor_exits_without_reconnecting_when_stopping(self, listener):
+        """stop() wakes the supervisor to exit, not to re-establish the connection."""
+        with patch.object(listener, '_connect', new=AsyncMock()) as connect:
+            listener._stopping = True
+            listener._disconnected.set()
+            cycles = await listener._supervise(max_iterations=1)
+
+        assert cycles == 0
+        connect.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_reconnect_backs_off_and_retries(self, listener):
+        """A database that is still down is retried, with growing delay."""
+        attempts = []
+
+        async def fail_twice_then_succeed():
+            attempts.append(1)
+            if len(attempts) < 3:
+                raise OSError("connection refused")
+
+        sleeps = []
+
+        async def fake_sleep(seconds):
+            sleeps.append(seconds)
+
+        with patch.object(listener, '_connect', new=AsyncMock(side_effect=fail_twice_then_succeed)), \
+             patch.object(listener, '_close_connection', new=AsyncMock()), \
+             patch('asyncio.sleep', new=fake_sleep):
+            ok = await listener._reconnect_with_backoff()
+
+        assert ok is True
+        assert len(attempts) == 3
+        assert sleeps == [1.0, 2.0]           # doubling
+        assert listener.reconnects == 1
+
+    @pytest.mark.asyncio
+    async def test_reconnect_backoff_is_capped(self, listener):
+        """Backoff does not grow without bound during a long outage."""
+        sleeps = []
+
+        async def fake_sleep(seconds):
+            sleeps.append(seconds)
+
+        with patch.object(listener, '_connect', new=AsyncMock(side_effect=OSError("down"))), \
+             patch.object(listener, '_close_connection', new=AsyncMock()), \
+             patch('asyncio.sleep', new=fake_sleep):
+            ok = await listener._reconnect_with_backoff(max_attempts=12)
+
+        assert ok is False
+        assert max(sleeps) == listener.BACKOFF_MAX_SECONDS
+        assert listener.reconnects == 0
+
+    @pytest.mark.asyncio
+    async def test_reconnect_records_outage_duration(self, listener):
+        """The gap is measured so the lost-notification window is reportable."""
+        listener._disconnected_at = datetime.now() - timedelta(seconds=30)
+
+        with patch.object(listener, '_connect', new=AsyncMock()), \
+             patch.object(listener, '_close_connection', new=AsyncMock()):
+            await listener._reconnect_with_backoff()
+
+        assert listener.last_gap_seconds is not None
+        assert 29 <= listener.last_gap_seconds <= 32
+        assert listener._disconnected_at is None
+
+    @pytest.mark.asyncio
+    async def test_close_connection_tolerates_dead_connection(self, listener):
+        """Cleaning up an already-terminated connection must not raise."""
+        connection = MagicMock()
+        connection.is_closed.return_value = False
+        connection.remove_listener = AsyncMock(side_effect=Exception("connection is closed"))
+        connection.close = AsyncMock()
+        listener.connection = connection
+
+        await listener._close_connection()
+
+        assert listener.connection is None
+
+    @pytest.mark.asyncio
+    async def test_start_registers_termination_listener(self, listener):
+        """
+        The termination listener is re-registered on every connection.
+
+        asyncpg clears its termination listeners as soon as it fires them, so a
+        registration that happened only once would survive exactly one drop.
+        """
+        connection = MagicMock()
+        connection.add_listener = AsyncMock()
+        connection.is_closed.return_value = False
+
+        with patch('asyncpg.connect', new=AsyncMock(return_value=connection)):
+            await listener._connect()
+
+        connection.add_listener.assert_awaited_once()
+        connection.add_termination_listener.assert_called_once_with(
+            listener._on_connection_lost
+        )