car_engine/

voice_turn.rs

//! Two-track voice turn orchestration.
//!
//! Quip's `SidecarReasoningService` + `MediaPlatformService.HandleUtteranceAsync`
//! ported to CAR. See [`docs/proposals/voice-sidecar-orchestration.md`] and
//! [`docs/proposals/voice-sidecar-orchestration-plan.md`] for the design
//! rationale and implementation plan.
//!
//! The two-track shape:
//!
//! ```text
//!                      user utterance
//!                            │
//!                            ▼
//!           ┌────────────────────────────────────┐
//!           │                                    │
//!        FAST_TURN_1 (<500ms first audio)    SIDECAR
//!        streaming LLM, voice-context        full LLM with tools (2-8s)
//!           │                                    │
//!           ▼                                    │
//!        TTS streaming → play              oneshot::Sender<SidecarResult>
//! ```
//!
//! The caller obtains a [`VoiceTurnHandle`], drains the fast stream into
//! TTS immediately, awaits the sidecar with a timeout, and plays the
//! result if it arrives in time. [`VoiceTurnControl`] is a cheap clonable
//! handle for cross-task cancellation (see plan §6.8).

use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use car_eventlog::{EventKind, EventLog};
use car_inference::{GenerateRequest, InferenceEngine, StreamEvent};
use serde_json::Value;
use tokio::sync::{mpsc, oneshot};
use tokio_util::sync::CancellationToken;

/// Mint a new turn id. Process-wide, monotonic, never reused.
fn next_turn_id() -> u64 {
    static COUNTER: AtomicU64 = AtomicU64::new(1);
    COUNTER.fetch_add(1, Ordering::Relaxed)
}

/// Result delivered by the sidecar track once the substantive answer
/// has been produced (or a fast-data path has hit).
#[derive(Debug, Clone)]
pub struct SidecarResult {
    /// The turn this result belongs to. Callers MUST gate playback
    /// on `result.turn_id == current_turn_id` so stale results from
    /// superseded turns are dropped after a barge-in.
    pub turn_id: u64,
    /// Substantive answer to play through TTS.
    pub text: String,
    /// Optional structured data (tool results, etc.) for callers that
    /// want to render something richer than text-to-speech.
    pub data: Option<serde_json::Value>,
}

/// Failure modes distinguishable for telemetry / fallback policy.
#[derive(Debug, thiserror::Error)]
pub enum VoiceTurnError {
    /// The underlying inference call failed. The string carries the
    /// upstream error's display form (we don't bubble the concrete
    /// `InferenceError` to keep this enum FFI-friendly).
    #[error("inference failed: {0}")]
    Inference(String),
    /// The turn was cancelled — typically a barge-in or supersession
    /// by a newer utterance.
    #[error("turn cancelled (barge-in or supersession)")]
    Cancelled,
}

/// Cheap, clonable control surface for an in-flight voice turn.
///
/// Stored separately from the receivers so an orchestrator can keep a
/// copy in `Mutex<Option<VoiceTurnControl>>` and cancel from a different
/// task than the one driving the streams. See plan §6.8.
#[derive(Clone)]
pub struct VoiceTurnControl {
    /// Unique id for this turn; threaded through [`SidecarResult`] and
    /// telemetry. Use it to drop stale results after barge-in.
    pub turn_id: u64,
    cancel: CancellationToken,
}

impl VoiceTurnControl {
    /// Cancel the in-flight turn. Idempotent.
    pub fn cancel(&self) {
        self.cancel.cancel();
    }

    /// Has the turn been cancelled?
    pub fn is_cancelled(&self) -> bool {
        self.cancel.is_cancelled()
    }
}

/// Two-track handle returned by [`dispatch_voice_turn`].
///
/// `fast` streams [`StreamEvent`]s for the fast track (text deltas,
/// tool-call markers, usage, done). `sidecar` resolves once with the
/// substantive answer or an error. `control` is a cheap clonable
/// surface that can be cached by the orchestrator for cancellation
/// from another task.
///
/// Caller MUST drive both receivers. Dropping `fast` cancels the fast
/// task; dropping `sidecar` cancels the sidecar task. To cancel both
/// without driving them, call [`Self::cancel`] (or
/// [`VoiceTurnControl::cancel`] on a stored clone).
pub struct VoiceTurnHandle {
    pub control: VoiceTurnControl,
    pub fast: mpsc::Receiver<StreamEvent>,
    pub sidecar: oneshot::Receiver<Result<SidecarResult, VoiceTurnError>>,
}

