From b4bbf036924a82d03f2cd8dbdb3f666106871cda Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 29 Oct 2026 04:49:44 -0400 Subject: [PATCH 1/4] Rehearse external writer stores under the selected disk scratch root --- src/agent_comms/cli_verbs/planned_reconcile.py | 4 +++- src/agent_comms/typed_table.py | 8 ++++++-- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/src/agent_comms/cli_verbs/planned_reconcile.py b/src/agent_comms/cli_verbs/planned_reconcile.py index d3d65ba021..793bf2cdde 100644 --- a/src/agent_comms/cli_verbs/planned_reconcile.py +++ b/src/agent_comms/cli_verbs/planned_reconcile.py @@ -3,6 +3,7 @@ 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.typed_table import ReconciledFile from agent_comms.typed_table import StoreRewrite from agent_comms.typed_table import ExternalStore @@ -23,7 +24,8 @@ 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) + ExternalStore.practice_openings( + root, scratch=StorageRule.load(root).scratch_base, excluding=self.carrying) rewrites: list[StoreRewrite] = [] for path, owner in sorted(ReconciledFile.existing_in(root).items()): schema = owner.schema_at(path) diff --git a/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py index bf78b95c9d..771fed3069 100644 --- a/src/agent_comms/typed_table.py +++ b/src/agent_comms/typed_table.py @@ -83,6 +83,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,9 +2828,12 @@ 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: + def practice_openings( + root: Path, *, scratch: Path, excluding: tuple[type[ExternalStore], ...] = (), + ) -> None: stores = set(ExternalStore.members_with(ExternalStore)) - set(excluding) - with TemporaryDirectory() as directory: + PrivateDirectoryRole.make_parents(scratch) + with TemporaryDirectory(dir=scratch) as directory: for store in stores: store.practice_opening(LiveFilesRequest(root), Path(directory)) From 5ee6d8468201f68be74ecda71f4d8788876ad420 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 29 Oct 2026 05:09:11 -0400 Subject: [PATCH 2/4] Share writer-store rehearsal across reconcile and carry --- src/agent_comms/cli_verbs/planned_reconcile.py | 4 +++- src/agent_comms/installer/probes/carry.py | 12 ++++-------- src/agent_comms/installer/upgrade.py | 4 +++- src/agent_comms/typed_table.py | 5 ++--- tests/test_reconcile_stores.py | 2 +- tests/test_writer_owned_stores.py | 9 ++++++--- 6 files changed, 19 insertions(+), 17 deletions(-) diff --git a/src/agent_comms/cli_verbs/planned_reconcile.py b/src/agent_comms/cli_verbs/planned_reconcile.py index 793bf2cdde..2c14f820a5 100644 --- a/src/agent_comms/cli_verbs/planned_reconcile.py +++ b/src/agent_comms/cli_verbs/planned_reconcile.py @@ -4,6 +4,7 @@ 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 @@ -24,8 +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.""" + stores = tuple(set(ExternalStore.members_with(ExternalStore)) - set(self.carrying)) ExternalStore.practice_openings( - root, scratch=StorageRule.load(root).scratch_base, excluding=self.carrying) + 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) diff --git a/src/agent_comms/installer/probes/carry.py b/src/agent_comms/installer/probes/carry.py index 3f746dfb4d..a4cbf81fc3 100644 --- a/src/agent_comms/installer/probes/carry.py +++ b/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) diff --git a/src/agent_comms/installer/upgrade.py b/src/agent_comms/installer/upgrade.py index 2fb20ed9dc..975f6d3968 100644 --- a/src/agent_comms/installer/upgrade.py +++ b/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))) diff --git a/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py index 771fed3069..566731be3e 100644 --- a/src/agent_comms/typed_table.py +++ b/src/agent_comms/typed_table.py @@ -2829,13 +2829,12 @@ def practice_opening(cls, files: FilesRequest, directory: Path) -> None: @staticmethod def practice_openings( - root: Path, *, scratch: Path, excluding: tuple[type[ExternalStore], ...] = (), + files: FilesRequest, *, scratch: Path, stores: tuple[type[ExternalStore], ...], ) -> None: - stores = set(ExternalStore.members_with(ExternalStore)) - set(excluding) 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 diff --git a/tests/test_reconcile_stores.py b/tests/test_reconcile_stores.py index f00650963d..74b763baec 100644 --- a/tests/test_reconcile_stores.py +++ b/tests/test_reconcile_stores.py @@ -400,7 +400,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), diff --git a/tests/test_writer_owned_stores.py b/tests/test_writer_owned_stores.py index 8015db4ae2..3731718c26 100644 --- a/tests/test_writer_owned_stores.py +++ b/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 From 86b415d67e0bba0a170d107ce9f0372ebb8d956e Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 29 Oct 2026 05:18:27 -0400 Subject: [PATCH 3/4] Remove the obsolete live-files import --- src/agent_comms/typed_table.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py index 566731be3e..db1a6f6bac 100644 --- a/src/agent_comms/typed_table.py +++ b/src/agent_comms/typed_table.py @@ -70,7 +70,6 @@ DeclaredFile, DurableFile, FilesRequest, - LiveFilesRequest, PublishedSQLiteFile, RegistryFile, RootFile, From 4bd9bce04182c19cab6102ce6c23d5bee86128f2 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 29 Oct 2026 05:23:56 -0400 Subject: [PATCH 4/4] Exercise current and retired bridge tables in rehearsal controls --- tests/test_reconcile_stores.py | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/tests/test_reconcile_stores.py b/tests/test_reconcile_stores.py index 74b763baec..67480f52c0 100644 --- a/tests/test_reconcile_stores.py +++ b/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: