PicoBot/src/agent/steering.rs

//! Bounded, session-owned mailbox for same-turn steering input.
//!
//! A mailbox is intentionally separate from the session work queue.  The
//! gateway can accept a normal user message or a durable Agent steer event
//! while a turn is running and place it here;
//! [`AgentLoop`](super::AgentLoop) drains it only at safe model boundaries
//! (after a complete tool batch, or before deciding that a response is
//! final).  The state transition performed by
//! [`TurnMailbox::drain_or_close`] is atomic with respect to producers,
//! which means an input is either accepted by the active turn or rejected so
//! the caller can put it on the next-turn queue -- never both and never
//! neither.
//!
//! Durable steer events use a two-phase admission: the session first
//! reserves an agent-lane entry, persists the `leased → admitted` transition
//! (with the turn id), and only then activates the reservation into the
//! drainable queue.  Reservations and drained-but-uncommitted entries carry
//! their storage lease token so an abandoned turn can return every admitted
//! event to `pending` instead of losing it.

use crate::bus::{ChatMessage, MediaRef, MessageSource};
use std::collections::VecDeque;
use std::sync::{Arc, Mutex};

/// Default maximum number of user steering messages accepted by one active
/// turn.
pub const DEFAULT_MAX_USER_STEERING_MESSAGES: usize = 32;
/// Default aggregate UTF-8 byte budget for pending user steering messages.
pub const DEFAULT_MAX_USER_STEERING_BYTES: usize = 64 * 1024;
/// Default maximum number of Agent steer events accepted by one active turn.
pub const DEFAULT_MAX_AGENT_STEERING_MESSAGES: usize = 8;
/// Default aggregate UTF-8 byte budget for pending Agent steer events.
pub const DEFAULT_MAX_AGENT_STEERING_BYTES: usize = 32 * 1024;

/// Origin of one mailbox input.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TurnInputSource {
    User,
    AgentSignal { run_id: String, agent_id: String },
    AgentCompletion { run_id: String, agent_id: String },
    AgentGroupCompletion { group_id: String },
}

impl TurnInputSource {
    pub fn is_agent(&self) -> bool {
        !matches!(self, Self::User)
    }
}

impl From<&TurnInputSource> for WakeupSource {
    fn from(source: &TurnInputSource) -> Self {
        match source {
            TurnInputSource::User => WakeupSource::UserSteer,
            TurnInputSource::AgentSignal { run_id, agent_id } => WakeupSource::AgentSignal {
                run_id: run_id.clone(),
                agent_id: agent_id.clone(),
            },
            TurnInputSource::AgentCompletion { run_id, agent_id } => {
                WakeupSource::AgentCompletion {
                    run_id: run_id.clone(),
                    agent_id: agent_id.clone(),
                }
            }
            TurnInputSource::AgentGroupCompletion { group_id } => {
                WakeupSource::AgentGroupCompletion {
                    group_id: group_id.clone(),
                }
            }
        }
    }
}

/// What woke a root-interactive sleep.  Queue wakes carry no content: the
/// model only learns a type/count, never the payload.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum WakeupSource {
    UserSteer,
    UserQueue,
    AgentSignal { run_id: String, agent_id: String },
    AgentCompletion { run_id: String, agent_id: String },
    AgentGroupCompletion { group_id: String },
    AgentQueue,
}

/// Snapshot published to sleeping root Turns whenever a new input is
/// durably admitted anywhere on the session's receive surface.
#[derive(Debug, Clone, Default)]
pub struct TurnWakeupState {
    pub revision: u64,
    pub pending_user_steer: usize,
    pub pending_user_queue: usize,
    pub pending_agent_steer: usize,
    pub pending_agent_queue: usize,
    pub latest_source: Option<WakeupSource>,
    /// Safe, model-visible preview for steer wakes only.  Queue wakes never
    /// carry content.
    pub latest_safe_preview: Option<String>,
}

impl TurnWakeupState {
    pub fn pending_total(&self) -> usize {
        self.pending_user_steer
            .saturating_add(self.pending_user_queue)
            .saturating_add(self.pending_agent_steer)
            .saturating_add(self.pending_agent_queue)
    }
}

