diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index 179d1b46b9..9c0c69a1ae 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index a45b943055..2174b1d780 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/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), diff --git a/codex-rs/app-server-protocol/src/protocol/v2/mod.rs b/codex-rs/app-server-protocol/src/protocol/v2/mod.rs index 518c1b5227..5bc7d96f00 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/mod.rs +++ b/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::*; diff --git a/codex-rs/app-server-protocol/src/protocol/v2/rollout.rs b/codex-rs/app-server-protocol/src/protocol/v2/rollout.rs new file mode 100644 index 0000000000..436ec47189 --- /dev/null +++ b/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 {} diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index beecd05576..a61cdb0427 100644 --- a/codex-rs/app-server/README.md +++ b/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 diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 73421175e8..6d9457157c 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/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) diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 9e73f535a6..3338d98af3 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/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; diff --git a/codex-rs/app-server/src/request_processors/rollout.rs b/codex-rs/app-server/src/request_processors/rollout.rs new file mode 100644 index 0000000000..8bf2eaef89 --- /dev/null +++ b/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, JSONRPCErrorError> { + if !self.thread_store.as_any().is::() { + 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())) + } +} diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 5d0cbd3be5..325dff2f8a 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/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"))] diff --git a/codex-rs/app-server/tests/suite/v2/rollout_compress_tests.rs b/codex-rs/app-server/tests/suite/v2/rollout_compress_tests.rs new file mode 100644 index 0000000000..46ad929a22 --- /dev/null +++ b/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(()) +}