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,37 @@ 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

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "library-desk"
-version = "1.9.0"
+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/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/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):

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
+        )