fix(state): rebuild legacy gateway_routing PK; guard session_store in dispatch hook

Two log-spam bugs found in live gateway logs:

1. gateway_routing UNIQUE-constraint spam (261 warnings in one errors.log):
   early builds of the #59203 routing-index migration created
   gateway_routing with 'session_key TEXT PRIMARY KEY' and no scope
   column. _reconcile_columns() ADDs the missing scope column but SQLite
   cannot ALTER a primary key, so the shipped composite
   PRIMARY KEY (scope, session_key) never lands on those databases. Both
   write paths then fail on every save:
   - save_gateway_routing_entry: 'ON CONFLICT clause does not match any
     PRIMARY KEY or UNIQUE constraint'
   - replace_gateway_routing_entries: 'UNIQUE constraint failed:
     gateway_routing.session_key' whenever the same session_key exists
     under another scope (e.g. test-suite scopes leaked into a live DB).
   New _heal_gateway_routing_pk() rebuilds the table once with the
   composite key, preserving rows (newest wins on collisions, NULL scope
   coalesced to ''). Same one-time-heal pattern as the #51646 active-
   column repair. Verified E2E against a copy of a real affected state.db.

2. pre_gateway_dispatch warned ''GatewayRunner' object has no attribute
   'session_store'' and silently dropped the hook for every message on
   partially-initialized runners (bare object.__new__ runners in tests,
   and any future init-order change). Pass
   getattr(self, 'session_store', None) so the hook always fires
   (pitfall #17 pattern).

Both regression tests fail without their fixes (sabotage-verified).

gateway/run.py

@@ -11077,7 +11077,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     "pre_gateway_dispatch",
                     event=event,
                     gateway=self,
-                    session_store=self.session_store,
+                    # getattr: bare-runner tests build GatewayRunner via
+                    # object.__new__ without __init__ (pitfall #17), and the
+                    # hook must not fail dispatch over a missing attribute.
+                    session_store=getattr(self, "session_store", None),
                 )
             except Exception as _hook_exc:
                 logger.warning("pre_gateway_dispatch invocation failed: %s", _hook_exc)

hermes_state.py

@@ -3490,6 +3490,79 @@ class SessionDB:
                             "reconcile %s.%s: %s", table_name, col_name, exc,
                         )
 
