Always use streamed remote compaction for supported providers (#44255)

## What changed

- Route manual and automatic remote compaction through the v2 implementation.
- Retire the `remote_compaction_v2` feature toggle while continuing to advertise it in the model client beta features header.
- Migrate compaction tests from `/responses/compact` to streamed `/responses` requests with `compaction_trigger` items and remove legacy parity coverage.

## Testing

Add streamed compaction coverage for tool search output trimming and session base instruction token estimates. Update app-server tests to verify started/completed notifications for both local and streamed remote compaction.

GitOrigin-RevId: cea392a6166a9b116ff83708fd9126d5546476e4

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

@@ -1088,28 +1088,20 @@ impl Session {
     ///
     /// `ModelClient` is session-scoped and intentionally does not depend on the full `Config`, so
     /// we precompute the comma-separated list of enabled experimental feature keys at session
-    /// creation time and thread it into the client.
+    /// creation time and thread it into the client. Remote compaction stays advertised unconditionally.
     fn build_model_client_beta_features_header(config: &Config) -> Option<String> {
-        let beta_features_header = FEATURES
-            .iter()
-            .filter_map(|spec| {
-                let advertise_in_model_client_header =
-                    spec.stage.experimental_menu_description().is_some()
-                        || spec.id == Feature::RemoteCompactionV2;
-                if advertise_in_model_client_header && config.features.enabled(spec.id) {
-                    Some(spec.key)
-                } else {
-                    None
-                }
-            })
-            .collect::<Vec<_>>()
-            .join(",");
-
-        if beta_features_header.is_empty() {
-            None
-        } else {
-            Some(beta_features_header)
-        }
+        Some(
+            FEATURES
+                .iter()
+                .filter(|spec| {
+                    spec.id == Feature::RemoteCompactionV2
+                        || (spec.stage.experimental_menu_description().is_some()
+                            && config.features.enabled(spec.id))
+                })
+                .map(|spec| spec.key)
+                .collect::<Vec<_>>()
+                .join(","),
+        )
     }
 
     async fn start_managed_network_proxy(

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

@@ -11314,7 +11314,7 @@ async fn make_remote_compaction_session(
 #[test_case(FirstAttempt::Succeeds; "primary succeeds")]
 #[test_case(FirstAttempt::Retries; "fallback executes")]
 #[tokio::test]
-async fn legacy_compaction_retains_only_the_selected_step(first_attempt: FirstAttempt) {
+async fn remote_compaction_v2_retains_only_the_selected_step(first_attempt: FirstAttempt) {
     let server = responses::start_mock_server().await;
     let (session, turn, events) = make_remote_compaction_session(&server.uri()).await;
     session
@@ -11360,9 +11360,15 @@ async fn legacy_compaction_retains_only_the_selected_step(first_attempt: FirstAt
         Some(Arc::as_ptr(&primary)),
     );
 
-    let success = ResponseTemplate::new(/*status*/ 200).set_body_json(json!({
-        "output": [{ "type": "compaction", "encrypted_content": "summary" }]
-    }));
+    let success = ResponseTemplate::new(/*status*/ 200)
+        .insert_header("content-type", "text/event-stream")
+        .set_body_string(responses::sse(vec![
+            json!({
+                "type": "response.output_item.done",
+                "item": { "type": "compaction", "encrypted_content": "summary" },
+            }),
+            responses::ev_completed("compact-response"),
+        ]));
     let replies = match first_attempt {
         FirstAttempt::Succeeds => vec![success],
         FirstAttempt::Retries => vec![
@@ -11371,12 +11377,13 @@ async fn legacy_compaction_retains_only_the_selected_step(first_attempt: FirstAt
             success,
         ],
     };
-    let requests = responses::mount_compact_response_sequence(&server, replies).await;
-    crate::compact_remote::run_inline_remote_auto_compact_task(
+    let requests = responses::mount_response_sequence(&server, replies).await;
+    let mut client_session = session.services.model_client.new_session();
+    crate::compact_remote_v2::run_inline_remote_auto_compact_task(
         Arc::clone(&session),
         Arc::clone(&primary),
         Some(Arc::clone(&fallback)),
-        Arc::new(OnceLock::new()),
+        &mut client_session,
         InitialContextInjection::DoNotInject,
         CompactionReason::ModelDownshift,
         CompactionPhase::PreTurn,
@@ -11405,6 +11412,12 @@ async fn legacy_compaction_retains_only_the_selected_step(first_attempt: FirstAt
             .collect::<Vec<_>>(),
         models,
     );
+    assert!(
+        requests
+            .requests()
+            .iter()
+            .all(|request| request.inputs_of_type("compaction_trigger").len() == 1),
+    );
     session.abort_all_tasks(TurnAbortReason::Interrupted).await;
     recv_terminal_event(&events, TerminalEventKind::TurnAborted).await;
 }

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

@@ -39,11 +39,9 @@ use codex_network_proxy::NetworkProtocol;
 use codex_protocol::config_types::ApprovalsReviewer;
 use codex_protocol::error::SandboxErr;
 use codex_protocol::models::AdditionalPermissionProfile as PermissionProfile;
-use codex_protocol::models::ContentItem;
 use codex_protocol::models::ContentItemKind;
 use codex_protocol::models::NetworkPermissions;
 use codex_protocol::models::ResponseInputItem;
-use codex_protocol::models::ResponseItem;
 use codex_protocol::protocol::AskForApproval;
 use codex_protocol::request_permissions::PermissionGrantScope;
 use codex_protocol::request_permissions::RequestPermissionProfile;
@@ -1053,7 +1051,7 @@ async fn guardian_allows_unified_exec_additional_permissions_requests_past_polic
 }
 
 #[tokio::test]
-async fn process_compacted_history_preserves_separate_guardian_developer_message() {
+async fn compaction_initial_context_preserves_separate_guardian_developer_message() {
     let (session, mut turn_context) = make_session_and_context().await;
     update_turn_settings_for_test(&mut turn_context, |settings| {
         update_selected_settings_for_test(settings, |selected| {
@@ -1084,40 +1082,20 @@ async fn process_compacted_history_preserves_separate_guardian_developer_message
         step_context,
     };
 
-    let (refreshed, _) = crate::compact_remote::process_compacted_history(
-        &session,
-        vec![
-            ResponseItem::Message {
-                id: None,
-                role: "developer".to_string(),
-                content: vec![ContentItem::InputText {
-                    text: "stale developer message".to_string(),
-                }],
-                phase: None,
-                internal_chat_message_metadata_passthrough: None,
-            },
-            ResponseItem::Message {
-                id: None,
-                role: "user".to_string(),
-                content: vec![ContentItem::InputText {
-                    text: "summary".to_string(),
-                }],
-                phase: None,
-                internal_chat_message_metadata_passthrough: None,
-            },
-        ],
-        &initial_context_injection,
-    )
-    .await;
+    let (refreshed, _) =
+        crate::compact::build_compaction_initial_context(&session, &initial_context_injection)
+            .await;
 
     let developer_messages = refreshed
         .iter()
-        .filter_map(|item| match item {
+        .filter_map(|envelope| match &envelope.item {
             ResponseItem::Message { role, content, .. } if role == "developer" => {
                 crate::content_items_to_text(content).map(|text| {
                     (
                         text,
-                        item.executed_tool_call_metadata()
+                        envelope
+                            .item
+                            .executed_tool_call_metadata()
                             .and_then(|metadata| metadata.content_item_kinds.clone()),
                     )
                 })
@@ -1126,11 +1104,6 @@ async fn process_compacted_history_preserves_separate_guardian_developer_message
         })
         .collect::<Vec<_>>();
 
-    assert!(
-        !developer_messages
-            .iter()
-            .any(|(message, _)| message.contains("stale developer message"))
-    );
     assert!(
         !developer_messages
             .iter()

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

@@ -9,7 +9,6 @@ use crate::client_common::Prompt;
 use crate::client_common::ResponseEvent;
 use crate::compact::InitialContextInjection;
 use crate::compact::run_inline_auto_compact_task;
-use crate::compact_remote::run_inline_remote_auto_compact_task;
 use crate::compact_remote_v2::run_inline_remote_auto_compact_task as run_inline_remote_auto_compact_task_v2;
 use crate::connectors;
 use crate::context::ContextualUserFragment;
@@ -1315,12 +1314,7 @@ async fn run_auto_compact(
     }
 
     match turn_context.provider.capabilities().remote_compaction {
-        RemoteCompactionSupport::V2
-            if turn_context
-                .config
-                .features
-                .enabled(Feature::RemoteCompactionV2) =>
-        {
+        RemoteCompactionSupport::V2 => {
             emit_compact_metric(
                 &sess.services.session_telemetry,
                 "remote_v2",
@@ -1337,23 +1331,6 @@ async fn run_auto_compact(
             )
             .await?;
         }
-        RemoteCompactionSupport::V2 => {
-            emit_compact_metric(
-                &sess.services.session_telemetry,
-                "remote",
-                /*manual*/ false,
-            );
-            run_inline_remote_auto_compact_task(
-                Arc::clone(sess),
-                step_context,
-                fallback_step_context,
-                client_session.turn_state(),
-                initial_context_injection,
-                reason,
-                phase,
-            )
-            .await?;
-        }
         RemoteCompactionSupport::Unsupported => {
             emit_compact_metric(
                 &sess.services.session_telemetry,