#000065 step 4: in-place .db.new → .db atomic promote (α-shape)

Reshard no longer creates a parallel "shards.v2/" directory. Builds
write to "<target_dir>/00X.db.new" alongside originals; when the
executor's validation passes, each .db.new is atomically renamed to
its final 00X.db name via os.replace (POSIX-atomic per file).

For an in-place migration (target_dir == source_dir, which is the
canonical use case), the rename REPLACES the original shard files.
~/.arborist/shards/ never contains a parallel "v2" or "next" or
"backup" directory — it always holds exactly one corpus, just with
the new topology after promote completes.

Validation gate (refuses to promote on mismatch):
  * --expected-row-counts <snapshot.json> threads the pre-migration
    snapshot's documents/chunks/edges totals through to the executor.
  * Tolerance: ±1% to absorb the dupe-collapse from INSERT OR IGNORE
    on cross-shard duplicate document_roots (166 dupes measured
    pre-migration; <0.005% drift).
  * Mismatch → RuntimeError, .db.new files left in place for
    inspection, no rename performed.

CLI:
  arborist corpus reshard --to 4 \
    --source-dir ~/.arborist/shards \
    --target-dir ~/.arborist/shards \
    --audit-events-ndjson /tmp/audit-events.ndjson \
    --expected-row-counts bench/results/pre-migration-snapshot.json

3 new tests:
  * test_target_dir_only_has_db_files_after_completion — no .db.new
    sidecars survive a successful run
  * test_inplace_reshard_overwrites_originals — target_dir ==
    source_dir works end-to-end; final files are the new hash-routed
    shards
  * test_validation_failure_leaves_new_files — bad expected counts
    trip the validation guard; .db.new files survive; no .db files
    promoted

Stale .db.new files from a prior failed run are unlinked before
opening fresh targets, so a partial-fail-and-retry is idempotent.
WAL/SHM sidecars removed at promote time so the new live .db
produces fresh sidecars on next open.

tests/test_migrate_executor.py

@@ -312,6 +312,100 @@ class TestMigrationAllAtOnce:
         assert list(target.glob("*.db")) == []
 
 
+class TestInPlacePromotion:
+    """Verify the .db.new → .db atomic-promote dance (α-shape)."""
+
+    def test_target_dir_only_has_db_files_after_completion(
+        self, source_shards: Path, audit_ndjson: Path, tmp_path: Path
+    ):
+        """No .new sidecars left after a successful run."""
+        target = tmp_path / "tgt"
+        plan = HydrationPlanner().plan(
+            source_dir=source_shards,
+            target_dir=target,
+            target_M=4,
+            free_bytes=200 * 1024 ** 3,
+        )
+        execute_plan(plan, audit_events_ndjson=audit_ndjson)
+        files = sorted(p.name for p in target.glob("*"))
+        # 4 final .db files; -wal/-shm sidecars after close are
+        # permissible but no .db.new files.
+        assert all(not f.endswith(".db.new") for f in files), files
+        assert sum(1 for f in files if f.endswith(".db")) == 4
+
+    def test_inplace_reshard_overwrites_originals(
+        self, source_shards: Path, audit_ndjson: Path
+    ):
+        """Plan with target_dir == source_dir overwrites originals."""
+        # Capture source roots so we can later detect topology change.
+        source_roots: set[str] = set()
+        for shard in source_shards.glob("*.db"):
+            c = sqlite3.connect(f"file:{shard}?mode=ro", uri=True)
+            try:
+                for (root,) in c.execute("SELECT document_root FROM documents"):
+                    source_roots.add(root)
+            finally:
+                c.close()
+
+        plan = HydrationPlanner().plan(
+            source_dir=source_shards,
+            target_dir=source_shards,   # in-place
+            target_M=4,
+            free_bytes=200 * 1024 ** 3,
+        )
+        execute_plan(plan, audit_events_ndjson=audit_ndjson)
+
+        # After in-place: target_dir has exactly M files (no .new, no
+        # parallel dir), each carrying corpus_shard_count = 4, and
+        # each doc lands on its hashed shard.
+        db_files = sorted(p.name for p in source_shards.glob("*.db"))
+        assert db_files == ["000.db", "001.db", "002.db", "003.db"]
+        assert not list(source_shards.glob("*.db.new"))
+
+        for tidx in range(4):
+            tdb = source_shards / f"{tidx:03d}.db"
+            c = sqlite3.connect(f"file:{tdb}?mode=ro", uri=True)
+            try:
+                # corpus_shard_count meta set.
+                row = c.execute(
+                    "SELECT value FROM meta WHERE key='corpus_shard_count'"
+                ).fetchone()
+                assert row and int(row[0]) == 4
+                # All docs route to this shard.
+                for (root,) in c.execute("SELECT document_root FROM documents"):
+                    assert shard_for_document(root, 4) == tidx
+            finally:
+                c.close()
+
+    def test_validation_failure_leaves_new_files(
+        self, source_shards: Path, audit_ndjson: Path, tmp_path: Path
+    ):
+        """When row-count validation fails, .db.new files survive for
+        inspection and no atomic rename runs."""
+        from arborist.migrate import execute_plan as run
+        target = tmp_path / "tgt"
+        plan = HydrationPlanner().plan(
+            source_dir=source_shards,
+            target_dir=target,
+            target_M=4,
+            free_bytes=200 * 1024 ** 3,
+        )
+        # Pass a deliberately wrong expected row count to trip the
+        # >1% tolerance guard.
+        with pytest.raises(RuntimeError, match="validation failed"):
+            run(
+                plan,
+                audit_events_ndjson=audit_ndjson,
+                expected_row_counts={"documents": 9_999_999},
+            )
+        # Inspection artefacts must remain.
+        assert sorted(p.name for p in target.glob("*.db.new")) == [
+            "000.db.new", "001.db.new", "002.db.new", "003.db.new",
+        ]
+        # No promoted .db files yet.
+        assert not list(target.glob("00[0-9].db"))
+
+
 class TestExecutorErrorCases:
     def test_missing_ndjson_raises(
         self, source_shards: Path, tmp_path: Path