P958: retain nominal ACP session IDs through goal publication peers

tests/support/core_adapters/mcp_turn_boundaries.py

@@ -16,7 +16,7 @@
 )
 from agent_comms.comms import wire
 from agent_comms.coordinator import Coordination
-from agent_comms.field_codec import TurnId
+from agent_comms.field_codec import AcpSessionId, FieldCodec, TurnId
 from agent_comms.launch_runtime import ThreadLaunch
 from agent_comms.pi_payloads import (
     McpLiveAcknowledgement, McpServerReceipt, ReadyMcpServerState, ConfirmMcpCallPolicy,
@@ -41,7 +41,7 @@ def notes(view):
 
 
 @asynccontextmanager
-async def turn_source(root: Path, session_id: str):
+async def turn_source(root: Path, session_id: AcpSessionId):
     """Use the original declaration, attachment and retirement owners once."""
     # This authored owner has its own registry. Inherited public caller names
     # must not select a different participant during same-process retirement.
@@ -87,7 +87,7 @@ async def send(*facts, session=None):
         assert result is None, result
         await pilot.pause()
 
-    session_id = agent.session_id
+    session_id = FieldCodec.decode(AcpSessionId, agent.session_id)
     source_root = agent.project_root_path.parent / "mcp-boundaries"
     async with turn_source(source_root, session_id) as producer:
         name = producer.sessions.require(session_id)

tests/support/peers/turn_source.py

@@ -9,12 +9,13 @@
 from agent_comms.acp import CommsAgent
 from agent_comms.comms import wire
 from agent_comms.coordinator import Coordination
+from agent_comms.field_codec import AcpSessionId
 from agent_comms.launch_root import LaunchRoot
 from agent_comms.launch_runtime import ThreadLaunch
 
 
 @asynccontextmanager
-async def turn_source(wire_root: Path, project: Path, session_id: str):
+async def turn_source(wire_root: Path, project: Path, session_id: AcpSessionId):
     """Retain declaration and attachment custody without sending native input."""
     with patch.dict(os.environ), LaunchRoot.from_path(wire_root).selected():
         ThreadLaunch.strip(os.environ)

tests/support/peers/typed_turn_events.py

@@ -16,10 +16,10 @@
 from agent_comms.agent_event_updates import AcpEventConsumer
 from agent_comms.controller import SessionUpdateSink
 from agent_comms.coordinator import Coordination
-from agent_comms.field_codec import TurnId
+from agent_comms.field_codec import AcpSessionId, TurnId
 from turn_source import turn_source
 
-SESSION = "pilot"
+SESSION = AcpSessionId("pilot")
 
 
 class Pipe(SessionUpdateSink):
@@ -53,7 +53,7 @@ async def produce(wire_root: Path, project: Path):
             assert await Coordination.run_worker(partial(producer.turns.current_turn_update, SESSION)) == before
             # An earlier finished cut: the current binding must refuse it.
             await consumer.dispatch(retired)
-        await producer.turns.goals.sync_goal_execution(SESSION, name)
+        await producer.turns.goals.sync_goal_execution(SESSION)
 
 
 if __name__ == "__main__":

tests/support/shared_stdio.py

@@ -20,14 +20,15 @@ def stdio_agent_definition():
 async def threads_owned_here(comms, project, names=("alpha", "beta")):
     """Declare `names` as threads of `project`, owned by this process through Core's ACP agent."""
     from agent_comms.acp import CommsAgent
+    from agent_comms.field_codec import AcpSessionId
 
     owner = CommsAgent(comms, runtime_enabled=True, auto_wake=False)
     try:
         for name in names:
             comms.threads.claim_thread(name, tags=frozenset({"acp"}), worktree=str(project))
             comms.registry.restore_stopped(comms.registry.snapshot(), (name,))
             thread = comms.owners.acquire_thread(name)
-            await owner.sessions.bind_owned(thread, name)
+            await owner.sessions.bind_owned(thread, AcpSessionId(name))
         yield owner
     finally:
         await owner.shutdown()