codex / codex-rs /core /src /session /input_queue.rs

Add files using upload-large-folder tool

52a9af3 verified about 2 months ago

23.2 kB

use crate::state::ActiveTurn;
use crate::state::MailboxDeliveryPhase;
use crate::state::TurnState;
use codex_diagnostics::Gauge;
use codex_diagnostics::GaugeGuard;
use codex_history::ResponseItemEnvelope;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::turn_input::TurnStartOptions;
use codex_protocol::user_input::UserInput;
use serde::Deserialize;
use serde::Serialize;
use std::collections::VecDeque;
use std::sync::Arc;
use tokio::sync::Mutex;
use tokio::sync::watch;

static PENDING_MAILBOX_MESSAGES: Gauge = Gauge::new("core.mailbox.pending");

/// Input consumed by a regular turn.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub enum TurnInput {
    UserInput {
        content: Vec<UserInput>,
        client_id: Option<String>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        acceptance_order: Option<u64>,
    },
    FunctionCallOutput(ResponseItem),
    // Preserve the existing serialized format while carrying injection API metadata
    // through the in-memory queue.
    ResponseItem(#[serde(with = "turn_input_response_item")] ResponseItemEnvelope),
    InterAgentCommunication(InterAgentCommunication),
}

mod turn_input_response_item {
    use super::ResponseItem;
    use super::ResponseItemEnvelope;
    use serde::Deserialize;
    use serde::Deserializer;
    use serde::Serialize;
    use serde::Serializer;
    use serde::ser::Error as _;

