fix(storage): skip backup manifest requirement for additive-only migrations

polylogue/archive/session/display_mixin.py

@@ -9,7 +9,7 @@
 from typing import TYPE_CHECKING
 
 from polylogue.archive.session.branch_type import BranchType
-from polylogue.types import SessionId
+from polylogue.core.types import SessionId
 
 if TYPE_CHECKING:
     pass

polylogue/cli/commands/maintenance/_migrate_tier.py

@@ -30,12 +30,12 @@
 @click.argument("tier", type=click.Choice(tuple(sorted(tier.value for tier in DURABLE_MIGRATION_TIERS))))
 @click.option(
     "--backup-manifest",
-    required=True,
+    required=False,
     type=click.Path(path_type=Path, exists=True),
-    help="Verified polylogue ops backup manifest or backup directory containing manifest.json.",
+    help="Verified backup manifest. Required only when a selected migration changes existing durable data.",
 )
 @click.option("--output-format", type=click.Choice(["plain", "json"]), default="plain", show_default=True)
-def migrate_tier_command(tier: str, backup_manifest: Path, output_format: str) -> None:
+def migrate_tier_command(tier: str, backup_manifest: Path | None, output_format: str) -> None:
     """Apply additive migrations for one durable archive tier.
 
     Derived tiers are intentionally excluded from this command; rebuild or
@@ -58,7 +58,7 @@ def migrate_tier_command(tier: str, backup_manifest: Path, output_format: str) -
                         "ok": False,
                         "tier": tier,
                         "path": str(path),
-                        "backup_manifest": str(backup_manifest),
+                        "backup_manifest": str(backup_manifest) if backup_manifest is not None else None,
                         "error": str(exc),
                     },
                     indent=2,
@@ -73,7 +73,7 @@ def migrate_tier_command(tier: str, backup_manifest: Path, output_format: str) -
         "ok": True,
         "tier": tier,
         "path": str(path),
-        "backup_manifest": str(backup_manifest),
+        "backup_manifest": str(backup_manifest) if backup_manifest is not None else None,
         "backup_receipt": str(result.backup_receipt) if result.backup_receipt is not None else None,
         "from_version": result.from_version,
         "to_version": result.to_version,

polylogue/storage/sqlite/migration_runner.py

@@ -24,6 +24,7 @@
 _MIGRATION_NAME_RE = re.compile(r"^(?P<version>\d{3,})_[a-z0-9_]+\.sql$")
 _VERIFICATION_RECEIPT_FILE = "verification-receipt.json"
 _SQLITE_SIDECAR_SUFFIXES = ("-wal", "-shm", "-journal")
+_ADDITIVE_NO_BACKUP_MARKER = "-- migration-safety: additive-no-backup"
 
 
 class MigrationError(RuntimeError):
@@ -36,6 +37,7 @@ class MigrationStep:
     version: int
     name: str
     sql: str
+    requires_backup: bool
 
 
 @dataclass(frozen=True, slots=True)
@@ -51,6 +53,17 @@ def _migration_package(tier: ArchiveTier) -> str:
     return f"polylogue.storage.sqlite.migrations.{tier.value}"
 
 
+def _requires_migration_backup(sql: str) -> bool:
+    """A migration opts out of the backup requirement only via a header directive.
+
+    Substring matching would waive the backup requirement if the marker text
+    ever appeared in a comment, SQL string literal, or later in the file --
+    require it to be the file's first non-blank line instead.
+    """
+    first_nonblank = next((line.strip() for line in sql.splitlines() if line.strip()), "")
+    return first_nonblank != _ADDITIVE_NO_BACKUP_MARKER
+
+
 def _load_migrations(tier: ArchiveTier) -> tuple[MigrationStep, ...]:
     if tier not in DURABLE_MIGRATION_TIERS:
         return ()
@@ -63,12 +76,14 @@ def _load_migrations(tier: ArchiveTier) -> tuple[MigrationStep, ...]:
         match = _MIGRATION_NAME_RE.match(item.name)
         if match is None:
             continue
+        sql = item.read_text(encoding="utf-8")
         steps.append(
             MigrationStep(
                 tier=tier,
                 version=int(match.group("version")),
                 name=item.name,
-                sql=item.read_text(encoding="utf-8"),
+                sql=sql,
+                requires_backup=_requires_migration_backup(sql),
             )
         )
     versions = [step.version for step in steps]
@@ -552,46 +567,99 @@ def _execute_migration_sql(conn: sqlite3.Connection, sql: str) -> None:
         raise MigrationError("migration SQL ended with an incomplete statement")
 
 
+def _pending_migration_steps(
+    conn: sqlite3.Connection,
+    tier: ArchiveTier,
+    *,
+    current_version: int,
+    target_version: int,
+) -> tuple[MigrationStep, ...]:
+    steps = tuple(step for step in _load_migrations(tier) if current_version < step.version <= target_version)
+    expected_versions = tuple(range(current_version + 1, target_version + 1))
+    actual_versions = tuple(step.version for step in steps)
+    if actual_versions != expected_versions:
+        raise MigrationError(
+            f"{tier.value} migration chain is incomplete: expected {expected_versions}, found {actual_versions}"
+        )
+    return steps
+
+
 def migrate_archive_tier(
     conn: sqlite3.Connection,
     tier: ArchiveTier,
     *,
-    backup_manifest: Path,
+    backup_manifest: Path | None,
 ) -> MigrationResult:
     """Apply additive migrations for one durable tier."""
     if tier not in DURABLE_MIGRATION_TIERS:
         raise MigrationError(f"{tier.value} tier does not support in-place migrations")
     _checkpoint_live_tier(conn)
-    validate_migration_backup_manifest(backup_manifest, tier, connection=conn)
+    target_version = ARCHIVE_VERSION_BY_TIER[tier]
+
+    # Lock-free precheck: fail fast (no wasted write-lock acquisition) when a
+    # backup manifest is required but missing. This read is intentionally not
+    # authoritative -- a concurrent migrate_archive_tier call could change the
+    # version before this one acquires BEGIN IMMEDIATE below, so every value
+    # computed here is re-derived from a fresh read once the lock is held.
+    precheck_version = int(conn.execute("PRAGMA user_version").fetchone()[0] or 0)
+    if precheck_version == target_version:
+        return MigrationResult(
+            tier=tier,
+            from_version=precheck_version,
+            to_version=target_version,
+            applied_versions=(),
+        )
+    if precheck_version == 0:
+        raise MigrationError(f"{tier.value} tier is empty; initialize it fresh instead of migrating")
+    if precheck_version > target_version:
+        raise MigrationError(
+            f"{tier.value} tier version {precheck_version} is newer than this runtime expects ({target_version})"
+        )
+    precheck_steps = _pending_migration_steps(
+        conn, tier, current_version=precheck_version, target_version=target_version
+    )
+    precheck_requires_backup = any(step.requires_backup for step in precheck_steps)
+    if precheck_requires_backup and backup_manifest is None:
+        raise MigrationError(f"{tier.value} migration requires a verified backup manifest")
+    if precheck_requires_backup:
+        # Baseline validation before acquiring the write lock. The paired
+        # post-lock call below re-validates with the same connection;
+        # _validate_live_source_fingerprint rejects a nonempty WAL, so a
+        # write that lands on the live tier between this call and BEGIN
+        # IMMEDIATE is caught as "changed before the migration lock" instead
+        # of migrating over data the verified backup never covered.
+        assert backup_manifest is not None
+        validate_migration_backup_manifest(backup_manifest, tier, connection=conn)
+
     try:
         conn.execute("BEGIN IMMEDIATE")
-        backup_receipt = validate_migration_backup_manifest(backup_manifest, tier, connection=conn)
+        # Authoritative re-read: a concurrent migration may have advanced (or
+        # completed) the tier between the precheck above and this lock
+        # acquisition. Recomputing here instead of trusting the precheck
+        # avoids failing the per-step version check below with a confusing
+        # "expected version N, found M" instead of the correct no-op result.
         current_version = int(conn.execute("PRAGMA user_version").fetchone()[0] or 0)
-        target_version = ARCHIVE_VERSION_BY_TIER[tier]
         if current_version == target_version:
             conn.rollback()
             return MigrationResult(
                 tier=tier,
                 from_version=current_version,
                 to_version=target_version,
                 applied_versions=(),
-                backup_receipt=backup_receipt,
             )
-        if current_version == 0:
-            raise MigrationError(f"{tier.value} tier is empty; initialize it fresh instead of migrating")
         if current_version > target_version:
             raise MigrationError(
                 f"{tier.value} tier version {current_version} is newer than this runtime expects ({target_version})"
             )
-
-        steps = tuple(step for step in _load_migrations(tier) if current_version < step.version <= target_version)
-        expected_versions = tuple(range(current_version + 1, target_version + 1))
-        actual_versions = tuple(step.version for step in steps)
-        if actual_versions != expected_versions:
-            raise MigrationError(
-                f"{tier.value} migration chain is incomplete: expected {expected_versions}, found {actual_versions}"
-            )
-
+        steps = _pending_migration_steps(conn, tier, current_version=current_version, target_version=target_version)
+        requires_backup = any(step.requires_backup for step in steps)
+        if requires_backup and backup_manifest is None:
+            raise MigrationError(f"{tier.value} migration requires a verified backup manifest")
+        backup_receipt = (
+            validate_migration_backup_manifest(backup_manifest, tier, connection=conn)
+            if requires_backup and backup_manifest is not None
+            else None
+        )
         start_version = current_version
         applied: list[int] = []
         for step in steps:

polylogue/storage/sqlite/migrations/source/010_sinex_publication_obligations.sql

@@ -1,3 +1,4 @@
+-- migration-safety: additive-no-backup
 -- Durable Sinex-backed-mode publication obligation ledger (polylogue-303r.2).
 --
 -- One row per (object_id, protocol_version, revision_id, manifest_digest)

polylogue/storage/sqlite/migrations/source/011_excised_content.sql

@@ -1,3 +1,4 @@
+-- migration-safety: additive-no-backup
 -- Durable removed-content ledger for standalone/off-mode excision
 -- (polylogue-27m). Purely additive: one new table, no existing column or
 -- constraint changes.

tests/unit/cli/test_archive_maintenance_cli.py

@@ -24,6 +24,7 @@
 )
 from polylogue.storage.sqlite.archive_tiers.archive_plan import ArchiveInitAction, ArchiveInitPlan
 from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier
+from polylogue.storage.sqlite.archive_tiers.source import SOURCE_DDL
 from polylogue.storage.sqlite.archive_tiers.source_write import write_source_raw_session_blob_ref
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
 from polylogue.storage.sqlite.archive_tiers.user import USER_SCHEMA_VERSION
@@ -1107,6 +1108,30 @@ def fake_init(plan: ArchiveInitPlan) -> ArchiveInitResult:
     ]
 
 
+def test_migrate_tier_cli_allows_additive_source_ledgers_without_backup(
+    cli_workspace: dict[str, Path], cli_runner: CliRunner
+) -> None:
+    source_db = cli_workspace["archive_root"] / "source.db"
+    with sqlite3.connect(source_db) as conn:
+        conn.executescript(SOURCE_DDL)
+        conn.execute("DROP TABLE excised_content")
+        conn.execute("DROP TABLE sinex_publication_obligations")
+        conn.execute("PRAGMA user_version = 9")
+        conn.commit()
+
+    result = cli_runner.invoke(
+        cli,
+        ["--plain", "ops", "maintenance", "migrate-tier", "source", "--output-format", "json"],
+        catch_exceptions=False,
+    )
+
+    assert result.exit_code == 0
+    payload = json.loads(result.output)
+    assert payload["backup_manifest"] is None
+    assert payload["backup_receipt"] is None
+    assert payload["applied_versions"] == [10, 11]
+
+
 def test_backup_verify_then_migrate_tier_cli_applies_user_migration_with_receipt(
     cli_workspace: dict[str, Path],
     cli_runner: CliRunner,

tests/unit/storage/test_durable_migrations.py

@@ -547,6 +547,45 @@ def test_source_tier_v1_migrates_to_current_without_native_uniqueness(
         conn.close()
 
 
+def test_source_additive_ledger_migrations_do_not_require_a_backup(tmp_path: Path) -> None:
+    db_path = tmp_path / "source.db"
+    with sqlite3.connect(db_path) as conn:
+        conn.executescript(SOURCE_DDL)
+        conn.execute("DROP TABLE excised_content")
+        conn.execute("DROP TABLE sinex_publication_obligations")
+        conn.execute("PRAGMA user_version = 9")
+        conn.commit()
+
+        result = migrate_archive_tier(conn, ArchiveTier.SOURCE, backup_manifest=None)
+
+        assert result.from_version == 9
+        assert result.to_version == SOURCE_SCHEMA_VERSION == 11
+        assert result.applied_versions == (10, 11)
+        assert result.backup_receipt is None
+        tables = {row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'")}
+        assert {"sinex_publication_obligations", "excised_content"} <= tables
+
+
+def test_additive_no_backup_marker_must_be_the_header_not_a_substring() -> None:
+    """CodeRabbit #2905: substring matching would waive the backup requirement
+    if the marker text ever appeared in a comment, string literal, or later in
+    the file. It must be the file's first non-blank line."""
+    header = migration_runner._ADDITIVE_NO_BACKUP_MARKER
+
+    assert (
+        migration_runner._requires_migration_backup(f"{header}\n-- a real migration\nCREATE TABLE t (x INTEGER);")
+        is False
+    )
+    assert migration_runner._requires_migration_backup(f"{header} trailing text\nCREATE TABLE t (x INTEGER);") is True
+    assert (
+        migration_runner._requires_migration_backup(f"-- unrelated comment\n{header}\nCREATE TABLE t (x INTEGER);")
+        is True
+    )
+    assert migration_runner._requires_migration_backup(f"CREATE TABLE t (x TEXT DEFAULT '{header}');") is True
+    assert migration_runner._requires_migration_backup("CREATE TABLE t (x INTEGER);") is True
+    assert migration_runner._requires_migration_backup(f"\n\n  {header}  \nCREATE TABLE t (x INTEGER);") is False
+
+
 def test_source_tier_v7_expands_origin_checks_with_verified_backup(
     workspace_env: dict[str, Path],
     tmp_path: Path,