tick.rs

// Per-tick pipeline — the core of the channel actor in Local mode.
//
// The recv_loop task reads the socket continuously and pushes RTP/DTMF
// packets to the jitter buffer. This tick pops one per 20ms cycle,
// decodes, runs subsystems (recorder, player, barge-in), and sends
// outbound (echo, player, DTMF, silence).
//
// `run` is intentionally a thin orchestrator — one phase per call — so
// the shape of a tick is legible without paging through 300 lines. Each
// phase is a named helper below. The order matches the C++ addon's
// handletick flow:
//
//   1.  housekeeping  : tick_count, DTLS handshake result poll
//   2.  inbound       : pop from jitter, SRTP-decrypt
//   3.  DTMF classify : rfc2833 packets short-circuit here
//   4.  player        : read next 160-sample frame if playing
//   5.  decode        : wire bytes → narrowband 8 kHz linear
//   6.  barge-in      : interrupt player on loud inbound
//   7.  pre-buffer    : capture audio under playrecord's prompt
//   8.  recorder/readers : write to disk / push to JS readers
//   9.  outbound      : send DTMF / player / echo / nothing
//  10.  idle check    : multi-tier timeouts, matches C++ checkidlerecv

use std::net::SocketAddr;
use std::sync::atomic::Ordering;

use super::actor::{activate_pending_recorder, Event, Subsystems, PREBUFFER_CAPACITY_SAMPLES};
use super::rtp::{self, RtpPacket};
use super::state::ChannelState;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TickOutcome {
    Continue,
    #[allow(dead_code)]
    Handshaking,
    Stop,
}

pub const IDLE_TICK_LIMIT: u64 = 50 * 20; // 20s — soft idle (no RTP with remote confirmed)
pub const HARD_TIMEOUT_NO_REMOTE: u64 = 50 * 60 * 60; // 1hr — remote() never called
pub const HARD_TIMEOUT_NO_RECV: u64 = 50 * 60 * 60 * 2; // 2hr — on hold (recv=false)

// 20 ms at 8 kHz — the canonical narrowband frame length used for
// recorder/reader frames when the inbound side is silent.
const FRAME_SAMPLES: usize = 160;

pub async fn run(state: &mut ChannelState, subs: &mut Subsystems) -> TickOutcome {
    state.tick_count += 1;
    poll_dtls_handshake(state);

    // Match C++ projectrtpchannel::handletick: tsout += G711PAYLOADBYTES
    // (160) at the start of every tick, *unconditionally*. The outbound
    // RTP timestamp is a media clock — if we let it stall during a
    // pause (no player, no echo, no DTMF), the first packet sent after
    // the pause carries a stale timestamp relative to the receiver's
    // expected media clock, and downstream jitter buffers treat it as
    // a late/duplicate packet (silent or stuttered playback at the far
    // end). See ivr_exit_fail_rtp.pcap: 960 ms of wall-clock pause
    // between two outbound audio packets, but ts_delta=160.
    state.out_ts = state.out_ts.wrapping_add(FRAME_SAMPLES as u32);

    // Mirrors C++ projectrtpchannel::handletick: drain every queued
    // RFC-2833 packet first, then process at most one audio packet for
    // the rest of the tick. Some RFC 2833 senders interleave one DTMF
    // packet per audio packet (same SSRC, shared SN counter) so the wire
    // doubles to ~100 pps for the burst's duration. Without this drain,
    // popping just one packet per 50 Hz tick would let the jitter buffer
    // grow by ~10 packets per burst and overflow on the next press —
    // exactly the symptom in ivr_dtmf_issue_9_1_9_9_3.pcap.
    let mut inbound_pkt: Option<RtpPacket> = None;
    let mut popped_any = false;
    loop {
        let Some(pk) = pop_and_decrypt_inbound(state) else {
            break;
        };
        popped_any = true;
        if pk.payload_type() == state.rfc2833_pt {
            classify_dtmf_inbound(state, subs, &pk).await;
            continue;
        }
        inbound_pkt = Some(pk);
        break;
    }
    if popped_any {
        state.ticks_without_rtp = 0;
    } else if state.direction.recv {
        // Match C++ checkidlerecv: freeze the no-RTP counter while
        // recv=false (on hold). Otherwise the counter races past the
        // 20s soft timeout during a hold and the channel is killed
        // with reason="idle-timeout" the moment it's remixed.
        state.ticks_without_rtp += 1;
    }

    let player_frame = tick_player(state, subs).await;
    let writer_frame = tick_writer(state, subs);
    // One slot, two possible sources — writer wins when both are set
    // (the Create handler also nixes any active player so this is
    // belt-and-braces).
    let out_frame = writer_frame.as_deref().or(player_frame.as_deref());

    let decoded = decode_inbound(state, inbound_pkt.as_ref());
    run_bargein(state, subs, decoded.as_deref()).await;
    accumulate_prebuffer(subs, decoded.as_deref());
    feed_recorders_and_readers(state, subs, decoded.as_deref(), out_frame).await;

    send_outbound(state, subs, out_frame, inbound_pkt.as_ref()).await;

    // Periodic RTCP (SR/RR + SDES) on the P+1 control socket. No-op
    // off-interval or until a remote is known.
    super::rtcp_tx::maybe_send_rtcp(state).await;

    check_idle_timeout(state)
}

