Simplify queued user message admission (#38092)

## What changed

- Resolve user message admission when Core accepts the input as a new turn or a steer, without waiting for rollout persistence.
- Remove persistence- and hook-specific admission errors and task bookkeeping.
- Delete queued messages after Core accepts them for turn processing, including messages subsequently stopped by a prompt hook.

## Testing

- Update concurrent admission and queue service tests for acceptance-based admission.

GitOrigin-RevId: 7dcc311cfecd3d710733b420f2add546edb86153

codex-rs/core/src/tasks/mod.rs

@@ -24,7 +24,6 @@ use tracing::trace_span;
 use tracing::warn;
 
 use crate::codex_thread::BackgroundTerminalInfo;
-use crate::codex_thread::TryStartTurnIfIdleRejectionReason;
 use crate::config::Config;
 use crate::context::ContextualUserFragment;
 use crate::session::TurnInput;
@@ -285,14 +284,8 @@ impl Session {
     ) {
         self.abort_all_tasks(TurnAbortReason::Replaced).await;
         self.clear_connector_selection().await;
-        self.start_task(
-            turn_context,
-            input,
-            task,
-            /*input_persisted*/ None,
-            MailboxParentProvenance::Ignore,
-        )
-        .await;
+        self.start_task(turn_context, input, task, MailboxParentProvenance::Ignore)
+            .await;
     }
 
     pub(crate) async fn start_task<T: SessionTask>(
@@ -300,9 +293,6 @@ impl Session {
         turn_context: Arc<TurnContext>,
         input: Vec<TurnInput>,
         task: T,
-        input_persisted: Option<
-            tokio::sync::oneshot::Sender<Result<(), TryStartTurnIfIdleRejectionReason>>,
-        >,
         mailbox_parent_provenance: MailboxParentProvenance,
     ) {
         let task: Arc<dyn AnySessionTask> = Arc::new(task);
@@ -421,7 +411,6 @@ impl Session {
             handle: AbortOnDropHandle::new(handle),
             kind: task_kind,
             task,
-            input_persisted,
             cancellation_token,
             turn_context: Arc::clone(&turn_context),
             _agent_execution_guard: agent_execution_guard,
@@ -486,7 +475,6 @@ impl Session {
             turn_context,
             Vec::new(),
             RegularTask::new(),
-            /*input_persisted*/ None,
             MailboxParentProvenance::Attribute,
         )
         .await;
@@ -596,24 +584,12 @@ impl Session {
         let turn_state = {
             let mut active = self.active_turn.lock().await;
             active.as_mut().and_then(|active_turn| {
-                let mut task = active_turn.task.take()?;
-                let task_ended_before_persistence =
-                    if let Some(sender) = task.input_persisted.take() {
-                        let _ = sender.send(Err(
-                            TryStartTurnIfIdleRejectionReason::TaskEndedBeforePersistence,
-                        ));
-                        true
-                    } else {
-                        false
-                    };
+                let task = active_turn.task.take()?;
                 task.handle.detach();
-                Some((
-                    Arc::clone(&active_turn.turn_state),
-                    task_ended_before_persistence,
-                ))
+                Some(Arc::clone(&active_turn.turn_state))
             })
         };
-        let Some((turn_state, mut task_ended_before_persistence)) = turn_state else {
+        let Some(turn_state) = turn_state else {
             return;
         };
         let pending_input = self
@@ -635,9 +611,6 @@ impl Session {
             PersistContext::Standard,
         )
         .await;
-        task_ended_before_persistence |= self
-            .pending_user_message_admissions
-            .complete_task_end(&turn_context.sub_id);
         // Emit token usage metrics.
         {
             // TODO(jif): drop this
@@ -787,9 +760,7 @@ impl Session {
             });
         let idle_cause = if matches!(abort_reason.as_ref(), Some(TurnAbortReason::Interrupted)) {
             ThreadIdleCause::Interrupted
-        } else if task_ended_before_persistence
-            || (abort_reason.is_none() && turn_context.terminal_error.lock().await.is_some())
-        {
+        } else if abort_reason.is_none() && turn_context.terminal_error.lock().await.is_some() {
             ThreadIdleCause::Failed
         } else {
             ThreadIdleCause::Completed
@@ -877,19 +848,12 @@ impl Session {
             .await
     }
 
-    async fn handle_task_abort(self: &Arc<Self>, mut task: RunningTask, reason: TurnAbortReason) {
+    async fn handle_task_abort(self: &Arc<Self>, task: RunningTask, reason: TurnAbortReason) {
         let sub_id = task.turn_context.sub_id.clone();
         if task.cancellation_token.is_cancelled() {
             return;
         }
 
-        if let Some(sender) = task.input_persisted.take() {
-            let _ = sender.send(Err(
-                TryStartTurnIfIdleRejectionReason::TaskEndedBeforePersistence,
-            ));
-        }
-        self.pending_user_message_admissions
-            .complete_task_end(&sub_id);
         trace!(task_kind = ?task.kind, sub_id, "aborting running task");
         task.cancellation_token.cancel();
         if reason == TurnAbortReason::Interrupted