mirror of
https://github.com/openai/codex.git
synced 2026-11-07 16:35:32 +00:00
Correlate image requests and explicit interrupts with turns (#36092)
## What changed - Send the active turn ID in the `x-codex-image-turn-id` header for image generation and edit requests. - Add `explicit_client_interrupt_requested_at_ms` to turn analytics. Record the earliest explicit interrupt request that receives a successful response, while excluding rejected requests. - Track errors returned by `turn/interrupt` so pending analytics state is cleaned up correctly. ## Testing - Cover turn ID headers for image generation and editing. - Cover accepted, retried, and rejected explicit interrupt requests in analytics tests. GitOrigin-RevId: 80fe996a161b4952b79c5c38238f75b7661b92af
This commit is contained in:
1 parent
88ec932e96
commit
6219b7c40f
10 files changed
+365
-17
No files matched your search
@@ -137,6 +137,7 @@ use codex_app_server_protocol::Turn;
|
|||||||
use codex_app_server_protocol::TurnCompletedNotification;
|
use codex_app_server_protocol::TurnCompletedNotification;
|
||||||
use codex_app_server_protocol::TurnDiffUpdatedNotification;
|
use codex_app_server_protocol::TurnDiffUpdatedNotification;
|
||||||
use codex_app_server_protocol::TurnError as AppServerTurnError;
|
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::TurnStartParams;
|
||||||
use codex_app_server_protocol::TurnStartedNotification;
|
use codex_app_server_protocol::TurnStartedNotification;
|
||||||
use codex_app_server_protocol::TurnStatus as AppServerTurnStatus;
|
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 {
|
fn no_active_turn_steer_error() -> JSONRPCErrorError {
|
||||||
JSONRPCErrorError {
|
JSONRPCErrorError {
|
||||||
code: -32600,
|
code: -32600,
|
||||||
@@ -4073,6 +4078,7 @@ fn turn_event_serializes_expected_shape() {
|
|||||||
num_input_images: 2,
|
num_input_images: 2,
|
||||||
is_first_turn: true,
|
is_first_turn: true,
|
||||||
status: Some(TurnStatus::Completed),
|
status: Some(TurnStatus::Completed),
|
||||||
|
explicit_client_interrupt_requested_at_ms: None,
|
||||||
turn_error: None,
|
turn_error: None,
|
||||||
codex_error_kind: None,
|
codex_error_kind: None,
|
||||||
codex_error_http_status_code: None,
|
codex_error_http_status_code: None,
|
||||||
@@ -4147,6 +4153,7 @@ fn turn_event_serializes_expected_shape() {
|
|||||||
"num_input_images": 2,
|
"num_input_images": 2,
|
||||||
"is_first_turn": true,
|
"is_first_turn": true,
|
||||||
"status": "completed",
|
"status": "completed",
|
||||||
|
"explicit_client_interrupt_requested_at_ms": null,
|
||||||
"turn_error": null,
|
"turn_error": null,
|
||||||
"codex_error_kind": null,
|
"codex_error_kind": null,
|
||||||
"codex_error_http_status_code": null,
|
"codex_error_http_status_code": null,
|
||||||
@@ -4942,7 +4949,7 @@ async fn turn_lifecycle_emits_failed_turn_event() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[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 reducer = AnalyticsReducer::default();
|
||||||
let mut out = Vec::new();
|
let mut out = Vec::new();
|
||||||
|
|
||||||
@@ -4955,6 +4962,28 @@ async fn turn_lifecycle_emits_interrupted_turn_event_without_error() {
|
|||||||
/*include_token_usage*/ false,
|
/*include_token_usage*/ false,
|
||||||
)
|
)
|
||||||
.await;
|
.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
|
reducer
|
||||||
.ingest(
|
.ingest(
|
||||||
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
|
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);
|
assert_eq!(out.len(), 1);
|
||||||
let payload = serde_json::to_value(&out[0]).expect("serialize turn event");
|
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"]["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"]["turn_error"], json!(null));
|
||||||
assert_eq!(payload["event_params"]["codex_error_kind"], 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]
|
#[tokio::test]
|
||||||
async fn turn_completed_without_started_notification_emits_null_started_at() {
|
async fn turn_completed_without_started_notification_emits_null_started_at() {
|
||||||
let mut reducer = AnalyticsReducer::default();
|
let mut reducer = AnalyticsReducer::default();
|
||||||
|
|||||||
@@ -29,6 +29,7 @@ use crate::facts::TurnCodexErrorFact;
|
|||||||
use crate::facts::TurnProfileFact;
|
use crate::facts::TurnProfileFact;
|
||||||
use crate::facts::TurnResolvedConfigFact;
|
use crate::facts::TurnResolvedConfigFact;
|
||||||
use crate::facts::TurnTokenUsageFact;
|
use crate::facts::TurnTokenUsageFact;
|
||||||
|
use crate::now_unix_millis;
|
||||||
use crate::reducer::AnalyticsReducer;
|
use crate::reducer::AnalyticsReducer;
|
||||||
use codex_app_server_protocol::ClientRequest;
|
use codex_app_server_protocol::ClientRequest;
|
||||||
use codex_app_server_protocol::ClientResponsePayload;
|
use codex_app_server_protocol::ClientResponsePayload;
|
||||||
@@ -312,6 +313,18 @@ impl AnalyticsEventsClient {
|
|||||||
request_id: RequestId,
|
request_id: RequestId,
|
||||||
request: &ClientRequest,
|
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!(
|
if !matches!(
|
||||||
request,
|
request,
|
||||||
ClientRequest::TurnStart { .. } | ClientRequest::TurnSteer { .. }
|
ClientRequest::TurnStart { .. } | ClientRequest::TurnSteer { .. }
|
||||||
@@ -519,6 +532,7 @@ impl AnalyticsEventsClient {
|
|||||||
| ClientResponsePayload::ThreadFork(_)
|
| ClientResponsePayload::ThreadFork(_)
|
||||||
| ClientResponsePayload::TurnStart(_)
|
| ClientResponsePayload::TurnStart(_)
|
||||||
| ClientResponsePayload::TurnSteer(_)
|
| ClientResponsePayload::TurnSteer(_)
|
||||||
|
| ClientResponsePayload::TurnInterrupt(_)
|
||||||
) {
|
) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -56,6 +56,8 @@ use codex_app_server_protocol::ThreadStartResponse;
|
|||||||
use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus;
|
use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus;
|
||||||
use codex_app_server_protocol::Turn;
|
use codex_app_server_protocol::Turn;
|
||||||
use codex_app_server_protocol::TurnDiffUpdatedNotification;
|
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::TurnStartParams;
|
||||||
use codex_app_server_protocol::TurnStartResponse;
|
use codex_app_server_protocol::TurnStartResponse;
|
||||||
use codex_app_server_protocol::TurnStatus as AppServerTurnStatus;
|
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 {
|
fn sample_thread_archive_request() -> ClientRequest {
|
||||||
ClientRequest::ThreadArchive {
|
ClientRequest::ThreadArchive {
|
||||||
request_id: RequestId::Integer(3),
|
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();
|
let ignored_request = sample_thread_archive_request();
|
||||||
client.track_request(
|
client.track_request(
|
||||||
/*connection_id*/ 7,
|
/*connection_id*/ 7,
|
||||||
@@ -606,6 +640,13 @@ fn track_request_only_enqueues_analytics_relevant_requests() {
|
|||||||
&ignored_request,
|
&ignored_request,
|
||||||
);
|
);
|
||||||
assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty)));
|
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]
|
#[test]
|
||||||
@@ -618,6 +659,7 @@ fn track_response_only_enqueues_analytics_relevant_responses() {
|
|||||||
(RequestId::Integer(3), sample_thread_fork_response()),
|
(RequestId::Integer(3), sample_thread_fork_response()),
|
||||||
(RequestId::Integer(4), sample_turn_start_response()),
|
(RequestId::Integer(4), sample_turn_start_response()),
|
||||||
(RequestId::Integer(5), sample_turn_steer_response()),
|
(RequestId::Integer(5), sample_turn_steer_response()),
|
||||||
|
(RequestId::Integer(6), sample_turn_interrupt_response()),
|
||||||
] {
|
] {
|
||||||
client.track_response(/*connection_id*/ 7, request_id, &response);
|
client.track_response(/*connection_id*/ 7, request_id, &response);
|
||||||
assert!(matches!(
|
assert!(matches!(
|
||||||
@@ -629,7 +671,7 @@ fn track_response_only_enqueues_analytics_relevant_responses() {
|
|||||||
|
|
||||||
client.track_response(
|
client.track_response(
|
||||||
/*connection_id*/ 7,
|
/*connection_id*/ 7,
|
||||||
RequestId::Integer(6),
|
RequestId::Integer(7),
|
||||||
&ClientResponsePayload::ThreadArchive(ThreadArchiveResponse {}),
|
&ClientResponsePayload::ThreadArchive(ThreadArchiveResponse {}),
|
||||||
);
|
);
|
||||||
assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty)));
|
assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty)));
|
||||||
|
|||||||
@@ -897,6 +897,9 @@ pub(crate) struct CodexTurnEventParams {
|
|||||||
pub(crate) num_input_images: usize,
|
pub(crate) num_input_images: usize,
|
||||||
pub(crate) is_first_turn: bool,
|
pub(crate) is_first_turn: bool,
|
||||||
pub(crate) status: Option<TurnStatus>,
|
pub(crate) status: Option<TurnStatus>,
|
||||||
|
/// 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<u64>,
|
||||||
pub(crate) turn_error: Option<CodexErrorInfo>,
|
pub(crate) turn_error: Option<CodexErrorInfo>,
|
||||||
pub(crate) codex_error_kind: Option<CodexErrKind>,
|
pub(crate) codex_error_kind: Option<CodexErrKind>,
|
||||||
pub(crate) codex_error_http_status_code: Option<u16>,
|
pub(crate) codex_error_http_status_code: Option<u16>,
|
||||||
|
|||||||
@@ -374,6 +374,12 @@ pub(crate) enum AnalyticsFact {
|
|||||||
request_id: RequestId,
|
request_id: RequestId,
|
||||||
request: Box<ClientRequest>,
|
request: Box<ClientRequest>,
|
||||||
},
|
},
|
||||||
|
ExplicitClientInterruptRequest {
|
||||||
|
connection_id: u64,
|
||||||
|
request_id: RequestId,
|
||||||
|
turn_id: String,
|
||||||
|
requested_at_ms: u64,
|
||||||
|
},
|
||||||
ClientResponse {
|
ClientResponse {
|
||||||
connection_id: u64,
|
connection_id: u64,
|
||||||
request_id: RequestId,
|
request_id: RequestId,
|
||||||
|
|||||||
@@ -336,6 +336,7 @@ impl ThreadMetadataState {
|
|||||||
enum RequestState {
|
enum RequestState {
|
||||||
TurnStart(PendingTurnStartState),
|
TurnStart(PendingTurnStartState),
|
||||||
TurnSteer(PendingTurnSteerState),
|
TurnSteer(PendingTurnSteerState),
|
||||||
|
ExplicitClientInterrupt(PendingTurnInterruptState),
|
||||||
}
|
}
|
||||||
|
|
||||||
struct PendingTurnStartState {
|
struct PendingTurnStartState {
|
||||||
@@ -350,6 +351,11 @@ struct PendingTurnSteerState {
|
|||||||
created_at: u64,
|
created_at: u64,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
struct PendingTurnInterruptState {
|
||||||
|
turn_id: String,
|
||||||
|
requested_at_ms: u64,
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
struct CompletedTurnState {
|
struct CompletedTurnState {
|
||||||
status: Option<TurnStatus>,
|
status: Option<TurnStatus>,
|
||||||
@@ -368,6 +374,7 @@ struct TurnState {
|
|||||||
token_usage: Option<TokenUsage>,
|
token_usage: Option<TokenUsage>,
|
||||||
profile: Option<TurnProfile>,
|
profile: Option<TurnProfile>,
|
||||||
completed: Option<CompletedTurnState>,
|
completed: Option<CompletedTurnState>,
|
||||||
|
explicit_client_interrupt_requested_at_ms: Option<u64>,
|
||||||
codex_error: Option<TurnCodexError>,
|
codex_error: Option<TurnCodexError>,
|
||||||
latest_diff: Option<String>,
|
latest_diff: Option<String>,
|
||||||
steer_count: usize,
|
steer_count: usize,
|
||||||
@@ -445,6 +452,20 @@ impl AnalyticsReducer {
|
|||||||
} => {
|
} => {
|
||||||
self.ingest_request(connection_id, request_id, *request);
|
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 {
|
AnalyticsFact::ClientResponse {
|
||||||
connection_id,
|
connection_id,
|
||||||
request_id,
|
request_id,
|
||||||
@@ -961,6 +982,21 @@ impl AnalyticsReducer {
|
|||||||
} => {
|
} => {
|
||||||
self.ingest_turn_steer_response(connection_id, request_id, response, out);
|
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,
|
out,
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
RequestState::ExplicitClientInterrupt(_) => {}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -2694,6 +2731,8 @@ fn codex_turn_event_params(
|
|||||||
num_input_images,
|
num_input_images,
|
||||||
is_first_turn,
|
is_first_turn,
|
||||||
status: completed.status,
|
status: completed.status,
|
||||||
|
explicit_client_interrupt_requested_at_ms: turn_state
|
||||||
|
.explicit_client_interrupt_requested_at_ms,
|
||||||
turn_error: completed.turn_error,
|
turn_error: completed.turn_error,
|
||||||
codex_error_kind: codex_error.map(|error| error.kind),
|
codex_error_kind: codex_error.map(|error| error.kind),
|
||||||
codex_error_http_status_code: codex_error.and_then(|error| error.http_status_code),
|
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)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use codex_app_server_protocol::JSONRPCErrorError;
|
||||||
use codex_protocol::models::SandboxEnforcement;
|
use codex_protocol::models::SandboxEnforcement;
|
||||||
use codex_protocol::permissions::FileSystemSandboxPolicy;
|
use codex_protocol::permissions::FileSystemSandboxPolicy;
|
||||||
use codex_protocol::permissions::NetworkSandboxPolicy;
|
use codex_protocol::permissions::NetworkSandboxPolicy;
|
||||||
use pretty_assertions::assert_eq;
|
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]
|
#[test]
|
||||||
fn managed_full_disk_with_restricted_network_reports_external_sandbox() {
|
fn managed_full_disk_with_restricted_network_reports_external_sandbox() {
|
||||||
let permission_profile = PermissionProfile::from_runtime_permissions_with_enforcement(
|
let permission_profile = PermissionProfile::from_runtime_permissions_with_enforcement(
|
||||||
|
|||||||
@@ -227,9 +227,11 @@ impl TurnRequestProcessor {
|
|||||||
request_id: &ConnectionRequestId,
|
request_id: &ConnectionRequestId,
|
||||||
params: TurnInterruptParams,
|
params: TurnInterruptParams,
|
||||||
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
|
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
|
||||||
self.turn_interrupt_inner(request_id, params)
|
let result = self.turn_interrupt_inner(request_id, params).await;
|
||||||
.await
|
if let Err(error) = &result {
|
||||||
.map(|response| response.map(Into::into))
|
self.track_error_response(request_id, error, /*error_type*/ None);
|
||||||
|
}
|
||||||
|
result.map(|response| response.map(Into::into))
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) async fn thread_realtime_start(
|
pub(crate) async fn thread_realtime_start(
|
||||||
|
|||||||
@@ -93,7 +93,7 @@ async fn standalone_image_generation_returns_saved_path_hint_to_model() -> Resul
|
|||||||
.with_env_overrides(&[("OPENAI_API_KEY", None)])
|
.with_env_overrides(&[("OPENAI_API_KEY", None)])
|
||||||
.build_initialized_with_timeout(DEFAULT_READ_TIMEOUT)
|
.build_initialized_with_timeout(DEFAULT_READ_TIMEOUT)
|
||||||
.await?;
|
.await?;
|
||||||
start_image_generation_turn(
|
let turn_id = start_image_generation_turn(
|
||||||
&mut mcp,
|
&mut mcp,
|
||||||
ThreadStartParams {
|
ThreadStartParams {
|
||||||
service_name: Some("chatgpt_cca".to_string()),
|
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")?,
|
.context("standalone image generation originator should be valid ASCII")?,
|
||||||
"chatgpt_cca"
|
"chatgpt_cca"
|
||||||
);
|
);
|
||||||
|
assert_image_turn_id_header(&image_request, &turn_id)?;
|
||||||
|
|
||||||
let requests = response_mock.requests();
|
let requests = response_mock.requests();
|
||||||
assert_eq!(requests.len(), 2);
|
assert_eq!(requests.len(), 2);
|
||||||
@@ -446,7 +447,7 @@ generatedImage(result);
|
|||||||
async fn start_image_generation_turn(
|
async fn start_image_generation_turn(
|
||||||
mcp: &mut TestAppServer,
|
mcp: &mut TestAppServer,
|
||||||
thread_start_params: ThreadStartParams,
|
thread_start_params: ThreadStartParams,
|
||||||
) -> Result<()> {
|
) -> Result<String> {
|
||||||
start_turn(
|
start_turn(
|
||||||
mcp,
|
mcp,
|
||||||
thread_start_params,
|
thread_start_params,
|
||||||
@@ -500,7 +501,7 @@ async fn run_image_edit_test(
|
|||||||
.with_env_overrides(&[("OPENAI_API_KEY", None)])
|
.with_env_overrides(&[("OPENAI_API_KEY", None)])
|
||||||
.build_initialized_with_timeout(DEFAULT_READ_TIMEOUT)
|
.build_initialized_with_timeout(DEFAULT_READ_TIMEOUT)
|
||||||
.await?;
|
.await?;
|
||||||
start_turn(
|
let turn_id = start_turn(
|
||||||
&mut mcp,
|
&mut mcp,
|
||||||
ThreadStartParams {
|
ThreadStartParams {
|
||||||
service_name: Some("chatgpt_cca".to_string()),
|
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")?,
|
.context("standalone image edit originator should be valid ASCII")?,
|
||||||
"chatgpt_cca"
|
"chatgpt_cca"
|
||||||
);
|
);
|
||||||
|
assert_image_turn_id_header(image_request, &turn_id)?;
|
||||||
Ok(image_request.body_json::<serde_json::Value>()?)
|
Ok(image_request.body_json::<serde_json::Value>()?)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
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(
|
async fn start_turn(
|
||||||
mcp: &mut TestAppServer,
|
mcp: &mut TestAppServer,
|
||||||
thread_start_params: ThreadStartParams,
|
thread_start_params: ThreadStartParams,
|
||||||
input: Vec<V2UserInput>,
|
input: Vec<V2UserInput>,
|
||||||
) -> Result<()> {
|
) -> Result<String> {
|
||||||
let thread_req = mcp
|
let thread_req = mcp
|
||||||
.send_thread_start_request_with_auto_env(thread_start_params)
|
.send_thread_start_request_with_auto_env(thread_start_params)
|
||||||
.await?;
|
.await?;
|
||||||
@@ -560,9 +574,10 @@ async fn start_turn(
|
|||||||
..Default::default()
|
..Default::default()
|
||||||
})
|
})
|
||||||
.await?;
|
.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(
|
async fn wait_for_image_generation_completed(
|
||||||
|
|||||||
@@ -7,6 +7,9 @@ use codex_login::default_client::add_originator_header;
|
|||||||
use codex_login::default_client::create_client;
|
use codex_login::default_client::create_client;
|
||||||
use codex_model_provider::SharedModelProvider;
|
use codex_model_provider::SharedModelProvider;
|
||||||
use http::HeaderMap;
|
use http::HeaderMap;
|
||||||
|
use http::HeaderValue;
|
||||||
|
|
||||||
|
const X_CODEX_IMAGE_TURN_ID_HEADER: &str = "x-codex-image-turn-id";
|
||||||
|
|
||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
pub(crate) struct CodexImagesBackend {
|
pub(crate) struct CodexImagesBackend {
|
||||||
@@ -46,26 +49,40 @@ impl CodexImagesBackend {
|
|||||||
pub(crate) async fn generate(
|
pub(crate) async fn generate(
|
||||||
&self,
|
&self,
|
||||||
request: ImageGenerationRequest,
|
request: ImageGenerationRequest,
|
||||||
|
turn_id: &str,
|
||||||
) -> Result<ImageResponse, String> {
|
) -> Result<ImageResponse, String> {
|
||||||
self.client()
|
self.client()
|
||||||
.await?
|
.await?
|
||||||
.generate(&request, image_request_headers(self.originator.as_deref()))
|
.generate(
|
||||||
|
&request,
|
||||||
|
image_request_headers(self.originator.as_deref(), turn_id),
|
||||||
|
)
|
||||||
.await
|
.await
|
||||||
.map_err(|err| err.to_string())
|
.map_err(|err| err.to_string())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Sends a standalone image edit request through the configured Images client.
|
/// Sends a standalone image edit request through the configured Images client.
|
||||||
pub(crate) async fn edit(&self, request: ImageEditRequest) -> Result<ImageResponse, String> {
|
pub(crate) async fn edit(
|
||||||
|
&self,
|
||||||
|
request: ImageEditRequest,
|
||||||
|
turn_id: &str,
|
||||||
|
) -> Result<ImageResponse, String> {
|
||||||
self.client()
|
self.client()
|
||||||
.await?
|
.await?
|
||||||
.edit(&request, image_request_headers(self.originator.as_deref()))
|
.edit(
|
||||||
|
&request,
|
||||||
|
image_request_headers(self.originator.as_deref(), turn_id),
|
||||||
|
)
|
||||||
.await
|
.await
|
||||||
.map_err(|err| err.to_string())
|
.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();
|
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 {
|
if let Some(originator) = originator {
|
||||||
add_originator_header(&mut headers, originator);
|
add_originator_header(&mut headers, originator);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -150,8 +150,8 @@ impl ImageGenerationTool {
|
|||||||
))
|
))
|
||||||
.await;
|
.await;
|
||||||
let result = match request {
|
let result = match request {
|
||||||
ImageRequest::Generate(request) => self.backend.generate(request).await,
|
ImageRequest::Generate(request) => self.backend.generate(request, &call.turn_id).await,
|
||||||
ImageRequest::Edit(request) => self.backend.edit(request).await,
|
ImageRequest::Edit(request) => self.backend.edit(request, &call.turn_id).await,
|
||||||
}
|
}
|
||||||
.map_err(|err| format!("image generation failed: {err}"))
|
.map_err(|err| format!("image generation failed: {err}"))
|
||||||
.and_then(|response| {
|
.and_then(|response| {
|
||||||
|
|||||||
Reference in new issue
Block a user