// ---- phase helpers --------------------------------------------------------

/// Pop one packet from the jitter buffer and SRTP-decrypt in place if an
/// inbound SRTP context is active. Returns `None` when the jitter is
/// empty *or* when decryption fails (bad MAC / replay) — in both cases
/// the tick treats this as "no inbound this cycle".
fn pop_and_decrypt_inbound(state: &mut ChannelState) -> Option<RtpPacket> {
    let mut pkt = state.jitter.lock().pop()?;
    if let Some(ref mut ctx) = state.srtp_decrypt {
        match ctx.decrypt_rtp(pkt.as_slice()) {
            Ok(decrypted) => {
                let n = decrypted.len().min(pkt.buf.len());
                pkt.buf[..n].copy_from_slice(&decrypted[..n]);
                pkt.len = n;
            }
            Err(_) => return None,
        }
    }
    Some(pkt)
}

/// Inspect `pk`'s payload type; if it's the rfc2833 DTMF PT, feed the
/// receiver, fire `telephone-event`, and if the player is interrupt-enabled
/// end it and activate any pending recorder. Returns `true` when the packet
/// was consumed as DTMF (audio processing should be skipped).
async fn classify_dtmf_inbound(
    state: &mut ChannelState,
    subs: &mut Subsystems,
    pk: &RtpPacket,
) -> bool {
    if pk.payload_type() != state.rfc2833_pt {
        return false;
    }
    let sn = pk.sequence_number();
    let ts = pk.timestamp();
    let payload = &pk.as_slice()[rtp::RTP_FIXED_HEADER_LEN..pk.len()];
    if let Some(digit) = subs.dtmf_recv.feed(sn, ts, payload) {
        let mut activate_recorder = false;
        if let Some(p) = subs.player.as_ref() {
            if p.interrupts() {
                subs.player = None;
                subs.bargein = None;
                state.pending_events.push(Event::Play {
                    state: super::actor::PlayState::End,
                    reason: Some("telephone-event".into()),
                });
                activate_recorder = true;
            }
        }
        state.pending_events.push(Event::Telephone { digit });
        if activate_recorder {
            let _ = activate_pending_recorder(subs, &mut state.pending_events).await;
        }
    }
    true
}

/// Read the next 160 samples from the active player, if any. If the
/// player finishes this tick, tear it down, emit `play/end`, and activate
/// any recorder queued by a prior `playrecord`. Returns the frame so
/// downstream phases (recorder write, outbound send) can use it.
async fn tick_player(state: &mut ChannelState, subs: &mut Subsystems) -> Option<Vec<i16>> {
    let mut player_frame: Option<Vec<i16>> = None;
    let mut player_just_ended = false;
    if let Some(player) = subs.player.as_mut() {
        // A paused player holds its position and produces nothing — the tick
        // sends silence and the player cannot finish until resumed.
        if player.is_paused() {
            return None;
        }
        let frame = player.read(FRAME_SAMPLES).await;
        if !frame.samples.is_empty() {
            player_frame = Some(frame.samples);
        }
        if player.is_finished() {
            subs.player = None;
            subs.bargein = None;
            state.pending_events.push(Event::Play {
                state: super::actor::PlayState::End,
                reason: Some("completed".into()),
            });
            player_just_ended = true;
        }
    }
    if player_just_ended {
        // Ordered after the `play/end` event so downstream sees play-end
        // then recording-start in the right sequence.
        let _ = activate_pending_recorder(subs, &mut state.pending_events).await;
    }
    player_frame
}

/// Pull the next 20 ms frame from an active write stream. Sibling of
/// `tick_player` — same role (outbound source) but fed from a JS
/// `Writable` rather than a WAV file. Returns None on underrun (tick
/// will emit silence) and tears the writer down when the JS side has
/// ended AND we've drained the last partial frame.
fn tick_writer(state: &mut ChannelState, subs: &mut Subsystems) -> Option<Vec<i16>> {
    let frame = {
        let w = subs.writer.as_mut()?;
        w.next_frame_8k()
    };
    // Take-out check: if the writer reports drained+ended, retire it and
    // emit a `play/end` so the JS event sink sees symmetry with `play`.
    let drained = subs
        .writer
        .as_ref()
        .is_some_and(|w| w.is_drained_and_ended());
    if drained {
        subs.writer = None;
        state.pending_events.push(Event::Play {
            state: super::actor::PlayState::End,
            reason: Some("completed".into()),
        });
    }
    frame
}

/// Feed the wire bytes into `codecx` and pull the narrowband 8 kHz linear
/// representation. Returns `None` when there's no inbound packet this tick.
/// The result is owned so multiple downstream phases can share it without
/// holding a `&mut codecx` borrow.
fn decode_inbound(state: &mut ChannelState, pkt: Option<&RtpPacket>) -> Option<Vec<i16>> {
    let pk = pkt?;
    state.codecx.feed_wire(pk.payload_type(), pk.payload());
    state.codecx.require_narrowband_8k().map(|s| s.to_vec())
}