/// Root-Turn-side receiver used by wake-aware tools (sleep).
#[derive(Debug, Clone)]
pub struct TurnWakeupHandle {
    pub receiver: tokio::sync::watch::Receiver<TurnWakeupState>,
}

/// Session-side publisher for the active Turn.  Admission points bump the
/// revision and `send_replace` AFTER the durable fact is visible, so a
/// waking sleep can always observe the input it was told about.
#[derive(Debug, Clone)]
pub struct TurnWakeupPublisher {
    sender: tokio::sync::watch::Sender<TurnWakeupState>,
}

impl TurnWakeupPublisher {
    pub fn new() -> Self {
        let (sender, _) = tokio::sync::watch::channel(TurnWakeupState::default());
        Self { sender }
    }

    pub fn subscribe(&self) -> TurnWakeupHandle {
        TurnWakeupHandle {
            receiver: self.sender.subscribe(),
        }
    }

    pub fn publish(&self, state: TurnWakeupState) {
        let mut state = state;
        state.revision = state.revision.saturating_add(1);
        let _ = self.sender.send_replace(state);
    }
}

impl Default for TurnWakeupPublisher {
    fn default() -> Self {
        Self::new()
    }
}

/// How the input reached the mailbox.  Queue inputs belong to the next Turn;
/// only Steer entries are drained by the active Turn.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum InputDelivery {
    Queue,
    Steer,
}

/// One steering input retained by the active Turn.
#[derive(Debug, Clone)]
pub struct TurnInput {
    pub id: String,
    pub sequence: u64,
    pub source: TurnInputSource,
    pub delivery: InputDelivery,
    pub content: String,
    pub media_refs: Vec<MediaRef>,
    /// Durable inbox event id; `Some` only for Agent steer events.
    pub durable_event_id: Option<String>,
    pub received_at: i64,
    /// Channel attribution for user inputs, preserved through the turn.
    pub message_source: Option<MessageSource>,
    /// Storage lease token of a durable Agent event; the mailbox keeps it so
    /// an abandoned turn can release the event back to `pending`.
    pub(crate) lease_token: Option<String>,
}

impl TurnInput {
    pub fn user(
        id: impl Into<String>,
        content: impl Into<String>,
        media_refs: Vec<MediaRef>,
        message_source: Option<MessageSource>,
        received_at: i64,
    ) -> Self {
        Self {
            id: id.into(),
            sequence: 0,
            source: TurnInputSource::User,
            delivery: InputDelivery::Steer,
            content: content.into(),
            media_refs,
            durable_event_id: None,
            received_at,
            message_source,
            lease_token: None,
        }
    }

