Give DM peers one typed encoding, fix stale tests, repin native build

src/agent_comms/assignment_states.py

@@ -235,7 +235,7 @@ def notification(
         sender_lookup: str,
         *,
         observation: RecipientActivity,
-        activity: AssignmentActivity | None = None,
+        activity: AssignmentActivity,
         updated_at_ms: int = 0,
         triage_inflight: bool = False,
     ) -> MessageNotification:
@@ -396,7 +396,7 @@ def notification(
         sender_lookup: str,
         *,
         observation: RecipientActivity,
-        activity: AssignmentActivity | None,
+        activity: AssignmentActivity,
         updated_at_ms: int = 0,
         triage_inflight: bool = False,
     ) -> MessageNotification:
@@ -474,7 +474,7 @@ def notification(
         sender_lookup: str,
         *,
         observation: RecipientActivity,
-        activity: AssignmentActivity | None,
+        activity: AssignmentActivity,
         updated_at_ms: int = 0,
         triage_inflight: bool = False,
     ) -> MessageNotification:

src/agent_comms/field_codec.py

@@ -1010,7 +1010,7 @@ def argument_options(cls) -> dict[str, object]:
 class DeclaredWireType(ExternalJsonShape, WireValue):
     """A member whose whole wire value is one declared JSON type, converted by its own
     from_wire and to_wire. Its shape and schema derive from that type; inherited from a
-    union, its shape would be the union holding it, and admitting would never return."""
+    union, its shape would be the union holding it, and the shape check would never return."""
 
     wire_type: ClassVar[object]
 
@@ -1949,16 +1949,16 @@ def encode(self, value: T) -> object: ...
     def wire_children(self) -> tuple[ValueDecoder[object], ...]:
         return ()
 
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         """Check outer wire shape without decoding nested values.
 
