Derive installer file behavior from declared owners

src/agent_comms/coordination_database.py

@@ -7,12 +7,12 @@
 import stat
 import tempfile
 from abc import abstractmethod
-from contextlib import AbstractContextManager, contextmanager, suppress
+from contextlib import AbstractContextManager, closing, contextmanager, suppress
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Self
 
-from agent_comms.coordination_errors import IntegrityViolationError
+from agent_comms.coordination_errors import CoordinationReadUnavailable, IntegrityViolationError
 from agent_comms.coordination_schema import CoordinatorTable
 from agent_comms.typed_table import (
     CarriedWriter,
@@ -21,9 +21,9 @@
     TableSchema,
 )
 
-from .declared_family import DeclaredFamily
 from .child_process import Platform
-from .locked_store import DEFAULT_STORE_CONTENTION, SQLiteFile, StoreContention
+from .declared_family import DeclaredFamily
+from .locked_store import DEFAULT_STORE_CONTENTION, StoreBusy, StoreContention
 
 
 class StoreFile(DeclaredFamily):
@@ -80,8 +80,8 @@ def schema_at(cls, path: Path) -> type[TableSchema]:
 
         return RegistryPrivacy.capture(path.parent).coordinator_schema()
 
-    @staticmethod
-    def observed(path: Path) -> StoreFile:
+    @classmethod
+    def observed(cls, path: Path) -> StoreFile:
         """Classify through SQLite without dropping another connection's process locks.
 
         Both rollback and WAL snapshots are readable. SQLite may maintain its own
@@ -92,19 +92,24 @@ def observed(path: Path) -> StoreFile:
             info = path.lstat()
             if not stat.S_ISREG(info.st_mode):
                 return InvalidStore("Coordinator is not a regular file")
-            database = SQLiteFile.is_database(path)
+            if not info.st_size:
+                return InvalidStore("Coordinator is empty")
+            with (
+                CoordinationReadUnavailable.checking(),
+                closing(cls.read_snapshot(path, StoreBusy())),
+            ):
+                pass
         except FileNotFoundError:
             return MissingStore()
         except PermissionError as error:
             return InvalidStore(str(error))
         except sqlite3.DatabaseError as error:
             return InvalidStore(str(error))
-        if not database:
-            return InvalidStore("Coordinator is not a SQLite database")
         return ReadableStore(path)
 
     def __init__(
-        self, path: str | os.PathLike[str],
+        self,
+        path: str | os.PathLike[str],
         contention: StoreContention = DEFAULT_STORE_CONTENTION,
     ) -> None:
         super().__init__(Path(path))

src/agent_comms/diagnostics.py

@@ -29,22 +29,27 @@
     from .threads import Thread
 
 
-@contextmanager
-def owner_process_output(root: Path, thread: Thread):
-    """Retain a detached owner's startup traceback in the existing diagnostics.
+class OwnerProcessOutput(DurableFile, PatternFile):
+    """Startup output retained by the owner launcher."""
 
