diff --git a/src/agent_comms/acp.py b/src/agent_comms/acp.py
index 30a39596cc..8f5819dec7 100644
--- a/src/agent_comms/acp.py
+++ b/src/agent_comms/acp.py
@@ -76,7 +76,7 @@
 from .coordination_database import CoordinationStore
 from .cursor_owner import CursorOwner
 from .cursor_publication import CursorPublication
-from .nominal.field_codec import AcpSessionId, ClaimId, FieldCodec, RootId
+from .nominal.field_codec import AcpSessionId, AcpSessionIdText, ClaimId, FieldCodec, RootId
 from .image_inputs import prompt_images
 from .input_drain import InputDrain
 from .config_options import ConfigOptions
@@ -192,7 +192,7 @@ class SdkSessionConnection(AgentSideConnection, Controller, SessionUpdateSink):
     """The SDK connection and controller publish one Core acceptance answer."""
 
     async def publish_update(self, *, session_id: AcpSessionId, update: BaseModel) -> bool:
-        await self.session_update(session_id=session_id, update=update)
+        await self.session_update(session_id=AcpSessionIdText.encode(session_id), update=update)
         return True
 
 
diff --git a/src/agent_comms/coordination_work.py b/src/agent_comms/coordination_work.py
new file mode 100644
index 0000000000..369e833d51
--- /dev/null
+++ b/src/agent_comms/coordination_work.py
@@ -0,0 +1,28 @@
+"""Joined blocking work without opening or importing coordination stores."""
+
+from collections.abc import Callable
+from contextvars import Context
+from functools import partial
+
+from .child_process import AsyncWork
+from .maintenance_barrier import MaintenanceHolder
+from .native_build_inputs import GitLaunch
+from .store_files import StoreLock
+
+
+class CoordinationWork:
+    """The shared resource context for blocking store and registry operations."""
+
+    @staticmethod
+    def worker_context() -> Context:
+        return GitLaunch.in_context(MaintenanceHolder.in_context(StoreLock.worker_context()))
+
+    @staticmethod
+    async def run_worker[Result](operation: Callable[[], Result]) -> Result:
+        """Wait for blocking work before its caller releases the enclosing resources.
+
+        The callback owns and closes every borrowed resource it opens. This
+        scheduling seam also supports registry-only acquisition/retirement;
+        those operations must not depend on opening a coordinator connection.
+        """
+        return await AsyncWork.run_async(partial(CoordinationWork.worker_context().run, operation))
diff --git a/src/agent_comms/coordinator.py b/src/agent_comms/coordinator.py
index 39b307b4f3..d4a619aa49 100644
--- a/src/agent_comms/coordinator.py
+++ b/src/agent_comms/coordinator.py
@@ -8,15 +8,15 @@
 from collections.abc import AsyncIterator, Callable
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import AbstractContextManager, ExitStack, asynccontextmanager
-from contextvars import Context, ContextVar
+from contextvars import ContextVar
 from functools import partial
 from pathlib import Path
 from typing import TYPE_CHECKING, ClassVar, Final, Self, TypeVar
 
 from .assignment_store import AssignmentStore
 from .attempt_store import AttemptStore
-from .child_process import AsyncWork, join_retirement
-from .native_build_inputs import GitLaunch
+from .child_process import join_retirement
+from .coordination_work import CoordinationWork
 from .coordination_session import (
     CoordinationSession,
     NativeCoordinationSession,
@@ -24,9 +24,8 @@
     system_clock,
 )
 from .execution_store import ExecutionStore
-from .maintenance_barrier import MaintenanceHolder
 from .participant_store import ParticipantStore
-from .store_files import DEFAULT_STORE_CONTENTION, StoreContention, StoreLock
+from .store_files import DEFAULT_STORE_CONTENTION, StoreContention
 from .working_memory_annotations import WorkingMemoryAnnotations
 
 if TYPE_CHECKING:
