Integrate the existing coupled submission sources

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

src/agent_comms/coordinator.py

@@ -8,25 +8,24 @@
 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,
     ReadNativeCoordinationSession,
     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]],

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

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

src/agent_comms/session_load.py

@@ -1,22 +1,23 @@
 """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:
     from .session_lifecycle import AttachedSessionLifecycle
     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)
@@ -32,17 +33,23 @@ def at_response(cls, metadata: object) -> 'SessionLoadPlan':
         return (EnsuringSessionLoadPlan() if witness is None
                 else FieldCodec.decode(ExistingSessionLoadPlan, witness))
 
-    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": ...
 
 
 @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 +60,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:
@@ -80,7 +94,7 @@ class FailedSessionLoadPlan(WitnessedSessionLoadPlan):
     """The original failed load command's witness; never an input permission."""
 
     @classmethod
-    def from_failure(cls, data: object) -> 'FailedSessionLoadPlan | None':
+    def from_failure(cls, data: object) -> FailedSessionLoadPlan | None:
         if not isinstance(data, dict):
             return None
         witness = data.get("agentCommsLoadFailure")

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

@@ -1,8 +1,8 @@
 """Thread presentation: declaration and persistence owners."""
 
 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
 
@@ -149,6 +149,12 @@
     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 +192,12 @@
     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.
@@ -292,3 +304,8 @@
                      name: str, sent_source_limit: int) -> ThreadPresentation:
         return self
 
+
+# Session-load annotations use the module namespace. Publish these declarations
+# before loading its coordinator dependencies, which consume our status classes.
+import agent_comms.session_load as session_load
+

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

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: