Derive thread backend obligations and selection from capability inheritance

plans/REMAINING-SLOP-PLAN.md

@@ -133,9 +133,46 @@ Three audits, one per catalog rung group, each finding read in the code and chec
 - `RecoveryCondition` matching a lifecycle by `declared_name` string equality across two families (`recovery_states.py:21-26`): `issubclass`.
 
 **6.4 Rosters and registries beside families (MEMB-1/2/4, about 120).**
-- Five hand-rolled `__init_subclass__` registries: `Carry.declared` (`installer/carry.py:45-54`), `ThreadRename.kinds` (`thread_management.py:557-567`, gone with 4.1), `BirthlessIncarnation.members` (`thread_identity.py:185-192`), `CliToolArguments.types` (`tool_calls.py:89-95`), `_start_phases`/`_end_events` (`pi_events.py:59-93`): `DeclaredFamily` or attributes through the MRO.
-- `PI_0_85_1_EVENT_KINDS`, 36 names re-typed beside `NativePiEvent.known_kinds()` (`pi_native_backend.py:63-81`, check at `agent_backend.py:428`): derive the union; the equality check goes.
-- `THREAD_CAPABILITIES` plus `serves_threads`/`serves_new_threads` flags plus `validate_backends` probing with `getattr` (`agent_backend.py:358-466`): `ThreadServingBackend(AgentBackend, Steers, Compacts)` and `NewThreadBackend`; ABC instantiation enforces the roles.
+
+Backend capability slice (#2623, integration base `acb2ddd0e`, 2026-10-23):
+deleted `THREAD_CAPABILITIES`, both serving flags, `validate_backends` and the
+old default-selection method. `ThreadServingBackend` inherits `Steers` and
+`Compacts`; both record default factories invoke `NewThreadBackend.selected`.
+Capability bases precede `AgentBackend`, whose concrete refusal method would
+otherwise hide the abstract turn obligation. Sealing attaches only concrete
+locations after validating unique ownership. An incomplete declaration cannot
+rebind a location. `NoSessionFile` constructs the shared `UnstartedRuntime`;
+both copied backend `unstarted_type` declarations are deleted.
+NRA census covers 1,528 Core/hooks/tools/tests and paired Toad modules:
+6,973 original classes, 6,752 projected classes, zero parse omissions; the
+221 unprojected originals remain visible. Source evidence is not a behavior
+proof. Afterward: 1,529 modules, 6,974 original/6,753 projected classes, zero
+parse omissions; Core remains 953 modules/5,341 original classes. The extra
+dependency module/class came from the concurrently edited Toad source tree.
+The before/after receipts remain in the owned backend-capability-rosters
+scratch directory. Focused source checks pass 27 cases (25.69s), including
+fresh CLI/UI imports, typed native/durable locations, ABC refusal without
+location mutation, worker backend reuse and a real durable host reading 260
+saved rows while its SQLite writer stays open. The initial 24-pass/3-failure
+run exposed an untyped session-file fixture, a stale caller roster and missing
+local Node dependencies; all three were corrected, without counting that run
+as green. Installed copied-agent DM/reply/tab-return qualification belongs to
+the later paired integration; this branch has not changed the live runtime.
+Review corrected the declaration tests: they now re-execute the actual backend
+declaration module through standard `runpy.run_path` in a fresh namespace,
+declare real nominal subclasses and invoke the actual seal lifecycle. They do
+not monkeypatch production sealing, registry contents or backend locations.
+The abstract-class exclusion/no-rebind, competing-location and zero/two-default
+cases pass (3 cases, 0.11s); fresh-entrypoint import checks also pass. Production
+is unchanged from the 27-case qualification; the class census above describes
+that implementation checkpoint before the test-fixture correction.
+The independent Pi producer/converter event comparison is retained;
+deriving expected producer events from the converters would erase its check.
+This slice does not close `RostersPhase` or edit its evidence declaration.
+
+- Five remaining hand-rolled `__init_subclass__` registration owners at this ordered checkpoint: `Carry.declared` (`installer/carry.py`), `ThreadRename.kinds` (`thread_management.py`; still present), `BirthlessIncarnation.members` (`thread_identity.py`), `CliToolArguments.types` (`tool_calls.py`), and `PiTurnPhase._start_phases`/`_end_events` (`pi_events.py`). Move registration through the existing nominal mechanisms or derive behavior through the MRO; preserve each boundary's actual keys and consumers. The accepted CLI and Birthless deliveries remove their own hooks when landed.
+- Preserve the independent Pi producer ABI contract: `PI_0_85_1_EVENT_KINDS` in `pi_native_backend.py` declares the pinned runtime's emitted kinds, while `NativePiEvent.known_kinds()` declares converter coverage. `AgentBackend.require_event_family` compares these independent answers and reports missing or extra converters. Deriving the expected producer kinds from converters would erase this check. Refactor the internal `_start_phases`/`_end_events` registration lifecycle without weakening producer/converter coverage; removal of the external manifest is not a completion condition.
+- Backend roles: removed by #2623 through `ThreadServingBackend(Steers, Compacts, AgentBackend)` and declaration-derived `NewThreadBackend.selected`; abstract methods enforce thread obligations. `PiDurableBackend` remains a non-thread backend.
 - Display names re-typed from class names in about 40 places (`tools.py` labels, `notification_progress.py:45-157`, `todos.py:1502/2012/2088/2605/2675`, `display_order.py:76-136`, `relationships.py:382/397`, `turn_phase.py:169/218`, `config_options.py:192-194`, `input_attempt.py:400/440/447` `public_status`): derived from the declared name, literal only where the text differs.
 - Sigils `#`/`@` spelled by hand at about 22 sites (`catalog_document.py:61-232`, `channels.py`, `channel_targets.py`, `goal_presentation.py:71-72`, `goal_actions.py:305/318`, …): owned by `Tag.channel` and `ThreadTarget.mention`.
 - Store and file names restated outside their class: `root / "wire"` in about 20 places across 17 modules (the virtual bus plan's done-grep for `_wire_lock_path` misses every `_store_lock(root / "wire")`; its grep gains that pattern), `~/.local/state/agent-comms` six times, bridge inbox dirs, `"registry.json"` at `installer/upgrade.py:566`, `".registry-owner-guard"` at `goal_history.py:208`, `active-route.json` five times, Pi `auth.json` in four modules: `X.path_in(root)` and one state-home owner.
@@ -226,7 +263,7 @@ Waves 4 to 8 were written on 2026-10-21 and declared no phase in `plan_phases.py
 | wire_renames | 4.6 | no `WireName("admission…` in src | one unit, Toad in lockstep |
 | owned_dispatch | 5.4 | no `TurnInboxReceiver`, no hand `separators=(",", ":")` | one unit |
 | kind_families | 6.3 | no `TagAction`, `ExecutionOrigin`, `WhenBusy`, `ControlClassification`, `MembershipChange` or `ViewKind` enum classes | by file |
-| rosters | 6.4 | no `PI_0_85_1_EVENT_KINDS`, `THREAD_CAPABILITIES` or `serves_new_threads` | one unit |
+| rosters | 6.4 | internal registries and capability/name/path rosters derive from their owners; retain independent Pi producer/converter coverage | by family; shared phase proof has one owner |
 | records | 7 | no `ConnectivityFacet`/`observe_connectivity`; no `nonlocal` in the five document updaters | one unit |
 
 9.4 is Tristan's ruleset change. 9.5, 9.6, 9.9, 9.10 and 9.11 are claude-ai's and are measured by their own numbers in this section: a touched-tests PR check, the full ratchet list in the job summary, commander calls under 3 s, one GitHub reader, and one stack lock. The order across plans stays Tristan's: the virtual bus batch first, then multiple inheritance, the commander-native plan, then this plan. Phases with no stored data run in parallel with all of them on disjoint files.

src/agent_comms/agent_backend.py

@@ -6,10 +6,10 @@
 ever sees the records declared here. A backend converts its runtime's own
 types to these records; no consumer outside a backend reads runtime types.
 
-Capabilities are mixin roles composed through the MRO. A backend that serves
-threads must implement every role in ``THREAD_CAPABILITIES``; the abstract
-methods enforce the signatures, and :meth:`AgentBackend.seal` checks the rest
-once every backend module is declared, then freezes the registry.
+Capabilities are mixin roles composed through the MRO. Thread-serving backends
+inherit ``ThreadServingBackend``; new-thread selection uses ``NewThreadBackend``.
+Their inherited abstract methods enforce the required behavior. Sealing binds
+each concrete backend to its location and freezes the registry.
 
 This module is deliberately light: it is imported by ``threads`` and must not
 import any runtime implementation at module level.
@@ -19,7 +19,7 @@
 
 import asyncio
 from abc import ABC, abstractmethod
-from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
+from collections.abc import AsyncIterator, Awaitable, Callable
 from dataclasses import dataclass
 from enum import Enum
 from types import MappingProxyType
@@ -355,12 +355,6 @@ async def fork(cls, parent: Thread, launcher: str, runtime: LaunchRuntime) -> Ru
         """
 
 
-THREAD_CAPABILITIES: tuple[type, ...] = (
-    Steers,
-    Compacts,
-)
-
-
 # ---------------------------------------------------------------------------
 # The family.
 # ---------------------------------------------------------------------------
@@ -369,15 +363,12 @@ async def fork(cls, parent: Thread, launcher: str, runtime: LaunchRuntime) -> Ru
 class AgentBackend(DeclaredFamily, affix="Backend"):
     """One agent runtime, keyed by ``declared_name`` ("pi_native", ...).
 
-    A backend declares its location type, whether it serves threads, whether
-    new threads get it, and the event kinds its runtime protocol emits. One
+    A backend declares its location type, capabilities through inheritance,
+    and the event kinds its runtime protocol emits. One
     instance holds one thread session's runtime process.
     """
 
     location_type: ClassVar[type[RuntimeLocation]]
-    unstarted_type: ClassVar[type[RuntimeLocation]]
-    serves_threads: ClassVar[bool] = True
-    serves_new_threads: ClassVar[bool] = False
     event_kinds: ClassVar[frozenset[str]]
     _sealed: ClassVar[bool] = False
 
@@ -391,7 +382,6 @@ def __init_subclass__(cls, **options: Any) -> None:
                 "before the Thread backend lookup seals its declarations."
             )
         super().__init_subclass__(**options)
-        cls.location_type._backend = cls
 
     @classmethod
     @abstractmethod
@@ -402,16 +392,22 @@ def for_worker(cls, runner: Any, session_id: str) -> Self:
     @abstractmethod
     def new_location(cls, root, worktree) -> RuntimeLocation: ...
 
-    @classmethod
-    def for_new_threads(cls) -> type[AgentBackend]:
-        """The backend a newly declared thread gets."""
-        (member,) = (m for m in AgentBackend.members_with(AgentBackend) if m.serves_new_threads)
-        return member
-
     @classmethod
     def seal(cls) -> None:
-        """Validate every declared backend and freeze the registry."""
-        validate_backends(dict(AgentBackend.__registry__).values())
+        """Bind concrete locations after validating all declaration relationships."""
+        NewThreadBackend.selected()
+        owners: dict[type[RuntimeLocation], type[AgentBackend]] = {}
+        for member in cls.members_with(cls):
+            location = member.location_type
+            if not issubclass(location, RuntimeLocation):
+                raise BackendDeclarationError(f"{member.__name__} declares no RuntimeLocation type")
+            if previous := owners.get(location):
+                raise BackendDeclarationError(
+                    f"location type {location.__name__} belongs to several backends "
+                    f"{previous.__name__}, {member.__name__}")
+            owners[location] = member
+        for location, owner in owners.items():
+            location._backend = owner
         AgentBackend.__registry__ = MappingProxyType(dict(AgentBackend.__registry__))  # type: ignore[assignment]
         AgentBackend._sealed = True
 
@@ -437,33 +433,23 @@ def _event_family_problems(backend: type, family: type[DeclaredFamily]) -> list[
     return problems
 
 
-def validate_backends(members: Iterable[type]) -> None:
-    """Check backend declarations; raise once, naming every missing piece."""
-    members = tuple(members)
-    problems: list[str] = []
-    owners: dict[type, list[str]] = {}
-    for member in members:
-        name = member.__name__
-        location = getattr(member, "location_type", None)
-        if not (isinstance(location, type) and issubclass(location, RuntimeLocation)):
-            problems.append(f"{name} declares no RuntimeLocation type")
-        else:
-            owners.setdefault(location, []).append(name)
-        if not isinstance(getattr(member, "event_kinds", None), frozenset):
-            problems.append(f"{name} declares no event_kinds")
-        if member.serves_threads:
-            missing = [role.__name__ for role in THREAD_CAPABILITIES if not issubclass(member, role)]
-            if missing:
-                problems.append(f"{name} serves threads but lacks roles {missing}")
-        unimplemented = sorted(getattr(member, "__abstractmethods__", ()))
-        if unimplemented:
-            problems.append(f"{name} leaves methods abstract {unimplemented}")
-    for location, names in owners.items():
-        if len(names) > 1:
-            problems.append(f"location type {location.__name__} belongs to several backends {names}")
-    new_thread_backends = [member.__name__ for member in members if member.serves_new_threads]
-    if len(new_thread_backends) != 1:
-        problems.append(f"exactly one backend must serve new threads, found {new_thread_backends}")
-    if problems:
-        raise BackendDeclarationError("; ".join(problems))
+class ThreadServingBackend(Steers, Compacts, AgentBackend, declared_name=None):
+    """A thread backend implements steering, turn lifecycle and compaction together.
+
+    The capability bases precede AgentBackend so its non-thread turn refusal
+    cannot satisfy the abstract turn contract.
+    """
+
+
+class NewThreadBackend(ThreadServingBackend, declared_name=None):
+    """The complete thread backend selected for newly registered threads."""
+
+    @classmethod
+    def selected(cls) -> type[Self]:
+        members = cls.members_with(cls)
+        if len(members) != 1:
+            raise BackendDeclarationError(
+                f"exactly one backend must serve new threads, found "
+                f"{[member.__name__ for member in members]}")
+        return members[0]
 

src/agent_comms/pi_durable_backend.py

@@ -37,7 +37,6 @@
 from .pi_payloads import PiOmitted, PiResponseData
 from .pi_vocabulary import ThinkingLevelField
 from .transcript_sources import TranscriptSource
-from .unstarted_runtime import UnstartedRuntime
 
 
 @dataclass(frozen=True)
@@ -206,9 +205,7 @@ class CompactDurable(ExactResponse, Compact):
 
 
 class PiDurableBackend(AgentBackend, Compacts):
-    serves_threads = False
     location_type = DurableDatabase
-    unstarted_type = UnstartedRuntime
     event_kinds = frozenset(kind for member in DurablePiEvent.members_with(DurablePiEvent) for kind in member.known_kinds())
 
     def __init__(self):

src/agent_comms/pi_native_backend.py

@@ -23,14 +23,13 @@
 from agent_comms.pi_payloads import PiOmitted
 
 from .agent_backend import (
-    AgentBackend,
     CompactionOutcome,
-    Compacts,
     FailedCompaction,
     Forks,
     InputContent,
     Interrupts,
     NothingToCompaction,
+    NewThreadBackend,
     PlacedCompaction,
     RuntimeLocation,
     TurnDelivery,
@@ -42,7 +41,6 @@
 from .native_session_reopen import NativeSessionIdentity
 from .native_transcript import NativeTranscript
 from .selected_session import SavedSelectedSession
-from .unstarted_runtime import UnstartedRuntime
 
 if TYPE_CHECKING:
     from .agent_events import AgentEvent, AgentInfo
@@ -179,9 +177,8 @@ async def close_idle(self) -> None:
 class PiNativeBackend(
     PersistentPiSession,
     Interrupts,
-    Compacts,
     Forks,
-    AgentBackend,
+    NewThreadBackend,
 ):
     """Pi 0.85.1 over RPC: one retained child per thread session in this worker.
 
@@ -191,8 +188,6 @@ class PiNativeBackend(
     """
 
     location_type: ClassVar[type[RuntimeLocation]] = PiSessionFile
-    unstarted_type: ClassVar[type[RuntimeLocation]] = UnstartedRuntime
-    serves_new_threads: ClassVar[bool] = True
     event_kinds: ClassVar[frozenset[str]] = PI_0_85_1_EVENT_KINDS
 
     def __init__(

src/agent_comms/thread_identity.py

@@ -16,7 +16,7 @@
 
 
 from agent_comms.participant_identity import ParticipantIdentity
-from .agent_backend import AgentBackend, RuntimeLocation
+from .agent_backend import AgentBackend, NewThreadBackend, RuntimeLocation
 from .channel_targets import Tag
 from .cli_options import JsonFlagRecord
 from .declared_family import DeclaredFamily, declared_words
@@ -306,7 +306,7 @@ class ThreadProvenance:
     session_file: SessionFile = field(default_factory=NoSessionFile)
     title: ThreadText = field(default_factory=NoThreadText)
     backend: Annotated[type[AgentBackend], WireOmitDefault()] = field(
-        default_factory=AgentBackend.for_new_threads
+        default_factory=NewThreadBackend.selected
     )
 
     @cached_property

src/agent_comms/thread_text.py

@@ -157,7 +157,8 @@ def applied(self, previous: SessionFile) -> SessionFile:
 @dataclass(frozen=True, slots=True)
 class NoSessionFile(AbsentText, SessionFile):
     def location(self, backend: type[AgentBackend]) -> RuntimeLocation:
-        return backend.unstarted_type(backend)
+        from .unstarted_runtime import UnstartedRuntime
+        return UnstartedRuntime(backend)
 
     def applied(self, previous: SessionFile) -> SessionFile:
         return previous

src/agent_comms/threads.py

@@ -11,7 +11,7 @@
 from typing import TYPE_CHECKING, ClassVar, Annotated
 
 
-from .agent_backend import AgentBackend
+from .agent_backend import AgentBackend, NewThreadBackend
 from .child_process import NoProcess, ProcessIdentity, ProcessState
 from .cli_result import ThreadToolCatalogResult
 from .context_segments.wake_values import WakeWorkContext
@@ -202,7 +202,7 @@ class Thread(DurableFile, PatternFile, ThreadProvenance):
         type[AgentBackend],
         WireOmitDefault(),
         RegistrationPolicy(InheritPrevious),
-    ] = field(default_factory=AgentBackend.for_new_threads, kw_only=True)
+    ] = field(default_factory=NewThreadBackend.selected, kw_only=True)
     turn_record: TurnRecord = field(default_factory=NoTurnRecord)
     channel_scope_generation: int = 0
     turn_generation: int = 0