//! # Slice-3 integration test — realtime brain end-to-end (spec §7.4 + §7.5)
//!
//! Drives the full stack **in-process** (no subprocess, no real OpenAI):
//! 1. **`MockRealtimeBrain`** — the in-process fake OpenAI Realtime WS server
//!    (`rutster_brain_realtime::MockRealtimeBrain`). Asserts the S4
//!    `turn_detection: null` invariant on `session.update` (spec §4.3).
//! 2. **A brain-process-equivalent task** running inside the test process.
//!    It replicates `crates/rutster-brain-realtime/src/main.rs`'s tap-side
//!    accept loop: WS handshake (hello), split sink+stream, dial
//!    `mock.url()` (the fake OpenAI), then run `run_openai_pump` to bridge
//!    tap frames ↔ OpenAI events. We inline this here (rather than refactoring
//!    `main.rs` to expose a library function) to keep the slice-3 surface
//!    change small — `run_openai_pump` is the load-bearing public API we
//!    consume; the ~30-line accept loop is glue.
//! 3. **`spawn_tap_engine`** (from `rutster::tap_engine`) — the core's
//!    per-session TapEngine, dialed against (2) by passing the brain's WS
//!    URL as `tap_url`. The `TapAudioPipe` it returns is the seam object
//!    we push PCM into + drain `audio_out` frames from.
//! 4. **Function-call dispatch** — removed from the slice-4 command-channel
//!    rewire; the `HangupTool` is still registered by `spawn_tap_engine`,
//!    but the old poll-task drain is gone, so this file does not exercise
//!    the hangup path end-to-end.
//!
//! # Why this isn't `cargo run -p rutster-brain-realtime --features=mock`
//! + a browser tab (spec §7.4 manual plan)
//!
//! Per AGENTS.md, browser-driven e2e (Selenium / Playwright / real
//! WebRTC peer) is **deferred post-slice-1**. This test exercises the
//! brain↔core wiring through the actual translator + protocol code with
//! zero browser dependencies — str0m's media loop is bypassed by pushing
//! `PcmFrame`s directly through the seam object (`TapAudioPipe`), the
//! same pattern slice-2's `tap_integration.rs` established.
//!
//! # Done-criteria coverage (spec §7.5)
//!
//! - #3 (mock-brain reply within ~250 ms) — covered by
//!   `audio_round_trip_pushes_pcm_and_receives_canned_response` (bounded
//!   to ~2 s to fail fast; the actual mock replies within ms).
//! - #5 (rutster-brain-realtime interop against the core) — covered by
//!   the test wiring `run_openai_pump` against `MockRealtimeBrain`.
//! - #6 (seam test) — N/A in this file; `loop_driver.rs` + `rtc_session.rs`
//!   byte-identity is a `git diff` gate, not a runtime assertion. Honored
//!   by not touching those files.
//! - #7 (S4 `turn_detection: null`) — covered by `mock`'s own green-path
//!   assertion (`MockRealtimeBrain::handle_connection` rejects a
//!   non-null `turn_detection` with a typed OpenAI-shaped error). This
//!   test exercises the brain→mock path so the assertion fires
//!   end-to-end: if `run_openai_pump` ever sends a non-null
//!   `turn_detection`, the mock closes the WS + the brain pump errors
//!   out, breaking the audio round-trip below.

use std::time::Duration;

use futures_util::{SinkExt, StreamExt};
use rutster::session_map::AppState;
use rutster::tap_engine::spawn_tap_engine;
use rutster_brain_realtime::{MockRealtimeBrain, openai_client::run_openai_pump};
use rutster_call_model::ChannelId;
use rutster_media::{AudioSink, AudioSource, PcmFrame};
use rutster_tap::{DecodedPayload, decode_envelope};
use tokio::net::TcpListener;
use tokio::sync::mpsc;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
use tracing::info;
use url::Url;

/// The handle returned by `start_brain_shim`. Drop to tear down: aborts the
/// accept-loop task + sends the shutdown signal. Mirrors the shape of
/// `MockRealtimeBrain`'s own handle so the test body can hold both + drop
/// both at the end.
struct BrainShim {
    addr: std::net::SocketAddr,
    shutdown: Option<tokio::sync::oneshot::Sender<()>>,
    join: tokio::task::JoinHandle<()>,
}

impl Drop for BrainShim {
    fn drop(&mut self) {
        if let Some(tx) = self.shutdown.take() {
            let _ = tx.send(());
        }
        self.join.abort();
    }
}

/// Start an in-process brain-process-equivalent WS server. Accepts tap WS
/// connections from `spawn_tap_engine`, does the hello handshake, dials the
/// MockRealtimeBrain, and runs `run_openai_pump` to bridge both legs.
/// Bound to an ephemeral port so the test never collides.
///
/// This is the body of `crates/rutster-brain-realtime/src/main.rs`'s
/// `main()` + `handle_tap_connection`, relocated into a test-local helper
/// (so the test doesn't need to spawn a subprocess for the brain binary).
async fn start_brain_shim(mock_url: String) -> BrainShim {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
    let join = tokio::spawn(async move {
        brain_accept_loop(listener, shutdown_rx, mock_url).await;
    });
    BrainShim {
        addr,
        shutdown: Some(shutdown_tx),
        join,
    }
}

