fix(llm): keep plugin-specific realtime handlers across a session swap

[priyam-garg]

_FallbackRealtimeSession only re-attached the fixed _FORWARDED_EVENTS allowlist to each new child, so handlers for provider-specific events (e.g. openai's openai_server_event_received) were silently dropped by restart_session() and by an availability fallback.

Forwarders are now created on demand: subscribing on the wrapper registers a forwarder for that event, and _bind replays every forwarder onto each new child, so plugin-specific subscribers survive a swap like the generic ones already did. A _child_bound flag keeps a subscription made mid-swap off the child that is being discarded.

Forwarders also became varargs, so a plugin event carrying more than one payload argument no longer loses all but the first.

Fixes #6559
Fixes #6556

[devin-ai-integration]

✅ Devin Review: No Issues Found

Devin Review analyzed this PR and found no bugs or issues to report.

[priyam-garg]

Some extra detail on the root cause and the one behavioral change reviewers will want to weigh in on.

Root cause

_forwarders was built once, at construction, from a hardcoded tuple:

self._forwarders: dict[EventTypes, Callable[[object], None]] = {
    event: _make_forwarder(event) for event in _FORWARDED_EVENTS
}

_bind() replays exactly that dict onto each new child, so only those seven generic events survive a swap. The adapter has no way to enumerate plugin event names up front — openai_server_event_received, openai_client_event_queued, and whatever the other plugins define — so a static allowlist can never be complete. That's why the fix moves forwarder creation to subscribe time rather than adding more names to the tuple.

The one behavioral change

Handlers have to be bound on the wrapper rather than on the child. Binding to _active attaches to one specific child object, which a swap then discards — that is the reported bug, and no amount of forwarding on our side can rescue a handler the caller attached directly to the session being thrown away.

rt_session = agent.realtime_llm_session

# before: bound to the child, lost on the next swap
rt_session._active.on("openai_server_event_received", handler)

# after: bound to the wrapper, re-attached to every new child
rt_session.on("openai_server_event_received", handler)

Worth calling out explicitly since it's a call-site change for anyone already working around this, though _active was never public API.

Design notes

• error is deliberately excluded from the dynamic forwarders. _on_child_error already re-emits it, sometimes re-stamped as recoverable, and a second forwarder would duplicate every error. The existing error tests (test_restart_emits_no_error, test_non_recoverable_error_forwarded_as_recoverable_while_fallback_remains) now run through the overridden on() and would catch a regression here.
• _child_bound covers the window between _unbind of the outgoing child and _bind of the incoming one. Subscribing during that window adds to the dict only; the upcoming _bind picks it up, so the handler never attaches to the child being discarded.
• Forwarders became varargs. The old _forward(ev) silently truncated any plugin event carrying more than one payload argument.
• The class generic widens to RealtimeSession[str] since plugin event names are only known to the plugin. That Literal["realtime_availability_changed"] was already inaccurate — the event is emitted on the adapter, never on the session.

Test evidence

Four of the five new tests fail on main and pass with the fix:

FAILED test_restart_preserves_plugin_event_subscribers
FAILED test_fallback_preserves_plugin_event_subscribers
FAILED test_plugin_event_subscribed_mid_swap_binds_to_new_child
FAILED test_forwards_multi_arg_plugin_events
4 failed, 37 passed

The fifth, test_plugin_event_detached_from_discarded_child, passes either way by construction — it's a regression guard against a fix that leaves the discarded child bound and double-delivers.

Scope of #6556

Flagging this so the auto-close isn't a surprise: #6556 scopes the ask to resubscribing when falling back to the same plugin-type model. This implementation resubscribes on every swap, including across plugin types — an openai_server_event_received forwarder would attach to, say, a Gemini session and simply never fire. Harmless, but broader than what was asked. Happy to gate it on type(old) is type(new) if you'd prefer the narrower behavior.

Possible follow-up

This makes the case for #6553 a bit stronger: callers now need session.on(...) on _FallbackRealtimeSession, a class whose name starts with an underscore. Left out of this PR to keep it focused, but happy to pick it up separately.

Verified with ruff format --check, ruff check, and mypy --strict on livekit.agents.llm. Note that the tests exercise the hermetic fakes in tests/fake_realtime.py; I haven't run the issue's exact repro against a live OpenAI realtime session, though the forwarding path is provider-agnostic.