-        This may admit a value decoding rejects, but must never reject one decoding accepts.
+        This may pass a value decoding rejects, but must never reject one decoding accepts.
         """
         return True
 
-    def admits_external(self, data: object) -> bool:
+    def fits_external(self, data: object) -> bool:
         """A sibling-aware external reader may accept more than the stored wire shape."""
-        return any(self.external_decoders(data)) or self.admits_wire(data)
+        return any(self.external_decoders(data)) or self.fits_wire(data)
 
     def selects_record(self, select: WireRecordFilter) -> bool:
         return False
@@ -2110,7 +2110,7 @@ class _ExactScalarDecoder(_ScalarDecoder):
     def accepts(self, data: object) -> bool:
         return type(data) is self.target
 
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         return self.accepts(data)
 
     def accepts_value(self, value: object) -> bool:
@@ -2158,11 +2158,11 @@ class _RealDecoder(_NumberDecoder):
     __slots__ = ()
     annotations = (float,)
 
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         return type(data) in (int, float)
 
     def accepts(self, data: object) -> bool:
-        return self.admits_wire(data) and math.isfinite(cast(float, data))
+        return self.fits_wire(data) and math.isfinite(cast(float, data))
 
 
 @dataclass(frozen=True, slots=True)
@@ -2180,9 +2180,15 @@ def wire_structure(self) -> ValueDecoder[object]:
     def wire_children(self) -> tuple[ValueDecoder[object], ...]:
         return (self.wire_structure,)
 
-    def admits_wire(self, data: object) -> bool:
+    @boundary_decoder
+    def fits_wire(self, data: object) -> bool:
         # Null handlers can change the structural null contract; leave that decision to decode.
-        return data is None or self.wire_structure.admits_wire(data)
+        return data is None or self.wire_structure.fits_wire(data)
+
+    def fits_external(self, data: object) -> bool:
+        # The representation reads through its wire structure, so an external read takes
+        # whatever that structure's external reader takes.
+        return ValueDecoder.fits_external(self, data) or self.wire_structure.fits_external(data)
 
     def _rewrite_records(
         self, data: object, change: WireRecordChange, select: WireRecordFilter,
@@ -2293,7 +2299,7 @@ def accepts_value(self, value: object) -> bool:
 
 
 class _FamilyNameDecoder(_TypeDecoder, _ChoiceDecoder[Any]):
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         return type(data) is str
 
     def encode(self, value: type[DeclaredFamily]) -> str:
@@ -2321,8 +2327,8 @@ def from_annotation(cls, target, args):
     def accepts_value(self, value: object) -> bool:
         return any(type(value) is type(item) and value == item for item in self.values)
 
-    def admits_wire(self, data: object) -> bool:
-        return any(plan.admits_wire(data) for plan in self.value_plans)
+    def fits_wire(self, data: object) -> bool:
+        return any(plan.fits_wire(data) for plan in self.value_plans)
 
     def encode(self, value: object) -> object:
         for item, plan in zip(self.values, self.value_plans, strict=True):
@@ -2348,8 +2354,8 @@ class UnionDecoder[T](_ContextDecoder[T]):
     def wire_children(self) -> tuple[ValueDecoder[object], ...]:
         return self.alternatives
 
-    def admits_wire(self, data: object) -> bool:
-        return any(alternative.admits_wire(data) for alternative in self.alternatives)
+    def fits_wire(self, data: object) -> bool:
+        return any(alternative.fits_wire(data) for alternative in self.alternatives)
 
     def _rewrite_records(
         self, data: object, change: WireRecordChange, select: WireRecordFilter,
@@ -2385,7 +2391,7 @@ def decode_external(self, data: object, record: Mapping[str, object]) -> T:
             return self.decode_bound(data, record)
         return self.decode_resolved(self.decode_candidates(
             lambda alternative: alternative.decode_external(data, record),
-            admits=lambda alternative: alternative.admits_external(data),
+            fits=lambda alternative: alternative.fits_external(data),
         ))
 
     @property
@@ -2407,18 +2413,19 @@ def encode(self, value: T) -> object:
     def decode_bound(self, data: object, context: Mapping[str, object]) -> T:
         return self.decode_candidates(
             lambda alternative: alternative.decode_bound(data, context),
-            admits=lambda alternative: alternative.admits_wire(data),
+            fits=lambda alternative: alternative.fits_wire(data),
         )
 
     def decode_candidates(
         self, decode: Callable[[ValueDecoder[T]], T], *,
-        admits: Callable[[ValueDecoder[T]], bool],
+        fits: Callable[[ValueDecoder[T]], bool],
     ) -> T:
         first_error = None
+        # Shapes only rule alternatives out; when they rule out every one, the first
+        # alternative's own decode states the refusal.
+        candidates = tuple(filter(fits, self.alternatives)) or self.alternatives[:1]
         try:
-            for alternative in self.alternatives:
-                if not admits(alternative):
-                    continue
+            for alternative in candidates:
                 try:
                     return decode(alternative)
                 except MalformedRecord as error:
@@ -2444,7 +2451,7 @@ class _ArrayDecoder(ValueDecoder[Any]):
     def wire_children(self) -> tuple[ValueDecoder[object], ...]:
         return self.items
 
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         return isinstance(data, list)
 
     def _rewrite_records(
@@ -2523,8 +2530,8 @@ class _VariadicTupleDecoder(_TupleDecoder):
 
 
 class _FixedTupleDecoder(_TupleDecoder):
-    def admits_wire(self, data: object) -> bool:
-        return super().admits_wire(data) and len(cast(list[object], data)) == len(self.items)
+    def fits_wire(self, data: object) -> bool:
+        return super().fits_wire(data) and len(cast(list[object], data)) == len(self.items)
 
     def plan_items(self, data: list[object]) -> Iterable[tuple[ValueDecoder[object], object]]:
         with MalformedRecord.refusing("Tuple length does not match declaration"):
@@ -2625,7 +2632,7 @@ def decode_keys(self, data, keys, selections):
 class _StringKeyDictDecoder(_DictDecoder):
     entries = staticmethod(_object_entries)
 
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         return isinstance(data, dict)
 
     @staticmethod
@@ -2646,7 +2653,7 @@ def decode_document(self, data: object) -> dict[str, object]:
 class _PairKeyDictDecoder(_DictDecoder):
     entries = staticmethod(_object_or_pair_entries)
 
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         return isinstance(data, (dict, list))
 
     @staticmethod
@@ -2664,7 +2671,7 @@ class _FamilyRecordDecoder[Member: DeclaredFamily](_ContextDecoder[Member]):
 
     target: type[Member]
 
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         return isinstance(data, dict) and isinstance(data.get(self.target.family_discriminator), str)
 
     @property
@@ -2713,9 +2720,9 @@ def decode_bound(self, data: object, context: Mapping[str, object]) -> Member:
 class _LeafDocumentDecoder(_FamilyRecordDecoder):
     """A directly declared leaf document omits its family selector."""
 
-    def admits_wire(self, data: object) -> bool:
-        return (super().admits_wire(data)
-                or _member_decoder(self.target).admits_wire(data))
+    def fits_wire(self, data: object) -> bool:
+        return (super().fits_wire(data)
+                or _member_decoder(self.target).fits_wire(data))
 
     def encode_document(self, value: DeclaredFamily) -> object:
         return _member_decoder(type(value)).encode(value)
@@ -2736,7 +2743,7 @@ class _RecordDecoder(_ContextDecoder[Any]):
 
     target: type
 
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         if not isinstance(data, dict):
             return False
         plan = FieldCodec._record_plan(self.target)

src/agent_comms/historical_views.py

@@ -13,7 +13,7 @@
 from typing import TYPE_CHECKING
 
 from .bus import Bus
-from .channel_targets import CommandTarget, NamedTarget, ThreadTarget
+from .channel_targets import NamedTarget, ThreadTarget
 from .channels import Channel
 from .coordination_errors import HistoricalSourceChanged
 from .errors import UnregisteredThreadError
@@ -194,11 +194,11 @@ def capture(self, snapshot: RegistryProvenance) -> ChannelDisplayScope:
 @dataclass(frozen=True, slots=True)
 class DMHistory(HistoryView):
     first: str
-    second: str | ThreadTarget
+    second: ThreadTarget
 
     def capture(self, snapshot: RegistryProvenance) -> DMDisplayScope:
         return DMDisplayScope.capture(
-            self.first, CommandTarget.named(self.second).selected_name(snapshot), snapshot)
+            self.first, self.second.selected_name(snapshot), snapshot)
 
 
 @dataclass(frozen=True, slots=True)
@@ -231,10 +231,8 @@ def live_page(self, presentation: BusPresentation, bounds: PageBounds) -> Messag
         from .read_basis import DMDisplayBasis
 
         bus = presentation.bus
-        target = CommandTarget.named(self.second)
-        target.require_dm_peer()
         with presentation.registry.store.reading_snapshot() as snapshot:
-            peer = target.selected_name(snapshot)
+            peer = self.second.selected_name(snapshot)
             missing = peer not in snapshot.threads
         if missing and bus.history.threads(peer):
             return MessagePage((), False, False, NoDisplaySelection())

src/agent_comms/history_views.py

@@ -119,7 +119,7 @@ def external(self):
         self.through.wire.require_source(AssignedSourceIdentity(str(self.views.root), self.thread.incarnation))
         page = DMDisplayBasis.fetch_page(
             self.views.bus, self.views.registry, self.views.root, self.viewer.name,
-            self.thread.name, worktree=self.viewer.worktree,
+            ThreadTarget(self.thread.name), worktree=self.viewer.worktree,
             bounds=PageBounds(LiveCursor(self.through.wire.sequence + 1).older()),
         )
         if page.messages:
@@ -216,7 +216,7 @@ def historical_threads(self, name: str | None = None) -> tuple[HistoricalThread,
         return self.bus.history.threads(name)
 
     def dm_display_page(
-        self, peer: str | ThreadTarget, *, worktree: str, bounds: PageBounds = NEWEST_PAGE
+        self, peer: ThreadTarget, *, worktree: str, bounds: PageBounds = NEWEST_PAGE
     ) -> MessagePage:
         viewer = self.messaging.user_identity(worktree).name
         return DMDisplayHistory(viewer, peer, worktree).page(self.presentation, bounds)

src/agent_comms/installer/backup.py

@@ -26,6 +26,7 @@
 from agent_comms.field_codec import FieldCodec, MalformedRecord, PathText
 from agent_comms.installer.interpreter_operation import RouteSelection
 from agent_comms.locked_store import DeclaredFile, LockedStore, RootFile, RuntimeFile, UndeclaredFile
+from agent_comms.maintenance_barrier import MaintenanceBarrier, MaintenanceOperator
 from agent_comms.private_path import FileObservation, FileRevision, MissingFile
 from agent_comms.registry_store import RegistryStore
 from agent_comms.store_files import AtomicFile
@@ -199,6 +200,21 @@ def interrupted(cls, root: Path, spill_dir: Path) -> tuple[StoreBackup, ...]:
         )
         return tuple(backup for backup in backups if not backup.original())
 
+    @classmethod
+    def complete_interrupted(cls, root: Path, spill_dir: Path) -> tuple[Path, ...]:
+        """Finish every restore a killed install left unfinished, before anything reads
+        the bus: take its abandoned maintenance hold, restore, release, then retire the
+        backup so the next install never meets it again."""
+        retired = []
+        for backup in cls.interrupted(root, spill_dir):
+            maintenance = MaintenanceBarrier(RegistryStore.at(root).path)
+            hold = maintenance.begin(MaintenanceOperator.current())
+            with hold.holding():
+                backup.restore()
+            maintenance.clear(hold)
+            retired.append(backup.retire())
+        return tuple(retired)
+
     def retire(self) -> Path:
         archived = self.path.with_name(f'{self.path.name}.restored-{time.time_ns()}')
         self.path.rename(archived)

src/agent_comms/installer/upgrade.py

@@ -771,6 +771,7 @@ def upgrade(checkout: Path, root: Path, target: str, installer: InstallerChoice,
     tree = TargetTree(checkout, scratch, target)
     bridges = report.bridges
     with commands, report.running(tree):
+        StoreBackup.complete_interrupted(root, spill_dir)
         installed = Stack(tree.path, commands)
         current = commands.run("git", "-C", checkout, "rev-parse", "HEAD").strip()
         ancestry = ["git", "-C", str(checkout), "merge-base", "--is-ancestor", current, target]

src/agent_comms/locked_store.py

@@ -59,6 +59,14 @@ class RootFile:
     filename: ClassVar[str]
     root_directory: ClassVar[str] = "."
 
+    def __init_subclass__(cls, **kwargs: object) -> None:
+        """A declared file is either retained across installations or rewritten by its
+        running owner; a declaration that is neither has no install behaviour at all."""
+        super().__init_subclass__(**kwargs)
+        if "filename" in vars(cls) and not issubclass(cls, (DurableFile, RuntimeFile)):
+            raise TypeError(f"{cls.__qualname__} declares {cls.filename!r} but is neither "
+                            "a DurableFile nor a RuntimeFile")
+
     @classmethod
     def path_in(cls, root: Path | str, **parts: str) -> Path:
         return Path(root) / cls.filename.format(**parts)
@@ -336,6 +344,12 @@ class DurableFile(RootFile):
     __slots__ = ()
 
 
+class RuntimeFile(RootFile):
+    """A bus file no carry writes: running owners rewrite it at any time, maintenance included."""
+
+    __slots__ = ()
+
+
 class BusFile(DurableFile):
     """Declared data and lifecycle files that rule out a fresh bus installation."""
 
@@ -386,7 +400,7 @@ def copy_group(cls, source: Path, target: Path) -> frozenset[Path]:
         return copied
 
 
-class RetiredCommanderFile(RootFile):
+class RetiredCommanderFile(DurableFile):
     """The retired commander file that tools/cutover/carry_commander_project.py consumes.
     The practice copy holds declared files only, so an undeclared input made the practice
     carry find nothing while the live carry deleted it. Deleted with that carry."""
@@ -642,12 +656,6 @@ def publish_change(cls, path: Path) -> None:
             ) from error
 
 
-class RuntimeFile(RootFile):
-    """A bus file no carry writes: running owners rewrite it at any time, maintenance included."""
-
-    __slots__ = ()
-
-
 @dataclass(frozen=True, slots=True)
 class LockedStore(FileStore, StoreLockOwner, JsonDocument[T], declared_name=None):
     """Ordered document transactions; callbacks never mutate or reenter."""