    pub(super) fn serialize<S>(
        item: &ResponseItemEnvelope,
        serializer: S,
    ) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        if item.metadata.is_some() {
            return Err(S::Error::custom(
                "annotated response items cannot cross the turn-input serialization boundary",
            ));
        }
        item.item.serialize(serializer)
    }

    pub(super) fn deserialize<'de, D>(deserializer: D) -> Result<ResponseItemEnvelope, D::Error>
    where
        D: Deserializer<'de>,
    {
        ResponseItem::deserialize(deserializer).map(ResponseItemEnvelope::new)
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum InputQueueActivity {
    Mailbox,
    Steer,
}

/// Turn-local pending input storage owned by the input queue flow.
#[derive(Default)]
pub(crate) struct TurnInputQueue {
    items: Vec<TurnInput>,
}

/// Session-scoped pending input storage and active-turn mailbox delivery coordination.
pub(crate) struct InputQueue {
    activity_tx: watch::Sender<InputQueueActivity>,
    mailbox_pending_mails: Mutex<VecDeque<PendingMailboxCommunication>>,
}

struct PendingMailboxCommunication {
    communication: InterAgentCommunication,
    start_options: TurnStartOptions,
    _diagnostics_guard: GaugeGuard,
}

impl InputQueue {
    pub(crate) fn new() -> Self {
        let (activity_tx, _) = watch::channel(InputQueueActivity::Mailbox);
        Self {
            activity_tx,
            mailbox_pending_mails: Mutex::new(VecDeque::new()),
        }
    }

    pub(crate) async fn subscribe_activity(
        &self,
        turn_state: Option<&Mutex<TurnState>>,
    ) -> (
        watch::Receiver<InputQueueActivity>,
        Option<InputQueueActivity>,
    ) {
        let activity_rx = self.activity_tx.subscribe();
        let has_pending_steer = if let Some(turn_state) = turn_state {
            turn_state.lock().await.pending_input.has_pending_input()
        } else {
            false
        };
        let pending_activity = if has_pending_steer {
            Some(InputQueueActivity::Steer)
        } else if self.has_pending_mailbox_items().await {
            Some(InputQueueActivity::Mailbox)
        } else {
            None
        };
        (activity_rx, pending_activity)
    }

    pub(crate) async fn enqueue_mailbox_communication(
        &self,
        communication: InterAgentCommunication,
        start_options: TurnStartOptions,
    ) {
        self.mailbox_pending_mails
            .lock()
            .await
            .push_back(PendingMailboxCommunication {
                communication,
                start_options,
                _diagnostics_guard: PENDING_MAILBOX_MESSAGES.track(),
            });
        self.activity_tx.send_replace(InputQueueActivity::Mailbox);
    }

    pub(crate) async fn has_pending_mailbox_items(&self) -> bool {
        !self.mailbox_pending_mails.lock().await.is_empty()
    }

    pub(crate) async fn has_trigger_turn_mailbox_items(&self) -> bool {
        self.mailbox_pending_mails
            .lock()
            .await
            .iter()
            .any(|mail| mail.communication.trigger_turn)
    }

    pub(crate) async fn drain_mailbox_input_items(&self) -> (Vec<TurnInput>, TurnStartOptions) {
        let pending_mails = self
            .mailbox_pending_mails
            .lock()
            .await
            .drain(..)
            .collect::<Vec<_>>();
        // A later follow-up supersedes the earlier choice, including an omitted choice.
        let mut start_options = pending_mails
            .iter()
            .rev()
            .find(|mail| mail.communication.trigger_turn)
            .map(|mail| mail.start_options.clone())
            .unwrap_or_default();
        start_options.parent_turn_id = pending_mails
            .iter()
            .filter(|mail| mail.communication.trigger_turn)
            .map(|mail| mail.start_options.parent_turn_id.as_deref())
            .reduce(|expected, candidate| expected.filter(|id| candidate == Some(*id)))
            .and_then(|id| id.filter(|id| !id.trim().is_empty()).map(str::to_string));
        start_options.root_turn_id = pending_mails
            .iter()
            .find(|mail| mail.communication.trigger_turn)
            .and_then(|mail| {
                mail.start_options
                    .parent_turn_id
                    .as_deref()
                    .filter(|id| !id.trim().is_empty())
                    .and(mail.start_options.root_turn_id.as_deref())
                    .filter(|id| !id.trim().is_empty())
            })
            .map(str::to_string);
        let items = pending_mails
            .into_iter()
            .map(|mail| TurnInput::InterAgentCommunication(mail.communication))
            .collect();
        (items, start_options)
    }

    pub(crate) async fn turn_state_for_sub_id(
        &self,
        active_turn: &Mutex<Option<ActiveTurn>>,
        sub_id: &str,
    ) -> Option<Arc<Mutex<TurnState>>> {
        let active = active_turn.lock().await;
        active.as_ref().and_then(|active_turn| {
            active_turn
                .task
                .as_ref()
                .is_some_and(|task| task.turn_context.sub_id == sub_id)
                .then(|| Arc::clone(&active_turn.turn_state))
        })
    }

    /// Clear any pending waiters and input buffered for the current turn.
    pub(crate) async fn clear_pending(&self, active_turn: &ActiveTurn) {
        let mut turn_state = active_turn.turn_state.lock().await;
        turn_state.clear_pending_waiters();
        turn_state.pending_input.items.clear();
    }

    pub(crate) async fn defer_mailbox_delivery_to_next_turn(
        &self,
        active_turn: &Mutex<Option<ActiveTurn>>,
        sub_id: &str,
    ) {
        let turn_state = self.turn_state_for_sub_id(active_turn, sub_id).await;
        let Some(turn_state) = turn_state else {
            return;
        };
        let mut turn_state = turn_state.lock().await;
        // Explicit same-turn work still needs a follow-up. Queue-only child mail does not: keep
        // it pending so task completion records it for the next turn without sampling again.
        if turn_state.pending_input.items.iter().any(|input| {
            !matches!(
                input,
                TurnInput::InterAgentCommunication(communication) if !communication.trigger_turn
            )
        }) {
            return;
        }
        turn_state.set_mailbox_delivery_phase(MailboxDeliveryPhase::NextTurn);
    }

    pub(crate) async fn accept_mailbox_delivery_for_current_turn(
        &self,
        active_turn: &Mutex<Option<ActiveTurn>>,
        sub_id: &str,
    ) {
        let turn_state = self.turn_state_for_sub_id(active_turn, sub_id).await;
        let Some(turn_state) = turn_state else {
            return;
        };
        self.accept_mailbox_delivery_for_turn_state(turn_state.as_ref())
            .await;
    }

    pub(super) async fn accept_mailbox_delivery_for_turn_state(
        &self,
        turn_state: &Mutex<TurnState>,
    ) {
        turn_state
            .lock()
            .await
            .accept_mailbox_delivery_for_current_turn();
    }

    pub(super) async fn extend_pending_input_and_accept_mailbox_delivery_for_turn_state(
        &self,
        turn_state: &Mutex<TurnState>,
        input: Vec<TurnInput>,
    ) {
        {
            let mut turn_state = turn_state.lock().await;
            turn_state.pending_input.items.extend(input);
            turn_state.accept_mailbox_delivery_for_current_turn();
        }
        self.activity_tx.send_replace(InputQueueActivity::Steer);
    }

    pub(crate) async fn extend_pending_input_for_turn_state(
        &self,
        turn_state: &Mutex<TurnState>,
        input: Vec<TurnInput>,
    ) {
        turn_state.lock().await.pending_input.items.extend(input);
    }

    pub(crate) async fn take_pending_input_for_turn_state(
        &self,
        turn_state: &Mutex<TurnState>,
    ) -> Vec<TurnInput> {
        turn_state.lock().await.pending_input.items.split_off(0)
    }

    #[expect(
        clippy::await_holding_invalid_type,
        reason = "active turn checks and turn state updates must remain atomic"
    )]
    pub(crate) async fn get_pending_input(
        &self,
        active_turn: &Mutex<Option<ActiveTurn>>,
    ) -> (Vec<TurnInput>, TurnStartOptions) {
        let (pending_input, accepts_mailbox_delivery) = {
            let mut active = active_turn.lock().await;
            match active.as_mut() {
                Some(active_turn) => {
                    let mut turn_state = active_turn.turn_state.lock().await;
                    let accepts_mailbox_delivery =
                        turn_state.accepts_mailbox_delivery_for_current_turn();
                    let pending_input = if accepts_mailbox_delivery {
                        turn_state.pending_input.items.split_off(0)
                    } else {
                        Vec::new()
                    };
                    (pending_input, accepts_mailbox_delivery)
                }
                None => (Vec::new(), true),
            }
        };
        if !accepts_mailbox_delivery {
            return (pending_input, TurnStartOptions::default());
        }
        let (mailbox_items, start_options) = self.drain_mailbox_input_items().await;
        if pending_input.is_empty() {
            (mailbox_items, start_options)
        } else {
            let mut pending_input = pending_input;
            pending_input.extend(mailbox_items);
            (pending_input, start_options)
        }
    }

    #[expect(
        clippy::await_holding_invalid_type,
        reason = "active turn checks and turn state reads must remain atomic"
    )]
    pub(crate) async fn has_pending_input(&self, active_turn: &Mutex<Option<ActiveTurn>>) -> bool {
        let (has_turn_pending_input, accepts_mailbox_delivery) = {
            let active = active_turn.lock().await;
            match active.as_ref() {
                Some(active_turn) => {
                    let turn_state = active_turn.turn_state.lock().await;
                    (
                        !turn_state.pending_input.items.is_empty(),
                        turn_state.accepts_mailbox_delivery_for_current_turn(),
                    )
                }
                None => (false, true),
            }
        };
        if !accepts_mailbox_delivery {
            return false;
        }
        if has_turn_pending_input {
            return true;
        }
        self.has_pending_mailbox_items().await
    }
}

