diff --git a/tests/native/test_native_goal_bar.py b/tests/native/test_native_goal_bar.py
index fd7e48bad..3ddb5100d 100644
--- a/tests/native/test_native_goal_bar.py
+++ b/tests/native/test_native_goal_bar.py
@@ -1,21 +1,28 @@
 """The goal bar follows the real owner: standby, a long goal that scrolls, edits made elsewhere, pause/resume and clear."""
 
+import asyncio
+import json
 import os
-from contextlib import AsyncExitStack
 
-from agent_comms import agent_events as events
-from agent_comms.owned_turn import OwnedTurn
 from agent_comms.child_process import ProcessIdentity
 from agent_comms.goal_actions import (
-    ActiveGoalAction, ClearGoalAction, EditGoalAction, GoalPrecondition, OwnerInvocable, PausedGoalAction,
+    ActiveGoalAction,
+    ClearGoalAction,
+    EditGoalAction,
+    GoalPrecondition,
+    OwnerInvocable,
+    PausedGoalAction,
+    SetGoalAction,
     StandbyGoalAction,
 )
 from agent_comms.goal_presentation import IdleGoalExecution
 from agent_comms.goals import NoGoal
+from rich.cells import cell_len
+from support.native import ControlledProvider, native_session
+from support.native_steps import goal_on_real_owner
 from textual.containers import VerticalScroll
 from textual.widgets import Static
 
-from support.native_steps import goal_on_real_owner
 from toad.screens.goal_details import GoalDetails
 from toad.screens.goal_edit import GoalEdit
 from toad.widgets.goal_bar import GoalBar, GoalControl, StandbyPulse
