openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-25 18:28:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

[codex-analytics] add steering metadata

Browse Source
This commit is contained in:
rhan-oai
2026-04-07 11:44:35 -07:00
parent 9976d49992
commit 150f8f0a7e
8 changed files with 781 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/analytics/src/analytics_client_tests.rs
+400 -2
View File
@@ -7,6 +7,7 @@ use crate::events::CodexPluginEventRequest;
use crate::events::CodexPluginUsedEventRequest;
use crate::events::CodexRuntimeMetadata;
use crate::events::CodexTurnEventRequest;
use crate::events::CodexTurnSteerEventRequest;
use crate::events::ThreadInitializationMode;
use crate::events::ThreadInitializedEvent;
use crate::events::ThreadInitializedEventParams;
@@ -14,11 +15,13 @@ use crate::events::TrackEventRequest;
use crate::events::codex_app_metadata;
use crate::events::codex_plugin_metadata;
use crate::events::codex_plugin_used_metadata;
use crate::events::codex_turn_steer_event_params;
use crate::events::subagent_thread_started_event_request;
use crate::facts::AnalyticsFact;
use crate::facts::AppInvocation;
use crate::facts::AppMentionedInput;
use crate::facts::AppUsedInput;
use crate::facts::CodexTurnSteerEvent;
use crate::facts::CustomAnalyticsFact;
use crate::facts::InvocationType;
use crate::facts::PluginState;
@@ -30,6 +33,9 @@ use crate::facts::SubAgentThreadStartedInput;
use crate::facts::TrackEventsContext;
use crate::facts::TurnResolvedConfigFact;
use crate::facts::TurnStatus;
use crate::facts::TurnSteerInput;
use crate::facts::TurnSteerRejectionReason;
use crate::facts::TurnSteerResult;
use crate::facts::TurnSubmissionType;
use crate::facts::TurnTokenUsageFact;
use crate::reducer::AnalyticsReducer;
@@ -243,6 +249,25 @@ fn sample_turn_resolved_config(turn_id: &str) -> TurnResolvedConfigFact {
}
}
fn sample_app_server_client_metadata() -> CodexAppServerClientMetadata {
CodexAppServerClientMetadata {
product_client_id: "codex-tui".to_string(),
client_name: Some("codex-tui".to_string()),
client_version: Some("1.0.0".to_string()),
rpc_transport: AppServerRpcTransport::Stdio,
experimental_api_enabled: None,
}
}
fn sample_runtime_metadata() -> CodexRuntimeMetadata {
CodexRuntimeMetadata {
codex_rs_version: "0.1.0".to_string(),
runtime_os: "macos".to_string(),
runtime_os_version: "15.3.1".to_string(),
runtime_arch: "aarch64".to_string(),
}
}
async fn ingest_turn_prerequisites(
reducer: &mut AnalyticsReducer,
out: &mut Vec<TrackEventRequest>,
@@ -1065,7 +1090,7 @@ fn turn_event_serializes_expected_shape() {
is_first_turn: true,
status: Some(TurnStatus::Completed),
turn_error: None,
steer_count: None,
steer_count: Some(0),
total_tool_call_count: None,
shell_command_count: None,
file_change_count: None,
@@ -1111,7 +1136,7 @@ fn turn_event_serializes_expected_shape() {
"is_first_turn": true,
"status": "completed",
"turn_error": null,
"steer_count": null,
"steer_count": 0,
"total_tool_call_count": null,
"shell_command_count": null,
"file_change_count": null,
@@ -1133,6 +1158,280 @@ fn turn_event_serializes_expected_shape() {
);
}
#[test]
fn turn_steer_event_serializes_expected_shape() {
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
};
let event = TrackEventRequest::TurnSteer(CodexTurnSteerEventRequest {
event_type: "codex_turn_steer_event",
event_params: codex_turn_steer_event_params(
sample_app_server_client_metadata(),
sample_runtime_metadata(),
&tracking,
CodexTurnSteerEvent {
expected_turn_id: Some("turn-2".to_string()),
accepted_turn_id: Some("turn-2".to_string()),
num_input_images: 2,
result: TurnSteerResult::Accepted,
rejection_reason: None,
created_at: 1_716_000_123,
},
),
});
let payload = serde_json::to_value(&event).expect("serialize turn steer event");
assert_eq!(payload["event_type"], json!("codex_turn_steer_event"));
assert_eq!(payload["event_params"]["thread_id"], json!("thread-2"));
assert_eq!(payload["event_params"]["expected_turn_id"], json!("turn-2"));
assert_eq!(payload["event_params"]["accepted_turn_id"], json!("turn-2"));
assert_eq!(
payload["event_params"]["app_server_client"],
json!({
"product_client_id": "codex-tui",
"client_name": "codex-tui",
"client_version": "1.0.0",
"rpc_transport": "stdio",
"experimental_api_enabled": null,
})
);
assert_eq!(
payload["event_params"]["runtime"],
json!({
"codex_rs_version": "0.1.0",
"runtime_os": "macos",
"runtime_os_version": "15.3.1",
"runtime_arch": "aarch64",
})
);
assert!(payload["event_params"].get("product_client_id").is_none());
assert_eq!(payload["event_params"]["num_input_images"], json!(2));
assert_eq!(payload["event_params"]["result"], json!("accepted"));
assert_eq!(payload["event_params"]["rejection_reason"], json!(null));
assert_eq!(payload["event_params"]["started_at"], json!(1_716_000_123));
assert!(payload["event_params"].get("created_at").is_none());
}
#[test]
fn rejected_turn_steer_event_serializes_expected_shape() {
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-3".to_string(),
turn_id: "turn-3".to_string(),
};
let event = TrackEventRequest::TurnSteer(CodexTurnSteerEventRequest {
event_type: "codex_turn_steer_event",
event_params: codex_turn_steer_event_params(
sample_app_server_client_metadata(),
sample_runtime_metadata(),
&tracking,
CodexTurnSteerEvent {
expected_turn_id: Some("turn-expected".to_string()),
accepted_turn_id: None,
num_input_images: 1,
result: TurnSteerResult::Rejected,
rejection_reason: Some(TurnSteerRejectionReason::ExpectedTurnMismatch),
created_at: 1_716_000_124,
},
),
});
let payload = serde_json::to_value(&event).expect("serialize rejected turn steer event");
assert_eq!(payload["event_type"], json!("codex_turn_steer_event"));
assert_eq!(payload["event_params"]["thread_id"], json!("thread-3"));
assert_eq!(
payload["event_params"]["expected_turn_id"],
json!("turn-expected")
);
assert_eq!(payload["event_params"]["accepted_turn_id"], json!(null));
assert_eq!(
payload["event_params"]["app_server_client"]["product_client_id"],
json!("codex-tui")
);
assert_eq!(
payload["event_params"]["runtime"]["codex_rs_version"],
json!("0.1.0")
);
assert!(payload["event_params"].get("product_client_id").is_none());
assert_eq!(payload["event_params"]["num_input_images"], json!(1));
assert_eq!(payload["event_params"]["result"], json!("rejected"));
assert_eq!(
payload["event_params"]["rejection_reason"],
json!("expected_turn_mismatch")
);
assert_eq!(payload["event_params"]["started_at"], json!(1_716_000_124));
assert!(payload["event_params"].get("created_at").is_none());
}
#[tokio::test]
async fn turn_steer_uses_connection_metadata() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ false,
/*include_started*/ false,
/*include_token_usage*/ false,
)
.await;
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::TurnSteer(TurnSteerInput {
tracking: TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
},
turn_steer: CodexTurnSteerEvent {
expected_turn_id: Some("turn-2".to_string()),
accepted_turn_id: Some("turn-2".to_string()),
num_input_images: 2,
result: TurnSteerResult::Accepted,
rejection_reason: None,
created_at: 1_716_000_125,
},
})),
&mut out,
)
.await;
assert_eq!(out.len(), 1);
let payload = serde_json::to_value(&out[0]).expect("serialize turn steer event");
assert_eq!(payload["event_type"], json!("codex_turn_steer_event"));
assert_eq!(
payload["event_params"]["app_server_client"],
json!({
"product_client_id": "codex-tui",
"client_name": "codex-tui",
"client_version": "1.0.0",
"rpc_transport": "stdio",
"experimental_api_enabled": null,
})
);
assert_eq!(
payload["event_params"]["runtime"],
json!({
"codex_rs_version": "0.1.0",
"runtime_os": "macos",
"runtime_os_version": "15.3.1",
"runtime_arch": "aarch64",
})
);
assert!(payload["event_params"].get("product_client_id").is_none());
}
#[tokio::test]
async fn rejected_turn_steer_uses_thread_connection_metadata() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
reducer
.ingest(
AnalyticsFact::Initialize {
connection_id: 7,
params: InitializeParams {
client_info: ClientInfo {
name: "codex-tui".to_string(),
title: None,
version: "1.0.0".to_string(),
},
capabilities: None,
},
product_client_id: "codex-tui".to_string(),
runtime: sample_runtime_metadata(),
rpc_transport: AppServerRpcTransport::Stdio,
},
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::Response {
connection_id: 7,
response: Box::new(sample_thread_start_response(
"thread-2", /*ephemeral*/ false, "gpt-5",
)),
},
&mut out,
)
.await;
out.clear();
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::TurnSteer(TurnSteerInput {
tracking: TrackEventsContext {
model_slug: String::new(),
thread_id: "thread-2".to_string(),
turn_id: String::new(),
},
turn_steer: CodexTurnSteerEvent {
expected_turn_id: None,
accepted_turn_id: None,
num_input_images: 1,
result: TurnSteerResult::Rejected,
rejection_reason: Some(TurnSteerRejectionReason::NoActiveTurn),
created_at: 1_716_000_126,
},
})),
&mut out,
)
.await;
assert_eq!(out.len(), 1);
let payload = serde_json::to_value(&out[0]).expect("serialize turn steer event");
assert_eq!(payload["event_type"], json!("codex_turn_steer_event"));
assert_eq!(
payload["event_params"]["app_server_client"]["product_client_id"],
json!("codex-tui")
);
assert_eq!(
payload["event_params"]["runtime"]["codex_rs_version"],
json!("0.1.0")
);
assert_eq!(payload["event_params"]["result"], json!("rejected"));
assert_eq!(
payload["event_params"]["rejection_reason"],
json!("no_active_turn")
);
}
#[tokio::test]
async fn turn_steer_does_not_emit_without_connection_metadata() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::TurnSteer(TurnSteerInput {
tracking: TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
},
turn_steer: CodexTurnSteerEvent {
expected_turn_id: Some("turn-2".to_string()),
accepted_turn_id: None,
num_input_images: 1,
result: TurnSteerResult::Rejected,
rejection_reason: Some(TurnSteerRejectionReason::NoActiveTurn),
created_at: 1_716_000_126,
},
})),
&mut out,
)
.await;
assert!(out.is_empty());
}
#[tokio::test]
async fn turn_lifecycle_emits_turn_event() {
let mut reducer = AnalyticsReducer::default();
@@ -1170,6 +1469,7 @@ async fn turn_lifecycle_emits_turn_event() {
);
assert_eq!(payload["event_params"]["num_input_images"], json!(1));
assert_eq!(payload["event_params"]["status"], json!("completed"));
assert_eq!(payload["event_params"]["steer_count"], json!(0));
assert_eq!(payload["event_params"]["started_at"], json!(455));
assert_eq!(payload["event_params"]["completed_at"], json!(456));
assert_eq!(payload["event_params"]["duration_ms"], json!(1234));
@@ -1183,6 +1483,104 @@ async fn turn_lifecycle_emits_turn_event() {
assert_eq!(payload["event_params"]["total_tokens"], json!(321));
}
#[tokio::test]
async fn accepted_steers_increment_turn_steer_count() {
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::Custom(CustomAnalyticsFact::TurnSteer(TurnSteerInput {
tracking: TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
},
turn_steer: CodexTurnSteerEvent {
expected_turn_id: Some("turn-2".to_string()),
accepted_turn_id: Some("turn-2".to_string()),
num_input_images: 0,
result: TurnSteerResult::Accepted,
rejection_reason: None,
created_at: 1,
},
})),
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::TurnSteer(TurnSteerInput {
tracking: TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
},
turn_steer: CodexTurnSteerEvent {
expected_turn_id: None,
accepted_turn_id: None,
num_input_images: 0,
result: TurnSteerResult::Rejected,
rejection_reason: Some(TurnSteerRejectionReason::NoActiveTurn),
created_at: 2,
},
})),
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::TurnSteer(TurnSteerInput {
tracking: TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
},
turn_steer: CodexTurnSteerEvent {
expected_turn_id: Some("turn-2".to_string()),
accepted_turn_id: Some("turn-2".to_string()),
num_input_images: 1,
result: TurnSteerResult::Accepted,
rejection_reason: None,
created_at: 3,
},
})),
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
let turn_event = out
.iter()
.find(|event| matches!(event, TrackEventRequest::TurnEvent(_)))
.expect("turn event should be emitted");
let payload = serde_json::to_value(turn_event).expect("serialize turn event");
assert_eq!(payload["event_params"]["steer_count"], json!(2));
}
#[tokio::test]
async fn queued_submission_type_emits_queued_turn_event() {
let mut reducer = AnalyticsReducer::default();
codex-rs/analytics/src/client.rs
+11
View File
@@ -6,6 +6,7 @@ use crate::facts::AnalyticsFact;
use crate::facts::AppInvocation;
use crate::facts::AppMentionedInput;
use crate::facts::AppUsedInput;
use crate::facts::CodexTurnSteerEvent;
use crate::facts::CustomAnalyticsFact;
use crate::facts::PluginState;
use crate::facts::PluginStateChangedInput;
@@ -14,6 +15,7 @@ use crate::facts::SkillInvokedInput;
use crate::facts::SubAgentThreadStartedInput;
use crate::facts::TrackEventsContext;
use crate::facts::TurnResolvedConfigFact;
use crate::facts::TurnSteerInput;
use crate::facts::TurnTokenUsageFact;
use crate::reducer::AnalyticsReducer;
use codex_app_server_protocol::ClientRequest;
@@ -203,6 +205,15 @@ impl AnalyticsEventsClient {
)));
}
pub fn track_turn_steer(&self, tracking: TrackEventsContext, turn_steer: CodexTurnSteerEvent) {
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::TurnSteer(
TurnSteerInput {
tracking,
turn_steer,
},
)));
}
pub fn track_plugin_installed(&self, plugin: PluginTelemetryMetadata) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
codex-rs/analytics/src/events.rs
+42
View File
@@ -1,9 +1,12 @@
use crate::facts::AppInvocation;
use crate::facts::CodexTurnSteerEvent;
use crate::facts::InvocationType;
use crate::facts::PluginState;
use crate::facts::SubAgentThreadStartedInput;
use crate::facts::TrackEventsContext;
use crate::facts::TurnStatus;
use crate::facts::TurnSteerRejectionReason;
use crate::facts::TurnSteerResult;
use crate::facts::TurnSubmissionType;
use codex_app_server_protocol::CodexErrorInfo;
use codex_login::default_client::originator;
@@ -41,6 +44,7 @@ pub(crate) enum TrackEventRequest {
AppMentioned(CodexAppMentionedEventRequest),
AppUsed(CodexAppUsedEventRequest),
TurnEvent(Box<CodexTurnEventRequest>),
TurnSteer(CodexTurnSteerEventRequest),
PluginUsed(CodexPluginUsedEventRequest),
PluginInstalled(CodexPluginEventRequest),
PluginUninstalled(CodexPluginEventRequest),
@@ -172,6 +176,25 @@ pub(crate) struct CodexTurnEventRequest {
pub(crate) event_params: CodexTurnEventParams,
}
#[derive(Serialize)]
pub(crate) struct CodexTurnSteerEventParams {
pub(crate) thread_id: String,
pub(crate) expected_turn_id: Option<String>,
pub(crate) accepted_turn_id: Option<String>,
pub(crate) app_server_client: CodexAppServerClientMetadata,
pub(crate) runtime: CodexRuntimeMetadata,
pub(crate) num_input_images: usize,
pub(crate) result: TurnSteerResult,
pub(crate) rejection_reason: Option<TurnSteerRejectionReason>,
pub(crate) started_at: u64,
}
#[derive(Serialize)]
pub(crate) struct CodexTurnSteerEventRequest {
pub(crate) event_type: &'static str,
pub(crate) event_params: CodexTurnSteerEventParams,
}
#[derive(Serialize)]
pub(crate) struct CodexPluginMetadata {
pub(crate) plugin_id: Option<String>,
@@ -263,6 +286,25 @@ pub(crate) fn codex_plugin_used_metadata(
}
}
pub(crate) fn codex_turn_steer_event_params(
app_server_client: CodexAppServerClientMetadata,
runtime: CodexRuntimeMetadata,
tracking: &TrackEventsContext,
turn_steer: CodexTurnSteerEvent,
) -> CodexTurnSteerEventParams {
CodexTurnSteerEventParams {
thread_id: tracking.thread_id.clone(),
expected_turn_id: turn_steer.expected_turn_id,
accepted_turn_id: turn_steer.accepted_turn_id,
app_server_client,
runtime,
num_input_images: turn_steer.num_input_images,
result: turn_steer.result,
rejection_reason: turn_steer.rejection_reason,
started_at: turn_steer.created_at,
}
}
pub(crate) fn thread_source_name(thread_source: &SessionSource) -> Option<&'static str> {
match thread_source {
SessionSource::Cli | SessionSource::VSCode | SessionSource::Exec => Some("user"),
codex-rs/analytics/src/facts.rs
+33
View File
@@ -81,6 +81,33 @@ pub enum TurnStatus {
Interrupted,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum TurnSteerResult {
Accepted,
Rejected,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum TurnSteerRejectionReason {
NoActiveTurn,
ExpectedTurnMismatch,
NonSteerableReview,
NonSteerableCompact,
EmptyInput,
}
#[derive(Clone)]
pub struct CodexTurnSteerEvent {
pub expected_turn_id: Option<String>,
pub accepted_turn_id: Option<String>,
pub num_input_images: usize,
pub result: TurnSteerResult,
pub rejection_reason: Option<TurnSteerRejectionReason>,
pub created_at: u64,
}
#[derive(Clone, Debug)]
pub struct SkillInvocation {
pub skill_name: String,
@@ -142,6 +169,7 @@ pub(crate) enum CustomAnalyticsFact {
SubAgentThreadStarted(SubAgentThreadStartedInput),
TurnResolvedConfig(Box<TurnResolvedConfigFact>),
TurnTokenUsage(Box<TurnTokenUsageFact>),
TurnSteer(TurnSteerInput),
SkillInvoked(SkillInvokedInput),
AppMentioned(AppMentionedInput),
AppUsed(AppUsedInput),
@@ -149,6 +177,11 @@ pub(crate) enum CustomAnalyticsFact {
PluginStateChanged(PluginStateChangedInput),
}
pub(crate) struct TurnSteerInput {
pub tracking: TrackEventsContext,
pub turn_steer: CodexTurnSteerEvent,
}
pub(crate) struct SkillInvokedInput {
pub tracking: TrackEventsContext,
pub invocations: Vec<SkillInvocation>,
codex-rs/analytics/src/lib.rs
+3
View File
@@ -6,12 +6,15 @@ mod reducer;
pub use client::AnalyticsEventsClient;
pub use events::AppServerRpcTransport;
pub use facts::AppInvocation;
pub use facts::CodexTurnSteerEvent;
pub use facts::InvocationType;
pub use facts::SkillInvocation;
pub use facts::SubAgentThreadStartedInput;
pub use facts::TrackEventsContext;
pub use facts::TurnResolvedConfigFact;
pub use facts::TurnStatus;
pub use facts::TurnSteerRejectionReason;
pub use facts::TurnSteerResult;
pub use facts::TurnSubmissionType;
pub use facts::TurnTokenUsageFact;
pub use facts::build_track_events_context;
codex-rs/analytics/src/reducer.rs
+74 -1
View File
@@ -7,6 +7,7 @@ use crate::events::CodexPluginUsedEventRequest;
use crate::events::CodexRuntimeMetadata;
use crate::events::CodexTurnEventParams;
use crate::events::CodexTurnEventRequest;
use crate::events::CodexTurnSteerEventRequest;
use crate::events::SkillInvocationEventParams;
use crate::events::SkillInvocationEventRequest;
use crate::events::ThreadInitializationMode;
@@ -16,20 +17,25 @@ use crate::events::TrackEventRequest;
use crate::events::codex_app_metadata;
use crate::events::codex_plugin_metadata;
use crate::events::codex_plugin_used_metadata;
use crate::events::codex_turn_steer_event_params;
use crate::events::plugin_state_event_type;
use crate::events::subagent_thread_started_event_request;
use crate::events::thread_source_name;
use crate::facts::AnalyticsFact;
use crate::facts::AppMentionedInput;
use crate::facts::AppUsedInput;
use crate::facts::CodexTurnSteerEvent;
use crate::facts::CustomAnalyticsFact;
use crate::facts::PluginState;
use crate::facts::PluginStateChangedInput;
use crate::facts::PluginUsedInput;
use crate::facts::SkillInvokedInput;
use crate::facts::SubAgentThreadStartedInput;
use crate::facts::TrackEventsContext;
use crate::facts::TurnResolvedConfigFact;
use crate::facts::TurnStatus;
use crate::facts::TurnSteerInput;
use crate::facts::TurnSteerResult;
use crate::facts::TurnTokenUsageFact;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ClientResponse;
@@ -57,6 +63,7 @@ pub(crate) struct AnalyticsReducer {
requests: HashMap<(u64, RequestId), RequestState>,
turns: HashMap<String, TurnState>,
connections: HashMap<u64, ConnectionState>,
thread_connections: HashMap<String, u64>,
}
struct ConnectionState {
@@ -89,6 +96,7 @@ struct TurnState {
started_at: Option<u64>,
token_usage: Option<TokenUsage>,
completed: Option<CompletedTurnState>,
steer_count: usize,
}
impl AnalyticsReducer {
@@ -135,6 +143,9 @@ impl AnalyticsReducer {
CustomAnalyticsFact::TurnTokenUsage(input) => {
self.ingest_turn_token_usage(*input, out);
}
CustomAnalyticsFact::TurnSteer(input) => {
self.ingest_turn_steer(input, out);
}
CustomAnalyticsFact::SkillInvoked(input) => {
self.ingest_skill_invoked(input, out).await;
}
@@ -198,6 +209,8 @@ impl AnalyticsReducer {
let ClientRequest::TurnStart { params, .. } = request else {
return;
};
self.thread_connections
.insert(params.thread_id.clone(), connection_id);
self.requests.insert(
(connection_id, request_id),
RequestState::TurnStart(PendingTurnStartState {
@@ -229,6 +242,7 @@ impl AnalyticsReducer {
started_at: None,
token_usage: None,
completed: None,
steer_count: 0,
});
turn_state.thread_id = Some(thread_id);
turn_state.num_input_images = Some(num_input_images);
@@ -250,6 +264,7 @@ impl AnalyticsReducer {
started_at: None,
token_usage: None,
completed: None,
steer_count: 0,
});
turn_state.thread_id = Some(input.thread_id);
turn_state.token_usage = Some(input.token_usage);
@@ -402,6 +417,7 @@ impl AnalyticsReducer {
started_at: None,
token_usage: None,
completed: None,
steer_count: 0,
});
turn_state.connection_id = Some(connection_id);
turn_state.thread_id = Some(pending_request.thread_id);
@@ -427,6 +443,7 @@ impl AnalyticsReducer {
started_at: None,
token_usage: None,
completed: None,
steer_count: 0,
});
turn_state.started_at = notification
.turn
@@ -445,6 +462,7 @@ impl AnalyticsReducer {
started_at: None,
token_usage: None,
completed: None,
steer_count: 0,
});
turn_state.completed = Some(CompletedTurnState {
status: analytics_turn_status(notification.turn.status),
@@ -478,6 +496,8 @@ impl AnalyticsReducer {
out: &mut Vec<TrackEventRequest>,
) {
let thread_source: SessionSource = thread.source.into();
self.thread_connections
.insert(thread.id.clone(), connection_id);
let Some(connection_state) = self.connections.get(&connection_id) else {
return;
};
@@ -500,6 +520,59 @@ impl AnalyticsReducer {
));
}
fn ingest_turn_steer(&mut self, input: TurnSteerInput, out: &mut Vec<TrackEventRequest>) {
let TurnSteerInput {
tracking,
turn_steer,
} = input;
if matches!(turn_steer.result, TurnSteerResult::Accepted)
&& let Some(accepted_turn_id) = turn_steer.accepted_turn_id.as_ref()
&& let Some(turn_state) = self.turns.get_mut(accepted_turn_id)
{
turn_state.steer_count += 1;
}
let Some((app_server_client, runtime)) =
self.connection_metadata_for_turn_steer(&tracking, &turn_steer)
else {
return;
};
out.push(TrackEventRequest::TurnSteer(CodexTurnSteerEventRequest {
event_type: "codex_turn_steer_event",
event_params: codex_turn_steer_event_params(
app_server_client,
runtime,
&tracking,
turn_steer,
),
}));
}
fn connection_metadata_for_turn_steer(
&self,
tracking: &TrackEventsContext,
turn_steer: &CodexTurnSteerEvent,
) -> Option<(CodexAppServerClientMetadata, CodexRuntimeMetadata)> {
let turn_connection_id = turn_steer
.accepted_turn_id
.as_ref()
.or(turn_steer.expected_turn_id.as_ref())
.or(Some(&tracking.turn_id))
.and_then(|turn_id| self.turns.get(turn_id))
.and_then(|turn_state| turn_state.connection_id);
let connection_id = turn_connection_id
.or_else(|| self.thread_connections.get(&tracking.thread_id).copied());
connection_id
.and_then(|connection_id| self.connections.get(&connection_id))
.map(|connection_state| {
(
connection_state.app_server_client.clone(),
connection_state.runtime.clone(),
)
})
}
fn maybe_emit_turn_event(&mut self, turn_id: &str, out: &mut Vec<TrackEventRequest>) {
let Some(turn_state) = self.turns.get(turn_id) else {
return;
@@ -585,7 +658,7 @@ fn codex_turn_event_params(
is_first_turn,
status: completed.status,
turn_error: completed.turn_error,
steer_count: None,
steer_count: Some(turn_state.steer_count),
total_tool_call_count: None,
shell_command_count: None,
file_change_count: None,
codex-rs/app-server/tests/suite/v2/turn_steer.rs
+50 -27
View File
@@ -6,6 +6,7 @@ use app_test_support::create_mock_responses_server_sequence;
use app_test_support::create_mock_responses_server_sequence_unchecked;
use app_test_support::create_shell_command_sse_response;
use app_test_support::to_response;
use app_test_support::write_mock_responses_config_toml_with_chatgpt_base_url;
use codex_app_server::INPUT_TOO_LARGE_ERROR_CODE;
use codex_app_server::INVALID_PARAMS_ERROR_CODE;
use codex_app_server_protocol::JSONRPCError;
@@ -23,6 +24,9 @@ use codex_protocol::user_input::MAX_USER_INPUT_TEXT_CHARS;
use tempfile::TempDir;
use tokio::time::timeout;
use super::analytics::enable_analytics_capture;
use super::analytics::wait_for_analytics_event;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[tokio::test]
@@ -32,7 +36,12 @@ async fn turn_steer_requires_active_turn() -> Result<()> {
std::fs::create_dir(&codex_home)?;
let server = create_mock_responses_server_sequence(vec![]).await;
create_config_toml(&codex_home, &server.uri())?;
write_mock_responses_config_toml_with_chatgpt_base_url(
&codex_home,
&server.uri(),
&server.uri(),
)?;
enable_analytics_capture(&server, &codex_home).await?;
let mut mcp = McpProcess::new(&codex_home).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
@@ -52,7 +61,7 @@ async fn turn_steer_requires_active_turn() -> Result<()> {
let steer_req = mcp
.send_turn_steer_request(TurnSteerParams {
thread_id: thread.id,
thread_id: thread.id.clone(),
input: vec![V2UserInput::Text {
text: "steer".to_string(),
text_elements: Vec::new(),
@@ -67,6 +76,21 @@ async fn turn_steer_requires_active_turn() -> Result<()> {
.await??;
assert_eq!(steer_err.error.code, -32600);
let event =
wait_for_analytics_event(&server, DEFAULT_READ_TIMEOUT, "codex_turn_steer_event").await?;
assert_eq!(event["event_params"]["thread_id"], thread.id);
assert_eq!(event["event_params"]["result"], "rejected");
assert_eq!(event["event_params"]["num_input_images"], 0);
assert_eq!(
event["event_params"]["expected_turn_id"],
"turn-does-not-exist"
);
assert_eq!(
event["event_params"]["accepted_turn_id"],
serde_json::Value::Null
);
assert_eq!(event["event_params"]["rejection_reason"], "no_active_turn");
Ok(())
}
@@ -95,7 +119,12 @@ async fn turn_steer_rejects_oversized_text_input() -> Result<()> {
"call_sleep",
)?])
.await;
create_config_toml(&codex_home, &server.uri())?;
write_mock_responses_config_toml_with_chatgpt_base_url(
&codex_home,
&server.uri(),
&server.uri(),
)?;
enable_analytics_capture(&server, &codex_home).await?;
let mut mcp = McpProcess::new(&codex_home).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
@@ -198,7 +227,12 @@ async fn turn_steer_returns_active_turn_id() -> Result<()> {
"call_sleep",
)?])
.await;
create_config_toml(&codex_home, &server.uri())?;
write_mock_responses_config_toml_with_chatgpt_base_url(
&codex_home,
&server.uri(),
&server.uri(),
)?;
enable_analytics_capture(&server, &codex_home).await?;
let mut mcp = McpProcess::new(&codex_home).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
@@ -258,31 +292,20 @@ async fn turn_steer_returns_active_turn_id() -> Result<()> {
let steer: TurnSteerResponse = to_response::<TurnSteerResponse>(steer_resp)?;
assert_eq!(steer.turn_id, turn.id);
let event =
wait_for_analytics_event(&server, DEFAULT_READ_TIMEOUT, "codex_turn_steer_event").await?;
assert_eq!(event["event_params"]["thread_id"], thread.id);
assert_eq!(event["event_params"]["result"], "accepted");
assert_eq!(event["event_params"]["num_input_images"], 0);
assert_eq!(event["event_params"]["expected_turn_id"], turn.id);
assert_eq!(event["event_params"]["accepted_turn_id"], turn.id);
assert_eq!(
event["event_params"]["rejection_reason"],
serde_json::Value::Null
);
mcp.interrupt_turn_and_wait_for_aborted(thread.id, steer.turn_id, DEFAULT_READ_TIMEOUT)
.await?;
Ok(())
}
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 = "danger-full-access"
model_provider = "mock_provider"
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "{server_uri}/v1"
wire_api = "responses"
request_max_retries = 0
stream_max_retries = 0
"#
),
)
}
codex-rs/core/src/codex.rs
+168 -18
View File
@@ -49,9 +49,13 @@ use chrono::Local;
use chrono::Utc;
use codex_analytics::AnalyticsEventsClient;
use codex_analytics::AppInvocation;
use codex_analytics::CodexTurnSteerEvent;
use codex_analytics::InvocationType;
use codex_analytics::SubAgentThreadStartedInput;
use codex_analytics::TrackEventsContext;
use codex_analytics::TurnResolvedConfigFact;
use codex_analytics::TurnSteerRejectionReason;
use codex_analytics::TurnSteerResult;
use codex_analytics::TurnSubmissionType;
use codex_analytics::build_track_events_context;
use codex_app_server_protocol::McpServerElicitationRequest;
@@ -240,6 +244,32 @@ impl SteerInputError {
},
}
}
fn to_turn_steer_rejection_reason(&self) -> TurnSteerRejectionReason {
match self {
Self::NoActiveTurn(_) => TurnSteerRejectionReason::NoActiveTurn,
Self::ExpectedTurnMismatch { .. } => TurnSteerRejectionReason::ExpectedTurnMismatch,
Self::ActiveTurnNotSteerable { turn_kind } => match turn_kind {
NonSteerableTurnKind::Review => TurnSteerRejectionReason::NonSteerableReview,
NonSteerableTurnKind::Compact => TurnSteerRejectionReason::NonSteerableCompact,
},
Self::EmptyInput => TurnSteerRejectionReason::EmptyInput,
}
}
}
struct AcceptedSteerInput {
turn_id: String,
tracking: TrackEventsContext,
expected_turn_id: String,
num_input_images: usize,
}
struct RejectedSteerInput {
error: SteerInputError,
tracking: TrackEventsContext,
expected_turn_id: Option<String>,
num_input_images: usize,
}
/// Notes from the previous real user turn.
@@ -4096,47 +4126,163 @@ impl Session {
input: Vec<UserInput>,
expected_turn_id: Option<&str>,
) -> Result<String, SteerInputError> {
if input.is_empty() {
return Err(SteerInputError::EmptyInput);
let created_at = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
match self
.try_append_input_to_active_turn(input, expected_turn_id)
.await
{
Ok(accepted) => {
self.services.analytics_events_client.track_turn_steer(
accepted.tracking,
CodexTurnSteerEvent {
expected_turn_id: Some(accepted.expected_turn_id),
accepted_turn_id: Some(accepted.turn_id.clone()),
num_input_images: accepted.num_input_images,
result: TurnSteerResult::Accepted,
rejection_reason: None,
created_at,
},
);
Ok(accepted.turn_id)
}
Err(rejected) => {
self.services.analytics_events_client.track_turn_steer(
rejected.tracking,
CodexTurnSteerEvent {
expected_turn_id: rejected.expected_turn_id,
accepted_turn_id: None,
num_input_images: rejected.num_input_images,
result: TurnSteerResult::Rejected,
rejection_reason: Some(rejected.error.to_turn_steer_rejection_reason()),
created_at,
},
);
Err(rejected.error)
}
}
}
async fn try_append_input_to_active_turn(
&self,
input: Vec<UserInput>,
expected_turn_id: Option<&str>,
) -> Result<AcceptedSteerInput, RejectedSteerInput> {
let thread_id = self.conversation_id.to_string();
let fallback_tracking = || {
build_track_events_context(
String::new(),
thread_id.clone(),
expected_turn_id.unwrap_or_default().to_string(),
)
};
if input.is_empty() {
return Err(RejectedSteerInput {
error: SteerInputError::EmptyInput,
tracking: fallback_tracking(),
expected_turn_id: expected_turn_id.map(str::to_string),
num_input_images: 0,
});
}
let num_input_images = input
.iter()
.filter(|item| matches!(item, UserInput::Image { .. } | UserInput::LocalImage { .. }))
.count();
let mut active = self.active_turn.lock().await;
let Some(active_turn) = active.as_mut() else {
return Err(SteerInputError::NoActiveTurn(input));
return Err(RejectedSteerInput {
error: SteerInputError::NoActiveTurn(input),
tracking: fallback_tracking(),
expected_turn_id: expected_turn_id.map(str::to_string),
num_input_images,
});
};
let Some((active_turn_id, _)) = active_turn.tasks.first() else {
return Err(SteerInputError::NoActiveTurn(input));
let Some((active_turn_id, task)) = active_turn.tasks.first() else {
return Err(RejectedSteerInput {
error: SteerInputError::NoActiveTurn(input),
tracking: fallback_tracking(),
expected_turn_id: expected_turn_id.map(str::to_string),
num_input_images,
});
};
let active_turn_id = active_turn_id.clone();
let tracking = build_track_events_context(
task.turn_context.model_info.slug.clone(),
thread_id.clone(),
task.turn_context.sub_id.clone(),
);
if let Some(expected_turn_id) = expected_turn_id
&& expected_turn_id != active_turn_id
{
return Err(SteerInputError::ExpectedTurnMismatch {
expected: expected_turn_id.to_string(),
actual: active_turn_id.clone(),
return Err(RejectedSteerInput {
error: SteerInputError::ExpectedTurnMismatch {
expected: expected_turn_id.to_string(),
actual: active_turn_id,
},
tracking,
expected_turn_id: Some(expected_turn_id.to_string()),
num_input_images,
});
}
match active_turn.tasks.first().map(|(_, task)| task.kind) {
Some(crate::state::TaskKind::Regular) => {}
Some(crate::state::TaskKind::Review) => {
return Err(SteerInputError::ActiveTurnNotSteerable {
turn_kind: NonSteerableTurnKind::Review,
return Err(RejectedSteerInput {
error: SteerInputError::ActiveTurnNotSteerable {
turn_kind: NonSteerableTurnKind::Review,
},
tracking,
expected_turn_id: expected_turn_id
.map(str::to_string)
.or(Some(active_turn_id)),
num_input_images,
});
}
Some(crate::state::TaskKind::Compact) => {
return Err(SteerInputError::ActiveTurnNotSteerable {
turn_kind: NonSteerableTurnKind::Compact,
return Err(RejectedSteerInput {
error: SteerInputError::ActiveTurnNotSteerable {
turn_kind: NonSteerableTurnKind::Compact,
},
tracking,
expected_turn_id: expected_turn_id
.map(str::to_string)
.or(Some(active_turn_id)),
num_input_images,
});
}
None => {
return Err(RejectedSteerInput {
error: SteerInputError::NoActiveTurn(input),
tracking: fallback_tracking(),
expected_turn_id: expected_turn_id.map(str::to_string),
num_input_images,
});
}
None => return Err(SteerInputError::NoActiveTurn(input)),
}
let mut turn_state = active_turn.turn_state.lock().await;
turn_state.push_pending_input(input.into());
turn_state.accept_mailbox_delivery_for_current_turn();
Ok(active_turn_id.clone())
drop(turn_state);
let expected_turn_id = expected_turn_id
.map(str::to_string)
.unwrap_or_else(|| active_turn_id.clone());
Ok(AcceptedSteerInput {
turn_id: active_turn_id,
tracking,
expected_turn_id,
num_input_images,
})
}
/// Returns the input if there was no task running to inject into.
@@ -4801,6 +4947,7 @@ fn submission_dispatch_span(sub: &Submission) -> tracing::Span {
/// Operation handlers
mod handlers {
use crate::codex::RejectedSteerInput;
use crate::codex::Session;
use crate::codex::SessionSettingsUpdate;
use crate::codex::SteerInputError;
@@ -4961,11 +5108,14 @@ mod handlers {
sess.maybe_emit_unknown_model_warning_for_turn(current_context.as_ref())
.await;
match sess
.steer_input(items.clone(), /*expected_turn_id*/ None)
.try_append_input_to_active_turn(items.clone(), /*expected_turn_id*/ None)
.await
{
Ok(_) => current_context.session_telemetry.user_prompt(&items),
Err(SteerInputError::NoActiveTurn(items)) => {
Err(RejectedSteerInput {
error: SteerInputError::NoActiveTurn(items),
..
}) => {
current_context.session_telemetry.user_prompt(&items);
sess.refresh_mcp_servers_if_requested(&current_context)
.await;
@@ -4976,10 +5126,10 @@ mod handlers {
)
.await;
}
Err(err) => {
Err(RejectedSteerInput { error, .. }) => {
sess.send_event_raw(Event {
id: sub_id,
msg: EventMsg::Error(err.to_error_event()),
msg: EventMsg::Error(error.to_error_event()),
})
.await;
}