Skip to content

fix(streaming): keep stop_* when a later message_delta omits them #1941

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tsumon wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(streaming): keep stop_* when a later message_delta omits them #1941

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 8 additions & 3 deletions src/anthropic/lib/streaming/_beta_messages.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
11 changes: 8 additions & 3 deletions src/anthropic/lib/streaming/_messages.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
75 changes: 74 additions & 1 deletion tests/lib/streaming/test_beta_messages.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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
Expand Down
67 changes: 66 additions & 1 deletion tests/lib/streaming/test_messages.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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
Expand Down