Use the declared family as the carry-step roster

src/agent_comms/installer/carry.py

@@ -15,7 +15,7 @@
 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
@@ -43,25 +43,15 @@ class Carried:
 
 
 @dataclass(frozen=True)
-class Carry(ABC):
+class Carry(DeclaredFamily):
     """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:
@@ -135,7 +125,7 @@ def at(cls, tree: Path) -> CarryChain:
         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]
+        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", "--",

src/agent_comms/installer/step_worker.py

@@ -22,7 +22,8 @@ def main() -> None:
     steps = []
     for name, encoded_step in FieldCodec.parse(encoded_steps):
         import_module(f"{STEPS.name}.{name}")
-        steps.append(FieldCodec.decode(Carry.declared[name], encoded_step))
+        step_type, = (step for step in Carry.members_with(Carry) if step.name() == name)
+        steps.append(FieldCodec.decode(step_type, encoded_step))
     Path(result_path).write_text(operation.encode_answers(
         tuple(operation.run(step) for step in steps)))