    /// Project this input into a provider-compatible user message.  Agent
    /// inputs stay hidden from client history (the Signal UI comes from the
    /// durable event projection) but keep their typed source for rendering
    /// and cancellation recovery.
    pub fn into_chat_message(self, turn_id: String, iteration: u32) -> ChatMessage {
        let (client_visibility, source) = match &self.source {
            TurnInputSource::User => {
                let visibility = crate::bus::ClientVisibility::Visible;
                let source = self.message_source.clone();
                (visibility, source)
            }
            TurnInputSource::AgentSignal { run_id, agent_id } => {
                let source = MessageSource {
                    kind: crate::bus::SourceKind::AgentSignal,
                    from_channel: None,
                    from_session: None,
                    from_user_id: None,
                    system_name: None,
                    task_id: self.durable_event_id.clone(),
                    from_run_id: Some(run_id.clone()),
                    from_agent_id: Some(agent_id.clone()),
                    group_id: None,
                };
                (crate::bus::ClientVisibility::Hidden, Some(source))
            }
            TurnInputSource::AgentCompletion { run_id, agent_id } => {
                let source = MessageSource {
                    kind: crate::bus::SourceKind::AgentCompletion,
                    from_channel: None,
                    from_session: None,
                    from_user_id: None,
                    system_name: None,
                    task_id: self.durable_event_id.clone(),
                    from_run_id: Some(run_id.clone()),
                    from_agent_id: Some(agent_id.clone()),
                    group_id: None,
                };
                (crate::bus::ClientVisibility::Hidden, Some(source))
            }
            TurnInputSource::AgentGroupCompletion { group_id } => {
                let source = MessageSource {
                    kind: crate::bus::SourceKind::AgentGroupCompletion,
                    from_channel: None,
                    from_session: None,
                    from_user_id: None,
                    system_name: None,
                    task_id: self.durable_event_id.clone(),
                    from_run_id: None,
                    from_agent_id: None,
                    group_id: Some(group_id.clone()),
                };
                (crate::bus::ClientVisibility::Hidden, Some(source))
            }
        };
        let mut message = ChatMessage::user(self.content);
        message.id = self.id;
        message.turn_id = Some(turn_id);
        message.iteration = Some(iteration);
        message.media_refs = self.media_refs;
        message.timestamp = self.received_at;
        message.client_visibility = client_visibility;
        message.source = source;
        message
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum MailboxPhase {
    Accepting,
    Closed,
}

#[derive(Debug)]
struct MailboxState {
    phase: MailboxPhase,
    pending: VecDeque<TurnInput>,
    reserved: VecDeque<TurnInput>,
    /// Drained at a safe boundary but not yet committed to durable history.
    /// Keeping their count/size reserved prevents concurrent producers from
    /// filling the capacity that an error retry may need to restore.
    in_flight: VecDeque<TurnInput>,
}

impl MailboxState {
    fn user_lane_used(&self) -> (usize, usize) {
        let mut count = 0usize;
        let mut bytes = 0usize;
        for input in self.pending.iter().chain(self.in_flight.iter()) {
            if !input.source.is_agent() {
                count = count.saturating_add(1);
                bytes = bytes.saturating_add(input_size_bytes(input));
            }
        }
        (count, bytes)
    }

    fn agent_lane_used(&self) -> (usize, usize) {
        let mut count = 0usize;
        let mut bytes = 0usize;
        for input in self
            .pending
            .iter()
            .chain(self.in_flight.iter())
            .chain(self.reserved.iter())
        {
            if input.source.is_agent() {
                count = count.saturating_add(1);
                bytes = bytes.saturating_add(input_size_bytes(input));
            }
        }
        (count, bytes)
    }
}

/// Error returned when the active turn cannot accept a steering input.
#[derive(Debug, Clone)]
pub enum SteeringPushError {
    /// The turn has reached a terminal boundary.  Route the input to the
    /// session's ordinary queue (or, for durable events, keep it pending).
    Closed,
    /// The mailbox is accepting input, but its bounded lane capacity is
    /// exhausted.  Route the input to the ordinary queue.
    Full,
}

/// Result of the atomic final-response boundary operation.
#[derive(Debug, Clone)]
pub enum SteeringDrain {
    /// One or more inputs were accepted and removed from the mailbox.  The
    /// mailbox remains open for a subsequent safe boundary.
    Messages(Vec<TurnInput>),
    /// No pending input existed.  The mailbox is now closed; later producers
    /// receive [`SteeringPushError::Closed`].
    Closed,
}

/// What an abandoned turn returns for requeue/release handling.
#[derive(Debug, Default)]
pub struct MailboxTake {
    /// Pending non-durable user inputs.
    pub user_inputs: Vec<TurnInput>,
    /// Durable Agent entries activated and drained or still pending:
    /// `(event_id, lease_token)` of `admitted` events.
    pub admitted_leases: Vec<(String, String)>,
    /// Reserved-but-not-activated Agent entries: `(event_id, lease_token)`
    /// of `leased` (or admitted) events.
    pub reserved_leases: Vec<(String, String)>,
}

impl MailboxTake {
    pub fn is_empty(&self) -> bool {
        self.user_inputs.is_empty()
            && self.admitted_leases.is_empty()
            && self.reserved_leases.is_empty()
    }

    pub fn len(&self) -> usize {
        self.user_inputs.len() + self.admitted_leases.len() + self.reserved_leases.len()
    }
}

/// Shared state for same-turn steering during one active AgentLoop
/// execution.
///
/// Cloning a mailbox is cheap and shares the same mutex-protected state.  In
/// practice the session stores an `Arc<TurnMailbox>` in its active-turn
/// handle and gives another clone to [`AgentTurnContext`](super::AgentTurnContext).
#[derive(Clone)]
pub struct TurnMailbox {
    state: Arc<Mutex<MailboxState>>,
    max_user_messages: usize,
    max_user_bytes: usize,
    max_agent_messages: usize,
    max_agent_bytes: usize,
}

impl std::fmt::Debug for TurnMailbox {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        formatter
            .debug_struct("TurnMailbox")
            .field("phase", &state.phase)
            .field("pending", &state.pending.len())
            .field("reserved", &state.reserved.len())
            .field("in_flight", &state.in_flight.len())
            .field("max_user_messages", &self.max_user_messages)
            .field("max_agent_messages", &self.max_agent_messages)
            .finish()
    }
}

impl TurnMailbox {
    /// Construct a mailbox using the product defaults (32 user/8 agent).
    pub fn new() -> Self {
        Self::with_limits(
            DEFAULT_MAX_USER_STEERING_MESSAGES,
            DEFAULT_MAX_USER_STEERING_BYTES,
            DEFAULT_MAX_AGENT_STEERING_MESSAGES,
            DEFAULT_MAX_AGENT_STEERING_BYTES,
        )
    }