/// Smoothed-RMS barge-in check. Only runs when a player is active *and*
/// we've seen enough inbound packets for the moving average to have
/// converged (100 packets ≈ 2 s — same constant the recorder uses).
async fn run_bargein(state: &mut ChannelState, subs: &mut Subsystems, decoded: Option<&[i16]>) {
    let Some(samples) = decoded else {
        return;
    };
    let Some(bi) = subs.bargein.as_mut() else {
        return;
    };
    if subs.player.is_none() {
        return;
    }
    if state.in_count.load(Ordering::Relaxed) < 100 {
        return;
    }

    let mut sum_sq: u64 = 0;
    for s in samples {
        let v = *s as i64;
        sum_sq += (v * v) as u64;
    }
    let rms = if samples.is_empty() {
        0
    } else {
        ((sum_sq / samples.len() as u64) as f64).sqrt() as i32
    };
    let smoothed = bi.power_ma.execute(rms.min(i16::MAX as i32) as i16) as i32;
    if smoothed <= bi.power_threshold {
        return;
    }

    subs.player = None;
    subs.bargein = None;
    state.pending_events.push(Event::Play {
        state: super::actor::PlayState::End,
        reason: Some("interrupted".into()),
    });
    let _ = activate_pending_recorder(subs, &mut state.pending_events).await;
}

/// During the prompt phase of a `playrecord` (player active *and*
/// recorder pending), accumulate inbound samples into the pre-buffer so
/// they can be flushed into the recorder on activation — captures the
/// caller's speech that started before the prompt finished.
fn accumulate_prebuffer(subs: &mut Subsystems, decoded: Option<&[i16]>) {
    if subs.player.is_none() || subs.pending_recorder.is_none() {
        return;
    }
    let Some(samples) = decoded else {
        return;
    };

    let total = subs.prebuffer.len() + samples.len();
    if total > PREBUFFER_CAPACITY_SAMPLES {
        let drop_n = total - PREBUFFER_CAPACITY_SAMPLES;
        for _ in 0..drop_n.min(subs.prebuffer.len()) {
            subs.prebuffer.pop_front();
        }
    }
    for s in samples {
        subs.prebuffer.push_back(*s);
    }
}

/// Resolve inbound / outbound sample slices for the tick (same L/R
/// convention C++ uses for the WAV recorder), then feed every recorder
/// and every audio reader. Garbage-collect readers whose JS consumer has
/// gone away.
///
/// Recorder vs reader semantics:
///   - **Recorder**: writes bytes only when there's *actual* audio
///     (inbound or player). Idle ticks are no-ops, so a channel that's
///     just holding open doesn't inflate the WAV file. Matches C++.
///   - **Reader**: fed every tick, including silence, so STT / caption
///     consumers get a continuous 20 ms stream — their FFT / VAD / ASR
///     pipelines rely on steady framing.
async fn feed_recorders_and_readers(
    state: &mut ChannelState,
    subs: &mut Subsystems,
    decoded: Option<&[i16]>,
    player_frame: Option<&[i16]>,
) {
    // Matches C++ `projectrtpsoundfile.cpp:760` — incodec = inbound
    // decoded, outcodec = player → else echo → else silence. Mono
    // recorders sum both sides; stereo interleaves L=in, R=out.
    let silence = [0i16; FRAME_SAMPLES];
    let in_s: &[i16] = decoded.unwrap_or(&silence);
    let out_s: &[i16] = if let Some(pf) = player_frame {
        pf
    } else if state.echo {
        in_s
    } else {
        &silence
    };

    // Recorder writes only when there's real audio. `state.echo` without
    // inbound produces no samples, so it doesn't qualify.
    let has_recordable_audio = decoded.is_some() || player_frame.is_some();
    if has_recordable_audio {
        if state.codecx.is_wideband() {
            // Native-rate recording on a wideband channel: true wideband
            // inbound (G.722 decode) + the outbound (player/echo/silence)
            // upsampled to 16 kHz so the WAV stays coherent. The recorder's
            // WAV rate is stamped to 16k at open to match.
            let in_wb = state
                .codecx
                .require_wideband_16k()
                .map(|s| s.to_vec())
                .unwrap_or_else(|| upsample_2x(in_s));
            let out_wb = upsample_2x(out_s);
            write_recorder_frames(state, subs, &in_wb, &out_wb).await;
        } else {
            write_recorder_frames(state, subs, in_s, out_s).await;
        }
    }
    // Readers manage their own rate (a 16k reader pulls wideband from codecx);
    // always hand them the 8 kHz narrowband here.
    feed_readers(state, subs, in_s, out_s);
}

/// Linear 2x upsample (8 kHz -> 16 kHz) used for the recorder's outbound leg
/// on a wideband channel, where the player/echo source is narrowband. The
/// inbound leg uses the codec's true wideband decode; this only covers the
/// out side so a stereo native recording stays sample-aligned.
fn upsample_2x(input: &[i16]) -> Vec<i16> {
    if input.is_empty() {
        return Vec::new();
    }
    let mut out = Vec::with_capacity(input.len() * 2);
    for i in 0..input.len() {
        let cur = input[i] as i32;
        let next = if i + 1 < input.len() {
            input[i + 1] as i32
        } else {
            cur
        };
        out.push(cur as i16);
        out.push(((cur + next) / 2) as i16);
    }
    out
}