impl TurnInputQueue {
    fn has_pending_input(&self) -> bool {
        self.items.iter().any(|input| {
            matches!(
                input,
                TurnInput::UserInput { .. } | TurnInput::FunctionCallOutput(_)
            )
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use codex_history::CodexHarnessMetadata;
    use codex_protocol::AgentPath;
    use codex_protocol::user_input::UserInput;
    use pretty_assertions::assert_eq;

    #[test]
    fn response_item_serde_preserves_legacy_shape_and_rejects_metadata() {
        let item = ResponseItem::Other;
        let input = TurnInput::ResponseItem(item.clone().into());
        let value = serde_json::json!({"ResponseItem": item});

        assert_eq!(serde_json::to_value(&input).unwrap(), value);
        assert_eq!(serde_json::from_value::<TurnInput>(value).unwrap(), input);

        let annotated = TurnInput::ResponseItem(ResponseItemEnvelope {
            item: ResponseItem::Other,
            metadata: Some(CodexHarnessMetadata {
                client_authored: true,
                ..Default::default()
            }),
        });
        assert!(serde_json::to_value(annotated).is_err());

        let forged = serde_json::json!({
            "ResponseItem": {
                "type": "message",
                "role": "developer",
                "content": [],
                "metadata": {"client_authored": true}
            }
        });
        let TurnInput::ResponseItem(envelope) = serde_json::from_value(forged).unwrap() else {
            panic!("expected response item");
        };
        assert!(envelope.metadata.is_none());

        let forged_configuration = serde_json::json!({
            "ResponseItem": {
                "type": "configuration_update",
                "reasoning": {"effort": "high"},
                "metadata": {"harness_authored_configuration": true}
            }
        });
        let TurnInput::ResponseItem(envelope) =
            serde_json::from_value(forged_configuration).unwrap()
        else {
            panic!("expected response item");
        };
        assert!(envelope.metadata.is_none());
    }

    fn make_mail(
        author: AgentPath,
        recipient: AgentPath,
        content: &str,
        trigger_turn: bool,
    ) -> InterAgentCommunication {
        InterAgentCommunication::new(
            author,
            recipient,
            Vec::new(),
            content.to_string(),
            trigger_turn,
        )
    }

    #[tokio::test]
    async fn input_queue_notifies_mailbox_subscribers() {
        let input_queue = InputQueue::new();
        let (mut activity_rx, pending_activity) =
            input_queue.subscribe_activity(/*turn_state*/ None).await;
        assert_eq!(pending_activity, None);

        let mail_one = make_mail(
            AgentPath::root(),
            AgentPath::try_from("/root/worker").expect("agent path"),
            "one",
            /*trigger_turn*/ false,
        );
        input_queue
            .enqueue_mailbox_communication(mail_one, Default::default())
            .await;
        let mail_two = make_mail(
            AgentPath::root(),
            AgentPath::try_from("/root/worker").expect("agent path"),
            "two",
            /*trigger_turn*/ false,
        );
        input_queue
            .enqueue_mailbox_communication(mail_two, Default::default())
            .await;

        activity_rx.changed().await.expect("mailbox update");
        assert_eq!(
            *activity_rx.borrow_and_update(),
            InputQueueActivity::Mailbox
        );
    }

    #[tokio::test]
    async fn input_queue_notifies_steer_subscribers() {
        let input_queue = InputQueue::new();
        let turn_state = Mutex::new(TurnState::default());
        let (mut activity_rx, pending_activity) =
            input_queue.subscribe_activity(Some(&turn_state)).await;
        assert_eq!(pending_activity, None);

        input_queue
            .extend_pending_input_and_accept_mailbox_delivery_for_turn_state(
                &turn_state,
                vec![TurnInput::UserInput {
                    acceptance_order: None,
                    content: vec![UserInput::Text {
                        text: "steer".to_string(),
                        text_elements: Vec::new(),
                    }],
                    client_id: None,
                }],
            )
            .await;

        activity_rx.changed().await.expect("steer update");
        assert_eq!(*activity_rx.borrow_and_update(), InputQueueActivity::Steer);
    }

    #[tokio::test]
    async fn input_queue_reports_already_pending_steer() {
        let input_queue = InputQueue::new();
        let turn_state = Mutex::new(TurnState::default());
        let passive_output = serde_json::from_value(serde_json::json!({
            "ResponseItem": {"type": "function_call_output", "name": "notify", "output": "passive"}
        }))
        .unwrap();
        input_queue
            .extend_pending_input_for_turn_state(&turn_state, vec![passive_output])
            .await;
        assert_eq!(
            input_queue.subscribe_activity(Some(&turn_state)).await.1,
            None
        );
        input_queue
            .extend_pending_input_and_accept_mailbox_delivery_for_turn_state(
                &turn_state,
                vec![TurnInput::UserInput {
                    acceptance_order: None,
                    content: vec![UserInput::Text {
                        text: "already pending".to_string(),
                        text_elements: Vec::new(),
                    }],
                    client_id: None,
                }],
            )
            .await;

        let (_activity_rx, pending_activity) =
            input_queue.subscribe_activity(Some(&turn_state)).await;

        assert_eq!(pending_activity, Some(InputQueueActivity::Steer));
    }

    #[tokio::test]
    async fn input_queue_drains_mailbox_in_delivery_order() {
        let input_queue = InputQueue::new();
        let mail_one = make_mail(
            AgentPath::root(),
            AgentPath::try_from("/root/worker").expect("agent path"),
            "one",
            /*trigger_turn*/ false,
        );
        let mail_two = make_mail(
            AgentPath::try_from("/root/worker").expect("agent path"),
            AgentPath::root(),
            "two",
            /*trigger_turn*/ true,
        );

        input_queue
            .enqueue_mailbox_communication(mail_one.clone(), Default::default())
            .await;
        input_queue
            .enqueue_mailbox_communication(mail_two.clone(), Default::default())
            .await;

        assert_eq!(
            input_queue.drain_mailbox_input_items().await.0,
            vec![
                TurnInput::InterAgentCommunication(mail_one),
                TurnInput::InterAgentCommunication(mail_two)
            ]
        );
        assert!(!input_queue.has_pending_mailbox_items().await);
    }

    #[tokio::test]
    async fn input_queue_uses_unambiguous_trigger_parent_and_first_root() {
        let (parent, peer, root, root2) = (Some("a"), Some("b"), Some("r"), Some("s"));
        for (pending_mails, expected_parent_turn_id, expected_root_turn_id) in [
            (Vec::new(), None, None),
            (vec![(false, Some("q"), root)], None, None),
            (vec![(true, Some(""), root)], None, None),
            (vec![(true, Some("   "), root)], None, None),
            (vec![(true, None, root)], None, None),
            (vec![(true, parent, None)], parent, None),
            (vec![(true, parent, Some(""))], parent, None),
            (vec![(true, parent, root), (true, peer, root)], None, root),
            (vec![(true, parent, root), (true, peer, root2)], None, root),
            (vec![(true, parent, root), (true, None, root)], None, root),
            (
                vec![(true, parent, root), (true, parent, root)],
                parent,
                root,
            ),
            (
                vec![(false, Some("q"), root2), (true, parent, root)],
                parent,
                root,
            ),
        ] {
            let input_queue = InputQueue::new();
            for (trigger_turn, parent_turn_id, root_turn_id) in pending_mails {
                input_queue
                    .enqueue_mailbox_communication(
                        make_mail(AgentPath::root(), AgentPath::root(), "task", trigger_turn),
                        TurnStartOptions {
                            parent_turn_id: parent_turn_id.map(str::to_string),
                            root_turn_id: root_turn_id.map(str::to_string),
                            ..Default::default()
                        },
                    )
                    .await;
            }
            let (_, start_options) = input_queue.drain_mailbox_input_items().await;
            assert_eq!(
                start_options.parent_turn_id.as_deref(),
                expected_parent_turn_id
            );
            assert_eq!(start_options.root_turn_id.as_deref(), expected_root_turn_id);
        }
    }

    #[tokio::test]
    async fn input_queue_uses_latest_followup_choice_and_ignores_queue_only_mail() {
        use codex_protocol::turn_input::CyberAccessProgram;

        for latest in [Some(CyberAccessProgram::Standard), None] {
            let input_queue = InputQueue::new();
            for (trigger_turn, program) in [
                (true, Some(CyberAccessProgram::DaybreakBlue)),
                (true, latest),
                (false, Some(CyberAccessProgram::DaybreakRed)),
            ] {
                input_queue
                    .enqueue_mailbox_communication(
                        make_mail(AgentPath::root(), AgentPath::root(), "task", trigger_turn),
                        TurnStartOptions {
                            cyber_access_program: program,
                            ..Default::default()
                        },
                    )
                    .await;
            }
            let (_, start_options) = input_queue.drain_mailbox_input_items().await;
            assert_eq!(start_options.cyber_access_program, latest);
        }
    }

    #[tokio::test]
    async fn input_queue_tracks_pending_trigger_turn_mail() {
        let input_queue = InputQueue::new();

        let queued_mail = make_mail(
            AgentPath::root(),
            AgentPath::try_from("/root/worker").expect("agent path"),
            "queued",
            /*trigger_turn*/ false,
        );
        input_queue
            .enqueue_mailbox_communication(queued_mail, Default::default())
            .await;
        assert!(!input_queue.has_trigger_turn_mailbox_items().await);

        let trigger_mail = make_mail(
            AgentPath::root(),
            AgentPath::try_from("/root/worker").expect("agent path"),
            "wake",
            /*trigger_turn*/ true,
        );
        input_queue
            .enqueue_mailbox_communication(trigger_mail, Default::default())
            .await;
        assert!(input_queue.has_trigger_turn_mailbox_items().await);
    }
}

	use crate::state::ActiveTurn;
	use crate::state::MailboxDeliveryPhase;
	use crate::state::TurnState;
	use codex_diagnostics::Gauge;
	use codex_diagnostics::GaugeGuard;
	use codex_history::ResponseItemEnvelope;
	use codex_protocol::models::ResponseItem;
	use codex_protocol::protocol::InterAgentCommunication;
	use codex_protocol::turn_input::TurnStartOptions;
	use codex_protocol::user_input::UserInput;
	use serde::Deserialize;
	use serde::Serialize;
	use std::collections::VecDeque;
	use std::sync::Arc;
	use tokio::sync::Mutex;
	use tokio::sync::watch;

	static PENDING_MAILBOX_MESSAGES: Gauge = Gauge::new("core.mailbox.pending");

	/// Input consumed by a regular turn.
	#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
	pub enum TurnInput {
	UserInput {
	content: Vec<UserInput>,
	client_id: Option<String>,
	#[serde(default, skip_serializing_if = "Option::is_none")]
	acceptance_order: Option<u64>,
	},
	FunctionCallOutput(ResponseItem),
	// Preserve the existing serialized format while carrying injection API metadata
	// through the in-memory queue.
	ResponseItem(#[serde(with = "turn_input_response_item")] ResponseItemEnvelope),
	InterAgentCommunication(InterAgentCommunication),
	}

	mod turn_input_response_item {
	use super::ResponseItem;
	use super::ResponseItemEnvelope;
	use serde::Deserialize;
	use serde::Deserializer;
	use serde::Serialize;
	use serde::Serializer;
	use serde::ser::Error as _;

	pub(super) fn serialize<S>(
	item: &ResponseItemEnvelope,
	serializer: S,
	) -> Result<S::Ok, S::Error>
	where
	S: Serializer,
	{
	if item.metadata.is_some() {
	return Err(S::Error::custom(
	"annotated response items cannot cross the turn-input serialization boundary",
	));
	}
	item.item.serialize(serializer)
	}

	pub(super) fn deserialize<'de, D>(deserializer: D) -> Result<ResponseItemEnvelope, D::Error>
	where
	D: Deserializer<'de>,
	{
	ResponseItem::deserialize(deserializer).map(ResponseItemEnvelope::new)
	}
	}

	#[derive(Clone, Copy, Debug, Eq, PartialEq)]
	pub(crate) enum InputQueueActivity {
	Mailbox,
	Steer,
	}

	/// Turn-local pending input storage owned by the input queue flow.
	#[derive(Default)]
	pub(crate) struct TurnInputQueue {
	items: Vec<TurnInput>,
	}

	/// Session-scoped pending input storage and active-turn mailbox delivery coordination.
	pub(crate) struct InputQueue {
	activity_tx: watch::Sender<InputQueueActivity>,
	mailbox_pending_mails: Mutex<VecDeque<PendingMailboxCommunication>>,
	}

	struct PendingMailboxCommunication {
	communication: InterAgentCommunication,
	start_options: TurnStartOptions,
	_diagnostics_guard: GaugeGuard,
	}

	impl InputQueue {
	pub(crate) fn new() -> Self {
	let (activity_tx, _) = watch::channel(InputQueueActivity::Mailbox);
	Self {
	activity_tx,
	mailbox_pending_mails: Mutex::new(VecDeque::new()),
	}
	}

	pub(crate) async fn subscribe_activity(
	&self,
	turn_state: Option<&Mutex<TurnState>>,
	) -> (
	watch::Receiver<InputQueueActivity>,
	Option<InputQueueActivity>,
	) {
	let activity_rx = self.activity_tx.subscribe();
	let has_pending_steer = if let Some(turn_state) = turn_state {
	turn_state.lock().await.pending_input.has_pending_input()
	} else {
	false
	};
	let pending_activity = if has_pending_steer {
	Some(InputQueueActivity::Steer)
	} else if self.has_pending_mailbox_items().await {
	Some(InputQueueActivity::Mailbox)
	} else {
	None
	};
	(activity_rx, pending_activity)
	}

	pub(crate) async fn enqueue_mailbox_communication(
	&self,
	communication: InterAgentCommunication,
	start_options: TurnStartOptions,
	) {
	self.mailbox_pending_mails
	.lock()
	.await
	.push_back(PendingMailboxCommunication {
	communication,
	start_options,
	_diagnostics_guard: PENDING_MAILBOX_MESSAGES.track(),
	});
	self.activity_tx.send_replace(InputQueueActivity::Mailbox);
	}

	pub(crate) async fn has_pending_mailbox_items(&self) -> bool {
	!self.mailbox_pending_mails.lock().await.is_empty()
	}

	pub(crate) async fn has_trigger_turn_mailbox_items(&self) -> bool {
	self.mailbox_pending_mails
	.lock()
	.await
	.iter()
	.any(\|mail\| mail.communication.trigger_turn)
	}

	pub(crate) async fn drain_mailbox_input_items(&self) -> (Vec<TurnInput>, TurnStartOptions) {
	let pending_mails = self
	.mailbox_pending_mails
	.lock()
	.await
	.drain(..)
	.collect::<Vec<_>>();
	// A later follow-up supersedes the earlier choice, including an omitted choice.
	let mut start_options = pending_mails
	.iter()
	.rev()
	.find(\|mail\| mail.communication.trigger_turn)
	.map(\|mail\| mail.start_options.clone())
	.unwrap_or_default();
	start_options.parent_turn_id = pending_mails
	.iter()
	.filter(\|mail\| mail.communication.trigger_turn)
	.map(\|mail\| mail.start_options.parent_turn_id.as_deref())
	.reduce(\|expected, candidate\| expected.filter(\|id\| candidate == Some(*id)))
	.and_then(\|id\| id.filter(\|id\| !id.trim().is_empty()).map(str::to_string));
	start_options.root_turn_id = pending_mails
	.iter()
	.find(\|mail\| mail.communication.trigger_turn)
	.and_then(\|mail\| {
	mail.start_options
	.parent_turn_id
	.as_deref()
	.filter(\|id\| !id.trim().is_empty())
	.and(mail.start_options.root_turn_id.as_deref())
	.filter(\|id\| !id.trim().is_empty())
	})
	.map(str::to_string);
	let items = pending_mails
	.into_iter()
	.map(\|mail\| TurnInput::InterAgentCommunication(mail.communication))
	.collect();
	(items, start_options)
	}

	pub(crate) async fn turn_state_for_sub_id(
	&self,
	active_turn: &Mutex<Option<ActiveTurn>>,
	sub_id: &str,
	) -> Option<Arc<Mutex<TurnState>>> {
	let active = active_turn.lock().await;
	active.as_ref().and_then(\|active_turn\| {
	active_turn
	.task
	.as_ref()
	.is_some_and(\|task\| task.turn_context.sub_id == sub_id)
	.then(\|\| Arc::clone(&active_turn.turn_state))
	})
	}

	/// Clear any pending waiters and input buffered for the current turn.
	pub(crate) async fn clear_pending(&self, active_turn: &ActiveTurn) {
	let mut turn_state = active_turn.turn_state.lock().await;
	turn_state.clear_pending_waiters();
	turn_state.pending_input.items.clear();
	}

	pub(crate) async fn defer_mailbox_delivery_to_next_turn(
	&self,
	active_turn: &Mutex<Option<ActiveTurn>>,
	sub_id: &str,
	) {
	let turn_state = self.turn_state_for_sub_id(active_turn, sub_id).await;
	let Some(turn_state) = turn_state else {
	return;
	};
	let mut turn_state = turn_state.lock().await;
	// Explicit same-turn work still needs a follow-up. Queue-only child mail does not: keep
	// it pending so task completion records it for the next turn without sampling again.
	if turn_state.pending_input.items.iter().any(\|input\| {
	!matches!(
	input,
	TurnInput::InterAgentCommunication(communication) if !communication.trigger_turn
	)
	}) {
	return;
	}
	turn_state.set_mailbox_delivery_phase(MailboxDeliveryPhase::NextTurn);
	}

	pub(crate) async fn accept_mailbox_delivery_for_current_turn(
	&self,
	active_turn: &Mutex<Option<ActiveTurn>>,
	sub_id: &str,
	) {
	let turn_state = self.turn_state_for_sub_id(active_turn, sub_id).await;
	let Some(turn_state) = turn_state else {
	return;
	};
	self.accept_mailbox_delivery_for_turn_state(turn_state.as_ref())
	.await;
	}

	pub(super) async fn accept_mailbox_delivery_for_turn_state(
	&self,
	turn_state: &Mutex<TurnState>,
	) {
	turn_state
	.lock()
	.await
	.accept_mailbox_delivery_for_current_turn();
	}

	pub(super) async fn extend_pending_input_and_accept_mailbox_delivery_for_turn_state(
	&self,
	turn_state: &Mutex<TurnState>,
	input: Vec<TurnInput>,
	) {
	{
	let mut turn_state = turn_state.lock().await;
	turn_state.pending_input.items.extend(input);
	turn_state.accept_mailbox_delivery_for_current_turn();
	}
	self.activity_tx.send_replace(InputQueueActivity::Steer);
	}

	pub(crate) async fn extend_pending_input_for_turn_state(
	&self,
	turn_state: &Mutex<TurnState>,
	input: Vec<TurnInput>,
	) {
	turn_state.lock().await.pending_input.items.extend(input);
	}

	pub(crate) async fn take_pending_input_for_turn_state(
	&self,
	turn_state: &Mutex<TurnState>,
	) -> Vec<TurnInput> {
	turn_state.lock().await.pending_input.items.split_off(0)
	}

	#[expect(
	clippy::await_holding_invalid_type,
	reason = "active turn checks and turn state updates must remain atomic"
	)]
	pub(crate) async fn get_pending_input(
	&self,
	active_turn: &Mutex<Option<ActiveTurn>>,
	) -> (Vec<TurnInput>, TurnStartOptions) {
	let (pending_input, accepts_mailbox_delivery) = {
	let mut active = active_turn.lock().await;
	match active.as_mut() {
	Some(active_turn) => {
	let mut turn_state = active_turn.turn_state.lock().await;
	let accepts_mailbox_delivery =
	turn_state.accepts_mailbox_delivery_for_current_turn();
	let pending_input = if accepts_mailbox_delivery {
	turn_state.pending_input.items.split_off(0)
	} else {
	Vec::new()
	};
	(pending_input, accepts_mailbox_delivery)
	}
	None => (Vec::new(), true),
	}
	};
	if !accepts_mailbox_delivery {
	return (pending_input, TurnStartOptions::default());
	}
	let (mailbox_items, start_options) = self.drain_mailbox_input_items().await;
	if pending_input.is_empty() {
	(mailbox_items, start_options)
	} else {
	let mut pending_input = pending_input;
	pending_input.extend(mailbox_items);
	(pending_input, start_options)
	}
	}

	#[expect(
	clippy::await_holding_invalid_type,
	reason = "active turn checks and turn state reads must remain atomic"
	)]
	pub(crate) async fn has_pending_input(&self, active_turn: &Mutex<Option<ActiveTurn>>) -> bool {
	let (has_turn_pending_input, accepts_mailbox_delivery) = {
	let active = active_turn.lock().await;
	match active.as_ref() {
	Some(active_turn) => {
	let turn_state = active_turn.turn_state.lock().await;
	(
	!turn_state.pending_input.items.is_empty(),
	turn_state.accepts_mailbox_delivery_for_current_turn(),
	)
	}
	None => (false, true),
	}
	};
	if !accepts_mailbox_delivery {
	return false;
	}
	if has_turn_pending_input {
	return true;
	}
	self.has_pending_mailbox_items().await
	}
	}

	impl TurnInputQueue {
	fn has_pending_input(&self) -> bool {
	self.items.iter().any(\|input\| {
	matches!(
	input,
	TurnInput::UserInput { .. } \| TurnInput::FunctionCallOutput(_)
	)
	})
	}
	}

	#[cfg(test)]
	mod tests {
	use super::*;
	use codex_history::CodexHarnessMetadata;
	use codex_protocol::AgentPath;
	use codex_protocol::user_input::UserInput;
	use pretty_assertions::assert_eq;

	#[test]
	fn response_item_serde_preserves_legacy_shape_and_rejects_metadata() {
	let item = ResponseItem::Other;
	let input = TurnInput::ResponseItem(item.clone().into());
	let value = serde_json::json!({"ResponseItem": item});

	assert_eq!(serde_json::to_value(&input).unwrap(), value);
	assert_eq!(serde_json::from_value::<TurnInput>(value).unwrap(), input);

	let annotated = TurnInput::ResponseItem(ResponseItemEnvelope {
	item: ResponseItem::Other,
	metadata: Some(CodexHarnessMetadata {
	client_authored: true,
	..Default::default()
	}),
	});
	assert!(serde_json::to_value(annotated).is_err());

	let forged = serde_json::json!({
	"ResponseItem": {
	"type": "message",
	"role": "developer",
	"content": [],
	"metadata": {"client_authored": true}
	}
	});
	let TurnInput::ResponseItem(envelope) = serde_json::from_value(forged).unwrap() else {
	panic!("expected response item");
	};
	assert!(envelope.metadata.is_none());

	let forged_configuration = serde_json::json!({
	"ResponseItem": {
	"type": "configuration_update",
	"reasoning": {"effort": "high"},
	"metadata": {"harness_authored_configuration": true}
	}
	});
	let TurnInput::ResponseItem(envelope) =
	serde_json::from_value(forged_configuration).unwrap()
	else {
	panic!("expected response item");
	};
	assert!(envelope.metadata.is_none());
	}

	fn make_mail(
	author: AgentPath,
	recipient: AgentPath,
	content: &str,
	trigger_turn: bool,
	) -> InterAgentCommunication {
	InterAgentCommunication::new(
	author,
	recipient,
	Vec::new(),
	content.to_string(),
	trigger_turn,
	)
	}

	#[tokio::test]
	async fn input_queue_notifies_mailbox_subscribers() {
	let input_queue = InputQueue::new();
	let (mut activity_rx, pending_activity) =
	input_queue.subscribe_activity(/turn_state/ None).await;
	assert_eq!(pending_activity, None);

	let mail_one = make_mail(
	AgentPath::root(),
	AgentPath::try_from("/root/worker").expect("agent path"),
	"one",
	/trigger_turn/ false,
	);
	input_queue
	.enqueue_mailbox_communication(mail_one, Default::default())
	.await;
	let mail_two = make_mail(
	AgentPath::root(),
	AgentPath::try_from("/root/worker").expect("agent path"),
	"two",
	/trigger_turn/ false,
	);
	input_queue
	.enqueue_mailbox_communication(mail_two, Default::default())
	.await;

	activity_rx.changed().await.expect("mailbox update");
	assert_eq!(
	*activity_rx.borrow_and_update(),
	InputQueueActivity::Mailbox
	);
	}

	#[tokio::test]
	async fn input_queue_notifies_steer_subscribers() {
	let input_queue = InputQueue::new();
	let turn_state = Mutex::new(TurnState::default());
	let (mut activity_rx, pending_activity) =
	input_queue.subscribe_activity(Some(&turn_state)).await;
	assert_eq!(pending_activity, None);

	input_queue
	.extend_pending_input_and_accept_mailbox_delivery_for_turn_state(
	&turn_state,
	vec![TurnInput::UserInput {
	acceptance_order: None,
	content: vec![UserInput::Text {
	text: "steer".to_string(),
	text_elements: Vec::new(),
	}],
	client_id: None,
	}],
	)
	.await;

	activity_rx.changed().await.expect("steer update");
	assert_eq!(*activity_rx.borrow_and_update(), InputQueueActivity::Steer);
	}

	#[tokio::test]
	async fn input_queue_reports_already_pending_steer() {
	let input_queue = InputQueue::new();
	let turn_state = Mutex::new(TurnState::default());
	let passive_output = serde_json::from_value(serde_json::json!({
	"ResponseItem": {"type": "function_call_output", "name": "notify", "output": "passive"}
	}))
	.unwrap();
	input_queue
	.extend_pending_input_for_turn_state(&turn_state, vec![passive_output])
	.await;
	assert_eq!(
	input_queue.subscribe_activity(Some(&turn_state)).await.1,
	None
	);
	input_queue
	.extend_pending_input_and_accept_mailbox_delivery_for_turn_state(
	&turn_state,
	vec![TurnInput::UserInput {
	acceptance_order: None,
	content: vec![UserInput::Text {
	text: "already pending".to_string(),
	text_elements: Vec::new(),
	}],
	client_id: None,
	}],
	)
	.await;

	let (_activity_rx, pending_activity) =
	input_queue.subscribe_activity(Some(&turn_state)).await;

	assert_eq!(pending_activity, Some(InputQueueActivity::Steer));
	}

	#[tokio::test]
	async fn input_queue_drains_mailbox_in_delivery_order() {
	let input_queue = InputQueue::new();
	let mail_one = make_mail(
	AgentPath::root(),
	AgentPath::try_from("/root/worker").expect("agent path"),
	"one",
	/trigger_turn/ false,
	);
	let mail_two = make_mail(
	AgentPath::try_from("/root/worker").expect("agent path"),
	AgentPath::root(),
	"two",
	/trigger_turn/ true,
	);

	input_queue
	.enqueue_mailbox_communication(mail_one.clone(), Default::default())
	.await;
	input_queue
	.enqueue_mailbox_communication(mail_two.clone(), Default::default())
	.await;

	assert_eq!(
	input_queue.drain_mailbox_input_items().await.0,
	vec![
	TurnInput::InterAgentCommunication(mail_one),
	TurnInput::InterAgentCommunication(mail_two)
	]
	);
	assert!(!input_queue.has_pending_mailbox_items().await);
	}

	#[tokio::test]
	async fn input_queue_uses_unambiguous_trigger_parent_and_first_root() {
	let (parent, peer, root, root2) = (Some("a"), Some("b"), Some("r"), Some("s"));
	for (pending_mails, expected_parent_turn_id, expected_root_turn_id) in [
	(Vec::new(), None, None),
	(vec![(false, Some("q"), root)], None, None),
	(vec![(true, Some(""), root)], None, None),
	(vec![(true, Some(" "), root)], None, None),
	(vec![(true, None, root)], None, None),
	(vec![(true, parent, None)], parent, None),
	(vec![(true, parent, Some(""))], parent, None),
	(vec![(true, parent, root), (true, peer, root)], None, root),
	(vec![(true, parent, root), (true, peer, root2)], None, root),
	(vec![(true, parent, root), (true, None, root)], None, root),
	(
	vec![(true, parent, root), (true, parent, root)],
	parent,
	root,
	),
	(
	vec![(false, Some("q"), root2), (true, parent, root)],
	parent,
	root,
	),
	] {
	let input_queue = InputQueue::new();
	for (trigger_turn, parent_turn_id, root_turn_id) in pending_mails {
	input_queue
	.enqueue_mailbox_communication(
	make_mail(AgentPath::root(), AgentPath::root(), "task", trigger_turn),
	TurnStartOptions {
	parent_turn_id: parent_turn_id.map(str::to_string),
	root_turn_id: root_turn_id.map(str::to_string),
	..Default::default()
	},
	)
	.await;
	}
	let (_, start_options) = input_queue.drain_mailbox_input_items().await;
	assert_eq!(
	start_options.parent_turn_id.as_deref(),
	expected_parent_turn_id
	);
	assert_eq!(start_options.root_turn_id.as_deref(), expected_root_turn_id);
	}
	}

	#[tokio::test]
	async fn input_queue_uses_latest_followup_choice_and_ignores_queue_only_mail() {
	use codex_protocol::turn_input::CyberAccessProgram;

	for latest in [Some(CyberAccessProgram::Standard), None] {
	let input_queue = InputQueue::new();
	for (trigger_turn, program) in [
	(true, Some(CyberAccessProgram::DaybreakBlue)),
	(true, latest),
	(false, Some(CyberAccessProgram::DaybreakRed)),
	] {
	input_queue
	.enqueue_mailbox_communication(
	make_mail(AgentPath::root(), AgentPath::root(), "task", trigger_turn),
	TurnStartOptions {
	cyber_access_program: program,
	..Default::default()
	},
	)
	.await;
	}
	let (_, start_options) = input_queue.drain_mailbox_input_items().await;
	assert_eq!(start_options.cyber_access_program, latest);
	}
	}

	#[tokio::test]
	async fn input_queue_tracks_pending_trigger_turn_mail() {
	let input_queue = InputQueue::new();

	let queued_mail = make_mail(
	AgentPath::root(),
	AgentPath::try_from("/root/worker").expect("agent path"),
	"queued",
	/trigger_turn/ false,
	);
	input_queue
	.enqueue_mailbox_communication(queued_mail, Default::default())
	.await;
	assert!(!input_queue.has_trigger_turn_mailbox_items().await);

	let trigger_mail = make_mail(
	AgentPath::root(),
	AgentPath::try_from("/root/worker").expect("agent path"),
	"wake",
	/trigger_turn/ true,
	);
	input_queue
	.enqueue_mailbox_communication(trigger_mail, Default::default())
	.await;
	assert!(input_queue.has_trigger_turn_mailbox_items().await);
	}
	}