diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index b3ed451e43..5314d1c9ef 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -137,6 +137,7 @@ use codex_app_server_protocol::Turn; use codex_app_server_protocol::TurnCompletedNotification; use codex_app_server_protocol::TurnDiffUpdatedNotification; use codex_app_server_protocol::TurnError as AppServerTurnError; +use codex_app_server_protocol::TurnInterruptResponse; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartedNotification; use codex_app_server_protocol::TurnStatus as AppServerTurnStatus; @@ -485,6 +486,10 @@ fn sample_turn_steer_response(turn_id: &str) -> ClientResponsePayload { }) } +fn sample_turn_interrupt_response() -> ClientResponsePayload { + ClientResponsePayload::TurnInterrupt(TurnInterruptResponse {}) +} + fn no_active_turn_steer_error() -> JSONRPCErrorError { JSONRPCErrorError { code: -32600, @@ -4073,6 +4078,7 @@ fn turn_event_serializes_expected_shape() { num_input_images: 2, is_first_turn: true, status: Some(TurnStatus::Completed), + explicit_client_interrupt_requested_at_ms: None, turn_error: None, codex_error_kind: None, codex_error_http_status_code: None, @@ -4147,6 +4153,7 @@ fn turn_event_serializes_expected_shape() { "num_input_images": 2, "is_first_turn": true, "status": "completed", + "explicit_client_interrupt_requested_at_ms": null, "turn_error": null, "codex_error_kind": null, "codex_error_http_status_code": null, @@ -4942,7 +4949,7 @@ async fn turn_lifecycle_emits_failed_turn_event() { } #[tokio::test] -async fn turn_lifecycle_emits_interrupted_turn_event_without_error() { +async fn rejected_turn_interrupt_does_not_tag_interrupted_turn_event() { let mut reducer = AnalyticsReducer::default(); let mut out = Vec::new(); @@ -4955,6 +4962,28 @@ async fn turn_lifecycle_emits_interrupted_turn_event_without_error() { /*include_token_usage*/ false, ) .await; + reducer + .ingest( + AnalyticsFact::ExplicitClientInterruptRequest { + connection_id: 7, + request_id: RequestId::Integer(4), + turn_id: "turn-2".to_string(), + requested_at_ms: 1716000000123, + }, + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::ErrorResponse { + connection_id: 7, + request_id: RequestId::Integer(4), + error: no_active_turn_steer_error(), + error_type: None, + }, + &mut out, + ) + .await; reducer .ingest( AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( @@ -4970,10 +4999,148 @@ async fn turn_lifecycle_emits_interrupted_turn_event_without_error() { assert_eq!(out.len(), 1); let payload = serde_json::to_value(&out[0]).expect("serialize turn event"); assert_eq!(payload["event_params"]["status"], json!("interrupted")); + assert_eq!( + payload["event_params"]["explicit_client_interrupt_requested_at_ms"], + json!(null) + ); assert_eq!(payload["event_params"]["turn_error"], json!(null)); assert_eq!(payload["event_params"]["codex_error_kind"], json!(null)); } +#[tokio::test] +async fn accepted_turn_interrupt_records_requested_at_on_turn_event() { + let mut reducer = AnalyticsReducer::default(); + let mut out = Vec::new(); + + ingest_turn_prerequisites( + &mut reducer, + &mut out, + /*include_initialize*/ true, + /*include_resolved_config*/ true, + /*include_started*/ true, + /*include_token_usage*/ false, + ) + .await; + reducer + .ingest( + AnalyticsFact::ExplicitClientInterruptRequest { + connection_id: 7, + request_id: RequestId::Integer(4), + turn_id: "turn-2".to_string(), + requested_at_ms: 1716000000123, + }, + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 7, + request_id: RequestId::Integer(4), + response: Box::new(sample_turn_interrupt_response()), + thread_originator: None, + }, + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-2", + "turn-2", + AppServerTurnStatus::Interrupted, + /*codex_error_info*/ None, + ))), + &mut out, + ) + .await; + + assert_eq!(out.len(), 1); + let payload = serde_json::to_value(&out[0]).expect("serialize turn event"); + assert_eq!( + payload["event_params"]["explicit_client_interrupt_requested_at_ms"], + json!(1716000000123_u64) + ); +} + +#[tokio::test] +async fn accepted_turn_interrupt_retries_preserve_earliest_requested_at() { + let mut reducer = AnalyticsReducer::default(); + let mut out = Vec::new(); + + ingest_turn_prerequisites( + &mut reducer, + &mut out, + /*include_initialize*/ true, + /*include_resolved_config*/ true, + /*include_started*/ true, + /*include_token_usage*/ false, + ) + .await; + reducer + .ingest( + AnalyticsFact::ExplicitClientInterruptRequest { + connection_id: 7, + request_id: RequestId::Integer(4), + turn_id: "turn-2".to_string(), + requested_at_ms: 1716000000123, + }, + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::ExplicitClientInterruptRequest { + connection_id: 7, + request_id: RequestId::Integer(5), + turn_id: "turn-2".to_string(), + requested_at_ms: 1716000000456, + }, + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 7, + request_id: RequestId::Integer(5), + response: Box::new(sample_turn_interrupt_response()), + thread_originator: None, + }, + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 7, + request_id: RequestId::Integer(4), + response: Box::new(sample_turn_interrupt_response()), + thread_originator: None, + }, + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-2", + "turn-2", + AppServerTurnStatus::Interrupted, + /*codex_error_info*/ None, + ))), + &mut out, + ) + .await; + + assert_eq!(out.len(), 1); + let payload = serde_json::to_value(&out[0]).expect("serialize turn event"); + assert_eq!( + payload["event_params"]["explicit_client_interrupt_requested_at_ms"], + json!(1716000000123_u64) + ); +} + #[tokio::test] async fn turn_completed_without_started_notification_emits_null_started_at() { let mut reducer = AnalyticsReducer::default(); diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index de9423bfca..b3df51fcd7 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -29,6 +29,7 @@ use crate::facts::TurnCodexErrorFact; use crate::facts::TurnProfileFact; use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnTokenUsageFact; +use crate::now_unix_millis; use crate::reducer::AnalyticsReducer; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ClientResponsePayload; @@ -312,6 +313,18 @@ impl AnalyticsEventsClient { request_id: RequestId, request: &ClientRequest, ) { + if let ClientRequest::TurnInterrupt { params, .. } = request { + if params.turn_id.is_empty() { + return; + } + self.record_fact(AnalyticsFact::ExplicitClientInterruptRequest { + connection_id, + request_id, + turn_id: params.turn_id.clone(), + requested_at_ms: now_unix_millis(), + }); + return; + } if !matches!( request, ClientRequest::TurnStart { .. } | ClientRequest::TurnSteer { .. } @@ -519,6 +532,7 @@ impl AnalyticsEventsClient { | ClientResponsePayload::ThreadFork(_) | ClientResponsePayload::TurnStart(_) | ClientResponsePayload::TurnSteer(_) + | ClientResponsePayload::TurnInterrupt(_) ) { return; } diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index 72072e87df..4391f2c661 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -56,6 +56,8 @@ use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus; use codex_app_server_protocol::Turn; use codex_app_server_protocol::TurnDiffUpdatedNotification; +use codex_app_server_protocol::TurnInterruptParams; +use codex_app_server_protocol::TurnInterruptResponse; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::TurnStatus as AppServerTurnStatus; @@ -465,6 +467,20 @@ fn sample_turn_steer_request() -> ClientRequest { } } +fn sample_turn_interrupt_request(turn_id: &str) -> ClientRequest { + ClientRequest::TurnInterrupt { + request_id: RequestId::Integer(3), + params: TurnInterruptParams { + thread_id: "thread-1".to_string(), + turn_id: turn_id.to_string(), + }, + } +} + +fn sample_turn_interrupt_response() -> ClientResponsePayload { + ClientResponsePayload::TurnInterrupt(TurnInterruptResponse {}) +} + fn sample_thread_archive_request() -> ClientRequest { ClientRequest::ThreadArchive { request_id: RequestId::Integer(3), @@ -599,6 +615,24 @@ fn track_request_only_enqueues_analytics_relevant_requests() { )); } + client.track_request( + /*connection_id*/ 7, + RequestId::Integer(3), + &sample_turn_interrupt_request("turn-1"), + ); + assert!(matches!( + receiver.try_recv(), + Ok(AnalyticsEventsQueueMessage::Fact(input)) + if matches!( + *input, + AnalyticsFact::ExplicitClientInterruptRequest { + ref turn_id, + requested_at_ms, + .. + } if turn_id == "turn-1" && requested_at_ms > 0 + ) + )); + let ignored_request = sample_thread_archive_request(); client.track_request( /*connection_id*/ 7, @@ -606,6 +640,13 @@ fn track_request_only_enqueues_analytics_relevant_requests() { &ignored_request, ); assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); + + client.track_request( + /*connection_id*/ 7, + RequestId::Integer(4), + &sample_turn_interrupt_request(""), + ); + assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); } #[test] @@ -618,6 +659,7 @@ fn track_response_only_enqueues_analytics_relevant_responses() { (RequestId::Integer(3), sample_thread_fork_response()), (RequestId::Integer(4), sample_turn_start_response()), (RequestId::Integer(5), sample_turn_steer_response()), + (RequestId::Integer(6), sample_turn_interrupt_response()), ] { client.track_response(/*connection_id*/ 7, request_id, &response); assert!(matches!( @@ -629,7 +671,7 @@ fn track_response_only_enqueues_analytics_relevant_responses() { client.track_response( /*connection_id*/ 7, - RequestId::Integer(6), + RequestId::Integer(7), &ClientResponsePayload::ThreadArchive(ThreadArchiveResponse {}), ); assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index e5ac0cb1d6..ae87b94fd3 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -897,6 +897,9 @@ pub(crate) struct CodexTurnEventParams { pub(crate) num_input_images: usize, pub(crate) is_first_turn: bool, pub(crate) status: Option, + /// Client wall-clock time for the first non-startup turn/interrupt request + /// that later received a successful response. + pub(crate) explicit_client_interrupt_requested_at_ms: Option, pub(crate) turn_error: Option, pub(crate) codex_error_kind: Option, pub(crate) codex_error_http_status_code: Option, diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index 519007b3f5..674653e485 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -374,6 +374,12 @@ pub(crate) enum AnalyticsFact { request_id: RequestId, request: Box, }, + ExplicitClientInterruptRequest { + connection_id: u64, + request_id: RequestId, + turn_id: String, + requested_at_ms: u64, + }, ClientResponse { connection_id: u64, request_id: RequestId, diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 7777e02397..099d3049e8 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -336,6 +336,7 @@ impl ThreadMetadataState { enum RequestState { TurnStart(PendingTurnStartState), TurnSteer(PendingTurnSteerState), + ExplicitClientInterrupt(PendingTurnInterruptState), } struct PendingTurnStartState { @@ -350,6 +351,11 @@ struct PendingTurnSteerState { created_at: u64, } +struct PendingTurnInterruptState { + turn_id: String, + requested_at_ms: u64, +} + #[derive(Clone)] struct CompletedTurnState { status: Option, @@ -368,6 +374,7 @@ struct TurnState { token_usage: Option, profile: Option, completed: Option, + explicit_client_interrupt_requested_at_ms: Option, codex_error: Option, latest_diff: Option, steer_count: usize, @@ -445,6 +452,20 @@ impl AnalyticsReducer { } => { self.ingest_request(connection_id, request_id, *request); } + AnalyticsFact::ExplicitClientInterruptRequest { + connection_id, + request_id, + turn_id, + requested_at_ms, + } => { + self.requests.insert( + (connection_id, request_id), + RequestState::ExplicitClientInterrupt(PendingTurnInterruptState { + turn_id, + requested_at_ms, + }), + ); + } AnalyticsFact::ClientResponse { connection_id, request_id, @@ -961,6 +982,21 @@ impl AnalyticsReducer { } => { self.ingest_turn_steer_response(connection_id, request_id, response, out); } + ClientResponse::TurnInterrupt { request_id, .. } => { + let Some(RequestState::ExplicitClientInterrupt(pending_request)) = + self.requests.remove(&(connection_id, request_id)) + else { + return; + }; + let turn_id = pending_request.turn_id; + let turn_state = self.turns.entry(turn_id.clone()).or_default(); + let earliest_requested_at_ms = turn_state + .explicit_client_interrupt_requested_at_ms + .get_or_insert(pending_request.requested_at_ms); + *earliest_requested_at_ms = + (*earliest_requested_at_ms).min(pending_request.requested_at_ms); + self.maybe_emit_turn_event(&turn_id, out).await; + } _ => {} } } @@ -1196,6 +1232,7 @@ impl AnalyticsReducer { out, ); } + RequestState::ExplicitClientInterrupt(_) => {} } } @@ -2694,6 +2731,8 @@ fn codex_turn_event_params( num_input_images, is_first_turn, status: completed.status, + explicit_client_interrupt_requested_at_ms: turn_state + .explicit_client_interrupt_requested_at_ms, turn_error: completed.turn_error, codex_error_kind: codex_error.map(|error| error.kind), codex_error_http_status_code: codex_error.and_then(|error| error.http_status_code), @@ -2853,11 +2892,54 @@ pub(crate) fn normalize_path_for_skill_id( #[cfg(test)] mod tests { use super::*; + use codex_app_server_protocol::JSONRPCErrorError; use codex_protocol::models::SandboxEnforcement; use codex_protocol::permissions::FileSystemSandboxPolicy; use codex_protocol::permissions::NetworkSandboxPolicy; use pretty_assertions::assert_eq; + #[tokio::test] + async fn rejected_turn_interrupt_removes_pending_analytics_request() { + let mut reducer = AnalyticsReducer::default(); + let mut out = Vec::new(); + let connection_id = 7; + let request_id = RequestId::Integer(4); + let request_key = (connection_id, request_id.clone()); + + reducer + .ingest( + AnalyticsFact::ExplicitClientInterruptRequest { + connection_id, + request_id: request_id.clone(), + turn_id: "turn-2".to_string(), + requested_at_ms: 1716000000123, + }, + &mut out, + ) + .await; + + assert!(reducer.requests.contains_key(&request_key)); + + reducer + .ingest( + AnalyticsFact::ErrorResponse { + connection_id, + request_id, + error: JSONRPCErrorError { + code: -32600, + message: "no active turn to interrupt".to_string(), + data: None, + }, + error_type: None, + }, + &mut out, + ) + .await; + + assert!(!reducer.requests.contains_key(&request_key)); + assert!(out.is_empty()); + } + #[test] fn managed_full_disk_with_restricted_network_reports_external_sandbox() { let permission_profile = PermissionProfile::from_runtime_permissions_with_enforcement( diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index eee0b0ae27..f2ca6132f3 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -227,9 +227,11 @@ impl TurnRequestProcessor { request_id: &ConnectionRequestId, params: TurnInterruptParams, ) -> Result, JSONRPCErrorError> { - self.turn_interrupt_inner(request_id, params) - .await - .map(|response| response.map(Into::into)) + let result = self.turn_interrupt_inner(request_id, params).await; + if let Err(error) = &result { + self.track_error_response(request_id, error, /*error_type*/ None); + } + result.map(|response| response.map(Into::into)) } pub(crate) async fn thread_realtime_start( diff --git a/codex-rs/app-server/tests/suite/v2/imagegen_extension.rs b/codex-rs/app-server/tests/suite/v2/imagegen_extension.rs index b56e78298b..e4af09a56f 100644 --- a/codex-rs/app-server/tests/suite/v2/imagegen_extension.rs +++ b/codex-rs/app-server/tests/suite/v2/imagegen_extension.rs @@ -93,7 +93,7 @@ async fn standalone_image_generation_returns_saved_path_hint_to_model() -> Resul .with_env_overrides(&[("OPENAI_API_KEY", None)]) .build_initialized_with_timeout(DEFAULT_READ_TIMEOUT) .await?; - start_image_generation_turn( + let turn_id = start_image_generation_turn( &mut mcp, ThreadStartParams { service_name: Some("chatgpt_cca".to_string()), @@ -144,6 +144,7 @@ async fn standalone_image_generation_returns_saved_path_hint_to_model() -> Resul .context("standalone image generation originator should be valid ASCII")?, "chatgpt_cca" ); + assert_image_turn_id_header(&image_request, &turn_id)?; let requests = response_mock.requests(); assert_eq!(requests.len(), 2); @@ -446,7 +447,7 @@ generatedImage(result); async fn start_image_generation_turn( mcp: &mut TestAppServer, thread_start_params: ThreadStartParams, -) -> Result<()> { +) -> Result { start_turn( mcp, thread_start_params, @@ -500,7 +501,7 @@ async fn run_image_edit_test( .with_env_overrides(&[("OPENAI_API_KEY", None)]) .build_initialized_with_timeout(DEFAULT_READ_TIMEOUT) .await?; - start_turn( + let turn_id = start_turn( &mut mcp, ThreadStartParams { service_name: Some("chatgpt_cca".to_string()), @@ -538,14 +539,27 @@ async fn run_image_edit_test( .context("standalone image edit originator should be valid ASCII")?, "chatgpt_cca" ); + assert_image_turn_id_header(image_request, &turn_id)?; Ok(image_request.body_json::()?) } +fn assert_image_turn_id_header(request: &wiremock::Request, expected_turn_id: &str) -> Result<()> { + let turn_id = request + .headers + .get("x-codex-image-turn-id") + .context("image request should include the current turn id")? + .to_str() + .context("image turn id should be valid ASCII")?; + uuid::Uuid::parse_str(turn_id).context("image turn id should be a UUID")?; + assert_eq!(turn_id, expected_turn_id); + Ok(()) +} + async fn start_turn( mcp: &mut TestAppServer, thread_start_params: ThreadStartParams, input: Vec, -) -> Result<()> { +) -> Result { let thread_req = mcp .send_thread_start_request_with_auto_env(thread_start_params) .await?; @@ -560,9 +574,10 @@ async fn start_turn( ..Default::default() }) .await?; - let _: TurnStartResponse = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(turn_req)).await??; + let TurnStartResponse { turn } = + timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(turn_req)).await??; - Ok(()) + Ok(turn.id) } async fn wait_for_image_generation_completed( diff --git a/codex-rs/ext/image-generation/src/backend.rs b/codex-rs/ext/image-generation/src/backend.rs index 761bb1f27d..734d9becb0 100644 --- a/codex-rs/ext/image-generation/src/backend.rs +++ b/codex-rs/ext/image-generation/src/backend.rs @@ -7,6 +7,9 @@ use codex_login::default_client::add_originator_header; use codex_login::default_client::create_client; use codex_model_provider::SharedModelProvider; use http::HeaderMap; +use http::HeaderValue; + +const X_CODEX_IMAGE_TURN_ID_HEADER: &str = "x-codex-image-turn-id"; #[derive(Clone)] pub(crate) struct CodexImagesBackend { @@ -46,26 +49,40 @@ impl CodexImagesBackend { pub(crate) async fn generate( &self, request: ImageGenerationRequest, + turn_id: &str, ) -> Result { self.client() .await? - .generate(&request, image_request_headers(self.originator.as_deref())) + .generate( + &request, + image_request_headers(self.originator.as_deref(), turn_id), + ) .await .map_err(|err| err.to_string()) } /// Sends a standalone image edit request through the configured Images client. - pub(crate) async fn edit(&self, request: ImageEditRequest) -> Result { + pub(crate) async fn edit( + &self, + request: ImageEditRequest, + turn_id: &str, + ) -> Result { self.client() .await? - .edit(&request, image_request_headers(self.originator.as_deref())) + .edit( + &request, + image_request_headers(self.originator.as_deref(), turn_id), + ) .await .map_err(|err| err.to_string()) } } -fn image_request_headers(originator: Option<&str>) -> HeaderMap { +fn image_request_headers(originator: Option<&str>, turn_id: &str) -> HeaderMap { let mut headers = HeaderMap::new(); + if let Ok(turn_id) = HeaderValue::from_str(turn_id) { + headers.insert(X_CODEX_IMAGE_TURN_ID_HEADER, turn_id); + } if let Some(originator) = originator { add_originator_header(&mut headers, originator); } diff --git a/codex-rs/ext/image-generation/src/tool.rs b/codex-rs/ext/image-generation/src/tool.rs index 6299c3e271..87d8a0d44f 100644 --- a/codex-rs/ext/image-generation/src/tool.rs +++ b/codex-rs/ext/image-generation/src/tool.rs @@ -150,8 +150,8 @@ impl ImageGenerationTool { )) .await; let result = match request { - ImageRequest::Generate(request) => self.backend.generate(request).await, - ImageRequest::Edit(request) => self.backend.edit(request).await, + ImageRequest::Generate(request) => self.backend.generate(request, &call.turn_id).await, + ImageRequest::Edit(request) => self.backend.edit(request, &call.turn_id).await, } .map_err(|err| format!("image generation failed: {err}")) .and_then(|response| {