Use declared disk scratch for writer-store rehearsal

src/agent_comms/cli_verbs/planned_reconcile.py

@@ -3,6 +3,8 @@
 from agent_comms.cli_commands import PlannedStoresResult
 from agent_comms.cli_support.store_reconcile import StoreReconcileCliCommand
 from agent_comms.field_codec import Argument
+from agent_comms.storage_rules import StorageRule
+from agent_comms.locked_store import LiveFilesRequest
 from agent_comms.typed_table import ReconciledFile
 from agent_comms.typed_table import StoreRewrite
 from agent_comms.typed_table import ExternalStore
@@ -23,7 +25,9 @@ class PlannedReconcileCliCommand(
 
     def answer(self, root: Path) -> PlannedStoresResult:
         """A store outside the bus that the target's writer could not open refuses the plan."""
-        ExternalStore.practice_openings(root, excluding=self.carrying)
+        stores = tuple(set(ExternalStore.members_with(ExternalStore)) - set(self.carrying))
+        ExternalStore.practice_openings(
+            LiveFilesRequest(root), scratch=StorageRule.load(root).scratch_base, stores=stores)
         rewrites: list[StoreRewrite] = []
         for path, owner in sorted(ReconciledFile.existing_in(root).items()):
             schema = owner.schema_at(path)

src/agent_comms/installer/probes/carry.py

@@ -9,7 +9,6 @@
 from contextlib import ExitStack
 from dataclasses import dataclass, field, replace
 from pathlib import Path
-from tempfile import TemporaryDirectory
 from typing import Annotated, ClassVar
 
 from typing_extensions import TypeForm
@@ -209,15 +208,12 @@ class WriterOwnedOpeningProbe(TargetProbe, RuntimeProbe[tuple[Annotated[Path, Pa
 
     files: FilesRequest
     stores: tuple[type[ExternalStore], ...]
+    scratch: Annotated[Path, PathText]
 
     def run(self) -> tuple[Annotated[Path, PathText], ...]:
-        paths: list[Path] = []
-        with TemporaryDirectory() as directory:
-            for store in self.stores:
-                store.practice_opening(self.files, Path(directory))
-                paths.extend(self.files.location(resource)
-                             for resource in store.resources(self.files))
-        return tuple(paths)
+        ExternalStore.practice_openings(self.files, scratch=self.scratch, stores=self.stores)
+        return tuple(self.files.location(resource)
+                     for store in self.stores for resource in store.resources(self.files))
 
 
 @dataclass(frozen=True)

src/agent_comms/installer/upgrade.py

@@ -1263,7 +1263,9 @@ def carry(
         """One migration sequence for rehearsal and the backed-up live bus."""
         steps = [StepRun(step.declared_name, carried) for step, carried in zip(
             pending, staged.call_steps(pending, MigrateOperation(apply=True)), strict=True)]
-        staged.probe(WriterOwnedOpeningProbe(self.files, self.writer_stores))
+        staged.probe(WriterOwnedOpeningProbe(
+            self.files, self.writer_stores,
+            scratch=StorageRule.load(self.files.source_root()).scratch_base))
         reconciled = staged.cli(self.root, ReconcileStoresCliCommand, *reconcile_arguments)
         steps.append(StepRun(ReconcileStoresCliCommand.declared_name, Carried(
             len(reconciled.rewritten), rewritten=reconciled.rewritten)))

src/agent_comms/typed_table.py

@@ -70,7 +70,6 @@
     DeclaredFile,
     DurableFile,
     FilesRequest,
-    LiveFilesRequest,
     PublishedSQLiteFile,
     RegistryFile,
     RootFile,
@@ -83,6 +82,7 @@
 from .maintenance_barrier import HolderWrittenFile, MaintenanceBarrier, MaintenanceHolder
 from .package_stamp import STAMP
 from .projection_source import ProducedRow, ProjectionOutputs, ProjectionSource
+from .private_path import PrivateDirectoryRole
 from .store_files import (
     BorrowedStoreLock,
     ExclusiveAtomicFile,
@@ -2827,11 +2827,13 @@ def practice_opening(cls, files: FilesRequest, directory: Path) -> None:
         """A file with no stored schema has no opening to practise."""
 
     @staticmethod
-    def practice_openings(root: Path, excluding: tuple[type[ExternalStore], ...] = ()) -> None:
-        stores = set(ExternalStore.members_with(ExternalStore)) - set(excluding)
-        with TemporaryDirectory() as directory:
+    def practice_openings(
+        files: FilesRequest, *, scratch: Path, stores: tuple[type[ExternalStore], ...],
+    ) -> None:
+        PrivateDirectoryRole.make_parents(scratch)
+        with TemporaryDirectory(dir=scratch) as directory:
             for store in stores:
-                store.practice_opening(LiveFilesRequest(root), Path(directory))
+                store.practice_opening(files, Path(directory))
 
     @classmethod
     @abstractmethod

tests/test_reconcile_stores.py

@@ -376,16 +376,19 @@ def test_planned_reconcile_refuses_a_bridge_delivery_store_its_declaration_canno
     monkeypatch.setenv("HOME", str(tmp_path / "home"))
     stored = tmp_path / "home" / CodexBridge.state_directory / "unit-old" / "delivery.sqlite3"
     stored.parent.mkdir(parents=True)
-    with closing(sqlite3.connect(stored)) as db:
-        db.execute('CREATE TABLE observed_part (id TEXT NOT NULL, number TEXT NOT NULL, '
-                   'sender TEXT NOT NULL, body TEXT NOT NULL, record_id TEXT NOT NULL, '
-                   'timestamp REAL NOT NULL, PRIMARY KEY (id, number))')
-        db.execute("INSERT INTO observed_part VALUES ('m', 'first', 'a', 'b', 'r', 0)")
-        db.commit()
     if readable:
-        stored.unlink()
         with closing(SQLiteFile.writer(stored)) as db, db:
             BridgeStateTable.install_schema(db, WriterOwnedFile())
+            db.execute('CREATE TABLE observed_part (id TEXT NOT NULL, number TEXT NOT NULL, '
+                       'sender TEXT NOT NULL, body TEXT NOT NULL, record_id TEXT NOT NULL, '
+                       'timestamp REAL NOT NULL, PRIMARY KEY (id, number))')
+            db.execute("INSERT INTO observed_part VALUES ('m', 'first', 'a', 'b', 'r', 0)")
+    else:
+        with closing(sqlite3.connect(stored)) as db:
+            db.execute('CREATE TABLE board_delivery (content TEXT NOT NULL, since TEXT, '
+                       'delivered_at REAL NOT NULL, PRIMARY KEY (content))')
+            db.execute("INSERT INTO board_delivery VALUES ('line', 'not-a-time', 0)")
+            db.commit()
     sibling = stored.parent.with_name("unit-second") / BridgeStateTable.filename
     sibling.parent.mkdir()
     with closing(SQLiteFile.writer(sibling)) as db, db:
@@ -400,7 +403,7 @@ def test_planned_reconcile_refuses_a_bridge_delivery_store_its_declaration_canno
     practice = PracticeCopy(root, tmp_path / "practice")
     practice.include(tuple(originals))
     copied = practice.request
-    probe = WriterOwnedOpeningProbe(copied, (BridgeDeliveryStore,))
+    probe = WriterOwnedOpeningProbe(copied, (BridgeDeliveryStore,), scratch=tmp_path / "scratch")
     copied_paths = tuple(copied.location(resource) for resource in resources)
     planned = subprocess.run(
         [sys.executable, "-m", "agent_comms.cli", "--root", str(root),

tests/test_writer_owned_stores.py

@@ -175,7 +175,8 @@ def test_writer_owned_carry_copies_migrates_and_restores_original_bytes(
     with pytest.raises(RelationViolationError):
         copied_step.require_confined(store.path())
     assert copied_step.carry(apply=True).rewritten == (name,)
-    assert WriterOwnedOpeningProbe(copy.request, (store,)).run() == copied_paths
+    assert WriterOwnedOpeningProbe(
+        copy.request, (store,), scratch=tmp_path / "scratch").run() == copied_paths
     for copied_path in copied_paths:
         with closing(store.read_snapshot(copied_path)) as db:
             assert row.select(db)[0].target == ChannelTarget("#comms")
@@ -219,12 +220,14 @@ def test_writer_owned_carry_preserves_its_declaration_across_cli_and_worker_boun
         "planned-reconcile", "--carrying",
     ]))
     assert empty.carrying == ()
-    probe = WriterOwnedOpeningProbe(LiveFilesRequest(tmp_path), (store,))
+    probe = WriterOwnedOpeningProbe(
+        LiveFilesRequest(tmp_path), (store,), scratch=tmp_path / "scratch")
     decoded = FieldCodec.decode(WriterOwnedOpeningProbe, FieldCodec.encode(probe))
     assert decoded == probe
     assert decoded.run() == (store.path(),)
     copied_root = tmp_path / "copied"
-    copied_probe = WriterOwnedOpeningProbe(RetainedCopyRequest(copied_root, tmp_path), (store,))
+    copied_probe = WriterOwnedOpeningProbe(
+        RetainedCopyRequest(copied_root, tmp_path), (store,), scratch=tmp_path / "scratch")
     copied_decoded = FieldCodec.decode(WriterOwnedOpeningProbe, FieldCodec.encode(copied_probe))
     assert copied_decoded.stores == (store,)
     assert copied_decoded.files.root == copied_root