diff --git a/tests/support/core_adapters/mcp_turn_boundaries.py b/tests/support/core_adapters/mcp_turn_boundaries.py index 2388628a6..afddaa240 100644 --- a/tests/support/core_adapters/mcp_turn_boundaries.py +++ b/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) diff --git a/tests/support/peers/turn_source.py b/tests/support/peers/turn_source.py index dcb6c37f5..ed2f62659 100644 --- a/tests/support/peers/turn_source.py +++ b/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) diff --git a/tests/support/peers/typed_turn_events.py b/tests/support/peers/typed_turn_events.py index 4c0a4afc6..346389f39 100644 --- a/tests/support/peers/typed_turn_events.py +++ b/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__": diff --git a/tests/support/shared_stdio.py b/tests/support/shared_stdio.py index 98b94df7f..a51af95ed 100644 --- a/tests/support/shared_stdio.py +++ b/tests/support/shared_stdio.py @@ -20,6 +20,7 @@ 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: @@ -27,7 +28,7 @@ async def threads_owned_here(comms, project, names=("alpha", "beta")): 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()