Support compact SessionStart hooks (#21272)

# Why

Compaction replaces the live conversation history, so hooks that use
`SessionStart` to re-inject durable model context need a way to run
again after that rewrite.

Related - #19905 adds dedicated compact lifecycle hooks

# What

- add `compact` as a supported `SessionStart` source and matcher value
- change pending `SessionStart` state from a single slot to a small FIFO
queue so `resume` / `startup` / `clear` can be preserved alongside a
later `compact`
- drain all queued `SessionStart` sources before the next model request,
preserving their original order

# Testing

The new integration coverage verifies both the basic `compact` matcher
path and the stacked `resume` -> `compact` case where both hooks
contribute `additionalContext` to the next model turn.

codex-rs/core/src/hook_runtime.rs

@@ -99,54 +99,57 @@ pub(crate) async fn run_pending_session_start_hooks(
     sess: &Arc<Session>,
     turn_context: &Arc<TurnContext>,
 ) -> bool {
-    let Some(session_start_source) = sess.take_pending_session_start_source().await else {
-        return false;
-    };
-
-    // Pending session-start hooks are reused to dispatch thread-spawn subagent
-    // starts. Other subagent sessions are internal/system work and do not run
-    // start hooks.
-    let target = match &turn_context.session_source {
-        SessionSource::SubAgent(SubAgentSource::ThreadSpawn { agent_role, .. })
-            if matches!(
-                session_start_source,
-                codex_hooks::SessionStartSource::Startup
-            ) =>
-        {
-            let agent_type = agent_role
-                .clone()
-                .unwrap_or_else(|| crate::agent::role::DEFAULT_ROLE_NAME.to_string());
-            StartHookTarget::SubagentStart {
-                turn_id: turn_context.sub_id.clone(),
-                agent_id: sess.thread_id().to_string(),
-                agent_type,
+    while let Some(session_start_source) = sess.take_pending_session_start_source().await {
+        // Pending session-start hooks are reused to dispatch thread-spawn subagent
+        // starts. Other subagent sessions are internal/system work and do not run
+        // start hooks.
+        let target = match &turn_context.session_source {
+            SessionSource::SubAgent(SubAgentSource::ThreadSpawn { agent_role, .. })
+                if matches!(
+                    session_start_source,
+                    codex_hooks::SessionStartSource::Startup
+                ) =>
+            {
+                let agent_type = agent_role
+                    .clone()
+                    .unwrap_or_else(|| crate::agent::role::DEFAULT_ROLE_NAME.to_string());
+                StartHookTarget::SubagentStart {
+                    turn_id: turn_context.sub_id.clone(),
+                    agent_id: sess.thread_id().to_string(),
+                    agent_type,
+                }
             }
+            SessionSource::SubAgent(_) => return false,
+            _ => StartHookTarget::SessionStart {
+                source: session_start_source,
+            },
+        };
+        let request = codex_hooks::SessionStartRequest {
+            session_id: sess.session_id().into(),
+            #[allow(deprecated)]
+            cwd: turn_context.cwd.clone(),
+            transcript_path: sess.hook_transcript_path().await,
+            model: turn_context.model_info.slug.clone(),
+            permission_mode: hook_permission_mode(turn_context),
+            target,
+        };
+        let hooks = sess.hooks();
+        let preview_runs = hooks.preview_session_start(&request);
+        if run_context_injecting_hook(
+            sess,
+            turn_context,
+            preview_runs,
+            hooks.run_session_start(request, Some(turn_context.sub_id.clone())),
+        )
+        .await
+        .record_additional_contexts(sess, turn_context)
+        .await
+        {
+            return true;
         }
-        SessionSource::SubAgent(_) => return false,
-        _ => StartHookTarget::SessionStart {
-            source: session_start_source,
-        },
-    };
-    let request = codex_hooks::SessionStartRequest {
-        session_id: sess.session_id().into(),
-        #[allow(deprecated)]
-        cwd: turn_context.cwd.clone(),
-        transcript_path: sess.hook_transcript_path().await,
-        model: turn_context.model_info.slug.clone(),
-        permission_mode: hook_permission_mode(turn_context),
-        target,
-    };
-    let hooks = sess.hooks();
-    let preview_runs = hooks.preview_session_start(&request);
-    run_context_injecting_hook(
-        sess,
-        turn_context,
-        preview_runs,
-        hooks.run_session_start(request, Some(turn_context.sub_id.clone())),
-    )
-    .await
-    .record_additional_contexts(sess, turn_context)
-    .await
+    }
+
+    false
 }
 
 /// Runs matching `PreToolUse` hooks before a tool executes.

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

@@ -2609,6 +2609,10 @@ impl Session {
             self.persist_rollout_items(&[RolloutItem::TurnContext(turn_context_item)])
                 .await;
         }
+        {
+            let mut state = self.state.lock().await;
+            state.queue_pending_session_start_source(codex_hooks::SessionStartSource::Compact);
+        }
         self.services.model_client.advance_window_generation();
     }
 

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

@@ -1184,7 +1184,7 @@ impl Session {
             Box::pin(sess.record_initial_history(initial_history)).await;
             {
                 let mut state = sess.state.lock().await;
-                state.set_pending_session_start_source(Some(session_start_source));
+                state.queue_pending_session_start_source(session_start_source);
             }
 
             Ok(sess)

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

@@ -4,6 +4,7 @@ use codex_protocol::models::AdditionalPermissionProfile;
 use codex_protocol::models::ResponseItem;
 use codex_sandboxing::policy_transforms::merge_permission_profiles;
 use std::collections::HashSet;
+use std::collections::VecDeque;
 
 use super::auto_compact_window::AutoCompactWindow;
 use super::auto_compact_window::AutoCompactWindowSnapshot;
@@ -33,7 +34,7 @@ pub(crate) struct SessionState {
     /// Startup prewarmed session prepared during session initialization.
     pub(crate) startup_prewarm: Option<SessionStartupPrewarmHandle>,
     pub(crate) active_connector_selection: HashSet<String>,
-    pub(crate) pending_session_start_source: Option<codex_hooks::SessionStartSource>,
+    pub(crate) pending_session_start_sources: VecDeque<codex_hooks::SessionStartSource>,
     granted_permissions: Option<AdditionalPermissionProfile>,
     next_turn_is_first: bool,
 }
@@ -52,7 +53,7 @@ impl SessionState {
             auto_compact_window: AutoCompactWindow::new(),
             startup_prewarm: None,
             active_connector_selection: HashSet::new(),
-            pending_session_start_source: None,
+            pending_session_start_sources: VecDeque::new(),
             granted_permissions: None,
             next_turn_is_first: true,
         }
@@ -218,17 +219,17 @@ impl SessionState {
         self.active_connector_selection.clear();
     }
 
-    pub(crate) fn set_pending_session_start_source(
+    pub(crate) fn queue_pending_session_start_source(
         &mut self,
-        value: Option<codex_hooks::SessionStartSource>,
+        value: codex_hooks::SessionStartSource,
     ) {
-        self.pending_session_start_source = value;
+        self.pending_session_start_sources.push_back(value);
     }
 
     pub(crate) fn take_pending_session_start_source(
         &mut self,
     ) -> Option<codex_hooks::SessionStartSource> {
-        self.pending_session_start_source.take()
+        self.pending_session_start_sources.pop_front()
     }
 
     pub(crate) fn record_granted_permissions(&mut self, permissions: AdditionalPermissionProfile) {