diff --git a/src/agent_comms/historical_views.py b/src/agent_comms/historical_views.py index cf07533b34..17c0e7fde2 100644 --- a/src/agent_comms/historical_views.py +++ b/src/agent_comms/historical_views.py @@ -18,7 +18,7 @@ from .coordination_errors import HistoricalSourceChanged from .errors import UnregisteredThreadError from .field_codec import FieldCodec -from .locked_store import DurableFile, RegistryFile, RootFile +from .locked_store import DurableFile, LockedFile, RegistryFile, RootFile from .message_page import ( NEWEST_PAGE, ConversationMessagePageRequest, @@ -47,7 +47,7 @@ from .registration import Registration from .registry_provenance import RegistryProvenance from .revision_views import BusSourceView -from .store_files import AtomicFile, _store_lock +from .store_files import AtomicFile from .thread_identity import UnrecordedThreadIncarnation from .threads import ThreadProvenance @@ -398,7 +398,7 @@ class HistoricalThread: thread: ThreadProvenance -class HistoryArchive(DurableFile): +class HistoryArchive(DurableFile, LockedFile): """Ordered immutable source snapshots; never a delivery/launch authority.""" filename = "history_sources.json" @@ -469,7 +469,7 @@ def attach(self, source_root: Path, *, source_read: HistorySource | None = None) source_root = source_root.resolve() if source_root == self.bus_path.parent.resolve(): raise ValueError("The live bus cannot be its own history source") - with _store_lock(self.path): + with self.locked(): sources = self.sources() existing = next((s for s in sources if s.original_root == str(source_root)), None) if existing is not None: diff --git a/src/agent_comms/locked_store.py b/src/agent_comms/locked_store.py index d6076ac4af..77f8c65660 100644 --- a/src/agent_comms/locked_store.py +++ b/src/agent_comms/locked_store.py @@ -116,9 +116,9 @@ def copy_file(source: str | Path, target: str | Path) -> Path: @classmethod def copy_group(cls, source: Path, target: Path) -> frozenset[Path]: - """Copy this owner's consistent file cut and name the source files included.""" - with _store_lock(source, mode=SharedLock()): - cls.copy_file(source, target) + """Copy this owner's file cut and name the source files included. A file no writer + locks is copied as it is: locking it would create a lock inode beside it.""" + cls.copy_file(source, target) return frozenset((source,)) @staticmethod @@ -319,6 +319,17 @@ class UndeclaredFile(RootFile): path: Path +class LockedFile(RootFile, StoreLockOwner, declared_name=None): + """A declared file whose writers hold its store lock, so its copy shares that lock.""" + + __slots__ = () + + @classmethod + def copy_group(cls, source: Path, target: Path) -> frozenset[Path]: + with _store_lock(source, mode=SharedLock()): + return super().copy_group(source, target) + + class FileStore(RootFile, ABC): """Construction belongs only to owners whose initializer takes a file.""" @@ -672,7 +683,7 @@ def publish_change(cls, path: Path) -> None: @dataclass(frozen=True, slots=True) -class LockedStore(FileStore, StoreLockOwner, JsonDocument[T], declared_name=None): +class LockedStore(FileStore, LockedFile, JsonDocument[T], declared_name=None): """Ordered document transactions; callbacks never mutate or reenter.""" path: Path diff --git a/src/agent_comms/store_order.py b/src/agent_comms/store_order.py index 02c5e4d500..b1e3ba8918 100644 --- a/src/agent_comms/store_order.py +++ b/src/agent_comms/store_order.py @@ -1,4 +1,5 @@ """One complete acquisition order, loaded before store instances open.""" +from .historical_views import HistoryArchive from .registry_store import RegistryStore from .catalog_store import ChannelCatalog from .read_ledger import ReadLedger @@ -13,7 +14,7 @@ from .store_files import StoreLockOwner STORE_LOCK_ORDER = ( - RegistryStore, ChannelCatalog, ReadLedger, + HistoryArchive, RegistryStore, ChannelCatalog, ReadLedger, RelationshipStore, RuntimeInfoStore, SharedLedger, CollaborationLedger, ConnectorStore, RuntimeChildrenStore, RootRuntimeStore, ServicePassStore, ) diff --git a/tests/test_thread_catalog_files.py b/tests/test_thread_catalog_files.py index 2630f12594..34c1b71403 100644 --- a/tests/test_thread_catalog_files.py +++ b/tests/test_thread_catalog_files.py @@ -2,7 +2,8 @@ import pytest -from agent_comms.locked_store import RootFile +from agent_comms.locked_store import LockedFile, RootFile +from agent_comms.read_ledger import ReadLedger from agent_comms.store_files import store_lock_path from agent_comms.threads import ThreadCatalog from cutover.carry_lock_chains import LockChains @@ -20,11 +21,32 @@ def test_session_locks_are_undeclared_and_never_locked(tmp_path): lock = store_lock_path(catalog) lock.touch() assert ThreadCatalog.matching_in(tmp_path) == frozenset((catalog,)) + RootFile.copy_group(lock, tmp_path / "copy") with pytest.raises(ValueError, match="never itself a lockable store"): - RootFile.copy_group(lock, tmp_path / "copy") + LockedFile.copy_group(lock, tmp_path / "locked-copy") assert not store_lock_path(lock).exists() +def test_copy_locks_only_files_their_writers_lock(tmp_path): + """A copy of a root creates a store lock only beside a store whose writers take it.""" + root, copy = tmp_path / "root", tmp_path / "copy" + ledger = root / "imported_sessions" / "selected-tool-ledger" + ledger.mkdir(parents=True) + plain = tuple(ledger / f"slot-{index}" for index in range(3)) + for path in plain: + path.write_text("{}") + store = ReadLedger.path_in(root) + store.write_text("{}") + for resource, owner in RootFile.owned_files(root, (*plain, store)).items(): + target = copy / resource.path + target.parent.mkdir(parents=True, exist_ok=True) + owner.copy_group(root / resource.path, target) + assert not any(ledger.glob(".*.lock")) + assert store_lock_path(store).exists() + assert sorted(path.name for path in (copy / ledger.relative_to(root)).iterdir()) == [ + path.name for path in plain] + + def test_lock_chains_are_deleted_and_store_locks_kept(tmp_path): lock = store_lock_path(_catalog(tmp_path)) chain = [lock]