diff --git a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py index 024783aabfb..03c842ddc99 100644 --- a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py +++ b/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, ...]: diff --git a/tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py b/tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py index b89a56041f9..1b0c789f55d 100644 --- a/tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py +++ b/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' diff --git a/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py b/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py index 4db85bc8f58..806ade53080 100644 --- a/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py +++ b/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())))