test(streaming): shape the AgentCore and LangGraph fixtures like captures from a real runtime and server

tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py

@@ -107,19 +107,31 @@ def _gemini_frames(ending: Ending) -> tuple[bytes, ...]:
     return (*content, usage) if finished else content
 
 
+def _event(name: str, frame: JsonValue) -> bytes:
+    return f"event: {name}\n".encode() + _data(frame)
+
+
 def _langgraph_frames() -> tuple[bytes, ...]:
-    """LangGraph ends a run by closing the stream after its metadata event."""
+    """A LangGraph server's run stream in its default mode, its run metadata first and then the state after each step"""
+    question: Final = {"type": "human", "content": "Answer in two steps."}
+    answers: Final = tuple({"type": "ai", "content": part} for part in _PARTIAL)
     return (
-        *(_data(["messages", [[{"type": "ai", "content": part}, {}]]]) for part in _PARTIAL),
-        _data(["metadata", {"run_id": "run-1"}]),
+        _event("metadata", {"run_id": "run-1", "attempt": 1}),
+        *(_event("values", {"messages": [question, *answers[:step]]}) for step in range(len(answers) + 1)),
     )
 
 
 def _agentcore_frames(ending: Ending) -> tuple[bytes, ...]:
-    """A runtime that is not Strands can end its turn on messageStop with no metadata or final message."""
-    content: Final = tuple(_data({"event": {"contentBlockDelta": {"delta": {"text": part}}}}) for part in _PARTIAL)
-    message_stop: Final = _data({"event": {"messageStop": {"stopReason": "end_turn"}}})
-    return (*content, message_stop) if ending == "ending-complete" else content
+    """A runtime that forwards Bedrock Converse events without the metadata event, so its turn ends on messageStop"""
+    start: Final = _data({"event": {"messageStart": {"role": "assistant"}}})
+    content: Final = tuple(
+        _data({"event": {"contentBlockDelta": {"delta": {"text": part}, "contentBlockIndex": 0}}}) for part in _PARTIAL
+    )
+    end: Final = (
+        _data({"event": {"contentBlockStop": {"contentBlockIndex": 0}}}),
+        _data({"event": {"messageStop": {"stopReason": "end_turn"}}}),
+    )
+    return (start, *content, *end) if ending == "ending-complete" else (start, *content)
 
 
 def _frames(family: Family, identity: str, ending: Ending) -> tuple[bytes, ...]:

tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py

@@ -1277,8 +1277,9 @@ _AGENTCORE_RUNTIME_URL: Final = (
     "arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A888602223428%3Aruntime%2Fhosted_agent_13sf6-cALnp38iZD/invocations"
 )
 _AGENTCORE_DELTAS: Final = (
-    'data: {"event":{"contentBlockDelta":{"delta":{"text":"Machine learning lets "}}}}\n\n'
-    'data: {"event":{"contentBlockDelta":{"delta":{"text":"computers learn from data."}}}}\n\n'
+    'data: {"event":{"messageStart":{"role":"assistant"}}}\n\n'
+    'data: {"event":{"contentBlockDelta":{"delta":{"text":"Machine learning lets "},"contentBlockIndex":0}}}\n\n'
+    'data: {"event":{"contentBlockDelta":{"delta":{"text":"computers learn from data."},"contentBlockIndex":0}}}\n\n'
 )
 
 
@@ -1308,7 +1309,11 @@ async def _stream_agentcore(body: str, sync_mode: bool) -> list[tuple[str | None
 @pytest.mark.parametrize(
     "ending",
     [
-        pytest.param('data: {"event":{"messageStop":{"stopReason":"end_turn"}}}\n\n', id="message_stop"),
+        pytest.param(
+            'data: {"event":{"contentBlockStop":{"contentBlockIndex":0}}}\n\n'
+            'data: {"event":{"messageStop":{"stopReason":"end_turn"}}}\n\n',
+            id="message_stop",
+        ),
         pytest.param(
             'data: {"event":{"messageStop":{"stopReason":"end_turn"}}}\n\n'
             'data: {"event":{"metadata":{"usage":{"inputTokens":9,"outputTokens":7,"totalTokens":16}}}}\n\n'

tests/unit/llms/langgraph/chat/test_langgraph_transformation.py

@@ -271,11 +271,13 @@ async def test_langgraph_streaming_acompletion_without_a_client_posts_through_th
 
 
 @pytest.mark.asyncio
-async def test_langgraph_async_stream_delivers_every_message_event_before_its_stop():
+async def test_langgraph_async_stream_delivers_every_step_of_the_run_before_its_stop():
     async def events():
-        yield b'data: ["messages", [[{"type": "ai", "content": "25 * 4"}, {}]]]\n\n'
-        yield b'data: ["messages", [[{"type": "ai", "content": " is 100"}, {}]]]\n\n'
-        yield b'data: ["metadata", {"run_id": "run-1"}]\n\n'
+        question = {"type": "human", "content": "What is 25 * 4?"}
+        answers = [{"type": "ai", "content": "25 * 4"}, {"type": "ai", "content": " is 100"}]
+        yield b'event: metadata\ndata: {"run_id": "run-1", "attempt": 1}\n\n'
+        for step in range(len(answers) + 1):
+            yield b"event: values\ndata: " + json.dumps({"messages": [question, *answers[:step]]}).encode() + b"\n\n"
 
     client = AsyncHTTPHandler(transport=httpx.MockTransport(lambda request: httpx.Response(200, content=events())))