diff --git a/polylogue/cli/commands/status.py b/polylogue/cli/commands/status.py index 0ee48253ee..17ba611b1a 100644 --- a/polylogue/cli/commands/status.py +++ b/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 diff --git a/polylogue/storage/sqlite/connection_profile.py b/polylogue/storage/sqlite/connection_profile.py index 51001d792c..cf026866c3 100644 --- a/polylogue/storage/sqlite/connection_profile.py +++ b/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: diff --git a/tests/unit/operations/test_archive_debt.py b/tests/unit/operations/test_archive_debt.py index e2e4c3cef3..42a62e3651 100644 --- a/tests/unit/operations/test_archive_debt.py +++ b/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 diff --git a/tests/unit/storage/test_connection_profile.py b/tests/unit/storage/test_connection_profile.py index 1c1d14d200..5095086178 100644 --- a/tests/unit/storage/test_connection_profile.py +++ b/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) diff --git a/tests/unit/storage/test_durable_change_train.py b/tests/unit/storage/test_durable_change_train.py index 4fe321677c..5eed9eeb90 100644 --- a/tests/unit/storage/test_durable_change_train.py +++ b/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())