impl VoiceTurnHandle {
    /// Convenience accessor for `self.control.turn_id`.
    pub fn turn_id(&self) -> u64 {
        self.control.turn_id
    }

    /// Cancel both tracks. Idempotent.
    pub fn cancel(&self) {
        self.control.cancel();
    }
}

/// Foreign-implemented data fetcher. When attached, the sidecar task
/// gives this fetcher the utterance first; on `Some(Ok(text))` the
/// LLM is skipped entirely and the text becomes the sidecar's answer.
///
/// On `None` (no match) or `Some(Err(e))` (match but fetch failed),
/// the sidecar falls through to the LLM path. Use this for the
/// "fast data path" optimization in quip's playbook: email/calendar
/// queries that are pure data lookups don't need an LLM round trip.
///
/// Implementations are typically host-side (calendar API, mail API)
/// and should already format their output for voice narration —
/// short, no markdown. See `car_voice::format_for_voice` for a
/// minimal helper.
#[async_trait::async_trait]
pub trait DirectDataFetcher: Send + Sync {
    /// Try to satisfy the utterance directly without invoking the LLM.
    ///
    /// - `Some(Ok(text))` → use this as the sidecar answer.
    /// - `Some(Err(msg))` → match recognised but fetch failed; fall
    ///   through to LLM (msg is logged, not surfaced).
    /// - `None` → not a fast-data candidate; fall through to LLM.
    async fn try_fetch(&self, utterance: &str) -> Option<Result<String, String>>;
}

/// Optional telemetry sink for voice turn events. Wraps an
/// `Arc<Mutex<EventLog>>` so emissions are thread-safe across the
/// fast and sidecar tasks. Pass `None` to disable.
#[derive(Clone)]
pub struct VoiceTelemetry {
    log: Arc<Mutex<EventLog>>,
}

impl VoiceTelemetry {
    /// Wrap an existing event log for voice telemetry.
    pub fn new(log: Arc<Mutex<EventLog>>) -> Self {
        Self { log }
    }

    /// Append a voice eventlog entry. Callers in `car-voice`'s
    /// orchestrator emit `VoiceBridgePlayed` and `VoiceSidecarTimedOut`
    /// through this; the engine-side spawn helpers emit the rest.
    pub fn emit(&self, kind: EventKind, turn_id: u64, extra: Vec<(&str, Value)>) {
        let mut data: HashMap<String, Value> = HashMap::new();
        data.insert("turn_id".to_string(), Value::from(turn_id));
        for (k, v) in extra {
            data.insert(k.to_string(), v);
        }
        if let Ok(mut guard) = self.log.lock() {
            guard.append(kind, None, None, data);
        }
    }
}

/// Dispatch a voice-turn utterance with a fast-and-sidecar split.
///
/// `engine` runs both inference calls. `fast_request` is streamed
/// through the returned `VoiceTurnHandle::fast` channel; `sidecar_request`
/// runs to completion and resolves the `sidecar` oneshot once.
///
/// Both inference requests must be pre-composed by the caller — including
/// any voice-context overlay (use [`car_voice::compose_voice_context`]).
/// This crate stays neutral on prompt construction.
///
/// Wraps [`dispatch_voice_turn_with_telemetry`] with no telemetry — use
/// the `_with_telemetry` variant when you want eventlog emissions.
pub fn dispatch_voice_turn(
    engine: Arc<InferenceEngine>,
    utterance: String,
    fast_request: GenerateRequest,
    sidecar_request: GenerateRequest,
) -> VoiceTurnHandle {
    dispatch_voice_turn_with_telemetry(engine, utterance, fast_request, sidecar_request, None)
}

