// crates/rutster/tests/barge_in_integration.rs
//
// PRIMARY-path barge-in e2e proof (slice-4 spec §3.4, §6.1, §7 done-criteria #8).
// SECONDARY-path barge-in e2e proof (slice-4 spec §3.1, §6.1, §7 done-criteria #9).
//
// The PRIMARY test constructs the reflex stack directly and proves that loud
// caller audio kills playout with only the local VAD as the trigger source — no
// brain advisory is involved. This is wedge #1 from README:98-100 and
// ARCHITECTURE.md:79-81.
//
// The SECONDARY test exercises slice-3's advisory plumbing end-to-end:
// MockRealtimeBrain -> translator -> tap protocol -> tap_client -> advisory_tx ->
// Reflex. The local VAD is intentionally kept silent (zeroed caller frames), so
// the kill can only be advisory-driven.

use std::sync::atomic::Ordering;
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::mock::{AdvisoryKind, AdvisoryTrigger, MockRealtimeBrain};
use rutster_brain_realtime::openai_client::run_openai_pump;
use rutster_call_model::ChannelId;
use rutster_media::{AudioSink, AudioSource, PcmFrame, VAD_DEBOUNCE_FRAMES};
use rutster_media::{LocalVadReflex, Reflex, ReflexMetrics};
use rutster_tap::{DecodedPayload, TapAudioPipe, TapMetrics, 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;

// === Brain-shim helpers (duplicated from realtime_integration.rs) ===
// The brain process's accept loop is inlined here so this test exercises the
// real OpenAI-client pump (`run_openai_pump`) against the mock brain without
// spawning a subprocess or depending on private helpers in another integration-
// test binary.

/// Handle returned by `start_brain_shim`. Drop to tear down.
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.
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,
    }
}

/// Accept loop: spawns a per-connection task for each tap WS dial.
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: handshake, split sink+stream, dial the mock
/// OpenAI side, and run `run_openai_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");

    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?;

    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;
            }
        }
    });

    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");

    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(())
}

#[tokio::test]
async fn primary_path_local_vad_kills_playout_without_brain() {
    let (tx_pcm_in, _rx_pcm_in) = mpsc::channel(32);
    let (tx_audio_out, rx_audio_out) = mpsc::channel(32);
    let tap_metrics = TapMetrics::new();
    let pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, tap_metrics);

    let (advisory_tx, advisory_rx) = mpsc::channel(16);
    let reflex_metrics = ReflexMetrics::new();
    let reflex = Reflex::new(pipe, advisory_rx, reflex_metrics.clone());
    let mut stack = LocalVadReflex::new(reflex, advisory_tx);

    // Pre-load a brain audio_out frame into the ring (drain via next_pcm_frame).
    tx_audio_out.send(PcmFrame::zeroed()).await.unwrap();
    let _ = stack.next_pcm_frame(); // drain into ring
    stack.next_pcm_frame(); // pop one (brain is playing)

    // Loud caller audio × N → local VAD trips.
    let mut loud = PcmFrame::zeroed();
    for s in loud.samples.iter_mut() {
        *s = 1000;
    }
    for _ in 0..VAD_DEBOUNCE_FRAMES {
        stack.on_pcm_frame(loud.clone());
    }

    // Next playout tick: kill applied, None returned (ring cleared).
    let f = stack.next_pcm_frame();
    assert!(
        f.is_none(),
        "primary-path local VAD must kill playout within 1 tick"
    );
    assert_eq!(
        reflex_metrics
            .barge_in_count
            .load(std::sync::atomic::Ordering::Relaxed),
        1
    );

    // Fresh brain audio_out → resume.
    tx_audio_out.send(PcmFrame::zeroed()).await.unwrap();
    let resumed = stack.next_pcm_frame();
    assert!(
        resumed.is_some(),
        "first fresh audio_out post-barge must resume playout"
    );
}

#[tokio::test]
async fn secondary_path_brain_advisory_kills_playout_as_backstop() {
    let _ = tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| "rutster=info".into()),
        )
        .try_init();

    let mut mock = MockRealtimeBrain::start().await.expect("mock bind ok");
    // Emit speech_started after the mock has received 2 audio_in frames.
    mock.set_advisory_schedule(vec![AdvisoryTrigger {
        after_audio_in_frames: 2,
        event: AdvisoryKind::SpeechStarted,
    }])
    .await;

    let shim = start_brain_shim(mock.url()).await;
    let tap_url = Url::parse(&format!("ws://{}/", shim.addr)).unwrap();

    let app_state = AppState::default();
    let session_id = ChannelId::new();
    let (advisory_tx, advisory_rx) = mpsc::channel::<rutster_media::AdvisoryEvent>(16);
    let reflex_metrics = ReflexMetrics::new();
    let (pipe, conn) = spawn_tap_engine(session_id, tap_url, app_state, advisory_tx.clone());
    // Compose the same stack the media thread wires on Connected:
    // LocalVadReflex (primary VAD) wrapping Reflex (advisory state machine).
    let reflex = Reflex::new(pipe, advisory_rx, reflex_metrics.clone());
    let mut stack = LocalVadReflex::new(reflex, advisory_tx);

    // Give the engine a moment to complete the tap handshake and dial the mock.
    tokio::time::sleep(Duration::from_millis(50)).await;

    let quiet = PcmFrame::zeroed();

    // Step 1 — prove the brain→core audio_out path is alive. Send one quiet
    // frame (below the local VAD threshold, so it cannot trip the primary
    // trigger) and wait for the mock's canned response.audio.delta.
    stack.on_pcm_frame(quiet.clone());
    let mut audio_path_alive = false;
    for _ in 0..25 {
        if stack.next_pcm_frame().is_some() {
            audio_path_alive = true;
            break;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    assert!(
        audio_path_alive,
        "mock brain must deliver at least one audio_out frame before the kill"
    );

    // Step 2 — trigger the advisory schedule. The mock counts
    // input_audio_buffer.append events; this second append reaches the
    // threshold and causes the mock to emit speech_started unprompted.
    stack.on_pcm_frame(quiet.clone());

    // Step 3 — bounded wait for the advisory to cross the real plumbing and
    // kill playout.
    let kill_deadline = tokio::time::Instant::now() + Duration::from_secs(2);
    let mut killed = false;
    while tokio::time::Instant::now() < kill_deadline {
        stack.next_pcm_frame();
        if reflex_metrics.barge_in_count.load(Ordering::Relaxed) == 1 {
            killed = true;
            break;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    assert!(
        killed,
        "brain advisory must trigger barge-in within the bounded window"
    );
    assert_eq!(
        reflex_metrics.barge_in_count.load(Ordering::Relaxed),
        1,
        "advisory-driven barge-in must fire exactly once"
    );

    // Playout is killed: even though the ring may still have brain frames in
    // flight, barge_in_flush cleared them before the Reflex returned None.
    assert!(
        stack.next_pcm_frame().is_none(),
        "playout must stay killed after the advisory"
    );

    // Step 4 — resume. Another quiet frame prompts the mock to send a fresh
    // audio_out delta; the first post-barge audio_out un-mutes playout.
    let resume_deadline = tokio::time::Instant::now() + Duration::from_secs(2);
    let mut resumed = false;
    while tokio::time::Instant::now() < resume_deadline {
        stack.on_pcm_frame(quiet.clone());
        if stack.next_pcm_frame().is_some() {
            resumed = true;
            break;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    assert!(
        resumed,
        "first fresh audio_out post-barge must resume playout"
    );

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