    /// Construct a mailbox with explicit bounded capacities.  Zero limits
    /// are allowed and make every push return [`SteeringPushError::Full`].
    pub fn with_limits(
        max_user_messages: usize,
        max_user_bytes: usize,
        max_agent_messages: usize,
        max_agent_bytes: usize,
    ) -> Self {
        Self {
            state: Arc::new(Mutex::new(MailboxState {
                phase: MailboxPhase::Accepting,
                pending: VecDeque::new(),
                reserved: VecDeque::new(),
                in_flight: VecDeque::new(),
            })),
            max_user_messages,
            max_user_bytes,
            max_agent_messages,
            max_agent_bytes,
        }
    }

    /// Return an `Arc` suitable for storing in Session and AgentTurnContext.
    pub fn new_shared() -> Arc<Self> {
        Arc::new(Self::new())
    }

    /// Try to accept one real user steering input.
    ///
    /// This operation and the final close operation use the same mutex.  A
    /// producer racing with `drain_or_close` therefore receives a
    /// deterministic result and can route a rejected input to the ordinary
    /// queue.
    pub fn try_push_user(&self, input: TurnInput) -> Result<(), SteeringPushError> {
        debug_assert!(!input.source.is_agent());
        let input_bytes = input_size_bytes(&input);
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if state.phase == MailboxPhase::Closed {
            return Err(SteeringPushError::Closed);
        }
        let (count, bytes) = state.user_lane_used();
        if count >= self.max_user_messages
            || bytes.saturating_add(input_bytes) > self.max_user_bytes
        {
            return Err(SteeringPushError::Full);
        }
        state.pending.push_back(input);
        Ok(())
    }

    /// Reserve an Agent steer event for the active Turn (two-phase admission,
    /// step 2).  The entry is not drainable until
    /// [`activate_reserved`](Self::activate_reserved) succeeds for the same
    /// Turn.  `lease_token` is the storage token of the `leased` event.
    pub fn try_reserve_steer(
        &self,
        input: TurnInput,
        lease_token: String,
    ) -> Result<(), SteeringPushError> {
        debug_assert!(input.source.is_agent());
        let input_bytes = input_size_bytes(&input);
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if state.phase == MailboxPhase::Closed {
            return Err(SteeringPushError::Closed);
        }
        let (count, bytes) = state.agent_lane_used();
        if count >= self.max_agent_messages
            || bytes.saturating_add(input_bytes) > self.max_agent_bytes
        {
            return Err(SteeringPushError::Full);
        }
        let mut input = input;
        input.lease_token = Some(lease_token);
        state.reserved.push_back(input);
        Ok(())
    }

    /// Activate all reservations into the drainable queue.  Returns the
    /// number activated.  Callers invoke this only after the durable
    /// `leased → admitted` transition succeeded for the same Turn.
    pub fn activate_reserved(&self) -> usize {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let count = state.reserved.len();
        let reserved: Vec<_> = state.reserved.drain(..).collect();
        state.pending.extend(reserved);
        count
    }

    /// Remove all reservations without activating them.  Returns
    /// `(event_id, lease_token)` pairs so the caller can release the
    /// still-leased events back to `pending`.
    pub fn cancel_reserved(&self) -> Vec<(String, String)> {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        state
            .reserved
            .drain(..)
            .filter_map(|input| {
                let token = input.lease_token.clone()?;
                Some((input.durable_event_id.unwrap_or_default(), token))
            })
            .collect()
    }