+    def _heal_gateway_routing_pk(self, cursor: sqlite3.Cursor) -> None:
+        """Rebuild ``gateway_routing`` when its PRIMARY KEY predates scoping.
+
+        Early builds of the routing-index migration (#59203) created the
+        table with ``session_key TEXT PRIMARY KEY`` and no ``scope`` column.
+        ``_reconcile_columns()`` ADDs the missing ``scope`` column on those
+        databases, but SQLite cannot ALTER a primary key, so the shipped
+        composite ``PRIMARY KEY (scope, session_key)`` never lands.  On such
+        tables every write path is broken:
+
+        * ``save_gateway_routing_entry`` fails with "ON CONFLICT clause does
+          not match any PRIMARY KEY or UNIQUE constraint" (its upsert targets
+          the composite key), and
+        * ``replace_gateway_routing_entries`` fails with "UNIQUE constraint
+          failed: gateway_routing.session_key" whenever the same session_key
+          exists under a different scope — the exact isolation the composite
+          key exists to provide.
+
+        Each failed save logs a warning and falls back to sessions.json,
+        so a legacy-shaped table produces endless per-save warning spam.
+        Rebuild it once, preserving rows.  On a session_key collision across
+        scopes (possible while the PK was wrong) the newest row wins.
+        """
+        try:
+            rows = cursor.execute(
+                'PRAGMA table_info("gateway_routing")'
+            ).fetchall()
+        except sqlite3.OperationalError:
+            return
+        if not rows:
+            return
+
+        def _col(row, idx, name):
+            return row[idx] if isinstance(row, (tuple, list)) else row[name]
+
+        pk_cols = [
+            _col(r, 1, "name")
+            for r in sorted(
+                (r for r in rows if _col(r, 5, "pk")),
+                key=lambda r: _col(r, 5, "pk"),
+            )
+        ]
+        if pk_cols == ["scope", "session_key"]:
+            return
+
+        logger.info(
+            "gateway_routing has legacy primary key %r; rebuilding with "
+            "composite (scope, session_key) key",
+            pk_cols,
+        )
+        cursor.execute(
+            "ALTER TABLE gateway_routing RENAME TO gateway_routing_legacy_pk"
+        )
+        cursor.execute(
+            """CREATE TABLE gateway_routing (
+    scope TEXT NOT NULL DEFAULT '',
+    session_key TEXT NOT NULL,
+    entry_json TEXT NOT NULL,
+    updated_at REAL NOT NULL,
+    PRIMARY KEY (scope, session_key)
+)"""
+        )
+        # INSERT OR REPLACE + updated_at ordering: if the broken PK ever let
+        # two scopes race over one session_key, keep the newest row per
+        # (scope, session_key) pair.
+        cursor.execute(
+            "INSERT OR REPLACE INTO gateway_routing "
+            "(scope, session_key, entry_json, updated_at) "
+            "SELECT COALESCE(scope, ''), session_key, entry_json, updated_at "
+            "FROM gateway_routing_legacy_pk ORDER BY updated_at ASC"
+        )
+        cursor.execute("DROP TABLE gateway_routing_legacy_pk")
+
     def _init_schema(self):
         """Create tables and FTS if they don't exist, reconcile columns.
 
@@ -3514,6 +3587,11 @@ class SessionDB:
         # column gets created here.
         self._reconcile_columns(cursor)
 
+        # Rebuild gateway_routing if it still carries the pre-scope PRIMARY
+        # KEY (session_key alone). ADD COLUMN cannot fix a PK, so this is
+        # the one table-shape repair reconciliation can't express.
+        self._heal_gateway_routing_pk(cursor)
+
         # Indexes that reference reconciler-added columns must be created
         # AFTER _reconcile_columns runs — declaring them in SCHEMA_SQL
         # makes the initial executescript fail on legacy DBs (the index's

tests/gateway/test_pre_gateway_dispatch.py

@@ -177,3 +177,35 @@ async def test_internal_events_bypass_hook(monkeypatch):
     # Even though the hook would say skip, internal events bypass it.
     await runner._handle_message(event)
     assert called["count"] == 0
+
+@pytest.mark.asyncio
+async def test_hook_fires_without_session_store_attribute(monkeypatch):
+    """A runner missing session_store still delivers the event to plugins.
+
+    Regression: the hook kwargs read ``self.session_store`` directly, so a
+    partially-initialized runner raised AttributeError inside the dispatch
+    try-block — the hook never fired, and every message logged
+    "pre_gateway_dispatch invocation failed: 'GatewayRunner' object has no
+    attribute 'session_store'". Plugins must receive the event (with
+    session_store=None) instead.
+    """
+    _clear_auth_env(monkeypatch)
+
+    seen = {}
+
+    def _fake_hook(name, **kwargs):
+        if name == "pre_gateway_dispatch":
+            seen["session_store"] = kwargs.get("session_store", "MISSING")
+            return [{"action": "skip", "reason": "plugin-handled"}]
+        return []
+
+    monkeypatch.setattr("hermes_cli.plugins.invoke_hook", _fake_hook)
+
+    runner, adapter = _make_runner(Platform.WHATSAPP)
+    del runner.session_store
+
+    result = await runner._handle_message(_make_event("hi"))
+    assert result is None
+    # Hook actually fired (skip short-circuited before auth) with a None store.
+    assert seen == {"session_store": None}
+    adapter.send.assert_not_awaited()

tests/test_hermes_state.py

@@ -7810,3 +7810,115 @@ class TestDisplayMetadataReadPaths:
         )
         assert db.get_messages_as_conversation("s1")[0]["display_metadata"] == self.META
 
+
+
+class TestGatewayRoutingPkHeal:
+    """Legacy gateway_routing tables (session_key-only PK) get rebuilt on open.
+
+    Early builds of the #59203 routing-index migration created gateway_routing
+    with ``session_key TEXT PRIMARY KEY`` and no ``scope`` column. The column
+    reconciler ADDs ``scope`` but cannot change the PK, so on those databases
+    every routing save failed ("ON CONFLICT clause does not match any PRIMARY
+    KEY or UNIQUE constraint" / "UNIQUE constraint failed:
+    gateway_routing.session_key") and spammed warnings on each save.
+    """
+
+    LEGACY_SQL = """
+        CREATE TABLE gateway_routing (
+            session_key TEXT PRIMARY KEY,
+            entry_json TEXT NOT NULL,
+            updated_at REAL NOT NULL
+        , "scope" TEXT DEFAULT '')
+    """
+
+    def _make_legacy_db(self, tmp_path, rows=()):
+        db_path = tmp_path / "state.db"
+        conn = sqlite3.connect(db_path)
+        conn.execute(self.LEGACY_SQL)
+        conn.executemany(
+            "INSERT INTO gateway_routing (scope, session_key, entry_json, updated_at) "
+            "VALUES (?, ?, ?, ?)",
+            list(rows),
+        )
+        conn.commit()
+        conn.close()
+        return db_path
+
+    def _pk_cols(self, db):
+        rows = db._conn.execute('PRAGMA table_info("gateway_routing")').fetchall()
+        cols = sorted(
+            ((r["pk"], r["name"]) for r in rows if r["pk"]),
+        )
+        return [name for _, name in cols]
+
+    def test_legacy_pk_rebuilt_to_composite(self, tmp_path):
+        db_path = self._make_legacy_db(
+            tmp_path, rows=[("/home/u/.hermes/sessions", "agent:main:telegram:dm:1", "{}", 1.0)]
+        )
+        db = SessionDB(db_path=db_path)
+        try:
+            assert self._pk_cols(db) == ["scope", "session_key"]
+            # Existing rows survive the rebuild.
+            entries = db.load_gateway_routing_entries(scope="/home/u/.hermes/sessions")
+            assert entries == {"agent:main:telegram:dm:1": "{}"}
+        finally:
+            db.close()
+
+    def test_upsert_and_cross_scope_replace_work_after_heal(self, tmp_path):
+        """The two write paths that failed on the legacy shape now succeed."""
+        db_path = self._make_legacy_db(
+            tmp_path, rows=[("scopeA", "agent:main:telegram:dm:1", "{}", 1.0)]
+        )
+        db = SessionDB(db_path=db_path)
+        try:
+            # save_gateway_routing_entry: composite ON CONFLICT upsert.
+            db.save_gateway_routing_entry(
+                "agent:main:telegram:dm:1", '{"v": 2}', scope="scopeA"
+            )
+            assert db.load_gateway_routing_entries(scope="scopeA") == {
+                "agent:main:telegram:dm:1": '{"v": 2}'
+            }
+            # replace_gateway_routing_entries: same session_key, other scope —
+            # the exact collision the 'UNIQUE constraint failed' spam came from.
+            db.replace_gateway_routing_entries(
+                {"agent:main:telegram:dm:1": '{"v": 3}'}, scope="scopeB"
+            )
+            assert db.load_gateway_routing_entries(scope="scopeB") == {
+                "agent:main:telegram:dm:1": '{"v": 3}'
+            }
+            # scopeA untouched by scopeB's replace.
+            assert db.load_gateway_routing_entries(scope="scopeA") == {
+                "agent:main:telegram:dm:1": '{"v": 2}'
+            }
+        finally:
+            db.close()
+
+    def test_cross_scope_collision_rows_all_survive(self, tmp_path):
+        """Rows that shared a session_key across scopes are preserved per scope."""
+        db_path = self._make_legacy_db(tmp_path)
+        # The legacy single-column PK forbids duplicate session_keys, so
+        # simulate what a healed multi-scope DB must support by inserting the
+        # collision AFTER the heal via public APIs (covered above) — here we
+        # instead verify a NULL scope from the reconciler-added column
+        # coalesces to '' rather than violating NOT NULL.
+        conn = sqlite3.connect(db_path)
+        conn.execute(
+            "INSERT INTO gateway_routing (scope, session_key, entry_json, updated_at) "
+            "VALUES (NULL, 'k-null-scope', '{}', 5.0)"
+        )
+        conn.commit()
+        conn.close()
+        db = SessionDB(db_path=db_path)
+        try:
+            assert db.load_gateway_routing_entries(scope="") == {"k-null-scope": "{}"}
+        finally:
+            db.close()
+
+    def test_current_shape_left_untouched(self, tmp_path, db):
+        """A DB born with the composite PK is not rebuilt (idempotence)."""
+        db.save_gateway_routing_entry("k1", "{}", scope="s")
+        assert self._pk_cols(db) == ["scope", "session_key"]
+        # Re-running the heal is a no-op.
+        cur = db._conn.cursor()
+        db._heal_gateway_routing_pk(cur)
+        assert db.load_gateway_routing_entries(scope="s") == {"k1": "{}"}