-    The child inherits this descriptor; closing the parent's copy does not end
-    capture. Each launch gets its own private file, without changing thread or
-    input disposition state.
-    """
-    directory = root / "diagnostics"
-    directory.mkdir(mode=0o700, exist_ok=True)
-    path = directory / f"owner-{thread.incarnation.lookup}-{uuid4().hex}.log"
-    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
-    with os.fdopen(descriptor, "w") as output:
-        output.write(f"Owner launch: {thread.name}\n")
-        output.flush()
-        yield output
+    filename = "diagnostics/owner-{lookup}-{id}.log"
+
+    @classmethod
+    @contextmanager
+    def open(cls, root: Path, thread: Thread):
+        """Retain a detached owner's startup traceback in the existing diagnostics.
+
+        The child inherits this descriptor; closing the parent's copy does not end
+        capture. Each launch gets its own private file, without changing thread or
+        input disposition state.
+        """
+        path = cls.path_in(root, lookup=thread.incarnation.lookup, id=uuid4().hex)
+        path.parent.mkdir(mode=0o700, exist_ok=True)
+        descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        with os.fdopen(descriptor, "w") as output:
+            output.write(f"Owner launch: {thread.name}\n")
+            output.flush()
+            yield output
 
 
 @dataclass

src/agent_comms/installer/backup.py

@@ -265,8 +265,9 @@ def take(cls, root: Path, route: Path, files: tuple[Path, ...], spill_dir: Path)
             files = tuple(dict.fromkeys((*files, *registry)))
         path = spill_dir / BackupName.now(root).value
         backup = cls(root, route, path)
-        selected = tuple(sorted(resource.path for resource in RootFile.owned_files(root, files)))
-        files = tuple(root / path for path in selected)
+        owners = {root / resource.path: owner
+                  for resource, owner in RootFile.owned_files(root, files).items()}
+        selected = tuple(sorted(path.relative_to(root) for path in owners))
         if path.exists():
             if backup.selected_files() != selected:
                 raise UpgradeRefused('Existing backup belongs to different resources')
@@ -279,7 +280,7 @@ def take(cls, root: Path, route: Path, files: tuple[Path, ...], spill_dir: Path)
         path.mkdir(mode=0o700, parents=True)
         AtomicFile(path / 'selected.json').write_text(json.dumps(FieldCodec.encode(
             selected, tuple[Annotated[Path, PathText], ...])))
-        contents = backup.capture(files)
+        contents = backup.capture(owners)
         with route_directory(route):
             if route.exists():
                 shutil.copy2(route, path / RouteFile.filename)
@@ -309,61 +310,59 @@ def description(self) -> str:
     def data_root(self) -> Path:
         return self.path / 'stores'
 
-    @staticmethod
-    def digest(path: Path) -> str:
-        return RootFile.content_owner(path).digest(path)
-
-    @staticmethod
-    def copy_file(source: str | Path, target: str | Path) -> Path:
-        """Copy one store through its SQLite snapshot, never through live WAL bytes."""
-        source, target = Path(source), Path(target)
-        owner = RootFile.content_owner(source)
-        owner.copy_file(source, target)
-        return target
-
     @classmethod
     def copy_tree(
         cls, source: Path, target: Path, *, excluded: Collection[Path] = ()
     ) -> None:
         """Copy committed files; their owners declare transient names before any stat."""
 
+        paths = tuple(path for path in source.rglob("*")
+                      if path.is_file() and not path.is_symlink())
+        owners = {source / resource.path: owner
+                  for resource, owner in RootFile.owned_files(source, paths).items()}
+        omitted = set(paths) - owners.keys() | set(excluded)
+        omitted.update(member for path, owner in owners.items()
+                       for member in owner.files(path) if member != path)
+
         def transient(directory: str, names: list[str]) -> set[str]:
-            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(path)
-                elif path.is_file() and not path.is_symlink():
-                    owner = RootFile.content_owner(path)
-                    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
-        )
+            return {path.name for path in (Path(directory) / name for name in names)
+                    if path in omitted or LockedStore.is_staging(path)
+                    or stat.S_ISSOCK(path.lstat().st_mode)}
 
-    def capture(self, files: tuple[Path, ...]) -> dict[Path, str]:
+        def copy_file(original: str, destination: str) -> Path:
+            return owners[Path(original)].copy_file(original, destination)
+
+        shutil.copytree(source, target, symlinks=True, ignore=transient, copy_function=copy_file)
+
+    def capture(self, owners: dict[Path, type[RootFile]]) -> dict[Path, str]:
         contents: dict[Path, str] = {}
-        for source in files:
+        for source, owner in owners.items():
             if not source.exists():
                 continue
             name = source.relative_to(self.root)
             target = self.data_root / name
             target.parent.mkdir(parents=True, exist_ok=True)
-            self.copy_file(source, target)
+            owner.copy_file(source, target)
             Platform.current().sync_file(target)
-            contents[name] = self.digest(target)
+            contents[name] = owner.digest(target)
             Platform.current().sync_directory(target.parent)
         return contents
 
-    def file_original(self, name: Path, saved: dict[Path, str]) -> bool:
+    def resource_owners(self, selected: tuple[Path, ...], saved: Iterable[Path]
+                        ) -> dict[Path, type[RootFile]]:
+        """Resolve the live declarations and the retained originals once per operation."""
+        owners = {resource.path: owner for resource, owner in RootFile.owned_files(
+            self.root, tuple(self.root / name for name in selected)).items()}
+        owners.update((resource.path, owner) for resource, owner in RootFile.owned_files(
+            self.data_root, tuple(self.data_root / name for name in saved)).items())
+        return owners
+
+    def file_original(self, name: Path, saved: dict[Path, str], owner: type[RootFile]) -> bool:
         """An unchanged selected file keeps its identity as well as its contents."""
         target = self.root / name
         if name not in saved:
             return not target.exists()
-        return target.exists() and self.digest(target) == saved[name]
+        return target.exists() and owner.digest(target) == saved[name]
 
     def original(self) -> bool:
         return self.route_original() and self.stores_original()
@@ -372,12 +371,14 @@ def stores_original(self) -> bool:
         if not (self.path / 'files.json').exists():
             return False
         saved = self.files()
-        return all(self.file_original(name, saved)
-                   for name in self.selected_files())
+        selected = self.selected_files()
+        owners = self.resource_owners(selected, saved)
+        return all(self.file_original(name, saved, owners[name]) for name in selected)
 
     def restore(self) -> None:
         saved = self.files()
         selected = self.selected_files()
+        owners = self.resource_owners(selected, saved)
         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.
@@ -386,11 +387,10 @@ def restore(self) -> None:
             *(name for name in registry if name in selected),
         )
         for name in ordered:
-            if self.file_original(name, saved):
+            if self.file_original(name, saved, owners[name]):
                 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
+            owner = owners[name]
             if name in saved:
                 descriptor, temporary = tempfile.mkstemp(prefix='.restore-', dir=target.parent)
                 os.close(descriptor)

src/agent_comms/locked_store.py

@@ -101,10 +101,6 @@ def matching_in(cls, root: Path) -> frozenset[Path]:
         path = cls.path_in(root)
         return frozenset((path,)) if path.is_file() else frozenset()
 
-    @staticmethod
-    def content_owner(path: Path) -> type[RootFile]:
-        """The file behavior for an arbitrary resource, including undeclared files."""
-        return SQLiteFile if SQLiteFile.is_database(path) else RootFile
 
     @staticmethod
     def copy_file(source: str | Path, target: str | Path) -> Path:
@@ -169,8 +165,8 @@ def owned_files(
         store declaration for it.
         """
         primaries = cls.existing_in(root)
