fix streaming handling for builtin assistants

[pmeier]

My patch in #425 is broken when using the Python API with asyncio.run, i.e. the regular use case. As implemented in #425, breaking out of the loop in .answer() means that the async iterator is not fully exhausted, i.e. AsyncStopIteration is never raised. This leads to the context manager that handles the HTTP request staying open longer than Chat.answer. When it is cleaned up eventually, internally it calls await request.aclose(). However, if at that point the async event loop is already killed, e.g. because the coroutine called by asyncio.run is finished, you'll get an error:

Task exception was never retrieved
future: <Task finished name='Task-5' coro=<<async_generator_athrow without __name__>()> exception=RuntimeError('aclose(): asynchronous generator is already running')>
RuntimeError: aclose(): asynchronous generator is already running
Exception ignored in: <coroutine object Response.aclose at 0x73f95efdfcc0>
Traceback (most recent call last):
  File "/home/philip/.conda/envs/ragna-deploy-dev/lib/python3.9/site-packages/httpx/_models.py", line 1008, in aclose
    await self.stream.aclose()
  File "/home/philip/.conda/envs/ragna-deploy-dev/lib/python3.9/site-packages/httpx/_client.py", line 155, in aclose
    await self._stream.aclose()
  File "/home/philip/.conda/envs/ragna-deploy-dev/lib/python3.9/site-packages/httpx/_transports/default.py", line 259, in aclose
    await self._httpcore_stream.aclose()
  File "/home/philip/.conda/envs/ragna-deploy-dev/lib/python3.9/site-packages/httpcore/_async/connection_pool.py", line 374, in aclose
    await self._stream.aclose()
  File "/home/philip/.conda/envs/ragna-deploy-dev/lib/python3.9/site-packages/httpcore/_synchronization.py", line 225, in __exit__
    self._anyio_shield.__exit__(exc_type, exc_value, traceback)
  File "/home/philip/.conda/envs/ragna-deploy-dev/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 406, in __exit__
    if current_task() is not self._host_task:
  File "/home/philip/.conda/envs/ragna-deploy-dev/lib/python3.9/asyncio/tasks.py", line 38, in current_task
    loop = events.get_running_loop()
RuntimeError: no running event loop

To fix this, this PR introduces the same pattern for our streaming handling that httpx and httpx_sse use as well: a context manager creates the stream and the stream can than be iterated on.

ragna/ragna/assistants/_http_api.py

Lines 83 to 89 in 7c5728a

	async with httpx_sse.aconnect_sse(
	self._client, method, url, **kwargs
	) as event_source:
	await self._assert_api_call_is_success(event_source.response)
	async for sse in event_source.aiter_sse():
	yield json.loads(sse.data)

ragna/ragna/assistants/_http_api.py

Lines 99 to 103 in 7c5728a

	async with self._client.stream(method, url, **kwargs) as response:
	await self._assert_api_call_is_success(response)
	async for chunk in response.aiter_lines():
	yield json.loads(chunk)

Basically, we switch from

async for data in self._call_api(...):
    ...

to

async with self._call_api(...) as stream:
    async for data in stream:
        ...

With this pattern, the cleanup is correctly handled when we exit .answer() and thus get rid of the error.

[nenb]

Seems sensible to me.

Nitpick: Could you add a unit test to prevent this regression in the future?

[pmeier]

Could you add a unit test to prevent this regression in the future?

I thought about it, but I honestly have trouble coming up with one that is not unreasonably complex. We need a server that we can hit for HTTP streaming. And the best option we have here is to start our own.

I'll see if we can have this in a reasonable way.

[pmeier]

Moved it to the generic test utils as this is no longer just needed for deploy.

[pmeier]

We had this before, but it was removed in #322. Probably can also be used by the UI tests, but we can do that in a follow-up.

[pmeier]

@nenb This test fails when I revert my patch in this PR. See the CI runs for 0b9211e.