diff --git a/src/anthropic/lib/streaming/_beta_messages.py b/src/anthropic/lib/streaming/_beta_messages.py index c7d3d6c2c..3a63e6baf 100644 --- a/src/anthropic/lib/streaming/_beta_messages.py +++ b/src/anthropic/lib/streaming/_beta_messages.py @@ -536,7 +536,8 @@ def accumulate_event( elif event.delta.type == "compaction_delta": if content.type == "compaction": content.content = event.delta.content - content.encrypted_content = event.delta.encrypted_content + if "encrypted_content" in event.delta.model_fields_set: + content.encrypted_content = event.delta.encrypted_content else: # we only want exhaustive checking for linters, not at runtime if TYPE_CHECKING: # type: ignore[unreachable] diff --git a/tests/lib/streaming/test_beta_messages.py b/tests/lib/streaming/test_beta_messages.py index fc2af491d..b090f10f9 100644 --- a/tests/lib/streaming/test_beta_messages.py +++ b/tests/lib/streaming/test_beta_messages.py @@ -3,6 +3,7 @@ import os import json from typing import Any, Set, Dict, List, Tuple, TypeVar, cast +from collections.abc import Iterator import httpx2 import pytest @@ -23,7 +24,7 @@ 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 .helpers import get_response, to_async_iter +from .helpers import get_response, load_fixture, to_async_iter base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010") api_key = "my-anthropic-api-key" @@ -385,6 +386,33 @@ def assert_input_transformations_response(message: BetaMessage, expected: List[D assert [entry.to_dict() for entry in message.input_transformations] == expected +def compaction_metadata_response(update: dict[str, str | None]) -> Iterator[bytes]: + response = load_fixture("compaction_response.txt") + delta = {"type": "compaction_delta", "content": "Updated summary.", **update} + event = {"type": "content_block_delta", "index": 0, "delta": delta} + response = response.replace( + "event: content_block_stop\n", + "event: content_block_delta\ndata: " + json.dumps(event) + "\n\nevent: content_block_stop\n", + 1, + ) + return iter([response.encode()]) + + +def assert_compaction_metadata( + events: list[ParsedBetaMessageStreamEvent], message: BetaMessage, expected: str | None +) -> None: + block = message.content[0] + assert block.type == "compaction" + assert block.content == "Updated summary." + assert block.encrypted_content == expected + snapshots = [event for event in events if isinstance(event, BetaCompactionEvent)] + assert len(snapshots) == 2 + assert snapshots[0].encrypted_content == "EpwBCioIDxgCEAEYASJALd_opaque_compaction_payload" + assert snapshots[1].encrypted_content == expected + serialized = message.to_dict() + assert cast(list[dict[str, object]], serialized["content"])[0]["encrypted_content"] == expected + + class TestSyncMessages: @pytest.mark.respx(base_url=base_url) def test_basic_response(self, respx_mock: MockRouter) -> None: @@ -569,6 +597,28 @@ def test_compaction(self, respx_mock: MockRouter) -> None: assert_compaction_response([event for event in stream], stream.get_final_message()) + @pytest.mark.respx(base_url=base_url) + @pytest.mark.parametrize( + "metadata,expected", + [ + ({}, "EpwBCioIDxgCEAEYASJALd_opaque_compaction_payload"), + ({"encrypted_content": "replacement"}, "replacement"), + ({"encrypted_content": ""}, ""), + ({"encrypted_content": None}, None), + ], + ids=["omitted", "replaced", "empty", "null"], + ) + def test_compaction_metadata_presence( + self, respx_mock: MockRouter, metadata: dict[str, str | None], expected: str | None + ) -> None: + respx_mock.post("/v1/messages").mock( + return_value=httpx2.Response(200, content=compaction_metadata_response(metadata)) + ) + with sync_client.beta.messages.stream( + max_tokens=1024, messages=[{"role": "user", "content": "Summarize"}], model="claude-opus-4-7" + ) as stream: + assert_compaction_metadata(list(stream), stream.get_final_message(), expected) + @pytest.mark.respx(base_url=base_url) def test_fallback_relabels_model(self, respx_mock: MockRouter) -> None: respx_mock.post("/v1/messages").mock( @@ -820,6 +870,29 @@ async def test_compaction(self, respx_mock: MockRouter) -> None: assert_compaction_response([event async for event in stream], await stream.get_final_message()) + @pytest.mark.asyncio + @pytest.mark.respx(base_url=base_url) + @pytest.mark.parametrize( + "metadata,expected", + [ + ({}, "EpwBCioIDxgCEAEYASJALd_opaque_compaction_payload"), + ({"encrypted_content": "replacement"}, "replacement"), + ({"encrypted_content": ""}, ""), + ({"encrypted_content": None}, None), + ], + ids=["omitted", "replaced", "empty", "null"], + ) + async def test_compaction_metadata_presence( + self, respx_mock: MockRouter, metadata: dict[str, str | None], expected: str | None + ) -> None: + respx_mock.post("/v1/messages").mock( + return_value=httpx2.Response(200, content=to_async_iter(compaction_metadata_response(metadata))) + ) + async with async_client.beta.messages.stream( + max_tokens=1024, messages=[{"role": "user", "content": "Summarize"}], model="claude-opus-4-7" + ) as stream: + assert_compaction_metadata([event async for event in stream], await stream.get_final_message(), expected) + @pytest.mark.asyncio @pytest.mark.respx(base_url=base_url) async def test_fallback_relabels_model(self, respx_mock: MockRouter) -> None: