fix(notify): unbreak PG test backend on the notify dispatcher suite

CI's postgres-backend run failed 11 of the new notify tests from #505.
Three independent issues:

1. Migration 053's ``services_notify`` trigger lives only in the
   alembic chain, but the test fixture in conftest.py calls
   ``init_storage(..., run_migrations=False)`` for speed.  That path
   skips migrations and relies on ``metadata.create_all`` for the
   table tree.  Previous alembic-only DDL (migrations 041 / 048
   ``CREATE INDEX CONCURRENTLY`` on workstreams) is performance-only,
   so tests never depended on it.  053's trigger is the first
   behaviorally-required alembic-only DDL in the project — without it
   ``register_service`` doesn't fire NOTIFY and the trigger-filter
   tests time out.

   Fix: declare the trigger function + trigger in ``_schema.py`` and
   attach them via ``sa.event.listen(services, "after_create", ...)``
   DDL events, gated on ``dialect == "postgresql"``.  The same SQL
   constants are imported by migration 053 so there's a single source
   of truth.  Test fixture stays unchanged — ``create_all`` now
   installs the trigger on fresh PG test DBs.  Migration covers the
   upgrade-on-existing-DB path; the two are mutually exclusive given
   ``create_tables = not run_migrations`` in ``init_storage``.

2. NotifyDispatcher tests fired ``storage.notify(...)`` immediately
   after ``d.start()`` and hit a race: the listener thread is
   concurrently calling ``psycopg.connect(listen_url)`` + ``LISTEN
   <channel>`` over the network, so the notify can land before any
   session is listening on the channel and PG drops it (pg_notify
   only routes to sessions LISTEN'ing at COMMIT time).

   Fix: dispatcher gains a ``_listener_ready: threading.Event`` set
   inside ``_listener_loop`` after each successful ``storage.listen``
   open and cleared on disconnect, plus a public
   ``wait_until_ready(timeout)`` method.  Tests use a new
   ``_start_ready(d)`` helper that calls ``start()`` + asserts ready.
   Production callers don't need this (real reactive traffic arrives
   well after startup), but it's the right primitive for any future
   "start dispatcher, immediately send" call site too.

3. ``TestSqliteNotify`` is misnamed — its tests run against whichever
   backend the ``storage`` fixture provides (PG by default in CI).
   Two of its assertions were SQLite-specific:
   ``assert got.pid == 0`` only holds for the synthetic in-process
   path (PG carries real backend PIDs), and
   ``test_synthetic_sweep_emits_after_interval`` is fundamentally
   SQLite-only (no sweep on the PG path).

   Fix: drop the pid assertion (channel + payload are the
   backend-agnostic invariants), add an ``_is_sqlite`` fixture mirror
   of ``_is_postgres``, and gate the sweep test on it.  The sweep
   test also moves from monkey-patching ``stream._sweep_interval`` to
   passing the ``sweep_interval`` kwarg that ``SQLiteBackend.listen``
   now accepts (from the earlier Copilot review fix).

Validated locally against a fresh ``turnstone_test`` PG DB: 263
storage + console + notify tests pass on PG, 257 on SQLite, mypy +
ruff clean.

tests/test_notify_dispatcher.py

@@ -43,12 +43,26 @@ def _wait_for(predicate, deadline_sec: float = 3.0) -> bool:
     return False
 
 
+def _start_ready(d, *, timeout: float = 5.0) -> None:
+    """``d.start()`` + assert the listener is actually listening.
+
+    Closes the start-vs-notify race for backends where ``storage.listen``
+    blocks on the network (Postgres ``LISTEN`` over a fresh psycopg
+    connection): without the sync, a same-thread ``storage.notify`` can
+    fire before the LISTEN registers and the notification is lost.
+    """
+    d.start()
+    if not d.wait_until_ready(timeout=timeout):
+        msg = f"dispatcher listener did not open within {timeout}s"
+        raise AssertionError(msg)
+
+
 class TestSubscribe:
     def test_subscribe_registers_handler(self, dispatcher_factory, storage):
         d = dispatcher_factory(channels=["alpha"])
         seen: list = []
         d.subscribe("alpha", lambda n: seen.append(n))
-        d.start()
+        _start_ready(d)
         # Fire a notify via the storage layer — dispatcher delivers to handler.
         storage.notify("alpha", "hello")
         assert _wait_for(lambda: any(n.payload == "hello" for n in seen))
@@ -62,7 +76,7 @@ class TestSubscribe:
         d = dispatcher_factory(channels=["alpha"])
         seen: list = []
         unsub = d.subscribe("alpha", lambda n: seen.append(n))
-        d.start()
+        _start_ready(d)
         storage.notify("alpha", "first")
         assert _wait_for(lambda: any(n.payload == "first" for n in seen))
         unsub()
