Async SSE client reuses httpx.AsyncClient across event loops -> "Event loop is closed"

[nthmost-orkes]

Re-scoped from conductor-oss/conductor#1563 (originally agentspan-ai/agentspan#318). That one was filed against the server repo by the Agentspan migration, but the code actually lives here in the Python SDK.

OrkesAgentClient._get_sse_async_client() (src/conductor/client/orkes/orkes_agent_client.py:274) caches one httpx.AsyncClient and only recreates it when is_closed is true. The catch: a client whose connection pool is bound to a closed event loop is not is_closed -- so it gets reused. Drive the async streaming API through a fresh asyncio.run(...) twice in one process (each asyncio.run spins up and tears down its own loop) and the second call dies with RuntimeError: Event loop is closed. The client is only reset in shutdown_async(), never per-stream.

Good news up front: the original repro (sync start() called twice) no longer reproduces -- the sync SSE path was rewritten on the synchronous requests library, so it never touches an async client. This is confined to the async API (stream_async() / _get_sse_async_client()).

Repro that drives the real getter (a tiny local keep-alive server is all it needs so a connection actually gets pooled and bound to loop A):

import asyncio, threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from conductor.client.configuration.configuration import Configuration
from conductor.client.orkes.orkes_agent_client import OrkesAgentClient

class H(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"   # keep-alive so httpx pools the connection
    def do_GET(self):
        b = b"ok"; self.send_response(200)
        self.send_header("Content-Length", str(len(b))); self.end_headers(); self.wfile.write(b)
    def log_message(self, *a): pass

srv = HTTPServer(("127.0.0.1", 0), H); port = srv.server_address[1]
threading.Thread(target=srv.serve_forever, daemon=True).start()
url = f"http://127.0.0.1:{port}/"

client = OrkesAgentClient(Configuration(server_api_url="http://localhost:8080/api"))

async def touch():
    c = client._get_sse_async_client()
    print("is_closed before use:", c.is_closed)
    print("GET ->", (await c.get(url)).status_code)

asyncio.run(touch())   # loop A -- opens + pools a connection bound to loop A
asyncio.run(touch())   # loop B -- reuses the cached client -> boom

Output:

is_closed before use: False
GET -> 200
is_closed before use: False
RuntimeError: Event loop is closed

Fix: in _get_sse_async_client(), also recreate the client when the current running loop differs from the one it was created on (track id(asyncio.get_running_loop()) next to the cached client). Confirmed locally that this makes the repro pass (both calls return 200).