/// Same as [`dispatch_voice_turn`] but emits voice eventlog kinds
/// (`VoiceFastTurnStarted`, `VoiceFastTurnEnded`, `VoiceSidecarResolved`,
/// `VoiceSidecarFailed`, `VoiceSidecarTimedOut`, `VoiceTurnCancelled`)
/// to the supplied telemetry sink as the turn progresses.
pub fn dispatch_voice_turn_with_telemetry(
    engine: Arc<InferenceEngine>,
    _utterance: String,
    fast_request: GenerateRequest,
    sidecar_request: GenerateRequest,
    telemetry: Option<VoiceTelemetry>,
) -> VoiceTurnHandle {
    let turn_id = next_turn_id();
    let cancel = CancellationToken::new();
    let (fast_tx, fast_rx) = mpsc::channel::<StreamEvent>(64);
    let (sidecar_tx, sidecar_rx) = oneshot::channel();

    if let Some(t) = telemetry.as_ref() {
        t.emit(EventKind::VoiceFastTurnStarted, turn_id, vec![]);
    }

    spawn_fast_task(
        engine.clone(),
        fast_request,
        fast_tx,
        cancel.clone(),
        turn_id,
        telemetry.clone(),
    );
    spawn_sidecar_task(
        engine,
        sidecar_request,
        sidecar_tx,
        cancel.clone(),
        turn_id,
        telemetry,
    );

    VoiceTurnHandle {
        control: VoiceTurnControl { turn_id, cancel },
        fast: fast_rx,
        sidecar: sidecar_rx,
    }
}

/// Variant of [`dispatch_voice_turn`] that skips the fast inference call.
///
/// Use when the caller has already played a hardcoded bridge phrase
/// (the "STRUCTURAL HALLUCINATION FIX" in quip's terms) and only needs
/// the sidecar's substantive answer. The returned handle's `fast`
/// channel is pre-closed so callers don't have to special-case the
/// shape.
pub fn dispatch_voice_turn_sidecar_only(
    engine: Arc<InferenceEngine>,
    utterance: String,
    sidecar_request: GenerateRequest,
) -> VoiceTurnHandle {
    dispatch_voice_turn_sidecar_only_with_telemetry(engine, utterance, sidecar_request, None)
}

/// Same as [`dispatch_voice_turn_sidecar_only`] but emits voice
/// eventlog kinds for the sidecar half of the turn.
pub fn dispatch_voice_turn_sidecar_only_with_telemetry(
    engine: Arc<InferenceEngine>,
    utterance: String,
    sidecar_request: GenerateRequest,
    telemetry: Option<VoiceTelemetry>,
) -> VoiceTurnHandle {
    dispatch_voice_turn_sidecar_only_with_classifier(
        engine,
        utterance,
        sidecar_request,
        None,
        telemetry,
    )
}

/// Variant of [`dispatch_voice_turn_sidecar_only`] that consults a
/// [`DirectDataFetcher`] before invoking the LLM. On a fetcher hit
/// the LLM call is skipped entirely — the fetcher's text becomes
/// the sidecar's `SidecarResult`. On miss, falls through to the
/// LLM-based sidecar exactly like the bare `_with_telemetry` variant.
pub fn dispatch_voice_turn_sidecar_only_with_classifier(
    engine: Arc<InferenceEngine>,
    utterance: String,
    sidecar_request: GenerateRequest,
    fetcher: Option<Arc<dyn DirectDataFetcher>>,
    telemetry: Option<VoiceTelemetry>,
) -> VoiceTurnHandle {
    let turn_id = next_turn_id();
    let cancel = CancellationToken::new();
    // Closed channel — recv() returns None immediately. Capacity 1 is
    // arbitrary; we never send anything.
    let (fast_tx, fast_rx) = mpsc::channel::<StreamEvent>(1);
    drop(fast_tx);
    let (sidecar_tx, sidecar_rx) = oneshot::channel();

    spawn_sidecar_task_classified(
        engine,
        utterance,
        sidecar_request,
        sidecar_tx,
        cancel.clone(),
        turn_id,
        fetcher,
        telemetry,
    );

    VoiceTurnHandle {
        control: VoiceTurnControl { turn_id, cancel },
        fast: fast_rx,
        sidecar: sidecar_rx,
    }
}

