test(schema): parity checks schema.py against a fresh install in both directions (#3395)

tests/registry.json

@@ -3954,6 +3954,19 @@
       ],
       "description": "A fan-out batch is joined from the DB instead of gathered in a coroutine (#2524 - #1081 Phase 4's async fan-out join + sync edge adapter). The sync aggregate is rebuilt from the batch's schedule_executions rows (fan_out_id + the caller's fan_out_task_id) rather than from execute_task return values, which a pull dispatch never fills. Pinned: the join wakes the caller only when no row is open, is idempotent under re-delivery, costs one PK read and no batch COUNT on non-fan-out terminals, and is reached from spawn_task_terminal_event with a guard independent of #2523's loop advance. A row is created at SLOT GRANT, never up front (#2524 review): a pre-created RUNNING row waiting behind the semaphore is bulk-FAILed by the #106 no-session sweep, and a QUEUED one is claimable by the backlog drain / pull workers while the service also dispatches it. The row carries the SUB-004 subscription snapshot. A QUEUED return leaves the batch open, and a sync call waits on the rows and answers with the worker's result. max_concurrency still paces push. execute_task's error_code reaches the aggregate (no column holds it). A deadline reports the undispatched tail as running, not failed, and never cancels the dispatch. The default wait budget covers ceil(N / min(max_concurrency, max_parallel_tasks)) waves. A raised dispatch is closed by a proper terminal writer (activity close + terminal event on a won CAS, nothing on a lost one). Also covers task_execution_service.dispatch_and_await_terminal, the sync edge adapter that unstranded a2a and operator_response: a push dispatch is returned untouched, a QUEUED one waits and rebuilds from the row, a timed-out wait returns FAILED/TIMEOUT without raising, and a vanished row falls back to the dispatch result."
     },
+    {
+      "file": "unit/test_schema_parity.py",
+      "feature": "#713, #3395",
+      "added": "2026-10-26",
+      "categories": [
+        "backend",
+        "database",
+        "migrations",
+        "unit",
+        "guard"
+      ],
+      "description": "schema.py vs the fresh-install init_database() lifecycle (migrations, init_schema, migrations) on in-memory SQLite, in both directions. Forward (#713): nothing a migration creates is missing from schema.py, with column-type drift. Reverse (#3395): every table, column, index and trigger schema.py declares exists after a fresh install, so a creating migration's CREATE TABLE losing a column, or a second-pass _atomic_rebuild dropping indexes or triggers, goes red. Blind to a column missing only on existing installs (an unregistered migration on a table only init_schema creates); names and types only, not index/trigger SQL or column constraints."
+    },
     {
       "file": "unit/test_2524_fanout_real_schema.py",
       "feature": "#2524",

tests/unit/test_schema_parity.py

@@ -2,24 +2,48 @@
 Schema Parity Test (test_schema_parity.py)
 
 Issue #713 — Fail PRs that add a migration without updating schema.py.
+Issue #3395 — and fail the reverse: an object schema.py declares that the
+fresh-install lifecycle never produces.
 
-Builds two in-memory SQLite snapshots and asserts they match:
+Builds two in-memory SQLite snapshots and compares them in BOTH directions:
 
   DB_schema = init_schema(empty)                                # canonical declarative
   DB_full   = run_all_migrations() + init_schema() + run_all_migrations()  # init_database() lifecycle
 
+DB_full replays the SQLite branch of ``database.py::init_database`` on an
+empty database: first migration pass, ``init_schema``, second migration pass.
 If ``schema.py`` is faithful, the two snapshots are identical (modulo the
 ``schema_migrations`` tracker table that ``run_all_migrations`` creates
 internally). Any drift fails with a clear message naming the offending
 table / column / index / trigger.
 
+* ``TestSchemaParity`` (forward, ``DB_full - DB_schema``): a migration created
+  something ``schema.py`` does not declare.
+* ``TestReverseSchemaParity`` (reverse, ``DB_schema - DB_full``): ``schema.py``
+  declares something a fresh install does not end up with. Two ways that
+  happens: an old creating migration's ``CREATE TABLE`` loses a column that
+  ``init_schema``'s ``IF NOT EXISTS`` then never adds back, or a rebuild
+  migration on the second pass (after ``init_schema``) drops indexes or
+  triggers, since ``_atomic_rebuild(..., indexes=())`` recreates none.
+
 The full ``init_database()`` lifecycle is required: many migrations
 short-circuit when their target table already exists (e.g.
-``_migrate_audit_log_table`` at ``migrations.py:1356-1364`` returns early
-when ``audit_log`` exists). Running migrations on an empty DB *first*
-forces each migration to create the full table+indexes+triggers it
-intended to, exposing schema.py omissions that the early-return would
-otherwise hide.
+``_migrate_audit_log_table`` returns early when ``audit_log`` exists).
+Running migrations on an empty DB *first* forces each migration to create
+the full table+indexes+triggers it intended to, exposing schema.py omissions
+that the early-return would otherwise hide.
+
+Shape and limits. Both fixtures are the fresh-install shape (an empty
+database, no rows, no prior install), so this file says nothing about a
+database that already exists. In particular the reverse check is blind to a
+column whose migration was never registered when its table is created only
+by ``init_schema``: on a fresh install that migration hits "no such table",
+is skipped, and ``init_schema`` then creates the table with the column, so
+both snapshots agree while every existing install never gets the column.
+That existing-install shape needs a database built by older code, which
+neither fixture is, so it is out of reach here (#3395). Only names (and,
+forward, column types) are compared, not index or trigger SQL or column
+defaults / NOT NULL / primary keys.
 """
 
 import importlib.util
@@ -108,7 +132,7 @@ def production_snapshot() -> dict:
     """Snapshot of the full ``init_database()`` lifecycle:
     ``run_all_migrations`` -> ``init_schema`` -> ``run_all_migrations``.
 