async fn write_recorder_frames(
    state: &mut ChannelState,
    subs: &mut Subsystems,
    in_s: &[i16],
    out_s: &[i16],
) {
    let chan_in_count = state.in_count.load(Ordering::Relaxed);
    let len = in_s.len().max(out_s.len());

    let mut i = 0;
    while i < subs.recorders.len() {
        let rec = &mut subs.recorders[i];
        let prev_state = rec.state();
        let frame = build_recorder_frame(rec.num_channels(), rec.direction(), in_s, out_s, len);
        // Power calc runs on the narrowband slice of the leg(s) being
        // recorded — inbound for "in"/"both" (matches C++ `codecx::power()`),
        // outbound for "out" so a gated out-only recording can still trigger.
        let power_s: &[i16] = match rec.direction() {
            super::recorder::RecordDirection::Out => out_s,
            _ => in_s,
        };
        let _ = rec.write_frame(&frame, power_s, Some(chan_in_count)).await;

        let new_state = rec.state();
        let file_str = rec.file().to_string_lossy().into_owned();
        if prev_state == super::recorder::RecorderState::Pending
            && new_state == super::recorder::RecorderState::Active
        {
            state.pending_events.push(Event::Record {
                state: super::actor::RecordState::Recording,
                reason: Some("abovepower".into()),
                file: Some(file_str.clone()),
                filesize: None,
            });
        }
        if rec.is_finished() {
            let reason_str = match rec.finish_reason() {
                Some(super::recorder::FinishReason::Completed) => "completed",
                Some(super::recorder::FinishReason::MaxDurationReached) => "timeout",
                Some(super::recorder::FinishReason::BelowPowerThreshold) => "belowpower",
                Some(super::recorder::FinishReason::ChannelClosed) => "channelclosed",
                Some(super::recorder::FinishReason::Requested) => "requested",
                None => "completed",
            };
            let size = rec.file_size();
            state.pending_events.push(Event::Record {
                state: super::actor::RecordState::Finished,
                reason: Some(reason_str.into()),
                file: Some(file_str),
                filesize: Some(size),
            });
            subs.recorders.remove(i);
            continue;
        }
        i += 1;
    }
}

/// Build one WAV frame honouring the recorder's direction.
///
/// direction=Both (default): mono = saturated sum, stereo = interleaved
/// L=in R=out — the legacy call-recording behaviour.
/// direction=In / Out: only that leg is written — mono takes the slice as
/// is, stereo duplicates it across L/R (same convention as the mono-direction
/// audio reader and the mixer's no-peer fallback). "In" is what STT captures
/// want: a concurrent playrecord must not transcribe its own prompt.
fn build_recorder_frame(
    num_channels: u16,
    direction: super::recorder::RecordDirection,
    in_s: &[i16],
    out_s: &[i16],
    len: usize,
) -> Vec<i16> {
    use super::recorder::RecordDirection;
    let single: Option<&[i16]> = match direction {
        RecordDirection::In => Some(in_s),
        RecordDirection::Out => Some(out_s),
        RecordDirection::Both => None,
    };
    if num_channels == 2 {
        let mut v = Vec::with_capacity(len * 2);
        for j in 0..len {
            match single {
                Some(s) => {
                    let val = s.get(j).copied().unwrap_or(0);
                    v.push(val);
                    v.push(val);
                }
                None => {
                    v.push(in_s.get(j).copied().unwrap_or(0));
                    v.push(out_s.get(j).copied().unwrap_or(0));
                }
            }
        }
        v
    } else {
        (0..len)
            .map(|j| match single {
                Some(s) => s.get(j).copied().unwrap_or(0),
                None => {
                    let a = in_s.get(j).copied().unwrap_or(0) as i32;
                    let b = out_s.get(j).copied().unwrap_or(0) as i32;
                    (a + b).clamp(i16::MIN as i32, i16::MAX as i32) as i16
                }
            })
            .collect()
    }
}

fn feed_readers(state: &mut ChannelState, subs: &mut Subsystems, in_s: &[i16], out_s: &[i16]) {
    for reader in subs.readers.iter_mut() {
        reader.feed(&mut state.codecx, Some(in_s), Some(out_s));
    }
    subs.readers.retain(|r| !r.is_closed());
}

/// Priority: DTMF outbound → player frame → echo → nothing. Matches C++
/// `postreadcb` ordering — DTMF wins because the RFC-2833 burst has to
/// ship in 20 ms slots; player beats echo because an explicit prompt
/// always overrides reflection; echo only fires on the tick an inbound
/// packet arrives so it stays in lock-step with the source.
async fn send_outbound(
    state: &mut ChannelState,
    subs: &mut Subsystems,
    player_frame: Option<&[i16]>,
    inbound_pkt: Option<&RtpPacket>,
) {
    if !state.direction.send {
        return;
    }
    let Some(remote) = state.get_remote_addr() else {
        return;
    };

    if let Some(pkt) = subs.dtmf_send.next_event(state.out_ts) {
        send_dtmf(state, &pkt, remote).await;
    } else if let Some(samples) = player_frame {
        send_player_frame(state, samples, remote).await;
    } else if state.echo {
        if let Some(pk) = inbound_pkt {
            send_echo(state, pk, remote).await;
        }
    }
}