-        declarations = {member: owner for path, owner in primaries.items()
-                        for member in owner.member_paths(path)}
+        declarations = {member: owner for owner, directory in cls.declarations_in(root)
+                        for member in owner.files_in(directory)}
         declarations.update(primaries)
         selected = frozenset(paths)
         owners: dict[Path, type[RootFile]] = {
@@ -179,7 +175,7 @@ def owned_files(
         companions = {member for path, owner in owners.items()
                       for member in (*owner.files(path), *owner.transient_paths(path))
                       if member != path}
-        owners.update((path, cls.content_owner(path) if path.exists() else RootFile)
+        owners.update((path, RootFile)
                       for path in selected - owners.keys() - companions)
         companions.update(member for path, owner in owners.items()
                           for member in (*owner.files(path), *owner.transient_paths(path))
@@ -528,20 +524,6 @@ def observing(
                 )
                 raise
 
-    @classmethod
-    def is_database(cls, path: Path) -> bool:
-        if path.stat().st_size == 0:
-            return False
-        # Even a header-only plain open/read/close can cancel locks owned by
-        # another SQLite connection in this process. Let SQLite own the fd.
-        with CoordinationReadUnavailable.checking():
-            try:
-                with closing(cls.read_snapshot(path, StoreBusy())):
-                    return True
-            except sqlite3.DatabaseError as error:
-                if error.sqlite_errorcode & 0xFF == sqlite3.SQLITE_NOTADB:
-                    return False
-                raise
 
     @classmethod
     def copy_file(cls, source: str | Path, target: str | Path) -> Path:

src/agent_comms/native_pi.py

@@ -37,7 +37,7 @@
 from .native_failures import NativePiUnavailable, NativeRequestFailure
 from .native_input_record import NativeInputCommit, NativeInputLocation
 from .native_package import MANIFEST, UNIT_INSTRUCTIONS, VerifiedNativePackage
-from .native_session_files import NativeSessionFiles
+from .native_session_files import NativeAgentSettings, NativeSessionFiles
 from .native_tool_call import NativeToolMode
 from .pi_vocabulary import NoThinkingOverride, PendingStopReason, ThinkingChoice
 from .private_path import (
@@ -591,8 +591,7 @@ def _tracked(
             raise NativePiUnavailable("Native Pi worktree is unavailable")
         # Native SettingsManager and AgentSession own tracked retry/compaction
         # policy. This directory isolates writable resources, not policy copies.
-        agent_dir = session_dir / ".native-pi-agent"
-        NativeSessionFiles.prepare_private_directory(agent_dir)
+        agent_dir = NativeAgentSettings.prepare_directory(session_dir)
         tool_arguments = selected_tool_mode.launch_arguments(package.path)
         arguments = [
             "--mode",

src/agent_comms/native_session_files.py

@@ -222,6 +222,19 @@ class SavedSessionFile(PatternFile):
 
     __slots__ = ()
 
+    @classmethod
+    def member_paths(cls, path: Path) -> frozenset[Path]:
+        return super().member_paths(path) | frozenset(NativeSessionLocks(path).members)
+
+    @classmethod
+    def transient_paths(cls, path: Path) -> frozenset[Path]:
+        return super().transient_paths(path) | frozenset((NativeSessionLocks(path).writer_lock,))
+
+    @classmethod
+    def files_in(cls, root: Path) -> frozenset[Path]:
+        return super().files_in(root) | frozenset(
+            root.glob(str(NativeSessionLocks(cls.pattern()).writer_lock)))
+
     @classmethod
     def sessions_in(cls, root: Path) -> tuple[NativeSessionLocks, ...]:
         """Every saved session under ``root``, found relative to it, so a practice copy's
@@ -245,3 +258,22 @@ class CheckSession(DurableFile, SavedSessionFile):
 
     __slots__ = ()
     filename = "native-sessions/{slot}/enrolled-{session_id}.jsonl"
+
+
+class NativeAgentSettings(DurableFile, PatternFile):
+    """Pi's SettingsManager writes this file in the session's private agent directory."""
+
+    directory_name = ".native-pi-agent"
+    filename = f"native-sessions/{{lookup}}/{directory_name}/settings.json"
+
+    @classmethod
+    def prepare_directory(cls, session_directory: Path) -> Path:
+        directory = session_directory / cls.directory_name
+        NativeSessionFiles.prepare_private_directory(directory)
+        return directory
+
+
+class RetainedNativeWriteMarker(DurableFile, PatternFile):
+    """The retired PR48 writer left this after an uncertain write; retain its evidence."""
+
+    filename = "sessions/{session}.pr48-writer.lock"

src/agent_comms/owner_lifecycle.py

@@ -27,7 +27,7 @@
 )
 from .cli_result import CliResult
 from .commander_heartbeat import CommanderHeartbeat
-from .diagnostics import owner_process_output
+from .diagnostics import OwnerProcessOutput
 from .errors import RelationViolationError
 from .launch_root import LaunchRoot
 from .input_disposition import InputDispositions
@@ -493,7 +493,7 @@ def _launch_owner(
         env = initial_input.launch(thread.launch()).apply(env)
         env[NativeArguments.binary_key] = agent_bin
         agent_args.apply(env)
-        with owner_process_output(self.root, thread) as output, SpawnedProcess.launching(
+        with OwnerProcessOutput.open(self.root, thread) as output, SpawnedProcess.launching(
             (str(launch_runtime.interpreter), "-m", "agent_comms.worker",
              *LaunchRoot.from_path(self.launch_authority.child_root).arguments()),
             env=env,