@@ -92,7 +106,7 @@ class TestDispatch:
         seen_b: list = []
         d.subscribe("alpha", lambda n: seen_a.append(n))
         d.subscribe("alpha", lambda n: seen_b.append(n))
-        d.start()
+        _start_ready(d)
         storage.notify("alpha", "shared")
         assert _wait_for(lambda: seen_a and seen_b)
         assert seen_a[0].payload == "shared"
@@ -108,7 +122,7 @@ class TestDispatch:
 
         d.subscribe("alpha", _broken)
         d.subscribe("alpha", lambda n: survived.append(n))
-        d.start()
+        _start_ready(d)
         storage.notify("alpha", "after_broken")
         # The second handler runs even though the first raised.
         assert _wait_for(lambda: any(n.payload == "after_broken" for n in survived))
@@ -119,7 +133,7 @@ class TestDispatch:
         seen_b: list = []
         d.subscribe("alpha", lambda n: seen_a.append(n))
         d.subscribe("beta", lambda n: seen_b.append(n))
-        d.start()
+        _start_ready(d)
         storage.notify("alpha", "for_a")
         storage.notify("beta", "for_b")
         assert _wait_for(lambda: seen_a and seen_b)
@@ -320,7 +334,7 @@ class TestCoalescing:
             coalesce_gate.wait(0.05)
 
         d.subscribe("alpha", _slow_handler)
-        d.start()
+        _start_ready(d)
         # Burst of 10 notifies on the same channel — should coalesce
         # down to many fewer handler invocations.
         for i in range(10):

tests/test_storage_notify.py

@@ -42,7 +42,9 @@ class TestSqliteNotify:
             storage.notify("services", '{"op": "INSERT"}')
             got = _drain_until(stream, lambda n: n.payload == '{"op": "INSERT"}')
             assert got.channel == "services"
-            assert got.pid == 0
+            # ``pid`` is 0 on the SQLite synthetic path and the sending
+            # backend's PID on Postgres — both are valid notify shapes,
+            # so don't assert on the value here.
 
     def test_notify_filters_by_channel(self, storage):
         with storage.listen(["services"]) as stream:
@@ -68,12 +70,12 @@ class TestSqliteNotify:
         assert stream.poll(0.05) == []
         stream.close()
 
-    def test_synthetic_sweep_emits_after_interval(self, storage):
-        # Force a short sweep interval via direct attribute override —
-        # the production default (``_SQLITE_NOTIFY_SWEEP_INTERVAL``) is
-        # tuned for an idle dev backstop and is way too long for a test.
-        with storage.listen(["services"]) as stream:
-            stream._sweep_interval = 0.1
+    def test_synthetic_sweep_emits_after_interval(self, storage, _is_sqlite):
+        # Synthetic sweep is fundamentally SQLite-specific — the PG path
+        # uses real ``LISTEN``/``NOTIFY`` and has no sweep tick.  Gate
+        # so the test doesn't false-fail by waiting for a "sweep" notify
+        # that the PG stream will never produce.
+        with storage.listen(["services"], sweep_interval=0.1) as stream:
             # First poll: not yet at the interval, so likely empty.
             stream.poll(0.05)
             # Wait past the interval, then poll again — should emit a
@@ -102,6 +104,14 @@ def _is_postgres(storage):
     return True
 
 
+@pytest.fixture
+def _is_sqlite(storage):
+    """Skip the wrapped test when the active backend isn't SQLite."""
+    if storage.__class__.__name__ != "SQLiteBackend":
+        pytest.skip("SQLite-specific test")
+    return True
+
+
 class TestPostgresNotify:
     def test_round_trip(self, storage, _is_postgres):
         # Open a listener, fire a notify on a regular pooled connection,

turnstone/console/notify_dispatcher.py

@@ -85,6 +85,14 @@ class NotifyDispatcher:
         self._dispatch_thread: threading.Thread | None = None
         self._dispatch_queue: queue.Queue[Notify | None] = queue.Queue(maxsize=_DISPATCH_QUEUE_MAX)
         self._drop_count = 0
+        # Set inside :meth:`_listener_loop` after each successful
+        # ``storage.listen`` open; cleared on disconnect.  Callers use
+        # :meth:`wait_until_ready` after :meth:`start` to block until the
+        # listener is actually listening (matters when the next caller
+        # action is a ``notify`` whose delivery requires the LISTEN to
+        # already be in place — e.g. tests, or any startup-path traffic
+        # that should be reactive from the first event).
+        self._listener_ready = threading.Event()
 
     @property
     def channels(self) -> list[str]:
@@ -127,6 +135,9 @@ class NotifyDispatcher:
                 return
             self._started = True
             self._stopping.clear()
