Integration batch: duplicates 5-8, root runtime owner, interrupted rollback, installer file identity

src/agent_comms/annotation_disclosure.py

@@ -5,7 +5,7 @@
 import re
 from abc import abstractmethod
 from dataclasses import dataclass
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, ClassVar
 
 from .declared_family import DeclaredFamily
 
@@ -43,15 +43,22 @@ def require_text(self, text: str) -> str:
         return text
 
 
-class WithheldAnnotations:
-    """Source text cannot be disclosed to an annotation provider."""
+class Annotations:
+    """The declared policy owns disclosure; messages and segments share its construction."""
+
+    disclosure_policy: ClassVar[type[DisclosurePolicy]]
 
     def disclosure_for(self, span: ContextSpan) -> DisclosurePolicy:
-        return WithheldDisclosure()
+        return self.disclosure_policy()
 
 
-class PublicInstructionAnnotations:
+class WithheldAnnotations(Annotations):
+    """Source text cannot be disclosed to an annotation provider."""
+
+    disclosure_policy = WithheldDisclosure
+
+
+class PublicInstructionAnnotations(Annotations):
     """Instruction text uses the existing credential-filtering disclosure policy."""
 
-    def disclosure_for(self, span: ContextSpan) -> DisclosurePolicy:
-        return PublicInstructionDisclosure()
+    disclosure_policy = PublicInstructionDisclosure

src/agent_comms/cli_commands.py

@@ -1765,7 +1765,7 @@ def apply(self, ctx: Comms) -> CliResult:
 class ReconciledStoresResult(CliResult):
     """The bus files whose schema the reconcile changed, relative to the bus root."""
 
-    rewritten: tuple[str, ...]
+    rewritten: tuple[Annotated[Path, PathText], ...]
 
 
 @dataclass(frozen=True)
@@ -1810,7 +1810,7 @@ def changes(owner: type[ReconciledFile], path: Path) -> bool:
 
     def answer(self, root: Path) -> ReconciledStoresResult:
         return ReconciledStoresResult(tuple(
-            str(path.relative_to(root))
+            path.relative_to(root)
             for path, owner in sorted(ReconciledFile.existing_in(root).items())
             if self.changes(owner, path)))
 

src/agent_comms/comms.py

@@ -14,6 +14,7 @@
 from .context_store import ContextStore
 from .goal_management import Goals
 from .history_views import HistoryViews
+from .launch_runtime import LaunchRuntime
 from .locked_store import RegistryFile
 from .maintenance_barrier import MaintenanceBarrier
 from .message_bus import MessageBus
@@ -22,6 +23,7 @@
 from .publisher import OwnerPublisher, Publisher
 from .registration import Registration
 from .relationships import ThreadRelationships
+from .root_runtime import RootRuntime
 from .thread_configuration import ThreadConfiguration
 from .thread_management import ThreadManagement
 from .transcripts import Transcripts
@@ -32,6 +34,7 @@ def __init__(
         self, root: CommsRoute | Path, publisher: type[Publisher] = OwnerPublisher,
     ) -> None:
         self.route = resolve_comms_route(root)
