Point stale queue and live-output tests at their owners

tests/textual/test_live_output_streams.py

@@ -3,12 +3,14 @@
 from agent_comms.routing import MessageRoute
 from support.agents import RecordedEvents
 
+from toad.acp.comms_updates import CommsUpdateConsumer
 from toad.core.events import CoreEvent, Update
 from toad.core_event_carrier import CoreEventMessage, dispatch_publication
-from toad.live_output import OutputStream, ResponseStream, ThoughtStream
-from toad.response_delivery import RoutedResponse, UnroutedResponse
+from toad.live_output import CompleteResponseStream, OutputStream, ResponseStream, ThoughtStream
+from toad.response_delivery import RoutedResponse
 from toad.widgets.agent_response import AgentResponse
 from toad.widgets.agent_thought import AgentThought
+from toad.widgets.committed_presentation import CapturedClaim
 
 
 class DiagnosticStream(OutputStream):
@@ -19,6 +21,11 @@ def create(self, fragment):
         return AgentThought(fragment)
 
 
+def consumed_stream(agent):
+    """The stream one incoming agent message gets from the update consumer that owns it."""
+    return CommsUpdateConsumer(agent, "session").response_stream()
+
+
 def painted_window(ui):
     window = ui.conversation.window.region
     return "\n".join(strip.crop(window.x, window.right).text
@@ -61,8 +68,8 @@ async def test_queued_client_replies_of_one_local_turn_form_one_reply(workspace)
         ui.attach_agent(agent)
         subscription = agent.controller.surface.subscription
         view.turns.start_client()
-        first = view.turns.owner.response_stream(UnroutedResponse())
-        second = view.turns.owner.response_stream(UnroutedResponse())
+        first = consumed_stream(agent)
+        second = consumed_stream(agent)
         view.turns.finish_client()
         await dispatch_publication(view, CoreEventMessage(Update("text", "Local queued", first), subscription))
         await dispatch_publication(view, CoreEventMessage(Update("text", " response", second), subscription))
@@ -71,14 +78,14 @@ async def test_queued_client_replies_of_one_local_turn_form_one_reply(workspace)
         assert local.body.text == "Local queued response"
 
 
-async def test_a_different_turn_id_or_owner_notice_starts_a_new_block(workspace):
+async def test_a_different_append_claim_or_notice_starts_a_new_block(workspace):
     async with workspace.app() as ui:
         view = ui.conversation
-        one = await view.output.append(ResponseStream(turn_id="one"), "First turn")
-        two = await view.output.append(ResponseStream(turn_id="two"), "Second turn")
+        one = await view.output.append(ResponseStream(source_claim=CapturedClaim()), "First reply")
+        two = await view.output.append(ResponseStream(source_claim=CapturedClaim()), "Second reply")
         assert one is not two
-        notice = await view.output.append(view.turns.owner.response_stream(UnroutedResponse()), "Owner notice")
-        another = await view.output.append(view.turns.owner.response_stream(UnroutedResponse()), "Next owner notice")
+        notice = await view.output.append(CompleteResponseStream(), "Owner notice")
+        another = await view.output.append(CompleteResponseStream(), "Next owner notice")
         assert notice is not another
 
 
@@ -89,7 +96,7 @@ async def test_an_update_published_by_another_agent_adds_no_block(workspace):
         ui.attach_agent(agent)
         subscription = agent.controller.surface.subscription
         view.turns.start_client()
-        queued = view.turns.owner.response_stream(UnroutedResponse())
+        queued = consumed_stream(agent)
         view.turns.finish_client()
         await dispatch_publication(view, CoreEventMessage(Update("text", "Local queued", queued), subscription))
         await view.output.finish(ResponseStream)

tests/textual/test_queue_request_results.py

@@ -8,7 +8,7 @@
 from agent_comms.acp_extension import (
     AvailableQueueProjection, QueueChangedUpdate, QueueItem, LiveQueueScope, encode_updates,
 )
-from agent_comms.field_codec import InputId
+from agent_comms.field_codec import AcpSessionId, InputId
 from agent_comms.child_process import ProcessIdentity
 from agent_comms.thread_identity import ClaimedIdentity, RecordedThreadIncarnation
 
@@ -20,7 +20,7 @@
 def response(session, next_owner=False):
     generation = 2 if next_owner else 1
     scope = LiveQueueScope(
-        session_id=session,
+        session_id=AcpSessionId(session),
         claimed_identity=ClaimedIdentity(RecordedThreadIncarnation(session, 1000.0), generation),
         owner=ProcessIdentity(1234, 1),
     )