Lock a copied file only when its writers lock it

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:

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

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,
 )

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]