    /// Drain currently pending inputs while leaving the mailbox open.
    ///
    /// This is used after a complete tool-call batch.  It intentionally does
    /// not close the mailbox: another input may steer a later iteration.
    pub fn drain(&self) -> Vec<TurnInput> {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        drain_pending_locked(&mut state)
    }

    /// Atomically drain pending inputs, or close the mailbox if it is empty.
    pub fn drain_or_close(&self) -> SteeringDrain {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if state.pending.is_empty() {
            state.phase = MailboxPhase::Closed;
            SteeringDrain::Closed
        } else {
            SteeringDrain::Messages(drain_pending_locked(&mut state))
        }
    }

    /// Close acceptance without dropping pending messages.  Session uses
    /// [`take_pending`](Self::take_pending) after AgentLoop returns to move
    /// those messages to the ordinary next-turn queue (for example when the
    /// maximum iteration budget is exhausted).
    pub fn close(&self) {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        state.phase = MailboxPhase::Closed;
    }

    /// Close acceptance and collect everything for requeue/release handling.
    /// Any in-flight batch is returned through `admitted_leases`; `/stop`
    /// uses this method to preserve its queue-clearing semantics for user
    /// input while still releasing durable events back to `pending`.
    pub fn close_and_take_pending(&self) -> MailboxTake {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        state.phase = MailboxPhase::Closed;
        take_all_locked(&mut state)
    }

    /// Take pending inputs without changing whether producers may still push.
    ///
    /// Normally used after `close()`; keeping this method explicit makes it
    /// possible for Session to transfer accepted-but-unprocessed input to its
    /// FIFO queue without opening a race with a new turn.
    pub fn take_pending(&self) -> MailboxTake {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        take_all_locked(&mut state)
    }

    /// Restore all inputs drained by AgentLoop since the last commit.  This
    /// is useful when the AgentLoop completed but Session's durable write
    /// then failed: the next retry/queue operation can replay the exact
    /// accepted inputs instead of silently losing them.
    pub fn restore_drained(&self) {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        restore_in_flight_locked(&mut state);
    }

    /// Restore inputs drained by AgentLoop when a provider/tool error makes
    /// the current invocation retry from persisted history.  The inputs are
    /// prepended in their original order and their reserved capacity is
    /// released.  `drain()`/`drain_or_close()` reserve capacity while a batch
    /// is in-flight, so this operation cannot overflow a bounded mailbox due
    /// to a racing producer.
    pub fn restore_front(&self, inputs: Vec<TurnInput>) {
        if inputs.is_empty() {
            return;
        }
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        for _ in 0..inputs.len() {
            state.in_flight.pop_back();
        }
        for input in inputs.into_iter().rev() {
            state.pending.push_front(input);
        }
    }

    /// Mark all previously drained inputs as durably committed.  Session
    /// calls this only after the complete Turn persistence transaction
    /// succeeds.  It is a no-op when no steering batch was consumed.
    pub fn commit_drained(&self) {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        state.in_flight.clear();
    }

    /// Durable event ids currently drained into this Turn but not yet
    /// committed.  Session consumes them atomically with the Turn commit.
    pub fn durable_in_flight_ids(&self) -> Vec<String> {
        let state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        state
            .in_flight
            .iter()
            .filter_map(|input| input.durable_event_id.clone())
            .collect()
    }

    pub fn is_closed(&self) -> bool {
        let state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        state.phase == MailboxPhase::Closed
    }

