Integrate captured submission targets and session reload ownership

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
 
 

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)
             )

src/agent_comms/session_lifecycle.py

@@ -245,7 +245,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),
@@ -351,7 +353,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),

src/agent_comms/session_load.py

@@ -1,5 +1,6 @@
 """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 dataclasses import dataclass
@@ -35,6 +36,12 @@
     def superseded_by(self, binding: ThreadOwnerBinding) -> bool:
         return False
 
+    async def reload_if_superseded(
+        self, binding: 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": ...
 
@@ -58,6 +65,13 @@
     def superseded_by(self, binding: ThreadOwnerBinding) -> bool:
         return self.binding.superseded_by(binding)
 
+    async def reload_if_superseded(
+        self, binding: LiveThreadOwnerBinding,
+        reload: Callable[[SessionLoadPlan], Awaitable[None]],
+    ) -> None:
+        if self.superseded_by(binding):
+            await reload(ExistingSessionLoadPlan(binding))
+
 
 @dataclass(frozen=True, slots=True)
 class ExistingSessionLoadPlan(WitnessedSessionLoadPlan):

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)
 

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
 
@@ -16,6 +16,7 @@
 from .message_reference import MessageReference
 
 if TYPE_CHECKING:
+    from .session_load import SessionLoadPlan
     from .presentation import MessageNotification
     from .transcripts import AcquiredThreadPresentation, TranscriptReadIdentity
     from agent_comms.registry_document import RegistrySnapshot
@@ -149,6 +150,11 @@ def replaces(self, original: LiveThreadOwnerBinding) -> bool:
     def superseded_by(self, replacement: ThreadOwnerBinding) -> bool:
         return False
 
+    async def reload_if_superseded(
+        self, previous: SessionLoadPlan, reload: Callable[[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 +192,11 @@ 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: SessionLoadPlan, reload: Callable[[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.

src/agent_comms/tools.py

@@ -460,7 +460,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,
             (

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:

tests/test_acp.py

@@ -167,8 +167,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)
@@ -194,17 +195,30 @@ async def test_snapshot_capability_replays_one_bounded_update(
         if renamed:
             agent._comms.threads.rename_managed_thread("proj", "renamed", owner=ProcessIdentity.capture(os.getpid()))
         sent = []
+        destinations = []
+        destination = AcpSessionId("snapshot-destination")
+        thread = agent._comms.registry.require("proj")
 
         class Client(SessionUpdateSink):
-            async def publish_update(self, **kwargs):
-                sent.append(kwargs["update"])
+            async def publish_update(self, *, session_id, update):
+                destinations.append(session_id)
+                sent.append(update)
                 return True
 
         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 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

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()

tests/test_checkout_submodules.py

@@ -478,7 +478,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)
@@ -506,8 +506,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)
@@ -539,8 +539,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())

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
 
 
 
@@ -29,8 +29,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:
@@ -39,24 +39,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()
@@ -70,7 +70,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 = [], [[], []]
 
@@ -92,13 +92,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)