Add an experimental rollout compression endpoint (#46020)

## What changed

Add `rollout/compress` to trigger a best-effort background compression pass for cold local rollouts, even when `features.local_thread_store_compression` is disabled. The method takes no parameters and immediately returns `{}` to acknowledge scheduling, not completion. Existing worker locks, concurrency limits, and cooldowns still apply.

Require the `experimentalApi` capability and reject non-local thread stores. Document the endpoint and the requirement that clients sharing the Codex home support compressed rollout files.

## Testing

Add integration tests for compression with the startup flag disabled, lossless rollout readback, experimental capability enforcement, and rejection of non-local thread stores.

GitOrigin-RevId: be73a3b3f37f3adc54512de8256abf684d5d8112

codex-rs/app-server-protocol/src/protocol/common.rs

@@ -715,6 +715,13 @@ client_request_definitions! {
         serialization: global("memory"),
         response: v2::MemoryResetResponse,
     },
+    #[experimental("rollout/compress")]
+    /// Start a best-effort background compression pass for cold local rollouts.
+    RolloutCompress => "rollout/compress" {
+        params: #[ts(type = "undefined")] #[serde(skip_serializing_if = "Option::is_none")] Option<()>,
+        serialization: None,
+        response: v2::RolloutCompressResponse,
+    },
     ThreadUnarchive => "thread/unarchive" {
         params: v2::ThreadUnarchiveParams,
         serialization: thread_id(params.thread_id),

codex-rs/app-server-protocol/src/protocol/v2/mod.rs

@@ -30,6 +30,7 @@ mod project;
 mod realtime;
 mod remote_control;
 mod review;
+mod rollout;
 mod thread;
 mod thread_attachment;
 mod thread_data;
@@ -68,6 +69,7 @@ pub use project::*;
 pub use realtime::*;
 pub use remote_control::*;
 pub use review::*;
+pub use rollout::*;
 pub use shared::*;
 pub use thread::*;
 pub use thread_attachment::*;

codex-rs/app-server-protocol/src/protocol/v2/rollout.rs

@@ -0,0 +1,13 @@
+//! Fire-and-forget maintenance requests for local rollout storage.
+
+use crate::JsonSchema;
+use crate::TS;
+use serde::Deserialize;
+use serde::Serialize;
+
+/// Acknowledges the compression trigger, not completion. Existing maintenance
+/// locks and cooldowns can cause the background pass to skip without doing work.
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct RolloutCompressResponse {}

codex-rs/app-server/README.md

@@ -136,6 +136,20 @@ Failures use the normal JSON-RPC error envelope with closed `{type, reason}` dat
 `invalidRequest`, `unavailable`, `cancelled`, or `failed`. UI clients branch on
 these values rather than message text. Native diagnostic payloads stay private.
 
+## Local rollout compression
+
+The experimental `rollout/compress` method takes no parameters and immediately
+returns `{}` after scheduling one best-effort background pass over the app-server's
+local rollout storage. It does not change `features.local_thread_store_compression`
+or require that startup flag to be enabled. Non-local thread stores do not support
+this method.
+
+The worker retains its existing cold-file checks, maintenance and writer locks,
+concurrency limit, and cooldown. Acknowledgement does not imply completion or that
+any files were compressed; failures are reported through existing logs and metrics.
+There are no progress notifications or cancellation API. Clients sharing this
+Codex home must support compressed rollout files, including shared histories.
+
 ## Managed model provider requirements
 
 Existing threads retain their provider configuration. Input RPCs reject requests when managed

codex-rs/app-server/src/message_processor.rs

@@ -1417,6 +1417,7 @@ impl MessageProcessor {
                 self.thread_processor.memory_status(params).await
             }
             ClientRequest::MemoryReset { .. } => self.thread_processor.memory_reset().await,
+            ClientRequest::RolloutCompress { .. } => self.thread_processor.rollout_compress(),
             ClientRequest::ThreadUnarchive { params, .. } => {
                 self.thread_processor
                     .thread_unarchive(request_id.clone(), params)

codex-rs/app-server/src/request_processors.rs

@@ -557,6 +557,7 @@ mod plugins;
 mod process_exec_processor;
 mod projects;
 mod remote_control_processor;
+mod rollout;
 mod search;
 mod thread_attachments;
 mod thread_enrichment;

codex-rs/app-server/src/request_processors/rollout.rs

@@ -0,0 +1,21 @@
+//! Triggers local rollout maintenance without waiting for the background pass.
+
+use super::ThreadRequestProcessor;
+use super::thread_processor::unsupported_thread_store_operation;
+use codex_app_server_protocol::ClientResponsePayload;
+use codex_app_server_protocol::JSONRPCErrorError;
+use codex_app_server_protocol::RolloutCompressResponse;
+use codex_thread_store::LocalThreadStore;
+
+impl ThreadRequestProcessor {
+    pub(crate) fn rollout_compress(
+        &self,
+    ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
+        if !self.thread_store.as_any().is::<LocalThreadStore>() {
+            return Err(unsupported_thread_store_operation("rollout/compress"));
+        }
+
+        codex_rollout::spawn_rollout_compression_worker(self.config.codex_home.to_path_buf());
+        Ok(Some(RolloutCompressResponse {}.into()))
+    }
+}

codex-rs/app-server/tests/suite/v2/mod.rs

@@ -103,6 +103,8 @@ mod request_user_input;
 mod request_validation;
 mod residency;
 mod review;
+#[path = "rollout_compress_tests.rs"]
+mod rollout_compress;
 mod rollout_migration;
 mod safety_check_downgrade;
 #[cfg(not(target_os = "windows"))]

codex-rs/app-server/tests/suite/v2/rollout_compress_tests.rs

@@ -0,0 +1,131 @@
+//! Exercises the compression trigger through the public app-server API.
+
+use std::time::Duration;
+
+use anyhow::Result;
+use app_test_support::DEFAULT_CLIENT_NAME;
+use app_test_support::TestAppServer;
+use app_test_support::create_fake_rollout;
+use app_test_support::rollout_path;
+use codex_app_server_protocol::ClientInfo;
+use codex_app_server_protocol::ClientRequest;
+use codex_app_server_protocol::InitializeCapabilities;
+use codex_app_server_protocol::JSONRPCErrorError;
+use codex_app_server_protocol::JSONRPCMessage;
+use codex_app_server_protocol::RequestId;
+use codex_app_server_protocol::RolloutCompressResponse;
+use pretty_assertions::assert_eq;
+use tempfile::TempDir;
+use tokio::time::timeout;
+
+const READ_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 20);
+
+#[tokio::test]
+async fn rollout_compress_runs_after_startup_with_compression_disabled() -> Result<()> {
+    let codex_home = TempDir::new()?;
+    let mut app_server = TestAppServer::builder()
+        .with_codex_home(codex_home.path())
+        .with_args(&["-c", "features.local_thread_store_compression=false"])
+        .build_initialized()
+        .await?;
+
+    let filename_ts = "2025-01-05T12-00-00";
+    let thread_id = create_fake_rollout(
+        codex_home.path(),
+        filename_ts,
+        "2025-01-05T12:00:00Z",
+        "Saved user message",
+        /*model_provider*/ None,
+        /*git_info*/ None,
+    )?;
+    let path = rollout_path(codex_home.path(), filename_ts, &thread_id);
+    let original = std::fs::read_to_string(&path)?;
+    let compressed_path = path.with_extension("jsonl.zst");
+
+    let response: RolloutCompressResponse = app_server
+        .request(|request_id| ClientRequest::RolloutCompress {
+            request_id,
+            params: None,
+        })
+        .await?;
+    assert_eq!(response, RolloutCompressResponse {});
+    timeout(READ_TIMEOUT, async {
+        while path.exists() || !compressed_path.exists() {
+            tokio::time::sleep(Duration::from_millis(/*millis*/ 10)).await;
+        }
+    })
+    .await?;
+
+    let mut reader = codex_rollout::open_rollout_line_reader(&path).await?;
+    let mut lines = Vec::new();
+    while let Some(line) = reader.next_line().await? {
+        lines.push(line);
+    }
+    assert_eq!(lines.join("\n") + "\n", original);
+    Ok(())
+}
+
+#[tokio::test]
+async fn rollout_compress_requires_experimental_capability() -> Result<()> {
+    let mut app_server = TestAppServer::builder().build().await?;
+    let initialization = app_server
+        .initialize_with_capabilities(
+            ClientInfo {
+                name: DEFAULT_CLIENT_NAME.to_string(),
+                title: None,
+                version: "0.1.0".to_string(),
+            },
+            Some(InitializeCapabilities {
+                experimental_api: false,
+                ..Default::default()
+            }),
+        )
+        .await?;
+    assert!(matches!(initialization, JSONRPCMessage::Response(_)));
+
+    let request_id = app_server
+        .send_raw_request("rollout/compress", /*params*/ None)
+        .await?;
+    let error = timeout(
+        READ_TIMEOUT,
+        app_server.read_stream_until_error_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    assert_eq!(
+        error.error,
+        JSONRPCErrorError {
+            code: -32600,
+            message: "rollout/compress requires experimentalApi capability".to_string(),
+            data: None,
+        }
+    );
+    Ok(())
+}
+
+#[tokio::test]
+async fn rollout_compress_rejects_non_local_thread_stores() -> Result<()> {
+    let mut app_server = TestAppServer::builder()
+        .with_args(&[
+            "-c",
+            "experimental_thread_store={type=\"in_memory\",id=\"rollout-compress-test\"}",
+        ])
+        .build_initialized()
+        .await?;
+    let request_id = app_server
+        .send_raw_request("rollout/compress", /*params*/ None)
+        .await?;
+    let error = timeout(
+        READ_TIMEOUT,
+        app_server.read_stream_until_error_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    assert_eq!(
+        error.error,
+        JSONRPCErrorError {
+            code: -32601,
+            message: "rollout/compress is not supported yet".to_string(),
+            data: None,
+        }
+    );
+    Ok(())
+}