standard_asr.runtime.streaming

Incorporate one event into the running result.

State commits only on the ADMITTED path: every suppression below returns before the method's tail, and the sticky detected_language is committed at that tail -- so a suppressed event changes nothing, not even session-level metadata (the same admitted-only commit discipline the session guard applies to its audio_processed_until cursor). Committing the language on entry let a suppressed supersede's detected_language rewrite result() while the event itself was refused -- an event stream and its reduction silently disagreeing about what happened.

Build the reduced transcription result.

Enter the async session's context.

Exception-safe: a context manager whose __enter__ raises never receives __exit__, so a failed enter MUST tear down the owned loop + thread started in __init__ itself -- otherwise an engine whose _open raises (bad credentials, unreachable host) would leak the bridge's background thread (no leak). The timeout path is already torn down inside _submit; the entered guard also covers a non-timeout raise (_open raising a regular exception).

Exit the async context and stop the owned loop.

Feed audio from a managed source.

Manually send one audio chunk.

Mark the end of manual audio input.

Iterate events synchronously.

Event waits are unbounded by design: a live, fed session may legitimately go arbitrarily long between events (user silence), so the pump must never manufacture a timeout for it -- a stuck pipeline already surfaces as a terminal event from the async side's own deadlines. The pump waits in short slices purely to detect the two failures those in-loop deadlines cannot report: the owned event-loop thread dying, and the loop being frozen by an engine running blocking (non-async) code. Brief blocking stalls are tolerated (several consecutive unresponsive probes are required), so only a persistently frozen loop tears the bridge down.

Reduce the session so far into a transcription result.

SERIALIZED WITH THE PRODUCER while the bridge is live: the reduction walks reducer state the producer task mutates (a supersede pops segments mid-walk), so it is submitted to the owned loop like every other bridge member -- asyncio's run-to-completion between awaits is the mutual exclusion, and the wrapper coroutine never awaits around the call. Running it on the caller's thread instead read that state concurrently: a rendering loop (for ev in sync: render(sync. result())) crashed with a spurious KeyError or returned a torn result mixing pre- and post-supersede segments. After teardown (__exit__, or a timed-out lifecycle call) no producer runs anymore, so the direct call is safe -- and keeps the result-after-the-with-block pattern working.

Return the session's standard-layer and lifecycle diagnostics.

Mirrors TranscriptionSession.diagnostics so a synchronously- driven session exposes the same parameter-gating / language-resolution / lifecycle-suppression diagnostics as the async one. Without this the sync bridge would silently drop a first-class, compliance-checked part of the session surface (the sync bridge is a faithful mirror of the async session). Serialized with the producer exactly like result (the snapshot copies guard state the producer task appends to); direct once torn down.

Whether the owned background event-loop thread is still running.

The bridge starts a dedicated event-loop thread in __init__ and MUST tear it down on close (__exit__) or on a failed __enter__ (from an external thread). After a clean lifecycle this returns False; a True here once the session is closed is a leaked loop thread. Exposed so the sync-bridge compliance check can assert on this bridge's own thread rather than diffing the whole process thread set (which would mis-flag a dependency's benign daemon thread as a leak).

Build a partial event.

Build a final event.

Build a closed finality event (a final with finality=closed).

Build a supersede event replacing old segments with new ones.

Lineage is set-to-set: old_ids/new_ids express the cardinality of the re-segmentation (which ids retire, which appear) but not a per-old->per-new mapping. On a merge+split (many->many) a UI cannot tell which specific old segment a given new segment descends from. This is a documented v1 limitation; the spec does not require a pairwise mapping, and the frozen-prefix-preservation invariant is enforced over the concatenated prefixes, not per pair. Per-pair edit-ops/diffs are a possible future direction (additive later).

Build a progress event (heartbeat / cursor / reconnect notice).

Build a terminal done event.

Build an error event.

Async-iterate fed audio chunks until the input ends.

Each yielded chunk is also retained in the bounded rolling audio buffer for possible replay after a reconnect.

Consuming from this iterator is the session's liveness anchor: every dequeue resets the done_timeout backstop, so an engine that legitimately emits no events during user silence stays alive for as long as it keeps reading audio here.

Surface a structured diagnostic from _produce (the streaming channel).

The streaming counterpart of the batch path's result.diagnostics: call this from your _produce to report a non-fatal note -- a best-effort degradation, an assumed parameter, a lossy fallback -- through the session's own diagnostics channel, so a WS client (mid-stream diagnostics frame) or a sync caller (diagnostics()) sees WHY. For a fatal condition, emit an error event instead -- this is for non-fatal notes that must not end the stream.

Recorded through the same bounded channel as the guard's own diagnostics: past the cap entries aggregate into the overflow summary rather than growing without bound, so a chatty producer cannot exhaust memory.

