fix(state): run table rebuilds as one write transaction

Port from qwibitai/nanoclaw#3766: hold the SQLite write lock across a
schema migration so two openers cannot interleave.

The SessionDB writer connection is autocommit, so _rebuild_table's
RENAME / CREATE / copy / DROP each committed on its own. A sibling
process (gateway + CLI opening the same state.db) whose SCHEMA_SQL
bootstrap ran between RENAME and CREATE recreated the table empty; the
rebuilder's CREATE then failed with "table already exists", the heal
was logged as "skipped", the rows stayed in *_legacy_pk and the live
table was empty — silent loss of routing / usage accounting rows.

BEGIN IMMEDIATE around the whole sequence (unless the caller already
owns a transaction) makes the sibling wait under its existing lock
patience, and a mid-rebuild failure rolls the RENAME back.

hermes_state_schema.py

@@ -781,7 +781,29 @@ class SessionSchemaMixin:
     @staticmethod
     def _rebuild_table(cursor: sqlite3.Cursor, table: str, legacy_name: str, ddl: str, copy_sql: str, indexes=()) -> None:
         """RENAME *table* to *legacy_name*, CREATE it fresh from *ddl*, copy rows back with
-        *copy_sql*, DROP the legacy copy, recreate *indexes*."""
+        *copy_sql*, DROP the legacy copy, recreate *indexes* — as ONE write transaction.
+
+        The writer connection is autocommit (``isolation_level=None``), so without an explicit
+        BEGIN each statement commits on its own and a sibling process opening the same state.db
+        between RENAME and CREATE runs SCHEMA_SQL's ``CREATE TABLE IF NOT EXISTS`` first: our
+        CREATE then fails with "table already exists", every row is stranded in *legacy_name*
+        and the live table is empty. BEGIN IMMEDIATE holds the write lock for the whole rebuild
+        so the sibling blocks (and retries under its lock patience) instead of interleaving;
+        any failure rolls the RENAME back."""
+        conn = cursor.connection
+        if conn.in_transaction:  # caller already owns the transaction
+            SessionSchemaMixin._rebuild_table_statements(cursor, table, legacy_name, ddl, copy_sql, indexes)
+            return
+        cursor.execute("BEGIN IMMEDIATE")
+        try:
+            SessionSchemaMixin._rebuild_table_statements(cursor, table, legacy_name, ddl, copy_sql, indexes)
+        except BaseException:
+            cursor.execute("ROLLBACK")
+            raise
+        cursor.execute("COMMIT")
+
+    @staticmethod
+    def _rebuild_table_statements(cursor, table, legacy_name, ddl, copy_sql, indexes) -> None:
         cursor.execute(f"ALTER TABLE {table} RENAME TO {legacy_name}")
         cursor.execute(ddl)
         cursor.execute(copy_sql)

tests/hermes_state/test_rebuild_table_atomic.py

@@ -0,0 +1,67 @@
+"""``_rebuild_table`` is one write transaction (port of qwibitai/nanoclaw#3766's idea:
+hold the write lock across a migration so two openers cannot interleave).
+
+The writer connection is autocommit, so RENAME / CREATE / copy / DROP used to commit one
+by one. A sibling process opening state.db between RENAME and CREATE runs SCHEMA_SQL's
+``CREATE TABLE IF NOT EXISTS`` first; the rebuilder's CREATE then fails with "table
+already exists", the rows are stranded in the ``*_legacy`` copy and the live table is
+empty — silent data loss on the gateway + CLI concurrent-open path.
+"""
+
+import sqlite3
+
+import pytest
+
+from hermes_state_schema import SessionSchemaMixin
+
+DDL = "CREATE TABLE t (a INTEGER, b TEXT, c INTEGER NOT NULL DEFAULT 0)"
+COPY = "INSERT INTO t (a, b) SELECT a, b FROM t_legacy"
+
+
+def _db(tmp_path):
+    conn = sqlite3.connect(tmp_path / "t.db", isolation_level=None, timeout=0.05)
+    conn.execute("PRAGMA journal_mode=WAL")
+    conn.execute("CREATE TABLE t (a INTEGER PRIMARY KEY, b TEXT)")
+    conn.executemany("INSERT INTO t VALUES (?, ?)", [(1, "x"), (2, "y")])
+    return conn
+
+
+def test_sibling_opener_between_rename_and_create_cannot_strand_rows(tmp_path):
+    conn = _db(tmp_path)
+    sibling = sqlite3.connect(tmp_path / "t.db", isolation_level=None, timeout=0.05)
+    seen = {}
+
+    class InterleavingCursor(sqlite3.Cursor):
+        def execute(self, sql, *args):
+            result = super().execute(sql, *args)
+            if sql.startswith("ALTER TABLE t RENAME"):
+                # A sibling's schema bootstrap fires in the RENAME→CREATE gap. Its snapshot still
+                # shows the uncommitted-away ``t``, so IF NOT EXISTS is a no-op, and any real write
+                # must be refused (busy) while the rebuild holds the write lock — never applied.
+                try:
+                    sibling.execute("CREATE TABLE IF NOT EXISTS t (a INTEGER, b TEXT, c INTEGER)")
+                    sibling.execute("INSERT INTO t (a, b) VALUES (3, 'z')")
+                    seen["sibling"] = "applied"
+                except sqlite3.OperationalError as exc:
+                    seen["sibling"] = "busy" if ("locked" in str(exc) or "busy" in str(exc)) else str(exc)
+            return result
+
+    SessionSchemaMixin._rebuild_table(conn.cursor(InterleavingCursor), "t", "t_legacy", DDL, COPY)
+    assert seen["sibling"] == "busy"
+    assert not conn.in_transaction
+    names = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")}
+    assert names == {"t"}
+    assert conn.execute("SELECT COUNT(*) FROM t").fetchone()[0] == 2
+    assert conn.execute("SELECT c FROM t WHERE a = 1").fetchone()[0] == 0  # new shape landed
+    sibling.close()
+    conn.close()
+
+
+def test_rebuild_failure_rolls_back_rename(tmp_path):
+    conn = _db(tmp_path)
+    with pytest.raises(sqlite3.OperationalError):
+        SessionSchemaMixin._rebuild_table(conn.cursor(), "t", "t_legacy", DDL, "INSERT INTO nope SELECT * FROM t_legacy")
+    assert not conn.in_transaction
+    assert conn.execute("SELECT COUNT(*) FROM t").fetchone()[0] == 2
+    assert conn.execute("SELECT name FROM sqlite_master WHERE name = 't_legacy'").fetchone() is None
+    conn.close()