diff --git a/src/anthropic/lib/streaming/_beta_messages.py b/src/anthropic/lib/streaming/_beta_messages.py index c7d3d6c2c..48ef12089 100644 --- a/src/anthropic/lib/streaming/_beta_messages.py +++ b/src/anthropic/lib/streaming/_beta_messages.py @@ -546,9 +546,14 @@ def accumulate_event( if content_block.type == "text" and is_given(output_format): content_block.parsed_output = parse_text(content_block.text, output_format) elif event.type == "message_delta": - current_snapshot.stop_reason = event.delta.stop_reason - current_snapshot.stop_sequence = event.delta.stop_sequence - current_snapshot.stop_details = event.delta.stop_details + # optional delta fields are omitted when not applicable, so a later + # message_delta must not overwrite a value an earlier delta already set + if event.delta.stop_reason is not None: + current_snapshot.stop_reason = event.delta.stop_reason + if event.delta.stop_sequence is not None: + current_snapshot.stop_sequence = event.delta.stop_sequence + if event.delta.stop_details is not None: + current_snapshot.stop_details = event.delta.stop_details if event.delta.container is not None: current_snapshot.container = event.delta.container current_snapshot.usage.output_tokens = event.usage.output_tokens diff --git a/src/anthropic/lib/streaming/_messages.py b/src/anthropic/lib/streaming/_messages.py index d036a04d1..2307438e8 100644 --- a/src/anthropic/lib/streaming/_messages.py +++ b/src/anthropic/lib/streaming/_messages.py @@ -512,9 +512,14 @@ def accumulate_event( if content_block.type == "text" and is_given(output_format): content_block.parsed_output = parse_text(content_block.text, output_format) elif event.type == "message_delta": - current_snapshot.stop_reason = event.delta.stop_reason - current_snapshot.stop_sequence = event.delta.stop_sequence - current_snapshot.stop_details = event.delta.stop_details + # optional delta fields are omitted when not applicable, so a later + # message_delta must not overwrite a value an earlier delta already set + if event.delta.stop_reason is not None: + current_snapshot.stop_reason = event.delta.stop_reason + if event.delta.stop_sequence is not None: + current_snapshot.stop_sequence = event.delta.stop_sequence + if event.delta.stop_details is not None: + current_snapshot.stop_details = event.delta.stop_details if event.delta.container is not None: current_snapshot.container = event.delta.container current_snapshot.usage.output_tokens = event.usage.output_tokens diff --git a/tests/lib/streaming/test_beta_messages.py b/tests/lib/streaming/test_beta_messages.py index fa14382e9..34bd366ac 100644 --- a/tests/lib/streaming/test_beta_messages.py +++ b/tests/lib/streaming/test_beta_messages.py @@ -19,9 +19,15 @@ ) from anthropic.types.beta.beta_tool_param import BetaToolParam from anthropic.resources.messages.messages import DEPRECATED_MODELS -from anthropic.lib.streaming._beta_messages import TRACKS_TOOL_INPUT, BetaMessageStream, BetaAsyncMessageStream +from anthropic.lib.streaming._beta_messages import ( + TRACKS_TOOL_INPUT, + BetaMessageStream, + BetaAsyncMessageStream, + accumulate_event, +) from anthropic.types.beta.beta_message_delta_usage import BetaMessageDeltaUsage from anthropic.types.beta.beta_raw_message_delta_event import Delta as BetaRawMessageDelta, BetaRawMessageDeltaEvent +from anthropic.types.beta.beta_raw_message_start_event import BetaRawMessageStartEvent from .helpers import get_response, to_async_iter @@ -923,6 +929,73 @@ def test_message_delta_fields_are_all_accumulated() -> None: } +def test_later_message_delta_omitting_stop_fields_keeps_snapshot() -> None: + # a usage-only later message_delta must not reset stop_* a prior delta already set + snap = accumulate_event( + event=BetaRawMessageStartEvent.model_validate( + { + "type": "message_start", + "message": { + "id": "msg_1", + "type": "message", + "role": "assistant", + "model": "claude-haiku-4-5", + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 10, "output_tokens": 0}, + }, + } + ), + current_snapshot=None, + json_bufs={}, + request_headers=httpx2.Headers(), + ) + snap = accumulate_event( + event=BetaRawMessageDeltaEvent.model_validate( + { + "type": "message_delta", + "delta": { + "stop_reason": "end_turn", + "stop_sequence": "STOP", + "stop_details": { + "type": "refusal", + "category": "cyber", + "explanation": "policy", + }, + }, + "usage": {"output_tokens": 5}, + } + ), + current_snapshot=snap, + json_bufs={}, + request_headers=httpx2.Headers(), + ) + assert snap.stop_reason == "end_turn" + assert snap.stop_sequence == "STOP" + assert snap.stop_details is not None + assert snap.stop_details.type == "refusal" + assert snap.usage.output_tokens == 5 + + snap = accumulate_event( + event=BetaRawMessageDeltaEvent.model_validate( + { + "type": "message_delta", + "delta": {}, + "usage": {"output_tokens": 9}, + } + ), + current_snapshot=snap, + json_bufs={}, + request_headers=httpx2.Headers(), + ) + assert snap.stop_reason == "end_turn" + assert snap.stop_sequence == "STOP" + assert snap.stop_details is not None + assert snap.stop_details.category == "cyber" + assert snap.usage.output_tokens == 9 + + @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) def test_stream_method_definition_in_sync(sync: bool) -> None: client: Anthropic | AsyncAnthropic = sync_client if sync else async_client diff --git a/tests/lib/streaming/test_messages.py b/tests/lib/streaming/test_messages.py index e7360f789..d5748b8bb 100644 --- a/tests/lib/streaming/test_messages.py +++ b/tests/lib/streaming/test_messages.py @@ -14,9 +14,10 @@ from anthropic.lib.streaming import InputJsonEvent, ParsedMessageStreamEvent from anthropic.types.message import Message from anthropic.resources.messages import DEPRECATED_MODELS -from anthropic.lib.streaming._messages import TRACKS_TOOL_INPUT +from anthropic.lib.streaming._messages import TRACKS_TOOL_INPUT, accumulate_event from anthropic.types.message_delta_usage import MessageDeltaUsage from anthropic.types.raw_message_delta_event import Delta as RawMessageDelta, RawMessageDeltaEvent +from anthropic.types.raw_message_start_event import RawMessageStartEvent from .helpers import get_response, to_async_iter @@ -665,6 +666,70 @@ def test_message_delta_fields_are_all_accumulated() -> None: } +def test_later_message_delta_omitting_stop_fields_keeps_snapshot() -> None: + # a usage-only later message_delta must not reset stop_* a prior delta already set + snap = accumulate_event( + event=RawMessageStartEvent.model_validate( + { + "type": "message_start", + "message": { + "id": "msg_1", + "type": "message", + "role": "assistant", + "model": "claude-haiku-4-5", + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 10, "output_tokens": 0}, + }, + } + ), + current_snapshot=None, + json_bufs={}, + ) + snap = accumulate_event( + event=RawMessageDeltaEvent.model_validate( + { + "type": "message_delta", + "delta": { + "stop_reason": "end_turn", + "stop_sequence": "STOP", + "stop_details": { + "type": "refusal", + "category": "cyber", + "explanation": "policy", + }, + }, + "usage": {"output_tokens": 5}, + } + ), + current_snapshot=snap, + json_bufs={}, + ) + assert snap.stop_reason == "end_turn" + assert snap.stop_sequence == "STOP" + assert snap.stop_details is not None + assert snap.stop_details.type == "refusal" + assert snap.usage.output_tokens == 5 + + snap = accumulate_event( + event=RawMessageDeltaEvent.model_validate( + { + "type": "message_delta", + "delta": {}, + "usage": {"output_tokens": 9}, + } + ), + current_snapshot=snap, + json_bufs={}, + ) + assert snap.stop_reason == "end_turn" + assert snap.stop_sequence == "STOP" + assert snap.stop_details is not None + assert snap.stop_details.category == "cyber" + assert snap.usage.output_tokens == 9 + + @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) def test_stream_method_definition_in_sync(sync: bool) -> None: client: Anthropic | AsyncAnthropic = sync_client if sync else async_client