/// Multi-tier idle check — matches C++ `checkidlerecv` in
/// `projectrtpchannel.cpp`. Three regimes:
///   * recv=true + remote_confirmed : soft 20 s ceiling on no-RTP
///   * recv=true + not confirmed    : hard 1 h zombie timeout
///   * recv=false (on hold)         : hard 2 h timeout
fn check_idle_timeout(state: &ChannelState) -> TickOutcome {
    if state.direction.recv {
        if state.remote_confirmed {
            if state.ticks_without_rtp >= IDLE_TICK_LIMIT {
                return TickOutcome::Stop;
            }
        } else if state.tick_count >= HARD_TIMEOUT_NO_REMOTE {
            return TickOutcome::Stop;
        }
    } else if state.tick_count >= HARD_TIMEOUT_NO_RECV {
        return TickOutcome::Stop;
    }
    TickOutcome::Continue
}

// ---- DTLS → SRTP plumbing -------------------------------------------------

/// Pick up the DTLS handshake result (if it's arrived) and build the
/// inbound / outbound SRTP contexts. Non-blocking `try_recv` — cheap to
/// call every tick. Must be called from whichever tick loop owns the
/// channel (Local `tick::run`, Mixer `mix_tick`) or the SRTP contexts
/// never get built and audio is silent after handshake.
pub(crate) fn poll_dtls_handshake(state: &mut ChannelState) {
    if let Some(rx) = state.dtls_result_rx.as_mut() {
        match rx.try_recv() {
            Ok(Some(result)) => {
                let keys = super::dtls_session::split_keying_material(
                    &result.keying_material,
                    result.profile,
                    result.is_client,
                );
                let (our_key, our_salt, profile) = super::dtls_session::local_srtp_params(&keys);
                if let Ok(enc) =
                    webrtc_srtp::context::Context::new(our_key, our_salt, profile, None, None)
                {
                    state.srtp_encrypt = Some(enc);
                }
                let (their_key, their_salt, _) = super::dtls_session::remote_srtp_params(&keys);
                if let Ok(dec) =
                    webrtc_srtp::context::Context::new(their_key, their_salt, profile, None, None)
                {
                    state.srtp_decrypt = Some(dec);
                }
                // Hand the keying material to the inbound RTCP loop so it can
                // build its own SRTCP decrypt context (SRTP/SRTCP replay state
                // is independent, so a dedicated context is correct here).
                if let Some(tx) = &state.srtp_key_tx {
                    let _ = tx.send(Some(keys.clone()));
                }
                state.srtp_keys = Some(keys);
                state.dtls_result_rx = None;
                state.dtls_handshake_abort = None;
            }
            Ok(None) => {
                state.dtls_result_rx = None;
                state.dtls_handshake_abort = None;
            }
            Err(tokio::sync::oneshot::error::TryRecvError::Empty) => {}
            Err(tokio::sync::oneshot::error::TryRecvError::Closed) => {
                state.dtls_result_rx = None;
                state.dtls_handshake_abort = None;
            }
        }
    }
}

// ---- outbound send primitives --------------------------------------------

async fn send_rtp(state: &mut ChannelState, pkt: &RtpPacket, remote: SocketAddr) {
    // Never emit plaintext on a channel that negotiated DTLS but has no keys
    // yet — a failed/in-progress handshake must not downgrade to cleartext.
    if state.secure_not_ready() {
        return;
    }
    // Payload octets (excludes the RTP header) — the RTCP SR octet count.
    let octets = pkt.payload_len() as u64;
    if let Some(ref mut ctx) = state.srtp_encrypt {
        if let Ok(encrypted) = ctx.encrypt_rtp(pkt.as_slice()) {
            if state.rtp_sock.send_to(&encrypted, remote).await.is_ok() {
                state.out_count += 1;
                state.out_octets += octets;
            }
        }
    } else if state.rtp_sock.send_to(pkt.as_slice(), remote).await.is_ok() {
        state.out_count += 1;
        state.out_octets += octets;
    }
}

async fn send_dtmf(
    state: &mut ChannelState,
    pkt: &super::dtmf::NextDtmfPacket,
    remote: SocketAddr,
) {
    let mut out = RtpPacket::new();
    out.init(state.ssrc);
    out.set_payload_type(state.rfc2833_pt);
    out.set_sequence_number(state.out_sn);
    // Use the sender-latched burst TS rather than state.out_ts. Within a
    // burst state.out_ts is effectively frozen in this path (DTMF wins
    // over audio), but the latched value is authoritative regardless.
    out.set_timestamp(pkt.timestamp);
    out.set_marker(pkt.marker);
    out.set_payload(&pkt.payload);
    state.out_sn = state.out_sn.wrapping_add(1);
    send_rtp(state, &out, remote).await;
}

