run_stream error when break early

[yihuang]

Initial Checks

• I confirm that I'm using the latest version of Pydantic AI
• I confirm that I searched for my issue in https://github.com/pydantic/pydantic-ai/issues before opening this issue

Description

Example Code

Minimal reproduce code:

import asyncio

from pydantic_ai import Agent
from pydantic_ai.messages import ModelMessage
from pydantic_ai.result import StreamedRunResult


async def main(model: str, prompt: str):
    agent = Agent(model=model)
    async with agent.run_stream(prompt) as result:
        async for content in result.stream_text(delta=True):
            print(content)

            # break early will trigger error
            break


if __name__ == "__main__":
    asyncio.run(main("deepseek:deepseek-chat", "Hello, how are you?"))

Python, Pydantic AI & LLM client version

Python 3.12.8

>>> pydantic_ai.__version__
'0.1.0'

>>> openai.__version__
'1.74.0'

[yihuang]

investigation shows the issue is group_by_temporal, I've tried to replace it with normal async iteration, the issue disappears.

Actually, just pass debounce_by=None will success.

[yihuang]

minimal reproduction code:

import asyncio
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import Any

from pydantic_ai._utils import group_by_temporal


async def stream():
    for i in range(10):
        print("streaming", i)
        yield i
        await asyncio.sleep(0.1)


@asynccontextmanager
async def run_stream():
    streamed_response = stream()
    yield StreamedRunResult(streamed_response)
    async for _ in streamed_response:
        pass


@dataclass
class StreamedRunResult:
    _response: Any

    async def stream(self):
        async with group_by_temporal(self._response, 0.2) as group_iter:
            async for items in group_iter:
                yield items


async def main():
    async with run_stream() as result:
        async for chunk in result.stream():
            print(chunk)
            break


asyncio.run(main())