fix(sarvam): close the streams the STT collects

[dhruvladia-sarvam]

The leak

STT.stream() gives every SpeechStream its own aiohttp.ClientSession ("a fresh session for this stream to avoid conflicts"), and only SpeechStream.aclose() closes it. The STT collects its streams into self._streams = weakref.WeakSet[SpeechStream](), but nothing ever reads that set, and the STT class defines no aclose() — so STT.aclose() resolves to the abstract no-op on the base class. The streams are never closed, their sessions stay open, and aiohttp reports unclosed client sessions once the streams are dropped.

The base contract asks for this: STT.aclose is documented as "Close the STT, and every stream/requests associated with it", and STT.__aexit__ delegates to it, so async with stt: currently leaks the streams. stt_streaming.py in this same package already implements the intended shape at line 466.

The change

Eleven lines on STT, mirroring stt_streaming.py:

async def aclose(self) -> None:
    for stream in list(self._streams):
        await stream.aclose()
    self._streams.clear()

The STT owns no session of its own (non-streaming paths resolve through utils.http_context.http_session()), so the streams are the whole of it.

Validation

• Regression tests added (tests/test_plugin_sarvam_stt.py, hermetic — the streams' _run is stubbed, no network): per-stream sessions are closed by STT.aclose(), the async with path closes them via __aexit__, and aclose() tolerates already-closed streams. With the fix reverted, the first two tests fail exactly as the leak predicts (plus the suite's leaked-task detection fires); with the fix, all 16 tests in the file pass.
• ruff check, ruff format --check, and strict mypy pass on the touched plugin file.
• Reproduced the pre-fix leak independently: aiohttp's Unclosed client session warnings appear once streams are dropped.

[Rehansanjay]

The strong-ownership change is the right call and it is a hole in what I had in #7012 — a WeakSet drops a finished stream before aclose() can snapshot it, which is exactly the case that leaks in practice. Same for removing only what the call actually closed rather than clearing the set.

I have rescoped #7012 to simplismart so the two do not overlap. It has the identical bug — weakref.WeakSet at stt.py:187, a per-stream aiohttp.ClientSession created in stream(), and no aclose() on the STT — and it is not covered here. I followed the approach in this PR there, including the self-discard in SpeechStream.aclose(), so the two plugins stay consistent.

[tinalenguyen]

hi, thanks for the pr! could you address this finding:

Nothing in livekit-agents/ calls STT.aclose(), so the new strong set is filled on every stream() and drained only by a detached, cancellable path — pinning dead streams and their open sessions for the process lifetime. That's a worse leak than the WeakSet it replaces.

Fix before merge: gather(..., return_exceptions=True) in aclose(), discard in a finally (not gated on session.closed), a weakref.finalize to close the session on GC, and repair the test branch that tolerates the regression. No API-level breaking changes.