@@ -44,7 +51,8 @@ def owner_edit(comms, session, text):
 
 async def test_standby_goal_pulses_and_its_long_text_scrolls_at_every_size(workspace, native_package):
     async with goal_on_real_owner(workspace, OBJECTIVE, size=(90, 35),
-                                  arrange=standing_by_for_a_peer(workspace)) as (ui, _agent, _owner, _session, _goal):
+                                  arrange=standing_by_for_a_peer(workspace)) as native:
+        ui = native.ui
         conversation = ui.conversation
         bar = conversation.query_one(GoalBar)
         await ui.until(lambda: "Standby" in str(bar.query_one(".goal-header", Static).render()),
@@ -76,9 +84,8 @@ async def test_standby_goal_pulses_and_its_long_text_scrolls_at_every_size(works
 
 
 async def test_edits_made_on_the_owner_reach_open_goal_screens_without_touching_a_draft(workspace, native_package):
-    async with goal_on_real_owner(workspace, OBJECTIVE, size=(90, 35)) as (
-            ui, _agent, owner, session, _goal):
-        comms = owner.turns.comms
+    async with goal_on_real_owner(workspace, OBJECTIVE, size=(90, 35)) as native:
+        ui, comms, session = native.ui, native.comms, native.thread
         app, pilot = ui.textual
         conversation = ui.conversation
         await pilot.click("#goal-history")
@@ -101,7 +108,8 @@ async def test_edits_made_on_the_owner_reach_open_goal_screens_without_touching_
 
 async def test_pause_resume_and_clear(workspace, native_package):
     async with goal_on_real_owner(workspace, OBJECTIVE, size=(90, 35),
-                                  arrange=standing_by_for_a_peer(workspace)) as (ui, _agent, _owner, _session, _goal):
+                                  arrange=standing_by_for_a_peer(workspace)) as native:
+        ui = native.ui
         conversation = ui.conversation
         bar = conversation.query_one(GoalBar)
         pulse = bar.query_one(StandbyPulse)
@@ -117,35 +125,73 @@ async def test_pause_resume_and_clear(workspace, native_package):
 
 
 async def test_failed_ledger_repaints_retry_then_active_without_rewriting_goal(workspace, native_package):
-    async with goal_on_real_owner(workspace, OBJECTIVE) as (
-            ui, _agent, owner, session, first):
+    (workspace.project / "proof.txt").write_text("A real tool reads this file.\n")
+
+    def reply(_request, number):
+        if number == 1:
+            return {"role": "assistant", "content": ""}, "stop"
+        if number == 2:
+            return {"role": "assistant", "tool_calls": [{
+                "index": 0, "id": "p448-read", "type": "function",
+                "function": {"name": "read", "arguments": json.dumps({"path": "proof.txt"})}}]}, "tool_calls"
+        provider.hold_next_request()
+        return {"role": "assistant", "content": "PRODUCTIVE_SUCCESSOR"}, "stop"
+
+    provider = ControlledProvider(reply=reply, request_budget=4)
+    async with native_session(workspace, provider=provider) as native:
+        ui, agent, comms, session = native.ui, native.agent, native.comms, native.thread
         app, pilot = ui.textual
         view = ui.conversation
         view.prompt.text = "Keep the draft across the failed turn"
         bar = view.query_one(GoalBar)
-        failed = OwnedTurn(owner.turns, session, session, "Original work", original_goal_id=first.id)
-        async with AsyncExitStack() as resources, AsyncExitStack() as permits:
-            assert await failed.acquire(resources, permits)
-            await failed.progress.done(events.Done("Controlled backend failure", False))
-        assert owner.turns.comms.registry.require(session).goal == first
-        await view.goal_observation.refresh()
-        await ui.until(lambda: "Blocked" in str(bar.header.render()), why="the failed ledger in the heading")
+        first = await agent.update_goal(SetGoalAction, OBJECTIVE)
+        await ui.until(lambda: "Blocked" in str(bar.header.render()), seconds=30,
+                       why="the unproductive native goal turn settles as failed")
+        assert comms.registry.require(session).goal == first
         await ui.until(lambda: str(bar.query_one("#goal-toggle").render()) == "Retry",
                        why="Retry from execution while the stored goal remains active")
         await pilot.click("#goal-history")
         await ui.until(lambda: isinstance(app.screen, GoalDetails), why="the open goal details")
         details = app.screen
-        successor = OwnedTurn(owner.turns, session, session, "Fresh productive input",
-                              original_goal_id=first.id, original_owner_input=True)
-        async with AsyncExitStack() as resources, AsyncExitStack() as permits:
-            assert await successor.acquire(resources, permits)
-            await successor.progress.tool_ended(events.ToolEnd("read", "read", True, "Source read"))
-            await successor.progress.done(events.Done("Productive successor", True))
-        assert owner.turns.comms.registry.require(session).goal == first
-        await view.goal_observation.refresh()
+        sending = asyncio.create_task(agent.send_prompt("Read proof.txt and report completion"))
+        try:
+            await ui.until(sending.done, seconds=30, why="the fresh productive native input finishes")
+            await sending
+        finally:
+            if not sending.done():
+                sending.cancel()
+            await asyncio.gather(sending, return_exceptions=True)
+        assert comms.registry.require(session).goal == first
         await ui.until(lambda: "Active" in str(details.query_one("#goal-current-heading", Static).render()),
                        why="the already-open details following productive settlement")
         await ui.press("escape")
         await ui.until(lambda: str(bar.query_one("#goal-toggle").render()) == "Pause",
                        why="the goal control following the ledger-only change")
         assert view.prompt.text == "Keep the draft across the failed turn"
+
+
+async def test_standby_goal_mention_opens_its_peer_and_returns_to_the_draft(workspace, native_package):
+    async with native_session(workspace) as native:
+        ui, comms = native.ui, native.comms
+        owner = comms.registry.require(native.thread)
+        comms.registry.declare(comms.registry.new_thread(
+            "peer", frozenset(), str(native.project), model=owner.model,
+            thinking_level=owner.thinking_level,
+        ))
+        await asyncio.to_thread(comms.owners.start, "peer")
+        goal = comms.goals.update_goal(native.thread, SetGoalAction(text="Coordinate with @peer"))
+        comms.goals.update_goal(native.thread, StandbyGoalAction(
+            wait_for=("peer",), expect=GoalPrecondition(goal_id=goal.id)))
+        original = ui.selected_tab
+        await ui.write_draft("Composer draft stays")
+        await ui.until_shows("Coordinate with @peer")
+        summary = ui.conversation.goal_bar.summary
+        offset = (cell_len(summary.goal_text.split("@peer")[0]), 0)
+        await ui.until(lambda: ui.app.screen.get_style_at(
+            summary.region.x + offset[0], summary.region.y,
+        ).meta.get("@click") is not None, why="the objective's bound goal mention")
+        assert await ui.pilot.click(summary, offset=offset)
+        await ui.until(lambda: ui.app.selected_session._comms_thread == "peer",
+                       why="the goal mention opens its peer")
+        await ui.select_tab(original)
+        assert ui.draft() == "Composer draft stays"
diff --git a/tests/native/test_native_goal_objective_edit.py b/tests/native/test_native_goal_objective_edit.py
index 83d6627ac..a7f01248c 100644
--- a/tests/native/test_native_goal_objective_edit.py
+++ b/tests/native/test_native_goal_objective_edit.py
@@ -2,15 +2,14 @@
 
 from agent_comms.acp_extension import GoalChangedUpdate
 from agent_comms.goal_actions import ActiveGoalAction, GoalPrecondition
+from support.native_steps import goal_on_real_owner
 from textual.widgets import Static
 
-from support.native_steps import goal_on_real_owner
 from toad.core.events import CommsUpdated
 from toad.screens.goal_details import GoalDetails
 from toad.screens.goal_edit import GoalEdit
 from toad.widgets.goal_text import GoalText
 
-
 OBJECTIVE = "Work on draft PR #17 and report remaining gaps. " + "Detailed acceptance criteria. " * 30
 
 
@@ -25,8 +24,9 @@ def goal_in_toad(workspace):
 
 
 async def test_objective_edits_save_to_the_owner_and_redraw_the_goal_bar(workspace, native_package):
-    async with goal_in_toad(workspace) as (ui, agent, owner, session, first):
-        comms = owner.turns.comms
+    async with goal_in_toad(workspace) as native:
+        ui, agent, comms, session = native.ui, native.agent, native.comms, native.thread
+        first = comms.registry.require(session).goal.require()
         app, pilot = ui.textual
         conversation = ui.conversation
         current = first
@@ -48,10 +48,10 @@ async def test_objective_edits_save_to_the_owner_and_redraw_the_goal_bar(workspa
             else:
                 await pilot.click("#goal-save")
             await ui.until(lambda: not isinstance(app.screen, GoalEdit), why="the editor to close")
-            await ui.until(lambda: comms.registry.require(session).goal.text == objective, why="the saved objective")
+            await ui.until(lambda expected=objective: comms.registry.require(session).goal.text == expected, why="the saved objective")
             current = comms.registry.require(session).goal
             assert current.id == first.id
-            await ui.until(lambda: conversation.goal_display.snapshot == current, why="the saved goal in the bar")
+            await ui.until(lambda expected=current: conversation.goal_display.snapshot == expected, why="the saved goal in the bar")
             summary = conversation.query_one(".goal-summary", GoalText)
             assert objective in str(summary.render()), str(summary.render())
             assert ui.shows(" ".join(objective.split()[:3])) and ui.shows(objective.split()[-1])
@@ -71,7 +71,8 @@ async def test_objective_edits_save_to_the_owner_and_redraw_the_goal_bar(workspa
 
 
 async def test_a_late_notice_of_an_older_goal_does_not_replace_the_current_one(workspace, native_package):
-    async with goal_in_toad(workspace) as (ui, agent, _owner, session, first):
+    async with goal_in_toad(workspace) as native:
+        ui, agent = native.ui, native.agent
         conversation = ui.conversation
         before = await agent.get_goal_snapshot()
         older = before.goal.require()
diff --git a/tests/native/test_native_goal_owner.py b/tests/native/test_native_goal_owner.py
index 322e7bd2f..f9f66ea37 100644
--- a/tests/native/test_native_goal_owner.py
+++ b/tests/native/test_native_goal_owner.py
@@ -1,57 +1,29 @@
-"""Toad's goal requests reach the real owner: setting a goal makes it launchable; an edit keeps its identity."""
+"""Goal requests reach the managed native owner through Toad's real controller."""
 
 import pytest
 from agent_comms.goal_actions import SetGoalAction
-from agent_comms.goal_attempts import GoalAttemptStore
+from support.native import native_session
 
-from support.native import owner_server, start_private_native_wire
-from toad.acp.transcript_reader import CoordinationTranscriptReader
 
+async def test_setting_a_goal_reaches_the_native_provider(workspace, native_package):
+    async with native_session(workspace) as native:
+        native.provider.hold_next_request()
+        goal = await native.agent.update_goal(SetGoalAction, "Finish the task")
+        assert goal.state.declared_name == "active"
+        await native.ui.until(native.provider.entered.is_set, seconds=30,
+                              why="the managed owner begins the goal")
+        assert native.comms.registry.require(native.thread).goal == goal
+        native.provider.release()
 
-class InProcessReader(CoordinationTranscriptReader):
-    """Asks the owner directly, as Core's observation service would."""
 
-    def __init__(self, comms):
-        super().__init__()
-        self.comms = comms
-
-    async def read(self, request):
-        return await request.ask(self.comms)
-
-
-async def test_setting_a_goal_makes_it_ready_to_launch(workspace, native_package, monkeypatch):
-    monkeypatch.setenv("AGENT_COMMS_AGENT_MODELS", "openrouter/fake")
-    comms = start_private_native_wire(workspace.roots.wire)
-    owner = owner_server(comms)
-    try:
-        session = (await owner.new_session(cwd=str(workspace.project))).session_id
-        agent = workspace.agents.detached_agent()
-        agent.coordination = comms.configuration.answer(session)
-        goal = await agent.update_goal(SetGoalAction, "Finish the task")
-        assert goal is not None and goal.state.declared_name == "active"
-        assert goal.text == "Finish the task"
-        generation = GoalAttemptStore(comms.root / "goal-private").snapshot(goal.id)
-        assert generation is not None and generation.lifecycle.ready
-        assert owner.turns.goals.open_goal_store().ready_grant(goal.id, generation.number)
-        assert comms.registry.require(session).goal == goal
-    finally:
-        await owner.shutdown()
-
-
-async def test_editing_a_goal_keeps_its_identity_and_refuses_a_stale_edit(workspace, native_package, monkeypatch):
-    monkeypatch.setenv("AGENT_COMMS_AGENT_MODELS", "openrouter/fake")
-    comms = start_private_native_wire(workspace.roots.wire)
-    owner = owner_server(comms)
-    try:
-        session = (await owner.new_session(cwd=str(workspace.project))).session_id
-        agent = workspace.agents.detached_agent()
-        agent.coordination = comms.configuration.answer(session)
-        agent.controller.transcripts = InProcessReader(comms)
+async def test_editing_a_goal_keeps_its_identity_and_refuses_a_stale_edit(workspace, native_package):
+    async with native_session(workspace) as native:
+        agent, comms, session = native.agent, native.comms, native.thread
         first = await agent.update_goal(SetGoalAction, "First objective")
         changed = await agent.edit_goal(first, "Revised objective")
         assert changed.id == first.id
         assert changed.revision == first.revision + 1
-        assert changed.state.declared_name == first.state.declared_name
+        assert changed.state == first.state
         assert comms.registry.require(session).goal == changed
         with pytest.raises(ValueError):
             await agent.edit_goal(first, "Stale overwrite")
@@ -62,5 +34,3 @@ async def test_editing_a_goal_keeps_its_identity_and_refuses_a_stale_edit(worksp
         assert not any(entry["after"] and entry["after"]["text"] == "Stale overwrite" for entry in history)
         snapshot = await agent.get_goal_snapshot()
         assert snapshot.goal == changed and snapshot.execution.goal_id == changed.id
-    finally:
-        await owner.shutdown()
diff --git a/tests/support/catalog.py b/tests/support/catalog.py
index f9758f0f8..07abfa1a9 100644
--- a/tests/support/catalog.py
+++ b/tests/support/catalog.py
@@ -38,6 +38,7 @@ def catalog_text() -> str:
     definition = {
         "identity": IDENTITY, "name": "A native catalog control", "short_name": "native-control",
         "protocol": "acp", "type": "coding", "author_name": "Private controlled test",
+        "accepts_comms_root": True,
         "description": "Local commands only",
         "help": "# Native catalog control\nPrivate controlled command data.",
     }
diff --git a/tests/support/core_adapters/human_input_observer.py b/tests/support/core_adapters/human_input_observer.py
index 29b7b010a..b319aa189 100644
--- a/tests/support/core_adapters/human_input_observer.py
+++ b/tests/support/core_adapters/human_input_observer.py
@@ -21,6 +21,7 @@
 from agent_comms.input_disposition import InputDispositions
 from agent_comms.native_entries import NativeEntry
 from agent_comms.retained_task_facts import HumanInputTaskFact
+from toad.agent import AgentBase
 from toad.agent_schema import AgentDefinition
 from toad.app import ToadApp
 
@@ -52,7 +53,7 @@ async def queue_followup(self, text):
         inputs = InputDispositions(self.comms.root / InputDispositions.filename)
         before = inputs.read()
         command = QueuePromptRequest(text, True)
-        await self.app.selected_session.conversation.agent.send_prompt(text, request=command)
+        await self.app.selected_session.conversation.agent_binding.send_prompt(text, request=command)
         await until(self.pilot, lambda: len(inputs.read().rows) == len(before.rows) + 1)
         row = inputs.read().lookup('acp:' + command.input_id)
         row.origin.require_human().author.require_registered(self.comms.registry.snapshot())
@@ -64,7 +65,7 @@ async def queue_followup(self, text):
     async def submit(self, text, *, image=None, while_running=None,
                      observation_seconds=30):
         view = self.app.selected_session.conversation
-        agent = view.agent
+        agent = view.agent_binding
         inputs = InputDispositions(self.comms.root / InputDispositions.filename)
         before = inputs.read()
         wire_sequence = self.comms.bus.log.latest_sequence()
@@ -174,6 +175,7 @@ def phase(name):
             'name': 'Original S2/S5 fixture', 'identity': 'original-s2-s5',
             'short_name': 's2-s5', 'protocol': 'acp',
             'run_command': {'*': shlex.join([sys.executable, '-m', 'agent_comms.acp'])},
+            'accepts_comms_root': True,
         })
         app = ToadApp(agent_data=definition, project_dir=subject.worktree,
                       agent_session_id=subject.name)
@@ -183,11 +185,11 @@ def phase(name):
             await until(pilot, lambda: app.selected_session is not None)
             await app.selected_session.wait_content_ready()
             view = app.selected_session.conversation
-            await until(pilot, lambda: view.agent is not None)
+            await until(pilot, lambda: isinstance(view.agent_binding, AgentBase))
             phase('actual_original_view_before_acp_attachment_settlement')
-            await until(pilot, view.agent.session.settled.is_set)
-            assert view.agent.session.connected
-            await until(pilot, lambda: view.agent.queue_attachment.scope.matches_session(view.agent.session_id))
+            await until(pilot, view.agent_binding.session.settled.is_set)
+            assert view.agent_binding.session.connected
+            await until(pilot, lambda: view.agent_binding.queue_attachment.scope.matches_session(view.agent_binding.session_id))
             observer = ActualS2Ingress(app, pilot, owner._comms, subject, output)
             phase('actual_acp_attached_original_queue_scope')
             observer.persist('same_original_owner_actual_toad_acp_attached')
diff --git a/tests/support/core_adapters/mcp_permission_observer.py b/tests/support/core_adapters/mcp_permission_observer.py
index a3defddd7..69fc606fd 100644
--- a/tests/support/core_adapters/mcp_permission_observer.py
+++ b/tests/support/core_adapters/mcp_permission_observer.py
@@ -165,7 +165,7 @@ async def open_observer(case: str, artifact_dir: Path) -> AsyncIterator[Observer
             agent = Agent(root / "project", AgentDefinition.decode(AGENT_DATA), SESSION_ID)
             observer = Observer(case, app, pilot, agent, session, root)
             try:
-                view.agent = agent
+                view.bind_agent(agent)
                 await exercise_boundaries(agent, view, pilot)
                 yield observer
                 assert len(observer.receipts) == 1
diff --git a/tests/support/native.py b/tests/support/native.py
index a7e65f8ad..15e58a5a5 100644
--- a/tests/support/native.py
+++ b/tests/support/native.py
@@ -264,12 +264,12 @@ async def native_session(workspace, *, provider: ControlledProvider | None = Non
     with provider.serving() as base_url:
         comms = start_private_native_wire(workspace.roots.wire)
         configure_provider(base_url, settings=settings)
-        comms.registry.declare(comms.registry.new_thread("beta", frozenset({"team"}), str(workspace.project),
+        comms.registry.declare(comms.registry.new_thread(NativeSession.thread, frozenset({"team"}), str(workspace.project),
                                       model=MODEL, thinking_level=OffThinkingLevel))
         try:
             if prepare is not None:
                 await prepare(comms, workspace.project, provider)
-            async with attached_app(workspace, comms, "beta", size=size, application=application,
+            async with attached_app(workspace, comms, NativeSession.thread, size=size, application=application,
                                     attached=attached) as (ui, agent):
                 yield NativeSession(ui, agent, comms, provider, workspace.project)
                 assert not provider.failures, provider.failures
@@ -284,13 +284,3 @@ async def native_session(workspace, *, provider: ControlledProvider | None = Non
 def without_provider(models: str = "test/model") -> None:
     """Configure native owners with a model nothing serves, for tests that must never ask one."""
     os.environ.update(AGENT_COMMS_AGENT_BIN="pi", AGENT_COMMS_AGENT_ARGS="", AGENT_COMMS_AGENT_MODELS=models)
-
-
-def owner_server(comms, *, provider: str = "openrouter", model: str = "fake"):
-    """Core's ACP server object in this process, able to start native owners on the private wire."""
-    from agent_comms.acp import CommsAgent
-    from agent_comms.native_arguments import GivenArguments
-
-    return CommsAgent(
-        comms, agent_args=GivenArguments(("--provider", provider, "--model", model)),
-        runtime_enabled=True, auto_wake=False)
diff --git a/tests/support/native_steps.py b/tests/support/native_steps.py
index dae4bee41..4d39148fd 100644
--- a/tests/support/native_steps.py
+++ b/tests/support/native_steps.py
@@ -133,34 +133,19 @@ async def channel_message_feedback(native, owner_tab: str) -> None:
 
 @asynccontextmanager
 async def goal_on_real_owner(workspace, objective: str, *, size=(110, 35), arrange=None):
-    """A real owner (Core's ACP server in this process, no model) with an active goal, shown in Toad
-    through an agent object bound to the owner's session.
+    """Arrange a goal before the existing native workflow starts its managed owner.
 
-    `arrange(comms, session, goal)` may change the goal on the owner before the view binds the agent.
-    Yields (ui, agent, Core owner, session, goal as first set).
+    Yields the existing NativeSession. The model uses ControlledProvider.
     """
-    import os
-
     from agent_comms.goal_actions import SetGoalAction
 
-    from support.native import owner_server, start_private_native_wire
-
-    os.environ["AGENT_COMMS_AGENT_MODELS"] = "openrouter/fake"
-    comms = start_private_native_wire(workspace.roots.wire)
-    owner = owner_server(comms)
-    try:
-        session = (await owner.new_session(cwd=str(workspace.project))).session_id
-        async with workspace.app(size=size) as ui:
-            agent = ui.detached_agent()
-            agent.coordination = comms.configuration.answer(session)
-            goal = await agent.update_goal(SetGoalAction, objective)
-            if arrange is not None:
-                arrange(comms, session, goal)
-            conversation = ui.conversation
-            conversation.bind_agent(agent)
-            agent.ready = True
-            await conversation.goal_observation.refresh()
-            await ui.step()
-            yield ui, agent, owner, session, goal
-    finally:
-        await owner.shutdown()
+    from support.native import NativeSession, native_session
+
+    async def prepare(comms, _project, _provider):
+        session = NativeSession.thread
+        goal = comms.goals.update_goal(session, SetGoalAction(text=objective))
+        if arrange is not None:
+            arrange(comms, session, goal)
+
+    async with native_session(workspace, prepare=prepare, size=size) as native:
+        yield native