@@ -96,7 +95,7 @@ async def run_read(
 STANDALONE_ROUTE: Final = StandaloneRoute()
 
 
-class Coordination(ExecutionStore, AttemptStore):
+class Coordination(CoordinationWork, ExecutionStore, AttemptStore):
     watch_period: ClassVar[float] = 60
     session_type = NativeCoordinationSession
     _worker: ContextVar[CoordinationRoute] = ContextVar(
@@ -115,20 +114,6 @@ async def worker(cls, path: str | Path) -> AsyncIterator[CoordinationWorker]:
             finally:
                 cls._worker.reset(token)
 
-    @staticmethod
-    def worker_context() -> Context:
-        return GitLaunch.in_context(MaintenanceHolder.in_context(StoreLock.worker_context()))
-
-    @staticmethod
-    async def run_worker(operation: Callable[[], Result]) -> Result:
-        """Wait for blocking work before its caller releases the enclosing resources.
-
-        The callback owns and closes every borrowed resource it opens. This
-        scheduling seam also supports registry-only acquisition/retirement;
-        those operations must not depend on opening a coordinator connection.
-        """
-        return await AsyncWork.run_async(partial(Coordination.worker_context().run, operation))
-
     @classmethod
     async def watch_worker(
         cls, watch: Watch, opened_wire: Callable[[], AbstractContextManager[Comms]],
diff --git a/src/agent_comms/message_notification.py b/src/agent_comms/message_notification.py
index f003630c9c..ede31de970 100644
--- a/src/agent_comms/message_notification.py
+++ b/src/agent_comms/message_notification.py
@@ -11,7 +11,7 @@
 from .read_ledger import ReadLedger
 from .frozen_recipient import FrozenRecipient
 from .channel_targets import CommandTarget
-from .nominal.field_codec import ThreadLookup
+from .nominal.field_codec import MessageId, ThreadLookup
 from .message_reference import MessageReference
 from .notification_assignment import NotificationAssignment
 from agent_comms.notification_progress import AssignmentResponse
@@ -42,7 +42,7 @@
 class NotificationWindow(ABC):
     """One window owns its evidence, batching and change contract."""
 
-    outcomes: Mapping[tuple[int, str], tuple[MessageNotification, ...]]
+    outcomes: Mapping[tuple[int, MessageId], tuple[MessageNotification, ...]]
     reads: tuple[NotificationAssignmentRead, ...]
     read_marks: tuple[NotificationReadMark, ...] = ()
 
@@ -270,7 +270,7 @@ def project_delivery_window(
         if len(sources) > NotificationSelection.window_limit:
             raise ValueError("Notification reads require a bounded visible message window")
         keys = {(source.message.seq, source.message.message_id) for source in sources}
-        result: dict[tuple[int, str], list[MessageNotification]] = {key: [] for key in keys}
+        result: dict[tuple[int, MessageId], list[MessageNotification]] = {key: [] for key in keys}
         if not keys:
             return CapturedNotificationWindow({}, (assignment_read,))
         rows = NotificationAssignment.resolve(assignment_read)
diff --git a/src/agent_comms/runtime_requests.py b/src/agent_comms/runtime_requests.py
index 1d76386664..344f39252b 100644
--- a/src/agent_comms/runtime_requests.py
+++ b/src/agent_comms/runtime_requests.py
@@ -161,7 +161,9 @@ async def observing(self, ctx: AttachedRuntimeRequestContext) -> AsyncIterator[N
         agent = ctx.server.agent
         ctx.server.clients.setdefault(ctx.session, set()).add(ctx.client)
         await agent.emit_session_identity(ctx.session, ctx.name, client=ctx.client)
-        async with agent.sessions.transcript.subscribe(self.thread, client=ctx.client):
+        async with agent.sessions.transcript.subscribe(
+            ctx.name, session_id=ctx.session, client=ctx.client,
+        ):
             await agent.turns.publish_turn_state(
                 AcpEventConsumer(agent, ctx.session, ctx.client)
             )
diff --git a/src/agent_comms/session_lifecycle.py b/src/agent_comms/session_lifecycle.py
index 3b6522c249..8860c90fc6 100644
--- a/src/agent_comms/session_lifecycle.py
+++ b/src/agent_comms/session_lifecycle.py
@@ -249,7 +249,9 @@ async def load_session(self, request: LoadSessionRequest) -> LoadSessionResponse
         if not thread.process_identity.runs_here():
             return await self.attach_owner(thread, session_id)
         await self._bind_loaded(thread, session_id)
-        await self.transcript.replay(session_id, self.runtime)
+        await self.transcript.replay(
+            thread.name, session_id=session_id, client=self.runtime,
+        )
         self.effects.inputs.ensure_live_drain(session_id)
         response = LoadSessionResponse(
             field_meta=await self.metadata(thread.name, session_id=session_id),
@@ -355,7 +357,9 @@ async def attach_owner(
                 self.proxies[session_id] = proxy
             except IncompatibleOwnerError as error:
                 # History is Core's own store; only owner requests need the owner's Core.
-                await self.transcript.replay(session_id, self.runtime)
+                await self.transcript.replay(
+                    thread.name, session_id=session_id, client=self.runtime,
+                )
                 self.proxies[session_id] = error.owner
                 return LoadSessionResponse(field_meta={
                     **await self.metadata(thread.name, session_id=session_id),
diff --git a/src/agent_comms/session_load.py b/src/agent_comms/session_load.py
index 50b5a215eb..af3bafa2ba 100644
--- a/src/agent_comms/session_load.py
+++ b/src/agent_comms/session_load.py
@@ -1,14 +1,15 @@
 """The declared session type checks owner permission when handling an ACP load."""
 from abc import abstractmethod
+from collections.abc import Awaitable, Callable
 from functools import partial
-from .coordinator import Coordination
+from .coordination_work import CoordinationWork
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
 from .nominal.declared_family import DeclaredFamily
 from .errors import RelationViolationError
 from .nominal.field_codec import FieldCodec
-from .thread_presentation import LiveThreadOwnerBinding, ThreadOwnerBinding
+import agent_comms.thread_presentation as thread_presentation
 from .native_arguments import GivenArguments
 
 if TYPE_CHECKING:
@@ -16,7 +17,7 @@
     from .threads import Thread
 
 
-class SessionLoadPlan(DeclaredFamily, affix='SessionLoadPlan'):
+class SessionLoadPlan(CoordinationWork, DeclaredFamily, affix='SessionLoadPlan'):
     @classmethod
     def at_ingress(cls, metadata: object) -> 'SessionLoadPlan':
         return EnsuringSessionLoadPlan() if metadata is None else FieldCodec.decode(cls, metadata)
@@ -24,17 +25,31 @@ def at_ingress(cls, metadata: object) -> 'SessionLoadPlan':
     def metadata(self) -> dict:
         return {"agentCommsLoad": FieldCodec.encode(self)}
 
+    @staticmethod
+    def witnessed(metadata: object, key: str, witness: type[SessionLoadPlan]) -> SessionLoadPlan:
+        """Foreign metadata carries a load witness under its key, or carries none."""
+        if isinstance(metadata, dict) and key in metadata:
+            return FieldCodec.decode(witness, metadata[key])
+        return EnsuringSessionLoadPlan()
+
     @classmethod
     def at_response(cls, metadata: object) -> 'SessionLoadPlan':
-        if not isinstance(metadata, dict):
-            return EnsuringSessionLoadPlan()
-        witness = metadata.get("agentCommsLoad")
-        return (EnsuringSessionLoadPlan() if witness is None
-                else FieldCodec.decode(ExistingSessionLoadPlan, witness))
+        return cls.witnessed(metadata, "agentCommsLoad", ExistingSessionLoadPlan)
+
+    @classmethod
+    def from_failure(cls, data: object) -> SessionLoadPlan:
+        """Decode foreign failure metadata into a plan with its own reload behavior."""
+        return cls.witnessed(data, "agentCommsLoadFailure", FailedSessionLoadPlan)
 
-    def superseded_by(self, binding: ThreadOwnerBinding) -> bool:
+    def superseded_by(self, binding: thread_presentation.ThreadOwnerBinding) -> bool:
         return False
 
+    async def reload_if_superseded(
+        self, binding: thread_presentation.LiveThreadOwnerBinding,
+        reload: Callable[[SessionLoadPlan], Awaitable[None]],
+    ) -> None:
+        """A plan without an owner witness cannot request an observed reload."""
+
     @abstractmethod
     async def resolve(self, lifecycle: "AttachedSessionLifecycle", thread: "Thread") -> "Thread": ...
 
@@ -42,7 +57,7 @@ async def resolve(self, lifecycle: "AttachedSessionLifecycle", thread: "Thread")
 @dataclass(frozen=True, slots=True)
 class EnsuringSessionLoadPlan(SessionLoadPlan):
     async def resolve(self, lifecycle: "AttachedSessionLifecycle", thread: "Thread") -> "Thread":
-        return await Coordination.run_worker(partial(
+        return await self.run_worker(partial(
             lifecycle.comms.owners.ensure_owner, thread.name,
             agent_bin=lifecycle.effects.inputs.configuration.launch.binary,
             agent_args=GivenArguments(lifecycle.effects.inputs.configuration.agent_args.argv),
@@ -53,18 +68,25 @@ async def resolve(self, lifecycle: "AttachedSessionLifecycle", thread: "Thread")
 class WitnessedSessionLoadPlan(SessionLoadPlan):
     """Original load command/response evidence, never current registry state."""
 
-    binding: ThreadOwnerBinding
+    binding: thread_presentation.ThreadOwnerBinding
 
-    def superseded_by(self, binding: ThreadOwnerBinding) -> bool:
+    def superseded_by(self, binding: thread_presentation.ThreadOwnerBinding) -> bool:
         return self.binding.superseded_by(binding)
 
+    async def reload_if_superseded(
+        self, binding: thread_presentation.LiveThreadOwnerBinding,
+        reload: Callable[[SessionLoadPlan], Awaitable[None]],
+    ) -> None:
+        if self.superseded_by(binding):
+            await reload(ExistingSessionLoadPlan(binding))
+
 
 @dataclass(frozen=True, slots=True)
 class ExistingSessionLoadPlan(WitnessedSessionLoadPlan):
-    binding: LiveThreadOwnerBinding
+    binding: thread_presentation.LiveThreadOwnerBinding
 
     async def resolve(self, lifecycle: "AttachedSessionLifecycle", thread: "Thread") -> "Thread":
-        entry = (await Coordination.run_worker(partial(
+        entry = (await self.run_worker(partial(
             lifecycle.comms.registry.entry, thread.name,
         ))).entry
         if thread.incarnation != self.binding.owner.incarnation:
@@ -79,13 +101,6 @@ async def resolve(self, lifecycle: "AttachedSessionLifecycle", thread: "Thread")
 class FailedSessionLoadPlan(WitnessedSessionLoadPlan):
     """The original failed load command's witness; never an input permission."""
 
-    @classmethod
-    def from_failure(cls, data: object) -> 'FailedSessionLoadPlan | None':
-        if not isinstance(data, dict):
-            return None
-        witness = data.get("agentCommsLoadFailure")
-        return None if witness is None else FieldCodec.decode(cls, witness)
-
     def failure_metadata(self) -> dict:
         return {"agentCommsLoadFailure": FieldCodec.encode(self)}
 
diff --git a/src/agent_comms/thread_management.py b/src/agent_comms/thread_management.py
index a44a68fa8e..91449300f6 100644
--- a/src/agent_comms/thread_management.py
+++ b/src/agent_comms/thread_management.py
@@ -58,7 +58,7 @@
 from .pi_vocabulary import NoThinkingOverride, ThinkingChoice, ThinkingLevelField
 from .registry_provenance import RegistryNames
 from .session_file import RegistrySessionLocation
-from .thread_identity import AgentThreadRole, ThreadRole
+from .thread_identity import AgentThreadRole, ThreadIncarnation, ThreadRole
 from .thread_model import DeclaredModel, ThreadModel, UndeclaredModel
 from .thread_owned_state import ThreadOwnedState
 from .thread_text import ForkedParent, GivenThreadText, NoThreadText, ThreadText
@@ -354,11 +354,13 @@ def rename_self(self, new_name: str, *, caller: ThreadCaller) -> RenameThreadRes
                 edit, caller.name, new_name, title=GivenThreadText(new_name.replace("-", " ").replace("_", " "))
             )
 
-    def set_project(self, name: str, path: str) -> ProjectChangeResult:
-        """Move one thread's project, preserving its owner and conversation."""
+    def set_project(self, target_thread: ThreadIncarnation, path: str) -> ProjectChangeResult:
+        """Move the captured thread's project, never a later occupant of its name."""
         if not path.strip():
             raise ValueError("Project path cannot be empty.")
-        original = self.registry.require(name)
+        original = self.registry.require(target_thread.name)
+        if original.incarnation != target_thread:
+            raise RelationViolationError("Thread incarnation changed before project selection")
         target = Path(path).expanduser()
         if not target.is_absolute():
             target = original.worktree.path / target
@@ -374,7 +376,7 @@ def move(current: Thread) -> Thread:
             current.require_original_project(original)
             return current if current.project_path == target else current.moved_to_project(project)
 
-        updated = self.registry.update_thread(name, move)
+        updated = self.registry.update_thread(original.name, move)
         previous, current = str(original.project_path), str(updated.project_path)
         return ProjectChangeResult(updated.name, previous, current, current != previous)
 
diff --git a/src/agent_comms/thread_presentation.py b/src/agent_comms/thread_presentation.py
index b372018ab3..30d0ba0cc4 100644
--- a/src/agent_comms/thread_presentation.py
+++ b/src/agent_comms/thread_presentation.py
@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field, fields, replace
 from typing import TYPE_CHECKING, ClassVar, NoReturn, Self
 
@@ -14,6 +14,7 @@
 from .child_process import ProcessIdentity
 
 from .message_reference import MessageReference
+import agent_comms.session_load as session_load
 
 if TYPE_CHECKING:
     from .presentation import MessageNotification
@@ -149,6 +150,12 @@ def replaces(self, original: LiveThreadOwnerBinding) -> bool:
     def superseded_by(self, replacement: ThreadOwnerBinding) -> bool:
         return False
 
+    async def reload_if_superseded(
+        self, previous: session_load.SessionLoadPlan,
+        reload: Callable[[session_load.SessionLoadPlan], Awaitable[None]],
+    ) -> None:
+        """An unavailable owner cannot supply a new existing-session load."""
+
     def recipient_activity(self, activity, snapshot, thread):
         from .agent_activity import UnavailableRecipientActivity
 
@@ -186,6 +193,12 @@ def recipient_activity(self, activity, snapshot, thread):
     def superseded_by(self, replacement: ThreadOwnerBinding) -> bool:
         return replacement.replaces(self)
 
+    async def reload_if_superseded(
+        self, previous: session_load.SessionLoadPlan,
+        reload: Callable[[session_load.SessionLoadPlan], Awaitable[None]],
+    ) -> None:
+        await previous.reload_if_superseded(self, reload)
+
     def replaces(self, original: LiveThreadOwnerBinding) -> bool:
         # Both operands are the ORIGINAL registry owner/process witnesses.
         # Queued-input generations use a different counter.
diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py
index f4693aa118..fcf00358ce 100644
--- a/src/agent_comms/tools.py
+++ b/src/agent_comms/tools.py
@@ -517,7 +517,7 @@ class CommsSetProjectTool(ThreadToolRequest):
     path: Annotated[str, *tool_field("New project directory")]
 
     def apply(self, comms: Comms, caller: ThreadCaller) -> ProjectSet:
-        result = comms.threads.set_project(caller.name, self.path)
+        result = comms.threads.set_project(caller.identity(comms), self.path)
         return ProjectSet(
             result.thread, result.previous, result.current, result.changed,
             (
diff --git a/src/agent_comms/transcript_updates.py b/src/agent_comms/transcript_updates.py
index 4c0f7029d4..edbf04f49a 100644
--- a/src/agent_comms/transcript_updates.py
+++ b/src/agent_comms/transcript_updates.py
@@ -17,6 +17,7 @@
 )
 from .coordinator import Coordination
 from .nominal.declared_family import DeclaredFamily
+from .nominal.field_codec import AcpSessionId
 from .order_tools import TranscriptPageRead
 from .turn_lease import TurnState
 from .ui_model.observation import ObservationProcess, ReadTranscript, SubscribeTranscript
@@ -31,7 +32,7 @@
 class TranscriptUpdate(DeclaredFamily, affix="TranscriptUpdate"):
     @abstractmethod
     async def publish(
-        self, session_id: str, client: SessionUpdateSink,
+        self, session_id: AcpSessionId, client: SessionUpdateSink,
     ) -> None: ...
 
 
@@ -40,7 +41,7 @@ class TurnTranscriptUpdate(TranscriptUpdate):
     state: TurnState
 
     async def publish(
-        self, session_id: str, client: SessionUpdateSink,
+        self, session_id: AcpSessionId, client: SessionUpdateSink,
     ) -> None:
         await client.publish_update(
             session_id=session_id,
@@ -74,25 +75,29 @@ async def observing(self) -> AsyncIterator[ObservationProcess]:
             if not self._subscriptions:
                 await reads.close()
 
-    async def replay(self, session_id: str, client: SessionUpdateSink) -> None:
+    async def replay(
+        self, thread: str, *, session_id: AcpSessionId, client: SessionUpdateSink,
+    ) -> None:
         async with self.observing() as reads:
             snapshot = await reads.read(
-                ReadTranscript(str(self.comms.root), TranscriptPageRead(thread=session_id)),
+                ReadTranscript(str(self.comms.root), TranscriptPageRead(thread=thread)),
             )
             await self._publish(session_id, snapshot, client)
 
     @asynccontextmanager
-    async def subscribe(self, session_id: str, client: SessionUpdateSink) -> AsyncIterator[None]:
+    async def subscribe(
+        self, thread: str, *, session_id: AcpSessionId, client: SessionUpdateSink,
+    ) -> AsyncIterator[None]:
         async with self.observing() as reads:
             snapshot = await reads.read(
-                SubscribeTranscript(str(self.comms.root), TranscriptPageRead(thread=session_id)),
+                SubscribeTranscript(str(self.comms.root), TranscriptPageRead(thread=thread)),
             )
             await self._publish(session_id, snapshot, client)
             yield
 
     async def _publish(
         self,
-        session_id: str,
+        session_id: AcpSessionId,
         snapshot: TranscriptSnapshotUpdate,
         destination: SessionUpdateSink,
     ) -> None:
diff --git a/src/agent_comms/ui_model/observation.py b/src/agent_comms/ui_model/observation.py
index 8cab7f71e7..2046316991 100644
--- a/src/agent_comms/ui_model/observation.py
+++ b/src/agent_comms/ui_model/observation.py
@@ -59,6 +59,7 @@
 from agent_comms.message_notification import NotificationWindow
 from agent_comms.notification_read import NotificationsRead, UnreadableNotificationsRead
 from agent_comms.message_reference import MessageReference
+from agent_comms.nominal.field_codec import MessageId
 from agent_comms.nominal.fold import Absorbs, Accumulates, Fold
 from agent_comms.notification_reads import NotificationRead
 from agent_comms.read_basis import DisplayCut, NoDisplaySelection
@@ -593,7 +594,7 @@ def read(self, comms: Comms) -> NotificationWindow:
         return self.operation.apply(comms, self.display)
 
     @property
-    def keys(self) -> tuple[tuple[int, str], ...]:
+    def keys(self) -> tuple[tuple[int, MessageId], ...]:
         """The outcome keys this read shows, as windows key their outcomes."""
         return tuple((reference.seq, reference.message_id) for reference in self.operation.references)
 
@@ -791,7 +792,7 @@ class NotificationBatch(BatchRead):
     observed: NotificationAnswer
 
     @cached_property
-    def keys(self) -> frozenset[tuple[int, str]]:
+    def keys(self) -> frozenset[tuple[int, MessageId]]:
         return frozenset(self.request.keys)
 
     def settled(self, family: ReadNotifications, current: list[NotificationBatch],
@@ -812,9 +813,9 @@ def __init__(self, request: ObserveNotifications):
         self.batches: tuple[NotificationBatch, ...] = ()
         self.answered: dict[ReadNotifications, tuple[NotificationBatch, ...]] = {}
 
-    def wanted(self) -> dict[ReadNotifications, frozenset[tuple[int, str]]]:
+    def wanted(self) -> dict[ReadNotifications, frozenset[tuple[int, MessageId]]]:
         """Every key some open request shows, by the request family that reads it."""
-        wanted: dict[ReadNotifications, frozenset[tuple[int, str]]] = {}
+        wanted: dict[ReadNotifications, frozenset[tuple[int, MessageId]]] = {}
         for request in self.request.reads:
             wanted[request.family] = wanted.get(request.family, frozenset()) | frozenset(request.keys)
         return wanted
@@ -1199,7 +1200,7 @@ class NotificationAnswer(Observed, Fold):
 
 @dataclass(frozen=True)
 class NotificationsObserved(Accumulates, NotificationAnswer):
-    outcomes: Mapping[tuple[int, str], tuple[MessageNotification, ...]]
+    outcomes: Mapping[tuple[int, MessageId], tuple[MessageNotification, ...]]
 
     def including(self, batch: NotificationsObserved) -> NotificationsObserved:
         """This request's answer with a readable batch's outcomes for the keys it shows."""
diff --git a/tests/test_acp.py b/tests/test_acp.py
index 228ac22f3c..ff791a5cf7 100644
--- a/tests/test_acp.py
+++ b/tests/test_acp.py
@@ -52,9 +52,11 @@ class RecordingClient(SessionUpdateSink):
 
     def __init__(self):
         self.updates = []
+        self.destinations = []
 
     async def publish_update(self, session_id=None, update=None, **kwargs):
         self.updates.append(update)
+        self.destinations.append(session_id)
         return True
 
 
@@ -178,8 +180,9 @@ async def test_compaction_details_preserve_markdown(self, tmp_path):
         assert "\x1b" not in backend.compaction_summary("\x1b[2J\nSafe")
 
     @pytest.mark.parametrize("renamed", [False, True])
+    @pytest.mark.parametrize("subscribed", [False, True])
     async def test_snapshot_capability_replays_one_bounded_update(
-        self, tmp_path, renamed, native_runtime_package
+        self, tmp_path, renamed, subscribed, native_runtime_package
     ):
         agent = self._agent(tmp_path)
         await agent.initialize(1)
@@ -206,11 +209,22 @@ async def test_snapshot_capability_replays_one_bounded_update(
             agent._comms.threads.rename_managed_thread("proj", "renamed", owner=ProcessIdentity.capture(os.getpid()))
         client = RecordingClient()
         sent = client.updates
+        destination = AcpSessionId("snapshot-destination")
+        thread = agent._comms.registry.require("proj")
         try:
-            await agent.sessions.transcript.replay("proj", client=client)
+            if subscribed:
+                async with agent.sessions.transcript.subscribe(
+                    thread.name, session_id=destination, client=client,
+                ):
+                    assert len(sent) == 1
+            else:
+                await agent.sessions.transcript.replay(
+                    thread.name, session_id=destination, client=client,
+                )
+            assert client.destinations == [destination]
             assert len(sent) == 1
             update = facts(sent[0].field_meta, TranscriptSnapshotUpdate)[0]
-            assert update.identity.requested_name == "proj"
+            assert update.identity.requested_name == thread.name
             assert update.identity.thread.incarnation == agent._comms.registry.require("proj").incarnation
             page = update.page
             events = page.events
diff --git a/tests/test_bulk_lifetimes.py b/tests/test_bulk_lifetimes.py
index 6fc60d3726..f587392c4c 100644
--- a/tests/test_bulk_lifetimes.py
+++ b/tests/test_bulk_lifetimes.py
@@ -98,7 +98,7 @@ def test_rehome_refuses_a_changed_original_before_publishing_the_staged_copy(
         rehomed = workers.submit(comms.threads.rehome_thread, paused_import, name=original.name)
         try:
             assert paused_import.snapshot.entered.wait(5)
-            moved = workers.submit(comms.threads.set_project, original.name, str(changed_project))
+            moved = workers.submit(comms.threads.set_project, original.incarnation, str(changed_project))
             moved.result(timeout=2)
         finally:
             paused_import.snapshot.proceed.set()
diff --git a/tests/test_checkout_submodules.py b/tests/test_checkout_submodules.py
index fe32655668..3a93e75108 100644
--- a/tests/test_checkout_submodules.py
+++ b/tests/test_checkout_submodules.py
@@ -520,7 +520,7 @@ def test_transfer_refuses_current_reviewer_then_keeps_departed_reviewers_history
     assert (git(source.path, 'rev-parse', 'HEAD'),
             git(source.path, 'status', '--porcelain')) == before
 
-    assert comms.threads.set_project(reviewer.name, str(reviewer_path)).changed
+    assert comms.threads.set_project(reviewer.incarnation, str(reviewer_path)).changed
     departed = comms.registry.require(reviewer.name)
     assert departed.incarnation == reviewer.incarnation
     assert departed.worktree == Checkout(reviewer_path)
@@ -548,8 +548,8 @@ def test_transfer_refuses_foreign_allocation_even_after_its_owner_changes_projec
         'reviewer', frozenset(), source, execution=ExternalThreadExecution))
     reviewer_path = rule.order_checkout(rule.roots_of(reviewer), 'P3')
     Checkout.add(repository, reviewer_path, NewBranchStart('reviewer-next', 'HEAD'))
-    assert comms.threads.set_project(worker.name, str(worker_path)).changed
-    assert comms.threads.set_project(reviewer.name, str(reviewer_path)).changed
+    assert comms.threads.set_project(worker.incarnation, str(worker_path)).changed
+    assert comms.threads.set_project(reviewer.incarnation, str(reviewer_path)).changed
     worker = comms.registry.require(worker.name)
     reviewer = comms.registry.require(reviewer.name)
     assert worker.worktree == Checkout(worker_path)
@@ -581,8 +581,8 @@ def test_transfer_accepts_callers_historical_launch_without_current_allocation(
         rule.require_transfer_source(
             worker, source, comms.registry.snapshot().threads.values(),
             comms.goals.todos.get('P1').repository)
-    assert comms.threads.set_project(worker.name, str(source)).changed
-    assert comms.threads.set_project(worker.name, str(next_project)).changed
+    assert comms.threads.set_project(worker.incarnation, str(source)).changed
+    assert comms.threads.set_project(worker.incarnation, str(next_project)).changed
     worker = comms.registry.require(worker.name)
     assert checkout in worker.earlier_projects
     assert not rule.owns_present_checkout(worker, source, RepositoryReads())
diff --git a/tests/test_project.py b/tests/test_project.py
index ed79f44c3e..f60d9c9274 100644
--- a/tests/test_project.py
+++ b/tests/test_project.py
@@ -16,7 +16,7 @@
 from delivery_owner_fixture import canonical_agent
 from agent_comms.comms import wire
 from agent_comms.acp_extension import CoordinationChangedUpdate, decode_updates
-from agent_comms.native_arguments import GivenArguments, GivenBinary
+from agent_comms.repo import Checkout
 
 
 async def test_project_update_is_published_and_old_saved_cwd_can_resume(
@@ -27,8 +27,8 @@ async def test_project_update_is_published_and_old_saved_cwd_can_resume(
     old.mkdir()
     new.mkdir()
     comms = wire(tmp_path / "wire")
-    agent = canonical_agent(comms, agent_bin=GivenBinary("/bin/echo"), agent_args=GivenArguments(()))
-    session = (await agent.new_session(str(old))).session_id
+    agent = canonical_agent(comms)
+    session = FieldCodec.decode(AcpSessionId, (await agent.new_session(str(old))).session_id)
     updates = []
 
     class Client:
@@ -37,24 +37,24 @@ async def publish_update(self, **kwargs):
             return True
 
     agent.on_connect(Client())
-    comms.threads.set_project(session, str(new))
+    comms.threads.set_project(comms.registry.require(session).incarnation, str(new))
     await agent.sessions.sync_identity(session)
     assert next(
         update.worktree
         for update in decode_updates(updates[-1].field_meta)
         if isinstance(update, CoordinationChangedUpdate)
-    ) == str(new)
+    ) == Checkout(new)
     await agent.shutdown()
-    resumed = canonical_agent(comms, agent_bin=GivenBinary("/bin/echo"), agent_args=GivenArguments(()))
+    resumed = canonical_agent(comms)
     try:
-        result = await resumed.load_session(str(old), session)
+        result = await resumed.load_session(str(old), FieldCodec.encode(session))
         assert next(
             update.worktree
             for update in decode_updates(result.field_meta)
             if isinstance(update, CoordinationChangedUpdate)
-        ) == str(new)
+        ) == Checkout(new)
         with pytest.raises(RequestError):
-            await resumed.load_session(str(tmp_path / "unrelated"), session)
+            await resumed.load_session(str(tmp_path / "unrelated"), FieldCodec.encode(session))
         assert len(comms.registry.all_threads()) == 1
     finally:
         await resumed.shutdown()
@@ -66,7 +66,7 @@ async def test_project_changes_reach_all_subscribed_clients(tmp_path, native_run
     old.mkdir()
     new.mkdir()
     comms = wire(tmp_path / "wire")
-    owner = canonical_agent(comms, agent_bin=GivenBinary("/bin/echo"), agent_args=GivenArguments(()), runtime_enabled=True)
+    owner = canonical_agent(comms, runtime_enabled=True)
     session = FieldCodec.decode(AcpSessionId, (await owner.new_session(str(old))).session_id)
     proxies, updates = [], [[], []]
 
@@ -88,13 +88,13 @@ async def publish_update(self, session_id, update):
                 update.worktree
                 for update in decode_updates(await proxy.subscribe())
                 if isinstance(update, CoordinationChangedUpdate)
-            ) == str(old)
-        comms.threads.set_project(session, str(new))
+            ) == Checkout(old)
+        comms.threads.set_project(comms.registry.require(session).incarnation, str(new))
         await owner.sessions.sync_identity(session)
         async with asyncio.timeout(2):
             while not all(
                 any(
-                    update.worktree == str(new)
+                    update.worktree == Checkout(new)
                     for value in values
                     for update in decode_updates(value.get("_meta"))
                     if isinstance(update, CoordinationChangedUpdate)
diff --git a/tests/test_project_resume.py b/tests/test_project_resume.py
index 75ac97cd5b..c3dbcaa365 100644
--- a/tests/test_project_resume.py
+++ b/tests/test_project_resume.py
@@ -31,14 +31,14 @@ def declared_owner(tmp_path):
 
 def test_project_change_while_idle_owes_no_turn(tmp_path):
     comms, _old, new = declared_owner(tmp_path)
-    assert comms.threads.set_project("owner", str(new)).changed
+    assert comms.threads.set_project(comms.registry.require("owner").incarnation, str(new)).changed
     assert not comms.registry.require("owner").project_resume_due
 
 
 def test_project_change_during_turn_is_recorded_until_the_next_turn_starts(tmp_path):
     comms, old, new = declared_owner(tmp_path)
     turn = comms.agents.begin_turn("owner", TurnId('1056fa0e005902e019b8d1d4fd75d3cb'))
-    comms.threads.set_project("owner", str(new))
+    comms.threads.set_project(comms.registry.require("owner").incarnation, str(new))
     assert comms.registry.require("owner").project_resume_due
     assert comms.agents.finish_turn(turn.turn_lease)
     # The turn ended and nothing ran afterwards: the registry still says so,
@@ -55,7 +55,7 @@ def test_project_change_during_turn_is_recorded_until_the_next_turn_starts(tmp_p
 def test_cancelled_turn_withdraws_the_recorded_resume(tmp_path):
     comms, _old, new = declared_owner(tmp_path)
     turn = comms.agents.begin_turn("owner", TurnId('1056fa0e005902e019b8d1d4fd75d3cb'))
-    comms.threads.set_project("owner", str(new))
+    comms.threads.set_project(comms.registry.require("owner").incarnation, str(new))
     comms.agents.transition_turn(turn.turn_lease, CancellingPhase())
     assert comms.agents.finish_turn(turn.turn_lease)
     current = comms.registry.require("owner")
@@ -75,12 +75,12 @@ async def test_message_after_project_change_starts_a_turn_in_the_new_project(tmp
 
             def reply(message):
                 return json.dumps(FieldCodec.encode((Message(
-                    session, message.exact_reply_target, f"Reply {len(sent)}",
+                    agent.sessions.require(session), message.exact_reply_target, f"Reply {len(sent)}",
                     InfoKind, timestamp=0,
                 ),)))
 
             def change_project_and_send_again():
-                changed = comms.threads.set_project(session, str(second))
+                changed = comms.threads.set_project(comms.registry.require(session).incarnation, str(second))
                 sent.append(comms.messaging.send("sender", session, "Second message").message)
                 return changed
 
diff --git a/tests/test_project_source_identity.py b/tests/test_project_source_identity.py
new file mode 100644
index 0000000000..db7f7cf43b
--- /dev/null
+++ b/tests/test_project_source_identity.py
@@ -0,0 +1,55 @@
+"""Project selection retains the caller's original registry identity."""
+from dataclasses import replace
+
+import pytest
+
+from agent_comms.comms import wire
+from agent_comms.errors import RelationViolationError
+from agent_comms.repo import Checkout
+from agent_comms.thread_text import GivenThreadText
+from agent_comms.threads import Thread
+
+
+@pytest.fixture
+def project_source(tmp_path):
+    current, destination = tmp_path / "current", tmp_path / "destination"
+    current.mkdir()
+    destination.mkdir()
+    comms = wire(tmp_path / "wire")
+    original = comms.registry.declare(Thread("source", frozenset(), Checkout(current)))
+    return comms, original, destination
+
+
+def test_captured_project_target_follows_its_original_rename(project_source):
+    comms, original, destination = project_source
+    comms.registry.rename(original.name, "renamed")
+    changed = comms.threads.set_project(original.incarnation, str(destination))
+    current = comms.registry.require("renamed")
+    assert changed.thread == current.name == "renamed"
+    assert current.incarnation == original.incarnation
+    assert current.worktree == Checkout(destination)
+    assert current.earlier_projects == (original.worktree,)
+
+
+def test_recreated_name_cannot_receive_the_original_project_change(project_source):
+    comms, original, destination = project_source
+    with comms.registry.store.editing() as edit:
+        edit.document.remove(original.name)
+        edit.commit()
+    replacement = comms.registry.declare(Thread(
+        original.name, frozenset(), original.worktree))
+    assert replacement.incarnation != original.incarnation
+    with pytest.raises(RelationViolationError):
+        comms.threads.set_project(original.incarnation, str(destination))
+    assert comms.registry.require(replacement.name) == replacement
+    assert not comms.registry.require(replacement.name).earlier_projects
+
+
+def test_same_birth_metadata_does_not_retire_a_project_target(project_source):
+    comms, original, destination = project_source
+    titled = comms.registry.declare(replace(original, title=GivenThreadText("new title")))
+    changed = comms.threads.set_project(original.incarnation, str(destination))
+    current = comms.registry.require(original.name)
+    assert changed.changed and current.worktree == Checkout(destination)
+    assert current.title == titled.title
+    assert current.incarnation == original.incarnation
diff --git a/tests/test_registration_inheritance.py b/tests/test_registration_inheritance.py
index 3befe58573..e192f8b105 100644
--- a/tests/test_registration_inheritance.py
+++ b/tests/test_registration_inheritance.py
@@ -93,9 +93,9 @@ def test_project_round_trip_excludes_current_project_from_history(tmp_path):
     first.mkdir()
     second.mkdir()
     comms = wire(tmp_path / 'wire')
-    comms.registry.declare(Thread('owner', frozenset(), Checkout(first)))
-    comms.threads.set_project('owner', str(second))
-    comms.threads.set_project('owner', str(first))
+    original = comms.registry.declare(Thread('owner', frozenset(), Checkout(first)))
+    comms.threads.set_project(original.incarnation, str(second))
+    comms.threads.set_project(original.incarnation, str(first))
     current = comms.registry.require('owner')
     assert current.worktree == Checkout(first)
     assert current.earlier_projects == (Checkout(second),)
@@ -109,7 +109,7 @@ def test_project_change_retains_executing_source_but_refuses_another_input(tmp_p
     comms.registry.declare(Thread('owner', frozenset(), Checkout(old), process_identity=ProcessIdentity.capture(os.getpid())))
     original = comms.agents.begin_turn('owner', TurnId('ab87f599dc0a50a56b7c65de71c39426'))
     try:
-        comms.threads.set_project('owner', str(new))
+        comms.threads.set_project(original.thread.incarnation, str(new))
         with pytest.raises(StaleFence):
             original.require_snapshot(comms.registry.snapshot(), 'New input refused')
         saved = str(old / 'native.jsonl')
diff --git a/tests/test_runtime.py b/tests/test_runtime.py
index 0a658750ec..c9acc0def6 100644
--- a/tests/test_runtime.py
+++ b/tests/test_runtime.py
@@ -250,7 +250,8 @@ async def publish_update(self, session_id, update):
         assert identity_position < replay_position
         assert len(snapshots) == 1
         session, snapshot = snapshots[0]
-        assert session == requested == snapshot.identity.requested_name
+        assert session == requested
+        assert snapshot.identity.requested_name == "renamed"
         assert snapshot.identity.thread.incarnation == comms.registry.require("renamed").incarnation
         assert [event.text for event in snapshot.page.events] == ["An original saved answer"]
     finally:
diff --git a/tests/test_session_load_plan.py b/tests/test_session_load_plan.py
index 77c79ad95f..b3a05a5461 100644
--- a/tests/test_session_load_plan.py
+++ b/tests/test_session_load_plan.py
@@ -1,4 +1,6 @@
-"""Public type names may change; the existing ACP metadata must not."""
+"""Original ACP load witnesses retain their wire contract and owner semantics."""
+
+import pytest
 
 from agent_comms.child_process import ProcessIdentity
 from agent_comms.nominal.field_codec import FieldCodec
@@ -9,7 +11,7 @@
     SessionLoadPlan,
 )
 from agent_comms.thread_identity import OwnerIdentity, RecordedThreadIncarnation
-from agent_comms.thread_presentation import LiveThreadOwnerBinding
+from agent_comms.thread_presentation import LiveThreadOwnerBinding, UnavailableThreadOwnerBinding
 
 
 def test_default_load_metadata_keeps_its_original_shape():
@@ -37,4 +39,50 @@ def test_witnessed_load_and_failure_metadata_keep_original_tags():
     assert error == {
         "agentCommsLoadFailure": {"kind": "failed", "binding": FieldCodec.encode(binding)}
     }
-    assert FailedSessionLoadPlan.from_failure(error) == failed
+    assert SessionLoadPlan.from_failure(error) == failed
+
+
+@pytest.mark.parametrize("plan_type", (ExistingSessionLoadPlan, FailedSessionLoadPlan))
+async def test_only_a_superseding_live_owner_supplies_an_existing_load(plan_type):
+    original = LiveThreadOwnerBinding(
+        OwnerIdentity(RecordedThreadIncarnation("original", 123.0), 2),
+        ProcessIdentity(1, 1),
+    )
+    previous = plan_type(original)
+    loads: list[SessionLoadPlan] = []
+
+    async def reload(plan: SessionLoadPlan) -> None:
+        loads.append(plan)
+
+    await UnavailableThreadOwnerBinding().reload_if_superseded(previous, reload)
+    await original.reload_if_superseded(previous, reload)
+    await LiveThreadOwnerBinding(
+        OwnerIdentity(original.owner.incarnation, 1), ProcessIdentity(2, 2),
+    ).reload_if_superseded(previous, reload)
+    await LiveThreadOwnerBinding(
+        OwnerIdentity(RecordedThreadIncarnation("replacement", 124.0), 3),
+        ProcessIdentity(2, 2),
+    ).reload_if_superseded(previous, reload)
+    assert not loads
+    renamed = original.owner.incarnation.renamed("renamed")
+    successor = LiveThreadOwnerBinding(OwnerIdentity(renamed, 3), ProcessIdentity(2, 2))
+    await successor.reload_if_superseded(previous, reload)
+    assert loads == [ExistingSessionLoadPlan(successor)]
+    assert previous.binding == original
+
+
+@pytest.mark.parametrize("metadata", (None, {}))
+async def test_an_ensuring_load_is_not_an_old_owner_witness(metadata):
+    binding = LiveThreadOwnerBinding(
+        OwnerIdentity(RecordedThreadIncarnation("source", 123.0), 2),
+        ProcessIdentity(1, 1),
+    )
+    loads: list[SessionLoadPlan] = []
+
+    async def reload(plan: SessionLoadPlan) -> None:
+        loads.append(plan)
+
+    plan = SessionLoadPlan.from_failure(metadata)
+    assert plan == EnsuringSessionLoadPlan()
+    await binding.reload_if_superseded(plan, reload)
+    assert not loads
diff --git a/tests/test_transcript_input_display.py b/tests/test_transcript_input_display.py
index 671f0e1d2e..64e5506565 100644
--- a/tests/test_transcript_input_display.py
+++ b/tests/test_transcript_input_display.py
@@ -14,8 +14,10 @@
 from agent_comms.comms import wire
 from conftest import installed_comms
 from agent_comms.native_entries import TranscriptProjection
+from agent_comms.nominal.field_codec import AcpSessionId
 from agent_comms.pi_payloads import PiMessage, UserMessage
 from agent_comms.threads import Thread
+from agent_comms.turn_item import FreshNativeIdText
 from agent_comms.transcript_routes import DisplayText, InputDisplay, NoDisplayText, SentInputText
 from bus_record_fixture import FIRST_INPUT_ID, SECOND_INPUT_ID, THIRD_INPUT_ID
 
@@ -31,7 +33,8 @@ def saved_row(native_id, text, *, images=False, role="user"):
     if images:
         content.append({"type": "image", "mimeType": "image/png", "data": "secret-image"})
     return transcript_records.message(
-        {"role": role, "inputId": str(native_id), "content": content}, id=native_id[:8]
+        {"role": role, "inputId": str(native_id), "content": content},
+        id=FreshNativeIdText.decode(native_id[:8]),
     )
 
 
@@ -109,14 +112,20 @@ async def test_acp_saved_transcript_replay_hides_only_owned_internal_input(tmp_p
     comms.transcripts.routes.record_input_display(SECOND_INPUT_ID, DisplayText("test2"))
     agent = CommsAgent(wire(comms.root))
     updates = []
+    destinations = []
+    destination = AcpSessionId("input-display-destination")
 
     class Client(SessionUpdateSink):
-        async def publish_update(self, **kwargs):
-            updates.append(kwargs["update"])
+        async def publish_update(self, *, session_id, update):
+            destinations.append(session_id)
+            updates.append(update)
             return True
 
     try:
-        await agent.sessions.transcript.replay("worker", client=Client())
+        await agent.sessions.transcript.replay(
+            "worker", session_id=destination, client=Client(),
+        )
+        assert destinations == [destination]
         facts = decode_updates(updates[0].field_meta)
         snapshot, = (fact for fact in facts if isinstance(fact, TranscriptSnapshotUpdate))
         events = snapshot.page.events