-    Mirrors ``src/backend/database.py:139-164`` exactly. The first migrations
+    Mirrors the SQLite branch of ``database.py::init_database``. The first migrations
     pass on an empty DB lets every migration create the table+indexes+triggers
     it intended to, exposing any short-circuit-hidden drift in schema.py.
     """
@@ -170,3 +194,56 @@ def test_no_missing_triggers(self, schema_snapshot: dict, production_snapshot: d
             f"not in init_schema(): {missing}. Add CREATE TRIGGER statements "
             "to src/backend/db/schema.py."
         )
+
+
+class TestReverseSchemaParity:
+    """Every object ``schema.py`` declares must exist after a fresh install (#3395).
+
+    The other shape this runs is the fresh install: both fixtures start from an
+    empty database, and the drift caught here (a creating migration's
+    ``CREATE TABLE`` losing a column, a second-pass rebuild dropping indexes or
+    triggers) exists only on that shape.
+    """
+
+    def test_no_schema_only_tables(self, schema_snapshot: dict, production_snapshot: dict) -> None:
+        missing = sorted(schema_snapshot["tables"] - production_snapshot["tables"])
+        assert not missing, (
+            "reverse schema drift: tables declared in init_schema() are absent "
+            f"after init_database() on a fresh install: {missing}. A migration "
+            "that runs after init_schema() dropped or renamed them."
+        )
+
+    def test_no_schema_only_columns(self, schema_snapshot: dict, production_snapshot: dict) -> None:
+        # Tables missing outright are reported by test_no_schema_only_tables.
+        missing_cols = sorted(
+            f"{table}.{col_name} ({col_type})"
+            for table, schema_cols in schema_snapshot["columns"].items()
+            if table in production_snapshot["columns"]
+            for col_name, col_type in schema_cols.items()
+            if col_name not in production_snapshot["columns"][table]
+        )
+        assert not missing_cols, (
+            "reverse schema drift: columns declared in init_schema() are absent "
+            "after init_database() on a fresh install:\n  " + "\n  ".join(missing_cols)
+            + "\nThe fresh-install lifecycle never creates them: register their "
+            "migration in MIGRATIONS (src/backend/db/migrations.py), or fix the "
+            "creating migration's CREATE TABLE."
+        )
+
+    def test_no_schema_only_indexes(self, schema_snapshot: dict, production_snapshot: dict) -> None:
+        missing = sorted(schema_snapshot["indexes"] - production_snapshot["indexes"])
+        assert not missing, (
+            "reverse schema drift: indexes declared in init_schema() are absent "
+            f"after init_database() on a fresh install: {missing}. A migration "
+            "that runs after init_schema() dropped them (check the indexes= "
+            "argument of _atomic_rebuild in src/backend/db/migrations.py)."
+        )
+
+    def test_no_schema_only_triggers(self, schema_snapshot: dict, production_snapshot: dict) -> None:
+        missing = sorted(schema_snapshot["triggers"] - production_snapshot["triggers"])
+        assert not missing, (
+            "reverse schema drift: triggers declared in init_schema() are absent "
+            f"after init_database() on a fresh install: {missing}. A migration "
+            "that runs after init_schema() dropped them (a table rebuild through "
+            "_atomic_rebuild drops its triggers; recreate them after the swap)."
+        )