Security -- this is engine-authored, client-facing output, like the transcript text itself. Every field (message / param / provided / effective) is forwarded to clients VERBATIM -- to a (possibly unauthenticated) WebSocket client as a diagnostics frame, and, for the batch counterpart result.diagnostics, in the REST response -- and is NOT redacted. The standard layer scrubs only the detail IT auto-captures (an error event's extra, which holds a scrubbed exception summary -- never a raw str(exc)); content you pass here is your contract to keep. NEVER pass a credential, API key, a URL with embedded auth, or raw exception text -- route sensitive operator detail to logging instead. (That asymmetry is deliberate: an error event's extra is auto-captured so the server drops it; this note you chose, so you own its safety.)

Return audio for re-feeding to a freshly re-established connection.

Engines call this on reconnect to re-send audio. For a replayable (list / tuple) source the COMPLETE source is returned -- replayability promises loss-free replay, so it must not be silently truncated to the rolling ring. For a non-replayable / live source only the bounded rolling window of recent chunks is available (older audio was evicted).

Record that an internal reconnect bridged a gap (engine-driven).

The base ALWAYS emits a progress(reconnect=True, gap_start, gap_end) event in order with produced events, followed -- IMMEDIATELY -- by a trailing error(code="content_lost", recoverable=True, gap_start, gap_end) IFF the engine passes content_lost=True.

The events are flushed promptly, the moment this is called, through the drop-proof path (_CoalescingBuffer.put_forced, the same sync primitive the base uses for synthesized terminals). This matters because an engine typically calls note_reconnect and then BLOCKS while it re-establishes the connection without yielding another event; deferring the flush to the next produced event would leave the consumer staring at a timeout/silence during a slow reconnect -- the opposite of "transparent but honest". note_reconnect runs in the producer-task coroutine on the session loop, so a synchronous flush is correct and ordered (the events land after already-emitted events and before any subsequent one). The _run_producer / finally drains remain as a harmless safety net -- they simply find nothing pending.

Content loss is an explicit engine determination, not something the base infers from rolling-buffer eviction: a live ring is always evicting, so eviction is the wrong signal (it would falsely claim permanent loss on every long live session). The engine -- which alone knows whether its reconnect + replay_buffer replay actually covered the gap -- sets content_lost=True only when audio the engine had not yet processed could not be replayed and is therefore truly lost. This mirrors the existing contract where segment_id / timestamps / detected language / speaker-label mapping continuity across the reconnect is likewise the engine's responsibility (the base never rewrites them). For speaker labels the engine MUST NOT reuse a pre-reconnect label without identity evidence; the safe default is to mint fresh labels and emit a speaker_labels_reset diagnostic via emit_diagnostic (the fidelity-warning counterpart of the content_lost error event).

Feed audio from a managed source (mutually exclusive with manual).

A bare bytes / bytearray is treated as a single audio chunk (not an iterable of chunks -- iterating it would yield int byte values). A non-async, re-iterable collection (list / tuple, or a wrapped bytes-like) is classified as replayable for reconnect purposes; an async source or a one-shot generator/iterator is non-replayable. Any AsyncIterable (the Pythonic __aiter__ protocol, not only the stricter AsyncIterator) is accepted and normalized via aiter.

Manually send one audio chunk (mutually exclusive with feed).

Manual sources are always treated as non-replayable (live input).

Mark the end of manual audio input (idempotent in manual mode).

Claims manual ownership if this is the first input call, so a later feed is correctly rejected as mixing.

Open resources and start the producer.

Anchors the max_session_seconds wall-clock origin here, at session establishment, so the absolute cap measures from when the session opened -- not from the first __anext__ -- and a re-entered iterator cannot reset it.

Tear down the producer, feed task, and engine resources.

Cancels the producer and feed tasks and AWAITS them (so no coroutine is still touching engine state when _close runs), then closes.

Return the event async iterator (single-consumer).

A session has exactly ONE event stream and exactly ONE consumer: the events live in a shared bounded buffer, and a second concurrent iterator would race the first for _CoalescingBuffer.get, silently splitting the stream between them (each iterator would see an arbitrary subset, so neither's result would match the stream -- breaking the stream == result invariant). Re-iterating would also reset the per-iteration deadline anchors. Both are programming errors, so the second call fails loudly instead of returning a competing iterator.

Return the diagnostics accumulated for this session so far.

The lifecycle-suppression channel is bounded: a misbehaving engine that trips a clamp on (nearly) every event cannot grow this list without limit over a long session. Once DEFAULT_MAX_GUARD_DIAGNOSTICS entries accumulate, further suppressions are aggregated into a single trailing diagnostics_truncated summary (per-code counts) instead of being retained individually -- so the overflow is reported honestly, never silently dropped.

Reduce the session so far into a transcription result.