fix: read an unprovisioned archive tier without reporting schema skew

polylogue/cli/commands/status.py

@@ -15,6 +15,7 @@
 import click
 
 from polylogue.cli.shared.types import AppEnv
+from polylogue.core.errors import SchemaSkewError
 from polylogue.logging import get_logger
 from polylogue.operations.status_protocol import StatusComponentRegistry, StatusComponentSpec
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
@@ -499,7 +500,11 @@ def _archive_one_tier_status(tier: str, path: Path) -> dict[str, Any]:
             status["table_count_precision"] = precision
         finally:
             conn.close()
-    except sqlite3.Error as exc:
+    except (sqlite3.Error, SchemaSkewError) as exc:
+        # Row detail is an extra on top of the shared probe. A tier this
+        # runtime cannot read still has reportable existence/size/version
+        # facts, so the skew is recorded as data rather than raised through
+        # the status surface.
         status["error"] = str(exc)
     return status
 

polylogue/storage/sqlite/connection_profile.py

@@ -579,6 +579,12 @@ def _schema_skew_remedy(tier: ArchiveTier) -> str:
     )
 
 
+def _tier_holds_no_schema(conn: sqlite3.Connection) -> bool:
+    """Report whether a tier file carries any non-internal schema object."""
+    row = conn.execute("SELECT 1 FROM sqlite_master WHERE name NOT LIKE 'sqlite_%' LIMIT 1").fetchone()
+    return row is None
+
+
 def _assert_schema_supported(conn: sqlite3.Connection, path: str | Path, tier: ArchiveTier | None) -> None:
     """Reject a known archive tier before any caller can issue SQL against it."""
     from polylogue.core.errors import SchemaSkew