    pub fn len(&self) -> usize {
        let state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        state.pending.len()
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    pub fn max_user_messages(&self) -> usize {
        self.max_user_messages
    }

    pub fn max_agent_messages(&self) -> usize {
        self.max_agent_messages
    }

    /// Currently pending user steer entries (wake-state hint).
    pub fn user_pending_count(&self) -> usize {
        let state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        state
            .pending
            .iter()
            .filter(|input| !input.source.is_agent())
            .count()
    }

    /// Currently pending agent steer entries, including reservations
    /// (wake-state hint).
    pub fn agent_pending_count(&self) -> usize {
        let state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        state
            .pending
            .iter()
            .chain(state.reserved.iter())
            .filter(|input| input.source.is_agent())
            .count()
    }
}

impl Default for TurnMailbox {
    fn default() -> Self {
        Self::new()
    }
}

/// Approximate the bounded payload size without serializing the complete
/// input.  Content, media paths/types and source fields are all untrusted
/// input; counting their UTF-8 bytes gives a conservative enough guard while
/// retaining the original input losslessly.
fn input_size_bytes(input: &TurnInput) -> usize {
    let mut bytes = input.id.len()
        + input.content.len()
        + input.durable_event_id.as_deref().map_or(0, str::len)
        + input.lease_token.as_deref().map_or(0, str::len);
    for media in &input.media_refs {
        bytes = bytes.saturating_add(media.path.len() + media.media_type.len());
    }
    if let Some(source) = input.message_source.as_ref() {
        bytes = bytes
            .saturating_add(source.from_channel.as_deref().map_or(0, str::len))
            .saturating_add(source.from_user_id.as_deref().map_or(0, str::len));
    }
    bytes
}

fn drain_pending_locked(state: &mut MailboxState) -> Vec<TurnInput> {
    let inputs: Vec<_> = state.pending.drain(..).collect();
    state.in_flight.extend(inputs.iter().cloned());
    inputs
}

fn take_all_locked(state: &mut MailboxState) -> MailboxTake {
    let mut take = MailboxTake::default();
    for input in state.pending.drain(..).chain(state.in_flight.drain(..)) {
        if input.source.is_agent() {
            if let Some(token) = input.lease_token.clone()
                && let Some(event_id) = input.durable_event_id.clone()
            {
                take.admitted_leases.push((event_id, token));
            }
        } else {
            take.user_inputs.push(input);
        }
    }
    for input in state.reserved.drain(..) {
        if let Some(token) = input.lease_token.clone()
            && let Some(event_id) = input.durable_event_id.clone()
        {
            take.reserved_leases.push((event_id, token));
        }
    }
    take
}

fn restore_in_flight_locked(state: &mut MailboxState) {
    if state.in_flight.is_empty() {
        return;
    }
    let inputs: Vec<_> = state.in_flight.drain(..).collect();
    for input in inputs.into_iter().rev() {
        state.pending.push_front(input);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use std::thread;

    fn user_input(id: &str, content: &str) -> TurnInput {
        TurnInput::user(id, content, Vec::new(), None, 1_000)
    }

    fn agent_signal_input(id: &str, event_id: &str) -> TurnInput {
        TurnInput {
            id: id.to_string(),
            sequence: 0,
            source: TurnInputSource::AgentSignal {
                run_id: "run-1".to_string(),
                agent_id: "researcher".to_string(),
            },
            delivery: InputDelivery::Steer,
            content: "signal summary".to_string(),
            media_refs: Vec::new(),
            durable_event_id: Some(event_id.to_string()),
            received_at: 1_000,
            message_source: None,
            lease_token: None,
        }
    }

    #[test]
    fn accepts_fifo_inputs_and_clone_shares_state() {
        let mailbox = TurnMailbox::with_limits(2, 100, 8, 100);
        let clone = mailbox.clone();
        mailbox.try_push_user(user_input("a", "one")).unwrap();
        clone.try_push_user(user_input("b", "two")).unwrap();
        assert_eq!(mailbox.len(), 2);
        let inputs = mailbox.drain();
        assert_eq!(inputs.len(), 2);
        assert!(!mailbox.is_closed());
    }

    #[test]
    fn user_and_agent_lanes_have_independent_capacity() {
        let mailbox = TurnMailbox::with_limits(1, 100_000, 1, 100_000);
        mailbox.try_push_user(user_input("a", "one")).unwrap();
        assert!(matches!(
            mailbox.try_push_user(user_input("b", "two")),
            Err(SteeringPushError::Full)
        ));
        mailbox
            .try_reserve_steer(agent_signal_input("s1", "evt-1"), "token-1".to_string())
            .unwrap();
        assert!(matches!(
            mailbox.try_reserve_steer(agent_signal_input("s2", "evt-2"), "token-2".to_string()),
            Err(SteeringPushError::Full)
        ));
        assert_eq!(mailbox.len(), 1);
    }

    #[test]
    fn reserved_entries_are_not_drainable_until_activation() {
        let mailbox = TurnMailbox::new();
        mailbox
            .try_reserve_steer(agent_signal_input("s1", "evt-1"), "token-1".to_string())
            .unwrap();
        assert!(mailbox.drain().is_empty());
        assert_eq!(mailbox.activate_reserved(), 1);
        assert_eq!(mailbox.drain().len(), 1);
    }

    #[test]
    fn cancel_reserved_returns_leases() {
        let mailbox = TurnMailbox::new();
        mailbox
            .try_reserve_steer(agent_signal_input("s1", "evt-1"), "token-1".to_string())
            .unwrap();
        let leases = mailbox.cancel_reserved();
        assert_eq!(leases, vec![("evt-1".to_string(), "token-1".to_string())]);
        assert!(mailbox.drain().is_empty());
    }

    #[test]
    fn close_take_releases_admitted_and_reserved_durable_events() {
        let mailbox = TurnMailbox::new();
        mailbox
            .try_reserve_steer(agent_signal_input("s1", "evt-1"), "token-1".to_string())
            .unwrap();
        mailbox.activate_reserved();
        mailbox
            .try_reserve_steer(agent_signal_input("s2", "evt-2"), "token-2".to_string())
            .unwrap();
        let drained = mailbox.drain();
        assert_eq!(drained.len(), 1);
        mailbox.try_push_user(user_input("u", "hello")).unwrap();

        let take = mailbox.close_and_take_pending();
        // evt-1 was drained (admitted), evt-2 still reserved (leased), the
        // user input is returned separately.
        assert_eq!(take.user_inputs.len(), 1);
        assert_eq!(
            take.admitted_leases,
            vec![("evt-1".to_string(), "token-1".to_string())]
        );
        assert_eq!(
            take.reserved_leases,
            vec![("evt-2".to_string(), "token-2".to_string())]
        );
        assert!(mailbox.try_push_user(user_input("late", "x")).is_err());
    }

    #[test]
    fn drained_capacity_is_reserved_until_commit_or_restore() {
        let mailbox = TurnMailbox::with_limits(1, 100_000, 8, 100_000);
        mailbox.try_push_user(user_input("a", "first")).unwrap();
        let drained = mailbox.drain();
        assert_eq!(drained.len(), 1);
        assert!(matches!(
            mailbox.try_push_user(user_input("b", "second")),
            Err(SteeringPushError::Full)
        ));
        mailbox.restore_front(drained);
        let take = mailbox.take_pending();
        assert_eq!(take.user_inputs[0].content, "first");

        mailbox.try_push_user(user_input("c", "committed")).unwrap();
        let _ = mailbox.drain();
        mailbox.commit_drained();
        mailbox
            .try_push_user(user_input("d", "after commit"))
            .unwrap();

        let drained = mailbox.drain();
        assert_eq!(drained[0].content, "after commit");
        mailbox.restore_drained();
        let take = mailbox.take_pending();
        assert_eq!(take.user_inputs[0].content, "after commit");
    }

    #[test]
    fn drain_or_close_is_atomic_and_preserves_close_race_semantics() {
        let mailbox = Arc::new(TurnMailbox::new());
        let producer = mailbox.clone();
        let close_result = thread::spawn(move || producer.drain_or_close())
            .join()
            .unwrap();
        assert!(matches!(close_result, SteeringDrain::Closed));
        assert!(matches!(
            mailbox.try_push_user(user_input("late", "x")),
            Err(SteeringPushError::Closed)
        ));
    }

    #[test]
    fn turn_input_projects_to_hidden_agent_message() {
        let input = agent_signal_input("id", "evt-1");
        let message = input.into_chat_message("turn-1".to_string(), 2);
        assert_eq!(message.role, "user");
        assert_eq!(message.turn_id.as_deref(), Some("turn-1"));
        assert_eq!(message.iteration, Some(2));
        assert_eq!(
            message.client_visibility,
            crate::bus::ClientVisibility::Hidden
        );
        let source = message.source.unwrap();
        assert!(matches!(source.kind, crate::bus::SourceKind::AgentSignal));
        assert_eq!(source.from_run_id.as_deref(), Some("run-1"));
        assert_eq!(source.task_id.as_deref(), Some("evt-1"));
    }
}