feat: disable notifier v2 and start turn on agent interaction (#15624)

Make the inter-agent communication start a turn

As part of this, we disable the v2 notifier to prevent some odd
behaviour where the agent restart working while you're talking to it for
example

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

@@ -30,7 +30,6 @@ use crate::hook_runtime::record_additional_contexts;
 use crate::hook_runtime::record_pending_input;
 use crate::models_manager::manager::ModelsManager;
 use crate::protocol::EventMsg;
-use crate::protocol::TokenUsage;
 use crate::protocol::TurnAbortReason;
 use crate::protocol::TurnAbortedEvent;
 use crate::protocol::TurnCompleteEvent;
@@ -192,11 +191,6 @@ impl Session {
         let cancellation_token = CancellationToken::new();
         let done = Arc::new(Notify::new());
 
-        let timer = turn_context
-            .session_telemetry
-            .start_timer(TURN_E2E_DURATION_METRIC, &[])
-            .ok();
-
         let done_clone = Arc::clone(&done);
         let handle = {
             let session_ctx = Arc::new(SessionTaskContext::new(Arc::clone(self)));
@@ -236,6 +230,21 @@ impl Session {
             )
         };
 
+        let queued_response_items = self.take_queued_response_items_for_next_turn().await;
+        let mut active = self.active_turn.lock().await;
+        let mut turn = ActiveTurn::default();
+        let mut turn_state = turn.turn_state.lock().await;
+        turn_state.token_usage_at_turn_start = token_usage_at_turn_start;
+        for item in queued_response_items {
+            turn_state.push_pending_input(item);
+        }
+        drop(turn_state);
+
+        let timer = turn_context
+            .session_telemetry
+            .start_timer(TURN_E2E_DURATION_METRIC, &[])
+            .ok();
+
         let running_task = RunningTask {
             done,
             handle: Arc::new(AbortOnDropHandle::new(handle)),
@@ -245,8 +254,8 @@ impl Session {
             turn_context: Arc::clone(&turn_context),
             _timer: timer,
         };
-        self.register_new_active_task(running_task, token_usage_at_turn_start)
-            .await;
+        turn.add_task(running_task);
+        *active = Some(turn);
     }
 
     pub(crate) async fn ensure_task_for_queued_response_items(self: &Arc<Self>) {
@@ -406,23 +415,6 @@ impl Session {
         self.send_event(turn_context.as_ref(), event).await;
     }
 
-    async fn register_new_active_task(
-        &self,
-        task: RunningTask,
-        token_usage_at_turn_start: TokenUsage,
-    ) {
-        let mut active = self.active_turn.lock().await;
-        let mut turn = ActiveTurn::default();
-        let mut turn_state = turn.turn_state.lock().await;
-        turn_state.token_usage_at_turn_start = token_usage_at_turn_start;
-        for item in self.take_queued_response_items_for_next_turn().await {
-            turn_state.push_pending_input(item);
-        }
-        drop(turn_state);
-        turn.add_task(task);
-        *active = Some(turn);
-    }
-
     async fn take_active_turn(&self) -> Option<ActiveTurn> {
         let mut active = self.active_turn.lock().await;
         active.take()

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

@@ -63,14 +63,22 @@ impl SessionTask for RegularTask {
                 Some(*prewarmed_client_session)
             }
         };
-        run_turn(
-            sess,
-            ctx,
-            input,
-            prewarmed_client_session,
-            cancellation_token,
-        )
-        .instrument(run_turn_span)
-        .await
+        let mut next_input = input;
+        let mut prewarmed_client_session = prewarmed_client_session;
+        loop {
+            let last_agent_message = run_turn(
+                Arc::clone(&sess),
+                Arc::clone(&ctx),
+                next_input,
+                prewarmed_client_session.take(),
+                cancellation_token.child_token(),
+            )
+            .instrument(run_turn_span.clone())
+            .await;
+            if !sess.has_pending_input().await {
+                return last_agent_message;
+            }
+            next_input = Vec::new();
+        }
     }
 }