+        RootRuntime.require_attachable(self.route.root, LaunchRuntime.running())
         self.route.observe_root()
         self.registry = Registration(RegistryFile.path_in(self.root))
         self.bus = MessageBus(

src/agent_comms/field_codec.py

@@ -1098,6 +1098,11 @@ def from_wire(cls, data):
             raise TypeError("A JSON shape factory must return its own family")
         return result
 
+    @classmethod
+    def wire_structure(cls, structural: ValueDecoder[object]) -> ValueDecoder[object]:
+        """The wire value is one owned member's value; a member owns only its own."""
+        return compile_decoder(Union[tuple(annotation for _, annotation in cls._json_owners.values())])
+
     @classmethod
     def schema(cls) -> dict[str, Any]:
         alternatives = [FieldCodec.value_schema(annotation)

src/agent_comms/installer/backup.py

@@ -16,15 +16,16 @@
 from functools import cache
 from pathlib import Path
 from string import Formatter
-from typing import ClassVar, Iterator, Self
+from typing import Annotated, ClassVar, Iterator, Self
 
 from agent_comms.child_process import Platform
-from agent_comms.active_route import ActiveRoute, _route_directory, route_directory
+from agent_comms.active_route import ActiveRoute, _route_directory, active_route_path, route_directory
 from agent_comms.coordination_errors import CoordinationError
-from agent_comms.field_codec import FieldCodec
+from agent_comms.field_codec import FieldCodec, PathText
 from agent_comms.installer.interpreter_operation import RouteSelection
-from agent_comms.locked_store import LockedStore, RootFile, RuntimeFile, SQLiteFile
+from agent_comms.locked_store import DeclaredFile, LockedStore, RootFile, RuntimeFile, UndeclaredFile
 from agent_comms.private_path import FileObservation, FileRevision, MissingFile
+from agent_comms.registry_store import RegistryStore
 from agent_comms.store_files import AtomicFile
 
 
@@ -55,21 +56,23 @@ def matches_extraction(self, copied: SavedFile) -> bool:
         return self.size == copied.size
 
 
-def bus_files(root: Path) -> dict[str, SavedFile]:
-    """Read saved file identity and modification metadata, never contents."""
+def bus_files(root: Path) -> dict[DeclaredFile | UndeclaredFile, dict[Path, SavedFile]]:
+    """Record owner identities and their content members' physical metadata."""
     excluded = RuntimeFile.paths_in(root)
-    saved = {}
+    observed: dict[Path, os.stat_result] = {}
     for path in root.rglob("*"):
-        if path in excluded or path.name.endswith((".lock", "-shm", "-journal")):
+        if path in excluded:
             continue
-        observed = path.lstat()
-        # Empty WAL files appear on a read-only SQLite open; they hold no committed frames.
-        if path.name.endswith("-wal") and observed.st_size == 0:
-            continue
-        if stat.S_ISREG(observed.st_mode):
-            saved[str(path.relative_to(root))] = SavedFile(
-                observed.st_size, observed.st_mtime_ns, observed.st_ino)
-    return saved
+        info = path.lstat()
+        if stat.S_ISREG(info.st_mode):
+            observed[path] = info
+    return {
+        resource: {
+            member.relative_to(root): SavedFile(
+                observed[member].st_size, observed[member].st_mtime_ns, observed[member].st_ino,
+            ) for member in owner.snapshot_members(root / resource.path, observed)
+        } for resource, owner in RootFile.owned_files(root, observed).items()
+    }
 
 
 @dataclass(frozen=True)
@@ -172,9 +175,27 @@ class StoreBackup(ResourceBackup):
     def spill_directory(self, configured: Path) -> Path:
         return self.path
 
-    def files(self) -> dict[str, str]:
-        return FieldCodec.decode(dict[str, str],
-                                 FieldCodec.parse((self.path / 'files.json').read_text()))
+    def files(self) -> dict[Path, str]:
+        stored = FieldCodec.decode(dict[str, str],
+                                   FieldCodec.parse((self.path / 'files.json').read_text()))
+        return {PathText.decode(path): digest for path, digest in stored.items()}
+
+    def selected_files(self) -> tuple[Annotated[Path, PathText], ...]:
+        return FieldCodec.decode(tuple[Annotated[Path, PathText], ...],
+                                 FieldCodec.parse((self.path / 'selected.json').read_text()))
+
+    @classmethod
+    def interrupted(cls, root: Path, spill_dir: Path) -> tuple[StoreBackup, ...]:
+        if not spill_dir.exists():
+            return ()
+        route = active_route_path()
+        backups = (
+            cls(root, route, spill_dir / name.value)
+            for name in sorted(BackupName.matching(
+                root, (path.name for path in spill_dir.iterdir() if path.is_dir())
+            ), reverse=True)
+        )
+        return tuple(backup for backup in backups if not backup.original())
 
     def retire(self) -> Path:
         archived = self.path.with_name(f'{self.path.name}.restored-{time.time_ns()}')
@@ -226,11 +247,15 @@ def route_original(self) -> bool:
 
     @classmethod
     def take(cls, root: Path, route: Path, files: tuple[Path, ...], spill_dir: Path) -> StoreBackup:
+        registry = RegistryStore.at(root).written
+        if any(path in files for path in registry):
+            files = tuple(dict.fromkeys((*files, *registry)))
         path = spill_dir / BackupName.now(root).value
         backup = cls(root, route, path)
-        selected = tuple(str(file.relative_to(root)) for file in files)
+        selected = tuple(sorted(resource.path for resource in RootFile.owned_files(root, files)))
+        files = tuple(root / path for path in selected)
         if path.exists():
-            if tuple(json.loads((path / 'selected.json').read_text())) != selected:
+            if backup.selected_files() != selected:
                 raise UpgradeRefused('Existing backup belongs to different resources')
             backup.files()  # An interrupted snapshot is never treated as complete.
             with route_directory(route):
@@ -239,13 +264,15 @@ def take(cls, root: Path, route: Path, files: tuple[Path, ...], spill_dir: Path)
                 backup.route_revision = FileRevision.capture(route)
             return backup
         path.mkdir(mode=0o700, parents=True)
-        AtomicFile(path / 'selected.json').write_text(json.dumps(selected))
+        AtomicFile(path / 'selected.json').write_text(json.dumps(FieldCodec.encode(
+            selected, tuple[Annotated[Path, PathText], ...])))
         contents = backup.capture(files)
         with route_directory(route):
             if route.exists():
                 shutil.copy2(route, path / 'active-route.json')
             backup.route_revision = FileRevision.capture(route)
-        AtomicFile(path / 'files.json').write_text(json.dumps(FieldCodec.encode(contents), indent=1))
+        AtomicFile(path / 'files.json').write_text(json.dumps(
+            {PathText.encode(file): digest for file, digest in contents.items()}, indent=1))
         return backup
 
     def retain(self, spill_dir: Path) -> tuple[Path, ...]:
@@ -288,29 +315,28 @@ def copy_tree(
         """Copy committed files; their owners declare transient names before any stat."""
 
         def transient(directory: str, names: list[str]) -> set[str]:
-            omitted: set[str] = set()
-            for name in sorted(names):
-                path = Path(directory) / name
-                if name in omitted or path in excluded or LockedStore.is_staging(path):
-                    omitted.add(name)
+            omitted: set[Path] = set()
+            for path in sorted(Path(directory) / name for name in names):
+                if path in omitted or path in excluded or LockedStore.is_staging(path):
+                    omitted.add(path)
                     continue
                 if stat.S_ISSOCK(path.lstat().st_mode):
-                    omitted.add(name)
+                    omitted.add(path)
                 elif path.is_file() and not path.is_symlink():
                     owner = RootFile.content_owner(path)
-                    omitted.update(companion.name for companion in owner.files(path)[1:])
-            return omitted
+                    omitted.update(set(owner.files(path)) - {path})
+            return {path.name for path in omitted}
 
         shutil.copytree(
             source, target, symlinks=True, ignore=transient, copy_function=cls.copy_file
         )
 
-    def capture(self, files: tuple[Path, ...]) -> dict[str, str]:
-        contents = {}
+    def capture(self, files: tuple[Path, ...]) -> dict[Path, str]:
+        contents: dict[Path, str] = {}
         for source in files:
             if not source.exists():
                 continue
-            name = str(source.relative_to(self.root))
+            name = source.relative_to(self.root)
             target = self.data_root / name
             target.parent.mkdir(parents=True, exist_ok=True)
             self.copy_file(source, target)
@@ -319,7 +345,7 @@ def capture(self, files: tuple[Path, ...]) -> dict[str, str]:
             Platform.current().sync_directory(target.parent)
         return contents
 
-    def file_original(self, name: str, saved: dict[str, str]) -> bool:
+    def file_original(self, name: Path, saved: dict[Path, str]) -> bool:
         """An unchanged selected file keeps its identity as well as its contents."""
         target = self.root / name
         if name not in saved:
@@ -334,14 +360,24 @@ def stores_original(self) -> bool:
             return False
         saved = self.files()
         return all(self.file_original(name, saved)
-                   for name in json.loads((self.path / 'selected.json').read_text()))
+                   for name in self.selected_files())
 
     def restore(self) -> None:
         saved = self.files()
-        for name in json.loads((self.path / 'selected.json').read_text()):
+        selected = self.selected_files()
+        registry = tuple(path.relative_to(self.root) for path in RegistryStore.at(self.root).written)
+        # The registry owns this ordering: restore its snapshot before publishing
+        # the matching guard, after every unrelated selected file.
+        ordered = (
+            *(name for name in selected if name not in registry),
+            *(name for name in registry if name in selected),
+        )
+        for name in ordered:
             if self.file_original(name, saved):
                 continue
             target = self.root / name
+            observed = self.data_root / name if name in saved else target
+            owner = RootFile.content_owner(observed) if observed.exists() else RootFile
             if name in saved:
                 descriptor, temporary = tempfile.mkstemp(prefix='.restore-', dir=target.parent)
                 os.close(descriptor)
@@ -353,7 +389,7 @@ def restore(self) -> None:
                     stage.unlink(missing_ok=True)
             else:
                 target.unlink(missing_ok=True)
-            for companion in SQLiteFile.files(target)[1:]:
+            for companion in set(owner.files(target)) - {target}:
                 companion.unlink(missing_ok=True)
         Platform.current().sync_directory(self.root)
         self.restore_route()

src/agent_comms/installer/carry.py

@@ -15,11 +15,10 @@
 import inspect
 import json
 import subprocess
-from abc import ABC, abstractmethod
+from abc import abstractmethod
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from graphlib import TopologicalSorter
-from importlib import import_module
 from pathlib import Path
 from typing import Annotated, ClassVar
 
@@ -37,40 +36,26 @@ class Carried:
 
     changed: int
     detail: dict[str, int | bool] = field(default_factory=dict)
-    rewritten: tuple[str, ...] = ()
+    rewritten: tuple[Annotated[Path, PathText], ...] = ()
     """Files under the bus root this run wrote or deleted; every other file stays
     byte-identical."""
 
 
 @dataclass(frozen=True)
-class Carry(ABC):
+class Carry(DeclaredFamily, affix="", discovery_package="cutover"):
     """One migration's decoded source, destination and retirement location."""
 
     root: Annotated[Path, PathText]
     original_root: Annotated[Path, PathText]
     spill_dir: Annotated[Path, PathText]
 
-    declared: ClassVar[dict[str, type[Carry]]] = {}
     precedes: ClassVar[tuple[type[Carry], ...]] = ()
     """Steps whose decoders require this step's output, regardless of addition date."""
-    """Every step imported so far, by its module name."""
-
-    def __init_subclass__(cls, **kwargs: object) -> None:
-        super().__init_subclass__(**kwargs)
-        if inspect.isabstract(cls):
-            return
-        if cls.name() in Carry.declared and Carry.declared[cls.name()] is not cls:
-            raise TypeError(f"{cls.name()} declares a second carry step")
-        Carry.declared[cls.name()] = cls
 
     @classmethod
     def module(cls) -> Path:
         return Path(inspect.getfile(cls)).resolve()
 
-    @classmethod
-    def name(cls) -> str:
-        return cls.module().stem
-
     @abstractmethod
     def backup_files(self) -> tuple[Annotated[Path, PathText], ...]:
         """The existing files this step rewrites or deletes, backed up before it applies."""
@@ -132,19 +117,19 @@ class CarryChain:
     def at(cls, tree: Path) -> CarryChain:
         """Import the tree's steps (its ``tools`` is on the path) and order them by the
         first-parent commit that added each file."""
-        directory = (tree / STEPS).resolve()
-        for path in sorted(directory.glob("*.py")):
-            import_module(f"{directory.name}.{path.stem}")
-        steps = [step for step in Carry.declared.values() if step.module().parent == directory]
+        tree = tree.resolve()
+        directory = tree / STEPS
+        steps = [step for step in Carry.members_with(Carry) if step.module().parent == directory]
         log = subprocess.check_output(
             ["git", "-C", str(tree), "log", "-m", "--first-parent", "--no-renames",
              "--diff-filter=A", "--reverse", "--name-only", "--format=", "HEAD", "--",
              str(STEPS)]).decode().split()
-        added = {Path(path).stem: position for position, path in reversed(list(enumerate(log)))}
-        unordered = sorted(step.name() for step in steps if step.name() not in added)
+        added = {Path(path): position for position, path in reversed(list(enumerate(log)))}
+        unordered = sorted(step.declared_name for step in steps
+                           if step.module().relative_to(tree) not in added)
         if unordered:
             raise ValueError(f"{unordered} are not on the main line, so nothing orders them")
-        ordered = sorted(steps, key=lambda step: added[step.name()])
+        ordered = sorted(steps, key=lambda step: added[step.module().relative_to(tree)])
         dependencies: dict[type[Carry], set[type[Carry]]] = {step: set() for step in ordered}
         for step in ordered:
             for later in step.precedes: