Delete backend relocation modules and unify their behavior owners

src/agent_comms/historical_views.py

@@ -49,7 +49,7 @@
 from .revision_views import BusSourceView
 from .store_files import _atomic_write_text, _store_lock
 from .thread_identity import UnrecordedThreadIncarnation
-from .thread_provenance import ThreadProvenance
+from .threads import ThreadProvenance
 
 if TYPE_CHECKING:
     from .catalog_document import CatalogDocument

src/agent_comms/pi_events.py

@@ -39,7 +39,6 @@
     UnknownPayload,
     UnreportedUsage,
 )
-from .pi_turn_phase import PiTurnPhase
 from .pi_vocabulary import CompactionReason, UnknownCompactionReason
 from .request_progress import RequestProgress
 from .tool_declaration import ToolDeclaration
@@ -57,9 +56,31 @@ def retry_progress(self) -> bool:
 
 
 @dataclass(frozen=True)
-class PiEvent(PiTurnPhase, OpenPayload, affix=""):
+class PiEvent(OpenPayload, affix=""):
     """One declaration-owned record received by the Pi RPC channel."""
 
+    _start_phases: ClassVar[dict[type[object], tuple[type[phases.TurnPhase], ...]]] = {}
+    _end_events: ClassVar[dict[type[object], frozenset[type[object]]]] = {}
+
+    def on(self, current: phases.TurnPhase) -> phases.TurnPhase:
+        for event_type in type(self).__mro__:
+            for member in self._start_phases.get(event_type, ()):
+                return self.enter(member, current)
+        return self.leave(current) if self.matches_end(current) else current
+
+    def enter(self, member: type[phases.TurnPhase], current: phases.TurnPhase) -> phases.TurnPhase:
+        return member()
+
+    def matches_end(self, current: phases.TurnPhase) -> bool:
+        return any(
+            event_type in self._end_events.get(member, frozenset())
+            for member in type(current).__mro__
+            for event_type in type(self).__mro__
+        )
+
+    def leave(self, current: phases.TurnPhase) -> phases.TurnPhase:
+        return phases.ModelWaitPhase()
+
     def __init_subclass__(
         cls, *, starts: type[phases.TurnPhase] | None = None,
         ends: type[phases.TurnPhase] | None = None, **kwargs,

src/agent_comms/plan_phases.py

@@ -13,9 +13,9 @@
 
 from agent_comms.declared_family import DeclaredFamily
 from agent_comms.installer.carry import Carry
-from agent_comms.pi_turn_phase import PiTurnPhase
+from agent_comms.pi_events import PiEvent
 from agent_comms.thread_identity import BirthlessIncarnation
-from agent_comms.thread_rename import ThreadRename
+from agent_comms.private_registry_guard import RegistryPrivacy
 from agent_comms.tool_calls import CliToolArguments
 from agent_comms.phase_evidence import (
     AllEvidence,
@@ -667,11 +667,11 @@ class RostersPhase(RemainingSlopPhase):
                 ClassFlagsAbsent(r"^\s*" + attribute + r"\s*[:=]", owners=(owner,))
                 for owner, attribute in (
                     (Carry, "declared"),
-                    (ThreadRename, "kinds"),
+                    (RegistryPrivacy, "kinds"),
                     (BirthlessIncarnation, "members"),
                     (CliToolArguments, "types"),
-                    (PiTurnPhase, "_start_phases"),
-                    (PiTurnPhase, "_end_events"),
+                    (PiEvent, "_start_phases"),
+                    (PiEvent, "_end_events"),
                 )
             ),
         )

src/agent_comms/presentation.py

@@ -75,7 +75,7 @@
 from .runtime_info import RuntimeInfo
 from .storage_room import RoomLedger
 from .thread_identity import ThreadIncarnation