async fn send_player_frame(state: &mut ChannelState, samples: &[i16], remote: SocketAddr) {
    // Feed linear samples into the codec bundle, lazily produce the
    // wire encoding for this channel's remote PT. The encoder state
    // (G.722 predictor, iLBC LP, etc.) lives on the bundle and
    // persists across ticks for a coherent outbound stream.
    state.codecx.feed_linear_8k(samples);
    let pt = state.remote_pt;
    let Some(payload) = state.codecx.require_wire_as(pt).map(|b| b.to_vec()) else {
        return;
    };
    let mut out = RtpPacket::new();
    out.init(state.ssrc);
    out.set_payload_type(state.remote_pt);
    out.set_sequence_number(state.out_sn);
    out.set_timestamp(state.out_ts);
    out.set_payload(&payload);
    state.out_sn = state.out_sn.wrapping_add(1);
    // out_ts is advanced once per tick at the top of `run` (mirrors
    // C++ incrtsout). Don't double-advance here.
    send_rtp(state, &out, remote).await;
}

async fn send_echo(state: &mut ChannelState, in_pk: &RtpPacket, remote: SocketAddr) {
    let mut out = RtpPacket::new();
    out.init(state.ssrc);
    out.set_payload_type(in_pk.payload_type());
    out.set_sequence_number(state.out_sn);
    out.set_timestamp(in_pk.timestamp());
    out.set_payload(in_pk.payload());
    state.out_sn = state.out_sn.wrapping_add(1);
    send_rtp(state, &out, remote).await;
}