@@ -593,6 +599,12 @@ def _assert_schema_supported(conn: sqlite3.Connection, path: str | Path, tier: A
     except KeyError as exc:
         raise ValueError(f"unknown archive tier: {resolved_tier!r}") from exc
     found = int(conn.execute("PRAGMA user_version").fetchone()[0])
+    if found == 0 and _tier_holds_no_schema(conn):
+        # A tier file with neither a version stamp nor any schema object has
+        # never been provisioned. Reading it is reading an absent tier: the
+        # caller fails on the missing table it asked for, which is a truthful
+        # not-provisioned answer, where skew would misreport corruption.
+        return
     if resolved_tier is ArchiveTier.INDEX and found == 0:
         return
     if found != expected:

tests/unit/operations/test_archive_debt.py

@@ -234,6 +234,16 @@ def test_archive_debt_preserves_unknown_embedding_message_counts(
     assert row.caveats == ("Run `polylogue ops embed status --detail` for bounded exact-count attempts.",)
 
 
+def _stamp_tier_version(path: Path, tier: ArchiveTier) -> None:
+    """Stamp a hand-built tier fixture with the version its schema emulates."""
+    conn = sqlite3.connect(path)
+    try:
+        conn.execute(f"PRAGMA user_version = {ARCHIVE_TIER_SPECS[tier].version}")
+        conn.commit()
+    finally:
+        conn.close()
+
+
 def _init_raw_materialization_fixture(root: Path) -> tuple[Path, Path, Path]:
     source_db = root / "source.db"
     index_db = root / "index.db"
@@ -440,6 +450,11 @@ def _init_raw_materialization_fixture(root: Path) -> tuple[Path, Path, Path]:
     with sqlite3.connect(index_db) as conn:
         conn.execute("CREATE TABLE sessions (session_id TEXT, origin TEXT, native_id TEXT, raw_id TEXT)")
 
+    # A tier holding tables must declare the schema version it emulates; the
+    # read guard treats a stamped-zero populated tier as skew.
+    _stamp_tier_version(source_db, ArchiveTier.SOURCE)
+    _stamp_tier_version(index_db, ArchiveTier.INDEX)
+
     return source_db, index_db, source_file
 
 

tests/unit/storage/test_connection_profile.py

@@ -228,12 +228,43 @@ def test_index_write_profiles_refuse_stale_sibling_before_attach(
     with sqlite3.connect(index_path) as connection:
         connection.execute(f"PRAGMA user_version = {ARCHIVE_VERSION_BY_TIER[ArchiveTier.INDEX]}")
     sibling_path = root / f"{sibling_tier.value}.db"
+    # A version this runtime cannot serve in either direction. Stepping one
+    # below the expected version collapses onto 0 for a version-1 tier, which
+    # is the never-provisioned sentinel rather than a skewed schema.
+    skewed_version = ARCHIVE_VERSION_BY_TIER[sibling_tier] + 1
     with sqlite3.connect(sibling_path) as connection:
-        connection.execute(f"PRAGMA user_version = {ARCHIVE_VERSION_BY_TIER[sibling_tier] - 1}")
+        connection.execute(f"PRAGMA user_version = {skewed_version}")
 
     with pytest.raises(SchemaSkew) as excinfo:
         factory(index_path)
 
     assert excinfo.value.tier == sibling_tier.value
     assert excinfo.value.expected == ARCHIVE_VERSION_BY_TIER[sibling_tier]
-    assert excinfo.value.found == ARCHIVE_VERSION_BY_TIER[sibling_tier] - 1
+    assert excinfo.value.found == skewed_version
+
+
+def test_unprovisioned_durable_tier_opens_instead_of_reporting_skew(tmp_path: Path) -> None:
+    """An empty tier file carries no schema, so it cannot be skewed against one.
+
+    Anti-vacuity: restoring the version-only comparison makes the open raise
+    ``SchemaSkew`` for ``found == 0``.
+    """
+    db_path = tmp_path / "source.db"
+    sqlite3.connect(db_path).close()
+
+    with connection_profile.open_readonly_connection(db_path) as connection:
+        assert connection.execute("PRAGMA user_version").fetchone() == (0,)
+
+
+def test_populated_durable_tier_at_version_zero_still_reports_skew(tmp_path: Path) -> None:
+    """A durable tier holding tables without a version stamp is skew, not a fresh file.
+
+    Anti-vacuity: exempting every ``found == 0`` tier regardless of its schema
+    makes this open succeed.
+    """
+    db_path = tmp_path / "source.db"
+    with sqlite3.connect(db_path) as connection:
+        connection.execute("CREATE TABLE raw_sessions (raw_id TEXT PRIMARY KEY)")
+
+    with pytest.raises(SchemaSkew, match="source schema skew"):
+        connection_profile.open_readonly_connection(db_path)

tests/unit/storage/test_durable_change_train.py

@@ -3483,6 +3483,11 @@ def test_bootstrap_reconciles_and_persists_interrupted_train_evidence(
         "CREATE TABLE durable_items (item_id TEXT PRIMARY KEY, payload TEXT NOT NULL) STRICT;"
     )
     monkeypatch.setattr(bootstrap, "ARCHIVE_DDL_BY_TIER", ddl)
+    # Fresh-DDL parity projects the runtime target back to the train's slot.
+    # Left unpatched, the runner replays the real source migration history over
+    # this synthetic tier and fails on tables the fixture never declares.
+    monkeypatch.setattr(migration_runner, "ARCHIVE_VERSION_BY_TIER", versions)
+    monkeypatch.setattr(migration_runner, "ARCHIVE_DDL_BY_TIER", ddl)
     db_path = tmp_path / "source.db"
     _create_current_database(db_path)
     train = _admitted(ArchiveTier.SOURCE, rider=_production_rider())
@@ -3518,6 +3523,11 @@ def test_bootstrap_finishes_persisted_applied_train_without_reapplying(
         "CREATE TABLE durable_items (item_id TEXT PRIMARY KEY, payload TEXT NOT NULL) STRICT;"
     )
     monkeypatch.setattr(bootstrap, "ARCHIVE_DDL_BY_TIER", ddl)
+    # Fresh-DDL parity projects the runtime target back to the train's slot.
+    # Left unpatched, the runner replays the real source migration history over
+    # this synthetic tier and fails on tables the fixture never declares.
+    monkeypatch.setattr(migration_runner, "ARCHIVE_VERSION_BY_TIER", versions)
+    monkeypatch.setattr(migration_runner, "ARCHIVE_DDL_BY_TIER", ddl)
     db_path = tmp_path / "source.db"
     _create_current_database(db_path)
     train = _admitted(ArchiveTier.SOURCE, rider=_production_rider())