Gate new turn submissions on host shutdown admission (#43943)

## Why

Hosts need to stop new turn-input work during shutdown without consuming pending input or preventing already-running delegated work from finishing.

## What changed

- Add an optional `TurnStartAdmission` extension gate, checked before reserving or starting a new turn. Hosts without a gate retain existing behavior.
- Return `NotSubmittedReason::ServerDraining` for refused starts and surface an app-server error instructing clients to reconnect and retry.
- Keep steering, parent-delegated subagent input, and memory-only mailbox wakeups available during drain, while gating automatic starts.
- Close realtime conversations with an ordered handoff, error, and close event sequence when a handoff is refused during drain.

## Testing

Add regression coverage for rejected input staying out of subsequent requests, persisted queue items remaining available for later starts, delegated agent and review work completing during drain, mailbox wakeups, and realtime handoff error ordering.

GitOrigin-RevId: 03dbdcd71eab200e597c0649e6eb39bd92dbc82f

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

@@ -7,6 +7,9 @@
 //!
 //! Persistent thread settings apply on Started and Steered. Turn start
 //! options only apply on Started.
+//! Host shutdown admission is checked before reserving or starting a new turn.
+//! Parent-delegated subagent input bypasses drain; automatic starts remain gated.
+//! Realtime drain refusals are returned to the fanout for ordered session teardown.
 
 use super::TurnInput;
 use super::session::Session;
@@ -28,6 +31,8 @@ use codex_protocol::protocol::ErrorEvent;
 use codex_protocol::protocol::Event;
 use codex_protocol::protocol::EventMsg;
 use codex_protocol::protocol::NonSteerableTurnKind;
+use codex_protocol::protocol::SessionSource;
+use codex_protocol::protocol::SubAgentSource;
 use codex_protocol::protocol::ThreadSettingsOverrides;
 use codex_protocol::turn_input::NotSubmittedReason;
 use codex_protocol::turn_input::TurnInput as SubmittedTurnInput;
@@ -276,6 +281,28 @@ async fn start_or_steer(
             Ok(TurnInputSubmission::Steered { turn_id })
         }
         Err(NotSubmittedReason::NoActiveTurn) => {
+            // MAv1 sends explicit input to spawned agents as part of an existing
+            // parent's work. Client RPCs are gated separately by the host.
+            let is_delegated_input = settings.start_options.parent_turn_id.is_some()
+                && matches!(
+                    session
+                        .state
+                        .lock()
+                        .await
+                        .session_configuration
+                        .session_source,
+                    SessionSource::SubAgent(SubAgentSource::ThreadSpawn { .. })
+                );
+            let _admission = if is_delegated_input {
+                None
+            } else {
+                let Some(admission) = session.services.extensions.admit_turn_start() else {
+                    return Ok(TurnInputSubmission::NotSubmitted {
+                        reason: NotSubmittedReason::ServerDraining,
+                    });
+                };
+                Some(admission)
+            };
             let Some(turn_context) = settings
                 .apply_started(session, submission_id.clone(), TurnStartKind::User)
                 .await?
@@ -337,6 +364,27 @@ async fn start_if_idle(
         });
     }
 
+    let _admission = session.services.extensions.admit_turn_start();
+    // A one-shot review delegate completes its already-running parent's work.
+    // Its explicit input carries parent lineage; automatic starts do not qualify.
+    if _admission.is_none()
+        && !(kind == TurnStartKind::User
+            && start.parent_turn_id.is_some()
+            && matches!(
+                session
+                    .state
+                    .lock()
+                    .await
+                    .session_configuration
+                    .session_source,
+                SessionSource::SubAgent(SubAgentSource::Review)
+            ))
+    {
+        return Ok(TurnInputSubmission::NotSubmitted {
+            reason: NotSubmittedReason::ServerDraining,
+        });
+    }
+
     let turn_state = {
         let mut active_turn = session.active_turn.lock().await;
         if active_turn.is_some() {
@@ -460,7 +508,10 @@ async fn steer(
 }
 
 impl Session {
-    pub(crate) async fn route_realtime_text_input(self: &Arc<Self>, text: String) {
+    pub(crate) async fn route_realtime_text_input(
+        self: &Arc<Self>,
+        text: String,
+    ) -> Result<(), &'static str> {
         let submission_id = Uuid::now_v7().to_string();
         let submission = handle(
             self,
@@ -478,6 +529,11 @@ impl Session {
         .await;
         match submission {
             Ok(TurnInputSubmission::Started { .. } | TurnInputSubmission::Steered { .. }) => {}
+            Ok(TurnInputSubmission::NotSubmitted {
+                reason: NotSubmittedReason::ServerDraining,
+            }) => {
+                return Err("Server is draining; retry the turn after reconnecting");
+            }
             Ok(TurnInputSubmission::NotSubmitted { reason }) => {
                 self.send_event_raw(Event {
                     id: submission_id,
@@ -497,6 +553,7 @@ impl Session {
                 .await;
             }
         }
+        Ok(())
     }
 
     async fn clear_reserved_idle_turn(&self, turn_state: &Arc<tokio::sync::Mutex<TurnState>>) {