From e8e327d0c86784da5042ed2257631362074476e3 Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Tue, 16 Dec 2025 23:36:32 -0800 Subject: [PATCH] add compact to app server --- .../src/protocol/common.rs | 4 + .../app-server-protocol/src/protocol/v2.rs | 7 ++ codex-rs/app-server/README.md | 1 + .../app-server/src/codex_message_processor.rs | 50 ++++++++ .../app-server/tests/common/mcp_process.rs | 10 ++ codex-rs/app-server/tests/suite/v2/compact.rs | 111 ++++++++++++++++++ codex-rs/app-server/tests/suite/v2/mod.rs | 1 + 7 files changed, 184 insertions(+) create mode 100644 codex-rs/app-server/tests/suite/v2/compact.rs diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index bd7fd8e28c..bbd13729ef 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -133,6 +133,10 @@ client_request_definitions! { params: v2::ReviewStartParams, response: v2::ReviewStartResponse, }, + CompactStart => "compact/start" { + params: v2::CompactStartParams, + response: v2::TurnStartResponse, + }, ModelList => "model/list" { params: v2::ModelListParams, diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index 2acea04e9a..9667171072 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -1203,6 +1203,13 @@ pub struct TurnStartParams { pub summary: Option, } +#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct CompactStartParams { + pub thread_id: String, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 460418c81e..520b10b7a6 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -75,6 +75,7 @@ Example (from OpenAI's official VSCode extension): - `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. - `turn/interrupt` — request cancellation of an in-flight turn by `(thread_id, turn_id)`; success is an empty `{}` response and the turn finishes with `status: "interrupted"`. - `review/start` — kick off Codex’s automated reviewer for a thread; responds like `turn/start` and emits `item/started`/`item/completed` notifications with `enteredReviewMode` and `exitedReviewMode` items, plus a final assistant `agentMessage` containing the review. +- `compact/start` — compact (summarize) a thread’s history to reduce context usage; responds like `turn/start` and emits `thread/compacted` when the new compacted history is installed. - `command/exec` — run a single command under the server sandbox without starting a thread/turn (handy for utilities and validation). - `model/list` — list available models (with reasoning effort options). - `skills/list` — list skills for one or more `cwd` values. diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 0bcac39a55..01ab8e3c4b 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -23,6 +23,7 @@ use codex_app_server_protocol::CancelLoginAccountStatus; use codex_app_server_protocol::CancelLoginChatGptResponse; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::CommandExecParams; +use codex_app_server_protocol::CompactStartParams; use codex_app_server_protocol::ConversationGitInfo; use codex_app_server_protocol::ConversationSummary; use codex_app_server_protocol::ExecOneOffCommandResponse; @@ -380,6 +381,9 @@ impl CodexMessageProcessor { ClientRequest::ReviewStart { request_id, params } => { self.review_start(request_id, params).await; } + ClientRequest::CompactStart { request_id, params } => { + self.compact_start(request_id, params).await; + } ClientRequest::NewConversation { request_id, params } => { // Do not tokio::spawn() to process new_conversation() // asynchronously because we need to ensure the conversation is @@ -2756,6 +2760,52 @@ impl CodexMessageProcessor { } } + async fn compact_start(&self, request_id: RequestId, params: CompactStartParams) { + let (_, conversation) = match self.conversation_from_thread_id(¶ms.thread_id).await { + Ok(v) => v, + Err(error) => { + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + let thread_id = params.thread_id; + let turn_id = conversation.submit(Op::Compact).await; + + match turn_id { + Ok(turn_id) => { + let turn = Turn { + id: turn_id, + items: vec![], + error: None, + status: TurnStatus::InProgress, + }; + + self.outgoing + .send_response(request_id, TurnStartResponse { turn: turn.clone() }) + .await; + + self.outgoing + .send_server_notification(ServerNotification::TurnStarted( + TurnStartedNotification { thread_id, turn }, + )) + .await; + } + Err(err) => { + self.outgoing + .send_error( + request_id, + JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + message: format!("failed to start compact: {err}"), + data: None, + }, + ) + .await; + } + } + } + fn build_review_turn(turn_id: String, display_text: &str) -> Turn { let items = if display_text.is_empty() { Vec::new() diff --git a/codex-rs/app-server/tests/common/mcp_process.rs b/codex-rs/app-server/tests/common/mcp_process.rs index e2da40bf2d..bada07d126 100644 --- a/codex-rs/app-server/tests/common/mcp_process.rs +++ b/codex-rs/app-server/tests/common/mcp_process.rs @@ -18,6 +18,7 @@ use codex_app_server_protocol::CancelLoginAccountParams; use codex_app_server_protocol::CancelLoginChatGptParams; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientNotification; +use codex_app_server_protocol::CompactStartParams; use codex_app_server_protocol::ConfigBatchWriteParams; use codex_app_server_protocol::ConfigReadParams; use codex_app_server_protocol::ConfigValueWriteParams; @@ -390,6 +391,15 @@ impl McpProcess { self.send_request("review/start", params).await } + /// Send a `compact/start` JSON-RPC request (v2). + pub async fn send_compact_start_request( + &mut self, + params: CompactStartParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("compact/start", params).await + } + /// Send a `cancelLoginChatGpt` JSON-RPC request. pub async fn send_cancel_login_chat_gpt_request( &mut self, diff --git a/codex-rs/app-server/tests/suite/v2/compact.rs b/codex-rs/app-server/tests/suite/v2/compact.rs new file mode 100644 index 0000000000..a4a607529b --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/compact.rs @@ -0,0 +1,111 @@ +use anyhow::Result; +use app_test_support::McpProcess; +use app_test_support::create_final_assistant_message_sse_response; +use app_test_support::create_mock_chat_completions_server_unchecked; +use app_test_support::to_response; +use codex_app_server_protocol::CompactStartParams; +use codex_app_server_protocol::ContextCompactedNotification; +use codex_app_server_protocol::JSONRPCNotification; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnCompletedNotification; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::TurnStatus; +use tempfile::TempDir; +use tokio::time::timeout; + +const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +#[tokio::test] +async fn compact_start_emits_context_compacted_notification() -> Result<()> { + let responses = vec![create_final_assistant_message_sse_response( + "compacted summary", + )?]; + let server = create_mock_chat_completions_server_unchecked(responses).await; + + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let thread_id = start_default_thread(&mut mcp).await?; + + let compact_req = mcp + .send_compact_start_request(CompactStartParams { + thread_id: thread_id.clone(), + }) + .await?; + let compact_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(compact_req)), + ) + .await??; + let TurnStartResponse { turn } = to_response::(compact_resp)?; + assert_eq!(turn.status, TurnStatus::InProgress); + let turn_id = turn.id.clone(); + + let compacted_notif: JSONRPCNotification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("thread/compacted"), + ) + .await??; + let compacted: ContextCompactedNotification = + serde_json::from_value(compacted_notif.params.expect("params must be present"))?; + assert_eq!(compacted.thread_id, thread_id); + assert_eq!(compacted.turn_id, turn_id); + + let completed_notif: JSONRPCNotification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + let completed: TurnCompletedNotification = + serde_json::from_value(completed_notif.params.expect("params must be present"))?; + assert_eq!(completed.thread_id, compacted.thread_id); + assert_eq!(completed.turn.id, turn_id); + assert_eq!(completed.turn.status, TurnStatus::Completed); + + Ok(()) +} + +async fn start_default_thread(mcp: &mut McpProcess) -> Result { + let thread_req = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let thread_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_req)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response::(thread_resp)?; + Ok(thread.id) +} + +fn create_config_toml(codex_home: &std::path::Path, server_uri: &str) -> std::io::Result<()> { + let config_toml = codex_home.join("config.toml"); + std::fs::write( + config_toml, + format!( + r#" +model = "mock-model" +approval_policy = "never" +sandbox_mode = "read-only" + +model_provider = "mock_provider" + +[model_providers.mock_provider] +name = "Mock provider" +base_url = "{server_uri}/v1" +wire_api = "chat" +request_max_retries = 0 +stream_max_retries = 0 +"# + ), + ) +} diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 16d2142b2e..4ce5322b18 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -1,4 +1,5 @@ mod account; +mod compact; mod config_rpc; mod model_list; mod rate_limits;