fn spawn_fast_task(
    engine: Arc<InferenceEngine>,
    request: GenerateRequest,
    out: mpsc::Sender<StreamEvent>,
    cancel: CancellationToken,
    turn_id: u64,
    telemetry: Option<VoiceTelemetry>,
) {
    tokio::spawn(async move {
        let cancelled_during = tokio::select! {
            biased;
            _ = cancel.cancelled() => {
                tracing::debug!(turn_id, "fast task cancelled before inference start");
                true
            }
            res = engine.generate_tracked_stream(request) => {
                match res {
                    Ok(mut rx) => {
                        relay_fast_stream(&mut rx, &out, &cancel, turn_id).await;
                        cancel.is_cancelled()
                    }
                    Err(e) => {
                        tracing::error!(turn_id, error=%e, "fast turn inference failed");
                        false
                    }
                }
            }
        };
        if let Some(t) = telemetry {
            if cancelled_during {
                t.emit(
                    EventKind::VoiceTurnCancelled,
                    turn_id,
                    vec![("track", "fast".into())],
                );
            } else {
                t.emit(EventKind::VoiceFastTurnEnded, turn_id, vec![]);
            }
        }
    });
}

async fn relay_fast_stream(
    rx: &mut mpsc::Receiver<StreamEvent>,
    out: &mpsc::Sender<StreamEvent>,
    cancel: &CancellationToken,
    turn_id: u64,
) {
    loop {
        tokio::select! {
            biased;
            _ = cancel.cancelled() => {
                tracing::debug!(turn_id, "fast stream cancelled mid-relay");
                break;
            }
            evt = rx.recv() => match evt {
                Some(e) => {
                    if out.send(e).await.is_err() {
                        // Receiver dropped — caller has stopped listening.
                        break;
                    }
                }
                None => break,
            }
        }
    }
}

fn spawn_sidecar_task_classified(
    engine: Arc<InferenceEngine>,
    utterance: String,
    request: GenerateRequest,
    sender: oneshot::Sender<Result<SidecarResult, VoiceTurnError>>,
    cancel: CancellationToken,
    turn_id: u64,
    fetcher: Option<Arc<dyn DirectDataFetcher>>,
    telemetry: Option<VoiceTelemetry>,
) {
    tokio::spawn(async move {
        // Try the direct fetcher first. On hit, skip the LLM entirely.
        if let Some(f) = fetcher.as_ref() {
            let fetch_outcome = tokio::select! {
                biased;
                _ = cancel.cancelled() => None,
                outcome = f.try_fetch(&utterance) => outcome,
            };
            match fetch_outcome {
                Some(Ok(text)) => {
                    let result = Ok(SidecarResult {
                        turn_id,
                        text: text.clone(),
                        data: None,
                    });
                    if let Some(t) = telemetry {
                        t.emit(
                            EventKind::VoiceSidecarResolved,
                            turn_id,
                            vec![
                                ("text_len", Value::from(text.len())),
                                ("source", "direct_fetch".into()),
                            ],
                        );
                    }
                    let _ = sender.send(result);
                    return;
                }
                Some(Err(e)) => {
                    tracing::debug!(turn_id, error=%e, "DirectDataFetcher errored; falling through to LLM");
                }
                None => { /* no match — fall through */ }
            }
            // If the fetch attempt was cancelled (the cancel branch ran),
            // honour that immediately rather than starting an LLM call.
            if cancel.is_cancelled() {
                let _ = sender.send(Err(VoiceTurnError::Cancelled));
                if let Some(t) = telemetry {
                    t.emit(
                        EventKind::VoiceTurnCancelled,
                        turn_id,
                        vec![("track", "sidecar".into())],
                    );
                }
                return;
            }
        }
        run_llm_sidecar(engine, request, sender, cancel, turn_id, telemetry).await;
    });
}

async fn run_llm_sidecar(
    engine: Arc<InferenceEngine>,
    request: GenerateRequest,
    sender: oneshot::Sender<Result<SidecarResult, VoiceTurnError>>,
    cancel: CancellationToken,
    turn_id: u64,
    telemetry: Option<VoiceTelemetry>,
) {
    let result = tokio::select! {
        biased;
        _ = cancel.cancelled() => Err(VoiceTurnError::Cancelled),
        res = engine.generate(request) => {
            res.map(|text| SidecarResult { turn_id, text, data: None })
               .map_err(|e| VoiceTurnError::Inference(e.to_string()))
        }
    };
    if let Some(t) = telemetry {
        match &result {
            Ok(r) => t.emit(
                EventKind::VoiceSidecarResolved,
                turn_id,
                vec![("text_len", Value::from(r.text.len()))],
            ),
            Err(VoiceTurnError::Cancelled) => {
                t.emit(
                    EventKind::VoiceTurnCancelled,
                    turn_id,
                    vec![("track", "sidecar".into())],
                );
            }
            Err(VoiceTurnError::Inference(e)) => {
                t.emit(
                    EventKind::VoiceSidecarFailed,
                    turn_id,
                    vec![("error", Value::from(e.clone()))],
                );
            }
        }
    }
    let _ = sender.send(result);
}