/// The accept loop: spawns a per-connection task for each tap WS dial.
/// Selects against the shutdown signal so dropping the handle stops the
/// loop (matches `MockRealtimeBrain::accept_loop`'s shape).
async fn brain_accept_loop(
    listener: TcpListener,
    mut shutdown: tokio::sync::oneshot::Receiver<()>,
    mock_url: String,
) {
    loop {
        tokio::select! {
            biased;
            _ = &mut shutdown => {
                info!("brain_shim accept loop shutting down");
                return;
            }
            res = listener.accept() => {
                let Ok((stream, peer)) = res else { continue };
                let url = mock_url.clone();
                tokio::spawn(async move {
                    if let Err(e) = handle_tap_connection(stream, peer, &url).await {
                        info!(%peer, error = ?e, "brain_shim connection ended");
                    }
                });
            }
        }
    }
}

/// Handle one tap WS connection. Same shape as
/// `rutster_brain_realtime::main::handle_tap_connection`: hello handshake,
/// split sink+stream, dial the OpenAI (mock) side, run the pump.
async fn handle_tap_connection(
    stream: tokio::net::TcpStream,
    peer: std::net::SocketAddr,
    openai_url: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let mut tap_ws = tokio_tungstenite::accept_async(stream).await?;
    info!(%peer, "brain_shim tap WS connection accepted");

    // === Tap handshake: first frame must be hello; send hello ack. ===
    let hello_in = tap_ws
        .next()
        .await
        .ok_or("tap connection closed before hello")??;
    let hello_text = hello_in.into_text().map_err(|_| "hello not text")?;
    let decoded = decode_envelope(&hello_text)?;
    let session_id = match decoded.payload {
        DecodedPayload::Hello(p) => p.session_id,
        _ => return Err("first tap frame not hello".into()),
    };
    info!(%peer, %session_id, "brain_shim tap hello received");
    let ack = rutster_tap::encode_hello(&session_id, 0, 0)?;
    tap_ws.send(Message::Text(ack)).await?;

    // === Two mpsc forwarders between tap WS and the OpenAI pump. ===
    // tap_via: tap frames flowing IN (the core sends audio_in etc.);
    //   pump_tap_tx: pump's outbound tap frames flowing OUT (audio_out,
    //   function_call, etc.) — written to tap_ws by the out_fwd task.
    let (tap_via, pump_tap_rx) = mpsc::channel::<String>(64);
    let (pump_tap_tx, mut tap_out_rx) = mpsc::channel::<String>(64);
    let (mut tap_sink, mut tap_stream) = tap_ws.split();

    let in_fwd = tokio::spawn(async move {
        while let Some(msg_res) = tap_stream.next().await {
            if let Ok(m) = msg_res {
                if let Ok(text) = m.into_text() {
                    if tap_via.send(text).await.is_err() {
                        break;
                    }
                }
            }
        }
    });
    let out_fwd = tokio::spawn(async move {
        while let Some(text) = tap_out_rx.recv().await {
            if tap_sink.send(Message::Text(text)).await.is_err() {
                break;
            }
        }
    });

    // === Dial the OpenAI (mock) side. ===
    let request = openai_url.into_client_request()?;
    let (openai_ws, _resp) = tokio_tungstenite::connect_async(request).await?;
    info!(%peer, %openai_url, "brain_shim OpenAI side connected");

    // Run the pump — this sends `session.update` first (MockRealtimeBrain
    // asserts turn_detection: null per S4), then bridges bidirectionally.
    let pump_result =
        run_openai_pump(openai_ws, pump_tap_rx, pump_tap_tx, "alloy".to_string()).await;
    info!(%peer, ?pump_result, "brain_shim pump exited");
    in_fwd.abort();
    out_fwd.abort();
    Ok(())
}

/// Spin up the full stack: MockRealtimeBrain + brain shim, return the URLs
/// and handles the test body needs. Wrapped in one helper so the test body
/// starts with one line of setup and reads cleanly top-to-bottom.
async fn spin_up_stack() -> (MockRealtimeBrain, BrainShim, Url) {
    let mock = MockRealtimeBrain::start().await.expect("mock bind ok");
    let mock_url = mock.url();
    let shim = start_brain_shim(mock_url).await;
    let tap_url = Url::parse(&format!("ws://{}/", shim.addr)).unwrap();
    (mock, shim, tap_url)
}