+            # Clear ready so a stop/start cycle's wait_until_ready only
+            # returns True after the new listener has actually opened.
+            self._listener_ready.clear()
             self._listener_thread = threading.Thread(
                 target=self._listener_loop,
                 name="notify-dispatcher-listener",
@@ -144,6 +155,23 @@ class NotifyDispatcher:
             channels=self._channels,
         )
 
+    def wait_until_ready(self, timeout: float = 5.0) -> bool:
+        """Block until the listener has opened its stream, or ``timeout`` elapses.
+
+        Returns ``True`` when the listener is ready (``LISTEN`` issued
+        for every declared channel on PG; subscriber queues registered
+        on SQLite), ``False`` on timeout.  Cleared automatically on
+        disconnect — call again after a reconnect to wait for the next
+        successful reopen.
+
+        Doesn't replace :meth:`start` — call ``start()`` first, then
+        ``wait_until_ready()`` for the explicit sync point.  Production
+        startup typically doesn't need this (the first real event tends
+        to arrive well after the listener is up); tests use it to close
+        the start-vs-notify race window.
+        """
+        return self._listener_ready.wait(timeout=timeout)
+
     def stop(self, timeout: float = 5.0) -> None:
         """Signal shutdown and join the worker threads.
 
@@ -206,6 +234,12 @@ class NotifyDispatcher:
                     if reconcile_pending:
                         self._synthesize_reconcile()
                         reconcile_pending = False
+                    # Signal ``wait_until_ready`` callers that LISTEN is
+                    # in place (PG) / subscriber queues are bound
+                    # (SQLite).  Must come AFTER the synthesize so any
+                    # post-reconnect reconcile reaches handlers before
+                    # the caller assumes "fresh notifies will deliver".
+                    self._listener_ready.set()
                     while not self._stopping.is_set():
                         batch = stream.poll(_LISTENER_POLL_TIMEOUT)
                         for n in batch:
@@ -213,6 +247,7 @@ class NotifyDispatcher:
             except NotifyConnectionError as exc:
                 if self._stopping.is_set():
                     return
+                self._listener_ready.clear()
                 log.warning(
                     "notify_dispatcher.connection_lost",
                     error=str(exc),
@@ -225,6 +260,7 @@ class NotifyDispatcher:
             except Exception:
                 if self._stopping.is_set():
                     return
+                self._listener_ready.clear()
                 log.exception("notify_dispatcher.listener_unexpected_error")
                 reconcile_pending = True
                 if self._stopping.wait(backoff):

turnstone/core/storage/_schema.py

@@ -251,6 +251,76 @@ services = sa.Table(
 
 sa.Index("idx_services_type_heartbeat", services.c.service_type, services.c.last_heartbeat)
 
+
+# -- Postgres NOTIFY trigger on services -----------------------------------
+#
+# Producer side of the ``services`` channel that the console
+# ``NotifyDispatcher`` listens on for reactive node discovery.  Fires on
+# real registry changes (INSERT, DELETE, UPDATE that changes ``url`` or
+# ``metadata``) and stays quiet on heartbeat-only UPDATEs so the 30s × N
+# nodes heartbeat tick doesn't flood the channel.
+#
+# Declared in the schema (not just in migration 053) so the ``after_create``
+# DDL event installs the trigger any time ``metadata.create_all`` builds
+# the ``services`` table — covering fresh dev databases and the test
+# fixture path (``run_migrations=False``).  Migration 053 covers the
+# upgrade-on-existing-DB path; the two are mutually exclusive given the
+# ``create_tables = not run_migrations`` switch in ``init_storage``, so
+# neither double-installs.  SQLite has no equivalent — the in-process
+# notify fan-out and synthetic-sweep covers the dev path consumer-side.
+
+SERVICES_NOTIFY_TRIGGER_FN_NAME = "turnstone_notify_services"
+SERVICES_NOTIFY_TRIGGER_NAME = "services_notify"
+
+SERVICES_NOTIFY_TRIGGER_FN_SQL = f"""
+CREATE OR REPLACE FUNCTION {SERVICES_NOTIFY_TRIGGER_FN_NAME}() RETURNS trigger AS $$
+BEGIN
+    -- Skip heartbeat-only UPDATEs: same url and metadata, only
+    -- ``last_heartbeat`` changed.  ``register_service`` is an UPSERT
+    -- (on_conflict_do_update), so node restarts that change url or
+    -- metadata MUST still fire — only no-op heartbeat ticks stay
+    -- quiet.  IS NOT DISTINCT FROM treats NULLs as equal so a row
+    -- with NULL metadata before/after doesn't trip the diff.
+    IF TG_OP = 'UPDATE'
+       AND OLD.url IS NOT DISTINCT FROM NEW.url
+       AND OLD.metadata IS NOT DISTINCT FROM NEW.metadata THEN
+        RETURN NULL;
+    END IF;
+
+    PERFORM pg_notify(
+        'services',
+        json_build_object(
+            'service_type', COALESCE(NEW.service_type, OLD.service_type),
+            'service_id',   COALESCE(NEW.service_id,   OLD.service_id),
+            'op',           TG_OP
+        )::text
+    );
+    RETURN NULL;
+END;
+$$ LANGUAGE plpgsql;
+"""
+
+SERVICES_NOTIFY_TRIGGER_SQL = f"""
+CREATE TRIGGER {SERVICES_NOTIFY_TRIGGER_NAME}
+AFTER INSERT OR UPDATE OR DELETE ON services
+FOR EACH ROW EXECUTE FUNCTION {SERVICES_NOTIFY_TRIGGER_FN_NAME}();
+"""
+
+sa.event.listen(
+    services,
+    "after_create",
+    sa.DDL(SERVICES_NOTIFY_TRIGGER_FN_SQL).execute_if(  # type: ignore[no-untyped-call]
+        dialect="postgresql"
+    ),
+)
+sa.event.listen(
+    services,
+    "after_create",
+    sa.DDL(SERVICES_NOTIFY_TRIGGER_SQL).execute_if(  # type: ignore[no-untyped-call]
+        dialect="postgresql"
+    ),
+)
+
 # ---------------------------------------------------------------------------
 # Node metadata (per-node key/value with source tracking)
 # ---------------------------------------------------------------------------

turnstone/core/storage/migrations/versions/053_services_notify_trigger.py

@@ -33,63 +33,30 @@ Create Date: 2026-05-10
 import sqlalchemy as sa
 from alembic import op
 
+from turnstone.core.storage._schema import (
+    SERVICES_NOTIFY_TRIGGER_FN_NAME,
+    SERVICES_NOTIFY_TRIGGER_FN_SQL,
+    SERVICES_NOTIFY_TRIGGER_NAME,
+    SERVICES_NOTIFY_TRIGGER_SQL,
+)
+
 revision = "053"
 down_revision = "052"
 branch_labels = None
 depends_on = None
 
 
-_TRIGGER_FN_NAME = "turnstone_notify_services"
-_TRIGGER_NAME = "services_notify"
-
-
-_TRIGGER_FN_SQL = f"""
-CREATE OR REPLACE FUNCTION {_TRIGGER_FN_NAME}() RETURNS trigger AS $$
-BEGIN
-    -- Skip heartbeat-only UPDATEs: same url and metadata, only
-    -- ``last_heartbeat`` changed.  ``register_service`` is an UPSERT
-    -- (on_conflict_do_update), so node restarts that change url or
-    -- metadata MUST still fire — only no-op heartbeat ticks stay
-    -- quiet.  IS NOT DISTINCT FROM treats NULLs as equal so a row
-    -- with NULL metadata before/after doesn't trip the diff.
-    IF TG_OP = 'UPDATE'
-       AND OLD.url IS NOT DISTINCT FROM NEW.url
-       AND OLD.metadata IS NOT DISTINCT FROM NEW.metadata THEN
-        RETURN NULL;
-    END IF;
-
-    PERFORM pg_notify(
-        'services',
-        json_build_object(
-            'service_type', COALESCE(NEW.service_type, OLD.service_type),
-            'service_id',   COALESCE(NEW.service_id,   OLD.service_id),
-            'op',           TG_OP
-        )::text
-    );
-    RETURN NULL;
-END;
-$$ LANGUAGE plpgsql;
-"""
-
-
-_TRIGGER_SQL = f"""
-CREATE TRIGGER {_TRIGGER_NAME}
-AFTER INSERT OR UPDATE OR DELETE ON services
-FOR EACH ROW EXECUTE FUNCTION {_TRIGGER_FN_NAME}();
-"""
-
-
 def upgrade() -> None:
     bind = op.get_bind()
     if bind.dialect.name != "postgresql":
         return
-    op.execute(sa.text(_TRIGGER_FN_SQL))
-    op.execute(sa.text(_TRIGGER_SQL))
+    op.execute(sa.text(SERVICES_NOTIFY_TRIGGER_FN_SQL))
+    op.execute(sa.text(SERVICES_NOTIFY_TRIGGER_SQL))
 
 
 def downgrade() -> None:
     bind = op.get_bind()
     if bind.dialect.name != "postgresql":
         return
-    op.execute(sa.text(f"DROP TRIGGER IF EXISTS {_TRIGGER_NAME} ON services"))
-    op.execute(sa.text(f"DROP FUNCTION IF EXISTS {_TRIGGER_FN_NAME}()"))
+    op.execute(sa.text(f"DROP TRIGGER IF EXISTS {SERVICES_NOTIFY_TRIGGER_NAME} ON services"))
+    op.execute(sa.text(f"DROP FUNCTION IF EXISTS {SERVICES_NOTIFY_TRIGGER_FN_NAME}()"))