fn spawn_sidecar_task(
    engine: Arc<InferenceEngine>,
    request: GenerateRequest,
    sender: oneshot::Sender<Result<SidecarResult, VoiceTurnError>>,
    cancel: CancellationToken,
    turn_id: u64,
    telemetry: Option<VoiceTelemetry>,
) {
    tokio::spawn(run_llm_sidecar(
        engine, request, sender, cancel, turn_id, telemetry,
    ));
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn turn_ids_are_monotonic_and_unique() {
        let a = next_turn_id();
        let b = next_turn_id();
        let c = next_turn_id();
        assert!(b > a);
        assert!(c > b);
    }

    #[test]
    fn control_cancel_is_observable() {
        let control = VoiceTurnControl {
            turn_id: 42,
            cancel: CancellationToken::new(),
        };
        assert!(!control.is_cancelled());
        let clone = control.clone();
        clone.cancel();
        assert!(control.is_cancelled());
    }

    #[test]
    fn handle_turn_id_delegates_to_control() {
        let (_tx, fast_rx) = mpsc::channel::<StreamEvent>(1);
        let (_stx, sidecar_rx) = oneshot::channel();
        let handle = VoiceTurnHandle {
            control: VoiceTurnControl {
                turn_id: 7,
                cancel: CancellationToken::new(),
            },
            fast: fast_rx,
            sidecar: sidecar_rx,
        };
        assert_eq!(handle.turn_id(), 7);
        assert!(!handle.control.is_cancelled());
        handle.cancel();
        assert!(handle.control.is_cancelled());
    }

    #[tokio::test]
    async fn closed_fast_channel_recv_is_none() {
        // Mirrors the channel shape that `dispatch_voice_turn_sidecar_only`
        // produces — a pre-closed mpsc so callers can `.recv()` without
        // special-casing the sidecar-only flavour.
        let (fast_tx, mut fast_rx) = mpsc::channel::<StreamEvent>(1);
        drop(fast_tx);
        assert!(fast_rx.recv().await.is_none());
    }

    #[tokio::test]
    async fn cancellation_propagates_to_relay_fast_stream() {
        // Wire a producer rx, an out channel, and a cancellation token.
        // Cancelling the token should stop the relay even while the
        // producer is still alive.
        let (in_tx, mut in_rx) = mpsc::channel::<StreamEvent>(8);
        let (out_tx, mut out_rx) = mpsc::channel::<StreamEvent>(8);
        let cancel = CancellationToken::new();

        // Producer pushes until the channel fills. Bounded mpsc back-pressure
        // means it will block; we cancel the relay before it drains.
        let producer = tokio::spawn(async move {
            for i in 0..100u32 {
                if in_tx
                    .send(StreamEvent::TextDelta(format!("d{i}")))
                    .await
                    .is_err()
                {
                    break;
                }
            }
        });

        let cancel_clone = cancel.clone();
        let relay = tokio::spawn(async move {
            relay_fast_stream(&mut in_rx, &out_tx, &cancel_clone, 1).await;
        });

        // Drain at least one event, then cancel.
        let first = out_rx.recv().await.expect("first event");
        match first {
            StreamEvent::TextDelta(_) => {}
            other => panic!("unexpected event: {other:?}"),
        }
        cancel.cancel();

        // Relay should exit promptly.
        tokio::time::timeout(std::time::Duration::from_secs(1), relay)
            .await
            .expect("relay did not exit after cancel")
            .expect("relay panicked");

        producer.abort();
    }

    #[tokio::test]
    async fn direct_fetcher_hit_skips_llm_and_resolves_sidecar() {
        struct Hit;
        #[async_trait::async_trait]
        impl DirectDataFetcher for Hit {
            async fn try_fetch(&self, _u: &str) -> Option<Result<String, String>> {
                Some(Ok("3 emails: Bob, Alice, Carol".to_string()))
            }
        }
        let cancel = CancellationToken::new();
        let (tx, rx) = oneshot::channel();
        let log = Arc::new(Mutex::new(EventLog::new()));
        let telemetry = VoiceTelemetry::new(log.clone());
        // We never reach the engine — its generate() would panic on a
        // bare InferenceEngine without backends. The fetcher hit path
        // must short-circuit before that.
        let dummy_engine = Arc::new(car_inference::InferenceEngine::new(
            car_inference::InferenceConfig::default(),
        ));
        spawn_sidecar_task_classified(
            dummy_engine,
            "any new email today".to_string(),
            GenerateRequest::default(),
            tx,
            cancel,
            99,
            Some(Arc::new(Hit)),
            Some(telemetry),
        );
        let r = rx.await.expect("oneshot delivered").expect("ok");
        assert_eq!(r.turn_id, 99);
        assert_eq!(r.text, "3 emails: Bob, Alice, Carol");
        // Resolved event tagged source=direct_fetch.
        let g = log.lock().unwrap();
        let evt = g.events().last().expect("event emitted");
        assert_eq!(evt.kind, EventKind::VoiceSidecarResolved);
        assert_eq!(evt.data.get("source"), Some(&Value::from("direct_fetch")));
    }

    #[tokio::test]
    async fn direct_fetcher_miss_falls_through_but_we_observe_no_short_circuit() {
        // We only verify that the fetch path enters the LLM branch by
        // confirming the spawn task hasn't resolved the oneshot via the
        // fetcher hit path. We can't drive a real LLM here, so we
        // cancel quickly and observe the Cancelled error instead of a
        // SidecarResult — proves the fetcher's None did not produce a
        // direct-resolve.
        struct Miss;
        #[async_trait::async_trait]
        impl DirectDataFetcher for Miss {
            async fn try_fetch(&self, _u: &str) -> Option<Result<String, String>> {
                None
            }
        }
        let cancel = CancellationToken::new();
        let (tx, rx) = oneshot::channel();
        let dummy_engine = Arc::new(car_inference::InferenceEngine::new(
            car_inference::InferenceConfig::default(),
        ));
        spawn_sidecar_task_classified(
            dummy_engine,
            "what's the weather".to_string(),
            GenerateRequest::default(),
            tx,
            cancel.clone(),
            100,
            Some(Arc::new(Miss)),
            None,
        );
        // Cancel before the LLM has a chance to fail; we expect Cancelled.
        cancel.cancel();
        match rx.await.expect("oneshot delivered") {
            Err(VoiceTurnError::Cancelled) => {}
            other => panic!("expected Cancelled after fetcher miss + cancel, got {other:?}"),
        }
    }

    #[test]
    fn telemetry_emit_appends_to_eventlog() {
        let log = Arc::new(Mutex::new(EventLog::new()));
        let telemetry = VoiceTelemetry::new(log.clone());
        telemetry.emit(EventKind::VoiceFastTurnStarted, 7, vec![]);
        telemetry.emit(
            EventKind::VoiceSidecarResolved,
            7,
            vec![("text_len", Value::from(42usize))],
        );
        let g = log.lock().unwrap();
        let events = g.events();
        assert_eq!(events.len(), 2);
        assert_eq!(events[0].kind, EventKind::VoiceFastTurnStarted);
        assert_eq!(events[0].data.get("turn_id"), Some(&Value::from(7u64)));
        assert_eq!(events[1].kind, EventKind::VoiceSidecarResolved);
        assert_eq!(events[1].data.get("text_len"), Some(&Value::from(42usize)));
    }

    #[tokio::test]
    async fn dropped_out_channel_stops_relay_without_cancel() {
        let (in_tx, mut in_rx) = mpsc::channel::<StreamEvent>(8);
        let (out_tx, out_rx) = mpsc::channel::<StreamEvent>(8);
        let cancel = CancellationToken::new();

        // Caller drops their receiver — the relay should exit on the
        // next send attempt.
        drop(out_rx);

        let cancel_clone = cancel.clone();
        let relay = tokio::spawn(async move {
            relay_fast_stream(&mut in_rx, &out_tx, &cancel_clone, 1).await;
        });

        // Push one event so the relay's send fails.
        in_tx
            .send(StreamEvent::TextDelta("x".into()))
            .await
            .unwrap();

        tokio::time::timeout(std::time::Duration::from_secs(1), relay)
            .await
            .expect("relay did not exit after out_rx drop")
            .expect("relay panicked");
    }
}