impl RtpPacket {
    pub fn as_mut_slice_for_fill(&mut self, n: usize) -> &mut [u8] {
        debug_assert!(n <= self.buf.capacity());
        self.len = n;
        &mut self.buf[..n]
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::channel::commands::Direction;
    use crate::channel::state::ChannelState;
    use tokio::net::UdpSocket;

    async fn fresh_state() -> (ChannelState, UdpSocket, SocketAddr) {
        let rtp_sock = UdpSocket::bind("127.0.0.1:0").await.unwrap();
        let rtp_addr = rtp_sock.local_addr().unwrap();
        let rtcp_sock = UdpSocket::bind("127.0.0.1:0").await.unwrap();
        let peer_sock = UdpSocket::bind("127.0.0.1:0").await.unwrap();
        let peer_addr = peer_sock.local_addr().unwrap();
        let mut state = ChannelState::new(1, rtp_addr, rtp_sock, rtcp_sock, 0xC0FFEE);
        state.direction = Direction {
            send: true,
            recv: true,
        };
        (state, peer_sock, peer_addr)
    }

    #[tokio::test]
    async fn tick_skips_send_without_remote() {
        let (mut state, peer_sock, _peer_addr) = fresh_state().await;
        let mut subs = crate::channel::actor::Subsystems::default();
        run(&mut state, &mut subs).await;

        let r = tokio::time::timeout(
            std::time::Duration::from_millis(30),
            peer_sock.recv_from(&mut [0u8; 2000]),
        )
        .await;
        assert!(r.is_err());
    }

    #[tokio::test]
    async fn tick_respects_direction_send_false() {
        let (mut state, peer_sock, peer_addr) = fresh_state().await;
        let mut subs = crate::channel::actor::Subsystems::default();
        state.set_remote_addr(peer_addr);
        state.direction.send = false;

        run(&mut state, &mut subs).await;

        let r = tokio::time::timeout(
            std::time::Duration::from_millis(30),
            peer_sock.recv_from(&mut [0u8; 2000]),
        )
        .await;
        assert!(r.is_err());
    }

    #[tokio::test]
    async fn tick_drains_burst_when_dtmf_interleaved_with_audio() {
        // Regression for ivr_dtmf_issue_9_1_9_9_3.pcap: some senders
        // interleave one DTMF body packet per audio packet within the
        // same SSRC, doubling the wire to ~100 pps for the burst's
        // duration. Popping just one packet per 50 Hz tick let the
        // jitter buffer overflow on subsequent presses; the tick must
        // drain every queued DTMF packet and one audio packet per call.
        //
        // Pattern: 11 bursts (9, 1, 9, 9, 3, 3, 1, 1, 1, 1, 2) — sequence
        // numbers, timestamps, and end-bits are taken verbatim from the
        // pcap, with audio-PT-0 packets interleaved between burst body /
        // end packets at the SNs the pcap captured (so the buffer ordering
        // matches the wire exactly).
        use crate::channel::dtmf::encode_event;
        use crate::channel::rtp::RtpPacket;

        let (mut state, _peer_sock, _peer_addr) = fresh_state().await;
        let mut subs = crate::channel::actor::Subsystems::default();
        state.set_remote_addr(_peer_addr);

        // Each burst: (ts, event_code, [(dtmf_sn, end_bit), ...]). The
        // audio SNs interleaved between burst body / end packets are
        // synthesised below so the buffer ordering matches the wire.
        type Burst<'a> = (u32, u8, &'a [(u16, bool)]);
        let bursts: &[Burst<'_>] = &[
            (
                36160,
                9,
                &[
                    (226, false),
                    (228, false),
                    (230, false),
                    (232, false),
                    (234, false),
                    (236, false),
                    (238, false),
                    (239, true),
                    (241, true),
                    (242, true),
                ],
            ),
            (
                113280,
                1,
                &[
                    (718, false),
                    (720, false),
                    (722, false),
                    (724, false),
                    (726, false),
                    (728, false),
                    (730, false),
                    (731, true),
                    (733, true),
                    (734, true),
                ],
            ),
            (
                116640,
                9,
                &[
                    (749, false),
                    (751, false),
                    (753, false),
                    (755, false),
                    (757, false),
                    (759, false),
                    (761, false),
                    (762, true),
                    (764, true),
                    (765, true),
                ],
            ),
            (
                120000,
                9,
                &[
                    (780, false),
                    (782, false),
                    (784, false),
                    (786, false),
                    (788, false),
                    (790, false),
                    (792, false),
                    (793, true),
                    (795, true),
                    (796, true),
                ],
            ),
            (
                123200,
                3,
                &[
                    (810, false),
                    (812, false),
                    (814, false),
                    (816, false),
                    (818, false),
                    (820, false),
                    (822, false),
                    (823, true),
                    (825, true),
                    (826, true),
                ],
            ),
            (
                179840,
                3,
                &[
                    (1174, false),
                    (1176, false),
                    (1178, false),
                    (1180, false),
                    (1182, false),
                    (1184, false),
                    (1186, false),
                    (1187, true),
                    (1189, true),
                    (1190, true),
                ],
            ),
            (
                232640,
                1,
                &[
                    (1514, false),
                    (1516, false),
                    (1518, false),
                    (1520, false),
                    (1522, false),
                    (1524, false),
                    (1526, false),
                    (1527, true),
                    (1529, true),
                    (1530, true),
                ],
            ),
            (
                235200,
                1,
                &[
                    (1540, false),
                    (1542, false),
                    (1544, false),
                    (1546, false),
                    (1548, false),
                    (1550, false),
                    (1551, true),
                    (1553, true),
                    (1554, true),
                ],
            ),
            (
                261440,
                1,
                &[
                    (1713, false),
                    (1715, false),
                    (1717, false),
                    (1719, false),
                    (1721, false),
                    (1723, false),
                    (1725, false),
                    (1726, true),
                    (1728, true),
                    (1729, true),
                ],
            ),
            (
                324800,
                1,
                &[
                    (2119, false),
                    (2121, false),
                    (2123, false),
                    (2125, false),
                    (2127, false),
                    (2129, false),
                    (2131, false),
                    (2132, true),
                    (2134, true),
                    (2135, true),
                ],
            ),
            (
                328480,
                2,
                &[
                    (2152, false),
                    (2154, false),
                    (2156, false),
                    (2158, false),
                    (2160, false),
                    (2162, false),
                    (2164, false),
                    (2165, true),
                    (2167, true),
                    (2168, true),
                ],
            ),
        ];
        let dtmf_pt = state.rfc2833_pt;

        // Build a time-indexed packet schedule that mirrors the wire:
        // 50 pps audio everywhere, plus 50 pps interleaved DTMF inside
        // each burst. Two packets per 20 ms tick during a burst, one
        // per tick otherwise — same shape as the pcap.
        //
        // The bug only fires when the buffer has carry-over from a
        // prior burst when the next burst starts; pushing each burst
        // into an empty buffer hides it. So we push exactly the wire's
        // packets per 20 ms slot, run one tick, repeat — letting the
        // buffer state at the start of each burst be whatever the
        // previous burst left it as.
        let mut all_dtmf_sns: std::collections::HashMap<u16, (u32, u8, bool)> =
            std::collections::HashMap::new();
        let mut min_sn = u16::MAX;
        let mut max_sn: u16 = 0;
        for (ts, ev, packets) in bursts {
            for (sn, end) in *packets {
                all_dtmf_sns.insert(*sn, (*ts, *ev, *end));
                if *sn < min_sn {
                    min_sn = *sn;
                }
                if *sn > max_sn {
                    max_sn = *sn;
                }
            }
        }

        // Walk SNs from min..=max. For SNs that are DTMF, push as DTMF.
        // Otherwise push as audio. Each push advances "wire time" by
        // 10 ms — DTMF and audio in a burst are 10 ms apart, audio in
        // gaps is 20 ms apart, so:
        //   - DTMF SN: was just preceded by an audio SN at the same
        //     20 ms tick → no wire-time advance.
        //   - Audio SN with no immediately-prior DTMF SN at the same
        //     tick: 20 ms advance.
        //   - Otherwise (audio SN preceded by paired DTMF SN): 10 ms
        //     advance.
        // Every full 20 ms of wire time, run one tick.
        let mut audio_ts: u32 = 32_000;
        let mut digits: Vec<char> = Vec::new();
        let mut wire_ms: u32 = 0;
        let mut tick_target_ms: u32 = 20;
        let mut prev_was_audio = false;
        for sn in min_sn..=max_sn {
            // If this SN is the second packet of a burst pair (audio
            // immediately preceded by DTMF at sn-1), wire delta is 10 ms.
            // Otherwise it's 20 ms (steady audio cadence).
            let is_dtmf = all_dtmf_sns.contains_key(&sn);
            let prev_dtmf = sn > min_sn && all_dtmf_sns.contains_key(&(sn - 1));
            let delta_ms = if is_dtmf && prev_was_audio {
                // DTMF arrives 10 ms after the paired audio
                10
            } else if !is_dtmf && prev_dtmf {
                // Audio arrives 10 ms after the paired DTMF
                10
            } else {
                20
            };
            wire_ms += delta_ms;

            // Push the packet.
            let mut p = RtpPacket::new();
            p.init(0xC0FFEE);
            p.set_sequence_number(sn);
            if let Some((ts, ev, end)) = all_dtmf_sns.get(&sn) {
                p.set_payload_type(dtmf_pt);
                p.set_timestamp(*ts);
                p.set_payload(&encode_event(*ev, *end, 10, 160));
            } else {
                p.set_payload_type(0);
                p.set_timestamp(audio_ts);
                p.set_payload(&[0u8; 160]);
                audio_ts = audio_ts.wrapping_add(160);
            }
            state.jitter.lock().push(p);
            prev_was_audio = !is_dtmf;

            // Fire one tick for every 20 ms of wire time elapsed.
            while wire_ms >= tick_target_ms {
                run(&mut state, &mut subs).await;
                for ev in state.pending_events.drain(..) {
                    if let crate::channel::actor::Event::Telephone { digit } = ev {
                        digits.push(digit);
                    }
                }
                tick_target_ms += 20;
            }
        }
        // Drain anything still in the buffer.
        for _ in 0..40 {
            run(&mut state, &mut subs).await;
            for ev in state.pending_events.drain(..) {
                if let crate::channel::actor::Event::Telephone { digit } = ev {
                    digits.push(digit);
                }
            }
        }

        assert_eq!(
            digits,
            vec!['9', '1', '9', '9', '3', '3', '1', '1', '1', '1', '2'],
            "expected full IVR digit sequence; got {:?}",
            digits
        );
    }

