From b579f52a469c190f603e23aced6d849b75147c53 Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Thu, 30 Jul 2026 16:38:33 +0200 Subject: [PATCH 1/4] fix(ai): preserve Anthropic messages stream API --- .sampo/changesets/roguish-baron-tuonetar.md | 5 + posthog/ai/anthropic/anthropic.py | 54 +++++++-- posthog/ai/anthropic/anthropic_async.py | 53 ++++++-- posthog/ai/stream.py | 50 +++++++- posthog/test/ai/anthropic/test_anthropic.py | 126 +++++++++++++++++++- 5 files changed, 265 insertions(+), 23 deletions(-) create mode 100644 .sampo/changesets/roguish-baron-tuonetar.md diff --git a/.sampo/changesets/roguish-baron-tuonetar.md b/.sampo/changesets/roguish-baron-tuonetar.md new file mode 100644 index 00000000..097c6943 --- /dev/null +++ b/.sampo/changesets/roguish-baron-tuonetar.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: patch +--- + +Preserve Anthropic messages.stream compatibility diff --git a/posthog/ai/anthropic/anthropic.py b/posthog/ai/anthropic/anthropic.py index a5a2299b..b8119d95 100644 --- a/posthog/ai/anthropic/anthropic.py +++ b/posthog/ai/anthropic/anthropic.py @@ -10,6 +10,7 @@ import uuid from typing import Any, Dict, List, Optional +from posthog.ai.stream import StreamWrapper from posthog.ai.types import StreamingContentBlock, TokenUsage, ToolInProgress from posthog.ai.utils import ( call_llm_and_track_usage, @@ -115,37 +116,74 @@ def stream( **kwargs: Arguments passed to Anthropic's ``messages.create`` API. Returns: - A streaming iterator yielding Anthropic events. + Anthropic's native streaming context manager. """ if posthog_trace_id is None: posthog_trace_id = str(uuid.uuid4()) - return self._create_streaming( + # Construct the provider resource directly so older Anthropic versions, + # whose stream manager delegates through ``self.create(stream=True)``, + # cannot re-enter our tracked ``create`` override. + manager = Messages(self._client).stream(**kwargs) + request_attribute = "_MessageStreamManager__api_request" + request = getattr(manager, request_attribute) + + def tracked_request(): + start_time = time.time() + response = request() + return self._track_streaming_response( + response, + posthog_distinct_id, + posthog_trace_id, + posthog_properties, + posthog_privacy_mode, + posthog_groups, + kwargs, + start_time, + ) + + setattr(manager, request_attribute, tracked_request) + return manager + + def _create_streaming( + self, + posthog_distinct_id: Optional[str], + posthog_trace_id: Optional[str], + posthog_properties: Optional[Dict[str, Any]], + posthog_privacy_mode: bool, + posthog_groups: Optional[Dict[str, Any]], + **kwargs: Any, + ): + start_time = time.time() + response = super().create(**kwargs) + return self._track_streaming_response( + response, posthog_distinct_id, posthog_trace_id, posthog_properties, posthog_privacy_mode, posthog_groups, - **kwargs, + kwargs, + start_time, ) - def _create_streaming( + def _track_streaming_response( self, + response: Any, posthog_distinct_id: Optional[str], posthog_trace_id: Optional[str], posthog_properties: Optional[Dict[str, Any]], posthog_privacy_mode: bool, posthog_groups: Optional[Dict[str, Any]], - **kwargs: Any, + kwargs: Dict[str, Any], + start_time: float, ): - start_time = time.time() usage_stats: TokenUsage = TokenUsage(input_tokens=0, output_tokens=0) accumulated_content = "" content_blocks: List[StreamingContentBlock] = [] tools_in_progress: Dict[str, ToolInProgress] = {} current_text_block: Optional[StreamingContentBlock] = None stop_reason: Optional[str] = None - response = super().create(**kwargs) def generator(): nonlocal usage_stats @@ -224,7 +262,7 @@ def generator(): stop_reason=stop_reason, ) - return generator() + return StreamWrapper(generator(), stream=response) def _capture_streaming_event( self, diff --git a/posthog/ai/anthropic/anthropic_async.py b/posthog/ai/anthropic/anthropic_async.py index 8ea07840..fb5eb075 100644 --- a/posthog/ai/anthropic/anthropic_async.py +++ b/posthog/ai/anthropic/anthropic_async.py @@ -95,7 +95,7 @@ async def create( **kwargs, ) - async def stream( + def stream( self, posthog_distinct_id: Optional[str] = None, posthog_trace_id: Optional[str] = None, @@ -116,37 +116,74 @@ async def stream( **kwargs: Arguments passed to Anthropic's async ``messages.create`` API. Returns: - An async streaming iterator yielding Anthropic events. + Anthropic's native async streaming context manager, without awaiting. """ if posthog_trace_id is None: posthog_trace_id = str(uuid.uuid4()) - return await self._create_streaming( + # Construct the provider resource directly so older Anthropic versions, + # whose stream manager delegates through ``self.create(stream=True)``, + # cannot re-enter our tracked ``create`` override. + manager = AsyncMessages(self._client).stream(**kwargs) + request_attribute = "_AsyncMessageStreamManager__api_request" + request = getattr(manager, request_attribute) + + async def tracked_request(): + start_time = time.time() + response = await request + return self._track_streaming_response( + response, + posthog_distinct_id, + posthog_trace_id, + posthog_properties, + posthog_privacy_mode, + posthog_groups, + kwargs, + start_time, + ) + + setattr(manager, request_attribute, tracked_request()) + return manager + + async def _create_streaming( + self, + posthog_distinct_id: Optional[str], + posthog_trace_id: Optional[str], + posthog_properties: Optional[Dict[str, Any]], + posthog_privacy_mode: bool, + posthog_groups: Optional[Dict[str, Any]], + **kwargs: Any, + ): + start_time = time.time() + response = await super().create(**kwargs) + return self._track_streaming_response( + response, posthog_distinct_id, posthog_trace_id, posthog_properties, posthog_privacy_mode, posthog_groups, - **kwargs, + kwargs, + start_time, ) - async def _create_streaming( + def _track_streaming_response( self, + response: Any, posthog_distinct_id: Optional[str], posthog_trace_id: Optional[str], posthog_properties: Optional[Dict[str, Any]], posthog_privacy_mode: bool, posthog_groups: Optional[Dict[str, Any]], - **kwargs: Any, + kwargs: Dict[str, Any], + start_time: float, ): - start_time = time.time() usage_stats: TokenUsage = TokenUsage(input_tokens=0, output_tokens=0) accumulated_content = "" content_blocks: List[StreamingContentBlock] = [] tools_in_progress: Dict[str, ToolInProgress] = {} current_text_block: Optional[StreamingContentBlock] = None stop_reason: Optional[str] = None - response = await super().create(**kwargs) async def generator(): nonlocal usage_stats diff --git a/posthog/ai/stream.py b/posthog/ai/stream.py index 4ed8ca94..e988cee1 100644 --- a/posthog/ai/stream.py +++ b/posthog/ai/stream.py @@ -1,10 +1,47 @@ """Shared async streaming utilities for PostHog AI wrappers.""" -from typing import Any, AsyncGenerator, Generic, Optional, TypeVar +from typing import Any, AsyncGenerator, Generator, Generic, Optional, TypeVar T = TypeVar("T") +class StreamWrapper(Generic[T]): + """Preserves a provider stream's helpers while tracking its iteration.""" + + def __init__( + self, + generator: Generator[T, None, None], + stream: Any, + ) -> None: + self._generator = generator + self._stream = stream + + def __iter__(self) -> "StreamWrapper[T]": + return self + + def __next__(self) -> T: + return next(self._generator) + + def __enter__(self) -> "StreamWrapper[T]": + return self + + def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: + self.close() + + def close(self) -> None: + try: + self._generator.close() + finally: + self._stream.close() + + def __getattr__(self, name: str) -> Any: + if name.startswith("_"): + raise AttributeError(name) + if name in ("send", "throw"): + return getattr(self._generator, name) + return getattr(self._stream, name) + + class AsyncStreamWrapper(Generic[T]): """Adds the async context manager protocol to a PostHog streaming generator. @@ -34,6 +71,10 @@ async def __aenter__(self) -> "AsyncStreamWrapper[T]": return self async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> bool: + await self.close() + return False + + async def close(self) -> None: # Close the generator first so its `finally` captures the event, even on # early exit. try/finally still closes the provider stream if that raises. try: @@ -46,11 +87,10 @@ async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> bool: if close is not None: await close() - return False + async def aclose(self) -> None: + await self.close() - # aclose/asend/athrow belong to the generator; provider streams expose - # close(), not these. Forwarding aclose() keeps it firing the event. - _GENERATOR_METHODS = ("aclose", "asend", "athrow") + _GENERATOR_METHODS = ("asend", "athrow") def __getattr__(self, name: str) -> Any: # Proxy only public attributes (e.g. `.response`) to the provider stream. diff --git a/posthog/test/ai/anthropic/test_anthropic.py b/posthog/test/ai/anthropic/test_anthropic.py index 62cdc24d..8c44c28c 100644 --- a/posthog/test/ai/anthropic/test_anthropic.py +++ b/posthog/test/ai/anthropic/test_anthropic.py @@ -1,13 +1,32 @@ +import inspect import json import os -from unittest.mock import patch +from unittest.mock import AsyncMock, Mock, patch import pytest from posthog import identify_context, new_context try: - from anthropic.types import CacheCreation, Message, Usage + from anthropic.lib.streaming._messages import ( + AsyncMessageStreamManager, + MessageStreamManager, + ) + from anthropic.types import ( + CacheCreation, + Message, + MessageDeltaUsage, + RawContentBlockDeltaEvent, + RawContentBlockStartEvent, + RawContentBlockStopEvent, + RawMessageDeltaEvent, + RawMessageStartEvent, + RawMessageStopEvent, + TextBlock, + TextDelta, + Usage, + ) + from anthropic.types.raw_message_delta_event import Delta from posthog.ai.anthropic import Anthropic, AnthropicBedrock, AsyncAnthropic from posthog.test.ai.utils import RecordingAsyncStream @@ -1744,6 +1763,22 @@ def test_integration_stop_reason(mock_client): assert props["$ai_input_tokens"] > 0 +class RecordingStream: + def __init__(self, items): + self._items = iter(items) + self.closed = False + self.response = "provider-response" + + def __iter__(self): + return self + + def __next__(self): + return next(self._items) + + def close(self): + self.closed = True + + def _anthropic_stream_events(): final = MockStreamEvent("message_delta") final.usage = MockUsage( @@ -1759,6 +1794,93 @@ def _anthropic_stream_events(): ] +def _anthropic_raw_stream_events(): + message = Message( + id="message-id", + type="message", + role="assistant", + content=[], + model="claude-3-opus-20240229", + usage=Usage(input_tokens=10, output_tokens=0), + stop_reason=None, + stop_sequence=None, + ) + return [ + RawMessageStartEvent(type="message_start", message=message), + RawContentBlockStartEvent( + type="content_block_start", + index=0, + content_block=TextBlock(type="text", text=""), + ), + RawContentBlockDeltaEvent( + type="content_block_delta", + index=0, + delta=TextDelta(type="text_delta", text="Hi"), + ), + RawContentBlockStopEvent(type="content_block_stop", index=0), + RawMessageDeltaEvent( + type="message_delta", + delta=Delta(stop_reason="end_turn", stop_sequence=None), + usage=MessageDeltaUsage(output_tokens=5), + ), + RawMessageStopEvent(type="message_stop"), + ] + + +def test_messages_stream_preserves_native_manager_helpers_close_and_tracking( + mock_client, +): + source = RecordingStream(_anthropic_raw_stream_events()) + client = Anthropic(api_key="test-key", posthog_client=mock_client) + client.post = Mock(return_value=source) + + response = client.messages.stream( + model="claude-haiku-4-5", + messages=[{"role": "user", "content": "Foo"}], + max_tokens=1, + posthog_distinct_id="test-user", + ) + + assert isinstance(response, MessageStreamManager) + with response as stream: + assert stream.response == "provider-response" + text = list(stream.text_stream) + + assert text == ["Hi"] + assert source.closed is True + assert mock_client.capture.call_count == 1 + assert mock_client.capture.call_args.kwargs["distinct_id"] == "test-user" + assert "posthog_distinct_id" not in client.post.call_args.kwargs + + +@pytest.mark.asyncio +async def test_async_messages_stream_preserves_provider_contract_and_manager( + mock_client, +): + source = RecordingAsyncStream(_anthropic_raw_stream_events()) + client = AsyncAnthropic(posthog_client=mock_client) + client.post = AsyncMock(return_value=source) + + response = client.messages.stream( + model="claude-haiku-4-5", + messages=[{"role": "user", "content": "Foo"}], + max_tokens=1, + posthog_distinct_id="test-user", + ) + + assert isinstance(response, AsyncMessageStreamManager) + assert not inspect.isawaitable(response) + async with response as stream: + assert stream.response == "provider-response" + text = [chunk async for chunk in stream.text_stream] + + assert text == ["Hi"] + assert source.closed is True + assert mock_client.capture.call_count == 1 + assert mock_client.capture.call_args.kwargs["distinct_id"] == "test-user" + assert "posthog_distinct_id" not in client.post.call_args.kwargs + + @pytest.mark.asyncio async def test_async_messages_create_streaming_supports_async_with(mock_client): """Regression test for #393: messages.create(stream=True) must support From 2db2ed392c177b75c5e0498c22646ec2b8783865 Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Thu, 30 Jul 2026 18:42:16 +0200 Subject: [PATCH 2/4] chore: update public API snapshot --- references/public_api_snapshot.txt | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt index d83fbb74..6573a31b 100644 --- a/references/public_api_snapshot.txt +++ b/references/public_api_snapshot.txt @@ -37,6 +37,7 @@ alias posthog.ai.anthropic.AsyncAnthropic -> posthog.ai.anthropic.anthropic_asyn alias posthog.ai.anthropic.AsyncAnthropicBedrock -> posthog.ai.anthropic.anthropic_providers.AsyncAnthropicBedrock alias posthog.ai.anthropic.AsyncAnthropicVertex -> posthog.ai.anthropic.anthropic_providers.AsyncAnthropicVertex alias posthog.ai.anthropic.anthropic.PostHogClient -> posthog.client.Client +alias posthog.ai.anthropic.anthropic.StreamWrapper -> posthog.ai.stream.StreamWrapper alias posthog.ai.anthropic.anthropic.StreamingContentBlock -> posthog.ai.types.StreamingContentBlock alias posthog.ai.anthropic.anthropic.TokenUsage -> posthog.ai.types.TokenUsage alias posthog.ai.anthropic.anthropic.ToolInProgress -> posthog.ai.types.ToolInProgress @@ -893,6 +894,7 @@ class posthog.ai.prompts.CachedPrompt(prompt: str, fetched_at: float, name: str, class posthog.ai.prompts.PromptResult(source: PromptSource, prompt: str, name: Optional[str] = None, version: Optional[int] = None, label: Optional[str] = None, config: Optional[Dict[str, Any]] = None) class posthog.ai.prompts.Prompts(posthog: Optional[Any] = None, *, personal_api_key: Optional[str] = None, project_api_key: Optional[str] = None, host: Optional[str] = None, default_cache_ttl_seconds: Optional[int] = None, capture_errors: bool = False) class posthog.ai.stream.AsyncStreamWrapper(generator: AsyncGenerator[T, None], stream: Optional[Any] = None) +class posthog.ai.stream.StreamWrapper(generator: Generator[T, None, None], stream: Any) class posthog.ai.types.FormattedFunctionCall class posthog.ai.types.FormattedImageContent class posthog.ai.types.FormattedMessage @@ -1228,6 +1230,9 @@ method posthog.ai.otel.processor.PostHogSpanProcessor.shutdown() -> None method posthog.ai.prompts.Prompts.clear_cache(name: Optional[str] = None, *, version: Optional[int] = None) -> None method posthog.ai.prompts.Prompts.compile(prompt: str, variables: PromptVariables) -> str method posthog.ai.prompts.Prompts.get(name: str, *, with_metadata: Optional[bool] = None, cache_ttl_seconds: Optional[int] = None, fallback: Optional[str] = None, version: Optional[int] = None, label: Optional[str] = None) -> Union[str, PromptResult] +method posthog.ai.stream.AsyncStreamWrapper.aclose() -> None +method posthog.ai.stream.AsyncStreamWrapper.close() -> None +method posthog.ai.stream.StreamWrapper.close() -> None method posthog.bucketed_rate_limiter.BucketedRateLimiter.consume_rate_limit(key: Hashable) -> bool method posthog.bucketed_rate_limiter.BucketedRateLimiter.stop() -> None method posthog.client.Client.alias(previous_id: str, distinct_id: Optional[str], timestamp: Optional[Union[datetime, str]] = None, uuid: Optional[str] = None, disable_geoip: Optional[bool] = None) -> Optional[str] From 9147cd0c980babb2b5f7b36bce1d19309616697f Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Fri, 31 Jul 2026 10:15:10 +0200 Subject: [PATCH 3/4] chore(ai): keep sync stream wrapper internal --- posthog/ai/anthropic/anthropic.py | 4 ++-- posthog/ai/stream.py | 6 +++--- references/public_api_snapshot.txt | 3 --- 3 files changed, 5 insertions(+), 8 deletions(-) diff --git a/posthog/ai/anthropic/anthropic.py b/posthog/ai/anthropic/anthropic.py index b8119d95..45f86951 100644 --- a/posthog/ai/anthropic/anthropic.py +++ b/posthog/ai/anthropic/anthropic.py @@ -10,7 +10,7 @@ import uuid from typing import Any, Dict, List, Optional -from posthog.ai.stream import StreamWrapper +from ..stream import _StreamWrapper from posthog.ai.types import StreamingContentBlock, TokenUsage, ToolInProgress from posthog.ai.utils import ( call_llm_and_track_usage, @@ -262,7 +262,7 @@ def generator(): stop_reason=stop_reason, ) - return StreamWrapper(generator(), stream=response) + return _StreamWrapper(generator(), stream=response) def _capture_streaming_event( self, diff --git a/posthog/ai/stream.py b/posthog/ai/stream.py index e988cee1..2499f6e6 100644 --- a/posthog/ai/stream.py +++ b/posthog/ai/stream.py @@ -5,7 +5,7 @@ T = TypeVar("T") -class StreamWrapper(Generic[T]): +class _StreamWrapper(Generic[T]): """Preserves a provider stream's helpers while tracking its iteration.""" def __init__( @@ -16,13 +16,13 @@ def __init__( self._generator = generator self._stream = stream - def __iter__(self) -> "StreamWrapper[T]": + def __iter__(self) -> "_StreamWrapper[T]": return self def __next__(self) -> T: return next(self._generator) - def __enter__(self) -> "StreamWrapper[T]": + def __enter__(self) -> "_StreamWrapper[T]": return self def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt index 6573a31b..67a22411 100644 --- a/references/public_api_snapshot.txt +++ b/references/public_api_snapshot.txt @@ -37,7 +37,6 @@ alias posthog.ai.anthropic.AsyncAnthropic -> posthog.ai.anthropic.anthropic_asyn alias posthog.ai.anthropic.AsyncAnthropicBedrock -> posthog.ai.anthropic.anthropic_providers.AsyncAnthropicBedrock alias posthog.ai.anthropic.AsyncAnthropicVertex -> posthog.ai.anthropic.anthropic_providers.AsyncAnthropicVertex alias posthog.ai.anthropic.anthropic.PostHogClient -> posthog.client.Client -alias posthog.ai.anthropic.anthropic.StreamWrapper -> posthog.ai.stream.StreamWrapper alias posthog.ai.anthropic.anthropic.StreamingContentBlock -> posthog.ai.types.StreamingContentBlock alias posthog.ai.anthropic.anthropic.TokenUsage -> posthog.ai.types.TokenUsage alias posthog.ai.anthropic.anthropic.ToolInProgress -> posthog.ai.types.ToolInProgress @@ -894,7 +893,6 @@ class posthog.ai.prompts.CachedPrompt(prompt: str, fetched_at: float, name: str, class posthog.ai.prompts.PromptResult(source: PromptSource, prompt: str, name: Optional[str] = None, version: Optional[int] = None, label: Optional[str] = None, config: Optional[Dict[str, Any]] = None) class posthog.ai.prompts.Prompts(posthog: Optional[Any] = None, *, personal_api_key: Optional[str] = None, project_api_key: Optional[str] = None, host: Optional[str] = None, default_cache_ttl_seconds: Optional[int] = None, capture_errors: bool = False) class posthog.ai.stream.AsyncStreamWrapper(generator: AsyncGenerator[T, None], stream: Optional[Any] = None) -class posthog.ai.stream.StreamWrapper(generator: Generator[T, None, None], stream: Any) class posthog.ai.types.FormattedFunctionCall class posthog.ai.types.FormattedImageContent class posthog.ai.types.FormattedMessage @@ -1232,7 +1230,6 @@ method posthog.ai.prompts.Prompts.compile(prompt: str, variables: PromptVariable method posthog.ai.prompts.Prompts.get(name: str, *, with_metadata: Optional[bool] = None, cache_ttl_seconds: Optional[int] = None, fallback: Optional[str] = None, version: Optional[int] = None, label: Optional[str] = None) -> Union[str, PromptResult] method posthog.ai.stream.AsyncStreamWrapper.aclose() -> None method posthog.ai.stream.AsyncStreamWrapper.close() -> None -method posthog.ai.stream.StreamWrapper.close() -> None method posthog.bucketed_rate_limiter.BucketedRateLimiter.consume_rate_limit(key: Hashable) -> bool method posthog.bucketed_rate_limiter.BucketedRateLimiter.stop() -> None method posthog.client.Client.alias(previous_id: str, distinct_id: Optional[str], timestamp: Optional[Union[datetime, str]] = None, uuid: Optional[str] = None, disable_geoip: Optional[bool] = None) -> Optional[str] From a86e144d3fb87b2a4d20048c8790e3110b43bf7c Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Sat, 1 Aug 2026 08:39:27 +0200 Subject: [PATCH 4/4] fix(ai): guard Anthropic stream manager internals --- posthog/ai/anthropic/anthropic.py | 4 ++- posthog/ai/anthropic/anthropic_async.py | 4 ++- posthog/test/ai/anthropic/test_anthropic.py | 32 +++++++++++++++++++++ 3 files changed, 38 insertions(+), 2 deletions(-) diff --git a/posthog/ai/anthropic/anthropic.py b/posthog/ai/anthropic/anthropic.py index 45f86951..a25bedc4 100644 --- a/posthog/ai/anthropic/anthropic.py +++ b/posthog/ai/anthropic/anthropic.py @@ -126,7 +126,9 @@ def stream( # cannot re-enter our tracked ``create`` override. manager = Messages(self._client).stream(**kwargs) request_attribute = "_MessageStreamManager__api_request" - request = getattr(manager, request_attribute) + request = getattr(manager, request_attribute, None) + if request is None: + return manager def tracked_request(): start_time = time.time() diff --git a/posthog/ai/anthropic/anthropic_async.py b/posthog/ai/anthropic/anthropic_async.py index fb5eb075..eb642de8 100644 --- a/posthog/ai/anthropic/anthropic_async.py +++ b/posthog/ai/anthropic/anthropic_async.py @@ -126,7 +126,9 @@ def stream( # cannot re-enter our tracked ``create`` override. manager = AsyncMessages(self._client).stream(**kwargs) request_attribute = "_AsyncMessageStreamManager__api_request" - request = getattr(manager, request_attribute) + request = getattr(manager, request_attribute, None) + if request is None: + return manager async def tracked_request(): start_time = time.time() diff --git a/posthog/test/ai/anthropic/test_anthropic.py b/posthog/test/ai/anthropic/test_anthropic.py index 8c44c28c..0fb840f8 100644 --- a/posthog/test/ai/anthropic/test_anthropic.py +++ b/posthog/test/ai/anthropic/test_anthropic.py @@ -1853,6 +1853,20 @@ def test_messages_stream_preserves_native_manager_helpers_close_and_tracking( assert "posthog_distinct_id" not in client.post.call_args.kwargs +def test_messages_stream_tolerates_provider_manager_internals_changing(mock_client): + manager = object() + client = Anthropic(api_key="test-key", posthog_client=mock_client) + + with patch("anthropic.resources.messages.Messages.stream", return_value=manager): + response = client.messages.stream( + model="claude-haiku-4-5", + messages=[{"role": "user", "content": "Foo"}], + max_tokens=1, + ) + + assert response is manager + + @pytest.mark.asyncio async def test_async_messages_stream_preserves_provider_contract_and_manager( mock_client, @@ -1881,6 +1895,24 @@ async def test_async_messages_stream_preserves_provider_contract_and_manager( assert "posthog_distinct_id" not in client.post.call_args.kwargs +def test_async_messages_stream_tolerates_provider_manager_internals_changing( + mock_client, +): + manager = object() + client = AsyncAnthropic(api_key="test-key", posthog_client=mock_client) + + with patch( + "anthropic.resources.messages.AsyncMessages.stream", return_value=manager + ): + response = client.messages.stream( + model="claude-haiku-4-5", + messages=[{"role": "user", "content": "Foo"}], + max_tokens=1, + ) + + assert response is manager + + @pytest.mark.asyncio async def test_async_messages_create_streaming_supports_async_with(mock_client): """Regression test for #393: messages.create(stream=True) must support