feat(voice): support preemptive generation for realtime models

[eh-steve]

Motivation

Preemptive generation (starting the reply on the eager transcript, before the turn is confirmed) currently only fires for cascaded llm.LLM pipelines — on_preemptive_generation early-returns for any RealtimeModel. Pipelines that pair a text-mode realtime model with an STT that emits eager end-of-turn transcripts (e.g. Cartesia Ink-2's PREFLIGHT_TRANSCRIPT) can't get the same head start, even though everything downstream of the guard (the _PreemptiveGeneration fingerprint/adoption lifecycle, and _realtime_generation_task's authorization-gated playout) already supports it.

The reason a naive guard-lift isn't enough: _realtime_reply_task waits for authorization before issuing rt_session.generate_reply, so an unscheduled preemptive speech would only start generating at adoption time — no head start. And unlike cascaded speculation (a stateless local task), a realtime speculation mutates remote state: a pushed conversation item and, on single-active-response APIs, the session's one response slot — so it needs a real rollback story.

What this PR does

• Eligibility: on_preemptive_generation now accepts realtime models via _can_preemptively_generate(): requires mutable_chat_context, no server-side turn detection, text-only output (the speculative reply is generated from the pushed transcript, keeping the turn's input modality consistent), a live rt_session, and no active response.
• Eager generation: _realtime_reply_task gains a preemptive mode that pushes the transcript into the session context and issues generate_reply() before the authorization gate; playout and the local history commit stay gated on authorization (adoption). On adoption, the existing fingerprint match in _user_turn_completed_task schedules the held speech and the already-streaming generation plays through the native _realtime_generation_task path.
• Exit-path-total rollback: the speculation publishes a cleanup future before touching the session (chained behind any previous speculation's cleanup), and every exit other than adoption cancels the speculative response, waits for the server to release it, and restores the pre-speculation chat context. Follow-up generations wait on the cleanup future so they can never run against a dirty session. A speculative generate_reply that fails before adoption wakes the parked task and marks the handle done, so adoption falls back to a normal generation.
• RealtimeSession.has_active_generation (new ABC property, default False; the OpenAI plugin already implements it, and the realtime fallback adapter proxies it): lets the framework serialize response creation. Realtime replies — including tool replies and post-barge-in generations — now wait for a cancelled response to clear before creating the next one, avoiding conversation_already_has_active_response rejections.
• Turn-commit fixes in _user_turn_completed_task: the final transcript is captured for the preemptive match before the realtime branch nulls user_message (previously this path would have crashed reading user_message.raw_text_content); commit_audio() is deferred while a speculation is pending (clear_audio() on adoption, commit_audio() on invalidation or any early exit such as StopResponse, via a try/finally so buffered audio can never leak into a later turn's commit); the barge-in cleanup no longer interrupts the in-flight speculative response, and a turn arriving with skip_reply (e.g. commit_user_turn()) cancels the pending speculation instead of stranding it.
• _realtime_reply_task now receives the user message as llm.ChatMessage (resolving the existing TODO), so metrics injected at adoption land on the committed message.

Testing

• 22 new tests in tests/test_realtime_preemptive.py: eager-generate-before-authorization, adoption commit + native generation, rollback of pending and already-created responses (with a duration regression bound), supersession serialization, pre-adoption generation failure fallback, follow-up serialization, eligibility gating, the remote-item mirroring hold-back (buffered + replayed on adoption, removed on rollback, pass-through for unrelated items), the speculation TTL, and five full-AgentSession e2e tests (adoption with a verified head start, invalidation via an on_user_turn_completed edit, StopResponse with the deferred audio commit resolved, skip_reply/commit_user_turn cancelling a pending speculation, and pause() completing with a parked speculation). FakeRealtimeSession now emits remote_item_added acks so the mirroring machinery is covered by the fakes.
• Existing suites green: test_agent_session.py (incl. cascaded preemptive latency tests), test_realtime_reply_chat_ctx.py (updated for the ChatMessage signature), test_realtime_fallback.py, test_realtime_message_metrics.py, test_realtime_adaptive_interruption.py, test_session_host.py, test_remote_session.py — 168 tests total. ruff and mypy clean on the changed files.

Notes / possible follow-ups

• The bounded poll in _wait_for_active_response_cleared could become an awaitable on the RealtimeSession ABC if you'd prefer event-based signaling; the poll was chosen because it's self-healing across session reconnects (no waiter lifecycle to manage in every plugin).
• Audio-output realtime models are excluded from speculation for now (input-modality consistency); could be relaxed later.
• The adoption guard now also skips a preemptive speech handle that already failed (speech_handle.done()), which for cascaded pipelines means a failed preemptive inference falls back to a normal generation instead of scheduling a dead speech — intentional, previously the turn went silent.
• An optional normalized transcript match (case/punctuation-insensitive) would increase adoption rates for eager-EOT STTs whose preflight and final transcripts differ only in punctuation; kept exact-match here to preserve existing semantics.

Made with Cursor

Live verification (real providers)

Verified end-to-end with Cartesia Ink-2 (via LiveKit Inference, turn_detection="stt") + OpenAI Realtime gpt-realtime-1.5 (text-only) on a room-less AgentSession, streaming synthesized user speech at real-time pace:

• Head start: the speculative generate_reply launches on Ink-2's eager transcript ~0.10-0.14s after the user stops speaking (sometimes before), while the turn only commits at ~0.59s — a consistent ~460-490ms head start across runs, all adopted (single generation per turn, clear_audio, correct reply, identical local/server history).
• Baseline (preemptive disabled): generate_reply launches within 1ms after the turn commit, as expected.
• End-to-end: agent speech started a median ~0.52s sooner after end-of-user-speech with preemptive generation enabled (0.74s vs 1.26s median across runs; TTS held constant).
• Rollback: forcing invalidation via an on_user_turn_completed edit produced the full rollback + fallback live: speculative response cancelled, its items removed server-side, the committed turn audio preserved, and the fallback generation answered the original question correctly.
• Barge-in + speculation: a mid-utterance pause produced two consecutive turns where the second speculation launched while the first reply was being interrupted — no conversation_already_has_active_response collisions (the serialization path).

Live testing also caught two integration issues the fake-based tests couldn't, now fixed in this PR: the plugin's remote_item_added mirroring placed speculative items into the local history mid-turn (invalidating the preemptive match on every real turn), so mirroring of speculation-owned items is held back until adoption; and the rollback now removes exactly the speculation's items from the live context instead of restoring a snapshot, so concurrent state (e.g. the committed turn audio) survives.

Note on supersession coverage: supersession (a second eager transcript superseding an in-flight speculation) could not be triggered against real providers — Ink-2's eager and final end-of-turn signals arrive ~0.5s apart, and _can_preemptively_generate() intentionally declines to re-speculate while the previous speculative response is still active (the stale speculation is then invalidated by the fingerprint at turn end: safe, just no head start). The rollback-chaining behavior is covered at the unit level (test_preemptive_supersession_serializes_cleanup).