fix: bootstrap durable operation fixtures

polylogue/operations/durable_change_train.py

@@ -723,9 +723,14 @@ def _audit_file_identity(path: Path) -> tuple[int, int]:
     return metadata.st_dev, metadata.st_ino
 
 
-def _audit_live_metadata(audit_path: Path) -> tuple[int, int, tuple[str, ...]]:
-    """Read the durable markers that remain valid after an in-place migration."""
-    uri = f"{audit_path.resolve(strict=False).as_uri()}?mode=ro"
+def _audit_live_metadata(
+    audit_path: Path,
+    *,
+    immutable: bool = False,
+) -> tuple[int, int, tuple[str, ...]]:
+    """Read durable markers, without mutating an immutable backup artifact."""
+    mode = "?mode=ro&immutable=1" if immutable else "?mode=ro"
+    uri = f"{audit_path.resolve(strict=False).as_uri()}{mode}"
     with closing(sqlite3.connect(uri, uri=True)) as connection:
         version = int(connection.execute("PRAGMA user_version").fetchone()[0] or 0)
         application_id = int(connection.execute("PRAGMA application_id").fetchone()[0] or 0)
@@ -1372,7 +1377,10 @@ def restore_adopted_audit_tier(
     expected_initial_version = adoption.get("audit_user_version")
     if not isinstance(expected_application_id, int) or not isinstance(expected_initial_version, int):
         raise MigrationError("audit adoption receipt lacks its durable SQLite markers")
-    backup_version, backup_application_id, backup_quick_check = _audit_live_metadata(manifest_path.parent / "audit.db")
+    backup_version, backup_application_id, backup_quick_check = _audit_live_metadata(
+        manifest_path.parent / "audit.db",
+        immutable=True,
+    )
     if (
         backup_version != artifact_version
         or backup_version != ARCHIVE_VERSION_BY_TIER[ArchiveTier.AUDIT]

tests/unit/cli/test_archive_maintenance_cli.py

@@ -3638,6 +3638,8 @@ def test_migrate_tier_cli_restores_adopted_audit_from_verified_full_evidence(
     assert restored.exit_code == 0, restored.output
     payload = json.loads(restored.stdout)
     assert payload["restore_receipt"].endswith(".committed.json")
+    backup_audit = Path(verified.output_path) / "audit.db"
+    assert all(not Path(f"{backup_audit}{suffix}").exists() for suffix in ("-wal", "-shm", "-journal"))
     with sqlite3.connect(audit_path) as connection:
         assert connection.execute("PRAGMA user_version").fetchone() == (ARCHIVE_VERSION_BY_TIER[ArchiveTier.AUDIT],)
         assert connection.execute("SELECT generation FROM audit_continuity_head").fetchone() == (2,)

tests/unit/cli/test_reset.py

@@ -12,7 +12,8 @@
 from click.testing import CliRunner
 
 from polylogue.cli import cli
-from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database
+from polylogue.storage.archive_identity import ArchiveLocation
+from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root, initialize_archive_database
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
 from tests.infra.cli_subprocess import run_cli, setup_isolated_workspace
 
@@ -30,9 +31,9 @@
 
 
 def _seed_archive_session(archive_root: Path, *, native_id: str, source_path: Path | None = None) -> str:
+    initialize_active_archive_root(archive_root)
     source_db = archive_root / "source.db"
-    index_db = archive_root / "index.db"
-    initialize_archive_database(source_db, ArchiveTier.SOURCE)
+    index_db = ArchiveLocation.resolve(archive_root).active_index_path
     initialize_archive_database(index_db, ArchiveTier.INDEX)
     session_id = f"codex-session:{native_id}"
     raw_id = f"raw-{native_id}"
@@ -414,6 +415,9 @@ def test_reset_session_records_archive_suppression_and_deletes_archive_row(
 
         archive_root = tmp_path / "archive"
         archive_root.mkdir()
+        active_index = tmp_path / "index-generation" / "index.db"
+        active_index.parent.mkdir()
+        (archive_root / ".index-active-pointer").write_text(str(active_index), encoding="utf-8")
         session_id = _seed_archive_session(archive_root, native_id="reset-one")
 
         with patch("polylogue.cli.commands.reset.archive_root", return_value=archive_root):
@@ -423,8 +427,10 @@ def test_reset_session_records_archive_suppression_and_deletes_archive_row(
         assert result.exit_code == 0
         assert "1 suppression" in result.output
         assert "1 archive row" in result.output
-        with sqlite3.connect(archive_root / "index.db") as conn:
+        with sqlite3.connect(active_index) as conn:
             assert conn.execute("SELECT COUNT(*) FROM sessions WHERE session_id = ?", (session_id,)).fetchone()[0] == 0
+        with sqlite3.connect(archive_root / "index.db") as conn:
+            assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0] == 0
         with sqlite3.connect(archive_root / "user.db") as conn:
             row = conn.execute(
                 "SELECT body_text, json_extract(value_json, '$.mode') FROM assertions WHERE kind = 'suppression' AND target_ref = ?",

tests/unit/storage/test_raw.py

@@ -21,7 +21,7 @@
 from polylogue.storage.raw.models import RawSessionStateUpdate
 from polylogue.storage.repository import SessionRepository
 from polylogue.storage.runtime import RawSessionRecord
-from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database
+from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root, initialize_archive_database
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
 from polylogue.storage.sqlite.async_sqlite import SQLiteBackend
 from polylogue.storage.sqlite.queries.raw_reads import get_capture_mode_resolution
@@ -70,8 +70,7 @@ async def test_save_raw_session_new(self, backend: SQLiteBackend) -> None:
             assert row[0] == 1
 
     async def test_repository_update_raw_state_uses_source_tier(self, tmp_path: Path) -> None:
-        initialize_archive_database(tmp_path / "source.db", ArchiveTier.SOURCE)
-        initialize_archive_database(tmp_path / "index.db", ArchiveTier.INDEX)
+        initialize_active_archive_root(tmp_path)
         source_backend = SQLiteBackend(db_path=tmp_path / "source.db")
         try:
             await source_backend.save_raw_session(

tests/unit/storage/test_store_ops.py

@@ -1782,14 +1782,21 @@ async def fake_to_thread(func: Callable[..., object], /, *args: object, **kwargs
 class TestInfraTagAssignment:
     """Property-based tests using the full TagAssignmentSpec strategy."""
 
-    @given(infra_tag_assignment_strategy(min_sessions=2, max_sessions=4))
+    @given(spec=infra_tag_assignment_strategy(min_sessions=2, max_sessions=4))
     @settings(max_examples=10, deadline=None)
-    async def test_tag_assignment_roundtrip_and_counts(self, spec: TagAssignmentSpec) -> None:
+    async def test_tag_assignment_roundtrip_and_counts(
+        self,
+        spec: TagAssignmentSpec,
+        empty_archive_template: Path,
+    ) -> None:
         """Strategy-generated tags are retrievable and counted consistently."""
-        with tempfile.TemporaryDirectory() as tmp_dir:
-            from tests.infra.archive_scenarios import archive_for_scenario_db, native_session_id_for
+        from tests.conftest import _clone_archive_template
+        from tests.infra.archive_scenarios import archive_for_scenario_db, native_session_id_for
 
-            db_path = Path(tmp_dir) / "index.db"
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            archive_root = Path(tmp_dir) / "archive"
+            _clone_archive_template(empty_archive_template, archive_root)
+            db_path = archive_root / "index.db"
             seed_session_graph(db_path, spec.sessions)
 
             repo = archive_for_scenario_db(db_path)