    #[test]
    fn recorder_frame_direction_both_mono_sums_and_stereo_interleaves() {
        use crate::channel::recorder::RecordDirection;
        let in_s = [100i16, 200];
        let out_s = [1000i16, 2000];
        assert_eq!(
            build_recorder_frame(1, RecordDirection::Both, &in_s, &out_s, 2),
            vec![1100, 2200]
        );
        assert_eq!(
            build_recorder_frame(2, RecordDirection::Both, &in_s, &out_s, 2),
            vec![100, 1000, 200, 2000]
        );
    }

    #[test]
    fn recorder_frame_direction_in_excludes_player_audio() {
        // The languagegate regression: a mono concurrent-playrecord STT
        // capture must contain only the caller, never the TTS prompt.
        use crate::channel::recorder::RecordDirection;
        let in_s = [100i16, 200];
        let out_s = [1000i16, 2000]; // the prompt
        assert_eq!(
            build_recorder_frame(1, RecordDirection::In, &in_s, &out_s, 2),
            vec![100, 200]
        );
        // Stereo single-direction duplicates across L/R (reader convention).
        assert_eq!(
            build_recorder_frame(2, RecordDirection::In, &in_s, &out_s, 2),
            vec![100, 100, 200, 200]
        );
    }

    #[test]
    fn recorder_frame_direction_out_excludes_inbound_audio() {
        use crate::channel::recorder::RecordDirection;
        let in_s = [100i16, 200];
        let out_s = [1000i16, 2000];
        assert_eq!(
            build_recorder_frame(1, RecordDirection::Out, &in_s, &out_s, 2),
            vec![1000, 2000]
        );
        assert_eq!(
            build_recorder_frame(2, RecordDirection::Out, &in_s, &out_s, 2),
            vec![1000, 1000, 2000, 2000]
        );
    }

    #[test]
    fn recorder_frame_direction_saturates_and_zero_fills() {
        use crate::channel::recorder::RecordDirection;
        // Both: sum clamps at i16 bounds; short slices zero-fill to len.
        let in_s = [i16::MAX, 5];
        let out_s = [i16::MAX];
        assert_eq!(
            build_recorder_frame(1, RecordDirection::Both, &in_s, &out_s, 2),
            vec![i16::MAX, 5]
        );
        assert_eq!(
            build_recorder_frame(1, RecordDirection::Out, &in_s, &out_s, 2),
            vec![i16::MAX, 0]
        );
    }

    #[test]
    fn upsample_2x_doubles_and_interpolates() {
        let input = [100i16, 200, 300, 400];
        let out = upsample_2x(&input);
        assert_eq!(out.len(), 8);
        assert_eq!(out[0], 100);
        assert_eq!(out[1], 150); // (100+200)/2
        assert_eq!(out[2], 200);
        assert_eq!(out[3], 250); // (200+300)/2
                                 // last sample duplicates (no next to interpolate toward)
        assert_eq!(out[7], 400);
        assert!(upsample_2x(&[]).is_empty());
    }
}