/// Push PCM into the core via the TapAudioPipe + bounded-wait for at least
/// one `audio_out` frame to come back. Returns the first frame received.
/// Used by `audio_round_trip_pushes_pcm_and_receives_canned_response`.
async fn push_pcm_and_wait_audio_out(pipe: &mut rutster_tap::TapAudioPipe) -> PcmFrame {
    // The MockRealtimeBrain replies to each `input_audio_buffer.append`
    // with a canned `response.audio.delta` (480 zeroed samples as base64
    // LE i16). The translator turns that into an `audio_out` tap frame,
    // which the engine pumps back into the pipe's playout ring. The
    // round-trip through both WS legs + the translator is bounded in ms;
    // we wait up to 4 s to fail fast without flaking on a slow CI box.
    let mut marker = PcmFrame::zeroed();
    // Non-zero marker so the test can assert "we got THIS frame back" if
    // we ever extend the mock to echo the input. Today the mock returns
    // zeros, but the marker documents intent.
    marker.samples[0] = 42;
    pipe.on_pcm_frame(marker.clone());

    for _ in 0..200 {
        tokio::time::sleep(Duration::from_millis(20)).await;
        if let Some(f) = pipe.next_pcm_frame() {
            return f;
        }
        // The engine's pump loop runs on tokio's shared pool; the playout
        // ring fills as the translate-audio-out cycle runs. Keep pushing
        // occasional frames so the brain has input to react to (mimics
        // a real call's continuous PCM stream).
        pipe.on_pcm_frame(marker.clone());
    }
    panic!("no audio_out frame within 4 s (mock brain should reply in ms)");
}

/// Slice-3 §7.5 #3: audio round-trip through the full stack. Pushes a
/// PCM frame through TapAudioPipe → engine → brain shim → run_openai_pump
/// → translator → `input_audio_buffer.append` → MockRealtimeBrain → canned
/// `response.audio.delta` → translator → `audio_out` tap frame → engine →
/// TapAudioPipe.next_pcm_frame. Catches any break in the audio pipeline.
#[tokio::test]
async fn audio_round_trip_pushes_pcm_and_receives_canned_response() {
    let _ = tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| "rutster=info,rutster_brain_realtime=info".into()),
        )
        .try_init();

    let (_mock, _shim, tap_url) = spin_up_stack().await;
    let app_state = AppState::default();
    let session_id = ChannelId::new();
    let (advisory_tx, _advisory_rx) =
        tokio::sync::mpsc::channel::<rutster_media::AdvisoryEvent>(16);
    let (mut pipe, conn) = spawn_tap_engine(session_id, tap_url, app_state, advisory_tx);

    let frame = push_pcm_and_wait_audio_out(&mut pipe).await;
    // MockRealtimeBrain sends 480 zeroed samples per response.audio.delta.
    // The translator decodes back into a PcmFrame of SAMPLES_PER_FRAME=480.
    assert_eq!(frame.samples.len(), 480, "frame should be 480 samples");

    // Teardown — close the engine + the test-spin handles drop on scope exit.
    let _ = conn.close_tx.send(());
    conn.join.abort();
}

/// Slice-3 §7.5 #7 (S4 end-to-end): the brain sends a `session.update`
/// with `turn_detection: null` on startup. The MockRealtimeBrain
/// rejects a non-null `turn_detection` with a typed OpenAI-shaped error
/// so the brain's pump would fail. Our end-to-end stack coming up
/// without errors IS the S4 end-to-end assertion (the mock would have
/// closed the OpenAI WS otherwise). This test reifies that contract:
/// audio round-trips only succeed because the brain's session.update
/// had turn_detection: null.
///
/// (The MockRealtimeBrain already has
/// `accepts_session_update_with_turn_detection_null` +
/// `rejects_session_update_with_turn_detection_set` as unit tests in
/// `mock.rs`. This test exercises the same path through the brain's
/// actual `run_openai_pump` handshake.)
#[tokio::test]
async fn s4_brain_sends_session_update_with_turn_detection_null_end_to_end() {
    let _ = tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| "rutster=info,rutster_brain_realtime=info".into()),
        )
        .try_init();

    let (_mock, _shim, tap_url) = spin_up_stack().await;
    let app_state = AppState::default();
    let session_id = ChannelId::new();
    let (advisory_tx, _advisory_rx) =
        tokio::sync::mpsc::channel::<rutster_media::AdvisoryEvent>(16);
    let (mut pipe, conn) = spawn_tap_engine(session_id, tap_url, app_state, advisory_tx);

    // If the brain sent turn_detection != null, MockRealtimeBrain would
    // close the OpenAI WS — the brain's pump would exit + the engine's
    // reconnect_attempts would increment. We assert the OPPOSITE: audio
    // round-trips, which is only possible if the pump stayed up. Bounded
    // to ~4 s to fail fast.
    let frame = push_pcm_and_wait_audio_out(&mut pipe).await;
    assert_eq!(frame.samples.len(), 480);

    // Reconnect attempts remain 0 — the engine didn't enter backoff,
    // which would have happened if the mock WS had closed.
    use std::sync::atomic::Ordering;
    let attempts = conn.metrics.reconnect_attempts.load(Ordering::Relaxed);
    assert_eq!(
        attempts, 0,
        "engine should not have reconnected (S4 honored)"
    );

    let _ = conn.close_tx.send(());
    conn.join.abort();
}