-from .thread_provenance import ThreadProvenance
+from .threads import ThreadProvenance
 from .thread_presentation import (
     ThreadOwnerBinding,
     ThreadPresentation,

src/agent_comms/private_registry_guard.py

@@ -29,7 +29,9 @@
 from typing import TYPE_CHECKING, Annotated, NoReturn
 
 from . import store_files
+from .coordination_contracts import MAX_IDENTIFIER_CHARS
 from .coordination_database import CoordinationStore
+from .coordination_errors import ParticipantNotRegistered
 from .coordinator import Coordination
 from .errors import RelationViolationError
 from .field_codec import FieldCodec, Id128Text, MalformedRecord, WireName
@@ -44,6 +46,7 @@
     from .notification_assignment import NotificationAssignmentRead
     from .private_nk_entrypoint import RootLaunch
     from .registry_document import RegistrySnapshot
+    from .thread_management import ThreadManagement
     from .threads import Thread
     from .typed_table import TableSchema
 
@@ -118,6 +121,15 @@ def _decode(data: bytes, root_id: bytes) -> tuple[int, int, bytes]:
 class RegistryPrivacy(ABC):
     """What a registry root's protocol marker decides, selected by the registry store."""
 
+    @abstractmethod
+    def rename(
+        self, threads: ThreadManagement, before: Thread, new_name: str
+    ) -> tuple[str, str]:
+        """Rename under this root's original registry and coordinator authority."""
+
+    def rename_applied(self, root: Path) -> None:
+        """Public roots have no private owner intent to complete."""
+
     @staticmethod
     def capture(root: Path) -> RegistryPrivacy:
         """Read original privacy authority under its registry's existing lock."""
@@ -174,6 +186,11 @@ def coordinator_schema(self) -> type[TableSchema]:
 
 
 class PublicRegistry(RegistryPrivacy):
+    def rename(
+        self, threads: ThreadManagement, before: Thread, new_name: str
+    ) -> tuple[str, str]:
+        return threads.registry.rename(before.name, new_name)
+
     def coordinator_schema(self) -> type[TableSchema]:
         from .coordination_schema import CoordinatorTable
 
@@ -231,6 +248,62 @@ def require_root(self, root_id: str, error: Exception) -> None:
 
 
 class PrivateRegistryGuard(DurableFile, WireWakeFile, RegistryPrivacy):
+    def rename(
+        self, threads: ThreadManagement, before: Thread, new_name: str
+    ) -> tuple[str, str]:
+        root, registry = threads.root, threads.registry
+        _require_no_private_owner_rename(root)
+        coordinator = CoordinationStore.path_in(root)
+        if not coordinator.exists() or before.name == new_name:
+            return registry.rename(before.name, new_name)
+        # Validate registry/SQL name refusals before renaming the private owner.
+        # Registry readers cannot select a different owner while both stores change.
+        registry.status(before.name).require_self_rename()
+        if registry.snapshot().taken(new_name, owner=before.name):
+            raise RelationViolationError(f"Thread name {new_name!r} is already in use.")
+        before.require_name(new_name)
+        if len(new_name) > MAX_IDENTIFIER_CHARS:
+            raise ValueError("Private coordinator owner name exceeds its bound.")
+        with Coordination(str(coordinator)) as store:
+            try:
+                person = store.participants.get(before.incarnation.require_recorded().lookup)
+            except ParticipantNotRegistered:
+                return registry.rename(before.name, new_name)
+            if not person.committed or person.owner_identity.owner_thread != before.name:
+                raise RelationViolationError(
+                    "Private coordinator owner differs from the registered owner."
+                )
+            generation = person.owner_identity.generation
+            PrivateOwnerRenameIntent(
+                lookup=person.lookup,
+                old=before.name,
+                new=new_name,
+                generation=generation,
+                wire_root_id=self.root_id.hex(),
+            ).write(root)
+            store.participants.rename_owner(person.lookup, new_name, expected_generation=generation)
+            try:
+                return registry.rename(before.name, new_name)
+            except BaseException as error:
+                # Registry persistence can fail after the SQL CAS. If its atomic snapshot still
+                # names the old owner, restore that name without revoking its active attempt.
+                uncertain = RelationViolationError(
+                    "Private owner rename is uncertain; inspect both authorities."
+                )
+                try:
+                    if registry.require(before.name).name == before.name:
+                        store.participants.rename_owner(
+                            person.lookup, before.name, expected_generation=generation
+                        )
+                except BaseException as compensation:
+                    # Ambiguous dual-store failure: the inspection needs both failures.
+                    raise uncertain from compensation
+                raise uncertain from error
+
+    def rename_applied(self, root: Path) -> None:
+        if PrivateOwnerRenameIntent.path_in(root).exists():
+            PrivateOwnerRenameIntent.complete(root)
+
     filename = ".registry-owner-guard"
 
     def coordinator_schema(self) -> type[TableSchema]:

src/agent_comms/registry_provenance.py

@@ -15,7 +15,7 @@
 from .errors import RelationViolationError, UnregisteredThreadError
 from .routing import DeliveryScope
 from .thread_identity import ThreadIncarnation
-from .thread_provenance import ThreadProvenance
+from .threads import ThreadProvenance
 
 if TYPE_CHECKING:
     from .audience_manifest import FrozenRecipient

src/agent_comms/thread_management.py

@@ -49,10 +49,17 @@
 from .session_file import SavedSessionFile
 from .thread_model import DeclaredModel, ThreadModel, UndeclaredModel
 from .thread_owned_state import ThreadOwnedState
-from .thread_rename import RenameThreadResult, ThreadRename
+from .tool_result import ToolResult
 from .threads import KeptModel, ModelChoice, NextTurnConfiguration, Thread, current_thread
 
 
+@dataclass(frozen=True, slots=True)
+class RenameThreadResult(ToolResult):
+    previous: str
+    current: str
+    changed: bool
+
+
 class TagChoice(SubclassUnion, affix="Tags"):
     """Select explicit, inherited or changed tags from an existing set."""
 
@@ -380,7 +387,21 @@ def _rename_thread(
     def _rename_thread_unlocked(
         self, name: str, new_name: str, *, title: ThreadText = NoThreadText()
     ) -> RenameThreadResult:
-        return ThreadRename.chosen(self, self.registry.require(name), new_name).applied(title)
+        privacy = self._privacy()
+        previous, current = privacy.rename(self, self.registry.require(name), new_name)
+        thread = self.registry.require(current)
+        titled = replace(thread, auto_title_pending=False, title=title.applied(thread.title))
+        if titled != thread:
+            self.registry.register(titled, self.registry.status(current))
+        changed = previous != current
+        if changed:
+            self.agents.activity.rename_thread(previous, current)
+            self.agents.runtime_info.rename_thread(previous, current)
+            self.ledger.rename_thread(previous, current)
+            with self.catalog.editing() as document:
+                document.rename_thread(previous, current)
+        privacy.rename_applied(self.root)
+        return RenameThreadResult(previous, current, changed)
 
     def heartbeat(self, name: str) -> None:
         self.registry.heartbeat(name)